Fix the defer namespace in golem.core.deferred

Prevent importing the reactor in tests

Author: mfranciszkiewicz

golem/core/deferred.py

@@ -1,30 +1,30 @@
 from queue import Queue, Empty
 from typing import Any
 
-from twisted.internet.defer import Deferred, inlineCallbacks, maybeDeferred
+from twisted.internet import defer
 from twisted.internet.task import deferLater
-from twisted.internet.threads import deferToThread
 from twisted.python.failure import Failure
 
 
 class DeferredSeq:
     def __init__(self, *fns) -> None:
         self._seq = list(fns)
 
-    def execute(self) -> Deferred:
+    def execute(self) -> defer.Deferred:
+        from twisted.internet.threads import deferToThread
         # The executed function cannot return a Deferred, hence the bool cast
         return deferToThread(lambda *_: bool(self._execute()))
 
-    @inlineCallbacks
+    @defer.inlineCallbacks
     def _execute(self) -> Any:
         arg = None
         for fn in self._seq:
-            arg = yield maybeDeferred(fn, arg)
+            arg = yield defer.maybeDeferred(fn, arg)
         return arg
 
 
 def chain_function(deferred, fn, *args, **kwargs):
-    result = Deferred()
+    result = defer.Deferred()
 
     def resolve(_):
         fn(*args, **kwargs).addCallbacks(result.callback,
@@ -36,7 +36,7 @@ def resolve(_):
 
 
 def sync_wait(deferred, timeout=10):
-    if not isinstance(deferred, Deferred):
+    if not isinstance(deferred, defer.Deferred):
         return deferred
 
     queue = Queue()
@@ -45,13 +45,13 @@ def sync_wait(deferred, timeout=10):
     try:
         result = queue.get(True, timeout)
     except Empty:
-        raise TimeoutError("Command timed out")
+        raise defer.TimeoutError("Command timed out")
 
     if isinstance(result, Failure):
         result.raiseException()
     return result
 
 
-def call_later(delay: int, callable, *args, **kwargs) -> None:
+def call_later(delay: int, fn, *args, **kwargs) -> None:
     from twisted.internet import reactor
-    deferLater(reactor, delay, callable, *args, **kwargs)
+    deferLater(reactor, delay, fn, *args, **kwargs)

tests/golem/core/test_deferred.py

@@ -7,6 +7,8 @@
 from golem.core.deferred import chain_function, DeferredSeq
 
 
+@mock.patch('twisted.internet.threads.deferToThread', lambda x: succeed(x()))
+@mock.patch('twisted.internet.reactor', mock.Mock(), create=True)
 class TestDeferredSeq(unittest.TestCase):
 
     def test_init_empty(self):
@@ -19,11 +21,9 @@ def test_init_with_functions(self):
         ]
         assert DeferredSeq(*fns)._seq == fns
 
-    @mock.patch('golem.core.deferred.deferToThread', lambda x: succeed(x()))
     def test_execute_empty(self):
         assert DeferredSeq().execute().result
 
-    @mock.patch('golem.core.deferred.deferToThread', lambda x: succeed(x()))
     def test_execute_functions(self):
         fn_1, fn_2 = mock.Mock(), mock.Mock()
         fns = [fn_1, fn_2]
@@ -43,7 +43,7 @@ def def2t(f, *args, **kwargs) -> Deferred:
             except Exception as exc:  # pylint: disable=broad-except
                 return fail(exc)
 
-        with mock.patch('golem.core.deferred.deferToThread', def2t):
+        with mock.patch('twisted.internet.threads.deferToThread', def2t):
             DeferredSeq(fn_1, fn_2, fn_3, fn_4).execute()
 
         assert fn_1.called

tests/golem/task/test_rpc.py

@@ -124,7 +124,7 @@ def test_create_task(self, *_):
         def execute(f, *args, **kwargs):
             return defer.succeed(f(*args, **kwargs))
 
-        with mock.patch('golem.core.deferred.deferToThread', execute):
+        with mock.patch('twisted.internet.threads.deferToThread', execute):
             result = self.provider.create_task(t.to_dict())
         rpc.enqueue_new_task.assert_called()
         self.assertEqual(result, ('task_id', None))
@@ -751,6 +751,7 @@ def setUp(self):
                 rpc.enqueue_new_task(self.client, self.task),
             )
 
+    @mock.patch('twisted.internet.reactor', mock.Mock())
     @mock.patch("golem.task.rpc.prepare_and_validate_task_dict")
     def test_create_task(self, mock_method, *_):
         t = dummytaskstate.DummyTaskDefinition()