mirror of
https://github.com/ets-labs/python-dependency-injector.git
synced 2025-05-23 14:59:08 +03:00
Stop running coroutine tests for Python 3.4
This commit is contained in:
parent
ff4803d6a5
commit
99e690a8a9
|
@ -19,25 +19,8 @@ def _example(arg1, arg2, arg3, arg4):
|
||||||
return arg1, arg2, arg3, arg4
|
return arg1, arg2, arg3, arg4
|
||||||
|
|
||||||
|
|
||||||
def _run(coro, debug=False):
|
def run(*args, **kwargs):
|
||||||
if asyncio._get_running_loop() is not None:
|
return asyncio.run(*args, **kwargs)
|
||||||
raise RuntimeError(
|
|
||||||
"asyncio.run() cannot be called from a running event loop")
|
|
||||||
|
|
||||||
if not coroutines.iscoroutine(coro):
|
|
||||||
raise ValueError("a coroutine was expected, got {!r}".format(coro))
|
|
||||||
|
|
||||||
loop = events.new_event_loop()
|
|
||||||
try:
|
|
||||||
events.set_event_loop(loop)
|
|
||||||
loop.set_debug(debug)
|
|
||||||
return loop.run_until_complete(coro)
|
|
||||||
finally:
|
|
||||||
try:
|
|
||||||
loop.run_until_complete(loop.shutdown_asyncgens())
|
|
||||||
finally:
|
|
||||||
events.set_event_loop(None)
|
|
||||||
loop.close()
|
|
||||||
|
|
||||||
|
|
||||||
class CoroutineTests(unittest.TestCase):
|
class CoroutineTests(unittest.TestCase):
|
||||||
|
@ -50,34 +33,34 @@ class CoroutineTests(unittest.TestCase):
|
||||||
|
|
||||||
def test_call_with_positional_args(self):
|
def test_call_with_positional_args(self):
|
||||||
provider = providers.Coroutine(_example, 1, 2, 3, 4)
|
provider = providers.Coroutine(_example, 1, 2, 3, 4)
|
||||||
self.assertTupleEqual(_run(provider()), (1, 2, 3, 4))
|
self.assertTupleEqual(run(provider()), (1, 2, 3, 4))
|
||||||
|
|
||||||
def test_call_with_keyword_args(self):
|
def test_call_with_keyword_args(self):
|
||||||
provider = providers.Coroutine(_example,
|
provider = providers.Coroutine(_example,
|
||||||
arg1=1, arg2=2, arg3=3, arg4=4)
|
arg1=1, arg2=2, arg3=3, arg4=4)
|
||||||
self.assertTupleEqual(_run(provider()), (1, 2, 3, 4))
|
self.assertTupleEqual(run(provider()), (1, 2, 3, 4))
|
||||||
|
|
||||||
def test_call_with_positional_and_keyword_args(self):
|
def test_call_with_positional_and_keyword_args(self):
|
||||||
provider = providers.Coroutine(_example,
|
provider = providers.Coroutine(_example,
|
||||||
1, 2,
|
1, 2,
|
||||||
arg3=3, arg4=4)
|
arg3=3, arg4=4)
|
||||||
self.assertTupleEqual(_run(provider()), (1, 2, 3, 4))
|
self.assertTupleEqual(run(provider()), (1, 2, 3, 4))
|
||||||
|
|
||||||
def test_call_with_context_args(self):
|
def test_call_with_context_args(self):
|
||||||
provider = providers.Coroutine(_example, 1, 2)
|
provider = providers.Coroutine(_example, 1, 2)
|
||||||
self.assertTupleEqual(_run(provider(3, 4)), (1, 2, 3, 4))
|
self.assertTupleEqual(run(provider(3, 4)), (1, 2, 3, 4))
|
||||||
|
|
||||||
def test_call_with_context_kwargs(self):
|
def test_call_with_context_kwargs(self):
|
||||||
provider = providers.Coroutine(_example, arg1=1)
|
provider = providers.Coroutine(_example, arg1=1)
|
||||||
self.assertTupleEqual(
|
self.assertTupleEqual(
|
||||||
_run(provider(arg2=2, arg3=3, arg4=4)),
|
run(provider(arg2=2, arg3=3, arg4=4)),
|
||||||
(1, 2, 3, 4),
|
(1, 2, 3, 4),
|
||||||
)
|
)
|
||||||
|
|
||||||
def test_call_with_context_args_and_kwargs(self):
|
def test_call_with_context_args_and_kwargs(self):
|
||||||
provider = providers.Coroutine(_example, 1)
|
provider = providers.Coroutine(_example, 1)
|
||||||
self.assertTupleEqual(
|
self.assertTupleEqual(
|
||||||
_run(provider(2, arg3=3, arg4=4)),
|
run(provider(2, arg3=3, arg4=4)),
|
||||||
(1, 2, 3, 4),
|
(1, 2, 3, 4),
|
||||||
)
|
)
|
||||||
|
|
||||||
|
@ -86,7 +69,7 @@ class CoroutineTests(unittest.TestCase):
|
||||||
.add_args(1, 2) \
|
.add_args(1, 2) \
|
||||||
.add_kwargs(arg3=3, arg4=4)
|
.add_kwargs(arg3=3, arg4=4)
|
||||||
|
|
||||||
self.assertTupleEqual(_run(provider()), (1, 2, 3, 4))
|
self.assertTupleEqual(run(provider()), (1, 2, 3, 4))
|
||||||
|
|
||||||
def test_set_args(self):
|
def test_set_args(self):
|
||||||
provider = providers.Coroutine(_example) \
|
provider = providers.Coroutine(_example) \
|
||||||
|
@ -244,7 +227,7 @@ class AbstractCoroutineTests(unittest.TestCase):
|
||||||
provider = providers.AbstractCoroutine(_abstract_example)
|
provider = providers.AbstractCoroutine(_abstract_example)
|
||||||
provider.override(providers.Coroutine(_example))
|
provider.override(providers.Coroutine(_example))
|
||||||
|
|
||||||
self.assertTrue(_run(provider(1, 2, 3, 4)), (1, 2, 3, 4))
|
self.assertTrue(run(provider(1, 2, 3, 4)), (1, 2, 3, 4))
|
||||||
|
|
||||||
def test_call_overridden_by_delegated_coroutine(self):
|
def test_call_overridden_by_delegated_coroutine(self):
|
||||||
@asyncio.coroutine
|
@asyncio.coroutine
|
||||||
|
@ -254,7 +237,7 @@ class AbstractCoroutineTests(unittest.TestCase):
|
||||||
provider = providers.AbstractCoroutine(_abstract_example)
|
provider = providers.AbstractCoroutine(_abstract_example)
|
||||||
provider.override(providers.DelegatedCoroutine(_example))
|
provider.override(providers.DelegatedCoroutine(_example))
|
||||||
|
|
||||||
self.assertTrue(_run(provider(1, 2, 3, 4)), (1, 2, 3, 4))
|
self.assertTrue(run(provider(1, 2, 3, 4)), (1, 2, 3, 4))
|
||||||
|
|
||||||
def test_call_not_overridden(self):
|
def test_call_not_overridden(self):
|
||||||
provider = providers.AbstractCoroutine(_example)
|
provider = providers.AbstractCoroutine(_example)
|
Loading…
Reference in New Issue
Block a user