|
| 1 | +import pytest |
| 2 | +from kubernetes.dynamic import DynamicClient |
| 3 | +from ocp_resources.network_policy import NetworkPolicy |
| 4 | +from simple_logger.logger import get_logger |
| 5 | +from timeout_sampler import TimeoutSampler |
| 6 | + |
| 7 | +from tests.model_registry.model_catalog.utils import get_postgres_pod_in_namespace |
| 8 | +from tests.model_registry.utils import ( |
| 9 | + wait_for_model_catalog_pod_ready_after_deletion, |
| 10 | +) |
| 11 | + |
| 12 | +LOGGER = get_logger(name=__name__) |
| 13 | + |
| 14 | + |
| 15 | +class TestModelCatalogDBSecret: |
| 16 | + def test_model_catalog_postgres_secret_exists(self, model_catalog_postgres_secret_values): |
| 17 | + """Test that model-catalog-postgres secret exists and is accessible""" |
| 18 | + assert model_catalog_postgres_secret_values, ( |
| 19 | + f"model-catalog-postgres secret should exist and be accessible: {model_catalog_postgres_secret_values}" |
| 20 | + ) |
| 21 | + |
| 22 | + @pytest.mark.dependency(name="test_model_catalog_postgres_password_recreation") |
| 23 | + def test_model_catalog_postgres_password_recreation( |
| 24 | + self, model_catalog_postgres_secret_values, recreated_model_catalog_postgres_secret |
| 25 | + ): |
| 26 | + """Test that secret recreation generates new password but preserves user/database name""" |
| 27 | + # Verify database-name and database-user did NOT change |
| 28 | + unchanged_keys = ["database-name", "database-user"] |
| 29 | + for key in unchanged_keys: |
| 30 | + assert model_catalog_postgres_secret_values[key] == recreated_model_catalog_postgres_secret[key], ( |
| 31 | + f"{key} should remain the same after secret recreation" |
| 32 | + ) |
| 33 | + |
| 34 | + # Verify database-password DID change (randomization working) |
| 35 | + assert ( |
| 36 | + model_catalog_postgres_secret_values["database-password"] |
| 37 | + != recreated_model_catalog_postgres_secret["database-password"] |
| 38 | + ), "database-password should be different after secret recreation (randomized)" |
| 39 | + |
| 40 | + LOGGER.info("Password randomization verified - new password generated on recreation") |
| 41 | + |
| 42 | + @pytest.mark.dependency(depends=["test_model_catalog_postgres_password_recreation"]) |
| 43 | + def test_model_catalog_pod_ready_after_secret_recreation( |
| 44 | + self, admin_client: DynamicClient, model_registry_namespace: str |
| 45 | + ): |
| 46 | + """Test that model catalog pod becomes ready after secret recreation""" |
| 47 | + # delete the postgres pod first |
| 48 | + get_postgres_pod_in_namespace(admin_client=admin_client, namespace=model_registry_namespace).delete() |
| 49 | + # Wait for model catalog pod to be ready after the secret deletion/recreation |
| 50 | + wait_for_model_catalog_pod_ready_after_deletion( |
| 51 | + client=admin_client, model_registry_namespace=model_registry_namespace |
| 52 | + ) |
| 53 | + LOGGER.info("Model catalog pod is ready after secret recreation") |
| 54 | + |
| 55 | + |
| 56 | +class TestModelCatalogDBNetworkPolicy: |
| 57 | + def test_postgres_network_policy_exists(self, model_catalog_postgres_network_policy): |
| 58 | + """Test that postgres NetworkPolicy exists and is accessible""" |
| 59 | + assert model_catalog_postgres_network_policy.exists, "model-catalog-postgres NetworkPolicy should exist" |
| 60 | + |
| 61 | + def test_postgres_network_policy_restricts_to_port_5432(self, model_catalog_postgres_network_policy): |
| 62 | + """Test that NetworkPolicy only allows TCP 5432 ingress""" |
| 63 | + spec = model_catalog_postgres_network_policy.instance.spec |
| 64 | + assert "Ingress" in spec.policyTypes, "NetworkPolicy should have Ingress policy type" |
| 65 | + assert len(spec.ingress) == 1, "NetworkPolicy should have exactly one ingress rule" |
| 66 | + |
| 67 | + port = spec.ingress[0].ports[0] |
| 68 | + assert port.port == 5432, "NetworkPolicy should allow only PostgreSQL port 5432" |
| 69 | + assert port.protocol == "TCP", "NetworkPolicy port should use TCP protocol" |
| 70 | + |
| 71 | + def test_postgres_network_policy_allows_only_catalog_pods(self, model_catalog_postgres_network_policy): |
| 72 | + """Test that only model-catalog pods can reach postgres""" |
| 73 | + from_selector = model_catalog_postgres_network_policy.instance.spec.ingress[0]["from"][ |
| 74 | + 0 |
| 75 | + ].podSelector.matchLabels |
| 76 | + assert from_selector["component"] == "model-catalog", ( |
| 77 | + "Only model-catalog pods should be allowed to access postgres" |
| 78 | + ) |
| 79 | + |
| 80 | + @pytest.mark.dependency(name="test_postgres_network_policy_recreation") |
| 81 | + def test_postgres_network_policy_recreated_after_deletion( |
| 82 | + self, |
| 83 | + admin_client: DynamicClient, |
| 84 | + model_catalog_postgres_network_policy, |
| 85 | + model_registry_namespace: str, |
| 86 | + ): |
| 87 | + """Test that operator recreates NetworkPolicy after deletion""" |
| 88 | + model_catalog_postgres_network_policy.delete() |
| 89 | + get_postgres_pod_in_namespace(admin_client=admin_client, namespace=model_registry_namespace).delete() |
| 90 | + for np in TimeoutSampler( |
| 91 | + wait_timeout=120, |
| 92 | + sleep=10, |
| 93 | + func=NetworkPolicy, |
| 94 | + client=admin_client, |
| 95 | + name="model-catalog-postgres", |
| 96 | + namespace=model_registry_namespace, |
| 97 | + ): |
| 98 | + if np.exists: |
| 99 | + LOGGER.info("NetworkPolicy has been recreated by operator") |
| 100 | + break |
0 commit comments