diff --git a/tests/unit/providers/test_coroutines_py35.py b/tests/unit/providers/test_coroutines_py35.py index c9109f26..a3697044 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 @@ -12,6 +9,19 @@ from dependency_injector import ( errors, ) +# Runtime import to get asyncutils module +import os +_TOP_DIR = os.path.abspath( + os.path.sep.join(( + os.path.dirname(__file__), + '../', + )), +) +import sys +sys.path.append(_TOP_DIR) + +from asyncutils import AsyncTestCase + async def _example(arg1, arg2, arg3, arg4): future = asyncio.Future() @@ -25,52 +35,6 @@ def run(main): 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):