diff --git a/tests/model_registry/model_catalog/db_check/conftest.py b/tests/model_registry/model_catalog/db_check/conftest.py index 736990381..fdb5c764c 100644 --- a/tests/model_registry/model_catalog/db_check/conftest.py +++ b/tests/model_registry/model_catalog/db_check/conftest.py @@ -1,5 +1,6 @@ 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 @@ -56,3 +57,14 @@ def recreated_model_catalog_postgres_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, + ) diff --git a/tests/model_registry/model_catalog/db_check/test_model_catalog_db_validation.py b/tests/model_registry/model_catalog/db_check/test_model_catalog_db_validation.py new file mode 100644 index 000000000..7a968a589 --- /dev/null +++ b/tests/model_registry/model_catalog/db_check/test_model_catalog_db_validation.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 diff --git a/tests/model_registry/model_catalog/db_check/test_model_catalog_secrets.py b/tests/model_registry/model_catalog/db_check/test_model_catalog_secrets.py deleted file mode 100644 index 088c39bf3..000000000 --- a/tests/model_registry/model_catalog/db_check/test_model_catalog_secrets.py +++ /dev/null @@ -1,50 +0,0 @@ -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""" - 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( - 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")