test: add test to initiate mr with secure db connection

conftest.py

@@ -32,11 +32,11 @@
     collect_rhoai_must_gather,
     get_base_dir,
 )
-
 from kubernetes.dynamic import DynamicClient
 from utilities.infra import get_operator_distribution, get_dsci_applications_namespace, get_data_science_cluster
 from ocp_resources.resource import get_client
 
+
 LOGGER = logging.getLogger(name=__name__)
 BASIC_LOGGER = logging.getLogger(name="basic")
 

tests/model_registry/conftest.py

@@ -174,16 +174,36 @@ def model_registry_instance(
 
 @pytest.fixture(scope="class")
 def model_registry_mysql_config(
-    model_registry_db_deployment: Deployment, model_registry_db_secret: Secret
+    request: FixtureRequest,
+    model_registry_db_deployment: Deployment,
+    model_registry_db_secret: Secret,
 ) -> dict[str, Any]:
-    return {
+    """
+    Fixture to build the MySQL config dictionary for Model Registry.
+    Expects request.param to be a dict. If 'sslRootCertificateConfigMap' is not present, it defaults to None.
+    If 'sslRootCertificateConfigMap' is present, it will be used to configure the MySQL connection.
+
+    Args:
+        request: The pytest request object
+        model_registry_db_deployment: The model registry db deployment
+        model_registry_db_secret: The model registry db secret
+
+    Returns:
+        dict[str, Any]: The MySQL config dictionary
+    """
+    param = request.param if hasattr(request, "param") else {}
+    config = {
         "host": f"{model_registry_db_deployment.name}.{model_registry_db_deployment.namespace}.svc.cluster.local",
         "database": model_registry_db_secret.string_data["database-name"],
         "passwordSecret": {"key": "database-password", "name": model_registry_db_deployment.name},
-        "port": 3306,
+        "port": param.get("port", 3306),
         "skipDBCreation": False,
         "username": model_registry_db_secret.string_data["database-user"],
     }
+    if "sslRootCertificateConfigMap" in param:
+        config["sslRootCertificateConfigMap"] = param["sslRootCertificateConfigMap"]
+
+    return config
 
 
 @pytest.fixture(scope="class")

tests/model_registry/constants.py

@@ -26,6 +26,7 @@ class ModelRegistryEndpoints:
     },
 }
 MR_INSTANCE_NAME: str = "model-registry"
+SECURE_MR_NAME: str = "secure-db-mr"
 ISTIO_CONFIG_DICT: dict[str, Any] = {
     "gateway": {"grpc": {"tls": {}}, "rest": {"tls": {}}},
 }
@@ -48,6 +49,11 @@ class ModelRegistryEndpoints:
     f"{Resource.ApiGroup.TEMPLATE_OPENSHIFT_IO}/expose-password": "'{.data[''database-password'']}'",
     f"{Resource.ApiGroup.TEMPLATE_OPENSHIFT_IO}/expose-username": "'{.data[''database-user'']}'",
 }
+
+CA_CONFIGMAP_NAME = "odh-trusted-ca-bundle"
+CA_MOUNT_PATH = "/etc/pki/ca-trust/extracted/pem"
+CA_FILE_PATH = f"{CA_MOUNT_PATH}/ca-bundle.crt"
+
 MODEL_REGISTRY_STANDARD_LABELS = {
     Annotations.KubernetesIo.NAME: MR_INSTANCE_NAME,
     Annotations.KubernetesIo.INSTANCE: MR_INSTANCE_NAME,

tests/model_registry/rest_api/conftest.py

@@ -1,11 +1,38 @@
-from typing import Any
-
+from typing import Any, Generator
+import os
+from kubernetes.dynamic import DynamicClient
 import pytest
-
 from tests.model_registry.rest_api.constants import MODEL_REGISTRY_BASE_URI
-from tests.model_registry.rest_api.utils import register_model_rest_api, execute_model_registry_patch_command
+from tests.model_registry.rest_api.utils import (
+    register_model_rest_api,
+    execute_model_registry_patch_command,
+)
 from utilities.constants import Protocols
+from ocp_resources.deployment import Deployment
+from tests.model_registry.utils import (
+    get_model_registry_deployment_template_dict,
+)
+from tests.model_registry.constants import (
+    DB_RESOURCES_NAME,
+    CA_MOUNT_PATH,
+    CA_FILE_PATH,
+    CA_CONFIGMAP_NAME,
+    OAUTH_PROXY_CONFIG_DICT,
+    MODEL_REGISTRY_STANDARD_LABELS,
+    SECURE_MR_NAME,
+)
+from ocp_resources.resource import ResourceEditor
+from ocp_resources.secret import Secret
+from ocp_resources.config_map import ConfigMap
+from simple_logger.logger import get_logger
+from ocp_resources.model_registry_modelregistry_opendatahub_io import ModelRegistry
+from pytest_testconfig import config as py_config
 from utilities.exceptions import MissingParameter
+import tempfile
+from tests.model_registry.rest_api.utils import generate_ca_and_server_cert
+from utilities.certificates_utils import create_k8s_secret, create_ca_bundle_with_router_cert
+
+LOGGER = get_logger(name=__name__)
 
 
 @pytest.fixture(scope="class")
@@ -63,3 +90,272 @@ def updated_model_registry_resource(
         headers=model_registry_rest_headers,
         data_json=request.param["data"],
     )
+
+
+@pytest.fixture(scope="class")
+def patch_invalid_ca(
+    admin_client: DynamicClient,
+    model_registry_namespace: str,
+    request: pytest.FixtureRequest,
+) -> Generator[str, Any, Any]:
+    """
+    Patches the odh-trusted-ca-bundle ConfigMap with an invalid CA certificate.
+    """
+    ca_configmap_name = request.param.get("ca_configmap_name", "odh-trusted-ca-bundle")
+    ca_file_name = request.param.get("ca_file_name", "invalid-ca.crt")
+    ca_file_path = f"{CA_MOUNT_PATH}/{ca_file_name}"
+    ca_data = {ca_file_name: "-----BEGIN CERTIFICATE-----\nINVALIDCERTIFICATE\n-----END CERTIFICATE-----"}
+    ca_configmap = ConfigMap(
+        client=admin_client,
+        name=ca_configmap_name,
+        namespace=model_registry_namespace,
+        ensure_exists=True,
+    )
+    patch = {
+        "metadata": {
+            "name": ca_configmap_name,
+            "namespace": model_registry_namespace,
+        },
+        "data": ca_data,
+    }
+    with ResourceEditor(patches={ca_configmap: patch}):
+        yield ca_file_path
+
+
+@pytest.fixture(scope="class")
+def mysql_template_with_ca(model_registry_db_secret: Secret) -> dict[str, Any]:
+    """
+    Patches the MySQL template with the CA file path and volume mount.
+
+    Args:
+        model_registry_db_secret: The secret for the model registry's MySQL database
+
+    Returns:
+        dict[str, Any]: The patched MySQL template
+    """
+    mysql_template = get_model_registry_deployment_template_dict(
+        secret_name=model_registry_db_secret.name,
+        resource_name=DB_RESOURCES_NAME,
+    )
+    mysql_template["spec"]["containers"][0]["args"].append(f"--ssl-ca={CA_FILE_PATH}")
+    mysql_template["spec"]["containers"][0]["volumeMounts"].append({
+        "mountPath": CA_MOUNT_PATH,
+        "name": CA_CONFIGMAP_NAME,
+        "readOnly": True,
+    })
+    mysql_template["spec"]["volumes"].append({"name": CA_CONFIGMAP_NAME, "configMap": {"name": CA_CONFIGMAP_NAME}})
+    return mysql_template
+
+
+@pytest.fixture(scope="class")
+def deploy_secure_mysql_and_mr(
+    model_registry_namespace: str,
+    model_registry_db_secret: Secret,
+    model_registry_db_deployment: Deployment,
+    model_registry_mysql_config: dict[str, Any],
+    mysql_template_with_ca: dict[str, Any],
+    patch_mysql_deployment_with_ssl_ca: Deployment,
+) -> Generator[ModelRegistry, None, None]:
+    """
+    Deploy a secure MySQL and Model Registry instance.
+
+    Args:
+        model_registry_namespace: The namespace of the model registry
+        model_registry_db_secret: The secret for the model registry's MySQL database
+        model_registry_db_deployment: The deployment for the model registry's MySQL database
+        model_registry_mysql_config: The MySQL config dictionary
+        mysql_template_with_ca: The MySQL template with the CA file path and volume mount
+    """
+    with ModelRegistry(
+        name=SECURE_MR_NAME,
+        namespace=model_registry_namespace,
+        label=MODEL_REGISTRY_STANDARD_LABELS,
+        grpc={},
+        rest={},
+        istio=None,
+        oauth_proxy=OAUTH_PROXY_CONFIG_DICT,
+        mysql=model_registry_mysql_config,
+        wait_for_resource=True,
+    ) as mr:
+        mr.wait_for_condition(condition="Available", status="True")
+        yield mr
+
+
+@pytest.fixture()
+def local_ca_bundle(request: pytest.FixtureRequest, admin_client: DynamicClient) -> Generator[str, Any, Any]:
+    """
+    Creates a local CA bundle file by fetching the CA bundle from a ConfigMap and appending the router CA from a Secret.
+    Args:
+        request: The pytest request object
+        admin_client: The admin client to get the CA bundle from a ConfigMap and append the router CA from a Secret.
+    Returns:
+        Generator[str, Any, Any]: A generator that yields the CA bundle path.
+    """
+    ca_bundle_path = getattr(request, "param", {}).get("cert_name", "ca-bundle.crt")
+    create_ca_bundle_with_router_cert(
+        client=admin_client,
+        namespace=py_config["model_registry_namespace"],
+        ca_bundle_path=ca_bundle_path,
+        cert_name=ca_bundle_path,
+    )
+    yield ca_bundle_path
+
+    os.remove(ca_bundle_path)
+
+
+@pytest.fixture(scope="class")
+def ca_configmap_for_test(
+    admin_client: DynamicClient,
+    model_registry_namespace: str,
+    mysql_ssl_artifact_paths: dict[str, Any],
+) -> Generator[ConfigMap, None, None]:
+    """
+    Creates a test-specific ConfigMap for the CA bundle, using the generated CA cert.
+
+    Args:
+        admin_client: The admin client to create the ConfigMap
+        model_registry_namespace: The namespace of the model registry
+        mysql_ssl_secrets: The artifacts and secrets for the MySQL SSL connection
+
+    Returns:
+        Generator[ConfigMap, None, None]: A generator that yields the ConfigMap instance.
+    """
+    with open(mysql_ssl_artifact_paths["ca_crt"], "r") as f:
+        ca_content = f.read()
+    if not ca_content:
+        LOGGER.info("CA content is empty")
+        raise Exception("CA content is empty")
+    cm_name = "mysql-ca-configmap"
+    with ConfigMap(
+        client=admin_client,
+        name=cm_name,
+        namespace=model_registry_namespace,
+        data={"ca-bundle.crt": ca_content},
+    ) as cm:
+        yield cm
+
+
+@pytest.fixture(scope="class")
+def patch_mysql_deployment_with_ssl_ca(
+    request: pytest.FixtureRequest,
+    admin_client: DynamicClient,
+    model_registry_namespace: str,
+    model_registry_db_deployment: Deployment,
+    mysql_ssl_secrets: dict[str, Any],
+    ca_configmap_for_test: ConfigMap,
+) -> Generator[Deployment, Any, Any]:
+    """
+    Patch the MySQL deployment to use the test CA bundle (mysql-ca-configmap),
+    and mount the server cert/key for SSL.
+    """
+    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"]
+    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
+    ])
+
+    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
+
+
+@pytest.fixture(scope="class")
+def mysql_ssl_artifact_paths() -> Generator[dict[str, str], None, None]:
+    """
+    Generates MySQL SSL certificate and key files in a temporary directory
+    and provides their paths.
+
+    Args:
+        admin_client: The admin client to create the ConfigMap
+        model_registry_namespace: The namespace of the model registry
+        mysql_ssl_artifacts_and_secrets: The artifacts and secrets for the MySQL SSL connection
+
+    Returns:
+        Generator[dict[str, str], None, None]: A generator that yields the CA certificate and key file paths.
+    """
+    with tempfile.TemporaryDirectory() as tmp_dir:
+        yield generate_ca_and_server_cert(tmp_dir=tmp_dir)
+
+
+@pytest.fixture(scope="class")
+def mysql_ssl_secrets(
+    admin_client: DynamicClient,
+    model_registry_namespace: str,
+    mysql_ssl_artifact_paths: dict[str, str],
+) -> Generator[dict[str, Secret], None, None]:
+    """
+    Creates Kubernetes secrets for MySQL SSL artifacts.
+
+    Args:
+        admin_client: The admin client to create the ConfigMap
+        model_registry_namespace: The namespace of the model registry
+        mysql_ssl_artifacts_and_secrets: The artifacts and secrets for the MySQL SSL connection
+
+    Returns:
+        Generator[dict[str, str], None, None]: A generator that yields the CA certificate and key file paths.
+    """
+    ca_secret = create_k8s_secret(
+        client=admin_client,
+        namespace=model_registry_namespace,
+        name="mysql-ca",
+        file_path=mysql_ssl_artifact_paths["ca_crt"],
+        key_name="ca.crt",
+    )
+    server_cert_secret = create_k8s_secret(
+        client=admin_client,
+        namespace=model_registry_namespace,
+        name="mysql-server-cert",
+        file_path=mysql_ssl_artifact_paths["server_crt"],
+        key_name="tls.crt",
+    )
+    server_key_secret = create_k8s_secret(
+        client=admin_client,
+        namespace=model_registry_namespace,
+        name="mysql-server-key",
+        file_path=mysql_ssl_artifact_paths["server_key"],
+        key_name="tls.key",
+    )
+
+    yield {
+        "ca_secret": ca_secret,
+        "server_cert_secret": server_cert_secret,
+        "server_key_secret": server_key_secret,
+    }