fix(tests): make threading tests deterministic

Replace time-based loops and sleep-waits with iteration counts and
thread.join(), eliminating flaky failures on slow CI runners.

Author: fgmacedo

tests/test_threading.py

@@ -34,9 +34,8 @@ def test_regression_443():
     """
     Test for https://github.com/fgmacedo/python-statemachine/issues/443
     """
-    time_collecting = 0.2
-    time_to_send = 0.125
-    time_sampling_current_state = 0.05
+    total_iterations = 4
+    send_at_iteration = 3  # 0-indexed: send before the 4th sample
 
     class TrafficLightMachine(StateChart):
         "A traffic light machine"
@@ -52,25 +51,20 @@ def __init__(self):
             self.statuses_history = []
             self.fsm = TrafficLightMachine()
             # set up thread
-            t = threading.Thread(target=self.recv_cmds)
-            t.start()
+            self.thread = threading.Thread(target=self.recv_cmds)
+            self.thread.start()
 
         def recv_cmds(self):
-            """Pretend we receive a command triggering a state change after Xs."""
-            waiting_time = 0
-            sent = False
-            while waiting_time < time_collecting:
-                if waiting_time >= time_to_send and not sent:
+            """Pretend we receive a command triggering a state change."""
+            for i in range(total_iterations):
+                if i == send_at_iteration:
                     self.fsm.cycle()
-                    sent = True
-
-                waiting_time += time_sampling_current_state
                 self.statuses_history.append(self.fsm.current_state_value)
-                time.sleep(time_sampling_current_state)
 
     c1 = Controller()
     c2 = Controller()
-    time.sleep(time_collecting + 0.01)
+    c1.thread.join()
+    c2.thread.join()
     assert c1.statuses_history == ["green", "green", "green", "yellow"]
     assert c2.statuses_history == ["green", "green", "green", "yellow"]
 
@@ -79,9 +73,8 @@ def test_regression_443_with_modifications():
     """
     Test for https://github.com/fgmacedo/python-statemachine/issues/443
     """
-    time_collecting = 0.2
-    time_to_send = 0.125
-    time_sampling_current_state = 0.05
+    total_iterations = 4
+    send_at_iteration = 3  # 0-indexed: send before the 4th sample
 
     class TrafficLightMachine(StateChart):
         "A traffic light machine"
@@ -98,42 +91,36 @@ def __init__(self, name):
             super().__init__()
 
         def beat(self):
-            waiting_time = 0
-            sent = False
-            while waiting_time < time_collecting:
-                if waiting_time >= time_to_send and not sent:
+            for i in range(total_iterations):
+                if i == send_at_iteration:
                     self.cycle()
-                    sent = True
-
                 self.statuses_history.append(f"{self.name}.{self.current_state_value}")
 
-                time.sleep(time_sampling_current_state)
-                waiting_time += time_sampling_current_state
-
     class Controller:
         def __init__(self, name):
             self.fsm = TrafficLightMachine(name)
             # set up thread
-            t = threading.Thread(target=self.fsm.beat)
-            t.start()
+            self.thread = threading.Thread(target=self.fsm.beat)
+            self.thread.start()
 
     c1 = Controller("c1")
     c2 = Controller("c2")
     c3 = Controller("c3")
-    time.sleep(time_collecting + 0.01)
+    c1.thread.join()
+    c2.thread.join()
+    c3.thread.join()
 
     assert c1.fsm.statuses_history == ["c1.green", "c1.green", "c1.green", "c1.yellow"]
     assert c2.fsm.statuses_history == ["c2.green", "c2.green", "c2.green", "c2.yellow"]
     assert c3.fsm.statuses_history == ["c3.green", "c3.green", "c3.green", "c3.yellow"]
 
 
-async def test_regression_443_with_modifications_for_async_engine():  # noqa: C901
+async def test_regression_443_with_modifications_for_async_engine():
     """
     Test for https://github.com/fgmacedo/python-statemachine/issues/443
     """
-    time_collecting = 0.2
-    time_to_send = 0.125
-    time_sampling_current_state = 0.05
+    total_iterations = 4
+    send_at_iteration = 3  # 0-indexed: send before the 4th sample
 
     class TrafficLightMachine(StateChart):
         "A traffic light machine"
@@ -153,35 +140,30 @@ def __init__(self, name):
             super().__init__()
 
         def beat(self):
-            waiting_time = 0
-            sent = False
-            while waiting_time < time_collecting:
-                if waiting_time >= time_to_send and not sent:
+            for i in range(total_iterations):
+                if i == send_at_iteration:
                     self.cycle()
-                    sent = True
-
                 self.statuses_history.append(f"{self.name}.{self.current_state_value}")
 
-                time.sleep(time_sampling_current_state)
-                waiting_time += time_sampling_current_state
-
     class Controller:
         def __init__(self, name):
             self.fsm = TrafficLightMachine(name)
 
         async def start(self):
             # set up thread
             await self.fsm.activate_initial_state()
-            t = threading.Thread(target=self.fsm.beat)
-            t.start()
+            self.thread = threading.Thread(target=self.fsm.beat)
+            self.thread.start()
 
     c1 = Controller("c1")
     c2 = Controller("c2")
     c3 = Controller("c3")
     await c1.start()
     await c2.start()
     await c3.start()
-    time.sleep(time_collecting + 0.01)
+    c1.thread.join()
+    c2.thread.join()
+    c3.thread.join()
 
     assert c1.fsm.statuses_history == ["c1.green", "c1.green", "c1.green", "c1.yellow"]
     assert c2.fsm.statuses_history == ["c2.green", "c2.green", "c2.green", "c2.yellow"]