-
-
Notifications
You must be signed in to change notification settings - Fork 4
/
test_lifespans.py
65 lines (44 loc) · 1.43 KB
/
test_lifespans.py
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
from contextlib import asynccontextmanager
from typing import AsyncIterator, List
import pytest
from xpresso import App, Dependant
from xpresso.routing.mount import Mount
from xpresso.testclient import TestClient
def test_lifespan_mounted_app() -> None:
class Counter(List[int]):
pass
@asynccontextmanager
async def lifespan(counter: Counter) -> AsyncIterator[None]:
counter.append(1)
yield
app = App(
routes=[Mount("/mounted-app", app=App(lifespan=lifespan))], lifespan=lifespan
)
counter = Counter()
app.container.register_by_type(Dependant(lambda: counter, scope="app"), Counter)
with TestClient(app):
pass
assert counter == [1, 1]
def test_lifespan_raises_exception_startup() -> None:
class SomeExc(Exception):
pass
@asynccontextmanager
async def lifespan() -> AsyncIterator[None]:
raise SomeExc
yield # type: ignore[unreachable]
app = App(lifespan=lifespan)
client = TestClient(app)
with pytest.raises(SomeExc):
client.__enter__()
def test_lifespan_raises_exception_shutdown() -> None:
class SomeExc(Exception):
pass
@asynccontextmanager
async def lifespan() -> AsyncIterator[None]:
yield
raise SomeExc
app = App(lifespan=lifespan)
client = TestClient(app)
client = client.__enter__()
with pytest.raises(SomeExc):
client.__exit__(None, None, None)