46 lines
1.4 KiB
Python
46 lines
1.4 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
import app as app_module
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_lifespan_uses_registered_services(
|
||
|
|
monkeypatch: pytest.MonkeyPatch,
|
||
|
|
) -> None:
|
||
|
|
initialized_services = [object(), object()]
|
||
|
|
calls: dict[str, object] = {}
|
||
|
|
|
||
|
|
async def _fake_initialize(service_names: list[str]) -> tuple[bool, list[object]]:
|
||
|
|
calls["init_names"] = service_names
|
||
|
|
return True, initialized_services
|
||
|
|
|
||
|
|
async def _fake_close(services: list[object]) -> bool:
|
||
|
|
calls["close_services"] = services
|
||
|
|
return True
|
||
|
|
|
||
|
|
monkeypatch.setattr(app_module, "initialize_registered_services", _fake_initialize)
|
||
|
|
monkeypatch.setattr(app_module, "close_registered_services", _fake_close)
|
||
|
|
|
||
|
|
context = app_module.lifespan(app_module.app)
|
||
|
|
await context.__aenter__()
|
||
|
|
await context.__aexit__(None, None, None)
|
||
|
|
|
||
|
|
assert calls["init_names"] == ["redis", "supabase"]
|
||
|
|
assert calls["close_services"] == initialized_services
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_lifespan_raises_when_initialization_failed(
|
||
|
|
monkeypatch: pytest.MonkeyPatch,
|
||
|
|
) -> None:
|
||
|
|
async def _fake_initialize(_: list[str]) -> tuple[bool, list[object]]:
|
||
|
|
return False, []
|
||
|
|
|
||
|
|
monkeypatch.setattr(app_module, "initialize_registered_services", _fake_initialize)
|
||
|
|
|
||
|
|
context = app_module.lifespan(app_module.app)
|
||
|
|
with pytest.raises(RuntimeError, match="Service initialization failed"):
|
||
|
|
await context.__aenter__()
|