Coordinate threading tests with events instead of sleeps

tests/test_threading.py

@@ -1,27 +1,34 @@
 import itertools
 import time
-from threading import Barrier, Thread
+from threading import Barrier, Event, Thread
 
 from loguru import logger
 
 
 class NonSafeSink:
-    def __init__(self, sleep_time, stop_time=0):
+    def __init__(self, sleep_time, stop_time=0, stopping=None):
         self.sleep_time = sleep_time
         self.stop_time = stop_time
         self.written = ""
         self.stopped = False
+        # Set once the sink is midway through an operation, so that another thread can
+        # act while it is still in progress without relying on timing. The "stopping"
+        # event can be shared between sinks to wait for the first one of a group.
+        self.writing = Event()
+        self.stopping = Event() if stopping is None else stopping
 
     def write(self, message):
         if self.stopped:
             raise RuntimeError("Can't write on stopped sink")
 
         length = len(message)
         self.written += message[:length]
+        self.writing.set()
         time.sleep(self.sleep_time)
         self.written += message[length:]
 
     def stop(self):
+        self.stopping.set()
         time.sleep(self.stop_time)
         self.stopped = True
 
@@ -64,7 +71,7 @@ def thread_1():
 
     def thread_2():
         barrier.wait()
-        time.sleep(0.5)
+        sink_1.writing.wait()
         logger.add(sink_2, format="{message}", catch=False)
         logger.info("ccc{}ddd", next(counter))
 
@@ -95,7 +102,7 @@ def thread_1():
 
     def thread_2():
         barrier.wait()
-        time.sleep(0.5)
+        sink.writing.wait()
         logger.remove(i)
         logger.info("ccc{}ddd", next(counter))
 
@@ -144,8 +151,10 @@ def thread_2():
 def test_safe_slow_removing_all_while_logging(capsys):
     barrier = Barrier(2)
 
+    stopping = Event()
+
     for _ in range(10):
-        sink = NonSafeSink(0.1, 0.1)
+        sink = NonSafeSink(0.1, 0.1, stopping=stopping)
         logger.add(sink, format="{message}", catch=False)
 
     def thread_1():
@@ -154,7 +163,7 @@ def thread_1():
 
     def thread_2():
         barrier.wait()
-        time.sleep(0.5)
+        stopping.wait()
         logger.info("Some message")
 
     threads = [Thread(target=thread_1), Thread(target=thread_2)]
@@ -173,7 +182,8 @@ def thread_2():
 def test_safe_writing_after_removing(capsys):
     barrier = Barrier(2)
 
-    logger.add(NonSafeSink(1), format="{message}", catch=False)
+    sink_1 = NonSafeSink(1)
+    logger.add(sink_1, format="{message}", catch=False)
     i = logger.add(NonSafeSink(1), format="{message}", catch=False)
 
     def write():
@@ -182,7 +192,7 @@ def write():
 
     def remove():
         barrier.wait()
-        time.sleep(0.5)
+        sink_1.writing.wait()
         logger.remove(i)
 
     threads = [Thread(target=write), Thread(target=remove)]