idk

Author: nycrat

src/software/gameplay_tests/simulated_test_runner.py

@@ -18,6 +18,11 @@
 LAUNCH_DELAY_S = 0.1
 WORLD_BUFFER_TIMEOUT = 0.5
 TICK_DURATION_S = DEFAULT_SIMULATOR_TICK_RATE_SECONDS_PER_TICK
+# Upper bound on how long _pre_run_setup will retry waiting for the simulator to
+# acknowledge the world state before giving up. The simulator normally responds
+# within WORLD_BUFFER_TIMEOUT, so this only trips on a genuinely wedged sim,
+# preventing an infinite setup loop from hanging an externally driven test.
+SETUP_TIMEOUT_S = 30
 
 
 class SimulatedTestRunner(TbotsTestRunner):
@@ -77,6 +82,7 @@ def _pre_run_setup(self, setup: (lambda: None)):
             world_state_received_buffer,
         )
 
+        setup_start_time = time.time()
         while True:
             setup()
 
@@ -86,6 +92,11 @@ def _pre_run_setup(self, setup: (lambda: None)):
                 )
             except queue.Empty:
                 # Did not receive a response within timeout period
+                if time.time() - setup_start_time > SETUP_TIMEOUT_S:
+                    raise TimeoutError(
+                        "Simulator did not acknowledge the world state within "
+                        f"{SETUP_TIMEOUT_S}s during test setup"
+                    )
                 continue
             else:
                 # Received a response from the simulator

src/software/gameplay_tests/tbots_test_runner.py

@@ -10,6 +10,10 @@
 
 PAUSE_AFTER_FAIL_DELAY_S = 3
 PROCESS_BUFFER_DELAY_S = 0.01
+# Extra wall-clock grace, on top of a test's own timeout, before an externally
+# driven test (run from the test selector widget) is declared hung. Generous so
+# it only ever trips on a genuinely wedged simulator, never on a slow machine.
+EXTERNAL_TEST_GRACE_S = 30
 
 
 class TbotsTestRunner:
@@ -43,6 +47,7 @@ def __init__(
         self.yellow_full_system_proto_unix_io = yellow_full_system_proto_unix_io
         self.gamecontroller = gamecontroller
         self.is_yellow_friendly = is_yellow_friendly
+        self._test_thread = None
         self.world_buffer = ThreadSafeBuffer(buffer_size=20, protobuf_type=World)
         self.primitive_set_buffer = ThreadSafeBuffer(
             buffer_size=1, protobuf_type=PrimitiveSet
@@ -188,6 +193,11 @@ def run_test(
         """
         self._pre_run_setup(setup)
 
+        # Install our thread excepthook so a failure on the test thread is
+        # captured into last_exception, and restore the previous hook afterwards
+        # so we don't leave a stale, short-lived runner installed on the
+        # long-lived process (e.g. the --test_mode GUI).
+        previous_excepthook = threading.excepthook
         threading.excepthook = self._excepthook
 
         args = (
@@ -197,23 +207,38 @@ def run_test(
             gc_cmd_with_delay,
         )
 
-        if self.thunderscope_is_external:
-            self._test_thread = threading.Thread(
-                target=self._runner, daemon=True, args=args
-            )
-            self._test_thread.start()
-        elif self.thunderscope:
-            run_test_thread = threading.Thread(
-                target=self._runner, daemon=True, args=args
-            )
-            run_test_thread.start()
-            self.thunderscope.show()
-            run_test_thread.join()
-
-            if self.last_exception:
-                pytest.fail(str(self.last_exception))
-        else:
-            self._runner(*args)
+        try:
+            if self.thunderscope_is_external:
+                # Thunderscope is owned/shown by an external owner (the test
+                # selector widget). We are already off the GUI thread here, so
+                # block until the test finishes to give run_test the same
+                # synchronous semantics as the other modes (callers may validate
+                # after run_test, or call run_test more than once). Bound the
+                # wait so a wedged simulator can't hang the caller forever.
+                self._test_thread = threading.Thread(
+                    target=self._runner, daemon=True, args=args
+                )
+                self._test_thread.start()
+                self._test_thread.join(timeout=test_timeout_s + EXTERNAL_TEST_GRACE_S)
+                if self._test_thread.is_alive():
+                    raise TimeoutError(
+                        f"Test '{self.test_name}' did not finish within "
+                        f"{test_timeout_s + EXTERNAL_TEST_GRACE_S:.0f}s"
+                    )
+            elif self.thunderscope:
+                run_test_thread = threading.Thread(
+                    target=self._runner, daemon=True, args=args
+                )
+                run_test_thread.start()
+                self.thunderscope.show()
+                run_test_thread.join()
+
+                if self.last_exception:
+                    pytest.fail(str(self.last_exception))
+            else:
+                self._runner(*args)
+        finally:
+            threading.excepthook = previous_excepthook
 
     @abstractmethod
     def _runner(
@@ -237,19 +262,26 @@ def _pre_run_setup(self, setup: (lambda: None)):
         """
         setup()
 
+    def ran_a_test(self) -> bool:
+        """Whether run_test() has actually started a test thread on this runner.
+
+        :return: True if a test was launched (in external mode)
+        """
+        return self._test_thread is not None
+
     def is_test_running(self) -> bool:
         """Check if a test is currently running in external mode.
 
         :return: True if a test thread is alive
         """
-        return hasattr(self, "_test_thread") and self._test_thread.is_alive()
+        return self._test_thread is not None and self._test_thread.is_alive()
 
     def wait_for_test(self, timeout: float | None = None) -> None:
         """Wait for the external test thread to finish.
 
         :param timeout: Maximum seconds to wait, or None for no timeout
         """
-        if hasattr(self, "_test_thread") and self._test_thread.is_alive():
+        if self._test_thread is not None and self._test_thread.is_alive():
             self._test_thread.join(timeout=timeout)
 
     def _stopper(self, delay=PROCESS_BUFFER_DELAY_S):
@@ -270,7 +302,15 @@ def _excepthook(self, args):
 
         :param args: The args passed in from the hook
         """
-        self._stopper(delay=PAUSE_AFTER_FAIL_DELAY_S)
+        # The post-fail pause exists to let the user see the failure before the
+        # runner closes Thunderscope. When Thunderscope is owned externally we
+        # never close it, so there's nothing to wait for.
+        delay = (
+            PROCESS_BUFFER_DELAY_S
+            if self.thunderscope_is_external
+            else PAUSE_AFTER_FAIL_DELAY_S
+        )
+        self._stopper(delay=delay)
         self.last_exception = args.exc_value
         raise self.last_exception
 

src/software/thunderscope/common/BUILD

@@ -75,6 +75,7 @@ py_library(
     name = "test_bridge",
     srcs = ["test_bridge.py"],
     deps = [
+        ":test_discovery",
         "//software/thunderscope:constants",
     ],
 )

src/software/thunderscope/common/test_bridge.py

@@ -1,8 +1,6 @@
-import importlib
-import importlib.util
-import sys
 import threading
 
+from software.thunderscope.common.test_discovery import load_test_module
 from software.thunderscope.constants import ProtoUnixIOTypes
 
 
@@ -24,6 +22,7 @@ def __init__(self):
         self.gamecontroller = None
         self.thunderscope = None
         self._running = False
+        self._lock = threading.Lock()
         self._current_runner = None
         self._on_test_completed = None
 
@@ -53,8 +52,6 @@ def run_test(self, item):
 
         :param item: TestItem from test_discovery
         """
-        if self._running:
-            return
         if item.test_type != "simulated":
             if self._on_test_completed:
                 self._on_test_completed(
@@ -63,7 +60,12 @@ def run_test(self, item):
                     "Field tests not supported in --test_mode yet",
                 )
             return
-        self._running = True
+        # Atomically claim the "running" slot so two near-simultaneous starts
+        # can't both spawn a runner against the shared simulator.
+        with self._lock:
+            if self._running:
+                return
+            self._running = True
         thread = threading.Thread(
             target=self._execute_test, args=(item,), daemon=True
         )
@@ -75,8 +77,9 @@ def _execute_test(self, item):
         )
 
         node_id = item.node_id or item.function_name or item.test_name
+        runner = None
         try:
-            module = self._import_test_module(item)
+            module = load_test_module(item.src_file)
             test_func = getattr(module, item.function_name)
             runner = SimulatedTestRunner(
                 test_name=node_id,
@@ -101,10 +104,16 @@ def _execute_test(self, item):
             else:
                 test_func(gameplay_test_runner=runner)
 
-            # run_test() starts the test in a background thread (Thunderscope is
-            # owned externally), so block here until it actually finishes before
-            # reporting the result.
+            # In external mode run_test() blocks until the test finishes (we are
+            # off the GUI thread), so this is a no-op guard. Treat a test
+            # function that never started a test as a failure rather than a
+            # silent pass.
             runner.wait_for_test()
+            if not runner.ran_a_test():
+                raise RuntimeError(
+                    "Test function returned without running a test "
+                    "(run_test was never called)"
+                )
             if runner.last_exception is not None:
                 raise runner.last_exception
 
@@ -114,17 +123,7 @@ def _execute_test(self, item):
             if self._on_test_completed:
                 self._on_test_completed(node_id, False, str(e))
         finally:
-            if self._current_runner is not None:
-                self._current_runner.unregister_observers()
-            self._running = False
+            if runner is not None:
+                runner.unregister_observers()
             self._current_runner = None
-
-    def _import_test_module(self, item):
-        """Dynamically import the test module from its source file."""
-        spec = importlib.util.spec_from_file_location(item.test_name, item.src_file)
-        if spec is None:
-            raise ImportError(f"Could not load spec for {item.src_file}")
-        module = importlib.util.module_from_spec(spec)
-        sys.modules[item.test_name] = module
-        spec.loader.exec_module(module)
-        return module
+            self._running = False