Updated service class unit tests

.github/workflows/linting.yml

@@ -26,7 +26,7 @@ jobs:
     - name: Install dependencies
       run: |
         python -m pip install --upgrade pip
-        python -m pip install flake8 pytest
+        python -m pip install flake8 pytest coverage
         if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
     - name: Lint with flake8
       run: |
@@ -39,4 +39,5 @@ jobs:
         DEBUG_API_ID: ${{ secrets.DEBUG_API_ID }}
         DEBUG_API_SECRET: ${{ secrets.DEBUG_API_SECRET }}
       run: |
-        pytest
+        coverage run --source src -m pytest -s -W ignore::Warning
+        coverage report

tests/test_authorization.py

@@ -1,9 +1,17 @@
 # A valid CrowdStrike Falcon API key is required to run these tests. 
+# You can store these values in your environment (this is the preferred method).
+# Example:
+#    export DEBUG_API_ID=CLIENT_ID_GOES_HERE
+#    export DEBUG_API_SECRET=CLIENT_SECRET_GOES_HERE
+#
+# You may also store these values locally in a configuration file.
+# DO NOT SUBMIT A COMMIT OR A PR THAT INCLUDES YOUR CONFIGURATION FILE.
 # API client ID & secret should be stored in tests/test.config in JSON format.
 # {
 #    "falcon_client_id": "CLIENT_ID_GOES_HERE",
 #    "falcon_client_secret": "CLIENT_SECRET_GOES_HERE"
 # }
+
 import json
 import os
 import sys
@@ -13,51 +21,62 @@
 from falconpy import api_complete as FalconSDK
 from falconpy import oauth2 as FalconAuth
 
-
 # The TestAuthorization class tests authentication and deauthentication
 # for both the Uber and Service classes.
 class TestAuthorization():
     def getConfig(self):
         #Grab our config parameters
-        try:
+        if "DEBUG_API_ID" in os.environ and "DEBUG_API_SECRET" in os.environ:
             self.config = {}
             self.config["falcon_client_id"] = os.getenv("DEBUG_API_ID")
             self.config["falcon_client_secret"] = os.getenv("DEBUG_API_SECRET")
-        except:
-            with open('%s/test.config' % os.path.dirname(os.path.abspath(__file__)), 'r') as file_config:
-                self.config = json.loads(file_config.read())
+            return True
+        else:
+            cur_path = os.path.dirname(os.path.abspath(__file__))
+            if os.path.exists('%s/test.config' % cur_path):
+                with open('%s/test.config' % cur_path, 'r') as file_config:
+                    self.config = json.loads(file_config.read())
+                return True
+            else:
+                return False
 
     def uberAuth(self):
-        self.getConfig()       
-        self.falcon = FalconSDK.APIHarness(creds={
-                "client_id": self.config["falcon_client_id"],
-                "client_secret": self.config["falcon_client_secret"]
-            }
-        )
-        self.falcon.authenticate()
-        if self.falcon.authenticated:
-            return True
+        status = self.getConfig()       
+        if status:
+            self.falcon = FalconSDK.APIHarness(creds={
+                    "client_id": self.config["falcon_client_id"],
+                    "client_secret": self.config["falcon_client_secret"]
+                }
+            )
+            self.falcon.authenticate()
+            if self.falcon.authenticated:
+                return True
+            else:
+                return False
         else:
             return False
 
     def uberRevoke(self):
         return self.falcon.deauthenticate()
 
     def serviceAuth(self):
-        self.getConfig()
-        self.authorization = FalconAuth.OAuth2(creds={
-            'client_id': self.config["falcon_client_id"],
-            'client_secret': self.config["falcon_client_secret"]
-        })
+        status = self.getConfig()
+        if status:
+            self.authorization = FalconAuth.OAuth2(creds={
+                'client_id': self.config["falcon_client_id"],
+                'client_secret': self.config["falcon_client_secret"]
+            })
 
-        try:
-            self.token = self.authorization.token()['body']['access_token']
+            try:
+                self.token = self.authorization.token()['body']['access_token']
+
+            except:
+                self.token = False
 
-        except:
-            self.token = False
-
-        if self.token:
-            return True
+            if self.token:
+                return True
+            else:
+                return False
         else:
             return False
 

tests/test_cloud_connect_aws.py

@@ -1,12 +1,10 @@
-# A valid CrowdStrike Falcon API key is required to run these tests. 
-# API client ID & secret should be stored in tests/test.config in JSON format.
-# {
-#    "falcon_client_id": "CLIENT_ID_GOES_HERE",
-#    "falcon_client_secret": "CLIENT_SECRET_GOES_HERE"
-# }
+# test_cloud_connect_aws.py
+# This class tests the cloud_connect_aws service class
+
 import json
 import os
 import sys
+import pytest
 # Authentication via the test_authorization.py
 from tests import test_authorization as Authorization
 
@@ -15,19 +13,58 @@
 # Classes to test - manually imported from sibling folder
 from falconpy import cloud_connect_aws as FalconAWS
 
+auth = Authorization.TestAuthorization()
+auth.serviceAuth()
+falcon = FalconAWS.Cloud_Connect_AWS(access_token=auth.token)
+AllowedResponses = [200, 429] #Adding rate-limiting as an allowed response for now
 
-# The TestCloudConnectAWS class tests the cloud_connect_aws service class
 class TestCloudConnectAWS:
     def serviceCCAWS_GetAWSSettings(self):
-        auth = Authorization.TestAuthorization()
-        auth.serviceAuth()
-        falcon = FalconAWS.Cloud_Connect_AWS(access_token=auth.token)
-        if falcon.GetAWSSettings()["status_code"] > 0:
-            auth.serviceRevoke()
+        if falcon.GetAWSSettings()["status_code"] in AllowedResponses:
             return True
         else:
-            auth.serviceRevoke()
             return False      
 
+    def serviceCCAWS_QueryAWSAccounts(self):
+        if falcon.QueryAWSAccounts()["status_code"] in AllowedResponses:
+            return True
+        else:
+            return False
+
+    @pytest.mark.skipif(falcon.QueryAWSAccounts(parameters={"limit":1})["status_code"] == 429, reason="API rate limit reached")
+    def serviceCCAWS_GetAWSAccounts(self):
+        if falcon.GetAWSAccounts(ids=falcon.QueryAWSAccounts(parameters={"limit":1})["body"]["resources"][0]["id"])["status_code"] in AllowedResponses:
+            return True
+        else:
+            return False
+
+    @pytest.mark.skipif(falcon.QueryAWSAccounts(parameters={"limit":1})["status_code"] == 429, reason="API rate limit reached")    
+    def serviceCCAWS_VerifyAWSAccountAccess(self):
+        if falcon.VerifyAWSAccountAccess(ids=falcon.QueryAWSAccounts(parameters={"limit":1})["body"]["resources"][0]["id"])["status_code"] in AllowedResponses:
+            return True
+        else:
+            return False
+
+    def serviceCCAWS_QueryAWSAccountsForIDs(self):
+        if falcon.QueryAWSAccountsForIDs(parameters={"limit":1})["status_code"] in AllowedResponses:
+            return True
+        else:
+            return False
+
     def test_GetAWSSettings(self):
         assert self.serviceCCAWS_GetAWSSettings() == True
+
+    def test_QueryAWSAccounts(self):
+        assert self.serviceCCAWS_QueryAWSAccounts() == True
+
+    def test_GetAWSAccounts(self):
+        assert self.serviceCCAWS_GetAWSAccounts() == True
+
+    def test_VerifyAWSAccountAccess(self):
+        assert self.serviceCCAWS_VerifyAWSAccountAccess() == True
+
+    def test_QueryAWSAccountsForIDs(self):
+        assert self.serviceCCAWS_QueryAWSAccountsForIDs() == True
+
+    def test_logout(self):
+        assert auth.serviceRevoke() == True

tests/test_detects.py

@@ -0,0 +1,55 @@
+# test_detects.py
+# This class tests the detects service class
+
+import json
+import os
+import sys
+import pytest
+# Authentication via the test_authorization.py
+from tests import test_authorization as Authorization
+
+#Import our sibling src folder into the path
+sys.path.append(os.path.abspath('src'))
+# Classes to test - manually imported from sibling folder
+from falconpy import detects as FalconDetections
+
+auth = Authorization.TestAuthorization()
+auth.serviceAuth()
+falcon = FalconDetections.Detects(access_token=auth.token)
+AllowedResponses = [200, 429] #Adding rate-limiting as an allowed response for now
+
+class TestDetects:
+
+    def serviceDetects_QueryDetects(self):
+        if falcon.QueryDetects(parameters={"limit":1})["status_code"] in AllowedResponses:
+            return True
+        else:
+            return False
+
+    @pytest.mark.skipif(falcon.QueryDetects(parameters={"limit":1})["status_code"] == 429, reason="API rate limit reached")
+    def serviceDetects_GetDetectSummaries(self):
+        if falcon.GetDetectSummaries(body={"ids":falcon.QueryDetects(parameters={"limit":1})["body"]["resources"]})["status_code"] in AllowedResponses:
+            return True
+        else:
+            return False
+
+    # def serviceDetects_GetAggregateDetects(self):
+    #     auth, falcon = self.authenticate()
+    #     if falcon.GetAggregateDetects(body={"ids":falcon.QueryDetects(parameters={"limit":1})["body"]["resources"]})["status_code"] in AllowedResponses:
+    #         auth.serviceRevoke()
+    #         return True
+    #     else:
+    #         auth.serviceRevoke()
+    #         return False
+
+    def test_QueryDetects(self):
+        assert self.serviceDetects_QueryDetects() == True
+
+    def test_GetDetectSummaries(self):
+        assert self.serviceDetects_GetDetectSummaries() == True
+
+    # def test_GetAggregateDetects(self):
+    #     assert self.serviceDetects_GetAggregateDetects() == True
+
+    def test_logout(self):
+        assert auth.serviceRevoke() == True

tests/test_device_control_policies.py

@@ -0,0 +1,72 @@
+# test_device_control_poligies.py
+# This class tests the device_control_policies service class
+
+import json
+import os
+import sys
+import pytest
+# Authentication via the test_authorization.py
+from tests import test_authorization as Authorization
+
+#Import our sibling src folder into the path
+sys.path.append(os.path.abspath('src'))
+# Classes to test - manually imported from sibling folder
+from falconpy import device_control_policies as FalconDeviceControlPolicy
+
+auth = Authorization.TestAuthorization()
+auth.serviceAuth()
+falcon = FalconDeviceControlPolicy.Device_Control_Policies(access_token=auth.token)
+AllowedResponses = [200, 429] #Adding rate-limiting as an allowed response for now
+
+class TestDeviceControlPolicy:
+
+    def serviceDeviceControlPolicies_queryDeviceControlPolicies(self):
+        if falcon.queryDeviceControlPolicies(parameters={"limit":1})["status_code"] in AllowedResponses:
+            return True
+        else:
+            return False
+
+    @pytest.mark.skipif(falcon.queryDeviceControlPolicies(parameters={"limit":1})["status_code"] == 429, reason="API rate limit reached")
+    def serviceDeviceControlPolicies_queryDeviceControlPolicyMembers(self):
+        if falcon.queryDeviceControlPolicyMembers(parameters={"id": falcon.queryDeviceControlPolicies(parameters={"limit":1})["body"]["resources"][0]})["status_code"] in AllowedResponses:
+            return True
+        else:
+            return False
+
+    @pytest.mark.skipif(falcon.queryDeviceControlPolicies(parameters={"limit":1})["status_code"] == 429, reason="API rate limit reached")
+    def serviceDeviceControlPolicies_getDeviceControlPolicies(self):
+        if falcon.getDeviceControlPolicies(ids=falcon.queryDeviceControlPolicies(parameters={"limit":1})["body"]["resources"][0])["status_code"] in AllowedResponses:
+            return True
+        else:
+            return False
+
+    def serviceDeviceControlPolicies_queryCombinedDeviceControlPolicies(self):
+        if falcon.queryCombinedDeviceControlPolicies(parameters={"limit":1})["status_code"] in AllowedResponses:
+            return True
+        else:
+            return False
+
+    @pytest.mark.skipif(falcon.queryCombinedDeviceControlPolicies(parameters={"limit":1})["status_code"] == 429, reason="API rate limit reached")
+    def serviceDeviceControlPolicies_queryCombinedDeviceControlPolicyMembers(self):
+        if falcon.queryCombinedDeviceControlPolicyMembers(parameters={"id": falcon.queryCombinedDeviceControlPolicies(parameters={"limit":1})["body"]["resources"][0]["id"]})["status_code"] in AllowedResponses:
+            return True
+        else:
+            return False
+
+    def test_queryDeviceControlPolicies(self):
+        assert self.serviceDeviceControlPolicies_queryDeviceControlPolicies() == True
+
+    def test_queryDeviceControlPolicyMembers(self):
+        assert self.serviceDeviceControlPolicies_queryDeviceControlPolicyMembers() == True
+
+    def test_getDeviceControlPolicies(self):
+        assert self.serviceDeviceControlPolicies_getDeviceControlPolicies() == True
+
+    def test_queryCombinedDeviceControlPolicies(self):
+        assert self.serviceDeviceControlPolicies_queryCombinedDeviceControlPolicies() == True
+
+    def test_queryCombinedDeviceControlPolicyMembers(self):
+        assert self.serviceDeviceControlPolicies_queryCombinedDeviceControlPolicyMembers() == True
+
+    def test_logout(self):
+        assert auth.serviceRevoke() == True

tests/test_event_streams.py

@@ -0,0 +1,49 @@
+# test_event_streams.py
+# This class tests the event_streams service class
+
+import json
+import os
+import sys
+import datetime
+import requests
+import pytest
+# Authentication via the test_authorization.py
+from tests import test_authorization as Authorization
+
+#Import our sibling src folder into the path
+sys.path.append(os.path.abspath('src'))
+# Classes to test - manually imported from sibling folder
+from falconpy import event_streams as FalconStream
+
+auth = Authorization.TestAuthorization()
+auth.serviceAuth()
+falcon = FalconStream.Event_Streams(access_token=auth.token)
+AllowedResponses = [200, 429] #Adding rate-limiting as an allowed response for now
+appId = "pytest-event_streams-unit-test"
+class TestEventStreams:
+
+    def serviceStream_listAvailableStreamsOAuth2(self):
+        if falcon.listAvailableStreamsOAuth2(parameters={"appId":appId})["status_code"] in AllowedResponses:
+            return True
+        else:
+            return False
+    @pytest.mark.skipif(falcon.listAvailableStreamsOAuth2(parameters={"appId":appId})["status_code"] == 429, reason="API rate limit reached")
+    def serviceStream_refreshActiveStreamSession(self):
+        avail = falcon.listAvailableStreamsOAuth2(parameters={"appId":appId})
+        t1 = datetime.datetime.utcnow().strftime('%a, %d %b %Y %H:%M:%S +0000')
+        headers = {'Authorization': 'Token %s' % (avail["body"]["resources"][0]["sessionToken"]["token"]), 'Date': t1, 'Connection': 'Keep-Alive'}
+        stream = requests.get(avail["body"]["resources"][0]["dataFeedURL"], headers=headers, stream=True)
+        with stream:
+            if falcon.refreshActiveStreamSession(parameters={"appId": appId, "action_name":"refresh_active_stream_session"}, partition=0)["status_code"] in AllowedResponses:
+                return True
+            else:
+                return False
+
+    def test_listAvailableStreamsOAuth2(self):
+        assert self.serviceStream_listAvailableStreamsOAuth2() == True
+
+    def test_refreshActiveStreamSession(self):
+        assert self.serviceStream_refreshActiveStreamSession() == True
+
+    def test_logout(self):
+        assert auth.serviceRevoke() == True