From 98f4c3d5740c70fde64349c2fd8ea81872853ce9 Mon Sep 17 00:00:00 2001
From: Roman Mogylatov <rmogilatov@gmail.com>
Date: Wed, 18 Nov 2020 14:13:46 -0500
Subject: [PATCH] Refactor async tests

---
 tests/unit/providers/test_coroutines_py35.py | 55 +------------------
 tests/unit/providers/utils.py                | 58 ++++++++++++++++++++
 2 files changed, 59 insertions(+), 54 deletions(-)
 create mode 100644 tests/unit/providers/utils.py

diff --git a/tests/unit/providers/test_coroutines_py35.py b/tests/unit/providers/test_coroutines_py35.py
index c9109f26..b50a7675 100644
--- a/tests/unit/providers/test_coroutines_py35.py
+++ b/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):
diff --git a/tests/unit/providers/utils.py b/tests/unit/providers/utils.py
new file mode 100644
index 00000000..5b0a3ca5
--- /dev/null
+++ b/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)