[gearbox] Add gearbox unit test

tests/conftest.py

@@ -244,7 +244,7 @@ def __init__(
         name: str = None,
         imgname: str = None,
         keeptb: bool = False,
-        fakeplatform: str = None,
+        env: list = [],
         log_path: str = None,
         max_cpu: int = 2,
         forcedvs: bool = None,
@@ -356,8 +356,6 @@ def __init__(
             self.mount = f"/var/run/redis-vs/{self.ctn_sw.name}"
             ensure_system(f"mkdir -p {self.mount}")
 
-            self.environment = [f"fake_platform={fakeplatform}"] if fakeplatform else []
-
             kwargs = {}
             if newctnname:
                 kwargs["name"] = newctnname
@@ -372,7 +370,7 @@ def __init__(
             self.ctn = self.client.containers.run(imgname,
                                                   privileged=True,
                                                   detach=True,
-                                                  environment=self.environment,
+                                                  environment=env,
                                                   network_mode=f"container:{self.ctn_sw.name}",
                                                   cpu_count=max_cpu,
                                                   **kwargs)
@@ -1235,7 +1233,7 @@ def __init__(
         namespace=None,
         imgname=None,
         keeptb=False,
-        fakeplatform=None,
+        env=[],
         log_path=None,
         max_cpu=2,
         forcedvs=None,
@@ -1244,7 +1242,7 @@ def __init__(
         self.ns = namespace
         self.chassbr = "br4chs"
         self.keeptb = keeptb
-        self.fakeplatform = fakeplatform
+        self.env = env
         self.topoFile = topoFile
         self.imgname = imgname
         self.ctninfo = {}
@@ -1303,7 +1301,7 @@ def find_all_ctns(self):
         for ctn in docker.from_env().containers.list():
             if ctn.name.endswith(suffix):
                 self.dvss[ctn.name] = DockerVirtualSwitch(ctn.name, self.imgname, self.keeptb,
-                                                          self.fakeplatform, log_path=ctn.name,
+                                                          self.env, log_path=ctn.name,
                                                           max_cpu=self.max_cpu, forcedvs=self.forcedvs,
                                                           vct=self)
         if self.chassbr is None and len(self.dvss) > 0:
@@ -1421,7 +1419,7 @@ def create_vct_ctn(self, ctndir):
             if ctnname not in self.dvss:
                 self.dvss[ctnname] = DockerVirtualSwitch(name=None, imgname=self.imgname,
                                                          keeptb=self.keeptb,
-                                                         fakeplatform=self.fakeplatform,
+                                                         env=self.env,
                                                          log_path=self.log_path,
                                                          max_cpu=self.max_cpu,
                                                          forcedvs=self.forcedvs,
@@ -1598,18 +1596,18 @@ def manage_dvs(request) -> str:
     buffer_model = request.config.getoption("--buffer_model")
     force_recreate = request.config.getoption("--force-recreate-dvs")
     dvs = None
-    curr_fake_platform = None   # lgtm[py/unused-local-variable]
+    curr_dvs_env = [] # lgtm[py/unused-local-variable]
 
     if using_persistent_dvs and force_recreate:
         pytest.fail("Options --dvsname and --force-recreate-dvs are mutually exclusive")
 
-    def update_dvs(log_path, new_fake_platform=None):
+    def update_dvs(log_path, new_dvs_env=[]):
         """
         Decides whether or not to create a new DVS
 
         Create a new the DVS in the following cases:
         1. CLI option `--force-recreate-dvs` was specified (recreate for every module)
-        2. The fake_platform has changed (this can only be set at container creation,
+        2. The dvs_env has changed (this can only be set at container creation,
            so it is necessary to spin up a new DVS)
         3. No DVS currently exists (i.e. first time startup)
 
@@ -1618,18 +1616,18 @@ def update_dvs(log_path, new_fake_platform=None):
         Returns:
             (DockerVirtualSwitch) a DVS object
         """
-        nonlocal curr_fake_platform, dvs
+        nonlocal curr_dvs_env, dvs
         if force_recreate or \
-           new_fake_platform != curr_fake_platform or \
+           new_dvs_env != curr_dvs_env or \
            dvs is None:
 
             if dvs is not None:
                 dvs.get_logs()
                 dvs.destroy()
 
-            dvs = DockerVirtualSwitch(name, imgname, keeptb, new_fake_platform, log_path, max_cpu, forcedvs, buffer_model = buffer_model)
+            dvs = DockerVirtualSwitch(name, imgname, keeptb, new_dvs_env, log_path, max_cpu, forcedvs, buffer_model = buffer_model)
 
-            curr_fake_platform = new_fake_platform
+            curr_dvs_env = new_dvs_env
 
         else:
             # First generate GCDA files for GCov
@@ -1654,11 +1652,11 @@ def update_dvs(log_path, new_fake_platform=None):
 
 @pytest.yield_fixture(scope="module")
 def dvs(request, manage_dvs) -> DockerVirtualSwitch:
-    fakeplatform = getattr(request.module, "DVS_FAKE_PLATFORM", None)
+    dvs_env = getattr(request.module, "DVS_ENV", [])
     name = request.config.getoption("--dvsname")
     log_path = name if name else request.module.__name__
 
-    return manage_dvs(log_path, fakeplatform)
+    return manage_dvs(log_path, dvs_env)
 
 @pytest.yield_fixture(scope="module")
 def vct(request):
@@ -1669,11 +1667,11 @@ def vct(request):
     imgname = request.config.getoption("--imgname")
     max_cpu = request.config.getoption("--max_cpu")
     log_path = vctns if vctns else request.module.__name__
-    fakeplatform = getattr(request.module, "DVS_FAKE_PLATFORM", None)
+    dvs_env = getattr(request.module, "DVS_ENV", [])
     if not topo:
         # use ecmp topology as default
         topo = "virtual_chassis/chassis_with_ecmp_neighbors.json"
-    vct = DockerVirtualChassisTopology(vctns, imgname, keeptb, fakeplatform, log_path, max_cpu,
+    vct = DockerVirtualChassisTopology(vctns, imgname, keeptb, dvs_env, log_path, max_cpu,
                                        forcedvs, topo)
     yield vct
     vct.get_logs(request.module.__name__)

tests/test_gearbox.py

@@ -0,0 +1,143 @@
+# This test suite covers the functionality of gearbox feature
+import time
+import os
+import pytest
+from swsscommon import swsscommon
+from dvslib.dvs_database import DVSDatabase
+from dvslib.dvs_common import PollingConfig, wait_for_result
+
+# module specific dvs env variables
+DVS_ENV = ["HWSKU=brcm_gearbox_vs"]
+
+class Gearbox(object):
+    def __init__(self, dvs):
+        db = swsscommon.DBConnector(swsscommon.APPL_DB, dvs.redis_sock, 0)
+        t = swsscommon.Table(db, "_GEARBOX_TABLE")
+        assert len(t.getKeys()) > 0
+        sr = t.getTableNameSeparator()
+
+        # "_GEARBOX_TABLE:phy:1"
+        # "_GEARBOX_TABLE:phy:1:ports:0"
+        # "_GEARBOX_TABLE:phy:1:lanes:200"
+        self.phys = {}
+        phy_table  = swsscommon.Table(db, sr.join([t.getKeyName(""), "phy"]))
+        for i in [x for x in phy_table.getKeys() if sr not in x]:
+            (status, fvs) = phy_table.get(i)
+            assert status == True
+            self.phys[i] = {"attrs" : dict(fvs)}
+
+            port_table = swsscommon.Table(db, sr.join([phy_table.getKeyName(i), "ports"]))
+            port_list = [x for x in port_table.getKeys() if sr not in x]
+            self.phys[i]["port_table"] = port_table
+            self.phys[i]["ports"] = {}
+            for j in port_list:
+                (status, fvs) = port_table.get(j)
+                assert status == True
+                self.phys[i]["ports"][j] = dict(fvs)
+
+            lane_table = swsscommon.Table(db, sr.join([phy_table.getKeyName(i), "lanes"]))
+            lane_list = [x for x in lane_table.getKeys() if sr not in x]
+            self.phys[i]["lanes"] = {}
+            for j in lane_list:
+                (status, fvs) = lane_table.get(j)
+                assert status == True
+                self.phys[i]["lanes"][j] = dict(fvs)
+
+        # "_GEARBOX_TABLE:interface:0"
+        self.interfaces = {}
+        intf_table = swsscommon.Table(db, sr.join([t.getKeyName(""), "interface"]))
+        for i in [x for x in intf_table.getKeys() if sr not in x]:
+            (status, fvs) = intf_table.get(i)
+            assert status == True
+            self.interfaces[i] = {"attrs" : dict(fvs)}
+
+    def SanityCheck(self, dvs, testlog):
+        """
+        Verify data integrity of Gearbox objects in APPL_DB
+        """
+        for i in self.interfaces:
+            phy_id = self.interfaces[i]["attrs"]["phy_id"]
+            assert phy_id in self.phys
+            assert self.interfaces[i]["attrs"]["index"] in self.phys[phy_id]["ports"]
+
+            for lane in self.interfaces[i]["attrs"]["system_lanes"].split(','):
+                assert lane in self.phys[phy_id]["lanes"]
+            for lane in self.interfaces[i]["attrs"]["line_lanes"].split(','):
+                assert lane in self.phys[phy_id]["lanes"]
+
+class GBAsic(DVSDatabase):
+    def __init__(self, db_id: int, connector: str, gearbox: Gearbox):
+        DVSDatabase.__init__(self, db_id, connector)
+        self.gearbox = gearbox
+        self.ports = {}
+        self._wait_for_gb_asic_db_to_initialize()
+
+        for connector in self.get_keys("ASIC_STATE:SAI_OBJECT_TYPE_PORT_CONNECTOR"):
+            fvs = self.get_entry("ASIC_STATE:SAI_OBJECT_TYPE_PORT_CONNECTOR", connector)
+            system_port_oid = fvs.get("SAI_PORT_CONNECTOR_ATTR_SYSTEM_SIDE_PORT_ID")
+            line_port_oid = fvs.get("SAI_PORT_CONNECTOR_ATTR_LINE_SIDE_PORT_ID")
+
+            fvs = self.get_entry("ASIC_STATE:SAI_OBJECT_TYPE_PORT", system_port_oid)
+            system_lanes = fvs.get("SAI_PORT_ATTR_HW_LANE_LIST").split(":")[-1]
+
+            fvs = self.get_entry("ASIC_STATE:SAI_OBJECT_TYPE_PORT", line_port_oid)
+            line_lanes = fvs.get("SAI_PORT_ATTR_HW_LANE_LIST").split(":")[-1]
+
+            for i in self.gearbox.interfaces:
+                intf = self.gearbox.interfaces[i]
+                if intf["attrs"]["system_lanes"] == system_lanes:
+                    assert intf["attrs"]["line_lanes"] == line_lanes
+                    self.ports[intf["attrs"]["index"]] = (system_port_oid, line_port_oid)
+
+        assert len(self.ports) == len(self.gearbox.interfaces)
+
+    def _wait_for_gb_asic_db_to_initialize(self) -> None:
+        """Wait up to 30 seconds for the default fields to appear in ASIC DB."""
+        def _verify_db_contents():
+            if len(self.get_keys("ASIC_STATE:SAI_OBJECT_TYPE_SWITCH")) != \
+               len(self.gearbox.phys):
+                return (False, None)
+
+            if len(self.get_keys("ASIC_STATE:SAI_OBJECT_TYPE_PORT")) != \
+               2 * len(self.gearbox.interfaces):
+                return (False, None)
+
+            if len(self.get_keys("ASIC_STATE:SAI_OBJECT_TYPE_PORT_CONNECTOR")) != \
+               len(self.gearbox.interfaces):
+                return (False, None)
+
+            return (True, None)
+
+        # Verify that GB ASIC DB has been fully initialized
+        init_polling_config = PollingConfig(2, 30, strict=True)
+        wait_for_result(_verify_db_contents, init_polling_config)
+
+
+class TestGearbox(object):
+    def test_GearboxSanity(self, dvs, testlog):
+        Gearbox(dvs).SanityCheck(dvs, testlog)
+
+    def test_GbAsicFEC(self, dvs, testlog):
+        gbasic = GBAsic(swsscommon.GB_ASIC_DB, dvs.redis_sock, Gearbox(dvs))
+
+        # set fec rs on port 0 of phy 1
+        fvs = swsscommon.FieldValuePairs([("system_fec","rs")])
+        gbasic.gearbox.phys["1"]["port_table"].set("0", fvs)
+        fvs = swsscommon.FieldValuePairs([("line_fec","rs")])
+        gbasic.gearbox.phys["1"]["port_table"].set("0", fvs)
+
+        """FIXME: uncomment it after GearboxOrch is implemented
+        # validate if fec rs is pushed to system/line port in gb asic db
+        system_port_oid, line_port_oid = gbasic.ports["0"]
+        expected_fields = {"SAI_PORT_ATTR_FEC_MODE":"SAI_PORT_FEC_MODE_RS"}
+        gbasic.wait_for_field_match("ASIC_STATE:SAI_OBJECT_TYPE_PORT", \
+                system_port_oid, expected_fields)
+        gbasic.wait_for_field_match("ASIC_STATE:SAI_OBJECT_TYPE_PORT", \
+                line_port_oid, expected_fields)
+        """
+
+
+# Add Dummy always-pass test at end as workaroud
+# for issue when Flaky fail on final test it invokes module tear-down before retrying
+def test_nonflaky_dummy():
+    pass

tests/test_mirror_ipv6_combined.py

@@ -3,7 +3,7 @@
 
 from swsscommon import swsscommon
 
-DVS_FAKE_PLATFORM = "broadcom"
+DVS_ENV = ["fake_platform=broadcom"]
 
 
 class TestMirror(object):

tests/test_mirror_ipv6_separate.py

@@ -3,7 +3,7 @@
 
 from swsscommon import swsscommon
 
-DVS_FAKE_PLATFORM = "mellanox"
+DVS_ENV = ["fake_platform=mellanox"]
 
 
 class TestMirror(object):