ets-labs/python-dependency-injector
1
1
Fork 0
mirror of https://github.com/ets-labs/python-dependency-injector.git synced 2025-07-31 10:19:48 +03:00
Code Issues Packages Projects Releases Wiki Activity

Refactor coroutine provider tests

Browse Source
This commit is contained in:
Roman Mogylatov 2018-10-16 17:56:15 +03:00
parent 06526d9457
commit 891e356197
1 changed files with 193 additions and 202 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

395
tests/unit/providers/test_coroutines_py3.py
View File

@ -1,9 +1,6 @@
"""Dependency injector coroutine providers unit tests.""" """Dependency injector coroutine providers unit tests."""
try: import asyncio
import asyncio
except ImportError:
asyncio = None
import unittest2 as unittest import unittest2 as unittest
@ -13,279 +10,273 @@ from dependency_injector import (
) )
if asyncio: @asyncio.coroutine
@asyncio.coroutine def _example(arg1, arg2, arg3, arg4):
def _example(arg1, arg2, arg3, arg4): future = asyncio.Future()
future = asyncio.Future() future.set_result(None)
future.set_result(None) yield from future
yield from future return arg1, arg2, arg3, arg4
return arg1, arg2, arg3, arg4
def _run(coro):
loop = asyncio.get_event_loop() def _run(coro):
return loop.run_until_complete(coro) loop = asyncio.get_event_loop()
else: return loop.run_until_complete(coro)
_example = None
class CoroutineTests(unittest.TestCase): class CoroutineTests(unittest.TestCase):
if asyncio: def test_init_with_coroutine(self):
def test_init_with_coroutine(self): self.assertTrue(providers.Coroutine(_example))
self.assertTrue(providers.Coroutine(_example))
def test_init_with_not_coroutine(self): def test_init_with_not_coroutine(self):
self.assertRaises(errors.Error, providers.Coroutine, lambda: None) self.assertRaises(errors.Error, providers.Coroutine, lambda: None)
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),
) )
def test_fluent_interface(self): def test_fluent_interface(self):
provider = providers.Coroutine(_example) \ provider = providers.Coroutine(_example) \
.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) \
.add_args(1, 2) \ .add_args(1, 2) \
.set_args(3, 4) .set_args(3, 4)
self.assertEqual(provider.args, tuple([3, 4])) self.assertEqual(provider.args, tuple([3, 4]))
def test_set_kwargs(self): def test_set_kwargs(self):
provider = providers.Coroutine(_example) \ provider = providers.Coroutine(_example) \
.add_kwargs(init_arg3=3, init_arg4=4) \ .add_kwargs(init_arg3=3, init_arg4=4) \
.set_kwargs(init_arg3=4, init_arg4=5) .set_kwargs(init_arg3=4, init_arg4=5)
self.assertEqual(provider.kwargs, dict(init_arg3=4, init_arg4=5)) self.assertEqual(provider.kwargs, dict(init_arg3=4, init_arg4=5))
def test_clear_args(self): def test_clear_args(self):
provider = providers.Coroutine(_example) \ provider = providers.Coroutine(_example) \
.add_args(1, 2) \ .add_args(1, 2) \
.clear_args() .clear_args()
self.assertEqual(provider.args, tuple()) self.assertEqual(provider.args, tuple())
def test_clear_kwargs(self): def test_clear_kwargs(self):
provider = providers.Coroutine(_example) \ provider = providers.Coroutine(_example) \
.add_kwargs(init_arg3=3, init_arg4=4) \ .add_kwargs(init_arg3=3, init_arg4=4) \
.clear_kwargs() .clear_kwargs()
self.assertEqual(provider.kwargs, dict()) self.assertEqual(provider.kwargs, dict())
def test_call_overridden(self): def test_call_overridden(self):
provider = providers.Coroutine(_example) provider = providers.Coroutine(_example)
provider.override(providers.Object((4, 3, 2, 1))) provider.override(providers.Object((4, 3, 2, 1)))
provider.override(providers.Object((1, 2, 3, 4))) provider.override(providers.Object((1, 2, 3, 4)))
self.assertTupleEqual(provider(), (1, 2, 3, 4)) self.assertTupleEqual(provider(), (1, 2, 3, 4))
def test_deepcopy(self): def test_deepcopy(self):
provider = providers.Coroutine(_example) provider = providers.Coroutine(_example)
provider_copy = providers.deepcopy(provider) provider_copy = providers.deepcopy(provider)
self.assertIsNot(provider, provider_copy) self.assertIsNot(provider, provider_copy)
self.assertIs(provider.provides, provider_copy.provides) self.assertIs(provider.provides, provider_copy.provides)
self.assertIsInstance(provider, providers.Coroutine) self.assertIsInstance(provider, providers.Coroutine)
def test_deepcopy_from_memo(self): def test_deepcopy_from_memo(self):
provider = providers.Coroutine(_example) provider = providers.Coroutine(_example)
provider_copy_memo = providers.Coroutine(_example) provider_copy_memo = providers.Coroutine(_example)
provider_copy = providers.deepcopy( provider_copy = providers.deepcopy(
provider, memo={id(provider): provider_copy_memo}) provider, memo={id(provider): provider_copy_memo})
self.assertIs(provider_copy, provider_copy_memo) self.assertIs(provider_copy, provider_copy_memo)
def test_deepcopy_args(self): def test_deepcopy_args(self):
provider = providers.Coroutine(_example) provider = providers.Coroutine(_example)
dependent_provider1 = providers.Callable(list) dependent_provider1 = providers.Callable(list)
dependent_provider2 = providers.Callable(dict) dependent_provider2 = providers.Callable(dict)
provider.add_args(dependent_provider1, dependent_provider2) provider.add_args(dependent_provider1, dependent_provider2)
provider_copy = providers.deepcopy(provider) provider_copy = providers.deepcopy(provider)
dependent_provider_copy1 = provider_copy.args[0] dependent_provider_copy1 = provider_copy.args[0]
dependent_provider_copy2 = provider_copy.args[1] dependent_provider_copy2 = provider_copy.args[1]
self.assertNotEqual(provider.args, provider_copy.args) self.assertNotEqual(provider.args, provider_copy.args)
self.assertIs(dependent_provider1.provides, self.assertIs(dependent_provider1.provides,
dependent_provider_copy1.provides) dependent_provider_copy1.provides)
self.assertIsNot(dependent_provider1, dependent_provider_copy1) self.assertIsNot(dependent_provider1, dependent_provider_copy1)
self.assertIs(dependent_provider2.provides, self.assertIs(dependent_provider2.provides,
dependent_provider_copy2.provides) dependent_provider_copy2.provides)
self.assertIsNot(dependent_provider2, dependent_provider_copy2) self.assertIsNot(dependent_provider2, dependent_provider_copy2)
def test_deepcopy_kwargs(self): def test_deepcopy_kwargs(self):
provider = providers.Coroutine(_example) provider = providers.Coroutine(_example)
dependent_provider1 = providers.Callable(list) dependent_provider1 = providers.Callable(list)
dependent_provider2 = providers.Callable(dict) dependent_provider2 = providers.Callable(dict)
provider.add_kwargs(a1=dependent_provider1, a2=dependent_provider2) provider.add_kwargs(a1=dependent_provider1, a2=dependent_provider2)
provider_copy = providers.deepcopy(provider) provider_copy = providers.deepcopy(provider)
dependent_provider_copy1 = provider_copy.kwargs['a1'] dependent_provider_copy1 = provider_copy.kwargs['a1']
dependent_provider_copy2 = provider_copy.kwargs['a2'] dependent_provider_copy2 = provider_copy.kwargs['a2']
self.assertNotEqual(provider.kwargs, provider_copy.kwargs) self.assertNotEqual(provider.kwargs, provider_copy.kwargs)
self.assertIs(dependent_provider1.provides, self.assertIs(dependent_provider1.provides,
dependent_provider_copy1.provides) dependent_provider_copy1.provides)
self.assertIsNot(dependent_provider1, dependent_provider_copy1) self.assertIsNot(dependent_provider1, dependent_provider_copy1)
self.assertIs(dependent_provider2.provides, self.assertIs(dependent_provider2.provides,
dependent_provider_copy2.provides) dependent_provider_copy2.provides)
self.assertIsNot(dependent_provider2, dependent_provider_copy2) self.assertIsNot(dependent_provider2, dependent_provider_copy2)
def test_deepcopy_overridden(self): def test_deepcopy_overridden(self):
provider = providers.Coroutine(_example) provider = providers.Coroutine(_example)
object_provider = providers.Object(object()) object_provider = providers.Object(object())
provider.override(object_provider) provider.override(object_provider)
provider_copy = providers.deepcopy(provider) provider_copy = providers.deepcopy(provider)
object_provider_copy = provider_copy.overridden[0] object_provider_copy = provider_copy.overridden[0]
self.assertIsNot(provider, provider_copy) self.assertIsNot(provider, provider_copy)
self.assertIs(provider.provides, provider_copy.provides) self.assertIs(provider.provides, provider_copy.provides)
self.assertIsInstance(provider, providers.Callable) self.assertIsInstance(provider, providers.Callable)
self.assertIsNot(object_provider, object_provider_copy) self.assertIsNot(object_provider, object_provider_copy)
self.assertIsInstance(object_provider_copy, providers.Object) self.assertIsInstance(object_provider_copy, providers.Object)
def test_repr(self): def test_repr(self):
provider = providers.Coroutine(_example) provider = providers.Coroutine(_example)
self.assertEqual(repr(provider), self.assertEqual(repr(provider),
'<dependency_injector.providers.' '<dependency_injector.providers.'
'Coroutine({0}) at {1}>'.format( 'Coroutine({0}) at {1}>'.format(
repr(_example), repr(_example),
hex(id(provider)))) hex(id(provider))))
class DelegatedCoroutineTests(unittest.TestCase): class DelegatedCoroutineTests(unittest.TestCase):
if asyncio: def test_inheritance(self):
def test_inheritance(self): self.assertIsInstance(providers.DelegatedCoroutine(_example),
self.assertIsInstance(providers.DelegatedCoroutine(_example), providers.Coroutine)
providers.Coroutine)
def test_is_provider(self): def test_is_provider(self):
self.assertTrue( self.assertTrue(
providers.is_provider(providers.DelegatedCoroutine(_example))) providers.is_provider(providers.DelegatedCoroutine(_example)))
def test_is_delegated_provider(self): def test_is_delegated_provider(self):
provider = providers.DelegatedCoroutine(_example) provider = providers.DelegatedCoroutine(_example)
self.assertTrue(providers.is_delegated(provider)) self.assertTrue(providers.is_delegated(provider))
def test_repr(self): def test_repr(self):
provider = providers.DelegatedCoroutine(_example) provider = providers.DelegatedCoroutine(_example)
self.assertEqual(repr(provider), self.assertEqual(repr(provider),
'<dependency_injector.providers.' '<dependency_injector.providers.'
'DelegatedCoroutine({0}) at {1}>'.format( 'DelegatedCoroutine({0}) at {1}>'.format(
repr(_example), repr(_example),
hex(id(provider)))) hex(id(provider))))
class AbstractCoroutineTests(unittest.TestCase): class AbstractCoroutineTests(unittest.TestCase):
if asyncio: def test_inheritance(self):
def test_inheritance(self): self.assertIsInstance(providers.AbstractCoroutine(_example),
self.assertIsInstance(providers.AbstractCoroutine(_example), providers.Coroutine)
providers.Coroutine)
def test_call_overridden_by_coroutine(self): def test_call_overridden_by_coroutine(self):
@asyncio.coroutine @asyncio.coroutine
def _abstract_example(): def _abstract_example():
raise RuntimeError('Should not be raised') raise RuntimeError('Should not be raised')
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
def _abstract_example(): def _abstract_example():
raise RuntimeError('Should not be raised') raise RuntimeError('Should not be raised')
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)
with self.assertRaises(errors.Error): with self.assertRaises(errors.Error):
provider(1, 2, 3, 4) provider(1, 2, 3, 4)
def test_override_by_not_coroutine(self): def test_override_by_not_coroutine(self):
provider = providers.AbstractCoroutine(_example) provider = providers.AbstractCoroutine(_example)
with self.assertRaises(errors.Error): with self.assertRaises(errors.Error):
provider.override(providers.Factory(object)) provider.override(providers.Factory(object))
def test_provide_not_implemented(self): def test_provide_not_implemented(self):
provider = providers.AbstractCoroutine(_example) provider = providers.AbstractCoroutine(_example)
with self.assertRaises(NotImplementedError): with self.assertRaises(NotImplementedError):
provider._provide((1, 2, 3, 4), dict()) provider._provide((1, 2, 3, 4), dict())
def test_repr(self): def test_repr(self):
provider = providers.AbstractCoroutine(_example) provider = providers.AbstractCoroutine(_example)
self.assertEqual(repr(provider), self.assertEqual(repr(provider),
'<dependency_injector.providers.' '<dependency_injector.providers.'
'AbstractCoroutine({0}) at {1}>'.format( 'AbstractCoroutine({0}) at {1}>'.format(
repr(_example), repr(_example),
hex(id(provider)))) hex(id(provider))))
class CoroutineDelegateTests(unittest.TestCase): class CoroutineDelegateTests(unittest.TestCase):
if asyncio: def setUp(self):
def setUp(self): self.delegated = providers.Coroutine(_example)
self.delegated = providers.Coroutine(_example) self.delegate = providers.CoroutineDelegate(self.delegated)
self.delegate = providers.CoroutineDelegate(self.delegated)
def test_is_delegate(self): def test_is_delegate(self):
self.assertIsInstance(self.delegate, providers.Delegate) self.assertIsInstance(self.delegate, providers.Delegate)
def test_init_with_not_callable(self): def test_init_with_not_callable(self):
self.assertRaises(errors.Error, self.assertRaises(errors.Error,
providers.CoroutineDelegate, providers.CoroutineDelegate,
providers.Object(object())) providers.Object(object()))