mirror of
https://github.com/ets-labs/python-dependency-injector.git
synced 2025-05-24 23:39:11 +03:00
Refactor provider tests
This commit is contained in:
parent
3f8af1cd56
commit
dea1033371
|
@ -1,9 +1,6 @@
|
||||||
"""Dependency injector coroutine providers unit tests."""
|
"""Dependency injector coroutine providers unit tests."""
|
||||||
|
|
||||||
import asyncio
|
import asyncio
|
||||||
import contextlib
|
|
||||||
import sys
|
|
||||||
import gc
|
|
||||||
|
|
||||||
import unittest2 as unittest
|
import unittest2 as unittest
|
||||||
|
|
||||||
|
@ -12,6 +9,19 @@ from dependency_injector import (
|
||||||
errors,
|
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):
|
async def _example(arg1, arg2, arg3, arg4):
|
||||||
future = asyncio.Future()
|
future = asyncio.Future()
|
||||||
|
@ -25,52 +35,6 @@ def run(main):
|
||||||
return loop.run_until_complete(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):
|
class CoroutineTests(AsyncTestCase):
|
||||||
|
|
||||||
def test_init_with_coroutine(self):
|
def test_init_with_coroutine(self):
|
||||||
|
|
Loading…
Reference in New Issue
Block a user