From 7094fdc356897c13fe68115ebefe4c31e2269aab Mon Sep 17 00:00:00 2001 From: Roman Mogylatov Date: Sun, 17 Oct 2021 21:48:09 -0400 Subject: [PATCH] Migrate coroutine providers --- tests/unit/providers/coroutines/__init__.py | 1 + tests/unit/providers/coroutines/common.py | 5 + .../test_abstract_coroutine_py35.py | 66 ++++ .../test_coroutine_delegate_py35.py | 17 + .../coroutines/test_coroutine_py35.py | 199 ++++++++++++ .../test_delegated_coroutine_py35.py | 26 ++ tests/unit/providers/test_coroutines_py35.py | 293 ------------------ 7 files changed, 314 insertions(+), 293 deletions(-) create mode 100644 tests/unit/providers/coroutines/__init__.py create mode 100644 tests/unit/providers/coroutines/common.py create mode 100644 tests/unit/providers/coroutines/test_abstract_coroutine_py35.py create mode 100644 tests/unit/providers/coroutines/test_coroutine_delegate_py35.py create mode 100644 tests/unit/providers/coroutines/test_coroutine_py35.py create mode 100644 tests/unit/providers/coroutines/test_delegated_coroutine_py35.py delete mode 100644 tests/unit/providers/test_coroutines_py35.py diff --git a/tests/unit/providers/coroutines/__init__.py b/tests/unit/providers/coroutines/__init__.py new file mode 100644 index 00000000..90a8072d --- /dev/null +++ b/tests/unit/providers/coroutines/__init__.py @@ -0,0 +1 @@ +"""Tests for coroutine providers.""" diff --git a/tests/unit/providers/coroutines/common.py b/tests/unit/providers/coroutines/common.py new file mode 100644 index 00000000..e092bc46 --- /dev/null +++ b/tests/unit/providers/coroutines/common.py @@ -0,0 +1,5 @@ +"""Common test artifacts.""" + + +async def example(arg1, arg2, arg3, arg4): + return arg1, arg2, arg3, arg4 diff --git a/tests/unit/providers/coroutines/test_abstract_coroutine_py35.py b/tests/unit/providers/coroutines/test_abstract_coroutine_py35.py new file mode 100644 index 00000000..4f098442 --- /dev/null +++ b/tests/unit/providers/coroutines/test_abstract_coroutine_py35.py @@ -0,0 +1,66 @@ +"""AbstractCoroutine provider tests.""" + +import asyncio + +from dependency_injector import providers, errors +from pytest import mark, raises + +from .common import example + + +def test_inheritance(): + assert isinstance(providers.AbstractCoroutine(example), providers.Coroutine) + + +@mark.asyncio +@mark.filterwarnings("ignore") +async def test_call_overridden_by_coroutine(): + @asyncio.coroutine + def abstract_example(): + raise RuntimeError("Should not be raised") + + provider = providers.AbstractCoroutine(abstract_example) + provider.override(providers.Coroutine(example)) + + result = await provider(1, 2, 3, 4) + assert result == (1, 2, 3, 4) + + +@mark.asyncio +@mark.filterwarnings("ignore") +async def test_call_overridden_by_delegated_coroutine(): + @asyncio.coroutine + def abstract_example(): + raise RuntimeError("Should not be raised") + + provider = providers.AbstractCoroutine(abstract_example) + provider.override(providers.DelegatedCoroutine(example)) + + result = await provider(1, 2, 3, 4) + assert result == (1, 2, 3, 4) + + +def test_call_not_overridden(): + provider = providers.AbstractCoroutine(example) + with raises(errors.Error): + provider(1, 2, 3, 4) + + +def test_override_by_not_coroutine(): + provider = providers.AbstractCoroutine(example) + with raises(errors.Error): + provider.override(providers.Factory(object)) + + +def test_provide_not_implemented(): + provider = providers.AbstractCoroutine(example) + with raises(NotImplementedError): + provider._provide((1, 2, 3, 4), dict()) + + +def test_repr(): + provider = providers.AbstractCoroutine(example) + assert repr(provider) == ( + "".format(repr(example), hex(id(provider))) + ) diff --git a/tests/unit/providers/coroutines/test_coroutine_delegate_py35.py b/tests/unit/providers/coroutines/test_coroutine_delegate_py35.py new file mode 100644 index 00000000..d21b30c7 --- /dev/null +++ b/tests/unit/providers/coroutines/test_coroutine_delegate_py35.py @@ -0,0 +1,17 @@ +"""CoroutineDelegate provider tests.""" + +from dependency_injector import providers, errors +from pytest import raises + +from .common import example + + +def test_is_delegate(): + provider = providers.Coroutine(example) + delegate = providers.CoroutineDelegate(provider) + assert isinstance(delegate, providers.Delegate) + + +def test_init_with_not_coroutine(): + with raises(errors.Error): + providers.CoroutineDelegate(providers.Object(object())) diff --git a/tests/unit/providers/coroutines/test_coroutine_py35.py b/tests/unit/providers/coroutines/test_coroutine_py35.py new file mode 100644 index 00000000..09a8a946 --- /dev/null +++ b/tests/unit/providers/coroutines/test_coroutine_py35.py @@ -0,0 +1,199 @@ +"""Coroutine provider tests.""" + +from dependency_injector import providers, errors +from pytest import mark, raises + +from .common import example + + +def test_init_with_coroutine(): + assert isinstance(providers.Coroutine(example), providers.Coroutine) + + +def test_init_with_not_coroutine(): + with raises(errors.Error): + providers.Coroutine(lambda: None) + + +@mark.asyncio +async def test_init_optional_provides(): + provider = providers.Coroutine() + provider.set_provides(example) + + result = await provider(1, 2, 3, 4) + + assert result == (1, 2, 3, 4) + assert provider.provides is example + + +def test_set_provides_returns_self(): + provider = providers.Coroutine() + assert provider.set_provides(example) is provider + + +@mark.asyncio +async def test_call_with_positional_args(): + provider = providers.Coroutine(example, 1, 2, 3, 4) + result = await provider() + assert result == (1, 2, 3, 4) + + +@mark.asyncio +async def test_call_with_keyword_args(): + provider = providers.Coroutine(example, arg1=1, arg2=2, arg3=3, arg4=4) + result = await provider() + assert result == (1, 2, 3, 4) + + +@mark.asyncio +async def test_call_with_positional_and_keyword_args(): + provider = providers.Coroutine(example, 1, 2, arg3=3, arg4=4) + result = await provider() + assert result == (1, 2, 3, 4) + + +@mark.asyncio +async def test_call_with_context_args(): + provider = providers.Coroutine(example, 1, 2) + result = await provider(3, 4) + assert result == (1, 2, 3, 4) + + +@mark.asyncio +async def test_call_with_context_kwargs(): + provider = providers.Coroutine(example, arg1=1) + result = await provider(arg2=2, arg3=3, arg4=4) + assert result == (1, 2, 3, 4) + + +@mark.asyncio +async def test_call_with_context_args_and_kwargs(): + provider = providers.Coroutine(example, 1) + result = await provider(2, arg3=3, arg4=4) + assert result == (1, 2, 3, 4) + + +@mark.asyncio +async def test_fluent_interface(): + provider = providers.Coroutine(example) \ + .add_args(1, 2) \ + .add_kwargs(arg3=3, arg4=4) + result = await provider() + assert result == (1, 2, 3, 4) + + +def test_set_args(): + provider = providers.Coroutine(example) \ + .add_args(1, 2) \ + .set_args(3, 4) + assert provider.args == (3, 4) + + +def test_set_kwargs(): + provider = providers.Coroutine(example) \ + .add_kwargs(init_arg3=3, init_arg4=4) \ + .set_kwargs(init_arg3=4, init_arg4=5) + assert provider.kwargs == dict(init_arg3=4, init_arg4=5) + + +def test_clear_args(): + provider = providers.Coroutine(example) \ + .add_args(1, 2) \ + .clear_args() + assert provider.args == tuple() + + +def test_clear_kwargs(): + provider = providers.Coroutine(example) \ + .add_kwargs(init_arg3=3, init_arg4=4) \ + .clear_kwargs() + assert provider.kwargs == dict() + + +def test_call_overridden(): + provider = providers.Coroutine(example) + + provider.override(providers.Object((4, 3, 2, 1))) + provider.override(providers.Object((1, 2, 3, 4))) + + assert provider() == (1, 2, 3, 4) + + +def test_deepcopy(): + provider = providers.Coroutine(example) + + provider_copy = providers.deepcopy(provider) + + assert provider is not provider_copy + assert provider.provides is provider_copy.provides + assert isinstance(provider, providers.Coroutine) + + +def test_deepcopy_from_memo(): + provider = providers.Coroutine(example) + provider_copy_memo = providers.Coroutine(example) + + provider_copy = providers.deepcopy(provider, memo={id(provider): provider_copy_memo}) + + assert provider_copy is provider_copy_memo + + +def test_deepcopy_args(): + provider = providers.Coroutine(example) + dependent_provider1 = providers.Callable(list) + dependent_provider2 = providers.Callable(dict) + + provider.add_args(dependent_provider1, dependent_provider2) + + provider_copy = providers.deepcopy(provider) + dependent_provider_copy1 = provider_copy.args[0] + dependent_provider_copy2 = provider_copy.args[1] + + assert dependent_provider1.provides is dependent_provider_copy1.provides + assert dependent_provider1 is not dependent_provider_copy1 + + assert dependent_provider2.provides is dependent_provider_copy2.provides + assert dependent_provider2 is not dependent_provider_copy2 + + +def test_deepcopy_kwargs(): + provider = providers.Coroutine(example) + dependent_provider1 = providers.Callable(list) + dependent_provider2 = providers.Callable(dict) + + provider.add_kwargs(a1=dependent_provider1, a2=dependent_provider2) + + provider_copy = providers.deepcopy(provider) + dependent_provider_copy1 = provider_copy.kwargs["a1"] + dependent_provider_copy2 = provider_copy.kwargs["a2"] + + assert dependent_provider1.provides is dependent_provider_copy1.provides + assert dependent_provider1 is not dependent_provider_copy1 + + assert dependent_provider2.provides is dependent_provider_copy2.provides + assert dependent_provider2 is not dependent_provider_copy2 + + +def test_deepcopy_overridden(): + provider = providers.Coroutine(example) + object_provider = providers.Object(object()) + + provider.override(object_provider) + + provider_copy = providers.deepcopy(provider) + object_provider_copy = provider_copy.overridden[0] + + assert provider is not provider_copy + assert provider.provides is provider_copy.provides + assert isinstance(provider, providers.Callable) + + assert object_provider is not object_provider_copy + assert isinstance(object_provider_copy, providers.Object) + + +def test_repr(): + provider = providers.Coroutine(example) + assert repr(provider) == ( + "".format(repr(example), hex(id(provider))) + ) diff --git a/tests/unit/providers/coroutines/test_delegated_coroutine_py35.py b/tests/unit/providers/coroutines/test_delegated_coroutine_py35.py new file mode 100644 index 00000000..96de0036 --- /dev/null +++ b/tests/unit/providers/coroutines/test_delegated_coroutine_py35.py @@ -0,0 +1,26 @@ +"""DelegatedCoroutine provider tests.""" + +from dependency_injector import providers + +from .common import example + + +def test_inheritance(): + assert isinstance(providers.DelegatedCoroutine(example), providers.Coroutine) + + +def test_is_provider(): + assert providers.is_provider(providers.DelegatedCoroutine(example)) is True + + +def test_is_delegated_provider(): + provider = providers.DelegatedCoroutine(example) + assert providers.is_delegated(provider) is True + + +def test_repr(): + provider = providers.DelegatedCoroutine(example) + assert repr(provider) == ( + "".format(repr(example), hex(id(provider))) + ) diff --git a/tests/unit/providers/test_coroutines_py35.py b/tests/unit/providers/test_coroutines_py35.py deleted file mode 100644 index 2ea88adc..00000000 --- a/tests/unit/providers/test_coroutines_py35.py +++ /dev/null @@ -1,293 +0,0 @@ -"""Dependency injector coroutine providers unit tests.""" - -import asyncio -import unittest -import warnings - -from dependency_injector import ( - providers, - errors, -) -from pytest import raises - -# 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() - future.set_result(None) - await future - return arg1, arg2, arg3, arg4 - - -def run(main): - loop = asyncio.get_event_loop() - return loop.run_until_complete(main) - - -class CoroutineTests(AsyncTestCase): - - def test_init_with_coroutine(self): - assert isinstance(providers.Coroutine(_example), providers.Coroutine) - - def test_init_with_not_coroutine(self): - with raises(errors.Error): - providers.Coroutine(lambda: None) - - def test_init_optional_provides(self): - provider = providers.Coroutine() - provider.set_provides(_example) - assert provider.provides is _example - assert run(provider(1, 2, 3, 4)) == (1, 2, 3, 4) - - def test_set_provides_returns_self(self): - provider = providers.Coroutine() - assert provider.set_provides(_example) is provider - - def test_call_with_positional_args(self): - provider = providers.Coroutine(_example, 1, 2, 3, 4) - assert self._run(provider()) == (1, 2, 3, 4) - - def test_call_with_keyword_args(self): - provider = providers.Coroutine(_example, arg1=1, arg2=2, arg3=3, arg4=4) - assert self._run(provider()) == (1, 2, 3, 4) - - def test_call_with_positional_and_keyword_args(self): - provider = providers.Coroutine(_example, 1, 2, arg3=3, arg4=4) - assert run(provider()) == (1, 2, 3, 4) - - def test_call_with_context_args(self): - provider = providers.Coroutine(_example, 1, 2) - assert self._run(provider(3, 4)) == (1, 2, 3, 4) - - def test_call_with_context_kwargs(self): - provider = providers.Coroutine(_example, arg1=1) - assert self._run(provider(arg2=2, arg3=3, arg4=4)) == (1, 2, 3, 4) - - def test_call_with_context_args_and_kwargs(self): - provider = providers.Coroutine(_example, 1) - assert self._run(provider(2, arg3=3, arg4=4)) == (1, 2, 3, 4) - - def test_fluent_interface(self): - provider = providers.Coroutine(_example) \ - .add_args(1, 2) \ - .add_kwargs(arg3=3, arg4=4) - assert self._run(provider()) == (1, 2, 3, 4) - - def test_set_args(self): - provider = providers.Coroutine(_example) \ - .add_args(1, 2) \ - .set_args(3, 4) - assert provider.args == (3, 4) - - def test_set_kwargs(self): - provider = providers.Coroutine(_example) \ - .add_kwargs(init_arg3=3, init_arg4=4) \ - .set_kwargs(init_arg3=4, init_arg4=5) - assert provider.kwargs == dict(init_arg3=4, init_arg4=5) - - def test_clear_args(self): - provider = providers.Coroutine(_example) \ - .add_args(1, 2) \ - .clear_args() - assert provider.args == tuple() - - def test_clear_kwargs(self): - provider = providers.Coroutine(_example) \ - .add_kwargs(init_arg3=3, init_arg4=4) \ - .clear_kwargs() - assert provider.kwargs == dict() - - def test_call_overridden(self): - provider = providers.Coroutine(_example) - - provider.override(providers.Object((4, 3, 2, 1))) - provider.override(providers.Object((1, 2, 3, 4))) - - assert provider() == (1, 2, 3, 4) - - def test_deepcopy(self): - provider = providers.Coroutine(_example) - - provider_copy = providers.deepcopy(provider) - - assert provider is not provider_copy - assert provider.provides is provider_copy.provides - assert isinstance(provider, providers.Coroutine) - - def test_deepcopy_from_memo(self): - provider = providers.Coroutine(_example) - provider_copy_memo = providers.Coroutine(_example) - - provider_copy = providers.deepcopy( - provider, memo={id(provider): provider_copy_memo}) - - assert provider_copy is provider_copy_memo - - def test_deepcopy_args(self): - provider = providers.Coroutine(_example) - dependent_provider1 = providers.Callable(list) - dependent_provider2 = providers.Callable(dict) - - provider.add_args(dependent_provider1, dependent_provider2) - - provider_copy = providers.deepcopy(provider) - dependent_provider_copy1 = provider_copy.args[0] - dependent_provider_copy2 = provider_copy.args[1] - - assert provider.args != provider_copy.args - - assert dependent_provider1.provides is dependent_provider_copy1.provides - assert dependent_provider1 is not dependent_provider_copy1 - - assert dependent_provider2.provides is dependent_provider_copy2.provides - assert dependent_provider2 is not dependent_provider_copy2 - - def test_deepcopy_kwargs(self): - provider = providers.Coroutine(_example) - dependent_provider1 = providers.Callable(list) - dependent_provider2 = providers.Callable(dict) - - provider.add_kwargs(a1=dependent_provider1, a2=dependent_provider2) - - provider_copy = providers.deepcopy(provider) - dependent_provider_copy1 = provider_copy.kwargs["a1"] - dependent_provider_copy2 = provider_copy.kwargs["a2"] - - assert provider.kwargs != provider_copy.kwargs - - assert dependent_provider1.provides is dependent_provider_copy1.provides - assert dependent_provider1 is not dependent_provider_copy1 - - assert dependent_provider2.provides is dependent_provider_copy2.provides - assert dependent_provider2 is not dependent_provider_copy2 - - def test_deepcopy_overridden(self): - provider = providers.Coroutine(_example) - object_provider = providers.Object(object()) - - provider.override(object_provider) - - provider_copy = providers.deepcopy(provider) - object_provider_copy = provider_copy.overridden[0] - - assert provider is not provider_copy - assert provider.provides is provider_copy.provides - assert isinstance(provider, providers.Callable) - - assert object_provider is not object_provider_copy - assert isinstance(object_provider_copy, providers.Object) - - def test_repr(self): - provider = providers.Coroutine(_example) - - assert repr(provider) == ( - "".format(repr(_example), hex(id(provider))) - ) - - -class DelegatedCoroutineTests(unittest.TestCase): - - def test_inheritance(self): - assert isinstance(providers.DelegatedCoroutine(_example), - providers.Coroutine) - - def test_is_provider(self): - assert providers.is_provider(providers.DelegatedCoroutine(_example)) is True - - def test_is_delegated_provider(self): - provider = providers.DelegatedCoroutine(_example) - assert providers.is_delegated(provider) is True - - def test_repr(self): - provider = providers.DelegatedCoroutine(_example) - - assert repr(provider) == ( - "".format(repr(_example), hex(id(provider))) - ) - - -class AbstractCoroutineTests(AsyncTestCase): - - def test_inheritance(self): - assert isinstance(providers.AbstractCoroutine(_example), - providers.Coroutine) - - def test_call_overridden_by_coroutine(self): - with warnings.catch_warnings(): - warnings.simplefilter("ignore") - - @asyncio.coroutine - def _abstract_example(): - raise RuntimeError("Should not be raised") - - provider = providers.AbstractCoroutine(_abstract_example) - provider.override(providers.Coroutine(_example)) - - assert self._run(provider(1, 2, 3, 4)) == (1, 2, 3, 4) - - def test_call_overridden_by_delegated_coroutine(self): - with warnings.catch_warnings(): - warnings.simplefilter("ignore") - - @asyncio.coroutine - def _abstract_example(): - raise RuntimeError("Should not be raised") - - provider = providers.AbstractCoroutine(_abstract_example) - provider.override(providers.DelegatedCoroutine(_example)) - - assert self._run(provider(1, 2, 3, 4)) == (1, 2, 3, 4) - - def test_call_not_overridden(self): - provider = providers.AbstractCoroutine(_example) - - with raises(errors.Error): - provider(1, 2, 3, 4) - - def test_override_by_not_coroutine(self): - provider = providers.AbstractCoroutine(_example) - - with raises(errors.Error): - provider.override(providers.Factory(object)) - - def test_provide_not_implemented(self): - provider = providers.AbstractCoroutine(_example) - - with raises(NotImplementedError): - provider._provide((1, 2, 3, 4), dict()) - - def test_repr(self): - provider = providers.AbstractCoroutine(_example) - - assert repr(provider) == ( - "".format(repr(_example), hex(id(provider))) - ) - - -class CoroutineDelegateTests(unittest.TestCase): - - def setUp(self): - self.delegated = providers.Coroutine(_example) - self.delegate = providers.CoroutineDelegate(self.delegated) - - def test_is_delegate(self): - assert isinstance(self.delegate, providers.Delegate) - - def test_init_with_not_callable(self): - with raises(errors.Error): - providers.CoroutineDelegate(providers.Object(object()))