Dynamic Port Breakout system test cases using CLI command

tests/conftest.py

@@ -1134,6 +1134,23 @@ def get_state_db(self) -> DVSDatabase:
 
         return self.state_db
 
+    def get_fvs_dict(self, fvs):
+        fvs_dict = {}
+        for fv in fvs:
+            fvs_dict.update({fv[0]:fv[1]})
+        return fvs_dict
+
+    def change_port_breakout_mode(self, intf_name, target_mode, options=""):
+        cmd = "config interface breakout %s %s -y %s"%(intf_name, target_mode, options)
+        self.runcmd(cmd)
+        time.sleep(2)
+
+    def verify_port_breakout_mode(self, intf_name, current_mode):
+        brkout_cfg_tbl = swsscommon.Table(self.cdb, "BREAKOUT_CFG")
+        (status, fvs) = brkout_cfg_tbl.get(intf_name)
+        assert(status == True)
+        fvs_dict = self.get_fvs_dict(fvs)
+        assert(fvs_dict["brkout_mode"] == current_mode)
 
 class DockerVirtualChassisTopology:
     def __init__(
@@ -1415,7 +1432,7 @@ def get_chassis_instance_port_statuses(self):
                 chassis_container_name = device_info["hostname"] + "." + self.ns
 
                 port_info = config["PORT"]
-            
+
             for port, config in port_info.items():
                 if "admin_status" not in config:
                     continue
@@ -1424,13 +1441,13 @@ def get_chassis_instance_port_statuses(self):
                     instance_to_port_status_map[chassis_container_name] = []
 
                 instance_to_port_status_map[chassis_container_name].append((port, config.get("admin_status")))
-            
+
             return instance_to_port_status_map
 
     def handle_chassis_connections(self):
         if self.oper != "create":
             return
-        
+
         instance_to_port_status_map = self.get_chassis_instance_port_statuses()
         for chassis_instance, port_statuses in instance_to_port_status_map.items():
             if chassis_instance not in self.dvss:
@@ -1480,7 +1497,6 @@ def verify_vct(self):
         print("vct verifications passed ? %s" % (ret1 and ret2))
         return ret1 and ret2
 
-
 @pytest.yield_fixture(scope="module")
 def dvs(request) -> DockerVirtualSwitch:
     if sys.version_info[0] < 3:
@@ -1586,12 +1602,10 @@ def create_dpb_config_file(dvs):
     cmd = "cp /tmp/dpb_config_db.json /etc/sonic/config_db.json"
     dvs.runcmd(cmd)
 
-
 def remove_dpb_config_file(dvs):
     cmd = "mv /etc/sonic/config_db.json.bak /etc/sonic/config_db.json"
     dvs.runcmd(cmd)
 
-
 @pytest.yield_fixture(scope="module")
 def dpb_setup_fixture(dvs):
     create_dpb_config_file(dvs)

tests/dvslib/dvs_acl.py

@@ -61,7 +61,7 @@ def create_acl_table(
         table_attrs = {
             "policy_desc": table_name,
             "type": table_type,
-            "ports": ",".join(ports)
+            "ports@": ",".join(ports)
         }
 
         if stage:
@@ -95,7 +95,7 @@ def update_acl_table_port_list(self, table_name: str, ports: List[str]) -> None:
             table_name: The name of the ACL table to update.
             ports: The new list of ports to bind to the ACL table.
         """
-        table_attrs = {"ports": ",".join(ports)}
+        table_attrs = {"ports@": ",".join(ports)}
         self.config_db.update_entry(self.CDB_ACL_TABLE_NAME, table_name, table_attrs)
 
     def remove_acl_table(self, table_name: str) -> None:

tests/dvslib/dvs_vlan.py

@@ -8,9 +8,9 @@ def __init__(self, adb, cdb, sdb, cntrdb, appdb):
         self.counters_db = cntrdb
         self.app_db = appdb
 
-    def create_vlan(self, vlan):
-        vlan = "Vlan{}".format(vlan)
-        vlan_entry = {"vlanid": vlan}
+    def create_vlan(self, vlanID):
+        vlan = "Vlan{}".format(vlanID)
+        vlan_entry = {"vlanid": vlanID}
         self.config_db.create_entry("VLAN", vlan, vlan_entry)
 
     def remove_vlan(self, vlan):

tests/port_dpb.py

@@ -225,7 +225,11 @@ def verify_asic_db(self):
             assert(fvs_dict['SAI_PORT_ATTR_HW_LANE_LIST'] == self.get_lanes_asic_db_str())
         assert(fvs_dict['SAI_PORT_ATTR_SPEED'] == str(self.get_speed()))
 
+    def verify_breakout_mode(self, breakout_mode):
+        self._dvs.get_config_db().wait_for_field_match("BREAKOUT_CFG", self.get_name(), {"brkout_mode": breakout_mode})
+
 class DPB():
+    MAX_LANES = 4
     def breakin(self, dvs, port_names):
         child_ports = []
         for pname in port_names:
@@ -313,3 +317,34 @@ def change_speed_and_verify(self, dvs, port_names, speed = 100000):
             p.verify_app_db()
             time.sleep(1)
             p.verify_asic_db()
+
+    def verify_port_breakout_mode(self, dvs, port_name, breakout_mode):
+        p = Port(dvs, port_name)
+        p.verify_breakout_mode(breakout_mode)
+
+    def get_child_ports(self, root_port, breakout_mode):
+
+        if '+' not in breakout_mode:
+            return self._get_child_ports(root_port, breakout_mode, self.MAX_LANES)
+
+        modes = breakout_mode.split('+')
+        child_ports = []
+        root_port_num = int(root_port.split('Ethernet')[1])
+        for mode in modes:
+            lanes = int(mode.split('(')[1].split(')')[0])
+            mode = mode.split('(')[0]
+            child_ports = child_ports + self._get_child_ports(root_port, mode, lanes)
+            root_port_num = root_port_num + lanes
+            root_port = 'Ethernet' + str(root_port_num)
+
+        return child_ports
+
+    def _get_child_ports(self, root_port, breakout_mode, lanes):
+        count = int(breakout_mode.split('x')[0])
+        port = int(root_port.split('Ethernet')[1])
+
+        child_ports = []
+        jump = int(lanes/count)
+        for i in range(0, lanes, jump):
+            child_ports.append('Ethernet{}'.format(port+i))
+        return child_ports