[xcvrd] Add Initial CPO Support

sonic-xcvrd/tests/test_cpo.py

@@ -0,0 +1,102 @@
+from unittest.mock import MagicMock, patch
+
+from xcvrd.xcvrd_utilities import common
+
+
+class TestPortDeviceResolver:
+    def test_is_cpo_port_no_chassis(self):
+        with patch.object(common, 'platform_chassis', None):
+            assert common.is_cpo_port(0) is False
+
+    def test_is_cpo_port_true_when_cpo_present(self):
+        chassis = MagicMock()
+        chassis.get_cpo.return_value = MagicMock()
+        with patch.object(common, 'platform_chassis', chassis):
+            assert common.is_cpo_port(3) is True
+        chassis.get_cpo.assert_called_with(3)
+
+    def test_is_cpo_port_false_when_not_cpo(self):
+        chassis = MagicMock()
+        chassis.get_cpo.return_value = None
+        with patch.object(common, 'platform_chassis', chassis):
+            assert common.is_cpo_port(3) is False
+
+    def test_is_cpo_port_swallows_not_implemented(self):
+        chassis = MagicMock()
+        chassis.get_cpo.side_effect = NotImplementedError
+        with patch.object(common, 'platform_chassis', chassis):
+            assert common.is_cpo_port(3) is False
+
+    def test_get_port_device_prefers_cpo(self):
+        chassis = MagicMock()
+        cpo = MagicMock()
+        chassis.get_cpo.return_value = cpo
+        with patch.object(common, 'platform_chassis', chassis):
+            assert common.get_port_device(1) is cpo
+        chassis.get_sfp.assert_not_called()
+
+    def test_get_port_device_falls_back_to_sfp(self):
+        chassis = MagicMock()
+        sfp = MagicMock()
+        chassis.get_cpo.return_value = None
+        chassis.get_sfp.return_value = sfp
+        with patch.object(common, 'platform_chassis', chassis):
+            assert common.get_port_device(1) is sfp
+
+    def test_get_port_device_none_when_unavailable(self):
+        with patch.object(common, 'platform_chassis', None):
+            assert common.get_port_device(1) is None
+
+
+class TestObjDictAccessors:
+    def _make_port_mapping(self, physical_ports=(0, 1, 2)):
+        port_mapping = MagicMock()
+        port_mapping.physical_to_logical = {p: ['Ethernet{}'.format(p * 4)] for p in physical_ports}
+        return port_mapping
+
+    def _make_obj_dict(self):
+        return {0: MagicMock(), 1: MagicMock(), 2: MagicMock()}
+
+    def test_accessors_are_complementary(self):
+        objs = self._make_obj_dict()
+        port_mapping = self._make_port_mapping()
+        with patch.object(common, 'is_cpo_port', side_effect=lambda p: p in (1,)), \
+             patch.object(common, 'get_port_device', side_effect=lambda p: objs[p]):
+            cpo = common.get_cpo_obj_dict(port_mapping)
+            pluggable = common.get_pluggable_obj_dict(port_mapping)
+        assert set(cpo) | set(pluggable) == set(objs)
+        assert set(cpo) & set(pluggable) == set()
+        assert set(cpo) == {1}
+        assert cpo[1] is objs[1]
+
+    def test_all_pluggable_when_no_cpo(self):
+        objs = self._make_obj_dict()
+        port_mapping = self._make_port_mapping()
+        with patch.object(common, 'is_cpo_port', return_value=False), \
+             patch.object(common, 'get_port_device', side_effect=lambda p: objs[p]):
+            assert common.get_cpo_obj_dict(port_mapping) == {}
+            assert set(common.get_pluggable_obj_dict(port_mapping)) == {0, 1, 2}
+
+    def test_accessors_return_empty_without_port_mapping(self):
+        with patch.object(common, 'get_port_device') as mock_get_port_device:
+            assert common.get_cpo_obj_dict(None) == {}
+            assert common.get_pluggable_obj_dict(None) == {}
+
+            port_mapping = MagicMock()
+            port_mapping.physical_to_logical = None
+            assert common.get_cpo_obj_dict(port_mapping) == {}
+            assert common.get_pluggable_obj_dict(port_mapping) == {}
+        mock_get_port_device.assert_not_called()
+
+    def test_accessors_skip_ports_raising_exceptions(self):
+        objs = self._make_obj_dict()
+
+        def mock_get_port_device(physical_port):
+            if physical_port == 2:
+                raise ValueError("Invalid port")
+            return objs[physical_port]
+
+        with patch.object(common, 'is_cpo_port', return_value=False), \
+             patch.object(common, 'get_port_device', side_effect=mock_get_port_device):
+            pluggable = common.get_pluggable_obj_dict(self._make_port_mapping())
+        assert set(pluggable.keys()) == {0, 1}

sonic-xcvrd/xcvrd/cmis/cmis_manager_task.py

@@ -46,19 +46,25 @@ class CmisManagerTask(threading.Thread):
     CMIS_MAX_HOST_LANES    = 8
     CMIS_EXPIRATION_BUFFER_MS = 2
 
-    def __init__(self, namespaces, port_mapping, main_thread_stop_event, skip_cmis_mgr=False, platform_chassis=None):
+    def __init__(self, namespaces, port_mapping, port_obj_dict, main_thread_stop_event, skip_cmis_mgr=False):
         threading.Thread.__init__(self)
         self.name = "CmisManagerTask"
         self.exc = None
         self.task_stopping_event = threading.Event()
         self.main_thread_stop_event = main_thread_stop_event
-        self.port_dict = {k: {"asic_id": v} for k, v in port_mapping.logical_to_asic.items()}
+        self.port_dict = {}
+        for lport, asic_id in port_mapping.logical_to_asic.items():
+            entry = {"asic_id": asic_id}
+            pports = port_mapping.get_logical_to_physical(lport)
+            if pports:
+                entry["index"] = pports[0]
+            self.port_dict[lport] = entry
         self.decomm_pending_dict = {}
         self.isPortInitDone = False
         self.isPortConfigDone = False
         self.skip_cmis_mgr = skip_cmis_mgr
         self.namespaces = namespaces
-        self.platform_chassis = platform_chassis
+        self.port_obj_dict = port_obj_dict
         self.xcvr_table_helper = XcvrTableHelper(self.namespaces)
         # Cache of gearbox line lanes dict, refreshed once per task_worker iteration.
         self._gearbox_lanes_dict = None
@@ -123,6 +129,10 @@ def on_port_update_event(self, port_change_event):
         if port_change_event.port_dict is None:
             return
 
+        owner_pport = pport if pport >= 0 else self.port_dict.get(lport, {}).get('index')
+        if owner_pport not in self.port_obj_dict:
+            return
+
         if port_change_event.event_type == port_change_event.PORT_SET:
             if lport not in self.port_dict:
                 self.port_dict[lport] = {"asic_id": port_change_event.asic_id,
@@ -1276,10 +1286,13 @@ def process_single_lport(self, lport, info):
         if pport < 0 or speed == 0 or len(lanes) < 1 or subport < 0:
             return
 
+        if pport not in self.port_obj_dict:
+            return
+
         host_lane_count = self.get_host_lane_count(lport, lanes)
 
         # double-check the HW presence before moving forward
-        sfp = self.platform_chassis.get_sfp(pport)
+        sfp = self.port_obj_dict[pport]
         if not sfp.get_presence():
             self.update_port_transceiver_status_table_sw_cmis_state(lport, CMIS_STATE_REMOVED)
             return
@@ -1355,8 +1368,8 @@ def task_worker(self):
         self.log_notice("Stopped")
 
     def run(self):
-        if self.platform_chassis is None:
-            self.log_notice("Platform chassis is not available, stopping...")
+        if not self.port_obj_dict:
+            self.log_notice("No SFP objects are available, stopping...")
             return
 
         if self.skip_cmis_mgr:
@@ -1370,7 +1383,8 @@ def run(self):
                 self.wait_for_port_config_done(namespace)
 
             for lport in self.port_dict.keys():
-                self.update_port_transceiver_status_table_sw_cmis_state(lport, CMIS_STATE_UNKNOWN)
+                if self.port_dict[lport].get('index') in self.port_obj_dict:
+                    self.update_port_transceiver_status_table_sw_cmis_state(lport, CMIS_STATE_UNKNOWN)
 
             self.task_worker()
         except Exception as e:

sonic-xcvrd/xcvrd/cpo/cpo_manager_task.py

@@ -0,0 +1,13 @@
+#!/usr/bin/env python3
+
+try:
+    from ..cmis.cmis_manager_task import CmisManagerTask
+except ImportError as e:
+    raise ImportError(str(e) + " - required module not found")
+
+
+class CpoManagerTask(CmisManagerTask):
+    def __init__(self, namespaces, port_mapping, port_obj_dict, main_thread_stop_event, skip_cpo_mgr=False):
+        super().__init__(namespaces, port_mapping, port_obj_dict, main_thread_stop_event,
+                         skip_cmis_mgr=skip_cpo_mgr)
+        self.name = "CpoManagerTask"

sonic-xcvrd/xcvrd/cpo/cpo_state_task.py

@@ -0,0 +1,12 @@
+#!/usr/bin/env python3
+
+try:
+    from ..xcvrd import SfpStateUpdateTask
+except ImportError as e:
+    raise ImportError(str(e) + " - required module not found")
+
+
+class CpoStateUpdateTask(SfpStateUpdateTask):
+    def __init__(self, namespaces, port_mapping, port_obj_dict, main_thread_stop_event, sfp_error_event):
+        super().__init__(namespaces, port_mapping, port_obj_dict, main_thread_stop_event, sfp_error_event)
+        self.name = "CpoStateUpdateTask"

sonic-xcvrd/xcvrd/cpo/dom_mgr.py

@@ -0,0 +1,10 @@
+#!/usr/bin/env python3
+
+try:
+    from ..dom.dom_mgr import DomInfoUpdateTask
+except ImportError as e:
+    raise ImportError(str(e) + " - required module not found")
+
+
+class CpoDomInfoUpdateTask(DomInfoUpdateTask):
+    name = "CpoDomInfoUpdateTask"

sonic-xcvrd/xcvrd/dom/dom_mgr.py

@@ -40,15 +40,15 @@ class DomInfoUpdateBase(threading.Thread):
 
     name = ''
 
-    def __init__(self, namespaces, port_mapping, sfp_obj_dict, main_thread_stop_event):
+    def __init__(self, namespaces, port_mapping, port_obj_dict, main_thread_stop_event):
         threading.Thread.__init__(self)
         self.exc = None
         self.task_stopping_event = threading.Event()
         self.main_thread_stop_event = main_thread_stop_event
         self.helper_logger = syslogger.SysLogger(SYSLOG_IDENTIFIER_DOMINFOUPDATETASK, enable_runtime_config=True)
         self.port_mapping = copy.deepcopy(port_mapping)
         self.namespaces = namespaces
-        self.sfp_obj_dict = sfp_obj_dict
+        self.port_obj_dict = port_obj_dict
 
     def log_debug(self, message):
         self.helper_logger.log_debug("{}".format(message))
@@ -147,17 +147,17 @@ class DomInfoUpdateTask(DomInfoUpdateBase):
         {'APPL_DB': 'PORT_TABLE', 'FILTER': ['flap_count']},
     ]
 
-    def __init__(self, namespaces, port_mapping, sfp_obj_dict, main_thread_stop_event, skip_cmis_mgr, dom_update_interval=None):
-        super().__init__(namespaces, port_mapping, sfp_obj_dict, main_thread_stop_event)
+    def __init__(self, namespaces, port_mapping, port_obj_dict, main_thread_stop_event, skip_cmis_mgr, dom_update_interval=None):
+        super().__init__(namespaces, port_mapping, port_obj_dict, main_thread_stop_event)
         self.skip_cmis_mgr = skip_cmis_mgr
         self.link_change_affected_ports = {}
         self.xcvr_table_helper = XcvrTableHelper(self.namespaces)
-        self.xcvrd_utils = XCVRDUtils(self.sfp_obj_dict, self.helper_logger)
-        self.dom_db_utils = DOMDBUtils(self.sfp_obj_dict, self.port_mapping, self.xcvr_table_helper, self.task_stopping_event, self.helper_logger)
+        self.xcvrd_utils = XCVRDUtils(self.port_obj_dict, self.helper_logger)
+        self.dom_db_utils = DOMDBUtils(self.port_obj_dict, self.port_mapping, self.xcvr_table_helper, self.task_stopping_event, self.helper_logger)
         self.db_utils = self.dom_db_utils
-        self.vdm_utils = VDMUtils(self.sfp_obj_dict, self.helper_logger)
-        self.vdm_db_utils = VDMDBUtils(self.sfp_obj_dict, self.port_mapping, self.xcvr_table_helper, self.task_stopping_event, self.helper_logger)
-        self.status_db_utils = StatusDBUtils(self.sfp_obj_dict, self.port_mapping, self.xcvr_table_helper, self.task_stopping_event, self.helper_logger)
+        self.vdm_utils = VDMUtils(self.port_obj_dict, self.helper_logger)
+        self.vdm_db_utils = VDMDBUtils(self.port_obj_dict, self.port_mapping, self.xcvr_table_helper, self.task_stopping_event, self.helper_logger)
+        self.status_db_utils = StatusDBUtils(self.port_obj_dict, self.port_mapping, self.xcvr_table_helper, self.task_stopping_event, self.helper_logger)
         self.dom_update_interval = self.DEFAULT_DOM_INFO_UPDATE_PERIOD_SECS
         if dom_update_interval is not None:
              if dom_update_interval < 0:
@@ -329,6 +329,9 @@ def task_worker(self):
                     self.log_notice("Stop event generated during DOM monitoring loop")
                     break
 
+                if physical_port not in self.port_obj_dict:
+                    continue
+
                 # Get the first logical port name since it corresponds to the first subport
                 # of the breakout group
                 logical_port_name = logical_ports[0]
@@ -443,6 +446,9 @@ def update_port_db_diagnostics_on_link_change(self, physical_port):
         if self.task_stopping_event.is_set():
             return
 
+        if physical_port not in self.port_obj_dict:
+            return
+
         logical_port_list = self.port_mapping.get_physical_to_logical(physical_port)
         if logical_port_list is None:
             self.log_warning("Update DB diagnostics during link change: Unknown physical port index {}".format(physical_port))
@@ -523,14 +529,15 @@ def on_remove_logical_port(self, port_change_event):
                                       self.xcvr_table_helper.get_firmware_info_tbl(port_change_event.asic_id)
                                       ])
 
+
 class DomThermalInfoUpdateTask(DomInfoUpdateBase):
     name = 'DomThermalInfoUpdateTask'
 
-    def __init__(self, namespaces, port_mapping, sfp_obj_dict, main_thread_stop_event, poll_interval):
-        super().__init__(namespaces, port_mapping, sfp_obj_dict, main_thread_stop_event)
+    def __init__(self, namespaces, port_mapping, port_obj_dict, main_thread_stop_event, poll_interval):
+        super().__init__(namespaces, port_mapping, port_obj_dict, main_thread_stop_event)
         self.poll_interval = poll_interval
         self.xcvr_table_helper = XcvrTableHelper(self.namespaces)
-        self.dom_db_utils = DOMDBUtils(self.sfp_obj_dict, self.port_mapping, self.xcvr_table_helper, self.task_stopping_event, self.helper_logger)
+        self.dom_db_utils = DOMDBUtils(self.port_obj_dict, self.port_mapping, self.xcvr_table_helper, self.task_stopping_event, self.helper_logger)
 
     def task_worker(self):
         self.log_notice("Start DOM thermal monitoring loop")
@@ -551,6 +558,9 @@ def task_worker(self):
                continue
 
             for physical_port, logical_ports in self.port_mapping.physical_to_logical.items():
+                if physical_port not in self.port_obj_dict:
+                    continue
+
                 # Get the first logical port name since it corresponds to the first subport
                 # of the breakout group
                 logical_port_name = logical_ports[0]