feat(tests): check_using_test_cluster: whitelisted node names (#27)

CHANGELOG.md

@@ -1,5 +1,10 @@
 # wiremind-kubernetes
 
+## v7.3.0 (2023-07-21)
+### Feature
+- tests: rename `check_not_using_wiremind_cluster` to `check_using_test_cluster`
+- test: check_using_test_cluster: also check for whitelisted node names in case of remote test cluster
+
 ## v7.2.0 (2023-06-22)
 ### Feature
 - kubernetes: support NetworkingV1Api

VERSION

@@ -1 +1 @@
-7.2.0
+7.3.0

pyproject.toml

@@ -21,3 +21,7 @@ exclude = '''
   | documentation
 )
 '''
+
+[tool.bandit]
+exclude_dirs = ["./.git", "./kubernetes", "./documentation", "./src/wiremind_kubernetes/tests", "./.venv", "./build"]
+skips = []

setup.cfg

@@ -26,6 +26,17 @@ no_implicit_optional = True
 show_error_codes = True
 files = src
 
+[isort]
+src_paths = src,tests
+profile = black
+line_length = 120
+sections=FUTURE,STDLIB,THIRDPARTY,FIRSTPARTY,LOCALFOLDER
+no_lines_before = LOCALFOLDER
+multi_line_output = 3
+include_trailing_comma = True
+use_parentheses = True
+force_grid_wrap = 0
+
 [tool:pytest]
 log_level=INFO
 # Deterministic ordering for tests; useful for pytest-xdist.

setup.py

@@ -1,7 +1,7 @@
 """
 wiremind-kubernetes
 """
-from setuptools import setup, find_packages
+from setuptools import find_packages, setup
 
 with open("VERSION") as version_file:
     version = version_file.read().strip()
@@ -18,6 +18,8 @@
     [
         "flake8",
         "black",
+        "isort",
+        "bandit",
         "flake8-mutable",
         "pip-tools>=3.7.0",
         "pyupgrade",

src/wiremind_kubernetes/__init__.py

@@ -1,7 +1,3 @@
 # noqa: F401
-from .kubernetes_helper import (  # noqa: F401
-    KubernetesHelper,
-    NamespacedKubernetesHelper,
-    KubernetesDeploymentManager,
-)
+from .kubernetes_helper import KubernetesDeploymentManager, KubernetesHelper, NamespacedKubernetesHelper  # noqa: F401
 from .utils import run_command  # noqa: F401

src/wiremind_kubernetes/kubernetes_helper.py

@@ -6,13 +6,16 @@
 import kubernetes
 
 from wiremind_kubernetes.exceptions import PodNotFound
-
 from .kube_config import load_kubernetes_config
 from .kubernetes_client_additional_arguments import (
-    AppV1ApiWithArguments, AutoscalingV1ApiWithArguments,
-    BatchV1ApiWithArguments, CoreV1ApiWithArguments,
-    CustomObjectsApiWithArguments, NetworkingV1ApiWithArguments,
-    RbacAuthorizationV1ApiWithArguments)
+    AppV1ApiWithArguments,
+    AutoscalingV1ApiWithArguments,
+    BatchV1ApiWithArguments,
+    CoreV1ApiWithArguments,
+    CustomObjectsApiWithArguments,
+    NetworkingV1ApiWithArguments,
+    RbacAuthorizationV1ApiWithArguments,
+)
 from .utils import retry_kubernetes_request, retry_kubernetes_request_no_ignore
 
 logger = logging.getLogger(__name__)
@@ -322,7 +325,7 @@ def start_pods(self) -> None:
             expected_scale: int
             if len(priority_dict):
                 scaled = True
-                for (name, expected_scale) in priority_dict.items():
+                for name, expected_scale in priority_dict.items():
                     self.re_enable_hpa(deployment_name=name)
                     self.scale_up_deployment(name, expected_scale)
         if scaled:

src/wiremind_kubernetes/tests/e2e_tests/conftest.py

@@ -8,7 +8,7 @@
 from pytest_mock import MockerFixture
 
 import wiremind_kubernetes
-from wiremind_kubernetes.tests.e2e_tests.helpers import check_not_using_wiremind_cluster
+from wiremind_kubernetes.tests.e2e_tests.helpers import check_using_test_cluster
 from wiremind_kubernetes.utils import run_command
 
 E2E_CLUSTER_MANIFESTS = "tests/e2e_tests/manifests"
@@ -27,7 +27,7 @@ def k8s_client_request_function(mocker: MockerFixture) -> Generator:
 
 @pytest.fixture(scope="session", autouse=True)
 def setUpE2E() -> None:
-    check_not_using_wiremind_cluster()
+    check_using_test_cluster()
 
 
 def delete_namespace() -> None:

src/wiremind_kubernetes/tests/e2e_tests/create_job_test.py

@@ -6,7 +6,6 @@
 from pytest_mock import MockerFixture
 
 from wiremind_kubernetes import KubernetesDeploymentManager
-
 from .conftest import TEST_NAMESPACE
 
 logger = logging.getLogger(__name__)

src/wiremind_kubernetes/tests/e2e_tests/helpers.py

@@ -3,33 +3,72 @@
 import subprocess
 import sys
 import urllib
-from typing import Any, Dict
+from typing import Any, Dict, List
+
+import kubernetes
 
 from wiremind_kubernetes import run_command
 
 logger = logging.getLogger(__name__)
 
+DEFAULT_TEST_IPS_WHITELISTED = [
+    "localhost",  # minikube
+    "127.0.0.1",  # kind
+    "kubernetes.docker.internal",  # Docker for Mac
+]
+
+DEFAULT_TEST_NODES_WHITELISTED = ["minikube", "kind-control-plane", "kind", "kind-worker"]
 
-def check_not_using_wiremind_cluster() -> None:
+
+def get_default_kube_context() -> str:
     """
-    Will sys.exit(1) if kubectl current context api server is not a test cluster (like kind, minikube, etc)
+    Retrieves the default kube context
     """
-    logger.info("[CLUSTER-CONFIG]: Making sure the tests are not running against the main cluster...")
+    try:
+        return kubernetes.config.list_kube_config_contexts()[1]["name"]
+    except kubernetes.config.config_exception.ConfigException:
+        # Inside a Container maybe. This will make it use the incluster config.
+        return ""
 
+
+def is_ip_whitelisted(*, ips_whitelisted: List[str]) -> bool:
     api_server = subprocess.check_output(
         "kubectl config view --minify | grep server | cut -f 2- -d ':' | tr -d ' '",
         shell=True,
-        universal_newlines=True,
+        text=True,
     )
-    whitelisted_api_server_hostname_list = [
-        "localhost",  # minikube
-        "127.0.0.1",  # kind
-        "kubernetes.docker.internal",  # Docker for Mac
-    ]
+
     hostname = urllib.parse.urlparse(api_server.lower().strip()).hostname
-    if hostname not in whitelisted_api_server_hostname_list:
-        logger.error("Attempted to run tests with non-test cluster %s, abort!", api_server)
+    return hostname in ips_whitelisted
+
+
+def is_node_whitelisted(*, nodes_whitelisted: List[str]) -> bool:
+    output, *_ = run_command("kubectl get nodes -o name", return_result=True)
+    cluster_nodes = [x for x in output.replace("node/", "").split("\n") if x != ""]
+    for node in cluster_nodes:
+        if node not in nodes_whitelisted:
+            return False
+    return True
+
+
+def check_using_test_cluster(
+    *,
+    ips_whitelisted: List[str] = DEFAULT_TEST_IPS_WHITELISTED,
+    nodes_whitelisted: List[str] = DEFAULT_TEST_NODES_WHITELISTED,
+) -> bool:
+    """
+    Will sys.exit(1) if kubectl current context api server is not a test cluster (like kind, minikube, etc)
+    """
+    logger.info("[CLUSTER-CONFIG]: Making sure the tests are running against a test cluster...")
+
+    kube_context = get_default_kube_context()
+
+    if not is_ip_whitelisted(ips_whitelisted=ips_whitelisted) and not is_node_whitelisted(
+        nodes_whitelisted=nodes_whitelisted
+    ):
+        logger.error(f"Attempted to run tests with non-test cluster <{kube_context}>, aborting!")
         sys.exit(1)
+    return True
 
 
 def get_k8s_username() -> str:
@@ -43,9 +82,7 @@ def get_k8s_username() -> str:
     | $user.name'"""
     # dex-k8s-authenticator sets the user to: user={{ .Username}}-{{.ClusterName }}`
     # https://github.com/mintel/dex-k8s-authenticator/blob/master/templates/linux-mac-common.html#L101
-    username = (
-        subprocess.check_output(command, shell=True, universal_newlines=True).replace('"', "").strip().split("-")[0]
-    )
+    username = subprocess.check_output(command, shell=True, text=True).replace('"', "").strip().split("-")[0]
     assert username
     return username