diff --git a/scripts/intfutil b/scripts/intfutil index 69472760d8..daf06ab04d 100755 --- a/scripts/intfutil +++ b/scripts/intfutil @@ -80,33 +80,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): @@ -307,6 +314,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 @@ -354,7 +386,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 """ @@ -367,8 +399,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 @@ -484,7 +516,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), @@ -501,7 +533,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), @@ -523,13 +555,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() diff --git a/tests/vlan_test.py b/tests/vlan_test.py index 8b0ce1b617..290d1ff081 100644 --- a/tests/vlan_test.py +++ b/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,6 +767,10 @@ 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"], @@ -755,6 +778,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, "PortChannel1001") + assert "routed" in switchport_status # Configure PortChannel1001 to trunk mode result = runner.invoke(config.config.commands["switchport"].commands["mode"], @@ -762,6 +788,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, "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)