chore(tests): add kubernetes provider tests (#3265)

2026-07-24 04:51:51 +00:00 · 2024-02-19 13:50:42 +01:00
parent 3cc367e0a3
commit 14bab496b5
4 changed files with 208 additions and 2 deletions
@@ -1,5 +1,6 @@
 import os
 import sys
+from argparse import Namespace
 from typing import Any, Optional

 from colorama import Fore, Style
@@ -17,9 +18,8 @@ class KubernetesProvider(Provider):
    audit_metadata: Optional[Any]
    audit_config: Optional[dict]

-    def __init__(self, arguments: dict):
+    def __init__(self, arguments: Namespace):
        logger.info("Instantiating Kubernetes Provider ...")
-
        self.api_client, self.context = self.setup_session(
            arguments.kubeconfig_file, arguments.context
        )
@@ -0,0 +1,21 @@
+from kubernetes import client
+
+from prowler.providers.common.models import Audit_Metadata
+from prowler.providers.kubernetes.lib.audit_info.models import Kubernetes_Audit_Info
+
+
+# Mocked Kubernetes Audit Info
+def set_mocked_audit_info():
+    audit_info = Kubernetes_Audit_Info(
+        api_client=client.ApiClient,
+        context=None,
+        audit_config=None,
+        audit_resources=[],
+        audit_metadata=Audit_Metadata(
+            services_scanned=0,
+            expected_checks=[],
+            completed_checks=0,
+            audit_progress=0,
+        ),
+    )
+    return audit_info
@@ -0,0 +1,172 @@
+import io
+import sys
+from argparse import Namespace
+from unittest.mock import MagicMock, patch
+
+from prowler.providers.kubernetes.kubernetes_provider_new import KubernetesProvider
+
+
+class TestKubernetesProvider:
+    @patch("kubernetes.client.RbacAuthorizationV1Api")
+    @patch("kubernetes.client.ApiClient")
+    @patch("kubernetes.config.load_kube_config")
+    @patch("kubernetes.config.load_incluster_config")
+    @patch("kubernetes.config.list_kube_config_contexts")
+    def test_setup_session(
+        self,
+        mock_list_kube_config_contexts,
+        mock_load_incluster_config,
+        mock_load_kube_config,
+        mock_api_client,
+        mock_rbac_api,
+    ):
+        # Mocking the return value of list_kube_config_contexts
+        mock_list_kube_config_contexts.return_value = (
+            [
+                {
+                    "name": "context_name",
+                    "context": {"cluster": "test-cluster", "user": "test-user"},
+                }
+            ],
+            0,
+        )
+
+        # Create a Namespace object for arguments
+        args = Namespace(kubeconfig_file="dummy_path", context=None, only_logs=False)
+
+        # Instantiate the KubernetesProvider with mocked arguments
+        provider = KubernetesProvider(args)
+
+        # Assert that an ApiClient has been created
+        assert provider.api_client is not None
+
+        # Assert that the context has been correctly set
+        assert provider.context == {
+            "name": "context_name",
+            "context": {"cluster": "test-cluster", "user": "test-user"},
+        }
+
+    @patch("kubernetes.client.RbacAuthorizationV1Api")
+    @patch("kubernetes.config.list_kube_config_contexts")
+    @patch("kubernetes.config.load_incluster_config")
+    @patch("kubernetes.config.load_kube_config")
+    def test_get_context_user_roles(
+        self,
+        mock_load_kube_config,
+        mock_load_incluster_config,
+        mock_list_kube_config_contexts,
+        mock_rbac_api,
+    ):
+        mock_list_kube_config_contexts.return_value = (
+            [
+                {
+                    "name": "context_name",
+                    "context": {"cluster": "test-cluster", "user": "test-user"},
+                }
+            ],
+            0,
+        )
+
+        # Mock the RbacAuthorizationV1Api methods
+        cluster_role_binding = MagicMock()
+        role_binding = MagicMock()
+        cluster_role_binding.list_cluster_role_binding.return_value = MagicMock(
+            items=[]
+        )
+        role_binding.list_role_binding_for_all_namespaces.return_value = MagicMock(
+            items=[]
+        )
+
+        mock_rbac_api.return_value = MagicMock(
+            list_cluster_role_binding=cluster_role_binding.list_cluster_role_binding,
+            list_role_binding_for_all_namespaces=role_binding.list_role_binding_for_all_namespaces,
+        )
+
+        args = Namespace(kubeconfig_file=None, context=None, only_logs=False)
+        provider = KubernetesProvider(args)
+
+        roles = provider.get_context_user_roles()
+
+        assert isinstance(roles, list)
+
+    @patch("kubernetes.client.RbacAuthorizationV1Api")
+    @patch("kubernetes.client.ApiClient")
+    @patch("kubernetes.config.load_kube_config")
+    @patch("kubernetes.config.load_incluster_config")
+    @patch("kubernetes.config.list_kube_config_contexts")
+    @patch("sys.stdout", new_callable=MagicMock)
+    def test_print_credentials(
+        self,
+        mock_stdout,
+        mock_list_kube_config_contexts,
+        mock_load_incluster_config,
+        mock_load_kube_config,
+        mock_api_client,
+        mock_rbac_api,
+    ):
+        mock_list_kube_config_contexts.return_value = (
+            [
+                {
+                    "name": "context_name",
+                    "context": {"cluster": "test-cluster", "user": "test-user"},
+                }
+            ],
+            0,
+        )
+
+        args = Namespace(kubeconfig_file=None, context=None, only_logs=False)
+        provider = KubernetesProvider(args)
+        provider.context = {
+            "context": {"cluster": "test-cluster", "user": "test-user"},
+            "namespace": "default",
+        }
+        provider.get_context_user_roles = MagicMock(return_value=["ClusterRole: admin"])
+
+        # Capture print output
+        captured_output = io.StringIO()
+        sys.stdout = captured_output
+
+        provider.print_credentials()
+
+        # Reset standard output
+        sys.stdout = sys.__stdout__
+
+        output = captured_output.getvalue()
+        assert "[test-cluster]" in output
+        assert "[test-user]" in output
+        assert "[default]" in output
+        assert "[ClusterRole: admin]" in output
+
+    @patch("kubernetes.client.RbacAuthorizationV1Api")
+    @patch("kubernetes.config.list_kube_config_contexts")
+    @patch("kubernetes.config.load_incluster_config")
+    @patch("kubernetes.config.load_kube_config")
+    def test_search_and_save_roles(
+        self,
+        mock_load_kube_config,
+        mock_load_incluster_config,
+        mock_list_kube_config_contexts,
+        mock_rbac_api,
+    ):
+        mock_list_kube_config_contexts.return_value = (
+            [
+                {
+                    "name": "context_name",
+                    "context": {"cluster": "test-cluster", "user": "test-user"},
+                }
+            ],
+            0,
+        )
+        mock_rbac_api.return_value.list_cluster_role_binding.return_value = MagicMock(
+            items=[]
+        )
+        mock_rbac_api.return_value.list_role_binding_for_all_namespaces.return_value = (
+            MagicMock(items=[])
+        )
+
+        args = Namespace(kubeconfig_file=None, context=None, only_logs=False)
+        provider = KubernetesProvider(args)
+        provider.context = {"context": {"user": "test-user"}}
+
+        roles = provider.search_and_save_roles([], [], "test-user", "ClusterRole")
+        assert isinstance(roles, list)
@@ -0,0 +1,13 @@
+from kubernetes import client
+
+from prowler.providers.kubernetes.lib.service.service import KubernetesService
+from tests.providers.kubernetes.audit_info_utils import set_mocked_audit_info
+
+
+class Test_KubernetesService:
+    def test_KubernetesService_init(self):
+        audit_info = set_mocked_audit_info()
+        service = KubernetesService(audit_info)
+
+        assert service.context is None
+        assert service.api_client == client.ApiClient