fix liveness test and add more type hints

Author: jvpasinatto

e2e-tests/conftest.py

@@ -1,4 +1,5 @@
 import os
+from typing import Dict
 import pytest
 import subprocess
 import logging
@@ -20,7 +21,7 @@
 
 
 @pytest.hookimpl(tryfirst=True, hookwrapper=True)
-def pytest_runtest_makereport(item, call):
+def pytest_runtest_makereport(item: pytest.Item, call: pytest.CallInfo) -> None:
     """Collect K8s resources when a test fails."""
     outcome = yield
     report = outcome.get_result()
@@ -81,7 +82,7 @@ def setup_env_vars() -> None:
 
 
 @pytest.fixture(scope="class")
-def test_paths(request):
+def test_paths(request: pytest.FixtureRequest) -> Dict[str, Path]:
     """Fixture to provide paths relative to the test file."""
     test_file = Path(request.fspath)
     test_dir = test_file.parent
@@ -159,7 +160,7 @@ def _create_namespace(namespace):
 
 
 @pytest.fixture(scope="class")
-def create_infra(test_paths, create_namespace):
+def create_infra(test_paths: Dict[str, Path], create_namespace):
     global _current_namespace
     created_namespaces = []
 
@@ -191,7 +192,7 @@ def _create_infra(test_name):
     _current_namespace = None
 
     if os.environ.get("SKIP_DELETE") == "1":
-        logger.info("SKIP_DELETE = 1. Skipping test environment cleanup")
+        logger.info("SKIP_DELETE=1. Skipping test environment cleanup")
         return
 
     def run_cmd(cmd: list[str]) -> None:
@@ -375,7 +376,81 @@ def deploy_cert_manager():
 
 
 @pytest.fixture(scope="class")
-def psmdb_client(test_paths) -> tools.MongoManager:
+def deploy_minio():
+    """Deploy MinIO and clean up after tests."""
+    service_name = "minio-service"
+    bucket = "operator-testing"
+
+    logger.info(f"Installing MinIO: {service_name}")
+
+    subprocess.run(["helm", "uninstall", service_name], capture_output=True, check=False)
+    subprocess.run(["helm", "repo", "remove", "minio"], capture_output=True, check=False)
+    subprocess.run(["helm", "repo", "add", "minio", "https://charts.min.io/"], check=True)
+
+    endpoint = f"http://{service_name}:9000"
+    minio_args = [
+        "helm", "install", service_name, "minio/minio",
+        "--version", os.environ.get("MINIO_VER"),
+        "--set", "replicas=1",
+        "--set", "mode=standalone",
+        "--set", "resources.requests.memory=256Mi",
+        "--set", "rootUser=rootuser",
+        "--set", "rootPassword=rootpass123",
+        "--set", "users[0].accessKey=some-access-key",
+        "--set", "users[0].secretKey=some-secret-key",
+        "--set", "users[0].policy=consoleAdmin",
+        "--set", "service.type=ClusterIP",
+        "--set", "configPathmc=/tmp/",
+        "--set", "securityContext.enabled=false",
+        "--set", "persistence.size=2G",
+        "--set", f"fullnameOverride={service_name}",
+        "--set", "serviceAccount.create=true",
+        "--set", f"serviceAccount.name={service_name}-sa",
+    ]
+
+    tools.retry(lambda: subprocess.run(minio_args, check=True), max_attempts=10, delay=60)
+
+    minio_pod = tools.kubectl_bin(
+        "get", "pods", f"--selector=release={service_name}",
+        "-o", "jsonpath={.items[].metadata.name}"
+    ).strip()
+    tools.wait_pod(minio_pod)
+
+    operator_ns = os.environ.get("OPERATOR_NS")
+    if operator_ns:
+        namespace = tools.kubectl_bin(
+            "config", "view", "--minify", "-o", "jsonpath={..namespace}"
+        ).strip()
+        tools.kubectl_bin(
+            "create", "svc", "-n", operator_ns, "externalname", service_name,
+            f"--external-name={service_name}.{namespace}.svc.cluster.local",
+            "--tcp=9000", check=False
+        )
+
+    logger.info(f"Creating MinIO bucket: {bucket}")
+    tools.kubectl_bin(
+        "run", "-i", "--rm", "aws-cli",
+        "--image=perconalab/awscli", "--restart=Never",
+        "--", "bash", "-c",
+        f"AWS_ACCESS_KEY_ID=some-access-key "
+        f"AWS_SECRET_ACCESS_KEY=some-secret-key "
+        f"AWS_DEFAULT_REGION=us-east-1 "
+        f"/usr/bin/aws --no-verify-ssl --endpoint-url {endpoint} s3 mb s3://{bucket}",
+    )
+
+    yield
+
+    try:
+        subprocess.run(
+            ["helm", "uninstall", service_name, "--wait", "--timeout", "60s"],
+            check=True,
+        )
+    except subprocess.CalledProcessError as e:
+        logger.warning(f"Failed to cleanup minio: {e}")
+
+
+@pytest.fixture(scope="class")
+def psmdb_client(test_paths: Dict[str, Path]) -> tools.MongoManager:
     """Deploy and get the client pod name."""
     tools.kubectl_bin("apply", "-f", f"{test_paths['conf_dir']}/client-70.yml")
 

e2e-tests/finalizer/test_finalizer.py

@@ -1,14 +1,15 @@
+from pathlib import Path
 import pytest
 import logging
 
 from lib import tools
-from typing import Dict, Union
+from typing import Callable, Dict, Union
 
 logger = logging.getLogger(__name__)
 
 
 @pytest.fixture(scope="class", autouse=True)
-def config(create_infra) -> Dict[str, Union[int, str]]:
+def config(create_infra: Callable[[str], str]) -> Dict[str, Union[int, str]]:
     """Configuration for tests"""
     return {
         "namespace": create_infra("finalizer"),
@@ -17,7 +18,7 @@ def config(create_infra) -> Dict[str, Union[int, str]]:
 
 
 @pytest.fixture(scope="class", autouse=True)
-def setup_tests(test_paths):
+def setup_tests(test_paths: Dict[str, Path]) -> None:
     """Setup test environment"""
     tools.kubectl_bin("apply", "-f", f"{test_paths['conf_dir']}/secrets_with_tls.yml")
 
@@ -26,13 +27,17 @@ class TestFinalizer:
     """Test MongoDB cluster finalizers"""
 
     @pytest.mark.dependency()
-    def test_create_cluster(self, config, test_paths):
+    def test_create_cluster(
+        self, config: Dict[str, Union[int, str]], test_paths: Dict[str, Path]
+    ) -> None:
         tools.apply_cluster(f"{test_paths['test_dir']}/conf/{config['cluster']}.yml")
         tools.wait_for_running(f"{config['cluster']}-rs0", 3, False)
         tools.wait_for_running(f"{config['cluster']}-cfg", 3)
 
     @pytest.mark.dependency(depends=["TestFinalizer::test_create_cluster"])
-    def test_kill_primary_should_elect_new_one(self, config, psmdb_client):
+    def test_kill_primary_should_elect_new_one(
+        self, config: Dict[str, Union[int, str]], psmdb_client: tools.MongoManager
+    ) -> None:
         primary = psmdb_client.get_mongo_primary(
             f"clusterAdmin:clusterAdmin123456@{config['cluster']}-rs0.{config['namespace']}",
             config["cluster"],
@@ -47,7 +52,7 @@ def test_kill_primary_should_elect_new_one(self, config, psmdb_client):
         assert new_primary != primary, "Primary did not change after killing the pod"
 
     @pytest.mark.dependency(depends=["TestFinalizer::test_kill_primary_should_elect_new_one"])
-    def test_delete_cluster(self, config):
+    def test_delete_cluster(self, config: Dict[str, Union[int, str]]) -> None:
         tools.kubectl_bin("delete", "psmdb", config["cluster"], "--wait=false")
         tools.wait_for_delete(f"psmdb/{config['cluster']}")
 

e2e-tests/init-deploy/test_init_deploy.py

@@ -4,13 +4,13 @@
 import logging
 
 from lib import tools
-from typing import Dict, Union
+from typing import Callable, Dict, Union
 
 logger = logging.getLogger(__name__)
 
 
 @pytest.fixture(scope="class", autouse=True)
-def config(create_infra) -> Dict[str, Union[int, str]]:
+def config(create_infra: Callable[[str], str]) -> Dict[str, Union[int, str]]:
     """Configuration for tests"""
     return {
         "namespace": create_infra("init-deploy"),

e2e-tests/lib/k8s_collector.py

@@ -22,9 +22,7 @@ def __init__(self, namespace: str, custom_resources: Optional[List[str]] = None)
 
     def kubectl(self, *args: str) -> str:
         """Run kubectl command and return stdout"""
-        result = subprocess.run(
-            ["kubectl", *args], capture_output=True, text=True, check=False
-        )
+        result = subprocess.run(["kubectl", *args], capture_output=True, text=True, check=False)
         return result.stdout if result.returncode == 0 else ""
 
     def kubectl_ns(self, *args: str) -> str:
@@ -47,12 +45,18 @@ def process_resource(self, resource_type: str) -> None:
         print(f"Processing {resource_type}...")
         base = f"{self.output_dir}/get/{resource_type}"
 
-        self.save(f"{base}/{resource_type}.txt", self.kubectl_ns("get", resource_type, "-o", "wide"))
+        self.save(
+            f"{base}/{resource_type}.txt", self.kubectl_ns("get", resource_type, "-o", "wide")
+        )
 
         for name in self.get_names(resource_type):
-            self.save(f"{self.output_dir}/describe/{resource_type}_{name}.txt",
-                      self.kubectl_ns("describe", resource_type, name))
-            self.save(f"{base}/{name}.yaml", self.kubectl_ns("get", resource_type, name, "-o", "yaml"))
+            self.save(
+                f"{self.output_dir}/describe/{resource_type}_{name}.txt",
+                self.kubectl_ns("describe", resource_type, name),
+            )
+            self.save(
+                f"{base}/{name}.yaml", self.kubectl_ns("get", resource_type, name, "-o", "yaml")
+            )
 
     def extract_pod_logs(self, pod: str) -> None:
         """Extract logs for all containers in a pod"""
@@ -74,8 +78,10 @@ def process_pods(self) -> None:
 
         pods = self.get_names("pods")
         for pod in pods:
-            self.save(f"{self.output_dir}/describe/pod_{pod}.txt",
-                      self.kubectl_ns("describe", "pod", pod))
+            self.save(
+                f"{self.output_dir}/describe/pod_{pod}.txt",
+                self.kubectl_ns("describe", "pod", pod),
+            )
             self.save(f"{base}/{pod}.yaml", self.kubectl_ns("get", "pod", pod, "-o", "yaml"))
 
         with ThreadPoolExecutor(max_workers=5) as executor:
@@ -84,10 +90,12 @@ def process_pods(self) -> None:
     def extract_events(self) -> None:
         """Extract namespace events"""
         print("Extracting events...")
-        self.save(f"{self.output_dir}/events/events.txt",
-                  self.kubectl_ns("get", "events", "-o", "wide"))
-        self.save(f"{self.output_dir}/events/events.json",
-                  self.kubectl_ns("get", "events", "-o", "json"))
+        self.save(
+            f"{self.output_dir}/events/events.txt", self.kubectl_ns("get", "events", "-o", "wide")
+        )
+        self.save(
+            f"{self.output_dir}/events/events.json", self.kubectl_ns("get", "events", "-o", "json")
+        )
 
     def extract_errors(self) -> None:
         """Extract error lines from all logs into summary"""
@@ -102,15 +110,17 @@ def extract_errors(self) -> None:
                     continue
                 path = os.path.join(root, file)
                 with open(path) as f:
-                    error_lines = [l for l in f if "error" in l.lower()]
+                    error_lines = [line for line in f if "error" in line.lower()]
                 if error_lines:
                     rel_path = os.path.relpath(path, logs_dir)
                     errors.append(f"=== {rel_path} ===\n" + "".join(error_lines))
 
         if errors:
-            self.save(f"{self.output_dir}/error_summary.log",
-                      f"Errors for {self.namespace} ({self.timestamp})\n{'='*50}\n\n" +
-                      "\n\n".join(errors))
+            self.save(
+                f"{self.output_dir}/error_summary.log",
+                f"Errors for {self.namespace} ({self.timestamp})\n{'=' * 50}\n\n"
+                + "\n\n".join(errors),
+            )
 
     def collect_all(self) -> None:
         """Main collection method"""
@@ -159,4 +169,3 @@ def get_namespace(test_log: str) -> str:
         print(f"Extracted namespace: {match.group(1)}", file=sys.stderr)
         return match.group(1)
     raise ValueError("Namespace not found in logs")
-

e2e-tests/lib/tools.py

@@ -463,7 +463,9 @@ def get_cr_version() -> str:
     """Get CR version from cr.yaml"""
     try:
         with open(
-            os.path.realpath(os.path.join(os.path.dirname(__file__), "..", "..", "deploy", "cr.yaml"))
+            os.path.realpath(
+                os.path.join(os.path.dirname(__file__), "..", "..", "deploy", "cr.yaml")
+            )
         ) as f:
             return next(line.split()[1] for line in f if "crVersion" in line)
     except (StopIteration, Exception) as e:
@@ -663,10 +665,10 @@ def get_expected_file(test_dir: str | Path, user: str) -> Dict[str, Any]:
             else:
                 raise FileNotFoundError(f"Expected file not found: {base_file}")
 
-        def clean_mongo_json(data: Dict[str, Any]):
+        def clean_mongo_json(data: Dict[str, Any]) -> Dict[str, Any]:
             """Remove timestamps and metadata from MongoDB response"""
 
-            def remove_timestamps(obj):
+            def remove_timestamps(obj: Dict[str, Any]) -> Dict[str, Any]:
                 if isinstance(obj, dict):
                     return {
                         k: remove_timestamps(v)

e2e-tests/liveness/test_liveness.py

@@ -1,17 +1,18 @@
 #!/usr/bin/env python3
 
+from pathlib import Path
 import pytest
 import logging
 import re
 
 from lib import tools
-from typing import Dict
+from typing import Callable, Dict
 
 logger = logging.getLogger(__name__)
 
 
 @pytest.fixture(scope="class", autouse=True)
-def config(create_infra) -> Dict[str, str]:
+def config(create_infra: Callable[[str], str]) -> Dict[str, str]:
     """Configuration for tests"""
     return {
         "namespace": create_infra("liveness"),
@@ -20,9 +21,15 @@ def config(create_infra) -> Dict[str, str]:
 
 
 @pytest.fixture(scope="class", autouse=True)
-def setup_tests(test_paths):
+def setup_tests(test_paths: Dict[str, Path], deploy_minio) -> None:
     """Setup test environment"""
-    tools.kubectl_bin("apply", "-f", f"{test_paths['conf_dir']}/secrets_with_tls.yml")
+    tools.kubectl_bin(
+        "apply",
+        "-f",
+        f"{test_paths['conf_dir']}/secrets.yml",
+        "-f",
+        f"{test_paths['conf_dir']}/minio-secret.yml",
+    )
 
 
 class TestLiveness:
@@ -37,7 +44,7 @@ def test_create_first_cluster(self, config, test_paths):
         )
 
     @pytest.mark.dependency(depends=["TestLiveness::test_create_first_cluster"])
-    def test_liveness_check_fails_with_invalid_ssl_option(self, config):
+    def test_liveness_check_fails_with_invalid_ssl_option(self, config: Dict[str, str]) -> None:
         tools.kubectl_bin(
             "exec",
             f"{config['cluster']}-rs0-0",
@@ -69,7 +76,9 @@ def test_liveness_check_fails_with_invalid_ssl_option(self, config):
     @pytest.mark.dependency(
         depends=["TestLiveness::test_liveness_check_fails_with_invalid_ssl_option"]
     )
-    def test_change_liveness_config(self, config, test_paths):
+    def test_change_liveness_config(
+        self, config: Dict[str, str], test_paths: Dict[str, Path]
+    ) -> None:
         tools.apply_cluster(f"{test_paths['test_dir']}/conf/{config['cluster']}-rs0-changed.yml")
 
         tools.wait_for_running(f"{config['cluster']}-rs0", 3)