Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
37 changes: 0 additions & 37 deletions tests/model_registry/model_registry/async_job/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,17 +5,14 @@
import pytest
import shortuuid
from kubernetes.dynamic import DynamicClient
from kubernetes.dynamic.exceptions import ResourceNotFoundError
from model_registry import ModelRegistry as ModelRegistryClient
from model_registry.types import RegisteredModel
from ocp_resources.config_map import ConfigMap
from ocp_resources.job import Job
from ocp_resources.role_binding import RoleBinding
from ocp_resources.secret import Secret
from ocp_resources.service import Service
from ocp_resources.service_account import ServiceAccount
from pytest import FixtureRequest
from pytest_testconfig import py_config

from tests.model_registry.model_registry.async_job.constants import (
ASYNC_JOB_ANNOTATIONS,
Expand Down Expand Up @@ -109,40 +106,6 @@ def oci_secret_for_async_job(
yield secret


@pytest.fixture(scope="class")
def async_upload_image(admin_client: DynamicClient) -> str:
"""
Get the async upload image dynamically from the model-registry-operator-parameters ConfigMap.

This fetches the image from the cluster at runtime instead of using a hardcoded value.

Args:
admin_client: Kubernetes client for resource access

Returns:
str: The async upload image URL from the ConfigMap

Raises:
KeyError: If the ConfigMap or the required key doesn't exist
"""
config_map = ConfigMap(
client=admin_client,
name="model-registry-operator-parameters",
namespace=py_config["applications_namespace"],
)

if not config_map.exists:
raise ResourceNotFoundError(
f"ConfigMap 'model-registry-operator-parameters' not found in"
f" namespace '{py_config['applications_namespace']}'"
)

try:
return config_map.instance.data["IMAGES_JOBS_ASYNC_UPLOAD"]
except KeyError as e:
raise KeyError(f"Key 'IMAGES_JOBS_ASYNC_UPLOAD' not found in ConfigMap data: {e}") from e


@pytest.fixture(scope="class")
def model_sync_async_job(
admin_client: DynamicClient,
Expand Down
36 changes: 36 additions & 0 deletions tests/model_registry/model_registry/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
from kubernetes.dynamic.exceptions import ResourceNotFoundError
from model_registry import ModelRegistry as ModelRegistryClient
from model_registry.types import RegisteredModel
from ocp_resources.config_map import ConfigMap
from ocp_resources.deployment import Deployment
from ocp_resources.namespace import Namespace
from ocp_resources.pod import Pod
Expand All @@ -17,6 +18,7 @@
from ocp_resources.service_account import ServiceAccount
from pyhelper_utils.shell import run_command
from pytest import FixtureRequest
from pytest_testconfig import config as py_config

from tests.model_registry.constants import (
MODEL_REGISTRY_POD_FILTER,
Expand Down Expand Up @@ -236,3 +238,37 @@ def mr_access_role_binding(
LOGGER.info(f"RoleBinding {binding.name} created successfully.")
yield binding
LOGGER.info(f"RoleBinding {binding.name} deletion initiated by context manager.")


@pytest.fixture(scope="class")
def async_upload_image(admin_client: DynamicClient) -> str:
"""
Get the async upload image dynamically from the model-registry-operator-parameters ConfigMap.

This fetches the image from the cluster at runtime instead of using a hardcoded value.

Args:
admin_client: Kubernetes client for resource access

Returns:
str: The async upload image URL from the ConfigMap

Raises:
KeyError: If the ConfigMap or the required key doesn't exist
"""
config_map = ConfigMap(
client=admin_client,
name="model-registry-operator-parameters",
namespace=py_config["applications_namespace"],
)

if not config_map.exists:
raise ResourceNotFoundError(
f"ConfigMap 'model-registry-operator-parameters' not found in"
f" namespace '{py_config['applications_namespace']}'"
)

try:
return config_map.instance.data["IMAGES_JOBS_ASYNC_UPLOAD"]
except KeyError as e:
raise KeyError(f"Key 'IMAGES_JOBS_ASYNC_UPLOAD' not found in ConfigMap data: {e}") from e
Loading
Loading