-
Notifications
You must be signed in to change notification settings - Fork 6
Expand file tree
/
Copy pathconftest.py
More file actions
110 lines (80 loc) · 3.24 KB
/
conftest.py
File metadata and controls
110 lines (80 loc) · 3.24 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
import time
from typing import Callable, Iterator
import pytest
from lagom import Container
from redis import Redis
from sqlalchemy import Engine, create_engine
from sqlalchemy.orm import Session, sessionmaker
from testcontainers.postgres import PostgresContainer # type: ignore
from testcontainers.redis import RedisContainer # type: ignore
from smartschedule import container as container_module
from smartschedule.allocation.allocation_facade import AllocationFacade
from smartschedule.allocation.capabilityscheduling.capability_finder import (
CapabilityFinder,
)
from smartschedule.allocation.cashflow.cash_flow_facade import CashFlowFacade
from smartschedule.availability.availability_facade import AvailabilityFacade
from smartschedule.planning.planning_facade import PlanningFacade
from smartschedule.resource.employee.employee_facade import EmployeeFacade
from smartschedule.shared.sqlalchemy_extensions import registry
@pytest.fixture(scope="session")
def postgres_container() -> Iterator[PostgresContainer]:
with PostgresContainer("postgres:15") as container:
yield container
@pytest.fixture(scope="session")
def redis_container() -> Iterator[RedisContainer]:
with RedisContainer("redis:6") as container:
yield container
@pytest.fixture(scope="session")
def engine_for_tests(postgres_container: PostgresContainer) -> Engine:
url = postgres_container.get_connection_url()
engine = create_engine(url, echo=True)
# There seems to be some race condition. Fine, let's wait.
start = time.monotonic()
while time.monotonic() - start < 30:
try:
engine.connect()
break
except Exception:
time.sleep(0.1)
registry.metadata.create_all(engine)
return engine
@pytest.fixture(scope="session")
def session_factory(engine_for_tests: Engine) -> Callable[[], Session]:
return sessionmaker(bind=engine_for_tests)
@pytest.fixture()
def session(session_factory: Callable[[], Session]) -> Iterator[Session]:
a_session = session_factory()
yield a_session
a_session.close()
@pytest.fixture()
def clean_redis(redis_container: RedisContainer) -> Iterator[None]:
client = redis_container.get_client()
client.flushall()
yield
@pytest.fixture()
def container(
session: Session, redis_container: RedisContainer, clean_redis: None
) -> Container:
container = container_module.build()
container[Session] = session
container[Redis] = redis_container.get_client()
return container
@pytest.fixture()
def availability_facade(container: Container) -> AvailabilityFacade:
return container.resolve(AvailabilityFacade)
@pytest.fixture()
def capability_finder(container: Container) -> CapabilityFinder:
return container.resolve(CapabilityFinder)
@pytest.fixture()
def employee_facade(container: Container) -> EmployeeFacade:
return container.resolve(EmployeeFacade)
@pytest.fixture()
def planning_facade(container: Container) -> PlanningFacade:
return container.resolve(PlanningFacade)
@pytest.fixture()
def allocation_facade(container: Container) -> AllocationFacade:
return container.resolve(AllocationFacade)
@pytest.fixture()
def cash_flow_facade(container: Container) -> CashFlowFacade:
return container.resolve(CashFlowFacade)