test: Topotest to validate PMSI V4 and V6 tunnel Id

Test 'test_imet' added for V4 and V4 to validate Tunnel ID PMSI

Signed-off-by: Mrinmoy Ghosh <[email protected]>

Author: mrin-g

tests/topotests/bgp_evpn_vxlan_topo1/test_bgp_evpn_vxlan.py

@@ -30,6 +30,7 @@
     evpn_mac_learn_test,
     evpn_mac_test_local_remote,
     evpn_show_vni_json_elide_ifindex,
+    evpn_check_bgp_imet,
 )
 from lib.topogen import Topogen, TopoRouter, get_topogen
 from lib.topolog import logger
@@ -632,6 +633,39 @@ def test_evpn_l3vni_vlan_bridge():
         assert "Type: L3" in output, assertmsg
 
 
+def test_imet():
+    """
+    Verify PMSI tunnel attribute info
+    """
+    tgen = get_topogen()
+    if tgen.routers_have_failure():
+        pytest.skip(tgen.errors)
+
+    dut_name = "PE1"
+    dut = tgen.gears[dut_name]
+    rd = "10.30.30.30:2"
+    prefix = "[3]:[0]:[32]:[10.30.30.30]"
+    pmsi_label = 101
+    pmsi_id = "10.30.30.30"
+    # Check Imet from PE2 to PE1
+    test_fn = partial(evpn_check_bgp_imet, dut, rd, prefix, pmsi_label, pmsi_id)
+    _, result = topotest.run_and_expect(test_fn, None, count=10, wait=3)
+    assertmsg = f"{dut_name} IMET not present/incorrect, result:{result}"
+    assert result is None, assertmsg
+
+    # Check Imet from PE1 to PE2
+    dut_name = "PE2"
+    dut = tgen.gears[dut_name]
+    rd = "10.10.10.10:2"
+    prefix = "[3]:[0]:[32]:[10.10.10.10]"
+    pmsi_label = 101
+    pmsi_id = "10.10.10.10"
+    test_fn = partial(evpn_check_bgp_imet, dut, rd, prefix, pmsi_label, pmsi_id)
+    _, result = topotest.run_and_expect(test_fn, None, count=10, wait=3)
+    assertmsg = f"{dut_name} IMET not present/incorrect, result:{result}"
+    assert result is None, assertmsg
+
+
 def test_remote_neigh_uninstall_on_vxlan_down():
     "Ensure remote neighs are removed when VxLAN if is down"
     tgen = get_topogen()

tests/topotests/bgp_evpn_vxlan_topo1/test_bgp_evpn_vxlan_v6_vtep.py

@@ -34,6 +34,7 @@
     evpn_mac_learn_test,
     evpn_mac_test_local_remote,
     evpn_show_vni_json_elide_ifindex,
+    evpn_check_bgp_imet,
 )
 from lib.topogen import Topogen, TopoRouter, get_topogen
 from lib.topolog import logger
@@ -117,10 +118,12 @@ def setup_module(mod):
     # For all registered routers, load the zebra configuration file
     for rname, router in router_list.items():
         router.load_config(
-            TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra_v6_vtep.conf".format(rname))
+            TopoRouter.RD_ZEBRA,
+            os.path.join(CWD, "{}/zebra_v6_vtep.conf".format(rname)),
         )
         router.load_config(
-            TopoRouter.RD_OSPF6, os.path.join(CWD, "{}/ospfd_v6_vtep.conf".format(rname))
+            TopoRouter.RD_OSPF6,
+            os.path.join(CWD, "{}/ospfd_v6_vtep.conf".format(rname)),
         )
         router.load_config(
             TopoRouter.RD_BGP, os.path.join(CWD, "{}/bgpd_v6_vtep.conf".format(rname))
@@ -299,6 +302,34 @@ def test_ip_pe2_learn():
     # tgen.mininet_cli()
 
 
+def test_imet():
+    tgen = get_topogen()
+
+    dut_name = "PE1"
+    dut = tgen.gears[dut_name]
+    rd = "10.30.30.30:2"
+    prefix = "[3]:[0]:[128]:[10:30:30::30]"
+    pmsi_label = 101
+    pmsi_id = "10:30:30::30"
+    # Check Imet from PE2 to PE1
+    test_fn = partial(evpn_check_bgp_imet, dut, rd, prefix, pmsi_label, pmsi_id)
+    _, result = topotest.run_and_expect(test_fn, None, count=10, wait=3)
+    assertmsg = f"{dut_name} IMET not present/incorrect, result:{result}"
+    assert result is None, assertmsg
+
+    # Check Imet from PE1 to PE2
+    dut_name = "PE2"
+    dut = tgen.gears[dut_name]
+    rd = "10.10.10.10:2"
+    prefix = "[3]:[0]:[128]:[10:10:10::10]"
+    pmsi_label = 101
+    pmsi_id = "10:10:10::10"
+    test_fn = partial(evpn_check_bgp_imet, dut, rd, prefix, pmsi_label, pmsi_id)
+    _, result = topotest.run_and_expect(test_fn, None, count=10, wait=3)
+    assertmsg = f"{dut_name} IMET not present/incorrect, result:{result}"
+    assert result is None, assertmsg
+
+
 def test_memory_leak():
     "Run the memory leak test and report results."
     tgen = get_topogen()

tests/topotests/lib/evpn.py

@@ -1558,6 +1558,34 @@ def evpn_trigger_host_arp(tgen, host_gateways, interface="swp1", count=3, interv
                 sleep(interval)
 
 
+#
+#
+#
+def evpn_check_bgp_imet(dut, rd, prefix, pmsi_label, pmsi_id):
+    """
+    Return error if the type-3 PMSI attr label or ID don't match the inputs
+    """
+    rd_routes_json = dut.vtysh_cmd(f"show bgp l2vpn evpn route rd {rd} type 3 json")
+    rd_routes = json.loads(rd_routes_json)
+
+    if not rd_routes:
+        return "Imet routes not found"
+
+    if rd not in rd_routes or prefix not in rd_routes[rd]:
+        return f"Imet routes not found for rd {rd} and {prefix}"
+    paths = rd_routes[rd][prefix]["paths"]
+    if not len(paths):
+        return f"Imet route paths routes not found for rd {rd} and {prefix}"
+    out_label = paths[0][0]["pmsi"].get("label", 0)
+    if out_label != pmsi_label:
+        return f"Imet PMSI Label mismatch Expected {pmsi_label} Got {out_label}"
+
+    out_id = paths[0][0]["pmsi"].get("id", "")
+    if out_id != pmsi_id:
+        return f"Imet PMSI Id mismatch Expected {pmsi_id} Got {out_id}"
+    return None
+
+
 def evpn_trigger_arp_scapy(tgen, host_gateways, interface="swp1"):
     """
     Trigger ARP using Scapy to populate MAC address tables in the EVPN fabric.