From 1c56d39b092c7094b2b9826f606888d4db656b45 Mon Sep 17 00:00:00 2001 From: Andrey Epifancev Date: Sat, 3 Oct 2026 18:42:54 +0400 Subject: [PATCH] =?UTF-8?q?test:=20=D1=80=D0=B5=D0=B3=D1=80=D0=B5=D1=81?= =?UTF-8?q?=D1=81=D0=B8=D0=BE=D0=BD=D0=BD=D1=8B=D0=B5=20=D1=82=D0=B5=D1=81?= =?UTF-8?q?=D1=82=D1=8B=20lifespan=20=E2=80=94=20=D1=81=D1=82=D0=B0=D1=80?= =?UTF-8?q?=D1=82=20=D0=BF=D1=80=D0=B8=20=D0=BD=D0=B5=D0=B4=D0=BE=D1=81?= =?UTF-8?q?=D1=82=D1=83=D0=BF=D0=BD=D1=8B=D1=85=20S3/NATS=20=D0=B8=20?= =?UTF-8?q?=D0=B7=D0=B0=D0=BA=D1=80=D1=8B=D1=82=D0=B8=D0=B5=20=D1=81=D0=BE?= =?UTF-8?q?=D0=B7=D0=B4=D0=B0=D0=BD=D0=BD=D1=8B=D1=85=20=D1=80=D0=B5=D1=81?= =?UTF-8?q?=D1=83=D1=80=D1=81=D0=BE=D0=B2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Fable 5.1 --- api/tests/test_app.py | 64 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 64 insertions(+) diff --git a/api/tests/test_app.py b/api/tests/test_app.py index 601ddff..6dc057b 100644 --- a/api/tests/test_app.py +++ b/api/tests/test_app.py @@ -93,3 +93,67 @@ async def test_healthcheck_nats_follows_connection_state( def test_router_prefix(): assert router.prefix == "/api/v1" + + +class _FakeNats: + def __init__(self) -> None: + self.closed = False + + async def close(self) -> None: + self.closed = True + + +class _FakeS3Client: + def __init__(self) -> None: + self.session_closed = False + + async def close_session(self) -> None: + self.session_closed = True + + +@pytest.fixture +def _no_smb(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(app_module.smbclient, "register_session", lambda *args, **kwargs: None) + monkeypatch.setattr(app_module.smbclient, "reset_connection_cache", lambda: None) + + +@pytest.mark.usefixtures("_no_smb") +async def test_lifespan_starts_when_s3_and_nats_unavailable(monkeypatch: pytest.MonkeyPatch): + async def unavailable() -> None: + return None + + monkeypatch.setattr(app_module, "_connect_s3", unavailable) + monkeypatch.setattr(app_module, "_connect_nats", unavailable) + reached_yield = False + + async with app_module.lifespan(app_module.app): + reached_yield = True + assert app_module.app.state.storage is None + assert app_module.app.state.nats is None + + assert reached_yield + + +@pytest.mark.usefixtures("_no_smb") +async def test_lifespan_closes_created_resources(monkeypatch: pytest.MonkeyPatch): + fake_nats = _FakeNats() + s3_client = _FakeS3Client() + storage = SimpleNamespace(client=s3_client) + + async def connect_s3() -> SimpleNamespace: + return storage + + async def connect_nats() -> _FakeNats: + return fake_nats + + monkeypatch.setattr(app_module, "_connect_s3", connect_s3) + monkeypatch.setattr(app_module, "_connect_nats", connect_nats) + + async with app_module.lifespan(app_module.app): + assert app_module.app.state.storage is storage + assert app_module.app.state.nats is fake_nats + assert not fake_nats.closed + assert not s3_client.session_closed + + assert fake_nats.closed + assert s3_client.session_closed