[202412] Fix setup_interfaces fixture and related bgp tests for v6 topo

tests/bgp/bgp_helpers.py

@@ -19,6 +19,7 @@
 from tests.common.helpers.parallel import reset_ansible_local_tmp
 from tests.common.helpers.parallel import parallel_run
 from tests.common.utilities import wait_until
+from tests.common.utilities import is_ipv6_only_topology
 from tests.bgp.traffic_checker import get_traffic_shift_state
 from tests.bgp.constants import TS_NORMAL
 from tests.common.devices.eos import EosHost
@@ -280,12 +281,15 @@ def bgp_allow_list_setup(tbinfo, nbrhosts, duthosts, rand_one_dut_hostname):
             downstream_namespace = neigh['namespace']
             break
 
+    is_v6_topo = is_ipv6_only_topology(tbinfo)
+
     setup_info = {
         'downstream': downstream,
         'downstream_namespace': downstream_namespace,
         'downstream_exabgp_port': downstream_exabgp_port,
         'downstream_exabgp_port_v6': downstream_exabgp_port_v6,
         'other_neighbors': other_neighbors,
+        'is_v6_topo': is_v6_topo,
     }
     yield setup_info
 
@@ -306,18 +310,21 @@ def update_routes(action, ptfip, port, route):
 
 
 def build_routes(tbinfo, prefix_list, expected_community):
-    nhipv4 = tbinfo['topo']['properties']['configuration_properties']['common']['nhipv4']
-    nhipv6 = tbinfo['topo']['properties']['configuration_properties']['common']['nhipv6']
+    nhipv4 = tbinfo['topo']['properties']['configuration_properties']['common'].get('nhipv4')
+    nhipv6 = tbinfo['topo']['properties']['configuration_properties']['common'].get('nhipv6')
     routes = []
     for list_name, prefixes in list(prefix_list.items()):
         logging.info('list_name: {}, prefixes: {}'.format(list_name, str(prefixes)))
         for prefix in prefixes:
             route = {}
             route['prefix'] = prefix
             if ipaddress.IPNetwork(prefix).version == 4:
-                route['nexthop'] = nhipv4
+                nhip = nhipv4
             else:
-                route['nexthop'] = nhipv6
+                nhip = nhipv6
+            if not nhip:
+                continue
+            route['nexthop'] = nhip
             if 'COMMUNITY' in list_name:
                 route['community'] = expected_community
             routes.append(route)
@@ -380,7 +387,9 @@ def check_routes_on_from_neighbor(setup_info, nbrhosts):
     Verify if there are routes on neighbor who announce them.
     """
     downstream = setup_info['downstream']
-    for prefixes in list(PREFIX_LISTS.values()):
+    for list_name, prefixes in list(PREFIX_LISTS.items()):
+        if setup_info['is_v6_topo'] and "v6" not in list_name.lower():
+            continue
         for prefix in prefixes:
             downstream_route = nbrhosts[downstream]['host'].get_route(prefix)
             route_entries = downstream_route['vrfs']['default']['bgpRouteEntries']
@@ -409,6 +418,8 @@ def check_other_neigh(nbrhosts, permit, node=None, results=None):
 
         prefix_results = []
         for list_name, prefixes in list(PREFIX_LISTS.items()):
+            if setup['is_v6_topo'] and "v6" not in list_name.lower():
+                continue
             for prefix in prefixes:
                 prefix_result = {'failed': False, 'prefix': prefix, 'reasons': []}
                 neigh_route = nbrhosts[node]['host'].get_route(prefix)['vrfs']['default']['bgpRouteEntries']
@@ -462,6 +473,8 @@ def check_other_neigh(nbrhosts, permit, node=None, results=None):
 
         prefix_results = []
         for list_name, prefixes in list(PREFIX_LISTS.items()):
+            if setup['is_v6_topo'] and "v6" not in list_name.lower():
+                continue
             for prefix in prefixes:
                 prefix_result = {'failed': False, 'prefix': prefix, 'reasons': []}
                 neigh_route = nbrhosts[node]['host'].get_route(prefix)['vrfs']['default']['bgpRouteEntries']

tests/bgp/route_checker.py

@@ -198,23 +198,23 @@ def parse_routes_process_vsonic(node=None, results=None):
     return all_routes
 
 
-def verify_only_loopback_routes_are_announced_to_neighs(dut_hosts, duthost, neigh_hosts, community):
+def verify_only_loopback_routes_are_announced_to_neighs(dut_hosts, duthost, neigh_hosts, community, is_v6_topo=False):
     """
     Verify only loopback routes with certain community are announced to neighs in TSA
     """
-    return verify_loopback_route_with_community(dut_hosts, duthost, neigh_hosts, 4, community) and \
+    return (is_v6_topo or verify_loopback_route_with_community(dut_hosts, duthost, neigh_hosts, 4, community)) and \
         verify_loopback_route_with_community(
             dut_hosts, duthost, neigh_hosts, 6, community)
 
 
 def assert_only_loopback_routes_announced_to_neighs(dut_hosts, duthost, neigh_hosts, community,
-                                                    error_msg=""):
+                                                    error_msg="", is_v6_topo=False):
     if not error_msg:
         error_msg = "Failed to verify only loopback routes are announced to neighbours"
 
     pytest_assert(
         wait_until(180, 10, 5, verify_only_loopback_routes_are_announced_to_neighs,
-                   dut_hosts, duthost, neigh_hosts, community),
+                   dut_hosts, duthost, neigh_hosts, community, is_v6_topo),
         error_msg
     )
 

tests/bgp/test_bgp_allow_list.py

@@ -51,13 +51,15 @@ def load_remove_allow_list(duthosts, bgp_allow_list_setup, rand_one_dut_hostname
     remove_allow_list(duthost, namespace, ALLOW_LIST_PREFIX_JSON_FILE)
 
 
-def check_routes_on_dut(duthost, namespace):
+def check_routes_on_dut(duthost, setup_info):
     """
     Verify routes on dut
     """
-    for prefixes in list(PREFIX_LISTS.values()):
+    for list_name, prefixes in list(PREFIX_LISTS.items()):
+        if setup_info['is_v6_topo'] and "v6" not in list_name.lower():
+            continue
         for prefix in prefixes:
-            dut_route = duthost.get_route(prefix, namespace)
+            dut_route = duthost.get_route(prefix, setup_info['downstream_namespace'])
             pytest_assert(dut_route, 'Route {} is not found on DUT'.format(prefix))
 
 
@@ -71,7 +73,7 @@ def test_default_allow_list_preconfig(duthosts, rand_one_dut_hostname, bgp_allow
     # All routes should be found on from neighbor.
     check_routes_on_from_neighbor(bgp_allow_list_setup, nbrhosts)
     # All routes should be found in dut.
-    check_routes_on_dut(duthost, bgp_allow_list_setup['downstream_namespace'])
+    check_routes_on_dut(duthost, bgp_allow_list_setup)
     # If permit is True, all routes should be forwarded and added drop_community and keep ori community.
     # If permit if False, all routes should not be forwarded.
     check_routes_on_neighbors_empty_allow_list(nbrhosts, bgp_allow_list_setup, permit)
@@ -86,7 +88,7 @@ def test_allow_list(duthosts, rand_one_dut_hostname, bgp_allow_list_setup, nbrho
     # All routes should be found on from neighbor.
     check_routes_on_from_neighbor(bgp_allow_list_setup, nbrhosts)
     # All routes should be found in dut.
-    check_routes_on_dut(duthost, bgp_allow_list_setup['downstream_namespace'])
+    check_routes_on_dut(duthost, bgp_allow_list_setup)
     # If permit is True, all routes should be forwarded. Routs that in allow list should not be add drop_community
     # and keep ori community.
     # If permit is False, Routes in allow_list should be forwarded and keep ori community, routes not in allow_list

tests/bgp/test_bgp_peer_shutdown.py

@@ -5,14 +5,15 @@
 import time
 
 import pytest
-from scapy.all import sniff, IP
+from scapy.all import sniff, IP, IPv6
 from scapy.contrib import bgp
 
 from tests.bgp.bgp_helpers import capture_bgp_packages_to_file, fetch_and_delete_pcap_file
 from tests.common.errors import RunAnsibleModuleFail
 from tests.common.helpers.bgp import BGPNeighbor
 from tests.common.helpers.constants import DEFAULT_NAMESPACE
 from tests.common.utilities import wait_until, delete_running_config
+from tests.common.utilities import is_ipv6_only_topology
 
 pytestmark = [
     pytest.mark.topology('t0', 't1', 't2'),
@@ -124,18 +125,20 @@ def is_neighbor_session_established(duthost, neighbor):
             and bgp_facts["bgp_neighbors"][neighbor.ip]["state"] == "established")
 
 
-def bgp_notification_packets(pcap_file):
+def bgp_notification_packets(pcap_file, is_v6_topo):
     """Get bgp notification packets from pcap file."""
+    ip_ver = IPv6 if is_v6_topo else IP
     packets = sniff(
         offline=pcap_file,
-        lfilter=lambda p: IP in p and bgp.BGPHeader in p and p[bgp.BGPHeader].type == 3,
+        lfilter=lambda p: ip_ver in p and bgp.BGPHeader in p and p[bgp.BGPHeader].type == 3,
     )
     return packets
 
 
-def match_bgp_notification(packet, src_ip, dst_ip, action, bgp_session_down_time):
+def match_bgp_notification(packet, src_ip, dst_ip, action, bgp_session_down_time, is_v6_topo):
     """Check if the bgp notification packet matches."""
-    if not (packet[IP].src == src_ip and packet[IP].dst == dst_ip):
+    ip_ver = IPv6 if is_v6_topo else IP
+    if not (packet[ip_ver].src == src_ip and packet[ip_ver].dst == dst_ip):
         return False
 
     bgp_fields = packet[bgp.BGPNotification].fields
@@ -188,10 +191,13 @@ def test_bgp_peer_shutdown(
     duthosts,
     enum_rand_one_per_hwsku_frontend_hostname,
     request,
+    tbinfo
 ):
     duthost = duthosts[enum_rand_one_per_hwsku_frontend_hostname]
     n0 = common_setup_teardown
-    announced_route = {"prefix": "10.10.100.0/27", "nexthop": n0.ip}
+    is_v6_topo = is_ipv6_only_topology(tbinfo)
+    announced_route = {"prefix": "fc00:10::/64", "nexthop": n0.ip} if is_v6_topo else \
+                      {"prefix": "10.10.100.0/27", "nexthop": n0.ip}
 
     for _ in range(TEST_ITERATIONS):
         try:
@@ -225,7 +231,7 @@ def test_bgp_peer_shutdown(
                     pytest.fail("Could not tear down bgp session")
 
             local_pcap_filename = fetch_and_delete_pcap_file(bgp_pcap, constants.log_dir, duthost, request)
-            bpg_notifications = bgp_notification_packets(local_pcap_filename)
+            bpg_notifications = bgp_notification_packets(local_pcap_filename, is_v6_topo)
             for bgp_packet in bpg_notifications:
                 logging.debug(
                     "bgp notification packet, capture time %s, packet details:\n%s",
@@ -234,7 +240,8 @@ def test_bgp_peer_shutdown(
                 )
 
                 bgp_session_down_time = get_bgp_down_timestamp(duthost, n0.namespace, n0.ip, timestamp_before_teardown)
-                if not match_bgp_notification(bgp_packet, n0.ip, n0.peer_ip, "cease", bgp_session_down_time):
+                if not match_bgp_notification(bgp_packet, n0.ip, n0.peer_ip, "cease", bgp_session_down_time,
+                                              is_v6_topo):
                     pytest.fail("BGP notification packet does not match expected values")
 
             announced_route_on_dut_after_shutdown = duthost.get_route(announced_route["prefix"], n0.namespace)