sonic-buildimage/platform/vs/tests/breakout/test_breakout_cli.py

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 ****")
        """