Skip to content

Commit 4504f96

Browse files
committed
refactor(workbenches): extract check for internal image registry presence
the check for the internal image registry presence on the cluster is extracted from the fixture to a function for common use
1 parent e890b33 commit 4504f96

File tree

2 files changed

+21
-21
lines changed

2 files changed

+21
-21
lines changed

tests/workbenches/conftest.py

Lines changed: 4 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -13,11 +13,11 @@
1313
from ocp_resources.persistent_volume_claim import PersistentVolumeClaim
1414
from ocp_resources.route import Route
1515
from ocp_resources.notebook import Notebook
16-
from ocp_resources.config_imageregistry_operator_openshift_io import Config
1716

1817
from utilities.constants import Labels
1918
from utilities import constants
2019
from utilities.constants import INTERNAL_IMAGE_REGISTRY_PATH
20+
from utilities.infra import check_internal_image_registry_available
2121

2222
LOGGER = get_logger(name=__name__)
2323

@@ -38,25 +38,6 @@ def users_persistent_volume_claim(
3838
yield pvc
3939

4040

41-
@pytest.fixture(scope="function")
42-
def internal_image_registry(
43-
admin_client: DynamicClient,
44-
) -> Generator[bool, None, None]:
45-
"""Check if internal image registry is available by checking the imageregistry config managementState"""
46-
try:
47-
# Access the imageregistry.operator.openshift.io/v1 Config resource named "cluster"
48-
config_instance = Config(client=admin_client, name="cluster")
49-
50-
management_state = config_instance.instance.spec.get("managementState", "").lower()
51-
is_available = management_state == "managed"
52-
53-
LOGGER.info(f"Image registry management state: {management_state}, available: {is_available}")
54-
yield is_available
55-
except (ResourceNotFoundError, Exception) as e:
56-
LOGGER.warning(f"Failed to check image registry config: {e}")
57-
yield False
58-
59-
6041
@pytest.fixture(scope="function")
6142
def minimal_image() -> Generator[str, None, None]:
6243
"""Provides a full image name of a minimal workbench image"""
@@ -68,7 +49,6 @@ def minimal_image() -> Generator[str, None, None]:
6849
def default_notebook(
6950
request: pytest.FixtureRequest,
7051
admin_client: DynamicClient,
71-
internal_image_registry: bool,
7252
minimal_image: str,
7353
) -> Generator[Notebook, None, None]:
7454
"""Returns a new Notebook CR for a given namespace, name, and image"""
@@ -85,6 +65,9 @@ def default_notebook(
8565
username = get_username(dyn_client=admin_client)
8666
assert username, "Failed to determine username from the cluster"
8767

68+
# Check internal image registry availability
69+
internal_image_registry = check_internal_image_registry_available(admin_client=admin_client)
70+
8871
# Set the image path based on internal image registry status
8972
minimal_image_path = (
9073
f"{INTERNAL_IMAGE_REGISTRY_PATH}/{py_config['applications_namespace']}/{minimal_image}"

utilities/infra.py

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@
2727
from ocp_resources.catalog_source import CatalogSource
2828
from ocp_resources.cluster_service_version import ClusterServiceVersion
2929
from ocp_resources.config_map import ConfigMap
30+
from ocp_resources.config_imageregistry_operator_openshift_io import Config
3031
from ocp_resources.console_cli_download import ConsoleCLIDownload
3132
from ocp_resources.data_science_cluster import DataScienceCluster
3233
from ocp_resources.deployment import Deployment
@@ -1237,3 +1238,19 @@ def download_oc_console_cli(tmpdir: LocalPath) -> str:
12371238
binary_path = os.path.join(tmpdir, extracted_filenames[0])
12381239
os.chmod(binary_path, stat.S_IRUSR | stat.S_IXUSR)
12391240
return binary_path
1241+
1242+
1243+
def check_internal_image_registry_available(admin_client: DynamicClient) -> bool:
1244+
"""Check if internal image registry is available by checking the imageregistry config managementState"""
1245+
try:
1246+
# Access the imageregistry.operator.openshift.io/v1 Config resource named "cluster"
1247+
config_instance = Config(client=admin_client, name="cluster")
1248+
1249+
management_state = config_instance.instance.spec.get("managementState", "").lower()
1250+
is_available = management_state == "managed"
1251+
1252+
LOGGER.info(f"Image registry management state: {management_state}, available: {is_available}")
1253+
return is_available
1254+
except (ResourceNotFoundError, Exception) as e:
1255+
LOGGER.warning(f"Failed to check image registry config: {e}")
1256+
return False

0 commit comments

Comments
 (0)