-
-
Notifications
You must be signed in to change notification settings - Fork 5
Expand file tree
/
Copy pathtest_alias.py
More file actions
134 lines (93 loc) · 4.68 KB
/
Copy pathtest_alias.py
File metadata and controls
134 lines (93 loc) · 4.68 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
import dataclasses
import pytest
from modern_di import Container, Group, Scope, providers
from modern_di.exceptions import (
AliasSourceNotRegisteredError,
CircularDependencyError,
ScopeNotInitializedError,
ValidationFailedError,
)
class AbstractRepository: ...
@dataclasses.dataclass(kw_only=True, slots=True, frozen=True)
class PostgresRepository(AbstractRepository):
dsn: str = "postgres://localhost"
class MyGroup(Group):
repo = providers.Factory(creator=PostgresRepository, cache_settings=providers.CacheSettings())
abstract_repo = providers.Alias(source_type=PostgresRepository, bound_type=AbstractRepository)
def test_alias_delegates_to_source() -> None:
container = Container(groups=[MyGroup], validate=True)
concrete = container.resolve(PostgresRepository)
abstract = container.resolve(AbstractRepository)
assert isinstance(abstract, PostgresRepository)
assert concrete is abstract
def test_alias_without_caching_returns_fresh_instance_per_call() -> None:
class G(Group):
repo = providers.Factory(creator=PostgresRepository)
abstract = providers.Alias(source_type=PostgresRepository, bound_type=AbstractRepository)
container = Container(groups=[G])
a = container.resolve(AbstractRepository)
b = container.resolve(PostgresRepository)
assert isinstance(a, PostgresRepository)
assert isinstance(b, PostgresRepository)
assert a is not b
def test_alias_respects_source_scope() -> None:
class G(Group):
repo = providers.Factory(scope=Scope.REQUEST, creator=PostgresRepository)
abstract = providers.Alias(source_type=PostgresRepository, bound_type=AbstractRepository)
app_container = Container(groups=[G])
with pytest.raises(ScopeNotInitializedError):
app_container.resolve(AbstractRepository)
request_container = app_container.build_child_container(scope=Scope.REQUEST)
instance = request_container.resolve(AbstractRepository)
assert isinstance(instance, PostgresRepository)
def test_alias_override_does_not_affect_source() -> None:
container = Container(groups=[MyGroup])
mock = PostgresRepository(dsn="mock-alias")
container.override(MyGroup.abstract_repo, mock)
assert container.resolve(AbstractRepository) is mock
assert container.resolve(PostgresRepository) is not mock
def test_source_override_propagates_through_alias() -> None:
container = Container(groups=[MyGroup])
mock = PostgresRepository(dsn="mock-source")
container.override(MyGroup.repo, mock)
assert container.resolve(PostgresRepository) is mock
assert container.resolve(AbstractRepository) is mock
def test_alias_missing_source_raises_on_resolve() -> None:
class G(Group):
abstract = providers.Alias(source_type=PostgresRepository, bound_type=AbstractRepository)
container = Container(groups=[G])
with pytest.raises(AliasSourceNotRegisteredError, match="PostgresRepository") as exc:
container.resolve(AbstractRepository)
assert exc.value.source_type is PostgresRepository
def test_alias_missing_source_raises_on_validate_provider() -> None:
class G(Group):
abstract = providers.Alias(source_type=PostgresRepository, bound_type=AbstractRepository)
container = Container(groups=[G])
with pytest.raises(AliasSourceNotRegisteredError, match="PostgresRepository"):
container.resolve_provider(G.abstract)
def test_alias_missing_source_raises_on_container_validate() -> None:
class G(Group):
abstract = providers.Alias(source_type=PostgresRepository, bound_type=AbstractRepository)
with pytest.raises(AliasSourceNotRegisteredError, match="PostgresRepository"):
Container(groups=[G], validate=True)
def test_alias_participates_in_cycle_detection() -> None:
class Iface: ...
@dataclasses.dataclass(kw_only=True, slots=True, frozen=True)
class Concrete(Iface):
dep: Iface
class G(Group):
concrete = providers.Factory(creator=Concrete)
iface_alias = providers.Alias(source_type=Concrete, bound_type=Iface)
with pytest.raises(ValidationFailedError) as exc:
Container(groups=[G], validate=True)
[issue] = exc.value.errors
assert isinstance(issue, CircularDependencyError)
assert "Concrete" in str(issue)
def test_alias_default_bound_type_is_source_type() -> None:
alias = providers.Alias(source_type=PostgresRepository)
assert alias.bound_type is PostgresRepository
def test_alias_repr() -> None:
alias = providers.Alias(source_type=PostgresRepository, bound_type=AbstractRepository, scope=Scope.REQUEST)
assert repr(alias) == (
f"Alias(source_type={PostgresRepository!r}, bound_type={AbstractRepository!r}, scope=<Scope.REQUEST: 3>)"
)