169 lines
7.6 KiB
Python
169 lines
7.6 KiB
Python
|
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 ****")
|
||
|
|
"""
|