add helper to block on threads finishing

carly/__init__.py

@@ -3,3 +3,4 @@
 from .clock import cancelDelayedCalls, advanceTime
 from .context import Context
 from .hook import hook, cleanup, decoder, register
+from .threads import waitForThreads

carly/threads.py

@@ -0,0 +1,18 @@
+from time import sleep
+
+from twisted.internet import reactor
+from twisted.internet.threads import deferToThread
+
+from .clock import withTimeout
+
+
+def pendingIsEmpty():
+    while True:
+        stats = reactor.threadpool._team.statistics()
+        if not (stats.backloggedWorkCount or stats.busyWorkerCount > 1):
+            break
+        sleep(0.001)
+
+
+def waitForThreads(timeout=None):
+    return withTimeout(deferToThread(pendingIsEmpty), timeout)

tests/test_threads.py

@@ -0,0 +1,46 @@
+from time import sleep
+
+from testfixtures import compare, ShouldRaise
+from twisted.internet.defer import inlineCallbacks, TimeoutError
+from twisted.internet.task import LoopingCall
+from twisted.internet.threads import deferToThread
+from twisted.trial.unittest import TestCase
+
+from carly import advanceTime, cancelDelayedCalls, waitForThreads
+
+
+def setState(state, duration):
+    sleep(duration)
+    state.fired += 1
+
+
+def work(state, duration):
+    deferToThread(setState, state, duration)
+
+
+class TestDeferToThread(TestCase):
+
+    fired = 0
+
+    @inlineCallbacks
+    def testWorkOne(self):
+        LoopingCall(work, self, 0.001).start(2)
+        advanceTime(seconds=0.2)
+        yield waitForThreads()
+        compare(self.fired, expected=1)
+        cancelDelayedCalls()
+
+    @inlineCallbacks
+    def testWorkTwo(self):
+        LoopingCall(work, self, 0.001).start(2)
+        advanceTime(seconds=0.2)
+        yield waitForThreads()
+        compare(self.fired, expected=1)
+        cancelDelayedCalls()
+
+    @inlineCallbacks
+    def testTimeout(self):
+        LoopingCall(work, self, duration=0.1).start(2)
+        with ShouldRaise(TimeoutError(0.02, 'Deferred')):
+            yield waitForThreads(timeout=0.02)
+        cancelDelayedCalls()