net, tests, udn: Add VM-to-pod connectivity test

libs/net/udn.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from collections.abc import Generator
 from typing import Final
 

tests/network/user_defined_network/conftest.py

@@ -1,7 +1,9 @@
 from collections.abc import Generator
+from typing import TYPE_CHECKING
 
 import pytest
 from kubernetes.dynamic import DynamicClient
+from ocp_resources.pod import Pod
 from ocp_resources.user_defined_network import Layer2UserDefinedNetwork
 
 from libs.net.ip import random_ipv4_address
@@ -13,8 +15,12 @@
 from libs.vm.vm import BaseVirtualMachine
 from tests.network.libs.vm_factory import udn_vm
 from utilities.constants.architecture import AMD_64, ARM_64
+from utilities.constants.networking import POD_CONTAINER_SPEC
 from utilities.infra import create_ns
 
+if TYPE_CHECKING:
+    from ocp_resources.namespace import Namespace
+
 
 @pytest.fixture(scope="module")
 def udn_namespace(admin_client):
@@ -145,3 +151,19 @@ def running_amd_and_arm_vms(
     """
     amd64_vm, arm64_vm = run_vms(vms=(amd64_udn_vm, arm64_udn_vm))
     return amd64_vm, arm64_vm
+
+
+@pytest.fixture(scope="class")
+def udn_pod(
+    admin_client: DynamicClient,
+    udn_namespace: Namespace,
+    namespaced_layer2_user_defined_network: Layer2UserDefinedNetwork,
+) -> Generator[Pod]:
+    with Pod(
+        name="udn-pod",
+        namespace=udn_namespace.name,
+        containers=[{**POD_CONTAINER_SPEC, "name": "udn-container"}],
+        client=admin_client,
+    ) as pod:
+        pod.wait_for_status(status=Pod.Status.RUNNING)
+        yield pod

tests/network/user_defined_network/libudn.py

@@ -0,0 +1,28 @@
+from __future__ import annotations
+
+import json
+from typing import TYPE_CHECKING
+
+from ocp_resources.resource import Resource
+
+from libs.net.vmspec import IpNotFound
+
+if TYPE_CHECKING:
+    from ocp_resources.pod import Pod
+
+
+def lookup_default_pod_ip(pod: Pod) -> str:
+    """Return the default network IP address of a pod.
+
+    Args:
+        pod: The pod to query.
+
+    Returns:
+        The IP address from the default network attachment.
+    """
+    network_status_annotation = f"{Resource.ApiGroup.K8S_V1_CNI_CNCF_IO}/network-status"
+    network_status = json.loads(pod.instance.metadata.annotations[network_status_annotation])
+    default_entry = next(entry for entry in network_status if entry.get("default"))
+    if not default_entry["ips"]:
+        raise IpNotFound(f"No IPs assigned to default UDN entry on pod {pod.name}")
+    return str(default_entry["ips"][0])

tests/network/user_defined_network/test_user_defined_network.py

@@ -8,6 +8,7 @@
 
 from libs.net.traffic_generator import is_tcp_connection
 from libs.net.vmspec import lookup_iface_status_ip, lookup_primary_network
+from tests.network.user_defined_network.libudn import lookup_default_pod_ip
 from utilities.constants.networking import PUBLIC_DNS_SERVER_IP
 from utilities.constants.pytest import QUARANTINED
 from utilities.constants.timeouts import TIMEOUT_1MIN
@@ -79,3 +80,9 @@ def test_connectivity_is_preserved_during_server_live_migration(
     ):
         migrate_vm_and_verify(vm=server.vm, client=admin_client)
         assert is_tcp_connection(server=server, client=client)
+
+    @pytest.mark.polarion("CNV-11432")
+    def test_vm_to_pod_connectivity_on_udn(self, vma_udn, udn_pod):
+        """Jira: https://redhat.atlassian.net/browse/CNV-51404 # <skip-jira-utils-check>"""
+        pod_ip = lookup_default_pod_ip(pod=udn_pod)
+        vma_udn.console(commands=[f"ping -c 3 {pod_ip}"], timeout=TIMEOUT_1MIN)