twisted#11707 fix trial -j / ctrl-c UnboundLocalError bug (twisted#…

…11897)

Fixes twisted#11707

src/twisted/newsfragments/11707.bugfix

@@ -0,0 +1,7 @@
+When interrupted with control-C, `trial -j` no longer obscures tracebacks for
+any errors caused by that interruption with an `UnboundLocalError` due to a bug
+in its own implementation.  Note that there are still several internal
+tracebacks that will be emitted upon exiting, because tearing down the test
+runner mid-suite is still not an entirely clean operation, but it should at
+least be possible to see errors reported from, for example, a test that is
+hanging more clearly.

src/twisted/trial/_dist/disttrial.py

@@ -13,7 +13,17 @@
 import sys
 from functools import partial
 from os.path import isabs
-from typing import Awaitable, Callable, Iterable, List, Sequence, TextIO, Union, cast
+from typing import (
+    Awaitable,
+    Callable,
+    Iterable,
+    List,
+    Optional,
+    Sequence,
+    TextIO,
+    Union,
+    cast,
+)
 from unittest import TestCase, TestSuite
 
 from attrs import define, field, frozen
@@ -441,15 +451,35 @@ async def runAndReport(n: int) -> DistReporter:
             await startedPool.join()
 
     def _run(self, test: Union[TestCase, TestSuite], untilFailure: bool) -> IReporter:
-        result: Union[Failure, DistReporter]
+        result: Union[Failure, DistReporter, None] = None
+        reactorStopping: bool = False
+        testsInProgress: Deferred[object]
 
-        def capture(r):
+        def capture(r: Union[Failure, DistReporter]) -> None:
             nonlocal result
             result = r
 
-        d = Deferred.fromCoroutine(self.runAsync(test, untilFailure))
-        d.addBoth(capture)
-        d.addBoth(lambda ignored: self._reactor.stop())
+        def maybeStopTests() -> Optional[Deferred[object]]:
+            nonlocal reactorStopping
+            reactorStopping = True
+            if result is None:
+                testsInProgress.cancel()
+                return testsInProgress
+            return None
+
+        def maybeStopReactor(result: object) -> object:
+            if not reactorStopping:
+                self._reactor.stop()
+            return result
+
+        self._reactor.addSystemEventTrigger("before", "shutdown", maybeStopTests)
+
+        testsInProgress = (
+            Deferred.fromCoroutine(self.runAsync(test, untilFailure))
+            .addBoth(capture)
+            .addBoth(maybeStopReactor)
+        )
+
         self._reactor.run()
 
         if isinstance(result, Failure):
@@ -458,7 +488,7 @@ def capture(r):
         # mypy can't see that raiseException raises an exception so we can
         # only get here if result is not a Failure, so tell mypy result is
         # certainly a DistReporter at this point.
-        assert isinstance(result, DistReporter)
+        assert isinstance(result, DistReporter), f"{result} is not DistReporter"
 
         # Unwrap the DistReporter to give the caller some regular IReporter
         # object.  DistReporter isn't type annotated correctly so fix it here.

src/twisted/trial/_dist/test/test_disttrial.py

@@ -31,7 +31,7 @@
 
 from twisted.internet import interfaces
 from twisted.internet.base import ReactorBase
-from twisted.internet.defer import Deferred, succeed
+from twisted.internet.defer import CancelledError, Deferred, succeed
 from twisted.internet.error import ProcessDone
 from twisted.internet.protocol import ProcessProtocol, Protocol
 from twisted.internet.test.modulehelpers import AlternateReactor
@@ -125,6 +125,11 @@ def stop(self):
         See L{IReactorCore.stop}.
         """
         MemoryReactorClock.stop(self)
+        # TODO: implementing this more comprehensively in MemoryReactor would
+        # be nice, this is rather hard-coded to disttrial's current
+        # implementation.
+        if "before" in self.triggers:
+            self.triggers["before"]["shutdown"][0][0]()
         self.stopCount += 1
 
     def run(self):
@@ -139,6 +144,9 @@ def run(self):
 
         for f, args, kwargs in self.whenRunningHooks:
             f(*args, **kwargs)
+        self.stop()
+        # do not count internal 'stop' against trial-initiated .stop() count
+        self.stopCount -= 1
 
 
 class CountingReactorTests(SynchronousTestCase):
@@ -461,6 +469,15 @@ def test_runUnexpectedError(self) -> None:
         assert_that(errors, has_length(1))
         assert_that(errors[0][1].type, equal_to(WorkerPoolBroken))
 
+    def test_runUnexpectedErrorCtrlC(self) -> None:
+        """
+        If the reactor is stopped by C-c (i.e. `run` returns before the test
+        case's Deferred has been fired) we should cancel the pending test run.
+        """
+        runner = self.getRunner(workerPoolFactory=LocalWorkerPool)
+        with self.assertRaises(CancelledError):
+            runner.run(self.suite)
+
     def test_runUnexpectedWorkerError(self) -> None:
         """
         If for some reason the worker process cannot run a test, the error is