Fix test_bgp_router_id.py for v6 topo

tests/bgp/test_bgp_router_id.py

@@ -6,6 +6,8 @@
 from tests.common.helpers.assertions import pytest_require, pytest_assert
 from tests.common.helpers.bgp import run_bgp_facts
 from tests.common.utilities import wait_until
+from tests.common.utilities import is_ipv6_only_topology
+from ipaddress import ip_interface
 
 
 pytestmark = [
@@ -17,8 +19,10 @@
 CUSTOMIZED_BGP_ROUTER_ID = "8.8.8.8"
 
 
-def verify_bgp(enum_asic_index, duthost, expected_bgp_router_id, neighbor_type, nbrhosts):
-    output = duthost.shell("show ip bgp summary", module_ignore_errors=True)["stdout"]
+def verify_bgp(enum_asic_index, duthost, expected_bgp_router_id, neighbor_type, nbrhosts, tbinfo):
+    is_v6_topo = is_ipv6_only_topology(tbinfo)
+    cmd = "show ipv6 bgp summary" if is_v6_topo else "show ip bgp summary"
+    output = duthost.shell(cmd, module_ignore_errors=True)["stdout"]
 
     # Verify router id from DUT itself
     pattern = r"BGP router identifier (\d+\.\d+\.\d+\.\d+)"
@@ -40,7 +44,8 @@ def verify_bgp(enum_asic_index, duthost, expected_bgp_router_id, neighbor_type,
     local_ip_map = {}
     cfg_facts = duthost.config_facts(host=duthost.hostname, source="running")['ansible_facts']
     for _, item in cfg_facts.get("BGP_NEIGHBOR", {}).items():
-        if "." in item["local_addr"]:
+        addr_char = ":" if is_v6_topo else "."
+        if addr_char in item["local_addr"]:
             local_ip_map[item["name"]] = item["local_addr"]
 
     for neighbor_name, nbrhost in nbrhosts.items():
@@ -50,9 +55,15 @@ def verify_bgp(enum_asic_index, duthost, expected_bgp_router_id, neighbor_type,
         ).format(neighbor_name, local_ip_map))
 
         if neighbor_type == "sonic":
-            cmd = "show ip neighbors {}".format(local_ip_map[neighbor_name])
+            if is_v6_topo:
+                cmd = "show ipv6 bgp neighbors {}".format(local_ip_map[neighbor_name])
+            else:
+                cmd = "show ip bgp neighbors {}".format(local_ip_map[neighbor_name])
         elif neighbor_type == "eos":
-            cmd = "/usr/bin/Cli -c \"show ip bgp neighbors {}\"".format(local_ip_map[neighbor_name])
+            if is_v6_topo:
+                cmd = "/usr/bin/Cli -c \"show ipv6 bgp peers {}\"".format(local_ip_map[neighbor_name])
+            else:
+                cmd = "/usr/bin/Cli -c \"show ip bgp neighbors {}\"".format(local_ip_map[neighbor_name])
         output = nbrhost["host"].shell(cmd, module_ignore_errors=True)['stdout']
         pattern = r"BGP version 4, remote router ID (\d+\.\d+\.\d+\.\d+)"
         match = re.search(pattern, output)
@@ -78,61 +89,84 @@ def loopback_ip(duthosts, enum_frontend_dut_hostname):
     yield loopback_ip
 
 
-def restart_bgp(duthost):
+@pytest.fixture()
+def loopback_ipv6(duthosts, enum_frontend_dut_hostname):
+    duthost = duthosts[enum_frontend_dut_hostname]
+    cfg_facts = duthost.config_facts(host=duthost.hostname, source="running")['ansible_facts']
+    loopback_ip = None
+    loopback_table = cfg_facts.get("LOOPBACK_INTERFACE", {})
+    for key in loopback_table.get("Loopback0", {}).keys():
+        if ":" in key:
+            loopback_ip = key.split("/")[0]
+    pytest_require(loopback_ip is not None, "Cannot get IPv6 address of Loopback0")
+    # If bgp_adv_lo_prefix_as_128 is false, a /64 prefix of IPv6 loopback addr is used
+    # i.e. fc00:1::32/128 -> fc00:1::/64
+    dev_meta = cfg_facts.get('DEVICE_METADATA', {})
+    bgp_adv_lo_prefix_as_128 = "false"
+    if "localhost" in dev_meta and "bgp_adv_lo_prefix_as_128" in dev_meta["localhost"]:
+        bgp_adv_lo_prefix_as_128 = dev_meta["localhost"]["bgp_adv_lo_prefix_as_128"]
+    if bgp_adv_lo_prefix_as_128.lower() != "true":
+        loopback_ip = str(ip_interface(loopback_ip + "/64").network.network_address)
+    yield loopback_ip
+
+
+def restart_bgp(duthost, tbinfo):
     duthost.reset_service("bgp")
     duthost.restart_service("bgp")
     pytest_assert(wait_until(100, 10, 10, duthost.is_service_fully_started_per_asic_or_host, "bgp"), "BGP not started.")
-    pytest_assert(wait_until(100, 10, 10, duthost.check_default_route, "bgp"), "Default route not ready")
+    pytest_assert(wait_until(100, 10, 10, duthost.check_default_route,
+                             ipv4=not is_ipv6_only_topology(tbinfo)), "Default route not ready")
     # After restarting bgp, add time wait for bgp_facts to fetch latest status
     time.sleep(20)
 
 
 @pytest.fixture()
-def router_id_setup_and_teardown(duthosts, enum_frontend_dut_hostname):
+def router_id_setup_and_teardown(duthosts, enum_frontend_dut_hostname, tbinfo):
     duthost = duthosts[enum_frontend_dut_hostname]
     duthost.shell("sonic-db-cli CONFIG_DB hset \"DEVICE_METADATA|localhost\" \"bgp_router_id\" \"{}\""
                   .format(CUSTOMIZED_BGP_ROUTER_ID), module_ignore_errors=True)
-    restart_bgp(duthost)
+    restart_bgp(duthost, tbinfo)
 
     yield
 
     duthost.shell("sonic-db-cli CONFIG_DB hdel \"DEVICE_METADATA|localhost\" \"bgp_router_id\"",
                   module_ignore_errors=True)
-    restart_bgp(duthost)
+    restart_bgp(duthost, tbinfo)
 
 
 @pytest.fixture(scope="function")
-def router_id_loopback_setup_and_teardown(duthosts, enum_frontend_dut_hostname, loopback_ip):
+def router_id_loopback_setup_and_teardown(duthosts, enum_frontend_dut_hostname, loopback_ip, tbinfo):
     duthost = duthosts[enum_frontend_dut_hostname]
     duthost.shell("sonic-db-cli CONFIG_DB hset \"DEVICE_METADATA|localhost\" \"bgp_router_id\" \"{}\""
                   .format(CUSTOMIZED_BGP_ROUTER_ID), module_ignore_errors=True)
     duthost.shell("sonic-db-cli CONFIG_DB del \"LOOPBACK_INTERFACE|Loopback0|{}/32\"".format(loopback_ip))
-    restart_bgp(duthost)
+    restart_bgp(duthost, tbinfo)
 
     yield
 
     duthost.shell("sonic-db-cli CONFIG_DB hdel \"DEVICE_METADATA|localhost\" \"bgp_router_id\"",
                   module_ignore_errors=True)
     duthost.shell("sonic-db-cli CONFIG_DB hset \"LOOPBACK_INTERFACE|Loopback0|{}/32\" \"NULL\" \"NULL\""
                   .format(loopback_ip), module_ignore_errors=True)
-    restart_bgp(duthost)
+    restart_bgp(duthost, tbinfo)
 
 
-def test_bgp_router_id_default(duthosts, enum_frontend_dut_hostname, enum_asic_index, nbrhosts, request, loopback_ip):
+def test_bgp_router_id_default(duthosts, enum_frontend_dut_hostname, enum_asic_index, nbrhosts, request, loopback_ip,
+                               tbinfo):
     # Test in default config, the BGP router id should be aligned with Loopback IPv4 address
     duthost = duthosts[enum_frontend_dut_hostname]
     neighbor_type = request.config.getoption("neighbor_type")
-    verify_bgp(enum_asic_index, duthost, loopback_ip, neighbor_type, nbrhosts)
+    verify_bgp(enum_asic_index, duthost, loopback_ip, neighbor_type, nbrhosts, tbinfo)
 
 
 def test_bgp_router_id_set(duthosts, enum_frontend_dut_hostname, enum_asic_index, nbrhosts, request, loopback_ip,
-                           router_id_setup_and_teardown):
+                           router_id_setup_and_teardown, tbinfo):
     # Test in the scenario that bgp_router_id and Loopback IPv4 address both exist in CONFIG_DB, the actual BGP router
     # ID should be aligned with bgp_router_id in CONFIG_DB. And the Loopback IPv4 address should be advertised to BGP
     # neighbor
     duthost = duthosts[enum_frontend_dut_hostname]
     neighbor_type = request.config.getoption("neighbor_type")
-    verify_bgp(enum_asic_index, duthost, CUSTOMIZED_BGP_ROUTER_ID, neighbor_type, nbrhosts)
+    verify_bgp(enum_asic_index, duthost, CUSTOMIZED_BGP_ROUTER_ID, neighbor_type, nbrhosts, tbinfo)
     # Verify Loopback ip has been advertised to neighbor
     cfg_facts = duthost.config_facts(host=duthost.hostname, source="running")['ansible_facts']
     for remote_ip in cfg_facts.get("BGP_NEIGHBOR", {}).keys():
@@ -153,10 +187,38 @@ def test_bgp_router_id_set(duthosts, enum_frontend_dut_hostname, enum_asic_index
         ).format(loopback_ip, output["stdout"]))
 
 
+def test_bgp_router_id_set_ipv6(duthosts, enum_frontend_dut_hostname, enum_asic_index, nbrhosts, request, loopback_ipv6,
+                                router_id_setup_and_teardown, tbinfo):
+    # Test in the scenario that bgp_router_id and Loopback IPv6 address both exist in CONFIG_DB, the actual BGP router
+    # ID should be aligned with bgp_router_id in CONFIG_DB. And the Loopback IPv6 address should be advertised to BGP
+    # neighbor
+    duthost = duthosts[enum_frontend_dut_hostname]
+    neighbor_type = request.config.getoption("neighbor_type")
+    verify_bgp(enum_asic_index, duthost, CUSTOMIZED_BGP_ROUTER_ID, neighbor_type, nbrhosts, tbinfo)
+    # Verify Loopback ip has been advertised to neighbor
+    cfg_facts = duthost.config_facts(host=duthost.hostname, source="running")['ansible_facts']
+    for remote_ip in cfg_facts.get("BGP_NEIGHBOR", {}).keys():
+        if ":" not in remote_ip or "FT2" in cfg_facts["BGP_NEIGHBOR"][remote_ip]["name"]:
+            continue
+        output = duthost.shell("show ipv6 bgp neighbor {} advertised-routes| grep {}".format(remote_ip, loopback_ipv6),
+                               module_ignore_errors=True)
+        pytest_assert(output["rc"] == 0, (
+            "Failed to check whether Loopback ipv6 address has been advertised. "
+            "Return code: {} "
+            "Output: {}"
+        ).format(output["rc"], output))
+
+        pytest_assert(loopback_ipv6 in output["stdout"], (
+            "Router advertised unexpected. "
+            "Expected loopback IP: {} "
+            "Actual output: {}"
+        ).format(loopback_ipv6, output["stdout"]))
+
+
 def test_bgp_router_id_set_without_loopback(duthosts, enum_frontend_dut_hostname, enum_asic_index, nbrhosts, request,
-                                            router_id_loopback_setup_and_teardown):
+                                            router_id_loopback_setup_and_teardown, tbinfo):
     # Test in the scenario that bgp_router_id specified but Loopback IPv4 address not set, BGP could work well and the
     # actual BGP router id should be aligned with CONFIG_DB
     duthost = duthosts[enum_frontend_dut_hostname]
     neighbor_type = request.config.getoption("neighbor_type")
-    verify_bgp(enum_asic_index, duthost, CUSTOMIZED_BGP_ROUTER_ID, neighbor_type, nbrhosts)
+    verify_bgp(enum_asic_index, duthost, CUSTOMIZED_BGP_ROUTER_ID, neighbor_type, nbrhosts, tbinfo)

tests/common/plugins/conditional_mark/tests_mark_conditions.yaml

@@ -366,25 +366,25 @@ bgp/test_bgp_router_id.py:
     conditions:
       - "'t0-isolated-d256u256s2' in topo_name and platform in ['x86_64-nvidia_sn5640-r0']"
 
-bgp/test_bgp_router_id.py::test_bgp_router_id_default:
-  xfail:
-    reason: "xfail for IPv6-only topologies, issue with trying to parse with IPv4 style"
+bgp/test_bgp_router_id.py::test_bgp_router_id_set[:
+  skip:
+    reason: "Skip for IPv6-only topologies"
     conditions:
-      - "https://github.com/sonic-net/sonic-mgmt/issues/19916 and '-v6-' in topo_name"
-
-bgp/test_bgp_router_id.py::test_bgp_router_id_set:
+      - "'-v6-' in topo_name"
   xfail:
-    reason: "xfail for IPv6-only topologies, issue with trying to parse with IPv4 style. Or test case has issue on the t0-isolated-d256u256s2 topo."
-    conditions_logical_operator: or
+    reason: "Test case has issue on the t0-isolated-d256u256s2 topo."
     conditions:
-      - "https://github.com/sonic-net/sonic-mgmt/issues/19916 and '-v6-' in topo_name"
       - "'t0-isolated-d256u256s2' in topo_name and platform in ['x86_64-nvidia_sn5640-r0']"
 
-bgp/test_bgp_router_id.py::test_bgp_router_id_set_without_loopback:
+bgp/test_bgp_router_id.py::test_bgp_router_id_set_ipv6:
+  skip:
+    reason: "Skip for topologies that are not IPv6-only"
+    conditions:
+      - "'-v6-' not in topo_name"
   xfail:
-    reason: "xfail for IPv6-only topologies, issue with trying to parse with IPv4 style"
+    reason: "Test case has issue on the t0-isolated-d256u256s2 topo."
     conditions:
-      - "https://github.com/sonic-net/sonic-mgmt/issues/19916 and '-v6-' in topo_name"
+      - "'t0-isolated-v6-d256u256s2' in topo_name and platform in ['x86_64-nvidia_sn5640-r0']"
 
 bgp/test_bgp_sentinel.py::test_bgp_sentinel[IPv4:
   skip: