add mariadb tests for MR and reorganize some fixtures

tests/conftest.py

@@ -10,12 +10,18 @@
 from _pytest._py.path import LocalPath
 from _pytest.legacypath import TempdirFactory
 from _pytest.tmpdir import TempPathFactory
+from kubernetes.dynamic.exceptions import ResourceNotFoundError
+
+from ocp_resources.cluster_service_version import ClusterServiceVersion
 from ocp_resources.config_map import ConfigMap
+from ocp_resources.deployment import Deployment
 from ocp_resources.dsc_initialization import DSCInitialization
+from ocp_resources.mariadb_operator import MariadbOperator
 from ocp_resources.node import Node
 from ocp_resources.pod import Pod
 from ocp_resources.secret import Secret
 from ocp_resources.service import Service
+from ocp_resources.subscription import Subscription
 from ocp_utilities.monitoring import Prometheus
 from pyhelper_utils.shell import run_command
 from pytest import FixtureRequest, Config
@@ -26,6 +32,7 @@
 from pytest_testconfig import config as py_config
 from simple_logger.logger import get_logger
 
+from ocp_utilities.operators import uninstall_operator, install_operator
 from utilities.certificates_utils import create_ca_bundle_file
 from utilities.data_science_cluster_utils import update_components_in_dsc
 from utilities.exceptions import ClusterLoginError
@@ -42,11 +49,14 @@
     Labels,
     MinIo,
     Protocols,
+    Timeout,
+    OPENSHIFT_OPERATORS,
 )
 from utilities.infra import update_configmap_data
 from utilities.logger import RedactedString
+from utilities.mariadb_utils import wait_for_mariadb_operator_deployments
 from utilities.minio import create_minio_data_connection_secret
-from utilities.operator_utils import get_csv_related_images
+from utilities.operator_utils import get_csv_related_images, get_cluster_service_version
 
 LOGGER = get_logger(name=__name__)
 
@@ -573,3 +583,59 @@ def autouse_fixtures(
 ) -> None:
     """Fixture to control the order of execution of some of the fixtures"""
     return
+
+
+@pytest.fixture(scope="session")
+def installed_mariadb_operator(admin_client: DynamicClient) -> Generator[None, Any, Any]:
+    operator_ns = Namespace(name="openshift-operators", ensure_exists=True)
+    operator_name = "mariadb-operator"
+
+    mariadb_operator_subscription = Subscription(client=admin_client, namespace=operator_ns.name, name=operator_name)
+
+    if not mariadb_operator_subscription.exists:
+        install_operator(
+            admin_client=admin_client,
+            target_namespaces=["openshift-operators"],
+            name=operator_name,
+            channel="alpha",
+            source="community-operators",
+            operator_namespace=operator_ns.name,
+            timeout=Timeout.TIMEOUT_15MIN,
+            install_plan_approval="Manual",
+            starting_csv=f"{operator_name}.v0.38.1",
+        )
+
+        deployment = Deployment(
+            client=admin_client,
+            namespace=operator_ns.name,
+            name=f"{operator_name}-helm-controller-manager",
+            wait_for_resource=True,
+        )
+        deployment.wait_for_replicas()
+    yield
+    uninstall_operator(
+        admin_client=admin_client, name=operator_name, operator_namespace=operator_ns.name, clean_up_namespace=False
+    )
+
+
+@pytest.fixture(scope="class")
+def mariadb_operator_cr(
+    admin_client: DynamicClient, installed_mariadb_operator: None
+) -> Generator[MariadbOperator, Any, Any]:
+    mariadb_csv: ClusterServiceVersion = get_cluster_service_version(
+        client=admin_client, prefix="mariadb", namespace=OPENSHIFT_OPERATORS
+    )
+    alm_examples: list[dict[str, Any]] = mariadb_csv.get_alm_examples()
+    mariadb_operator_cr_dict: dict[str, Any] = next(
+        example for example in alm_examples if example["kind"] == "MariadbOperator"
+    )
+    if not mariadb_operator_cr_dict:
+        raise ResourceNotFoundError(f"No MariadbOperator dict found in alm_examples for CSV {mariadb_csv.name}")
+
+    mariadb_operator_cr_dict["metadata"]["namespace"] = OPENSHIFT_OPERATORS
+    with MariadbOperator(kind_dict=mariadb_operator_cr_dict) as mariadb_operator_cr:
+        mariadb_operator_cr.wait_for_condition(
+            condition="Deployed", status=mariadb_operator_cr.Condition.Status.TRUE, timeout=Timeout.TIMEOUT_10MIN
+        )
+        wait_for_mariadb_operator_deployments(mariadb_operator=mariadb_operator_cr)
+        yield mariadb_operator_cr

tests/model_explainability/trustyai_service/conftest.py

@@ -6,7 +6,6 @@
 from kubernetes.dynamic.exceptions import ResourceNotFoundError
 from ocp_resources.cluster_service_version import ClusterServiceVersion
 from ocp_resources.config_map import ConfigMap
-from ocp_resources.deployment import Deployment
 from ocp_resources.inference_service import InferenceService
 from ocp_resources.maria_db import MariaDB
 from ocp_resources.mariadb_operator import MariadbOperator
@@ -18,9 +17,7 @@
 from ocp_resources.service import Service
 from ocp_resources.service_account import ServiceAccount
 from ocp_resources.serving_runtime import ServingRuntime
-from ocp_resources.subscription import Subscription
 from ocp_resources.trustyai_service import TrustyAIService
-from ocp_utilities.operators import install_operator, uninstall_operator
 
 from tests.model_explainability.trustyai_service.constants import (
     TAI_DATA_CONFIG,
@@ -41,7 +38,6 @@
     wait_for_isvc_deployment_registered_by_trustyai_service,
 )
 from tests.model_explainability.trustyai_service.utils import (
-    wait_for_mariadb_operator_deployments,
     create_trustyai_service,
     wait_for_mariadb_pods,
     TRUSTYAI_SERVICE_NAME,
@@ -52,14 +48,10 @@
 )
 from utilities.logger import RedactedString
 from utilities.operator_utils import get_cluster_service_version
-
-from utilities.constants import Timeout, KServeDeploymentType, Labels
+from utilities.constants import KServeDeploymentType, Labels, OPENSHIFT_OPERATORS, MARIADB
 from utilities.inference_utils import create_isvc
 from utilities.infra import update_configmap_data, create_inference_token
 
-OPENSHIFT_OPERATORS: str = "openshift-operators"
-
-MARIADB: str = "mariadb"
 DB_CREDENTIALS_SECRET_NAME: str = "db-credentials"
 DB_NAME: str = "trustyai_db"
 DB_USERNAME: str = "trustyai_user"
@@ -142,64 +134,6 @@ def db_credentials_secret(admin_client: DynamicClient, model_namespace: Namespac
         yield db_credentials
 
 
-@pytest.fixture(scope="session")
-def installed_mariadb_operator(admin_client: DynamicClient) -> Generator[None, Any, Any]:
-    operator_ns = Namespace(name="openshift-operators", ensure_exists=True)
-    operator_name = "mariadb-operator"
-
-    mariadb_operator_subscription = Subscription(client=admin_client, namespace=operator_ns.name, name=operator_name)
-
-    if not mariadb_operator_subscription.exists:
-        install_operator(
-            admin_client=admin_client,
-            target_namespaces=[],
-            name=operator_name,
-            channel="alpha",
-            source="community-operators",
-            operator_namespace=operator_ns.name,
-            timeout=Timeout.TIMEOUT_15MIN,
-            install_plan_approval="Manual",
-            starting_csv=f"{operator_name}.v0.37.1",
-        )
-
-        deployment = Deployment(
-            client=admin_client,
-            namespace=operator_ns.name,
-            name=f"{operator_name}-helm-controller-manager",
-            wait_for_resource=True,
-        )
-        deployment.wait_for_replicas()
-
-    yield
-    uninstall_operator(
-        admin_client=admin_client, name=operator_name, operator_namespace=operator_ns.name, clean_up_namespace=False
-    )
-
-
-@pytest.fixture(scope="class")
-def mariadb_operator_cr(
-    admin_client: DynamicClient, installed_mariadb_operator: None
-) -> Generator[MariadbOperator, Any, Any]:
-    mariadb_csv: ClusterServiceVersion = get_cluster_service_version(
-        client=admin_client, prefix="mariadb", namespace=OPENSHIFT_OPERATORS
-    )
-    alm_examples: list[dict[str, Any]] = mariadb_csv.get_alm_examples()
-    mariadb_operator_cr_dict: dict[str, Any] = next(
-        example for example in alm_examples if example["kind"] == "MariadbOperator"
-    )
-
-    if not mariadb_operator_cr_dict:
-        raise ResourceNotFoundError(f"No MariadbOperator dict found in alm_examples for CSV {mariadb_csv.name}")
-
-    mariadb_operator_cr_dict["metadata"]["namespace"] = OPENSHIFT_OPERATORS
-    with MariadbOperator(kind_dict=mariadb_operator_cr_dict) as mariadb_operator_cr:
-        mariadb_operator_cr.wait_for_condition(
-            condition="Deployed", status=mariadb_operator_cr.Condition.Status.TRUE, timeout=Timeout.TIMEOUT_10MIN
-        )
-        wait_for_mariadb_operator_deployments(mariadb_operator=mariadb_operator_cr)
-        yield mariadb_operator_cr
-
-
 @pytest.fixture(scope="class")
 def mariadb(
     admin_client: DynamicClient,

tests/model_registry/conftest.py

@@ -148,11 +148,8 @@ def model_registry_db_secret(
 @pytest.fixture(scope="class")
 def model_registry_db_deployment(
     pytestconfig: Config,
-    admin_client: DynamicClient,
     model_registry_namespace: str,
     model_registry_db_secret: Secret,
-    model_registry_db_pvc: PersistentVolumeClaim,
-    model_registry_db_service: Service,
     teardown_resources: bool,
     is_model_registry_oauth: bool,
 ) -> Generator[Deployment, Any, Any]:
@@ -183,7 +180,17 @@ def model_registry_db_deployment(
 
 
 @pytest.fixture(scope="class")
-def model_registry_instance(
+def model_registry_mysql_metadata_db(
+    model_registry_db_secret: Secret,
+    model_registry_db_pvc: PersistentVolumeClaim,
+    model_registry_db_service: Service,
+    model_registry_db_deployment: Deployment,
+) -> Deployment:
+    return model_registry_db_deployment
+
+
+@pytest.fixture(scope="class")
+def model_registry_instance_mysql(
     pytestconfig: Config,
     admin_client: DynamicClient,
     model_registry_namespace: str,
@@ -228,29 +235,22 @@ def model_registry_instance(
 
 
 @pytest.fixture(scope="class")
-def model_registry_mysql_config(
-    request: FixtureRequest,
-    model_registry_db_deployment: Deployment,
-    model_registry_db_secret: Secret,
-) -> dict[str, Any]:
+def model_registry_mysql_config(request: FixtureRequest, model_registry_namespace: str) -> dict[str, Any]:
     """
     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
-
+        model_registry_namespace: The model registry namespoce
     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",
+        "host": f"{DB_RESOURCES_NAME}.{model_registry_namespace}.svc.cluster.local",
         "database": MODEL_REGISTRY_DB_SECRET_STR_DATA["database-name"],
-        "passwordSecret": {"key": "database-password", "name": model_registry_db_deployment.name},
+        "passwordSecret": {"key": "database-password", "name": DB_RESOURCES_NAME},
         "port": param.get("port", 3306),
         "skipDBCreation": False,
         "username": MODEL_REGISTRY_DB_SECRET_STR_DATA["database-user"],
@@ -262,45 +262,23 @@ def model_registry_mysql_config(
 
 
 @pytest.fixture(scope="class")
-def model_registry_instance_service(
-    admin_client: DynamicClient,
-    model_registry_namespace: str,
-    model_registry_instance: ModelRegistry,
-) -> Service:
-    """
-    Get the service for the regular model registry instance.
-    Args:
-        admin_client: The admin client
-        model_registry_namespace: The namespace where the model registry is deployed
-        model_registry_instance: The model registry instance to get the service for
-    Returns:
-        Service: The service for the model registry instance
-    """
-    return get_mr_service_by_label(
-        client=admin_client, namespace_name=model_registry_namespace, mr_instance=model_registry_instance
+def model_registry_instance_rest_endpoint(admin_client: DynamicClient, model_registry_namespace: str) -> str:
+    return get_endpoint_from_mr_service(
+        svc=get_mr_service_by_label(
+            client=admin_client,
+            namespace_name=model_registry_namespace,
+            mr_instance=ModelRegistry(name=MR_INSTANCE_NAME, namespace=model_registry_namespace),
+        ),
+        protocol=Protocols.REST,
     )
 
 
-@pytest.fixture(scope="class")
-def model_registry_instance_rest_endpoint(
-    model_registry_instance_service: Service,
-) -> str:
-    """
-    Get the REST endpoint for the model registry instance.
-    Args:
-        model_registry_instance_service: The service for the model registry instance
-    Returns:
-        str: The REST endpoint for the model registry instance
-    """
-    return get_endpoint_from_mr_service(svc=model_registry_instance_service, protocol=Protocols.REST)
-
-
 @pytest.fixture(scope="class")
 def updated_dsc_component_state_scope_class(
     pytestconfig: Config,
+    admin_client: DynamicClient,
     request: FixtureRequest,
     dsc_resource: DataScienceCluster,
-    admin_client: DynamicClient,
     teardown_resources: bool,
     is_model_registry_oauth: bool,
 ) -> Generator[DataScienceCluster, Any, Any]:

tests/model_registry/image_validation/test_verify_rhoai_images.py

@@ -28,7 +28,12 @@
     ],
     indirect=True,
 )
-@pytest.mark.usefixtures("updated_dsc_component_state_scope_class")
+@pytest.mark.usefixtures(
+    "updated_dsc_component_state_scope_class",
+    "is_model_registry_oauth",
+    "model_registry_mysql_metadata_db",
+    "model_registry_instance_mysql",
+)
 @pytest.mark.downstream_only
 class TestModelRegistryImages:
     """
@@ -43,7 +48,7 @@ class TestModelRegistryImages:
     def test_verify_model_registry_images(
         self: Self,
         admin_client: DynamicClient,
-        model_registry_instance: ModelRegistry,
+        model_registry_instance_mysql: ModelRegistry,
         model_registry_operator_pod: Pod,
         model_registry_instance_pod: Pod,
         related_images_refs: Set[str],

tests/model_registry/negative_tests/test_db_migration.py

@@ -1,8 +1,8 @@
 import pytest
 from typing import Self
 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 ocp_resources.pod import Pod
 from utilities.constants import DscComponents
 from tests.model_registry.constants import MR_INSTANCE_NAME
@@ -27,12 +27,13 @@
     ],
     indirect=True,
 )
-@pytest.mark.usefixtures("updated_dsc_component_state_scope_class")
+@pytest.mark.usefixtures(
+    "updated_dsc_component_state_scope_class", "model_registry_mysql_metadata_db", "model_registry_instance_mysql"
+)
 class TestDBMigration:
     def test_db_migration_negative(
         self: Self,
         admin_client: DynamicClient,
-        model_registry_instance: ModelRegistry,
         model_registry_db_instance_pod: Pod,
         set_mr_db_dirty: int,
         delete_mr_deployment: None,