[action] [PR:3788] Switchport mode update for 'show interfaces status'

scripts/intfutil

@@ -89,33 +89,40 @@ def get_sub_port_intf_list(config_db):
     return sub_intf_list
 
 
-def get_interface_vlan_dict(config_db):
+def get_interface_sw_mode_dict(config_db, front_panel_ports_list):
     """
-    Get info from REDIS ConfigDB and create interface to vlan mapping
+    Get info from REDIS ConfigDB and create interface to swithport mode mapping
     """
-    get_int_vlan_configdb_info = config_db.get_table('VLAN_MEMBER')
-    int_list = []
-    vlan_list = []
-    for line in get_int_vlan_configdb_info:
-        vlan_number = line[0]
-        interface = line[1]
-        int_list.append(interface)
-        vlan_list.append(vlan_number)
-    int_to_vlan_dict = dict(zip(int_list, vlan_list))
-    return int_to_vlan_dict
+    vlan_member_table = config_db.get_table('VLAN_MEMBER')
 
+    vlan_member_keys = []
+    for _, key in vlan_member_table:
+        vlan_member_keys.append(key)
 
-def config_db_vlan_port_keys_get(int_to_vlan_dict, front_panel_ports_list, intf_name):
+    intf_to_sw_mode_dict = {}
+    for intf_name in front_panel_ports_list:
+        port = config_db.get_entry('PORT', intf_name)
+        if "mode" in port:
+            mode = port['mode']
+        elif intf_name in vlan_member_keys:
+            mode = 'trunk'
+        else:
+            mode = 'routed'
+        intf_to_sw_mode_dict[intf_name] = mode
+
+    return intf_to_sw_mode_dict
+
+
+def config_db_vlan_port_keys_get(intf_to_sw_mode_dict, intf_to_po_dict, intf_name):
     """
     Get interface vlan value and return it.
     """
-    vlan = "routed"
-    if intf_name in front_panel_ports_list:
-        if intf_name in int_to_vlan_dict.keys():
-            vlan = int_to_vlan_dict[intf_name]
-            if "Vlan" in vlan:
-                vlan = "trunk"
-    return vlan
+    mode = "routed"
+    if intf_name in intf_to_po_dict.keys():
+        mode = intf_to_po_dict[intf_name]
+    elif intf_name in intf_to_sw_mode_dict.keys():
+        mode = intf_to_sw_mode_dict[intf_name]
+    return mode
 
 
 def appl_db_keys_get(appl_db, front_panel_ports_list, intf_name):
@@ -316,6 +323,31 @@ def create_po_int_dict(po_int_tuple_list):
     po_int_dict = tuple_to_dict(po_int_tuple_list, temp_dict)
     return po_int_dict
 
+def create_po_to_sw_mode_dict(config_db, po_int_tuple_list):
+    """
+    This function takes the portchannel to interface tuple
+    and converts that into an interface to portchannel dictionary
+    with the portchannels as the key and the mode as the values.
+    """
+    vlan_member_table = config_db.get_table('VLAN_MEMBER')
+
+    vlan_member_keys = []
+    for _, key in vlan_member_table:
+        vlan_member_keys.append(key)
+
+    po_to_sw_mode_dict = {}
+    for po, intf in po_int_tuple_list:
+        portchannel = config_db.get_entry('PORTCHANNEL', po)
+        if "mode" in portchannel:
+            mode = portchannel['mode']
+        elif po in vlan_member_keys:
+            mode = 'trunk'
+        else:
+            mode = 'routed'
+
+        po_to_sw_mode_dict[po] = mode
+    return po_to_sw_mode_dict
+
 def create_int_to_portchannel_dict(po_int_tuple_list):
     """
     This function takes the portchannel to interface tuple
@@ -363,7 +395,7 @@ def po_speed_dict(po_int_dict, appl_db):
         po_speed_dict = {}
         return po_speed_dict
 
-def appl_db_portchannel_status_get(appl_db, config_db, po_name, status_type, portchannel_speed_dict, combined_int_to_vlan_po_dict=None):
+def appl_db_portchannel_status_get(appl_db, config_db, po_name, status_type, portchannel_speed_dict, po_to_sw_mode_dict=None):
     """
     Get the port status
     """
@@ -376,8 +408,8 @@ def appl_db_portchannel_status_get(appl_db, config_db, po_name, status_type, por
             return "N/A"
         return status
     if status_type == "vlan":
-        if combined_int_to_vlan_po_dict and po_name in combined_int_to_vlan_po_dict.keys():
-            status = "trunk"
+        if po_to_sw_mode_dict and po_name in po_to_sw_mode_dict.keys():
+            status = po_to_sw_mode_dict[po_name]
         else:
             status = "routed"
         return status
@@ -491,7 +523,7 @@ class IntfStatus(object):
                                 appl_db_port_status_get(self.db, key, PORT_MTU_STATUS),
                                 appl_db_port_status_get(self.db, key, PORT_FEC),
                                 appl_db_port_status_get(self.db, key, PORT_ALIAS),
-                                config_db_vlan_port_keys_get(self.combined_int_to_vlan_po_dict, self.front_panel_ports_list, key),
+                                config_db_vlan_port_keys_get(self.intf_to_sw_mode_dict, self.int_po_dict, key),
                                 appl_db_port_status_get(self.db, key, PORT_OPER_STATUS),
                                 appl_db_port_status_get(self.db, key, PORT_ADMIN_STATUS),
                                 port_optics_get(self.db, key, PORT_OPTICS_TYPE),
@@ -508,7 +540,7 @@ class IntfStatus(object):
                                 appl_db_portchannel_status_get(self.db, self.config_db, po, PORT_MTU_STATUS, self.portchannel_speed_dict),
                                 appl_db_portchannel_status_get(self.db, self.config_db, po, PORT_FEC, self.portchannel_speed_dict),
                                 appl_db_portchannel_status_get(self.db, self.config_db, po, PORT_ALIAS, self.portchannel_speed_dict),
-                                appl_db_portchannel_status_get(self.db, self.config_db, po, "vlan", self.portchannel_speed_dict, self.combined_int_to_vlan_po_dict),
+                                appl_db_portchannel_status_get(self.db, self.config_db, po, "vlan", self.portchannel_speed_dict, self.po_to_sw_mode_dict),
                                 appl_db_portchannel_status_get(self.db, self.config_db, po, PORT_OPER_STATUS, self.portchannel_speed_dict),
                                 appl_db_portchannel_status_get(self.db, self.config_db, po, PORT_ADMIN_STATUS, self.portchannel_speed_dict),
                                 appl_db_portchannel_status_get(self.db, self.config_db, po, PORT_OPTICS_TYPE, self.portchannel_speed_dict),
@@ -530,13 +562,13 @@ class IntfStatus(object):
     def get_intf_status(self):
         self.front_panel_ports_list = get_frontpanel_port_list(self.config_db)
         self.appl_db_keys = appl_db_keys_get(self.db, self.front_panel_ports_list, None)
-        self.int_to_vlan_dict = get_interface_vlan_dict(self.config_db)
+        self.intf_to_sw_mode_dict = get_interface_sw_mode_dict(self.config_db, self.front_panel_ports_list)
         self.get_raw_po_int_configdb_info = get_raw_portchannel_info(self.config_db)
         self.portchannel_list = get_portchannel_list(self.get_raw_po_int_configdb_info)
         self.po_int_tuple_list = create_po_int_tuple_list(self.get_raw_po_int_configdb_info)
         self.po_int_dict = create_po_int_dict(self.po_int_tuple_list)
         self.int_po_dict = create_int_to_portchannel_dict(self.po_int_tuple_list)
-        self.combined_int_to_vlan_po_dict = merge_dicts(self.int_to_vlan_dict, self.int_po_dict)
+        self.po_to_sw_mode_dict = create_po_to_sw_mode_dict(self.config_db, self.po_int_tuple_list)
         self.portchannel_speed_dict = po_speed_dict(self.po_int_dict, self.db)
         self.portchannel_keys = self.portchannel_speed_dict.keys()
 

tests/vlan_test.py

@@ -13,6 +13,10 @@
 from importlib import reload
 import utilities_common.bgp_util as bgp_util
 
+root_path = os.path.dirname(os.path.abspath(__file__))
+modules_path = os.path.dirname(root_path)
+scripts_path = os.path.join(modules_path, "scripts")
+
 IP_VERSION_PARAMS_MAP = {
     "ipv4": {
         "table": "VLAN"
@@ -296,12 +300,24 @@
 """
 
 
+def get_intf_switchport_status(self, output: str, interface: str) -> str:
+    for line in output.splitlines():
+        line = line.strip()
+        if not line or line.startswith("Interface") or line.startswith("----"):
+            continue
+        parts = line.split()
+        if parts[0] == interface and len(parts) >= 2:
+            return parts[1]
+    return "interface not found"
+
+
 class TestVlan(object):
     _old_run_bgp_command = None
 
     @classmethod
     def setup_class(cls):
-        os.environ['UTILITIES_UNIT_TESTING'] = "1"
+        os.environ["PATH"] += os.pathsep + scripts_path
+        os.environ['UTILITIES_UNIT_TESTING'] = "2"
         # ensure that we are working with single asic config
         cls._old_run_bgp_command = bgp_util.run_bgp_command
         bgp_util.run_bgp_command = mock.MagicMock(
@@ -718,6 +734,9 @@ def test_config_vlan_add_portchannel_member_with_switchport_modes(self):
         print(result.exit_code)
         print(result.output)
         assert result.exit_code == 0
+        result = runner.invoke(show.cli.commands["interfaces"].commands["switchport"].commands["status"], obj=db)
+        switchport_status = get_intf_switchport_status(self, result.output, "PortChannel0001")
+        assert "routed" in switchport_status
 
         # Configure PortChannel0001 to routed mode again; should give error as it is already in routed mode
         result = runner.invoke(config.config.commands["switchport"].commands["mode"],
@@ -748,20 +767,30 @@ def test_config_vlan_add_portchannel_member_with_switchport_modes(self):
         print(result.exit_code)
         print(result.output)
         assert result.exit_code == 0
+        result = runner.invoke(show.cli.commands["interfaces"].commands["switchport"].commands["status"], obj=db)
+        print(result.output)
+        switchport_status = get_intf_switchport_status(self, result.output, "PortChannel1001")
+        assert "access" in switchport_status
 
         # Configure PortChannel1001 back to routed mode
         result = runner.invoke(config.config.commands["switchport"].commands["mode"],
                                ["routed", "PortChannel1001"], obj=db)
         print(result.exit_code)
         print(result.output)
         assert result.exit_code == 0
+        result = runner.invoke(show.cli.commands["interfaces"].commands["switchport"].commands["status"], obj=db)
+        switchport_status = get_intf_switchport_status(self, result.output, "PortChannel1001")
+        assert "routed" in switchport_status
 
         # Configure PortChannel1001 to trunk mode
         result = runner.invoke(config.config.commands["switchport"].commands["mode"],
                                ["trunk", "PortChannel1001"], obj=db)
         print(result.exit_code)
         print(result.output)
         assert result.exit_code == 0
+        result = runner.invoke(show.cli.commands["interfaces"].commands["switchport"].commands["status"], obj=db)
+        switchport_status = get_intf_switchport_status(self, result.output, "PortChannel1001")
+        assert "trunk" in switchport_status
 
         # Add back PortChannel1001 tagged member to Vlan4000
         result = runner.invoke(config.config.commands["vlan"].commands["member"].commands["add"],
@@ -1176,6 +1205,10 @@ def test_config_add_del_vlan_and_vlan_member_with_switchport_modes(self, mock_re
         print(result.output)
         assert result.exit_code == 0
         assert "Ethernet20 switched to access mode" in result.output
+        result = runner.invoke(show.cli.commands["interfaces"].commands["switchport"].commands["status"], obj=db)
+        print(result.output)
+        switchport_status = get_intf_switchport_status(self, result.output, "Ethernet20")
+        assert "access" in switchport_status
 
         # configure Ethernet20 to access mode again; should give error as it is already in access mode
         result = runner.invoke(config.config.commands["switchport"].commands["mode"], ["access", "Ethernet20"], obj=db)
@@ -1215,6 +1248,10 @@ def test_config_add_del_vlan_and_vlan_member_with_switchport_modes(self, mock_re
         print(result.output)
         traceback.print_tb(result.exc_info[2])
         assert result.exit_code == 0
+        result = runner.invoke(show.cli.commands["interfaces"].commands["switchport"].commands["status"], obj=db)
+        print(result.output)
+        switchport_status = get_intf_switchport_status(self, result.output, "Ethernet20")
+        assert "trunk" in switchport_status
 
         # show output
         result = runner.invoke(show.cli.commands["vlan"].commands["brief"], [], obj=db)
@@ -1256,6 +1293,11 @@ def test_config_add_del_vlan_and_vlan_member_with_switchport_modes(self, mock_re
         assert result.exit_code == 0
         assert "Ethernet20 switched to routed mode" in result.output
 
+        result = runner.invoke(show.cli.commands["interfaces"].commands["switchport"].commands["status"], obj=db)
+        print(result.output)
+        switchport_status = get_intf_switchport_status(self, result.output, "Ethernet20")
+        assert "routed" in switchport_status
+
         # del 1001
         result = runner.invoke(config.config.commands["vlan"].commands["del"], ["1001"], obj=db)
         print(result.exit_code)
@@ -1287,6 +1329,11 @@ def test_config_add_del_with_switchport_modes_changes_output(
         assert result.exit_code == 0
         assert "Ethernet20 switched to trunk mode" in result.output
 
+        result = runner.invoke(show.cli.commands["interfaces"].commands["switchport"].commands["status"], obj=db)
+        print(result.output)
+        switchport_status = get_intf_switchport_status(self, result.output, "Ethernet20")
+        assert "trunk" in switchport_status
+
         # add Ethernet64 to vlan 1001 but Ethernet64 is in routed mode will give error
         result = runner.invoke(config.config.commands["vlan"].commands["member"].commands["add"],
                                ["1001", "Ethernet64"], obj=db)
@@ -1302,6 +1349,10 @@ def test_config_add_del_with_switchport_modes_changes_output(
         print(result.output)
         assert result.exit_code == 0
         assert "Ethernet64 switched to trunk mode" in result.output
+        result = runner.invoke(show.cli.commands["interfaces"].commands["switchport"].commands["status"], obj=db)
+        print(result.output)
+        switchport_status = get_intf_switchport_status(self, result.output, "Ethernet20")
+        assert "trunk" in switchport_status
 
         # add Ethernet64 to vlan 1001
         result = runner.invoke(config.config.commands["vlan"].commands["member"].commands["add"],
@@ -1332,6 +1383,11 @@ def test_config_add_del_with_switchport_modes_changes_output(
         assert result.exit_code == 0
         assert "Ethernet64 switched to access mode" in result.output
 
+        result = runner.invoke(show.cli.commands["interfaces"].commands["switchport"].commands["status"], obj=db)
+        print(result.output)
+        switchport_status = get_intf_switchport_status(self, result.output, "Ethernet64")
+        assert "access" in switchport_status
+
         # show output
         result = runner.invoke(show.cli.commands["vlan"].commands["brief"], [], obj=db)
         print(result.exit_code)