test: add test to initiate mr with secure db connection

tests/model_registry/constants.py

@@ -43,3 +43,8 @@ class ModelRegistryEndpoints:
     f"{Resource.ApiGroup.TEMPLATE_OPENSHIFT_IO}/expose-password": "'{.data[''database-password'']}'",
     f"{Resource.ApiGroup.TEMPLATE_OPENSHIFT_IO}/expose-username": "'{.data[''database-user'']}'",
 }
+
+SECURE_MR_NAME = "secure-db-mr"
+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"

tests/model_registry/rest_api/conftest.py

@@ -1,10 +1,23 @@
-from typing import Any
-
+from typing import Any, Generator
+import os
+import base64
+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 utilities.constants import Protocols
+from ocp_resources.deployment import Deployment
+from ocp_resources.service import Service
+from tests.model_registry.utils import get_model_registry_deployment_template_dict, create_secure_model_registry
+from tests.model_registry.constants import DB_RESOURCES_NAME, CA_MOUNT_PATH, CA_FILE_PATH, CA_CONFIGMAP_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 import ModelRegistry
+
+
+LOGGER = get_logger(name=__name__)
 
 
 @pytest.fixture(scope="class")
@@ -47,3 +60,127 @@ def updated_model_artifact(
         headers=model_registry_rest_headers,
         data_json=request.param,
     )
+
+
+@pytest.fixture(scope="function")
+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,
+    )
+    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="function")
+def model_registry_instance_ca(
+    model_registry_namespace: str,
+    model_registry_db_secret: Secret,
+    model_registry_db_service: Service,
+    patch_invalid_ca: str,
+) -> Generator[ModelRegistry, Any, Any]:
+    """
+    Deploys a Model Registry instance with a custom CA certificate.
+    """
+    ca_file_path = patch_invalid_ca
+    mr = create_secure_model_registry(
+        model_registry_namespace=model_registry_namespace,
+        model_registry_db_service=model_registry_db_service,
+        model_registry_db_secret=model_registry_db_secret,
+        ca_file_path=ca_file_path,
+    )
+    yield mr
+
+
+@pytest.fixture(scope="class")
+def deploy_secure_mysql_and_mr(
+    model_registry_namespace: str,
+    model_registry_db_secret: Secret,
+    model_registry_db_service: Service,
+    model_registry_db_deployment: Deployment,
+) -> Generator[ModelRegistry, None, None]:
+    """
+    Deploys MySQL with SSL/TLS and a Model Registry configured to use a secure DB connection.
+
+    Ensures the odh-trusted-ca-bundle ConfigMap is mounted in both MySQL and Model Registry deployments,
+    and the --ssl-ca argument is set to the correct path.
+    """
+    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}})
+
+    patch = {"spec": {"template": mysql_template["spec"]}}
+
+    with ResourceEditor(patches={model_registry_db_deployment: patch}):
+        with create_secure_model_registry(
+            model_registry_namespace=model_registry_namespace,
+            model_registry_db_service=model_registry_db_service,
+            model_registry_db_secret=model_registry_db_secret,
+            ca_file_path=CA_FILE_PATH,
+        ) 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.
+    """
+    namespace = getattr(request, "param", {}).get("namespace", "test-model-registry-namespace")
+    ca_bundle_path = getattr(request, "param", {}).get("ca_bundle_path", "ca-bundle.crt")
+    cert_name = getattr(request, "param", {}).get("cert_name", "ca-bundle.crt")
+
+    cm = ConfigMap(client=admin_client, name="odh-trusted-ca-bundle", namespace=namespace)
+    ca_bundle_content = cm.instance.data.get(cert_name)
+    with open(ca_bundle_path, "w") as f:
+        f.write(ca_bundle_content)
+
+    try:
+        router_secret = Secret(client=admin_client, name="router-ca", namespace="openshift-ingress-operator")
+        router_ca_b64 = router_secret.instance.data.get("tls.crt")
+        if router_ca_b64:
+            router_ca_content = base64.b64decode(router_ca_b64).decode("utf-8")
+            with open(ca_bundle_path, "r") as bundle:
+                bundle_content = bundle.read()
+            if router_ca_content not in bundle_content:
+                with open(ca_bundle_path, "a") as bundle_append:
+                    bundle_append.write("\n" + router_ca_content)
+    except Exception:
+        pytest.fail(
+            "router-ca secret not found in openshift-ingress-operator. Proceeding without appending ingress CA."
+        )
+    yield ca_bundle_path
+
+    try:
+        os.remove(ca_bundle_path)
+    except FileNotFoundError:
+        pass

tests/model_registry/rest_api/test_model_registry_secure_db.py

@@ -0,0 +1,95 @@
+import pytest
+import requests
+from typing import Self
+from tests.model_registry.rest_api.utils import register_model_rest_api
+from tests.model_registry.rest_api.constants import MODEL_REGISTER_DATA
+from utilities.constants import DscComponents
+from simple_logger.logger import get_logger
+from tests.model_registry.utils import generate_random_name
+
+
+LOGGER = get_logger(name=__name__)
+
+
+@pytest.mark.parametrize(
+    "updated_dsc_component_state_scope_class",
+    [
+        pytest.param(
+            {
+                "component_patch": {
+                    DscComponents.MODELREGISTRY: {
+                        "managementState": DscComponents.ManagementState.MANAGED,
+                        "registriesNamespace": "test-model-registry-namespace",
+                    },
+                },
+            },
+        ),
+    ],
+    indirect=True,
+)
+@pytest.mark.usefixtures("updated_dsc_component_state_scope_class")
+class TestModelRegistryWithSecureDB:
+    """
+    Test suite for validating Model Registry functionality with a secure MySQL database connection (SSL/TLS).
+    Includes tests for both invalid and valid CA certificate scenarios.
+    """
+
+    # Implements RHOAIENG-26150
+    @pytest.mark.usefixtures("model_registry_instance_ca")
+    @pytest.mark.parametrize(
+        "patch_invalid_ca",
+        [{"ca_configmap_name": "odh-trusted-ca-bundle", "ca_file_name": "invalid-ca.crt"}],
+        indirect=True,
+    )
+    def test_register_model_with_invalid_ca(
+        self: Self,
+        patch_invalid_ca: dict[str, str],
+        model_registry_rest_url: str,
+        model_registry_rest_headers: dict[str, str],
+    ) -> None:
+        """
+        Test that model registration fails with an SSLError when the Model Registry is deployed
+        with an invalid CA certificate.
+        """
+        model_name = generate_random_name(prefix="model-rest-api")
+        MODEL_REGISTER_DATA["register_model_data"]["name"] = model_name
+        with pytest.raises(requests.exceptions.SSLError) as exc_info:
+            register_model_rest_api(
+                model_registry_rest_url=model_registry_rest_url,
+                model_registry_rest_headers=model_registry_rest_headers,
+                data_dict=MODEL_REGISTER_DATA,
+                verify=True,
+            )
+        assert "certificate verify failed" in str(exc_info.value), (
+            f"Expected SSL certificate verification failure, got: {exc_info.value}"
+        )
+
+    # Implements RHOAIENG-26150
+    @pytest.mark.usefixtures("deploy_secure_mysql_and_mr")
+    @pytest.mark.parametrize(
+        "local_ca_bundle",
+        [{"cert_name": "ca-bundle.crt"}, {"cert_name": "odh-ca-bundle.crt"}],
+        indirect=True,
+    )
+    def test_register_model_with_default_ca(
+        self: Self,
+        model_registry_rest_url: str,
+        model_registry_rest_headers: dict[str, str],
+        local_ca_bundle: str,
+    ) -> None:
+        """
+        Deploys Model Registry with a secure MySQL DB (SSL/TLS), registers a model, and checks functionality.
+        Uses a CA bundle file for SSL verification by passing it directly to the verify parameter.
+        """
+        model_name = generate_random_name(prefix="model-rest-api")
+        MODEL_REGISTER_DATA["register_model_data"]["name"] = model_name
+        result = register_model_rest_api(
+            model_registry_rest_url=model_registry_rest_url,
+            model_registry_rest_headers=model_registry_rest_headers,
+            data_dict=MODEL_REGISTER_DATA,
+            verify=local_ca_bundle,
+        )
+        assert result["register_model"].get("id"), "Model registration failed with secure DB connection."
+        for k, v in MODEL_REGISTER_DATA["register_model_data"].items():
+            assert result["register_model"][k] == v, f"Expected {k}={v}, got {result[k]}"
+        LOGGER.info(f"Model registered successfully with secure DB using {local_ca_bundle}")

tests/model_registry/rest_api/utils.py

@@ -29,8 +29,10 @@ def execute_model_registry_patch_command(
         raise
 
 
-def execute_model_registry_post_command(url: str, headers: dict[str, str], data_json: dict[str, Any]) -> dict[Any, Any]:
-    resp = requests.post(url=url, json=data_json, headers=headers, verify=False, timeout=60)
+def execute_model_registry_post_command(
+    url: str, headers: dict[str, str], data_json: dict[str, Any], verify: bool | str = False
+) -> dict[Any, Any]:
+    resp = requests.post(url=url, json=data_json, headers=headers, verify=verify, timeout=60)
     LOGGER.info(f"url: {url}, status code: {resp.status_code}, rep: {resp.text}")
 
     if resp.status_code not in [200, 201]:
@@ -60,13 +62,17 @@ def execute_model_registry_get_command(url: str, headers: dict[str, str]) -> dic
 
 
 def register_model_rest_api(
-    model_registry_rest_url: str, model_registry_rest_headers: dict[str, str], data_dict: dict[str, Any]
+    model_registry_rest_url: str,
+    model_registry_rest_headers: dict[str, str],
+    data_dict: dict[str, Any],
+    verify: bool | str = False,
 ) -> dict[str, Any]:
     # register a model
     register_model = execute_model_registry_post_command(
         url=f"{model_registry_rest_url}{MODEL_REGISTRY_BASE_URI}registered_models",
         headers=model_registry_rest_headers,
         data_json=data_dict["register_model_data"],
+        verify=verify,
     )
     # create associated model version:
     model_data = data_dict["model_version_data"]
@@ -75,12 +81,14 @@ def register_model_rest_api(
         url=f"{model_registry_rest_url}{MODEL_REGISTRY_BASE_URI}model_versions",
         headers=model_registry_rest_headers,
         data_json=model_data,
+        verify=verify,
     )
     # create associated model artifact
     model_artifact = execute_model_registry_post_command(
         url=f"{model_registry_rest_url}{MODEL_REGISTRY_BASE_URI}model_versions/{model_version['id']}/artifacts",
         headers=model_registry_rest_headers,
         data_json=data_dict["model_artifact_data"],
+        verify=verify,
     )
     LOGGER.info(
         f"Successfully registered model: {register_model}, with version: {model_version} and "

tests/model_registry/utils.py

@@ -10,9 +10,10 @@
 from simple_logger.logger import get_logger
 from timeout_sampler import TimeoutExpiredError, TimeoutSampler
 from kubernetes.dynamic.exceptions import NotFoundError
-from tests.model_registry.constants import MR_DB_IMAGE_DIGEST
+from tests.model_registry.constants import MR_DB_IMAGE_DIGEST, ISTIO_CONFIG_DICT, DB_RESOURCES_NAME, SECURE_MR_NAME
 from utilities.exceptions import ProtocolNotSupportedError, TooManyServicesError
 from utilities.constants import Protocols, Annotations
+from ocp_resources.secret import Secret
 
 ADDRESS_ANNOTATION_PREFIX: str = "routing.opendatahub.io/external-address-"
 
@@ -264,3 +265,38 @@ 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 create_secure_model_registry(
+    model_registry_namespace: str,
+    model_registry_db_service: Service,
+    model_registry_db_secret: Secret,
+    ca_file_path: str,
+) -> ModelRegistry:
+    """
+    Helper to create a ModelRegistry with secure MySQL connection.
+    Returns a context manager yielding the ModelRegistry resource.
+    """
+    return ModelRegistry(
+        name=SECURE_MR_NAME,
+        namespace=model_registry_namespace,
+        label={
+            Annotations.KubernetesIo.NAME: SECURE_MR_NAME,
+            Annotations.KubernetesIo.INSTANCE: SECURE_MR_NAME,
+            Annotations.KubernetesIo.PART_OF: "model-registry-operator",
+            Annotations.KubernetesIo.CREATED_BY: "model-registry-operator",
+        },
+        grpc={},
+        rest={},
+        istio=ISTIO_CONFIG_DICT,
+        mysql={
+            "host": f"{model_registry_db_service.name}.{model_registry_db_service.namespace}.svc.cluster.local",
+            "database": model_registry_db_secret.string_data["database-name"],
+            "passwordSecret": {"key": "database-password", "name": DB_RESOURCES_NAME},
+            "port": 3306,
+            "skipDBCreation": False,
+            "username": model_registry_db_secret.string_data["database-user"],
+            "ssl_ca": ca_file_path,
+        },
+        wait_for_resource=True,
+    )