add tests for post_port_active_apsel_to_db()

sonic-net · Jun 27, 2023 · b64604c · b64604c
1 parent 62845ab
commit b64604c
Showing 1 changed file with 61 additions and 0 deletions.
diff --git a/sonic-xcvrd/tests/test_xcvrd.py b/sonic-xcvrd/tests/test_xcvrd.py
@@ -764,6 +764,67 @@ def get_host_lane_assignment_option_side_effect(app):
         appl = task.get_cmis_application_desired(mock_xcvr_api, host_lane_count, speed)
         assert task.get_cmis_host_lanes_mask(mock_xcvr_api, appl, host_lane_count, subport) == expected
 
+    def test_CmisManagerTask_post_port_active_apsel_to_db(self):
+        mock_xcvr_api = MagicMock()
+        mock_xcvr_api.get_active_apsel_hostlane = MagicMock(side_effect=[
+            NotImplementedError,
+            {
+             'ActiveAppSelLane1': 1,
+             'ActiveAppSelLane2': 1,
+             'ActiveAppSelLane3': 1,
+             'ActiveAppSelLane4': 1,
+             'ActiveAppSelLane5': 1,
+             'ActiveAppSelLane6': 1,
+             'ActiveAppSelLane7': 1,
+             'ActiveAppSelLane8': 1
+            },
+            {
+             'ActiveAppSelLane1': 2,
+             'ActiveAppSelLane2': 2,
+             'ActiveAppSelLane3': 2,
+             'ActiveAppSelLane4': 2,
+             'ActiveAppSelLane5': 2,
+             'ActiveAppSelLane6': 2,
+             'ActiveAppSelLane7': 2,
+             'ActiveAppSelLane8': 2
+            }
+        ])
+        int_tbl = Table("STATE_DB", TRANSCEIVER_INFO_TABLE)
+
+        port_mapping = PortMapping()
+        stop_event = threading.Event()
+        task = CmisManagerTask(DEFAULT_NAMESPACE, port_mapping, stop_event)
+        task.xcvr_table_helper.get_intf_tbl = MagicMock(return_value=int_tbl)
+
+        # case: NotImplementedError
+        lport = "Ethernet0"
+        host_lanes_mask = 0xc
+        ret = task.post_port_active_apsel_to_db(mock_xcvr_api, lport, host_lanes_mask)
+        assert int_tbl.getKeys() == []
+
+        # case: partial lanes update
+        lport = "Ethernet0"
+        host_lanes_mask = 0xc
+        ret = task.post_port_active_apsel_to_db(mock_xcvr_api, lport, host_lanes_mask)
+        assert int_tbl.getKeys() == ["Ethernet0"]
+        assert dict(int_tbl.mock_dict["Ethernet0"]) == {'active_apsel_hostlane3': '1',
+                                                        'active_apsel_hostlane4': '1'}
+        # case: full lanes update
+        lport = "Ethernet8"
+        host_lanes_mask = 0xff
+        task.post_port_active_apsel_to_db(mock_xcvr_api, lport, host_lanes_mask)
+        assert int_tbl.getKeys() == ["Ethernet0", "Ethernet8"]
+        assert dict(int_tbl.mock_dict["Ethernet0"]) == {'active_apsel_hostlane3': '1',
+                                                        'active_apsel_hostlane4': '1'}
+        assert dict(int_tbl.mock_dict["Ethernet8"]) == {'active_apsel_hostlane1': '2',
+                                                        'active_apsel_hostlane2': '2',
+                                                        'active_apsel_hostlane3': '2',
+                                                        'active_apsel_hostlane4': '2',
+                                                        'active_apsel_hostlane5': '2',
+                                                        'active_apsel_hostlane6': '2',
+                                                        'active_apsel_hostlane7': '2',
+                                                        'active_apsel_hostlane8': '2'}
+
     @patch('xcvrd.xcvrd.platform_chassis')
     @patch('xcvrd.xcvrd_utilities.port_mapping.subscribe_port_update_event', MagicMock(return_value=(None, None)))
     @patch('xcvrd.xcvrd_utilities.port_mapping.handle_port_update_event', MagicMock())