[Storage/Network] Specify architecture for internal HTTP server for multiarch clusters

libs/vm/factory.py

@@ -1,12 +1,11 @@
 from __future__ import annotations
 
 from kubernetes.dynamic import DynamicClient
-from pytest_testconfig import config as py_config
 
 from libs.vm.spec import CPU, Devices, Domain, Memory, Metadata, Template, VMISpec, VMSpec
 from libs.vm.vm import BaseVirtualMachine, container_image, containerdisk_storage
 from utilities import constants as constants_module
-from utilities.constants.architecture import MULTIARCH
+from utilities.architecture import get_multiarch_cpu_arch
 from utilities.constants.images import OS_FLAVOR_FEDORA, ArchImages
 
 
@@ -42,8 +41,7 @@ def _fill_vm_spec_defaults(spec: VMSpec | None) -> VMSpec:
 
     vmi_spec = spec.template.spec
 
-    if not vmi_spec.architecture and py_config.get("cluster_type") == MULTIARCH:
-        cpu_arch = py_config.get("cpu_arch")
+    if not vmi_spec.architecture and (cpu_arch := get_multiarch_cpu_arch()):
         vmi_spec.architecture = cpu_arch
     vmi_spec.domain.devices = vmi_spec.domain.devices or Devices(rng={})
     vmi_spec.domain.devices.disks = vmi_spec.domain.devices.disks or []

tests/storage/conftest.py

@@ -3,6 +3,7 @@
 """
 
 import base64
+import copy
 import ipaddress
 import logging
 import os
@@ -42,9 +43,10 @@
     is_hpp_cr_legacy,
 )
 from tests.utils import create_cirros_vm
+from utilities.architecture import get_multiarch_cpu_arch
 from utilities.artifactory import get_artifactory_config_map, get_artifactory_secret
 from utilities.constants import Images
-from utilities.constants.cluster import CNV_TEST_SERVICE_ACCOUNT
+from utilities.constants.cluster import CNV_TEST_SERVICE_ACCOUNT, KUBERNETES_ARCH_LABEL
 from utilities.constants.components import CDI_OPERATOR, CDI_UPLOADPROXY
 from utilities.constants.images import OS_FLAVOR_FEDORA, OS_FLAVOR_RHEL
 from utilities.constants.instance_types import PREFERENCE_STR, U1_SMALL
@@ -167,11 +169,15 @@ def internal_http_deployment(cnv_tests_utilities_namespace, admin_client):
     Deploy internal HTTP server Deployment into the cnv_tests_utilities_namespace namespace.
     This Deployment deploys a pod that runs an HTTP server
     """
+    template = copy.deepcopy(INTERNAL_HTTP_TEMPLATE)
+    if cpu_arch := get_multiarch_cpu_arch():
+        template["spec"]["nodeSelector"] = {KUBERNETES_ARCH_LABEL: cpu_arch}
+
     with Deployment(
         name="internal-http",
         namespace=cnv_tests_utilities_namespace.name,
         selector=INTERNAL_HTTP_SELECTOR,
-        template=INTERNAL_HTTP_TEMPLATE,
+        template=template,
         replicas=1,
         client=admin_client,
     ) as dep:

utilities/architecture.py

@@ -3,6 +3,7 @@
 from functools import cache
 
 from ocp_resources.node import Node
+from pytest_testconfig import config as py_config
 
 from utilities.cluster import cache_admin_client
 from utilities.exceptions import UnsupportedCPUArchitectureError
@@ -42,3 +43,23 @@ def get_cluster_architecture() -> set[str]:
             "Cluster architecture could not be determined (no nodes found and env var unset)."
         )
     return cluster_archs
+
+
+def get_multiarch_cpu_arch() -> str | None:
+    """
+    Returns the target CPU architecture on multiarch clusters with a single --cpu-arch.
+
+    When --cpu-arch=ARCH1,ARCH2, py_config["cpu_arch"] is never set, so this returns None.
+
+    Returns:
+        str | None: The CPU architecture string (e.g. "arm64") if running on a multiarch
+            cluster with a single target arch, None otherwise.
+    """
+    # Lazy import to avoid circular dependency
+    # TODO: remove when/if utilities modules are refactored
+    from utilities.constants.architecture import MULTIARCH  # noqa: PLC0415
+
+    cpu_arch = py_config.get("cpu_arch")
+    if cpu_arch and py_config.get("cluster_type") == MULTIARCH:
+        return cpu_arch
+    return None

utilities/storage.py

@@ -34,9 +34,9 @@
 import utilities.infra
 import utilities.virt as virt_util
 from utilities import console
+from utilities.architecture import get_multiarch_cpu_arch
 from utilities.artifactory import get_test_artifact_server_url
 from utilities.constants import Images
-from utilities.constants.architecture import MULTIARCH
 from utilities.constants.components import HPP_POOL
 from utilities.constants.images import OS_FLAVOR_WINDOWS
 from utilities.constants.networking import POD_CONTAINER_SPEC
@@ -139,10 +139,7 @@ def construct_datavolume_source_dict(
         source_spec: dict[str, Any] = {"http": {"url": url}}
     elif source == "registry":
         registry_spec: dict[str, Any] = {"url": url}
-        # For multi-arch cluster and single --cpu-arch=ARCH, explicitly set the registry platform architecture
-        # For --cpu-arch=ARCH1,ARCH2, py_config["cpu_arch"] is never set
-        cpu_arch = py_config.get("cpu_arch")
-        if cpu_arch and py_config.get("cluster_type") == MULTIARCH:
+        if cpu_arch := get_multiarch_cpu_arch():
             registry_spec["platform"] = {"architecture": cpu_arch}
         source_spec = {"registry": registry_spec}
     elif source == "pvc":

utilities/unittests/test_architecture.py

@@ -7,7 +7,7 @@
 
 import pytest
 
-from utilities.architecture import get_cluster_architecture
+from utilities.architecture import get_cluster_architecture, get_multiarch_cpu_arch
 from utilities.exceptions import UnsupportedCPUArchitectureError
 
 
@@ -165,3 +165,21 @@ def test_get_cluster_architecture_skips_cluster_on_exit_flag(self, mock_node_cla
             assert result == {"amd64"}
             mock_cache_client.assert_not_called()
             mock_node_class.get.assert_not_called()
+
+
+class TestGetMultiarchCpuArch:
+    @patch.dict("utilities.architecture.py_config", {"cpu_arch": "arm64", "cluster_type": "multiarch"})
+    def test_returns_arch_on_multiarch_cluster_with_single_arch(self):
+        assert get_multiarch_cpu_arch() == "arm64"
+
+    @patch.dict("utilities.architecture.py_config", {"cpu_arch": "arm64", "cluster_type": "standard"})
+    def test_returns_none_on_non_multiarch_cluster(self):
+        assert get_multiarch_cpu_arch() is None
+
+    @patch.dict("utilities.architecture.py_config", {"cluster_type": "multiarch"})
+    def test_returns_none_when_cpu_arch_not_set(self):
+        assert get_multiarch_cpu_arch() is None
+
+    @patch.dict("utilities.architecture.py_config", {})
+    def test_returns_none_when_no_config(self):
+        assert get_multiarch_cpu_arch() is None

utilities/unittests/test_storage.py

@@ -107,8 +107,8 @@ def test_registry_source_with_secret_and_cert(self):
             }
         }
 
-    @patch.dict("utilities.storage.py_config", {"cpu_arch": "arm64", "cluster_type": "multiarch"})
-    def test_registry_source_multiarch_with_cpu_arch(self):
+    @patch("utilities.storage.get_multiarch_cpu_arch", return_value="arm64")
+    def test_registry_source_multiarch_with_cpu_arch(self, _mock_get_arch):
         result = construct_datavolume_source_dict(source="registry", url="docker://registry.example.com/image:latest")
         assert result == {
             "registry": {
@@ -117,14 +117,8 @@ def test_registry_source_multiarch_with_cpu_arch(self):
             }
         }
 
-    @patch.dict("utilities.storage.py_config", {"cpu_arch": "arm64", "cluster_type": "standard"})
-    def test_registry_source_non_multiarch_no_platform(self):
-        result = construct_datavolume_source_dict(source="registry", url="docker://registry.example.com/image:latest")
-        assert result == {"registry": {"url": "docker://registry.example.com/image:latest"}}
-        assert "platform" not in result["registry"]
-
-    @patch.dict("utilities.storage.py_config", {"cluster_type": "multiarch"})
-    def test_registry_source_multiarch_no_cpu_arch_no_platform(self):
+    @patch("utilities.storage.get_multiarch_cpu_arch", return_value=None)
+    def test_registry_source_non_multiarch_no_platform(self, _mock_get_arch):
         result = construct_datavolume_source_dict(source="registry", url="docker://registry.example.com/image:latest")
         assert result == {"registry": {"url": "docker://registry.example.com/image:latest"}}
         assert "platform" not in result["registry"]