test: New tests to validate postgres password autogeneration

Author: dbasunag

tests/model_registry/model_catalog/db_check/__init__.py

@@ -0,0 +1 @@
+# Database check tests for model catalog

tests/model_registry/model_catalog/db_check/conftest.py

@@ -0,0 +1,58 @@
+import pytest
+from kubernetes.dynamic import DynamicClient
+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) -> Secret:
+    """Get the model-catalog-postgres secret from model registry namespace"""
+    return Secret(
+        client=admin_client,
+        name="model-catalog-postgres",
+        namespace=py_config["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)

tests/model_registry/model_catalog/db_check/test_model_catalog_secrets.py

@@ -0,0 +1,52 @@
+import pytest
+from kubernetes.dynamic import DynamicClient
+from simple_logger.logger import get_logger
+
+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__)
+
+
+def test_model_catalog_postgres_secret_exists(model_catalog_postgres_secret_values):
+    """Test that model-catalog-postgres secret exists and is accessible"""
+    secret_values = model_catalog_postgres_secret_values
+    assert secret_values, "model-catalog-postgres secret should exist and be accessible"
+
+    # Log the keys (not values) for debugging
+    LOGGER.info(f"Secret contains keys: {list(secret_values.keys())}")
+
+
+@pytest.mark.dependency(name="test_model_catalog_postgres_password_recreation")
+def test_model_catalog_postgres_password_recreation(
+    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(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).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")

tests/model_registry/model_catalog/db_check/utils.py

@@ -0,0 +1,30 @@
+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
+
+    LOGGER.info(f"Captured secret with keys: {list(secret_values.keys())}")
+    return secret_values