test: [3.3] New tests to validate postgres password autogeneration

Signed-off-by: Debarati Basu-Nag <[email protected]>

Author: dbasunag

tests/model_registry/model_catalog/db_check/__init__.py

@@ -0,0 +1,70 @@
+import pytest
+from kubernetes.dynamic import DynamicClient
+from ocp_resources.network_policy import NetworkPolicy
+from ocp_resources.secret import Secret
+from pytest_testconfig import config as py_config
+from simple_logger.logger import get_logger
+from timeout_sampler import TimeoutSampler
+
+from .utils import extract_secret_values
+
+LOGGER = get_logger(name=__name__)
+
+
+@pytest.fixture(scope="class")
+def model_catalog_postgres_secret(admin_client: DynamicClient, model_registry_namespace: str) -> Secret:
+    """Get the model-catalog-postgres secret from model registry namespace"""
+    return Secret(
+        client=admin_client,
+        name="model-catalog-postgres",
+        namespace=model_registry_namespace,
+        ensure_exists=True,
+    )
+
+
+@pytest.fixture(scope="class")
+def model_catalog_postgres_secret_values(model_catalog_postgres_secret: Secret) -> dict[str, str]:
+    """Capture current values of model-catalog-postgres secret in model registry namespace"""
+    return extract_secret_values(secret=model_catalog_postgres_secret)
+
+
+@pytest.fixture(scope="class")
+def recreated_model_catalog_postgres_secret(
+    admin_client: DynamicClient, model_catalog_postgres_secret: Secret
+) -> dict[str, str]:
+    """Delete model-catalog-postgres secret and wait for it to be recreated"""
+    model_registry_namespace = py_config["model_registry_namespace"]
+    resource_name = "model-catalog-postgres"
+
+    LOGGER.info(f"Deleting secret {resource_name} in namespace {model_registry_namespace}")
+    model_catalog_postgres_secret.delete()
+
+    # Wait for the secret to be recreated by the operator
+    LOGGER.info(f"Waiting for secret {resource_name} to be recreated...")
+
+    recreated_secret = None
+    for secret in TimeoutSampler(
+        wait_timeout=120,
+        sleep=10,
+        func=Secret,
+        client=admin_client,
+        name=resource_name,
+        namespace=model_registry_namespace,
+    ):
+        if secret.exists:
+            LOGGER.info(f"Secret {resource_name} has been recreated")
+            recreated_secret = secret
+            break
+
+    return extract_secret_values(secret=recreated_secret)
+
+
+@pytest.fixture(scope="class")
+def model_catalog_postgres_network_policy(admin_client: DynamicClient, model_registry_namespace: str) -> NetworkPolicy:
+    """Get the model-catalog-postgres NetworkPolicy from model registry namespace"""
+    return NetworkPolicy(
+        client=admin_client,
+        name="model-catalog-postgres",
+        namespace=model_registry_namespace,
+        ensure_exists=True,
+    )

tests/model_registry/model_catalog/db_check/conftest.py

@@ -0,0 +1,100 @@
+import pytest
+from kubernetes.dynamic import DynamicClient
+from ocp_resources.network_policy import NetworkPolicy
+from simple_logger.logger import get_logger
+from timeout_sampler import TimeoutSampler
+
+from tests.model_registry.model_catalog.utils import get_postgres_pod_in_namespace
+from tests.model_registry.utils import (
+    wait_for_model_catalog_pod_ready_after_deletion,
+)
+
+LOGGER = get_logger(name=__name__)
+
+
+class TestModelCatalogDBSecret:
+    def test_model_catalog_postgres_secret_exists(self, model_catalog_postgres_secret_values):
+        """Test that model-catalog-postgres secret exists and is accessible"""
+        assert model_catalog_postgres_secret_values, (
+            f"model-catalog-postgres secret should exist and be accessible: {model_catalog_postgres_secret_values}"
+        )
+
+    @pytest.mark.dependency(name="test_model_catalog_postgres_password_recreation")
+    def test_model_catalog_postgres_password_recreation(
+        self, model_catalog_postgres_secret_values, recreated_model_catalog_postgres_secret
+    ):
+        """Test that secret recreation generates new password but preserves user/database name"""
+        # Verify database-name and database-user did NOT change
+        unchanged_keys = ["database-name", "database-user"]
+        for key in unchanged_keys:
+            assert model_catalog_postgres_secret_values[key] == recreated_model_catalog_postgres_secret[key], (
+                f"{key} should remain the same after secret recreation"
+            )
+
+        # Verify database-password DID change (randomization working)
+        assert (
+            model_catalog_postgres_secret_values["database-password"]
+            != recreated_model_catalog_postgres_secret["database-password"]
+        ), "database-password should be different after secret recreation (randomized)"
+
+        LOGGER.info("Password randomization verified - new password generated on recreation")
+
+    @pytest.mark.dependency(depends=["test_model_catalog_postgres_password_recreation"])
+    def test_model_catalog_pod_ready_after_secret_recreation(
+        self, admin_client: DynamicClient, model_registry_namespace: str
+    ):
+        """Test that model catalog pod becomes ready after secret recreation"""
+        # delete the postgres pod first
+        get_postgres_pod_in_namespace(admin_client=admin_client, namespace=model_registry_namespace).delete()
+        # Wait for model catalog pod to be ready after the secret deletion/recreation
+        wait_for_model_catalog_pod_ready_after_deletion(
+            client=admin_client, model_registry_namespace=model_registry_namespace
+        )
+        LOGGER.info("Model catalog pod is ready after secret recreation")
+
+
+class TestModelCatalogDBNetworkPolicy:
+    def test_postgres_network_policy_exists(self, model_catalog_postgres_network_policy):
+        """Test that postgres NetworkPolicy exists and is accessible"""
+        assert model_catalog_postgres_network_policy.exists, "model-catalog-postgres NetworkPolicy should exist"
+
+    def test_postgres_network_policy_restricts_to_port_5432(self, model_catalog_postgres_network_policy):
+        """Test that NetworkPolicy only allows TCP 5432 ingress"""
+        spec = model_catalog_postgres_network_policy.instance.spec
+        assert "Ingress" in spec.policyTypes, "NetworkPolicy should have Ingress policy type"
+        assert len(spec.ingress) == 1, "NetworkPolicy should have exactly one ingress rule"
+
+        port = spec.ingress[0].ports[0]
+        assert port.port == 5432, "NetworkPolicy should allow only PostgreSQL port 5432"
+        assert port.protocol == "TCP", "NetworkPolicy port should use TCP protocol"
+
+    def test_postgres_network_policy_allows_only_catalog_pods(self, model_catalog_postgres_network_policy):
+        """Test that only model-catalog pods can reach postgres"""
+        from_selector = model_catalog_postgres_network_policy.instance.spec.ingress[0]["from"][
+            0
+        ].podSelector.matchLabels
+        assert from_selector["component"] == "model-catalog", (
+            "Only model-catalog pods should be allowed to access postgres"
+        )
+
+    @pytest.mark.dependency(name="test_postgres_network_policy_recreation")
+    def test_postgres_network_policy_recreated_after_deletion(
+        self,
+        admin_client: DynamicClient,
+        model_catalog_postgres_network_policy,
+        model_registry_namespace: str,
+    ):
+        """Test that operator recreates NetworkPolicy after deletion"""
+        model_catalog_postgres_network_policy.delete()
+        get_postgres_pod_in_namespace(admin_client=admin_client, namespace=model_registry_namespace).delete()
+        for np in TimeoutSampler(
+            wait_timeout=120,
+            sleep=10,
+            func=NetworkPolicy,
+            client=admin_client,
+            name="model-catalog-postgres",
+            namespace=model_registry_namespace,
+        ):
+            if np.exists:
+                LOGGER.info("NetworkPolicy has been recreated by operator")
+                break

tests/model_registry/model_catalog/db_check/test_model_catalog_db_validation.py

@@ -0,0 +1,29 @@
+import base64
+import binascii
+
+from ocp_resources.secret import Secret
+from simple_logger.logger import get_logger
+
+LOGGER = get_logger(name=__name__)
+
+
+def extract_secret_values(secret: Secret) -> dict[str, str]:
+    """Extract and decode secret data values from a Secret object.
+
+    Args:
+        secret: The Secret object to extract values from
+
+    Returns:
+        Dict mapping secret keys to decoded string values
+    """
+    secret_values = {}
+    if secret.instance.data:
+        for key, encoded_value in secret.instance.data.items():
+            try:
+                decoded_value = base64.b64decode(s=encoded_value).decode(encoding="utf-8")
+                secret_values[key] = decoded_value
+            except (binascii.Error, UnicodeDecodeError) as e:
+                LOGGER.warning(f"Failed to decode secret key '{key}': {e}")
+                secret_values[key] = encoded_value  # Keep encoded if decode fails
+
+    return secret_values

tests/model_registry/model_catalog/db_check/utils.py

@@ -1,5 +1,6 @@
 from typing import Any
 
+from kubernetes.dynamic import DynamicClient
 from simple_logger.logger import get_logger
 
 from ocp_resources.pod import Pod
@@ -9,25 +10,30 @@
 LOGGER = get_logger(name=__name__)
 
 
-def get_postgres_pod_in_namespace(namespace: str = "rhoai-model-registries") -> Pod:
+def get_postgres_pod_in_namespace(admin_client: DynamicClient, namespace: str = "rhoai-model-registries") -> Pod:
     """Get the PostgreSQL pod for model catalog database."""
-    postgres_pods = list(Pod.get(namespace=namespace, label_selector="app.kubernetes.io/name=model-catalog-postgres"))
+    postgres_pods = list(
+        Pod.get(
+            dyn_client=admin_client, namespace=namespace, label_selector="app.kubernetes.io/name=model-catalog-postgres"
+        )
+    )
     assert postgres_pods, f"No PostgreSQL pod found in namespace {namespace}"
     return postgres_pods[0]
 
 
-def execute_database_query(query: str, namespace: str = "rhoai-model-registries") -> str:
+def execute_database_query(query: str, admin_client: DynamicClient, namespace: str = "rhoai-model-registries") -> str:
     """
     Execute a SQL query against the model catalog database.
 
     Args:
         query: SQL query to execute
+        admin_client: DynamicClient for Kubernetes API access
         namespace: OpenShift namespace containing the PostgreSQL pod
 
     Returns:
         Raw database query result as string
     """
-    postgres_pod = get_postgres_pod_in_namespace(namespace=namespace)
+    postgres_pod = get_postgres_pod_in_namespace(admin_client=admin_client, namespace=namespace)
 
     return postgres_pod.execute(
         command=["psql", "-U", "catalog_user", "-d", "model_catalog", "-c", query],