[CPO] Extend ChassisBase to support CPO ports

sonic_platform_base/chassis_base.py

@@ -6,6 +6,7 @@
 """
 
 import sys
+from sonic_py_common import device_info
 from . import device_base
 from . import sfp_base
 
@@ -62,9 +63,18 @@ def __init__(self):
         self._current_sensor_list = []
 
         # List of SfpBase-derived objects representing all sfps
-        # available on the chassis
+        # available on the chassis, indexed by physical port.
         self._sfp_list = []
 
+        # List of CpoBase-derived objects representing all CPO ports
+        # available on the chassis, indexed by physical port.
+        self._cpo_list = []
+
+        # Set once the port lists have been checked for consistency. The check
+        # is deferred until the lists are first accessed, since subclasses
+        # populate them after ChassisBase.__init__ has returned.
+        self._port_lists_validated = False
+
         # Object derived from WatchdogBase for interacting with hardware watchdog
         self._watchdog = None
 
@@ -80,6 +90,12 @@ def __init__(self):
         # SED (Self-Encrypting Drive) password management
         self._sed_mgmt = None
 
+        # On platforms that provide a cpo.json file, populate self._cpo_list
+        # based on the device topology described in that file
+        cpo_data = device_info.get_cpo_data()
+        if cpo_data:
+            self.construct_cpo_devices(cpo_data)
+
     def get_base_mac(self):
         """
         Retrieves the base MAC address for the chassis
@@ -701,14 +717,61 @@ def get_current_sensor(self, index):
     # SFP methods
     ##############################################
 
+    def construct_cpo_devices(self, cpo_data):
+        """
+        Construct objects representing the devices driving traffic through
+        a front panel port on the chassis based on topology data in
+        cpo.json
+
+        Subclasses should implement this method on platforms that provide
+        a cpo.json file.
+
+        Args:
+            cpo_data: device topology data parsed from cpo.json
+        """
+        raise NotImplementedError
+
+    def _validate_port_lists(self):
+        """
+        Check that _sfp_list and _cpo_list follow the indexing convention:
+        both are indexed by physical front panel port, with None at the
+        indices driven by the other technology. A platform which has only one
+        kind of port need only populate the corresponding list.
+        """
+        if self._port_lists_validated:
+            return
+
+        if not self._sfp_list or not self._cpo_list:
+            # This is a platform that only has one type of transceiver technology.
+            # We do not need to perform validation against both lists, because only
+            # one will be used.
+            return
+
+        if len(self._sfp_list) != len(self._cpo_list):
+            raise RuntimeError(
+                "_sfp_list (length {}) and _cpo_list (length {}) must be the "
+                "same length: both are indexed by physical front panel port, "
+                "with None at the indices driven by the other technology".format(
+                    len(self._sfp_list), len(self._cpo_list)))
+
+        for index, (sfp, cpo) in enumerate(zip(self._sfp_list, self._cpo_list)):
+            if sfp is not None and cpo is not None:
+                raise RuntimeError(
+                    "Physical port {} has both an SFP and a CPO object; each "
+                    "port must appear in exactly one of _sfp_list and "
+                    "_cpo_list".format(index))
+
+        self._port_lists_validated = True
+
     def get_num_sfps(self):
         """
         Retrieves the number of sfps available on this chassis
 
         Returns:
             An integer, the number of sfps available on this chassis
         """
-        return len(self._sfp_list)
+        self._validate_port_lists()
+        return sum(1 for sfp in self._sfp_list if sfp is not None)
 
     def get_all_sfps(self):
         """
@@ -718,6 +781,7 @@ def get_all_sfps(self):
             A list of objects derived from SfpBase representing all sfps
             available on this chassis
         """
+        self._validate_port_lists()
         return [ sfp for sfp in self._sfp_list if sfp is not None ]
 
     def get_sfp(self, index):
@@ -732,10 +796,12 @@ def get_sfp(self, index):
                    0 for Ethernet0, 1 for Ethernet4 and so on for another platform.
 
         Returns:
-            An object dervied from SfpBase representing the specified sfp
+            An object dervied from SfpBase representing the specified sfp,
+            or None if the port is not an SFP port
         """
         sfp = None
 
+        self._validate_port_lists()
         try:
             sfp = self._sfp_list[index]
         except IndexError:
@@ -744,6 +810,48 @@ def get_sfp(self, index):
 
         return sfp
 
+    def get_num_cpos(self):
+        """
+        Retrieves the number of CPO ports available on this chassis
+
+        Returns:
+            An integer, the number of CPO ports available on this chassis
+        """
+        self._validate_port_lists()
+        return sum(1 for cpo in self._cpo_list if cpo is not None)
+
+    def get_all_cpos(self):
+        """
+        Retrieves all CPO ports available on this chassis
+
+        Returns:
+            A list of objects derived from CpoBase representing all CPO ports
+            available on this chassis
+        """
+        self._validate_port_lists()
+        return [cpo for cpo in self._cpo_list if cpo is not None]
+
+    def get_cpo(self, index):
+        """
+        Retrieves the CPO port corresponding to physical port <index>, if
+        that port is driven by CPO devices.
+
+        Args:
+            index: An integer (>=0), the physical port index (same indexing as
+                   get_sfp()).
+
+        Returns:
+            An object derived from CpoBase representing the specified CPO port,
+            or None if the port is not a CPO port.
+        """
+        cpo = None
+        self._validate_port_lists()
+        try:
+            cpo = self._cpo_list[index]
+        except IndexError:
+            sys.stderr.write("CPO index {} out of range (0-{})\n".format(
+                             index, len(self._cpo_list)-1))
+        return cpo
 
     def get_port_or_cage_type(self, index):
         """

sonic_platform_base/sonic_xcvr/cpo/cpo_base.py

@@ -3,6 +3,7 @@
 from enum import Enum
 from typing import Optional
 
+from sonic_platform_base import device_base
 from sonic_platform_base.sonic_xcvr.xcvr_eeprom import XcvrEeprom
 from sonic_platform_base.sonic_xcvr.eeprom_rw import EepromReadWriteMixin
 
@@ -36,7 +37,7 @@ def create_api(self):
         raise NotImplementedError
 
 
-class CpoDeviceBase(EepromReadWriteMixin):
+class CpoDeviceBase(device_base.DeviceBase, EepromReadWriteMixin):
     def __init__(self, hardware_id: CpoHardwareInfo, bank: int = 0):
         self.bank = bank
         self.hardware_id = hardware_id
@@ -55,16 +56,25 @@ def get_api(self):
             self.refresh_api()
         return self._api
 
+    def remove_api(self):
+        self._api = None
+
 
-class CpoBase:
+class CpoBase(device_base.DeviceBase):
     def __init__(self, hardware_id: CpoHardwareInfo, oe: "OeBase", elsfp: "ElsfpBase"):
         self.hardware_id = hardware_id
         self.oe = oe
         self.elsfp = elsfp
 
-# TODO: Implement CPO-specific methods
-#     def do_fiber_check(self, lane):
-#         self.oe.get_api().do_fiber_check(lane)
-#
-#     def tx_disable(self, lane):
-#         self.elsfp.get_api().tx_disable(lane)
+    def refresh_xcvr_api(self):
+        self.oe.refresh_api()
+        self.elsfp.refresh_api()
+
+    def get_xcvr_api(self):
+        # We always default to the OE API for CPO. If the ELSFP API is required,
+        # then that can be accessed via self.elsfp.get_api() directly.
+        return self.oe.get_api()
+
+    def remove_xcvr_api(self):
+        self.oe.remove_api()
+        self.elsfp.remove_api()

tests/chassis_base_test.py

@@ -1,7 +1,17 @@
+import pytest
+from unittest import mock
+
 from sonic_platform_base.chassis_base import ChassisBase
 
 class TestChassisBase:
 
+    @pytest.fixture(autouse=True)
+    def _mock_cpo_data(self):
+        with mock.patch("sonic_py_common.device_info.get_cpo_data",
+                        return_value=None) as mock_get:
+            self.mock_get_cpo_data = mock_get
+            yield
+
     def test_reboot_cause(self):
         chassis = ChassisBase()
         assert(chassis.REBOOT_CAUSE_POWER_LOSS == "Power Loss")
@@ -161,3 +171,85 @@ def test_pdbs_multiple_and_negative_index(self, capsys):
         assert chassis.get_pdb(-4) is None
         err_neg = capsys.readouterr().err
         assert "PDB index -4 out of range (0-2)" in err_neg
+
+    def test_no_cpo_data(self):
+        chassis = ChassisBase()
+        self.mock_get_cpo_data.assert_called_once()
+        assert chassis.get_num_cpos() == 0
+
+    def test_cpo_data_base_class_not_implemented(self):
+        self.mock_get_cpo_data.return_value = {"devices": {}, "interfaces": {}}
+        with pytest.raises(NotImplementedError):
+            ChassisBase()
+
+    def test_construct_cpo_list_for_topology(self):
+        cpo_data = {
+            "devices": {
+                "OE1": {"device_type": "optical_engine"},
+                "ELS1": {"device_type": "external_laser_source"},
+            },
+            "interfaces": {
+                "Ethernet0": {
+                    "associated_devices": [
+                        {"device_id": "OE1", "bank": 0},
+                        {"device_id": "ELS1", "bank": 0},
+                    ]
+                }
+            },
+        }
+        self.mock_get_cpo_data.return_value = cpo_data
+
+        class CpoChassis(ChassisBase):
+            def construct_cpo_devices(self, cpo_data):
+                for interface in cpo_data["interfaces"]:
+                    self._cpo_list.append(interface)
+
+        chassis = CpoChassis()
+        assert chassis.get_num_cpos() == 1
+        assert chassis.get_cpo(0) == "Ethernet0"
+
+    def test_sfp_counts_only_valid_objects(self):
+        chassis = ChassisBase()
+        sfp2 = mock.MagicMock()
+        sfp3 = mock.MagicMock()
+        chassis._sfp_list = [None, None, sfp2, sfp3]
+
+        assert chassis.get_num_sfps() == 2
+        assert chassis.get_all_sfps() == [sfp2, sfp3]
+        assert chassis.get_sfp(0) is None
+        assert chassis.get_sfp(2) is sfp2
+
+    def test_port_lists_valid(self):
+        chassis = ChassisBase()
+        chassis._sfp_list = [None, None, mock.MagicMock(), mock.MagicMock()]
+        chassis._cpo_list = [mock.MagicMock(), mock.MagicMock(), None, None]
+
+        assert chassis.get_num_sfps() == 2
+        assert chassis.get_num_cpos() == 2
+
+    def test_port_lists_length_mismatch(self):
+        chassis = ChassisBase()
+        chassis._sfp_list = [None, None, mock.MagicMock(), mock.MagicMock()]
+        chassis._cpo_list = [mock.MagicMock(), mock.MagicMock()]
+
+        with pytest.raises(RuntimeError, match="must be the same length"):
+            chassis.get_num_cpos()
+
+        with pytest.raises(RuntimeError, match="must be the same length"):
+            chassis.get_num_sfps()
+
+    def test_port_lists_double_claimed_port(self):
+        chassis = ChassisBase()
+        chassis._sfp_list = [None, mock.MagicMock(), mock.MagicMock()]
+        chassis._cpo_list = [mock.MagicMock(), mock.MagicMock(), None]
+
+        with pytest.raises(RuntimeError,
+                           match="Physical port 1 has both an SFP and a CPO object"):
+            chassis.get_num_sfps()
+
+    def test_port_lists_single_technology_not_checked(self):
+        chassis = ChassisBase()
+        chassis._sfp_list = [None, mock.MagicMock(), mock.MagicMock()]
+
+        assert chassis.get_num_sfps() == 2
+        assert chassis.get_num_cpos() == 0

tests/sonic_xcvr/test_cpo_base.py

@@ -68,3 +68,15 @@ def test_init(self):
         assert cpo.hardware_id is hardware_id
         assert cpo.oe is oe
         assert cpo.elsfp is elsfp
+
+    def test_get_xcvr_api_returns_oe_api(self):
+        hardware_id = CpoHardwareInfo(oe_id=SOME_OE_ID, elsfp_id=SOME_ELSFP_ID)
+        oe = OeBase(hardware_id)
+        elsfp = ElsfpBase(hardware_id)
+        oe_api = MagicMock()
+        oe.get_api = MagicMock(return_value=oe_api)
+
+        cpo = CpoBase(hardware_id, oe, elsfp)
+
+        assert cpo.get_xcvr_api() is oe_api
+        oe.get_api.assert_called_with()