Move cache_admin_client to be public (#2890)

##### Short description:
Admin Client is used to CRUD cluster resources.
`admin_client` provides this client, and all functions calling
ocp_resources resources must get a client as an attributes.
There are, however, places in the code (i.e in pytest-native fixtures or
during pre-execution hooks) that the `admin_client` cannot be
accessible.
While there are pytest hacks to use the fixture, we should take a
simple, clear approach.

The usage must be very limited, the existing `_cache_admin_client` is
now public to allow re-using it in places described above.

##### More details:

##### What this PR does / why we need it:

##### Which issue(s) this PR fixes:

##### Special notes for reviewer:

##### jira-ticket:
<!-- full-ticket-url needs to be provided. This would add a link to the
pull request to the jira and close it when the pull request is merged
If the task is not tracked by a Jira ticket, just write "NONE".
-->


<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

* **Tests**
* Expanded architecture detection tests with comprehensive coverage for
multiple CPU architectures and node configurations.
  * Updated test infrastructure to support new client caching mechanism.

* **Chores**
* Centralized client management through a new cached utility function
for improved resource efficiency and consistency across codebase.

<sub>✏️ Tip: You can customize this high-level summary in your review
settings.</sub>
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: rnetser

conftest.py

@@ -21,11 +21,11 @@
 from _pytest.reports import CollectReport, TestReport
 from _pytest.runner import CallInfo
 from kubernetes.dynamic.exceptions import ConflictError
-from ocp_resources.resource import get_client
 from pyhelper_utils.shell import run_command
 from pytest import Item
 from pytest_testconfig import config as py_config
 
+import utilities.cluster
 import utilities.infra
 from libs.storage.config import StorageClassConfig
 from utilities.bitwarden import get_cnv_tests_secret_by_name
@@ -792,7 +792,7 @@ def _update_os_related_config():
         py_config["version_explorer_url"] = get_cnv_version_explorer_url(pytest_config=session.config)
         if not session.config.getoption("--skip-artifactory-check"):
             py_config["server_url"] = py_config["server_url"] or get_artifactory_server_url(
-                cluster_host_url=get_client().configuration.host
+                cluster_host_url=utilities.cluster.cache_admin_client().configuration.host
             )
             py_config["servers"] = {
                 name: _server.format(server=py_config["server_url"]) for name, _server in py_config["servers"].items()

tests/conftest.py

@@ -74,6 +74,7 @@
 from tests.utils import download_and_extract_tar, update_cluster_cpu_model
 from utilities.artifactory import get_artifactory_header, get_http_image_url, get_test_artifact_server_url
 from utilities.bitwarden import get_cnv_tests_secret_by_name
+from utilities.cluster import cache_admin_client
 from utilities.constants import (
     AAQ_NAMESPACE_LABEL,
     AMD,
@@ -305,7 +306,7 @@ def admin_client():
     """
     Get DynamicClient
     """
-    return get_client()
+    return cache_admin_client()
 
 
 @pytest.fixture(scope="session")

utilities/architecture.py

@@ -1,7 +1,8 @@
 import os
 
 from ocp_resources.node import Node
-from ocp_resources.resource import get_client
+
+from utilities.cluster import cache_admin_client
 
 
 def get_cluster_architecture() -> str:
@@ -21,7 +22,8 @@ def get_cluster_architecture() -> str:
 
     if not arch:
         # TODO: merge with `get_nodes_cpu_architecture`
-        nodes: list[Node] = list(Node.get(dyn_client=get_client()))
+        # cache_admin_client is used here as this function is used to get the architecture when initialing pytest config
+        nodes: list[Node] = list(Node.get(dyn_client=cache_admin_client()))
         nodes_cpu_arch = {node.labels[KUBERNETES_ARCH_LABEL] for node in nodes}
         arch = next(iter(nodes_cpu_arch))
 

utilities/cluster.py

@@ -0,0 +1,19 @@
+from functools import cache
+
+from kubernetes.dynamic import DynamicClient
+from ocp_resources.resource import get_client
+
+
+@cache
+def cache_admin_client() -> DynamicClient:
+    """Get admin_client once and reuse it
+
+    This usage of this function is limited ONLY in places where `client` cannot be passed as an argument.
+    For example: in pytest native fixtures in conftest.py.
+
+    Returns:
+        DynamicClient: admin_client
+
+    """
+
+    return get_client()

utilities/pytest_matrix_utils.py

@@ -5,12 +5,10 @@ def foo_matrix(matrix):
     return matrix
 """
 
-from functools import cache
-
-from kubernetes.dynamic import DynamicClient
-from ocp_resources.resource import get_client
 from ocp_resources.storage_class import StorageClass
 
+from utilities.cluster import cache_admin_client
+
 
 def snapshot_matrix(matrix):
     matrix_to_return = []
@@ -51,7 +49,7 @@ def hpp_matrix(matrix):
         # Using `get_client` explicitly as this function is dynamically called (like other functions in the module).
         # The other functions do not need a client.
         if (
-            StorageClass(client=_cache_admin_client(), name=[*storage_class][0]).instance.provisioner
+            StorageClass(client=cache_admin_client(), name=[*storage_class][0]).instance.provisioner
             in hpp_sc_provisioners
         ):
             matrix_to_return.append(storage_class)
@@ -75,17 +73,3 @@ def immediate_matrix(matrix):
         if storage_class[storage_class_name]["wffc"] is False:
             matrix_to_return.append(storage_class)
     return matrix_to_return
-
-
-@cache
-def _cache_admin_client() -> DynamicClient:
-    """Get admin_client once and reuse it
-
-    This usage of this function is limited to places where `client` cannot be passed as an argument.
-
-    Returns:
-        DynamicClient: admin_client
-
-    """
-
-    return get_client()

utilities/unittests/test_architecture.py

@@ -3,75 +3,96 @@
 """Unit tests for architecture module"""
 
 import os
-import sys
-from pathlib import Path
 from unittest.mock import MagicMock, patch
 
 import pytest
 
-# Add utilities to Python path for imports
-sys.path.insert(0, str(Path(__file__).parent.parent))
-
-from architecture import get_cluster_architecture
+from utilities.architecture import get_cluster_architecture
 
 
 class TestGetClusterArchitecture:
     """Test cases for get_cluster_architecture function"""
 
-    def test_get_cluster_architecture_from_env(self):
-        """Test getting architecture from environment variable"""
+    def test_get_cluster_architecture_from_env_arm64(self):
+        """Test getting architecture from environment variable - arm64"""
         with patch.dict(os.environ, {"OPENSHIFT_VIRTUALIZATION_TEST_IMAGES_ARCH": "arm64"}):
             result = get_cluster_architecture()
             assert result == "arm64"
 
-    @patch("architecture.Node")
-    @patch("architecture.get_client")
-    def test_get_cluster_architecture_from_nodes_x86_64(self, mock_get_client, mock_node_class):
+    def test_get_cluster_architecture_from_env_x86_64(self):
+        """Test getting architecture from environment variable - x86_64"""
+        with patch.dict(os.environ, {"OPENSHIFT_VIRTUALIZATION_TEST_IMAGES_ARCH": "x86_64"}):
+            result = get_cluster_architecture()
+            assert result == "x86_64"
+
+    def test_get_cluster_architecture_from_env_s390x(self):
+        """Test getting architecture from environment variable - s390x"""
+        with patch.dict(os.environ, {"OPENSHIFT_VIRTUALIZATION_TEST_IMAGES_ARCH": "s390x"}):
+            result = get_cluster_architecture()
+            assert result == "s390x"
+
+    def test_get_cluster_architecture_from_env_amd64_converts_to_x86_64(self):
+        """Test that amd64 from env is converted to x86_64"""
+        with patch.dict(os.environ, {"OPENSHIFT_VIRTUALIZATION_TEST_IMAGES_ARCH": "amd64"}):
+            result = get_cluster_architecture()
+            assert result == "x86_64"
+
+    @patch("utilities.architecture.cache_admin_client")
+    @patch("utilities.architecture.Node")
+    def test_get_cluster_architecture_from_nodes_x86_64(self, mock_node_class, mock_cache_client):
         """Test getting architecture from nodes - x86_64"""
         # Clear env var to force reading from nodes
         with patch.dict(os.environ, {}, clear=True):
-            # Mock node with x86_64 architecture
+            os.environ.pop("OPENSHIFT_VIRTUALIZATION_TEST_IMAGES_ARCH", None)
+
+            # Mock node with amd64 architecture
             mock_node = MagicMock()
             mock_node.labels = {"kubernetes.io/arch": "amd64"}
             mock_node_class.get.return_value = [mock_node]
+            mock_cache_client.return_value = MagicMock()
 
             result = get_cluster_architecture()
 
             # Should convert amd64 to x86_64
             assert result == "x86_64"
             mock_node_class.get.assert_called_once()
+            mock_cache_client.assert_called_once()
 
-    @patch("architecture.Node")
-    @patch("architecture.get_client")
-    def test_get_cluster_architecture_from_nodes_arm64(self, mock_get_client, mock_node_class):
+    @patch("utilities.architecture.cache_admin_client")
+    @patch("utilities.architecture.Node")
+    def test_get_cluster_architecture_from_nodes_arm64(self, mock_node_class, mock_cache_client):
         """Test getting architecture from nodes - arm64"""
         with patch.dict(os.environ, {}, clear=True):
+            os.environ.pop("OPENSHIFT_VIRTUALIZATION_TEST_IMAGES_ARCH", None)
+
             # Mock node with arm64 architecture
             mock_node = MagicMock()
             mock_node.labels = {"kubernetes.io/arch": "arm64"}
             mock_node_class.get.return_value = [mock_node]
+            mock_cache_client.return_value = MagicMock()
 
             result = get_cluster_architecture()
 
             assert result == "arm64"
 
-    @patch("architecture.Node")
-    @patch("architecture.get_client")
-    def test_get_cluster_architecture_from_nodes_s390x(self, mock_get_client, mock_node_class):
+    @patch("utilities.architecture.cache_admin_client")
+    @patch("utilities.architecture.Node")
+    def test_get_cluster_architecture_from_nodes_s390x(self, mock_node_class, mock_cache_client):
         """Test getting architecture from nodes - s390x"""
         with patch.dict(os.environ, {}, clear=True):
+            os.environ.pop("OPENSHIFT_VIRTUALIZATION_TEST_IMAGES_ARCH", None)
+
             # Mock node with s390x architecture
             mock_node = MagicMock()
             mock_node.labels = {"kubernetes.io/arch": "s390x"}
             mock_node_class.get.return_value = [mock_node]
+            mock_cache_client.return_value = MagicMock()
 
             result = get_cluster_architecture()
 
             assert result == "s390x"
 
-    @patch("architecture.Node")
-    @patch("architecture.get_client")
-    def test_get_cluster_architecture_unsupported(self, mock_get_client, mock_node_class):
+    def test_get_cluster_architecture_unsupported(self):
         """Test unsupported architecture raises ValueError"""
         with (
             patch.dict(os.environ, {"OPENSHIFT_VIRTUALIZATION_TEST_IMAGES_ARCH": "unsupported"}),
@@ -82,18 +103,41 @@ def test_get_cluster_architecture_unsupported(self, mock_get_client, mock_node_c
         ):
             get_cluster_architecture()
 
-    @patch("architecture.Node")
-    @patch("architecture.get_client")
-    def test_get_cluster_architecture_multiple_nodes(self, mock_get_client, mock_node_class):
+    @patch("utilities.architecture.cache_admin_client")
+    @patch("utilities.architecture.Node")
+    def test_get_cluster_architecture_multiple_nodes(self, mock_node_class, mock_cache_client):
         """Test getting architecture with multiple nodes of same arch"""
         with patch.dict(os.environ, {}, clear=True):
+            os.environ.pop("OPENSHIFT_VIRTUALIZATION_TEST_IMAGES_ARCH", None)
+
             # Mock multiple nodes with same architecture
             mock_node1 = MagicMock()
             mock_node1.labels = {"kubernetes.io/arch": "amd64"}
             mock_node2 = MagicMock()
             mock_node2.labels = {"kubernetes.io/arch": "amd64"}
             mock_node_class.get.return_value = [mock_node1, mock_node2]
+            mock_cache_client.return_value = MagicMock()
 
             result = get_cluster_architecture()
 
             assert result == "x86_64"
+
+    @patch("utilities.architecture.cache_admin_client")
+    @patch("utilities.architecture.Node")
+    def test_get_cluster_architecture_uses_cache_admin_client(self, mock_node_class, mock_cache_client):
+        """Test that cache_admin_client is used when getting nodes"""
+        with patch.dict(os.environ, {}, clear=True):
+            os.environ.pop("OPENSHIFT_VIRTUALIZATION_TEST_IMAGES_ARCH", None)
+
+            mock_client = MagicMock()
+            mock_cache_client.return_value = mock_client
+
+            mock_node = MagicMock()
+            mock_node.labels = {"kubernetes.io/arch": "amd64"}
+            mock_node_class.get.return_value = [mock_node]
+
+            get_cluster_architecture()
+
+            # Verify cache_admin_client was called and passed to Node.get
+            mock_cache_client.assert_called_once()
+            mock_node_class.get.assert_called_once_with(dyn_client=mock_client)

utilities/unittests/test_pytest_matrix_utils.py

@@ -120,7 +120,7 @@ def test_online_resize_matrix_empty_matrix(self):
 class TestHppMatrix:
     """Test cases for hpp_matrix function"""
 
-    @patch("utilities.pytest_matrix_utils._cache_admin_client")
+    @patch("utilities.infra.cache_admin_client")
     @patch("utilities.pytest_matrix_utils.StorageClass")
     def test_hpp_matrix_with_hpp_provisioner(self, mock_storage_class, mock_cache_admin_client):
         """Test hpp_matrix filters storage classes with HPP provisioner"""
@@ -156,9 +156,7 @@ def storage_class_side_effect(client, name):
         assert len(result) == 1
         assert {"hpp-sc": {"other": "value"}} in result
 
-    @patch("utilities.pytest_matrix_utils._cache_admin_client")
-    @patch("utilities.pytest_matrix_utils.StorageClass")
-    def test_hpp_matrix_empty_matrix(self, mock_storage_class, mock_cache_admin_client):
+    def test_hpp_matrix_empty_matrix(self):
         """Test hpp_matrix with empty matrix"""
         matrix = []