Refactor test cases and add new accounts test case

configurations/system/tests.py

@@ -13,6 +13,7 @@
 from .tests_cases.security_risks_tests import SecurityRisksTests
 from .tests_cases.vuln_scan_tests import VulnerabilityScanningTests
 from .tests_cases.workflows_tests import WorkflowsTests
+from .tests_cases.accounts_tests import AccountsTests
 
 
 def all_tests_names():
@@ -34,6 +35,7 @@ def all_tests_names():
     tests.extend(TestUtil.get_class_methods(SeccompProfileTests))
     tests.extend(TestUtil.get_class_methods(WorkflowsTests))
     tests.extend(TestUtil.get_class_methods(RegistryTests))
+    tests.extend(TestUtil.get_class_methods(AccountsTests))
     return tests
 
 
@@ -70,6 +72,8 @@ def get_test(test_name):
         return WorkflowsTests().__getattribute__(test_name)()
     if test_name in TestUtil.get_class_methods(RegistryTests):
         return RegistryTests().__getattribute__(test_name)()
+    if test_name in TestUtil.get_class_methods(AccountsTests):
+        return AccountsTests().__getattribute__(test_name)()
 
 
 ALL_TESTS = all_tests_names()

configurations/system/tests_cases/accounts_tests.py

@@ -0,0 +1,14 @@
+import inspect
+
+from .structures import KubescapeConfiguration
+
+
+class AccountsTests(object):
+
+    @staticmethod
+    def accounts():
+        from tests_scripts.accounts.accounts import Accounts
+        return KubescapeConfiguration(
+            name=inspect.currentframe().f_code.co_name,
+            test_obj=Accounts,
+        )

infrastructure/backend_api.py

@@ -142,6 +142,13 @@ class NotExistingCustomer(Exception):
 API_WEBHOOKS = "/api/v1/notifications/teams"
 API_TEAMS_TEST_MESSAGE = "/api/v1/notifications/teams/testMessage"
 
+API_ACCOUNTS = "/api/v1/accounts"
+API_ACCOUNTS_CLOUD_LIST = "/api/v1/accounts/cloud/list"
+API_ACCOUNTS_KUBERNETES_LIST = "/api/v1/accounts/kubernetes/list"
+API_ACCOUNTS_AWS_REGIONS = "/api/v1/accounts/aws/regions"
+API_ACCOUNTS_CLOUD_UNIQUEVALUES = "/api/v1/accounts/cloud/uniquevalues"
+API_ACCOUNTS_KUBERNETES_UNIQUEVALUES = "/api/v1/accounts/kubernetes/uniquevalues"
+
 
 
 def deco_cookie(func):
@@ -3085,6 +3092,79 @@ def test_webhook_message(self, body):
         return r.json()
 
 
+    def get_cloud_accounts(self,  body=None, **kwargs):
+        url = API_ACCOUNTS_CLOUD_LIST
+        if body is None:
+            body = {"pageSize": 150, "pageNum": 1}
+
+        params = {"customerGUID": self.selected_tenant_id}
+        if kwargs:
+            params.update(**kwargs)
+        r = self.post(url, params=params, json=body)
+        if not 200 <= r.status_code < 300:
+            raise Exception(
+                'Error accessing cloud accounts. Customer: "%s" (code: %d, message: %s)' % (
+                    self.customer, r.status_code, r.text))
+        return r.json()
+
+    def get_kubernetes_accounts(self,  body=None, **kwargs):
+        url = API_ACCOUNTS_KUBERNETES_LIST
+        if body is None:
+            body = {"pageSize": 150, "pageNum": 1}
+
+        params = {"customerGUID": self.selected_tenant_id}
+        if kwargs:
+            params.update(**kwargs)
+        r = self.post(url, params=params, json=body)
+        if not 200 <= r.status_code < 300:
+            raise Exception(
+                'Error accessing cloud accounts. Customer: "%s" (code: %d, message: %s)' % (
+                    self.customer, r.status_code, r.text))
+        return r.json()
+
+
+    def create_cloud_account(self, body, provider):
+        url = API_ACCOUNTS
+        params = {"customerGUID": self.selected_tenant_id,
+                  "provider": provider}
+        r = self.post(url, params=params, json=body)
+        if not 200 <= r.status_code < 300:
+            raise Exception(
+                'Error creating cloud account. Customer: "%s" (code: %d, message: %s)' % (
+                    self.customer, r.status_code, r.text))
+        return r.json()
+
+    def delete_cloud_account(self, guid):
+        url = API_ACCOUNTS
+        params = {"customerGUID": self.selected_tenant_id}
+        body = {
+            "innerFilters": [
+                {
+                    "guid": guid
+                }
+            ]
+        }
+        r = self.delete(url, params=params, json=body)
+        if not 200 <= r.status_code < 300:
+            raise Exception(
+                'Error deleting cloud account. Customer: "%s" (code: %d, message: %s)' % (
+                    self.customer, r.status_code, r.text))
+        return r.json()
+
+
+
+    def update_cloud_account(self, body, provider):
+        url = API_ACCOUNTS
+        params = {"customerGUID": self.selected_tenant_id,
+                  "provider": provider}
+        r = self.put(url, params=params, json=body)
+        if not 200 <= r.status_code < 300:
+            raise Exception(
+                'Error updating cloud account. Customer: "%s" (code: %d, message: %s)' % (
+                    self.customer, r.status_code, r.text))
+        return r.json()
+
+
 
 
 

system_test_mapping.json

@@ -1689,5 +1689,17 @@
     "description": "Checks workflows configurations",
     "skip_on_environment": "",
     "owner": "[email protected]"
+  },
+  "accounts": {
+    "target": [
+      "Backend"
+    ],
+    "target_repositories": [
+      "cadashboardbe",
+      "config-service"
+    ],
+    "description": "Checks accounts",
+    "skip_on_environment": "",
+    "owner": ""
   }
 }

tests_scripts/accounts/accounts.py

@@ -0,0 +1,93 @@
+
+from systest_utils import Logger, statics
+from tests_scripts.helm.base_helm import BaseHelm
+
+
+
+
+class Accounts(BaseHelm):
+    def __init__(self, test_obj=None, backend=None, kubernetes_obj=None, test_driver=None):
+        super().__init__(test_driver=test_driver, test_obj=test_obj, backend=backend, kubernetes_obj=kubernetes_obj)
+
+
+        self.helm_kwargs = {
+            "capabilities.vulnerabilityScan": "disable",
+            "grypeOfflineDB.enabled": "false",
+            "capabilities.relevancy": "disabled",
+            "capabilities.runtimeObservability": "disable",
+            "capabilities.malwareDetection": "disable",
+            "capabilities.runtimeDetection": "disable",
+            "alertCRD.installDefault": False,
+            "alertCRD.scopeClustered": False,
+        }
+        test_helm_kwargs = self.test_obj.get_arg("helm_kwargs")
+        if test_helm_kwargs:
+            self.helm_kwargs.update(test_helm_kwargs)
+
+        self.fw_name = None
+        self.cluster = None
+        self.wait_for_agg_to_end = False
+
+
+    def start(self):
+        """
+        Agenda:
+        1. Install kubescape with helm-chart
+        2. Validate accounts kubernetes list.
+        3. Validate accounts kubernetes uniquevalues.
+        4. Create bad arn cloud account with cspm.
+        5. Create good arn cloud account with cspm.
+        6. Validate accounts cloud with cspm list.
+        7. Validate accounts cloud with cspm uniquevalues.
+        8. Edit cloud account with cspm.
+        9. validate cloud account after edit.
+        10. Delete cloud account with cspm.
+        11. Validate cloud account after delete.
+        """
+
+        assert self.backend is not None, f'the test {self.test_driver.test_name} must run with backend'
+        self.cluster, self.namespace = self.setup(apply_services=False)
+
+        Logger.logger.info('Stage 1: Install kubescape with helm-chart')
+        self.install_kubescape(helm_kwargs=self.helm_kwargs)
+
+        Logger.logger.info('Stage 2: Validate accounts kubernetes list')
+        self.validate_accounts_kubernetes_list(self.cluster)
+
+
+        return self.cleanup()
+
+
+    def cleanup(self, **kwargs):
+        return super().cleanup(**kwargs)
+
+
+    def install_kubescape(self, helm_kwargs: dict = None):
+        self.add_and_upgrade_armo_to_repo()
+        self.install_armo_helm_chart(helm_kwargs=helm_kwargs)
+        self.verify_running_pods(namespace=statics.CA_NAMESPACE_FROM_HELM_NAME)
+
+
+    def validate_accounts_kubernetes_list(self, cluster:str):
+        """
+        Validate accounts kubernetes list.
+        """
+
+        body = {
+            "pageSize": 100,
+            "pageNum": 1,
+            "innerFilters": [{
+                "cluster": cluster
+            }]
+        }
+
+        r, t = self.wait_for_report(
+            self.backend.get_kubernetes_accounts, 
+            timeout=180,
+            body=body
+            )
+
+        assert "response" in r, f"response not in {r}"
+        assert len(r["response"]) > 0, f"response is empty"
+        assert r["response"][0]["cluster"] == cluster, f"cluster is not {cluster}"
+