Commit dd2a1b32 authored by Xavier Barbosa's avatar Xavier Barbosa

auto injector

parent 785a1b44
Pipeline #1041 failed with stage
in 75 minutes and 59 seconds
......@@ -123,7 +123,7 @@ Singleton mode can be disabled per service::
Current services are automatically exposed inside functions::
def func():
return current_services()
return current_injector()
assert func() is None
assert (await services.apply(func)) is services
......
from .bases import Injector, annotate, current_services
from .bases import Injector, annotate, attr, current_injector
from ._version import get_versions
__all__ = ["Injector", "annotate"]
__version__ = get_versions()["version"]
del get_versions
from ._version import get_versions
__version__ = get_versions()["version"]
del get_versions
__all__ = ["__version__", "Injector", "annotate", "attr", "current_injector"]
This diff is collapsed.
......@@ -15,3 +15,7 @@ max-complexity = 10
max-line-length=99
verbose=1
[pycodestyle]
max_line_length=99
import pytest
from knighted import Injector, annotate, current_services
from knighted import Injector, annotate, current_injector
import asyncio
from contextlib import contextmanager
from time import time_ns, sleep
......@@ -30,6 +30,8 @@ def timed():
@pytest.mark.asyncio
async def test_instance_factory(services):
print("caca")
@services.factory("foo")
def foo_factory():
return "I am foo"
......@@ -132,27 +134,6 @@ async def test_load_in_parallel(services):
assert timer.duration == pytest.approx(1000000000, rel=1e-2)
@pytest.mark.asyncio
async def test_partial(services):
@services.factory("foo")
def foo_factory():
return "I am foo"
@services.factory("bar")
def bar_factory():
return "I am bar"
@annotate("foo", "bar")
def fun(foo, bar):
return {"foo": foo, "bar": bar}
assert len(services.services) == 0
part = services.partial(fun)
assert len(services.services) == 0
assert (await part()) == {"foo": "I am foo", "bar": "I am bar"}
assert len(services.services) == 2
@pytest.mark.asyncio
async def test_class_factory():
class MyInjector(Injector):
......@@ -322,9 +303,10 @@ async def test_not_singleton(services):
assert result1 != result2
@pytest.mark.asyncio
async def test_service_context(services):
def func():
return current_services()
return current_injector()
assert func() is None
assert (await services.apply(func)) is services
import pytest
from knighted import Injector, annotate, current_injector, attr
@pytest.fixture
def services():
class MyInjector(Injector):
pass
return MyInjector()
@pytest.mark.asyncio
async def test_sync_async(services):
@services.factory("foo")
def foo_factory():
return "I am foo"
@annotate("foo")
async def fun(foo):
return {"foo": foo}
assert await services.apply(fun) == {"foo": "I am foo"}
@pytest.mark.asyncio
async def test_sync_sync(services):
@services.factory("foo")
def foo_factory():
return "I am foo"
@annotate("foo")
def fun(foo):
return {"foo": foo}
assert await services.apply(fun) == {"foo": "I am foo"}
@pytest.mark.asyncio
async def test_async_async(services):
@services.factory("foo")
async def foo_factory():
return "I am foo"
@annotate("foo")
async def fun(foo):
return {"foo": foo}
assert await services.apply(fun) == {"foo": "I am foo"}
@pytest.mark.asyncio
async def test_auto_sync_async(services):
@services.factory("foo")
def foo_factory():
return "I am foo"
@annotate("foo")
async def fun(foo):
return {"foo": foo}
result = await services.apply(fun)
assert result == {"foo": "I am foo"}
@pytest.mark.asyncio
async def test_auto_async_async(services):
@services.factory("foo")
async def foo_factory():
return "I am foo"
@services.factory("bar")
async def bar_factory():
return "I am bar"
@annotate("bar")
async def fun1(bar):
return bar
@annotate("foo")
async def fun(foo):
bar = await current_injector().apply(fun1)
return {"foo": foo, "bar": bar}
result = await services.apply(fun)
assert result == {"foo": "I am foo", "bar": "I am bar"}
@pytest.mark.asyncio
async def test_noauto_partial_async_async(services):
@services.factory("foo")
async def foo_factory():
return "I am foo"
@services.factory("bar")
async def bar_factory():
return "I am bar"
@annotate("bar")
async def fun1(bar):
return bar
@annotate("foo")
async def fun(foo):
bar = await current_injector().apply(fun1)
return {"foo": foo, "bar": bar}
with pytest.raises(TypeError):
await fun()
@pytest.mark.asyncio
async def test_descriptor_1_decorated(services):
@services.factory("foo")
async def foo_factory():
return "I am foo"
class Toto:
cache = attr("foo")
toto = Toto
with services.auto(), pytest.raises(Exception):
await toto.cache
@pytest.mark.asyncio
async def test_descriptor_2_decorated(services):
@services.factory("foo")
async def foo_factory():
return "I am foo"
class Toto:
cache = attr("foo")
toto = Toto()
with services.auto():
cache = await toto.cache
assert cache == "I am foo"
@pytest.mark.asyncio
async def test_descriptor_1_not_decorated(services):
@services.factory("foo")
async def foo_factory():
return "I am foo"
class Toto:
cache = attr("foo")
toto = Toto
with pytest.raises(Exception):
toto.cache # Must fails
@pytest.mark.asyncio
async def test_descriptor_2_not_decorated(services):
@services.factory("foo")
async def foo_factory():
return "I am foo"
class Toto:
cache = attr("foo")
toto = Toto()
with pytest.raises(LookupError):
toto.cache
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment