tests.experimental.test_async_context_based_executions.SomeWrapperDep.__init__() - Code Metrics - Inspection of "Fix bug where context singletons existed after cle..." - meadsteve/lagom - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 7dd739...b601db )

by Steve

created 2022-11-08 08:00 UTC

SomeWrapperDep.init() A

↳ Parent: tests.experimental.test_async_context_based_executions

Complexity

Conditions

Size

Total Lines	2
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	2
nop	2
dl	0
loc	2
rs	10
c	0
b	0
f	0

from contextlib import asynccontextmanager
from typing import ContextManager, AsyncGenerator, Awaitable

import pytest

from lagom import Container, dependency_definition
from lagom.decorators import context_dependency_definition
from lagom.exceptions import InvalidDependencyDefinition
from lagom.experimental.context_based import AsyncContextContainer, AwaitableSingleton


class SomeDep:
    global_clean_up_has_happened = False


class SomeWrapperDep:
    global_clean_up_has_happened = False

    def __init__(self, dep: SomeDep):
        pass


class SomeNotProperlySetupDef:
    pass


class Thing:
    contents: str

    def __init__(self, contents: str):
        self.contents = contents


container = Container()


@dependency_definition(container)
@asynccontextmanager
async def _load_a_some_dep_then_clean() -> AsyncGenerator[SomeDep, None]:
    try:
        yield SomeDep()
    finally:
        SomeDep.global_clean_up_has_happened = True


@context_dependency_definition(container)
async def _load_a_some_wrapper_dep_then_clean(
    c,
) -> AsyncGenerator[SomeWrapperDep, None]:
    try:
        yield SomeWrapperDep(await c[Awaitable[SomeDep]])
    finally:
        SomeWrapperDep.global_clean_up_has_happened = True


@pytest.mark.asyncio
async def test_clean_up_of_loaded_contexts_happens_on_container_exit():
    SomeDep.global_clean_up_has_happened = False

    async with AsyncContextContainer(
        container, context_types=[SomeDep]
    ) as context_container:
        assert isinstance(await context_container[Awaitable[SomeDep]], SomeDep)
        assert not SomeDep.global_clean_up_has_happened
    assert SomeDep.global_clean_up_has_happened


@pytest.mark.asyncio
async def test_context_instances_are_not_singletons():
    async with AsyncContextContainer(
        container, context_types=[SomeDep]
    ) as context_container:
        one = context_container[Awaitable[SomeDep]]
        two = context_container[Awaitable[SomeDep]]
        assert one is not two


@pytest.mark.asyncio
async def test_context_instances_can_be_made_singletons():
    SomeDep.global_clean_up_has_happened = False
    async with AsyncContextContainer(
        container, context_types=[], context_singletons=[SomeDep]
    ) as context_container:
        one = await context_container[AwaitableSingleton[SomeDep]].get()
        two = await context_container[AwaitableSingleton[SomeDep]].get()
        assert one is two
    assert SomeDep.global_clean_up_has_happened


@pytest.mark.asyncio
async def test_context_instance_singletons_only_have_a_lifespan_of_the_with():
    SomeDep.global_clean_up_has_happened = False
    context_container = AsyncContextContainer(
        container, context_types=[], context_singletons=[SomeDep]
    )
    async with context_container as c:
        one = await c[AwaitableSingleton[SomeDep]].get()
    async with context_container as c:
        two = await c[AwaitableSingleton[SomeDep]].get()
    assert one is not two


@pytest.mark.asyncio
async def test_clean_up_of_loaded_contexts_happens_recursively_on_container_exit():
    SomeDep.global_clean_up_has_happened = False
    SomeWrapperDep.global_clean_up_has_happened = False

    async with AsyncContextContainer(
        container, context_types=[SomeDep, SomeWrapperDep]
    ) as context_container:
        await context_container[Awaitable[SomeDep]]
        assert isinstance(
            await context_container[Awaitable[SomeWrapperDep]], SomeWrapperDep
        )
        assert not SomeDep.global_clean_up_has_happened
        assert not SomeWrapperDep.global_clean_up_has_happened

    assert SomeDep.global_clean_up_has_happened
    assert SomeWrapperDep.global_clean_up_has_happened


@pytest.mark.asyncio
async def test_it_fails_if_the_dependencies_arent_defined_correctly():
    with pytest.raises(InvalidDependencyDefinition) as failure:
        async with AsyncContextContainer(
            container, context_types=[SomeNotProperlySetupDef]
        ) as context_container:
            context_container.resolve(SomeNotProperlySetupDef)
    assert f"A ContextManager[{SomeNotProperlySetupDef}] should be defined" in str(
        failure.value
    )


@pytest.mark.asyncio

async def test_it_works_with_actual_context_managers():
    fresh_container = Container(container)

    class ThingManager:
        def __enter__(self):
            return Thing("managed thing")

        def __exit__(self, exc_type, exc_val, exc_tb):
            pass

    fresh_container[ContextManager[Thing]] = ThingManager  # type: ignore

    async with AsyncContextContainer(
        fresh_container, context_types=[Thing]

    ) as context_container:
        assert context_container.resolve(Thing).contents == "managed thing"


@pytest.mark.asyncio

async def test_it_works_with_actual_context_managers_as_singletons():
    fresh_container = Container(container)

    class ThingManager:
        def __enter__(self):
            return Thing("managed thing")

        def __exit__(self, exc_type, exc_val, exc_tb):
            pass

    fresh_container[ContextManager[Thing]] = ThingManager  # type: ignore

    async with AsyncContextContainer(
        fresh_container, context_types=[], context_singletons=[Thing]

    ) as context_container:
        assert context_container.resolve(Thing).contents == "managed thing"


@pytest.mark.asyncio
async def test_the_container_can_be_reused():
    original = AsyncContextContainer(container, context_types=[SomeDep])
    async with original as context_container_1:
        a = await context_container_1.resolve(Awaitable[SomeDep])
    async with original as context_container_2:
        b = await context_container_2.resolve(Awaitable[SomeDep])
    assert a != b


@pytest.mark.asyncio
async def test_the_container_can_be_nested_though_this_has_no_meaning():
    original = AsyncContextContainer(container, context_types=[SomeDep])
    async with original as context_container_1:
        a = await context_container_1.resolve(Awaitable[SomeDep])
        async with context_container_1 as context_container_2:
            b = context_container_2.resolve(Awaitable[SomeDep])
    assert a != b


1		from contextlib import asynccontextmanager
2		from typing import ContextManager, AsyncGenerator, Awaitable
3
4		import pytest
5
6		from lagom import Container, dependency_definition
7		from lagom.decorators import context_dependency_definition
8		from lagom.exceptions import InvalidDependencyDefinition
9		from lagom.experimental.context_based import AsyncContextContainer, AwaitableSingleton
10
11
12		class SomeDep:
13		global_clean_up_has_happened = False
14
15
16		class SomeWrapperDep:
17		global_clean_up_has_happened = False
18
19		def __init__(self, dep: SomeDep):
20		pass
21
22
23		class SomeNotProperlySetupDef:
24		pass
25
26
27		class Thing:
28		contents: str
29
30		def __init__(self, contents: str):
31		self.contents = contents
32
33
34		container = Container()
35
36
37		@dependency_definition(container)
38		@asynccontextmanager
39		async def _load_a_some_dep_then_clean() -> AsyncGenerator[SomeDep, None]:
40		try:
41		yield SomeDep()
42		finally:
43		SomeDep.global_clean_up_has_happened = True
44
45
46		@context_dependency_definition(container)
47		async def _load_a_some_wrapper_dep_then_clean(
48		c,
49		) -> AsyncGenerator[SomeWrapperDep, None]:
50		try:
51		yield SomeWrapperDep(await c[Awaitable[SomeDep]])
52		finally:
53		SomeWrapperDep.global_clean_up_has_happened = True
54
55
56		@pytest.mark.asyncio
57		async def test_clean_up_of_loaded_contexts_happens_on_container_exit():
58		SomeDep.global_clean_up_has_happened = False
59
60		async with AsyncContextContainer(
61		container, context_types=[SomeDep]
62		) as context_container:
63		assert isinstance(await context_container[Awaitable[SomeDep]], SomeDep)
64		assert not SomeDep.global_clean_up_has_happened
65		assert SomeDep.global_clean_up_has_happened
66
67
68		@pytest.mark.asyncio
69		async def test_context_instances_are_not_singletons():
70		async with AsyncContextContainer(
71		container, context_types=[SomeDep]
72		) as context_container:
73		one = context_container[Awaitable[SomeDep]]
74		two = context_container[Awaitable[SomeDep]]
75		assert one is not two
76
77
78		@pytest.mark.asyncio
79		async def test_context_instances_can_be_made_singletons():
80		SomeDep.global_clean_up_has_happened = False
81		async with AsyncContextContainer(
82		container, context_types=[], context_singletons=[SomeDep]
83		) as context_container:
84		one = await context_container[AwaitableSingleton[SomeDep]].get()
85		two = await context_container[AwaitableSingleton[SomeDep]].get()
86		assert one is two
87		assert SomeDep.global_clean_up_has_happened
88
89
90		@pytest.mark.asyncio
91		async def test_context_instance_singletons_only_have_a_lifespan_of_the_with():
92		SomeDep.global_clean_up_has_happened = False
93		context_container = AsyncContextContainer(
94		container, context_types=[], context_singletons=[SomeDep]
95		)
96		async with context_container as c:
97		one = await c[AwaitableSingleton[SomeDep]].get()
98		async with context_container as c:
99		two = await c[AwaitableSingleton[SomeDep]].get()
100		assert one is not two
101
102
103		@pytest.mark.asyncio
104		async def test_clean_up_of_loaded_contexts_happens_recursively_on_container_exit():
105		SomeDep.global_clean_up_has_happened = False
106		SomeWrapperDep.global_clean_up_has_happened = False
107
108		async with AsyncContextContainer(
109		container, context_types=[SomeDep, SomeWrapperDep]
110		) as context_container:
111		await context_container[Awaitable[SomeDep]]
112		assert isinstance(
113		await context_container[Awaitable[SomeWrapperDep]], SomeWrapperDep
114		)
115		assert not SomeDep.global_clean_up_has_happened
116		assert not SomeWrapperDep.global_clean_up_has_happened
117
118		assert SomeDep.global_clean_up_has_happened
119		assert SomeWrapperDep.global_clean_up_has_happened
120
121
122		@pytest.mark.asyncio
123		async def test_it_fails_if_the_dependencies_arent_defined_correctly():
124		with pytest.raises(InvalidDependencyDefinition) as failure:
125		async with AsyncContextContainer(
126		container, context_types=[SomeNotProperlySetupDef]
127		) as context_container:
128		context_container.resolve(SomeNotProperlySetupDef)
129		assert f"A ContextManager[{SomeNotProperlySetupDef}] should be defined" in str(
130		failure.value
131		)
132
133
134	View Code Duplication	@pytest.mark.asyncio
		0 ignored issues – show Duplication introduced 2022-11-05 09:49 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
135		async def test_it_works_with_actual_context_managers():
136		fresh_container = Container(container)
137
138		class ThingManager:
139		def __enter__(self):
140		return Thing("managed thing")
141
142		def __exit__(self, exc_type, exc_val, exc_tb):
143		pass
144
145		fresh_container[ContextManager[Thing]] = ThingManager # type: ignore
146
147		async with AsyncContextContainer(
148		fresh_container, context_types=[Thing]
		0 ignored issues – show Comprehensibility Best Practice introduced 2022-11-05 09:49 UTC by Report Bug Copy Issue Report The variable `Thing` does not seem to be defined. Loading history...
149		) as context_container:
150		assert context_container.resolve(Thing).contents == "managed thing"
151
152
153	View Code Duplication	@pytest.mark.asyncio
		0 ignored issues – show Duplication introduced 2022-11-05 09:49 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
154		async def test_it_works_with_actual_context_managers_as_singletons():
155		fresh_container = Container(container)
156
157		class ThingManager:
158		def __enter__(self):
159		return Thing("managed thing")
160
161		def __exit__(self, exc_type, exc_val, exc_tb):
162		pass
163
164		fresh_container[ContextManager[Thing]] = ThingManager # type: ignore
165
166		async with AsyncContextContainer(
167		fresh_container, context_types=[], context_singletons=[Thing]
		0 ignored issues – show Comprehensibility Best Practice introduced 2022-11-05 09:49 UTC by Report Bug Copy Issue Report The variable `Thing` does not seem to be defined. Loading history...
168		) as context_container:
169		assert context_container.resolve(Thing).contents == "managed thing"
170
171
172		@pytest.mark.asyncio
173		async def test_the_container_can_be_reused():
174		original = AsyncContextContainer(container, context_types=[SomeDep])
175		async with original as context_container_1:
176		a = await context_container_1.resolve(Awaitable[SomeDep])
177		async with original as context_container_2:
178		b = await context_container_2.resolve(Awaitable[SomeDep])
179		assert a != b
180
181
182		@pytest.mark.asyncio
183		async def test_the_container_can_be_nested_though_this_has_no_meaning():
184		original = AsyncContextContainer(container, context_types=[SomeDep])
185		async with original as context_container_1:
186		a = await context_container_1.resolve(Awaitable[SomeDep])
187		async with context_container_1 as context_container_2:
188		b = context_container_2.resolve(Awaitable[SomeDep])
189		assert a != b
190

meadsteve / lagom

Push — master ( 7dd739...b601db )

SomeWrapperDep.__init__() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

SomeWrapperDep.init() A