fix: refactor mysql ssl deployment

Author: fege

tests/model_registry/rest_api/conftest.py

@@ -2,7 +2,8 @@
 import os
 from kubernetes.dynamic import DynamicClient
 import pytest
-from tests.model_registry.rest_api.constants import MODEL_REGISTRY_BASE_URI
+import copy
+from tests.model_registry.rest_api.constants import MODEL_REGISTRY_BASE_URI, MODEL_REGISTER_DATA
 from tests.model_registry.rest_api.utils import (
     register_model_rest_api,
     execute_model_registry_patch_command,
@@ -11,7 +12,11 @@
 from ocp_resources.deployment import Deployment
 from tests.model_registry.utils import (
     get_model_registry_deployment_template_dict,
+    apply_mysql_args_and_volume_mounts,
+    add_mysql_certs_volumes_to_deployment,
+    generate_random_name,
 )
+
 from tests.model_registry.constants import (
     DB_RESOURCES_NAME,
     CA_MOUNT_PATH,
@@ -250,51 +255,20 @@ def patch_mysql_deployment_with_ssl_ca(
     """
     CA_CONFIGMAP_NAME = request.param.get("ca_configmap_name", "mysql-ca-configmap")
     CA_MOUNT_PATH = request.param.get("ca_mount_path", "/etc/mysql/ssl")
-    deployment = Deployment(
-        client=admin_client,
-        name=model_registry_db_deployment.name,
-        namespace=model_registry_namespace,
-    )
-    deployment.wait_for_condition(condition="Available", status="True")
-    original_deployment = deployment.instance.to_dict()
-    spec = original_deployment["spec"]["template"]["spec"]
+    deployment = model_registry_db_deployment.instance.to_dict()
+    spec = deployment["spec"]["template"]["spec"]
     my_sql_container = next(container for container in spec["containers"] if container["name"] == "mysql")
     assert my_sql_container is not None, "Mysql container not found"
-    mysql_args = list(my_sql_container.get("args", []))
-    mysql_args.extend([
-        f"--ssl-ca={CA_MOUNT_PATH}/ca/ca-bundle.crt",
-        f"--ssl-cert={CA_MOUNT_PATH}/server_cert/tls.crt",
-        f"--ssl-key={CA_MOUNT_PATH}/server_key/tls.key",
-    ])
-
-    volumes_mounts = list(my_sql_container.get("volumeMounts", []))
-    volumes_mounts.extend([
-        {"name": CA_CONFIGMAP_NAME, "mountPath": f"{CA_MOUNT_PATH}/ca", "readOnly": True},
-        {
-            "name": "mysql-server-cert",
-            "mountPath": f"{CA_MOUNT_PATH}/server_cert",
-            "readOnly": True,
-        },
-        {
-            "name": "mysql-server-key",
-            "mountPath": f"{CA_MOUNT_PATH}/server_key",
-            "readOnly": True,
-        },
-    ])
 
-    my_sql_container["args"] = mysql_args
-    my_sql_container["volumeMounts"] = volumes_mounts
-    volumes = list(spec["volumes"])
-    volumes.extend([
-        {"name": CA_CONFIGMAP_NAME, "configMap": {"name": CA_CONFIGMAP_NAME}},
-        {"name": "mysql-server-cert", "secret": {"secretName": "mysql-server-cert"}},  # pragma: allowlist secret
-        {"name": "mysql-server-key", "secret": {"secretName": "mysql-server-key"}},  # pragma: allowlist secret
-    ])
+    my_sql_container = apply_mysql_args_and_volume_mounts(
+        my_sql_container=my_sql_container, ca_configmap_name=CA_CONFIGMAP_NAME, ca_mount_path=CA_MOUNT_PATH
+    )
+    volumes = add_mysql_certs_volumes_to_deployment(spec=spec, ca_configmap_name=CA_CONFIGMAP_NAME)
 
     patch = {"spec": {"template": {"spec": {"volumes": volumes, "containers": [my_sql_container]}}}}
-    with ResourceEditor(patches={deployment: patch}):
-        deployment.wait_for_condition(condition="Available", status="True")
-        yield deployment
+    with ResourceEditor(patches={model_registry_db_deployment: patch}):
+        model_registry_db_deployment.wait_for_condition(condition="Available", status="True")
+        yield model_registry_db_deployment
 
 
 @pytest.fixture(scope="class")
@@ -359,3 +333,17 @@ def mysql_ssl_secrets(
         "server_cert_secret": server_cert_secret,
         "server_key_secret": server_key_secret,
     }
+
+
+@pytest.fixture(scope="function")
+def model_data_for_test() -> Generator[dict[str, Any], None, None]:
+    """
+    Generates a model data for the test.
+
+    Returns:
+        dict[str, Any]: The model data for the test
+    """
+    model_name = generate_random_name(prefix="model-rest-api")
+    model_data = copy.deepcopy(MODEL_REGISTER_DATA)
+    model_data["register_model_data"]["name"] = model_name
+    yield model_data

tests/model_registry/rest_api/test_model_registry_secure_db.py

@@ -1,11 +1,8 @@
 import pytest
 import requests
-from typing import Self
-import copy
+from typing import Self, Any
 from pytest_testconfig import config as py_config
 from tests.model_registry.rest_api.utils import register_model_rest_api, validate_resource_attributes
-from tests.model_registry.rest_api.constants import MODEL_REGISTER_DATA
-from tests.model_registry.utils import generate_random_name
 from tests.model_registry.constants import CA_MOUNT_PATH
 from tests.model_registry.utils import get_mr_service_by_label, get_endpoint_from_mr_service
 from kubernetes.dynamic import DynamicClient
@@ -62,6 +59,7 @@ def test_register_model_with_invalid_ca(
         model_registry_rest_headers: dict[str, str],
         local_ca_bundle: str,
         deploy_secure_mysql_and_mr: ModelRegistry,
+        model_data_for_test: dict[str, Any],
     ):
         """
         Test that model registration fails with an SSLError when the Model Registry is deployed
@@ -72,15 +70,11 @@ def test_register_model_with_invalid_ca(
         )
         model_registry_rest_url = get_endpoint_from_mr_service(svc=service, protocol=Protocols.REST)
 
-        model_name = generate_random_name(prefix="model-rest-api")
-        model_data = copy.deepcopy(MODEL_REGISTER_DATA)
-        model_data["register_model_data"]["name"] = model_name
-
         with pytest.raises(requests.exceptions.SSLError) as exc_info:
             register_model_rest_api(
                 model_registry_rest_url=f"https://{model_registry_rest_url}",
                 model_registry_rest_headers=model_registry_rest_headers,
-                data_dict=model_data,
+                data_dict=model_data_for_test,
                 verify=local_ca_bundle,
             )
         assert "SSLError" in str(exc_info.value), (
@@ -109,24 +103,22 @@ def test_register_model_with_valid_ca(
         model_registry_rest_headers: dict[str, str],
         local_ca_bundle: str,
         deploy_secure_mysql_and_mr: ModelRegistry,
+        model_data_for_test: dict[str, Any],
     ):
         service = get_mr_service_by_label(
             client=admin_client, namespace_name=model_registry_namespace, mr_instance=deploy_secure_mysql_and_mr
         )
         model_registry_rest_url = get_endpoint_from_mr_service(svc=service, protocol=Protocols.REST)
-        model_name = generate_random_name(prefix="model-rest-api")
-        model_data = copy.deepcopy(MODEL_REGISTER_DATA)
-        model_data["register_model_data"]["name"] = model_name
 
         result = register_model_rest_api(
             model_registry_rest_url=f"https://{model_registry_rest_url}",
             model_registry_rest_headers=model_registry_rest_headers,
-            data_dict=model_data,
+            data_dict=model_data_for_test,
             verify=local_ca_bundle,
         )
         assert result["register_model"].get("id"), "Model registration failed with secure DB connection."
         validate_resource_attributes(
-            expected_params=model_data["register_model_data"],
+            expected_params=model_data_for_test["register_model_data"],
             actual_resource_data=result["register_model"],
             resource_name="register_model",
         )

tests/model_registry/utils.py

@@ -261,3 +261,72 @@ def generate_random_name(prefix: str = "", length: int = 8) -> str:
 
 def generate_namespace_name(file_path: str) -> str:
     return (file_path.removesuffix(".py").replace("/", "-").replace("_", "-"))[-63:].split("-", 1)[-1]
+
+
+def add_mysql_certs_volumes_to_deployment(
+    spec: dict[str, Any],
+    ca_configmap_name: str,
+) -> list[dict[str, Any]]:
+    """
+    Adds the MySQL certs volumes to the deployment.
+
+    Args:
+        spec: The spec of the deployment
+        ca_configmap_name: The name of the CA configmap
+
+    Returns:
+        The volumes with the MySQL certs volumes added
+    """
+
+    volumes = list(spec["volumes"])
+    volumes.extend([
+        {"name": ca_configmap_name, "configMap": {"name": ca_configmap_name}},
+        {"name": "mysql-server-cert", "secret": {"secretName": "mysql-server-cert"}},  # pragma: allowlist secret
+        {"name": "mysql-server-key", "secret": {"secretName": "mysql-server-key"}},  # pragma: allowlist secret
+    ])
+
+    return volumes
+
+
+def apply_mysql_args_and_volume_mounts(
+    my_sql_container: dict[str, Any],
+    ca_configmap_name: str,
+    ca_mount_path: str,
+) -> dict[str, Any]:
+    """
+    Applies the MySQL args and volume mounts to the MySQL container.
+
+    Args:
+        my_sql_container: The MySQL container
+        ca_configmap_name: The name of the CA configmap
+        ca_mount_path: The mount path of the CA
+
+    Returns:
+        The MySQL container with the MySQL args and volume mounts applied
+    """
+
+    mysql_args = list(my_sql_container.get("args", []))
+    mysql_args.extend([
+        f"--ssl-ca={ca_mount_path}/ca/ca-bundle.crt",
+        f"--ssl-cert={ca_mount_path}/server_cert/tls.crt",
+        f"--ssl-key={ca_mount_path}/server_key/tls.key",
+    ])
+
+    volumes_mounts = list(my_sql_container.get("volumeMounts", []))
+    volumes_mounts.extend([
+        {"name": ca_configmap_name, "mountPath": f"{ca_mount_path}/ca", "readOnly": True},
+        {
+            "name": "mysql-server-cert",
+            "mountPath": f"{ca_mount_path}/server_cert",
+            "readOnly": True,
+        },
+        {
+            "name": "mysql-server-key",
+            "mountPath": f"{ca_mount_path}/server_key",
+            "readOnly": True,
+        },
+    ])
+
+    my_sql_container["args"] = mysql_args
+    my_sql_container["volumeMounts"] = volumes_mounts
+    return my_sql_container

utilities/certificates_utils.py

@@ -149,7 +149,7 @@ def create_ca_bundle_with_router_cert(
     Creates a CA bundle file by fetching the CA bundle from a ConfigMap and appending the router CA from a Secret.
 
     Args:
-        client: The admin client to get the CA bundle from a ConfigMap and append the router CA from a Secret.
+        client: The client to get the CA bundle from a ConfigMap and append the router CA from a Secret.
         namespace: The namespace of the ConfigMap and Secret.
         ca_bundle_path: The path to the CA bundle file.
         cert_name: The name of the certificate in the ConfigMap.