Refactor async tests

tests/unit/providers/test_coroutines_py35.py

@@ -1,9 +1,6 @@
 """Dependency injector coroutine providers unit tests."""
 
 import asyncio
-import contextlib
-import sys
-import gc
 
 import unittest2 as unittest
 
@@ -11,6 +8,7 @@ from dependency_injector import (
     providers,
     errors,
 )
+from .utils import AsyncTestCase, run
 
 
 async def _example(arg1, arg2, arg3, arg4):
@@ -20,57 +18,6 @@ async def _example(arg1, arg2, arg3, arg4):
     return arg1, arg2, arg3, arg4
 
 
-def run(main):
-    loop = asyncio.get_event_loop()
-    return loop.run_until_complete(main)
-
-
-def setup_test_loop(
-        loop_factory=asyncio.new_event_loop
-) -> asyncio.AbstractEventLoop:
-    loop = loop_factory()
-    try:
-        module = loop.__class__.__module__
-        skip_watcher = 'uvloop' in module
-    except AttributeError:  # pragma: no cover
-        # Just in case
-        skip_watcher = True
-    asyncio.set_event_loop(loop)
-    if sys.platform != "win32" and not skip_watcher:
-        policy = asyncio.get_event_loop_policy()
-        watcher = asyncio.SafeChildWatcher()  # type: ignore
-        watcher.attach_loop(loop)
-        with contextlib.suppress(NotImplementedError):
-            policy.set_child_watcher(watcher)
-    return loop
-
-
-def teardown_test_loop(loop: asyncio.AbstractEventLoop,
-                       fast: bool=False) -> None:
-    closed = loop.is_closed()
-    if not closed:
-        loop.call_soon(loop.stop)
-        loop.run_forever()
-        loop.close()
-
-    if not fast:
-        gc.collect()
-
-    asyncio.set_event_loop(None)
-
-
-class AsyncTestCase(unittest.TestCase):
-
-    def setUp(self):
-        self.loop = setup_test_loop()
-
-    def tearDown(self):
-        teardown_test_loop(self.loop)
-
-    def _run(self, f):
-        return self.loop.run_until_complete(f)
-
-
 class CoroutineTests(AsyncTestCase):
 
     def test_init_with_coroutine(self):

tests/unit/providers/utils.py

@@ -0,0 +1,58 @@
+"""Test utils."""
+
+import asyncio
+import contextlib
+import sys
+import gc
+import unittest
+
+
+def run(main):
+    loop = asyncio.get_event_loop()
+    return loop.run_until_complete(main)
+
+
+def setup_test_loop(
+        loop_factory=asyncio.new_event_loop
+) -> asyncio.AbstractEventLoop:
+    loop = loop_factory()
+    try:
+        module = loop.__class__.__module__
+        skip_watcher = 'uvloop' in module
+    except AttributeError:  # pragma: no cover
+        # Just in case
+        skip_watcher = True
+    asyncio.set_event_loop(loop)
+    if sys.platform != 'win32' and not skip_watcher:
+        policy = asyncio.get_event_loop_policy()
+        watcher = asyncio.SafeChildWatcher()  # type: ignore
+        watcher.attach_loop(loop)
+        with contextlib.suppress(NotImplementedError):
+            policy.set_child_watcher(watcher)
+    return loop
+
+
+def teardown_test_loop(loop: asyncio.AbstractEventLoop,
+                       fast: bool=False) -> None:
+    closed = loop.is_closed()
+    if not closed:
+        loop.call_soon(loop.stop)
+        loop.run_forever()
+        loop.close()
+
+    if not fast:
+        gc.collect()
+
+    asyncio.set_event_loop(None)
+
+
+class AsyncTestCase(unittest.TestCase):
+
+    def setUp(self):
+        self.loop = setup_test_loop()
+
+    def tearDown(self):
+        teardown_test_loop(self.loop)
+
+    def _run(self, f):
+        return self.loop.run_until_complete(f)