Add image validation for TrustyAI Operator and service image

tests/conftest.py

@@ -2,7 +2,7 @@
 import os
 import shutil
 from ast import literal_eval
-from typing import Any, Callable, Generator
+from typing import Any, Callable, Generator, Set
 
 import pytest
 import shortuuid
@@ -45,6 +45,7 @@
 )
 from utilities.infra import update_configmap_data
 from utilities.minio import create_minio_data_connection_secret
+from utilities.operator_utils import get_csv_related_images
 
 LOGGER = get_logger(name=__name__)
 
@@ -537,3 +538,10 @@ def prometheus(admin_client: DynamicClient) -> Prometheus:
         ),  # TODO: Verify SSL with appropriate certs
         bearer_token=get_openshift_token(),
     )
+
+
+@pytest.fixture(scope="session")
+def related_images_refs(admin_client: DynamicClient) -> Set[str]:
+    related_images = get_csv_related_images(admin_client=admin_client)
+    related_images_refs = {img["image"] for img in related_images}
+    return related_images_refs

tests/model_explainability/trustyai_service/conftest.py

@@ -1,4 +1,4 @@
-from typing import Generator, Any
+from typing import Any, Generator
 
 import pytest
 import yaml
@@ -21,21 +21,21 @@
 from ocp_resources.subscription import Subscription
 from ocp_resources.trustyai_service import TrustyAIService
 from ocp_utilities.operators import install_operator, uninstall_operator
+from pytest_testconfig import config as py_config
 
 from tests.model_explainability.trustyai_service.trustyai_service_utils import (
     wait_for_isvc_deployment_registered_by_trustyai_service,
 )
 from tests.model_explainability.trustyai_service.utils import (
-    wait_for_mariadb_operator_deployments,
+    TRUSTYAI_SERVICE_NAME,
     create_trustyai_service,
+    wait_for_mariadb_operator_deployments,
     wait_for_mariadb_pods,
-    TRUSTYAI_SERVICE_NAME,
 )
-from utilities.operator_utils import get_cluster_service_version
-
-from utilities.constants import Timeout, KServeDeploymentType, ApiGroups, Labels, Ports
+from utilities.constants import ApiGroups, KServeDeploymentType, Labels, Ports, Timeout
 from utilities.inference_utils import create_isvc
-from utilities.infra import update_configmap_data, create_inference_token
+from utilities.infra import create_inference_token, update_configmap_data
+from utilities.operator_utils import get_cluster_service_version
 
 OPENSHIFT_OPERATORS: str = "openshift-operators"
 
@@ -455,3 +455,15 @@ def isvc_getter_token_secret(
 @pytest.fixture(scope="class")
 def isvc_getter_token(isvc_getter_service_account: ServiceAccount, isvc_getter_token_secret: Secret) -> str:
     return create_inference_token(model_service_account=isvc_getter_service_account)
+
+
+@pytest.fixture(scope="class")
+def trustyai_operator_configmap(
+    admin_client: DynamicClient,
+) -> ConfigMap:
+    return ConfigMap(
+        client=admin_client,
+        namespace=py_config["applications_namespace"],
+        name=f"{TRUSTYAI_SERVICE_NAME}-operator-config",
+        ensure_exists=True,
+    )

tests/model_explainability/trustyai_service/test_trustyai_service.py

@@ -1,7 +1,15 @@
+from typing import Set
+
 import pytest
+from ocp_resources.config_map import ConfigMap
 from ocp_resources.namespace import Namespace
+from ocp_resources.trustyai_service import TrustyAIService
 
-from tests.model_explainability.trustyai_service.utils import validate_trustyai_service_db_conn_failure
+from tests.model_explainability.trustyai_service.utils import (
+    validate_trustyai_service_db_conn_failure,
+    validate_trustyai_operator_image,
+    validate_trustyai_service_images,
+)
 
 
 @pytest.mark.parametrize(
@@ -25,3 +33,46 @@ def test_trustyai_service_with_invalid_db_cert(
         namespace=model_namespace,
         label_selector=f"app.kubernetes.io/instance={trustyai_service_with_invalid_db_cert.name}",
     )
+
+
+@pytest.mark.parametrize(
+    "model_namespace",
+    [
+        pytest.param(
+            {"name": "test-validate-trustyai-images"},
+        )
+    ],
+    indirect=True,
+)
+@pytest.mark.smoke
+class TestValidateTrustyAIImages:
+    """Test to validate if operator and service image for TrustyAIService matches related images in the CSV.
+    Also validate if the image is pinned using a digest and is sourced from registry.redhat.io.
+    """
+
+    def test_validate_trustyai_operator_image(
+        self,
+        admin_client,
+        model_namespace: Namespace,
+        related_images_refs: Set[str],
+        trustyai_operator_configmap: ConfigMap,
+    ):
+        return validate_trustyai_operator_image(
+            client=admin_client,
+            related_images_refs=related_images_refs,
+            tai_operator_configmap_data=trustyai_operator_configmap.instance.data,
+        )
+
+    def test_validate_trustyai_service_image(
+        self,
+        admin_client,
+        model_namespace: Namespace,
+        related_images_refs: Set[str],
+        trustyai_service_with_pvc_storage: TrustyAIService,
+    ):
+        return validate_trustyai_service_images(
+            client=admin_client,
+            related_images_refs=related_images_refs,
+            model_namespace=model_namespace,
+            label_selector=f"app.kubernetes.io/instance={trustyai_service_with_pvc_storage.name}",
+        )

tests/model_explainability/trustyai_service/utils.py

@@ -1,23 +1,26 @@
-from typing import Generator, Any, Optional
 import re
+from typing import Any, Generator, Optional, Set
 
 from kubernetes.dynamic import DynamicClient
 from ocp_resources.deployment import Deployment
-from ocp_resources.mariadb_operator import MariadbOperator
 from ocp_resources.maria_db import MariaDB
+from ocp_resources.mariadb_operator import MariadbOperator
 from ocp_resources.namespace import Namespace
 from ocp_resources.pod import Pod
 from ocp_resources.trustyai_service import TrustyAIService
+from pytest_testconfig import config as py_config
 from simple_logger.logger import get_logger
-from timeout_sampler import TimeoutSampler
+from timeout_sampler import TimeoutSampler, retry
+
 from tests.model_explainability.trustyai_service.trustyai_service_utils import TRUSTYAI_SERVICE_NAME
 from utilities.constants import Timeout
-from timeout_sampler import retry
-
 from utilities.exceptions import TooManyPodsError, UnexpectedFailureError
+from utilities.general import validate_image_format, wait_for_pods_by_labels, validate_container_images
 
 LOGGER = get_logger(name=__name__)
 
+TAI_OPERATOR_DEPLOYMENT_NAME = f"{TRUSTYAI_SERVICE_NAME}-operator-controller-manager"
+
 
 def wait_for_mariadb_operator_deployments(mariadb_operator: MariadbOperator) -> None:
     expected_deployment_names: list[str] = [
@@ -147,3 +150,39 @@ def create_trustyai_service(
         if wait_for_replicas:
             trustyai_deployment.wait_for_replicas()
         yield trustyai_service
+
+
+def validate_trustyai_operator_image(
+    client: DynamicClient, related_images_refs: Set[str], tai_operator_configmap_data: dict[str, str]
+) -> None:
+    """Validates the TrustyAI operator image.
+    Checks if:
+        - container image matches that of the operator configmap.
+        - image is present in relatedImages of CSV.
+        - image complies with OpenShift AI requirements i.e. sourced from registry.redhat.io and pinned w/o tags.
+    """
+    tai_operator_deployment = Deployment(
+        client=client,
+        name=TAI_OPERATOR_DEPLOYMENT_NAME,
+        namespace=py_config["applications_namespace"],
+        wait_for_resource=True,
+    )
+    tai_operator_image = tai_operator_deployment.instance.spec.template.spec.containers[0].image
+    assert tai_operator_image == tai_operator_configmap_data["trustyaiOperatorImage"]
+    assert tai_operator_image in related_images_refs
+    image_valid, error_message = validate_image_format(image=tai_operator_image)
+    assert image_valid, error_message
+
+
+def validate_trustyai_service_images(
+    client: DynamicClient,
+    related_images_refs: Set[str],
+    model_namespace: Namespace,
+    label_selector: str,
+) -> None:
+    """Validates the TrustyAI service container images."""
+    trustyai_service_pod = wait_for_pods_by_labels(
+        admin_client=client, namespace=model_namespace.name, label_selector=label_selector, expected_num_pods=1
+    )
+    validation_errors = validate_container_images(pod=trustyai_service_pod, valid_image_refs=related_images_refs)
+    assert len(validation_errors) == 0, validation_errors

tests/model_registry/conftest.py

@@ -297,7 +297,7 @@ def model_registry_operator_pod(admin_client: DynamicClient) -> Generator[Pod, A
         namespace=py_config["applications_namespace"],
         label_selector=f"{Labels.OpenDataHubIo.NAME}={MR_OPERATOR_NAME}",
         expected_num_pods=1,
-    )[0]
+    )
 
 
 @pytest.fixture()
@@ -308,7 +308,7 @@ def model_registry_instance_pod(admin_client: DynamicClient) -> Generator[Pod, A
         namespace=py_config["model_registry_namespace"],
         label_selector=f"app={MR_INSTANCE_NAME}",
         expected_num_pods=1,
-    )[0]
+    )
 
 
 @pytest.fixture(scope="package", autouse=True)

utilities/general.py

@@ -222,17 +222,17 @@ def wait_for_pods_by_labels(
     namespace: str,
     label_selector: str,
     expected_num_pods: int,
-) -> list[Pod]:
+) -> list[Pod] | Pod:
     """
-    Get pods by label selector in a namespace.
+    Get pods or a pod by label selector in a namespace.
 
     Args:
         admin_client: The admin client to use for pod retrieval
         namespace: The namespace to search in
         label_selector: The label selector to filter pods
         expected_num_pods: The expected number of pods to be found
     Returns:
-        List of matching pods
+        List of matching pods or a single pod if expected_num_pods is 1
 
     Raises:
         ResourceNotFoundError: If no pods are found
@@ -247,8 +247,8 @@ def wait_for_pods_by_labels(
     if not pods:
         raise ResourceNotFoundError(f"No pods found with label selector {label_selector} in namespace {namespace}")
     if len(pods) != expected_num_pods:
-        raise UnexpectedResourceCountError(f"Expected {expected_num_pods} pods, found {len(pods)}")
-    return pods
+        raise UnexpectedResourceCountError(f"Expected {expected_num_pods} pod(s), found {len(pods)}")
+    return pods[0] if expected_num_pods == 1 else pods
 
 
 def validate_container_images(