net, evpn: add source provider migration test

libs/vm/spec.py

@@ -77,6 +77,7 @@ class Interface:
     passtBinding: dict[Any, Any] | None = None  # noqa: N815
     binding: NetBinding | None = None
     state: str | None = None
+    macAddress: str | None = None  # noqa: N815
 
 
 @dataclass

tests/network/bgp/evpn/conftest.py

@@ -11,11 +11,16 @@
 from ocp_resources.resource import ResourceEditor
 from ocp_resources.vtep import VTEP
 
-from libs.net.ip import random_ipv4_address, random_ipv6_address
+from libs.net.ip import random_ipv4_address
 from libs.net.traffic_generator import TcpServer
-from libs.net.udn import UDN_BINDING_DEFAULT_PLUGIN_NAME, create_udn_namespace
+from libs.net.udn import UDN_BINDING_DEFAULT_PLUGIN_NAME, create_udn_namespace, 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 Devices, Metadata
 from libs.vm.vm import BaseVirtualMachine
 from tests.network.bgp.evpn.libevpn import (
+    CUDN_EVPN_SUBNET_IPV6,
+    EVPN_CUDN_NET_SEED,
     EndpointTcpClient,
     EvpnEndpoint,
     cudn_evpn_subnets,
@@ -41,8 +46,9 @@
 EVPN_ADVERTISE_LABEL: Final[dict] = {"advertise": "evpn"}
 APP_EVPN_CUDN_LABEL: Final[dict] = {**EVPN_ADVERTISE_LABEL, "app": "cudn-evpn"}
 CUDN_EVPN_BGP_LABEL: Final[dict] = {"cudn-bgp": "evpn"}
-EXTERNAL_L2_ENDPOINT_IPV4: Final[str] = f"{random_ipv4_address(net_seed=5, host_address=250)}/24"
-EXTERNAL_L2_ENDPOINT_IPV6: Final[str] = f"{random_ipv6_address(net_seed=5, host_address=250)}/64"
+EXTERNAL_L2_ENDPOINT_IPV4: Final[str] = f"{random_ipv4_address(net_seed=EVPN_CUDN_NET_SEED, host_address=250)}/24"
+EXTERNAL_L2_ENDPOINT_IPV6: Final[str] = f"{ipaddress.ip_network(CUDN_EVPN_SUBNET_IPV6, strict=False)[250]}/64"
+EXTERNAL_L2_ENDPOINT_MAC: Final[str] = "02:00:05:00:fa:00"
 EXTERNAL_L3_ENDPOINT_IPV4: Final[str] = "192.168.100.100/24"
 EXTERNAL_L3_ENDPOINT_IPV6: Final[str] = "fd01:1234:5678::64/64"
 EXTERNAL_L3_GATEWAY_IPV4: Final[str] = "192.168.100.1/24"
@@ -233,9 +239,10 @@ def external_l2_endpoint(
         pod=frr_external_pod.pod,
         vni=EVPN_MAC_VRF_VNI,
         endpoint_ips=[EXTERNAL_L2_ENDPOINT_IPV4, EXTERNAL_L2_ENDPOINT_IPV6],
+        mac_address=EXTERNAL_L2_ENDPOINT_MAC,
     )
     yield endpoint
-    teardown_evpn_l2_endpoint(pod=frr_external_pod.pod)
+    teardown_evpn_l2_endpoint(pod=frr_external_pod.pod, vni=EVPN_MAC_VRF_VNI)
 
 
 @pytest.fixture(scope="module")
@@ -269,3 +276,30 @@ def evpn_routed_l3_active_connections(
 ) -> Generator[list[tuple[EndpointTcpClient, TcpServer]]]:
     with evpn_workloads_active_connections(endpoint=external_l3_endpoint, vm=vm_evpn_target) as connections:
         yield connections
+
+
+@pytest.fixture()
+def vm_source_provider(
+    external_l2_endpoint: EvpnEndpoint,
+    namespace_evpn: Namespace,
+    cudn_evpn_layer2: libcudn.ClusterUserDefinedNetwork,
+    admin_client: DynamicClient,
+    frr_external_pod: ExternalFrrPodInfo,
+) -> Generator[BaseVirtualMachine]:
+    spec = base_vmspec()
+    network_name = "udn-network"
+    iface, network = udn_primary_network(name=network_name, binding=UDN_BINDING_DEFAULT_PLUGIN_NAME)
+    iface.macAddress = external_l2_endpoint.mac_address
+    spec.template.spec.domain.devices = Devices(interfaces=[iface])
+    spec.template.spec.networks = [network]
+    spec.template.metadata = Metadata(
+        labels=EXTERNAL_FRR_POD_LABEL,
+        annotations={"network.kubevirt.io/addresses": json.dumps({network_name: external_l2_endpoint.ip_addresses})},
+    )
+    label, *_ = EXTERNAL_FRR_POD_LABEL.items()
+    spec.template.spec.affinity = new_pod_anti_affinity(
+        label=label, namespaces=[frr_external_pod.pod.namespace, namespace_evpn.name]
+    )
+
+    with fedora_vm(namespace=namespace_evpn.name, name="vm-source-provider", client=admin_client, spec=spec) as vm:
+        yield vm

tests/network/bgp/evpn/libevpn.py

@@ -24,8 +24,9 @@
 
 LOGGER = logging.getLogger(__name__)
 
-CUDN_EVPN_SUBNET_IPV4: str = f"{random_ipv4_address(net_seed=5, host_address=0)}/24"
-CUDN_EVPN_SUBNET_IPV6: str = f"{random_ipv6_address(net_seed=5, host_address=0)}/64"
+EVPN_CUDN_NET_SEED: int = 5
+CUDN_EVPN_SUBNET_IPV4: str = f"{random_ipv4_address(net_seed=EVPN_CUDN_NET_SEED, host_address=0)}/24"
+CUDN_EVPN_SUBNET_IPV6: str = f"{random_ipv6_address(net_seed=EVPN_CUDN_NET_SEED, host_address=0)}/64"
 
 _BRIDGE_NAME: str = "br0"
 _VXLAN_NAME: str = "vxlan0"
@@ -51,6 +52,7 @@ class EvpnEndpoint:
     pod: Pod
     ip_addresses: list[str]
     netns_name: str
+    mac_address: str | None = None
 
 
 class EndpointTcpClient(PodTcpClient):
@@ -143,6 +145,7 @@ def deploy_evpn_l2_endpoint(
     pod: Pod,
     vni: int,
     endpoint_ips: list[str],
+    mac_address: str | None = None,
 ) -> EvpnEndpoint:
     """Creates a stretched L2 endpoint on the shared SVD bridge.
 
@@ -155,32 +158,46 @@ def deploy_evpn_l2_endpoint(
         pod: The FRR pod hosting the endpoint.
         vni: MAC-VRF VNI (must match CUDN's macVRF VNI).
         endpoint_ips: IPs with prefix length (e.g. ["10.0.5.250/24", "fd00::fa/64"]).
+        mac_address: Explicit MAC for the endpoint interface (locally-administered).
 
     Returns:
         EvpnEndpoint.
     """
-    commands = _build_l2_endpoint_commands(vni=vni, endpoint_ips=endpoint_ips)
+    commands = _build_l2_endpoint_commands(vni=vni, endpoint_ips=endpoint_ips, mac_address=mac_address)
     for command in commands:
         pod.execute(command=shlex.split(command), container=NET_TOOLS_CONTAINER_NAME)
 
     bare_ips = [ip.split("/")[0] for ip in endpoint_ips]
     LOGGER.info(f"EVPN L2 endpoint deployed: {bare_ips} in namespace {_L2_ENDPOINT_NETNS}")
 
-    return EvpnEndpoint(pod=pod, ip_addresses=bare_ips, netns_name=_L2_ENDPOINT_NETNS)
+    return EvpnEndpoint(pod=pod, ip_addresses=bare_ips, netns_name=_L2_ENDPOINT_NETNS, mac_address=mac_address)
 
 
-def teardown_evpn_l2_endpoint(pod: Pod) -> None:
-    """Removes the EVPN L2 endpoint (netns, veth) from the FRR pod."""
+def teardown_evpn_l2_endpoint(pod: Pod, vni: int) -> None:
+    """Removes the EVPN L2 endpoint (netns, veth, VLAN/VNI mappings) from the FRR pod.
+
+    Args:
+        pod: The FRR pod hosting the endpoint.
+        vni: MAC-VRF VNI used during deployment.
+    """
     for cmd in [
         f"ip netns delete {_L2_ENDPOINT_NETNS}",
         f"ip link delete {_L2_VETH_POD_SIDE}",
+        f"bridge vlan del dev {_VXLAN_NAME} vid {_L2_VID} tunnel_info id {vni}",
+        f"bridge vni del dev {_VXLAN_NAME} vni {vni}",
+        f"bridge vlan del dev {_VXLAN_NAME} vid {_L2_VID}",
+        f"bridge vlan del dev {_BRIDGE_NAME} vid {_L2_VID} self",
     ]:
         pod.execute(command=shlex.split(cmd), container=NET_TOOLS_CONTAINER_NAME, ignore_rc=True)
 
     LOGGER.info(f"EVPN L2 endpoint removed: namespace={_L2_ENDPOINT_NETNS}")
 
 
-def _build_l2_endpoint_commands(vni: int, endpoint_ips: list[str]) -> list[str]:
+def _build_l2_endpoint_commands(
+    vni: int,
+    endpoint_ips: list[str],
+    mac_address: str | None = None,
+) -> list[str]:
     return [
         f"bridge vlan add dev {_BRIDGE_NAME} vid {_L2_VID} self",
         f"bridge vlan add dev {_VXLAN_NAME} vid {_L2_VID}",
@@ -193,6 +210,11 @@ def _build_l2_endpoint_commands(vni: int, endpoint_ips: list[str]) -> list[str]:
         f"ip netns add {_L2_ENDPOINT_NETNS}",
         f"ip link set {_L2_VETH_EP_SIDE} netns {_L2_ENDPOINT_NETNS}",
         *(f"ip netns exec {_L2_ENDPOINT_NETNS} ip addr add {ip} dev {_L2_VETH_EP_SIDE}" for ip in endpoint_ips),
+        *(
+            [f"ip netns exec {_L2_ENDPOINT_NETNS} ip link set dev {_L2_VETH_EP_SIDE} address {mac_address}"]
+            if mac_address
+            else []
+        ),
         f"ip netns exec {_L2_ENDPOINT_NETNS} ip link set {_L2_VETH_EP_SIDE} up",
         f"ip netns exec {_L2_ENDPOINT_NETNS} ip link set lo up",
     ]

tests/network/bgp/evpn/test_evpn_connectivity.py

@@ -20,11 +20,21 @@
 
 import pytest
 
+from libs.net.ip import random_ipv4_address, random_ipv6_address
 from libs.net.traffic_generator import active_tcp_connections, is_tcp_connection
 from libs.net.vmspec import lookup_primary_network
-from tests.network.bgp.evpn.libevpn import assert_evpn_workloads_connectivity, evpn_workloads_active_connections
+from tests.network.bgp.evpn.libevpn import (
+    EVPN_CUDN_NET_SEED,
+    assert_evpn_workloads_connectivity,
+    deploy_evpn_l2_endpoint,
+    evpn_workloads_active_connections,
+    teardown_evpn_l2_endpoint,
+)
 from utilities.virt import migrate_vm_and_verify
 
+_L2_ENDPOINT_IPV4: str = f"{random_ipv4_address(net_seed=EVPN_CUDN_NET_SEED, host_address=249)}/24"
+_L2_ENDPOINT_IPV6: str = f"{random_ipv6_address(net_seed=EVPN_CUDN_NET_SEED, host_address=249)}/64"
+
 pytestmark = [
     pytest.mark.bgp,
     pytest.mark.ipv4,
@@ -176,7 +186,15 @@ def test_connectivity_after_udn_vm_cold_reboot(
 
 
 @pytest.mark.polarion("CNV-15233")
-def test_source_provider_migration():
+@pytest.mark.order("last")
+def test_source_provider_migration(
+    external_l3_endpoint,
+    cudn_evpn_layer2,
+    vm_source_provider,
+    vm_evpn_target,
+    frr_external_pod,
+    subtests,
+):
     """
     Scenario emulates a migration of an external workload (Source Provider) into the OCP cluster as a CUDN VM,
     while preserving its IP and MAC addresses, and maintaining connectivity.
@@ -185,6 +203,7 @@ def test_source_provider_migration():
     - External Source Provider L2 and L3 endpoints.
     - Running connectivity reference VM with a primary EVPN-enabled CUDN.
     - TCP connectivity exists between the connectivity reference VM and the external L2 and L3 endpoints.
+      Precondition is verified in preceding tests.
 
     Steps:
     1. Shut down/remove the external L2 endpoint.
@@ -194,6 +213,23 @@ def test_source_provider_migration():
     Expected:
     - New connections are established after new UDN VM deployment.
     """
+    mac_vrf_vni = cudn_evpn_layer2.instance.spec.network.evpn.macVRF.vni
+
+    teardown_evpn_l2_endpoint(pod=frr_external_pod.pod, vni=mac_vrf_vni)
 
+    vm_source_provider.start(wait=True)
+    vm_source_provider.wait_for_agent_connected()
 
-test_source_provider_migration.__test__ = False
+    new_l2_endpoint = deploy_evpn_l2_endpoint(
+        pod=frr_external_pod.pod,
+        vni=mac_vrf_vni,
+        endpoint_ips=[_L2_ENDPOINT_IPV4, _L2_ENDPOINT_IPV6],
+    )
+
+    assert_evpn_workloads_connectivity(
+        target_vm=vm_evpn_target,
+        ref_vm=vm_source_provider,
+        l2_endpoint=new_l2_endpoint,
+        l3_endpoint=external_l3_endpoint,
+        subtests=subtests,
+    )