[chassis] Enhance test_default_route.py for chassis

tests/route/test_default_route.py

@@ -7,6 +7,7 @@
 from tests.common.config_reload import config_reload
 from tests.common.helpers.assertions import pytest_assert, pytest_require
 from tests.common.utilities import wait_until
+from tests.common.utilities import find_duthost_on_role
 
 
 pytestmark = [
@@ -18,17 +19,17 @@
 
 def get_upstream_neigh_type(topo):
     if 't0' in topo or 'dualtor' in topo:
-        return 't1'
+        return 'T1'
     elif 't1' in topo:
-        return 't2'
+        return 'T2'
     elif 't2' in topo:
-        return 't3'
+        return 'T3'
     elif 'm0' in topo:
-        return 'm1'
+        return 'M1'
     else:
         return None
 
-def get_upstream_neigh(tb):
+def get_upstream_neigh(tb, device_neigh_metadata):
     """
     Get the information for upstream neighbors present in the testbed
 
@@ -44,7 +45,9 @@ def get_upstream_neigh(tb):
         return upstream_neighbors
 
     for neigh_name, neigh_cfg in topo_cfg_facts.iteritems():
-        if neigh_type not in neigh_name.lower():
+        if neigh_type not in neigh_name:
+            continue
+        if neigh_type == 'T3' and device_neigh_metadata[neigh_name]['type'] == 'AZNGHub':
             continue
         interfaces = neigh_cfg.get('interfaces', {})
         ipv4_addr = None
@@ -70,12 +73,12 @@ def get_uplink_ns(tbinfo, bgp_name_to_ns_mapping):
     neigh_type = get_upstream_neigh_type(tbinfo['topo']['name'])
     asics = set()
     for name, asic in bgp_name_to_ns_mapping.items():
-        if neigh_type not in name.lower():
+        if neigh_type not in name:
             continue
         asics.add(asic)
     return asics
 
-def verify_default_route_in_app_db(duthost, tbinfo, af, uplink_ns):
+def verify_default_route_in_app_db(duthost, tbinfo, af, uplink_ns, device_neigh_metadata):
     """
     Verify the nexthops for the default routes match the ip interfaces
     configured on the peer device 
@@ -97,7 +100,7 @@ def verify_default_route_in_app_db(duthost, tbinfo, af, uplink_ns):
     pytest_assert(nexthops is not None, "Default route has not nexthops")
     logging.info("nexthops in app_db {}".format(nexthops) )
 
-    upstream_neigh = get_upstream_neigh(tbinfo)
+    upstream_neigh = get_upstream_neigh(tbinfo, device_neigh_metadata)
     pytest_assert(upstream_neigh is not None, "No upstream neighbors in the testbed")
 
     if af == 'ipv4':
@@ -112,13 +115,13 @@ def verify_default_route_in_app_db(duthost, tbinfo, af, uplink_ns):
 
 
 
-def test_default_route_set_src(duthosts, enum_rand_one_per_hwsku_frontend_hostname, enum_asic_index, tbinfo):
+def test_default_route_set_src(duthosts, tbinfo):
     """
     check if ipv4 and ipv6 default src address match Loopback0 address
 
     """
-    duthost = duthosts[enum_rand_one_per_hwsku_frontend_hostname]
-    asichost = duthost.asic_instance(enum_asic_index)
+    duthost = find_duthost_on_role(duthosts, get_upstream_neigh_type(tbinfo['topo']['name']) , tbinfo)
+    asichost = duthost.asic_instance(0 if duthost.is_multi_asic else None)
 
     config_facts = asichost.config_facts(host=duthost.hostname, source="running")['ansible_facts']
 
@@ -148,13 +151,13 @@ def test_default_route_set_src(duthosts, enum_rand_one_per_hwsku_frontend_hostna
     pytest_assert(rtinfo['set_src'] == lo_ipv6.ip, \
             "default v6 route set src to wrong IP {} != {}".format(rtinfo['set_src'], lo_ipv6.ip))
 
-def test_default_ipv6_route_next_hop_global_address(duthosts, enum_rand_one_per_hwsku_frontend_hostname, enum_asic_index):
+def test_default_ipv6_route_next_hop_global_address(duthosts, tbinfo):
     """
     check if ipv6 default route nexthop address uses global address
 
     """
-    duthost = duthosts[enum_rand_one_per_hwsku_frontend_hostname]
-    asichost = duthost.asic_instance(enum_asic_index)
+    duthost = find_duthost_on_role(duthosts, get_upstream_neigh_type(tbinfo['topo']['name']) , tbinfo)
+    asichost = duthost.asic_instance(0 if duthost.is_multi_asic else None)
 
     rtinfo = asichost.get_ip_route_info(ipaddress.ip_network(u"::/0"))
     pytest_assert(len(rtinfo['nexthops']) > 0, "cannot find ipv6 nexthop for default route")
@@ -163,7 +166,7 @@ def test_default_ipv6_route_next_hop_global_address(duthosts, enum_rand_one_per_
                 "use link local address {} for nexthop".format(nh[0]))
 
 
-def test_default_route_with_bgp_flap(duthosts, enum_rand_one_per_hwsku_frontend_hostname, tbinfo):
+def test_default_route_with_bgp_flap(duthosts, tbinfo):
     """
     Check the default route present in app_db has the correct nexthops ip
     Check the default route is removed when the bgp sessions are shutdown
@@ -174,7 +177,7 @@ def test_default_route_with_bgp_flap(duthosts, enum_rand_one_per_hwsku_frontend_
             "Skip this testcase since this topology {} has no default routes"\
                 .format(tbinfo['topo']['name']))
 
-    duthost = duthosts[enum_rand_one_per_hwsku_frontend_hostname]
+    duthost = find_duthost_on_role(duthosts, get_upstream_neigh_type(tbinfo['topo']['name']) , tbinfo)
 
     config_facts  = duthost.config_facts(host=duthost.hostname, source="running")['ansible_facts']
     bgp_neighbors = config_facts.get('BGP_NEIGHBOR', {})
@@ -187,20 +190,21 @@ def test_default_route_with_bgp_flap(duthosts, enum_rand_one_per_hwsku_frontend_
 
     af_list = ['ipv4', 'ipv6']
 
-    # verify the default route is correct in the app db
-    for af in af_list:
-        verify_default_route_in_app_db(duthost, tbinfo, af, uplink_ns)
-
-    duthost.command("sudo config bgp shutdown all")
-    if not wait_until(120, 2, 0, duthost.is_bgp_state_idle):
-        pytest.fail(
-            'BGP Shutdown Timeout: BGP sessions not shutdown after 120 seconds')
-
-    # give some more time for default route to be removed
-    if not wait_until(120, 2, 0, duthost.is_default_route_removed_from_app_db, uplink_ns):
-        pytest.fail(
-            'Default route is not removed from APP_DB')
-
-    duthost.command("sudo config bgp startup all")
-    if not wait_until(300, 10, 0, duthost.check_bgp_session_state, bgp_neighbors.keys()):
-        pytest.fail("not all bgp sessions are up after config reload")
+    try:
+        # verify the default route is correct in the app db
+        for af in af_list:
+            verify_default_route_in_app_db(duthost, tbinfo, af, uplink_ns, config_facts['DEVICE_NEIGHBOR_METADATA'])
+
+        duthost.command("sudo config bgp shutdown all")
+        if not wait_until(120, 2, 0, duthost.is_bgp_state_idle):
+            pytest.fail(
+                'BGP Shutdown Timeout: BGP sessions not shutdown after 120 seconds')
+
+        # give some more time for default route to be removed
+        if not wait_until(120, 2, 0, duthost.is_default_route_removed_from_app_db, uplink_ns):
+            pytest.fail(
+                'Default route is not removed from APP_DB')
+    finally:
+        duthost.command("sudo config bgp startup all")
+        if not wait_until(300, 10, 0, duthost.check_bgp_session_state, bgp_neighbors.keys()):
+            pytest.fail("not all bgp sessions are up after config reload")