CherryPicked: [cnv-4.22] [net] Multi-arch tests with VMs with primary UDN interfaces

tests/network/libs/vm_factory.py

@@ -5,6 +5,7 @@
 from libs.net.udn import udn_primary_network
 from libs.vm.affinity import new_pod_anti_affinity
 from libs.vm.factory import base_vmspec, fedora_vm
+from libs.vm.spec import Affinity
 from libs.vm.vm import BaseVirtualMachine
 
 
@@ -15,16 +16,38 @@ def udn_vm(
     binding: str,
     template_labels: dict | None = None,
     anti_affinity_namespaces: list[str] | None = None,
+    affinity: Affinity | None = None,
+    architecture: str | None = None,
 ) -> BaseVirtualMachine:
+    """Create a Fedora VM connected to a primary UDN using the specified binding.
+
+    Args:
+        namespace_name: Namespace in which the VM will be created.
+        name: Name of the VM.
+        client: Kubernetes dynamic client.
+        binding: UDN binding plugin name (e.g. UDN_BINDING_DEFAULT_PLUGIN_NAME).
+        template_labels: Optional labels to add to the VM pod template, also used as anti-affinity key.
+        anti_affinity_namespaces: Optional namespaces to scope the pod anti-affinity rule.
+        affinity: Optional explicit affinity rules. When set, takes precedence over auto-generated anti-affinity from template_labels.
+        architecture: Optional specific desired architecture of the target VM image.
+
+    Returns:
+        Configured BaseVirtualMachine object (not yet started).
+    """
     spec = base_vmspec()
+    if architecture is not None:
+        spec.template.spec.architecture = architecture
     iface, network = udn_primary_network(name="udn-primary", binding=binding)
     spec.template.spec.domain.devices.interfaces = [iface]  # type: ignore
     spec.template.spec.networks = [network]
+    if affinity is not None:
+        spec.template.spec.affinity = affinity
     if template_labels:
         spec.template.metadata.labels = spec.template.metadata.labels or {}  # type: ignore
         spec.template.metadata.labels.update(template_labels)  # type: ignore
-        # Use the first label key and first value as the anti-affinity label to use:
-        label, *_ = template_labels.items()
-        spec.template.spec.affinity = new_pod_anti_affinity(label=label, namespaces=anti_affinity_namespaces)
+        if affinity is None:
+            # Use the first label key and first value as the anti-affinity label to use:
+            label, *_ = template_labels.items()
+            spec.template.spec.affinity = new_pod_anti_affinity(label=label, namespaces=anti_affinity_namespaces)
 
     return fedora_vm(namespace=namespace_name, name=name, client=client, spec=spec)

tests/network/user_defined_network/conftest.py

@@ -1,8 +1,16 @@
+from collections.abc import Generator
+
 import pytest
+from kubernetes.dynamic import DynamicClient
 from ocp_resources.user_defined_network import Layer2UserDefinedNetwork
+from utilities.constants.architecture import AMD_64, ARM_64
 
 from libs.net.ip import random_ipv4_address
+from libs.net.udn import UDN_BINDING_DEFAULT_PLUGIN_NAME
 from libs.vm import affinity
+from libs.vm.oper import run_vms
+from libs.vm.vm import BaseVirtualMachine
+from tests.network.libs.vm_factory import udn_vm
 from utilities.infra import create_ns
 
 
@@ -35,3 +43,53 @@ def namespaced_layer2_user_defined_network(admin_client, udn_namespace):
 @pytest.fixture(scope="module")
 def udn_affinity_label():
     return affinity.new_label(key_prefix="udn")
+
+
+@pytest.fixture(scope="class")
+def arm64_udn_vm(
+    admin_client: DynamicClient,
+    namespaced_layer2_user_defined_network: Layer2UserDefinedNetwork,
+) -> Generator[BaseVirtualMachine]:
+    """
+    ARM64 VM with UDN as primary interface.
+    """
+    with udn_vm(
+        namespace_name=namespaced_layer2_user_defined_network.namespace,
+        name="arm64-udn-vm",
+        client=admin_client,
+        binding=UDN_BINDING_DEFAULT_PLUGIN_NAME,
+        architecture=ARM_64,
+    ) as vm:
+        yield vm
+
+
+@pytest.fixture(scope="class")
+def amd64_udn_vm(
+    admin_client: DynamicClient,
+    namespaced_layer2_user_defined_network: Layer2UserDefinedNetwork,
+) -> Generator[BaseVirtualMachine]:
+    """
+    AMD64 VM with UDN as primary interface.
+    """
+    with udn_vm(
+        namespace_name=namespaced_layer2_user_defined_network.namespace,
+        name="amd64-udn-vm",
+        client=admin_client,
+        binding=UDN_BINDING_DEFAULT_PLUGIN_NAME,
+        architecture=AMD_64,
+    ) as vm:
+        yield vm
+
+
+@pytest.fixture(scope="class")
+def running_amd_and_arm_vms(
+    amd64_udn_vm: BaseVirtualMachine, arm64_udn_vm: BaseVirtualMachine
+) -> tuple[BaseVirtualMachine, BaseVirtualMachine]:
+    """
+    Start AMD64 and ARM64 UDN VMs in parallel.
+
+    Returns:
+        Tuple of (amd64_vm, arm64_vm) both running with agent connected.
+    """
+    amd64_vm, arm64_vm = run_vms(vms=(amd64_udn_vm, arm64_udn_vm))
+    return amd64_vm, arm64_vm

tests/network/user_defined_network/test_user_defined_network_multiarch.py

@@ -7,6 +7,9 @@
 
 import pytest
 
+from libs.net.traffic_generator import client_server_active_connection, is_tcp_connection
+from libs.net.vmspec import lookup_primary_network
+
 
 @pytest.mark.multiarch
 @pytest.mark.single_nic
@@ -25,7 +28,7 @@ class TestMultiArchUdn:
     """
 
     @pytest.mark.polarion("CNV-15942")
-    def test_udn_connectivity_amd_client_to_arm_server(self):
+    def test_udn_connectivity_amd_client_to_arm_server(self, running_amd_and_arm_vms):
         """
         Test UDN connectivity between VMs on different architectures - client on AMD, server on ARM.
 
@@ -35,9 +38,16 @@ def test_udn_connectivity_amd_client_to_arm_server(self):
         Expected:
             - TCP connection succeeds
         """
+        amd64_udn_vm, arm64_udn_vm = running_amd_and_arm_vms
+        with client_server_active_connection(
+            client_vm=amd64_udn_vm,
+            server_vm=arm64_udn_vm,
+            spec_logical_network=lookup_primary_network(vm=arm64_udn_vm).name,
+        ) as (client, server):
+            assert is_tcp_connection(server=server, client=client)
 
     @pytest.mark.polarion("CNV-15970")
-    def test_udn_connectivity_arm_client_to_amd_server(self):
+    def test_udn_connectivity_arm_client_to_amd_server(self, running_amd_and_arm_vms):
         """
         Test UDN connectivity between VMs on different architectures - client on ARM, server on AMD.
 
@@ -47,6 +57,10 @@ def test_udn_connectivity_arm_client_to_amd_server(self):
         Expected:
             - TCP connection succeeds
         """
-
-    test_udn_connectivity_arm_client_to_amd_server.__test__ = False
-    test_udn_connectivity_amd_client_to_arm_server.__test__ = False
+        amd64_udn_vm, arm64_udn_vm = running_amd_and_arm_vms
+        with client_server_active_connection(
+            client_vm=arm64_udn_vm,
+            server_vm=amd64_udn_vm,
+            spec_logical_network=lookup_primary_network(vm=amd64_udn_vm).name,
+        ) as (client, server):
+            assert is_tcp_connection(server=server, client=client)