Add test that verifies one thread having logging supressed while it handles a

duaneg · duaneg · commit bc4bd11705c6 · 2025-04-17T22:34:35.000+12:00
message does not block a different thread from handling a message on the same
logger.
diff --git a/Lib/test/test_logging.py b/Lib/test/test_logging.py
@@ -4200,6 +4200,64 @@ def emit(self, record):
         logger.setLevel(logging.DEBUG)
         logger.debug("boom")
 
+    @threading_helper.requires_working_threading()
+    def test_thread_supression_noninterference(self):
+        lock = threading.Lock()
+        logger = logging.getLogger("test_thread_supression_noninterference")
+
+        # Block on the first call, allow others through
+        #
+        # NOTE: We need to bypass the base class's lock, otherwise that will
+        #       block multiple calls to the same handler itself.
+        class BlockOnceHandler(TestHandler):
+            def __init__(self, barrier):
+                super().__init__(support.Matcher())
+                self.barrier = barrier
+
+            def createLock(self):
+                self.lock = None
+
+            def handle(self, record):
+                self.emit(record)
+
+            def emit(self, record):
+                if self.barrier:
+                    barrier = self.barrier
+                    self.barrier = None
+                    barrier.wait()
+                    with lock:
+                        pass
+                super().emit(record)
+                logger.info("blow up if not supressed")
+
+        barrier = threading.Barrier(2)
+        handler = BlockOnceHandler(barrier)
+        logger.addHandler(handler)
+        logger.setLevel(logging.DEBUG)
+
+        t1 = threading.Thread(target=logger.debug, args=("1",))
+        with lock:
+
+            # Ensure first thread is blocked in the handler, hence supressing logging...
+            t1.start()
+            barrier.wait()
+
+            # ...but the second thread should still be able to log...
+            t2 = threading.Thread(target=logger.debug, args=("2",))
+            t2.start()
+            t2.join(timeout=3)
+
+            self.assertEqual(len(handler.buffer), 1)
+            self.assertTrue(handler.matches(levelno=logging.DEBUG, message='2'))
+
+            # The first thread should still be blocked here
+            self.assertTrue(t1.is_alive())
+
+        # Now the lock has been released the first thread should complete
+        t1.join()
+        self.assertEqual(len(handler.buffer), 2)
+        self.assertTrue(handler.matches(levelno=logging.DEBUG, message='1'))
+
 class ManagerTest(BaseTest):
     def test_manager_loggerclass(self):
         logged = []
