Increase unit test coverage

Signed-off-by: Zhaohui Sun <zhaohuisun@microsoft.com>

scripts/caclmgrd

@@ -137,6 +137,7 @@ class ControlPlaneAclManager(daemon_base.DaemonBase):
         self.update_thread[DEFAULT_NAMESPACE] = None
         self.lock[DEFAULT_NAMESPACE] = threading.Lock()
         self.num_changes[DEFAULT_NAMESPACE] = 0
+        self.exception_queue = Queue()
 
         if device_info.is_multi_npu():
             swsscommon.SonicDBConfig.load_sonic_global_db_config()
@@ -855,7 +856,7 @@ class ControlPlaneAclManager(daemon_base.DaemonBase):
             self.run_commands(dualtor_iptables_cmds)
 
 
-    def check_and_update_control_plane_acls(self, namespace, num_changes, exception_queue):
+    def check_and_update_control_plane_acls(self, namespace, num_changes):
         """
         This function is intended to be spawned in a separate thread.
         Its purpose is to prevent unnecessary iptables updates if we receive
@@ -897,7 +898,7 @@ class ControlPlaneAclManager(daemon_base.DaemonBase):
             self.log_error("Exception occured at {} thread due to {}".format(threading.current_thread().getName(), repr(e)))
             exc_type, exc_value, exc_traceback = sys.exc_info()
             full_traceback = traceback.format_exception(exc_type, exc_value, exc_traceback)
-            exception_queue.put((namespace, repr(e), full_traceback))  # Add the exception to the queue
+            self.exception_queue.put((namespace, repr(e), full_traceback))  # Add the exception to the queue
         finally:
             new_config_db_connector.close("CONFIG_DB")
 
@@ -996,7 +997,6 @@ class ControlPlaneAclManager(daemon_base.DaemonBase):
         # set up state_db connector
         state_db_connector = swsscommon.DBConnector("STATE_DB", 0)
         config_db_connector = swsscommon.DBConnector("CONFIG_DB", 0)
-        exception_queue = Queue()
 
         if self.DualToR:
             self.log_info("Dual ToR mode")
@@ -1045,10 +1045,9 @@ class ControlPlaneAclManager(daemon_base.DaemonBase):
         while True:
             # Periodically check for exceptions from child threads
             try:
-                namespace, error, exc_info = exception_queue.get_nowait()  # Non-blocking
+                namespace, error, exc_info = self.exception_queue.get_nowait()  # Non-blocking
                 self.log_error("Exception in namespace '{}': {}".format(namespace, error))
                 self.log_error("Full traceback from child thread:")
-                import pdb; pdb.set_trace()
                 for tb_line in exc_info:
                     for tb_line_split in tb_line.splitlines():
                         self.log_error(tb_line_split)
@@ -1149,7 +1148,7 @@ class ControlPlaneAclManager(daemon_base.DaemonBase):
                     if not self.update_thread[namespace]:
                         self.log_info("Spawning ACL update thread for namepsace '{}' ...".format(namespace))
                         self.update_thread[namespace] = threading.Thread(target=self.check_and_update_control_plane_acls,
-                                                                        args=(namespace, self.num_changes[namespace], exception_queue))
+                                                                        args=(namespace, self.num_changes[namespace]))
                         self.update_thread[namespace].start()
 
 # ============================= Functions =============================

tests/caclmgrd/caclmgrd_bfd_test.py

@@ -53,8 +53,7 @@ def test_caclmgrd_bfd(self, test_name, test_data, fs):
                 caclmgrd_daemon.bfdAllowed = True
                 mocked_subprocess.Popen.reset_mock()
                 caclmgrd_daemon.num_changes[''] = 1
-                exception_queue = Queue()
-                caclmgrd_daemon.check_and_update_control_plane_acls('', 1, exception_queue)
+                caclmgrd_daemon.check_and_update_control_plane_acls('', 1)
 
                 #Ensure BFD rules are installed before ip2me rules to avoid traffic loss during update of control plane acl rules
                 bfd_ipv4_idx = 0

tests/caclmgrd/caclmgrd_scale_test.py

@@ -48,6 +48,5 @@ def test_caclmgrd_scale(self, test_name, test_data, fs):
 
                 caclmgrd_daemon = self.caclmgrd.ControlPlaneAclManager("caclmgrd")
                 caclmgrd_daemon.num_changes[''] = 150
-                exception_queue = Queue()
-                caclmgrd_daemon.check_and_update_control_plane_acls('', 150, exception_queue)
+                caclmgrd_daemon.check_and_update_control_plane_acls('', 150)
                 mocked_subprocess.Popen.assert_has_calls(test_data["expected_subprocess_calls"], any_order=True)

tests/caclmgrd/caclmgrd_test.py

@@ -59,11 +59,10 @@ def test_update_control_plane_acls_exception(self, mock_update):
         manager.lock = {"": threading.Lock()}
         manager.num_changes = {"": 0}
         manager.update_thread = {"": None}
-        exception_queue = Queue()
         manager.num_changes[""] = 1
-        manager.check_and_update_control_plane_acls("", 0, exception_queue)
-        self.assertFalse(exception_queue.empty())
-        exc_info = exception_queue.get()
+        manager.check_and_update_control_plane_acls("", 0)
+        self.assertFalse(manager.exception_queue.empty())
+        exc_info = manager.exception_queue.get()
         self.assertEqual(exc_info[0], "")
         self.assertIn("Test exception", exc_info[1])
 
@@ -86,6 +85,7 @@ def test_run(
         mock_swsscommon.SonicDBConfig.getDbId.side_effect = lambda db_name: (
             6 if db_name == "STATE_DB" else 1
         )
+        mock_kill.return_value = None
         mock_state_db_connector = MagicMock()
         mock_config_db_connector = MagicMock()
         mock_swsscommon.DBConnector.side_effect = [mock_state_db_connector, mock_config_db_connector, mock_state_db_connector] 
@@ -169,3 +169,110 @@ def test_run(
         manager.update_dhcp_acl_for_mark_change.assert_called()
         manager.update_dhcp_acl.assert_called()
         manager.setup_dhcp_chain.assert_called()
+
+
+    @patch("caclmgrd.swsscommon")
+    @patch("os.geteuid", return_value=0)
+    @patch("os.kill")
+    @patch("signal.SIGKILL", return_value=9)
+    @patch("sys.exit")
+    def test_run_exception(
+        self,
+        mock_exit,
+        mock_sigkill,
+        mock_kill,
+        mock_geteuid,
+        mock_swsscommon
+    ):
+        mock_kill.return_value = None
+        mock_sigkill.return_value = 9
+        mock_geteuid.return_value = 0
+        mock_exit.return_value = None
+
+        mock_swsscommon.SonicDBConfig.getDbId.side_effect = lambda db_name: (
+            6 if db_name == "STATE_DB" else 1
+        )
+        mock_state_db_connector = MagicMock()
+        mock_config_db_connector = MagicMock()
+        mock_swsscommon.DBConnector.side_effect = [mock_state_db_connector, mock_config_db_connector, mock_state_db_connector]
+        mock_swsscommon.Select.OBJECT = 1
+        mock_swsscommon.Select.return_value.select.return_value = (
+            mock_swsscommon.Select.OBJECT,
+            MagicMock(),
+        )
+        mock_swsscommon.Select.return_value.removeSelectable.return_value = MagicMock()
+
+        mock_swsscommon.SubscriberStateTable.return_value.select.return_value = (
+            mock_swsscommon.Select.OBJECT,
+            MagicMock(),
+        )
+        mock_swsscommon.SubscriberStateTable.return_value.getTableNameSeparator.return_value = "|"
+        pop_values = [
+            ("key1", "SET", [("mark", "0x11"), ("field2", "value2")]),
+            (None, None, None),
+            ("key2", "DEL", []),
+            (None, None, None),
+            ("key3", "SET", [("mark", "0x11")]),
+            (None, None, None),
+            ("SSH_ONLY", "SET", (('policy_desc', 'SSH_ONLY'), ('services', 'SSH'), ('stage', 'ingress'), ('type', 'CTRLPLANE'))),
+            ('', None, None),
+            ("key5", "SET", [("mark", "0x11"), ("field4", "value4")]),
+            ('', None, None),
+            ("key6", "SET", [("mark", "0x11"), ("field5", "value5")]),
+            ('', None, None),
+            ("SSH_ONLY", "SET", (('policy_desc', 'SSH_ONLY'), ('services', 'SSH'), ('stage', 'ingress'), ('type', 'CTRLPLANE'))),
+            ('', None, None),
+            ("SSH_ONLY", "SET", (('policy_desc', 'SSH_ONLY'), ('services', 'SSH'), ('stage', 'ingress'), ('type', 'CTRLPLANE'))),
+            ('', None, None),
+        ]
+        mock_swsscommon.SubscriberStateTable.return_value.pop.side_effect = pop_values
+        mock_swsscommon.CastSelectableToRedisSelectObj.return_value.getDbConnector.return_value.getNamespace.return_value = ""
+
+        mock_swsscommon.CastSelectableToRedisSelectObj.return_value.getDbConnector.return_value.getDbId.side_effect = [6, 4, 4]
+
+        # Creating an instance of ControlPlaneAclManager
+        self.caclmgrd.ControlPlaneAclManager.get_namespace_mgmt_ip = MagicMock()
+        self.caclmgrd.ControlPlaneAclManager.get_namespace_mgmt_ipv6 = MagicMock()
+        manager = self.caclmgrd.ControlPlaneAclManager("caclmgrd")
+
+        # Setting necessary attributes
+        manager.log_info = MagicMock()
+        manager.log_error = MagicMock()
+        manager.DualToR = True
+        manager.iptables_cmd_ns_prefix = {"": []}
+        manager.lock = {"": threading.Lock()}
+        manager.num_changes = {"": 2}
+        manager.update_thread = {"": threading.Thread()}
+        manager.bfdAllowed = False
+        manager.VxlanAllowed = True
+        manager.VxlanSrcIP = ""
+        manager.MUX_CABLE_TABLE = "MUX_CABLE_TABLE"
+        manager.BFD_SESSION_TABLE = "BFD_SESSION_TABLE"
+        manager.VXLAN_TUNNEL_TABLE = "VXLAN_TUNNEL_TABLE"
+        manager.ACL_TABLE = "ACL_TABLE"
+        manager.ACL_RULE = "ACL_RULE"
+        manager.ACL_TABLE_TYPE_CTRLPLANE = "CTRLPLANE"
+
+        # Mocking methods
+        manager.removeSelectable = MagicMock()
+        manager.allow_bfd_protocol = MagicMock()
+        manager.update_control_plane_acls = MagicMock()
+        manager.allow_vxlan_port = MagicMock()
+        manager.block_vxlan_port = MagicMock()
+        manager.update_dhcp_acl_for_mark_change = MagicMock()
+        manager.update_dhcp_acl = MagicMock()
+        manager.setup_dhcp_chain = MagicMock()
+        manager.exception_queue = Queue()
+        namespace = ""
+        error = "Simulated exception"
+        exc_info = ["Traceback (most recent call last):", "  File \"mock.py\", line 1, in <module>", "    raise Exception('Simulated exception')"]
+        manager.exception_queue.put((namespace, error, exc_info))
+        try:
+            manager.run()
+        except StopIteration as e:
+            # This is expected to happen
+            pass
+
+        # Asserting the method calls
+        manager.update_control_plane_acls.assert_called()
+        mock_kill.assert_called()

tests/caclmgrd/caclmgrd_vxlan_test.py

@@ -59,7 +59,7 @@ def test_caclmgrd_vxlan(self, test_name, test_data, fs):
                 caclmgrd_daemon.allow_vxlan_port('', data)
                 mocked_subprocess.Popen.reset_mock()
                 caclmgrd_daemon.num_changes[''] = 1
-                exception_queue = Queue()
-                caclmgrd_daemon.check_and_update_control_plane_acls('', 1, exception_queue)
+                caclmgrd_daemon.exception_queue = Queue()
+                caclmgrd_daemon.check_and_update_control_plane_acls('', 1)
                 mocked_subprocess.Popen.assert_has_calls(test_data["expected_add_subprocess_calls"], any_order=True)