from swsscommon import swsscommon import time import os import json import ast import pytest import collections @pytest.mark.usefixtures('dpb_setup_fixture') class TestBreakoutCli(object): def setup_db(self, dvs): self.cdb = swsscommon.DBConnector(4, dvs.redis_sock, 0) def read_Json(self, dvs): test_dir = os.path.dirname(os.path.realpath(__file__)) sample_output_file = os.path.join(test_dir, 'sample_output', 'sample_new_port_config.json') with open(sample_output_file, 'rb') as fh: fh_data = json.load(fh) if not fh_data: return False expected = ast.literal_eval(json.dumps(fh_data)) return expected def breakout(self, dvs, interface, brkout_mode): (exitcode, result) = dvs.runcmd("config interface breakout {} {} -y".format(interface, brkout_mode)) if result.strip("\n")[0] == "[ERROR] Breakout feature is not available without platform.json file" : pytest.skip("**** This test is not needed ****") root_dir = os.path.dirname('/') (exitcode, output_dict) = dvs.runcmd("jq '.' new_port_config.json") if output_dict is None: raise Exception("Breakout output cant be None") output_dict = ast.literal_eval(output_dict.strip()) return output_dict # Check Initial Brakout Mode def test_InitialBreakoutMode(self, dvs, testlog): self.setup_db(dvs) output_dict = {} brkoutTbl = swsscommon.Table(self.cdb, "BREAKOUT_CFG") brkout_entries = brkoutTbl.getKeys() assert len(brkout_entries) == 32 for key in brkout_entries: (status, fvs) = brkoutTbl.get(key) assert status brkout_mode = fvs[0][1] output_dict[key] = brkout_mode output = collections.OrderedDict(sorted(output_dict.items(), key=lambda t: t[0])) expected_dict = \ {'Ethernet8': '1x100G[40G]', 'Ethernet0': '1x100G[40G]', 'Ethernet4': '1x100G[40G]', \ 'Ethernet108': '1x100G[40G]', 'Ethernet100': '1x100G[40G]', 'Ethernet104': '1x100G[40G]', \ 'Ethernet68': '1x100G[40G]', 'Ethernet96': '1x100G[40G]', 'Ethernet124': '1x100G[40G]', \ 'Ethernet92': '1x100G[40G]', 'Ethernet120': '1x100G[40G]', 'Ethernet52': '1x100G[40G]', \ 'Ethernet56': '1x100G[40G]', 'Ethernet76': '1x100G[40G]', 'Ethernet72': '1x100G[40G]', \ 'Ethernet32': '1x100G[40G]', 'Ethernet16': '1x100G[40G]', 'Ethernet36': '1x100G[40G]', \ 'Ethernet12': '1x100G[40G]', 'Ethernet28': '1x100G[40G]', 'Ethernet88': '1x100G[40G]', \ 'Ethernet116': '1x100G[40G]', 'Ethernet80': '1x100G[40G]', 'Ethernet112': '1x100G[40G]', \ 'Ethernet84': '1x100G[40G]', 'Ethernet48': '1x100G[40G]', 'Ethernet44': '1x100G[40G]', \ 'Ethernet40': '1x100G[40G]', 'Ethernet64': '1x100G[40G]', 'Ethernet60': '1x100G[40G]', \ 'Ethernet20': '1x100G[40G]', 'Ethernet24': '1x100G[40G]'} expected = collections.OrderedDict(sorted(expected_dict.items(), key=lambda t: t[0])) assert output == expected # Breakout Cli Test Mode def test_breakout_modes(self, dvs): expected = self.read_Json(dvs) assert expected print("**** Breakout Cli test Starts ****") output_dict = self.breakout(dvs, 'Ethernet0', '2x50G') expected_dict = expected["Ethernet0_2x50G"] assert output_dict == expected_dict print("**** 1X100G --> 2x50G passed ****") output_dict = self.breakout(dvs, 'Ethernet4', '4x25G[10G]') expected_dict = expected["Ethernet4_4x25G"] assert output_dict == expected_dict print("**** 1X100G --> 4x25G[10G] passed ****") output_dict = self.breakout(dvs, 'Ethernet8', '2x25G(2)+1x50G(2)') expected_dict = expected["Ethernet8_2x25G_1x50G"] assert output_dict == expected_dict print("**** 1X100G --> 2x25G(2)+1x50G(2) passed ****") output_dict = self.breakout(dvs, 'Ethernet12', '1x50G(2)+2x25G(2)') expected_dict = expected["Ethernet12_1x50G_2x25G"] assert output_dict == expected_dict print("**** 1X100G --> 1x50G(2)+2x25G(2) passed ****") # TODOFIX: remove comments once #4442 PR got merged and # yang model for DEVICE_METADATA becomes available. # As below test cases are dependent on DEVICE_METADATA to go # from a non-default breakout mode to a different breakout mode. """ output_dict = self.breakout(dvs, 'Ethernet0', '1x100G[40G]') expected_dict = expected["Ethernet0_1x100G"] assert output_dict == expected_dict print("**** 2x50G --> 1x100G[40G] passed ****") output_dict = self.breakout(dvs, 'Ethernet0', '4x25G[10G]') expected_dict = expected["Ethernet0_4x25G"] assert output_dict == expected_dict print("**** 1X100G --> 4x25G[10G] passed ****") output_dict = self.breakout(dvs, 'Ethernet0', '1x100G[40G]') expected_dict = expected["Ethernet0_1x100G"] assert output_dict == expected_dict print("**** 4x25G[10G] --> 1x100G[40G] passed ****") output_dict = self.breakout(dvs, 'Ethernet4', '2x50G') print("**** 1X100G --> 2x50G mode change ****") output_dict = self.breakout(dvs, 'Ethernet4', '4x25G[10G]') expected_dict = expected["Ethernet4_4x25G"] assert output_dict == expected_dict print("**** 2X50G --> 4x25G[10G] passed ****") output_dict = self.breakout(dvs, 'Ethernet4', '2x50G') expected_dict = expected["Ethernet4_2x50G"] assert output_dict == expected_dict print("**** 4x25G[10G] --> 2X50G passed ****") output_dict = self.breakout(dvs, 'Ethernet4', '1x100G[40G]') print("**** 2x50G -- > 1X100G mode change ****") output_dict = self.breakout(dvs, 'Ethernet0', '2x25G(2)+1x50G(2)') expected_dict = expected["Ethernet0_2x25G_1x50G"] assert output_dict == expected_dict print("**** 1x100G[40G] --> 2x25G(2)+1x50G(2) passed ****") output_dict = self.breakout(dvs, 'Ethernet0', '1x100G[40G]') expected_dict = expected["Ethernet0_1x100G"] assert output_dict == expected_dict print("**** 2x25G(2)+1x50G(2) --> 1x100G[40G] passed ****") output_dict = self.breakout(dvs, 'Ethernet0', '1x50G(2)+2x25G(2)') expected_dict = expected["Ethernet0_1x50G_2x25G"] assert output_dict == expected_dict print("**** 1x100G[40G] --> 1x50G(2)+2x25G(2) passed ****") output_dict = self.breakout(dvs, 'Ethernet0', '1x100G[40G]') expected_dict = expected["Ethernet0_1x100G"] assert output_dict == expected_dict print("**** 1x50G(2)+2x25G(2) --> 1x100G[40G] passed ****") output_dict = self.breakout(dvs, 'Ethernet8', '2x50G') print("**** 1x100G[40G] --> 2x50G mode change ****") output_dict = self.breakout(dvs, 'Ethernet8', '1x50G(2)+2x25G(2)') expected_dict = expected["Ethernet8_1x50G_2x25G"] assert output_dict == expected_dict print("**** 2x50G --> 2x25G(2)+1x50G(2) passed ****") output_dict = self.breakout(dvs, 'Ethernet8', '2x25G(2)+1x50G(2)') expected_dict = expected["Ethernet8_2x25G_1x50G"] assert output_dict == expected_dict print("**** 1x50G(2)+2x25G(2) --> 2x25G(2)+1x50G(2) passed ****") output_dict = self.breakout(dvs, 'Ethernet8', '1x100G[40G]') expected_dict = expected["Ethernet8_1x100G"] assert output_dict == expected_dict print("**** 2x25G(2)+1x50G(2) --> 1x100G[40G] passed ****") """