Fix event-loop mismatch masking OpenCloud token as "not found" on first click

Each call into opencloud_token_store is wrapped in its own asyncio.run()
(app/app_config.py, main.py's token popover), so every call gets a fresh
event loop. The asyncpg connection pool is bound to the loop it was created
in, so the next asyncio.run() could try to reuse a connection from an
already-closed loop and fail — unrelated to actual DB availability, but the
broad except Exception in resolve_opencloud_access_token() silently
swallowed it and fell through to "token not found", masking the real cause.

- add_token/get_token/remove_token/list_usernames now call engine.dispose()
  in a finally block so each asyncio.run() starts with a clean pool.
- resolve_opencloud_access_token() now surfaces the real exception via
  st.error() before falling back to the .env token (kept only for local dev
  without a DB).
- Added tests/unit/test_opencloud_token_store.py covering that dispose() is
  called after every store operation, including when commit() raises.
This commit is contained in:
Raykov-MS 2026-09-17 16:56:08 +03:00
parent 7b90daa396
commit 0bdb83d2f3
4 changed files with 218 additions and 29 deletions

View File

@ -59,6 +59,12 @@ def resolve_opencloud_access_token() -> str | None:
except OpenCloudTokenStoreError as exc: except OpenCloudTokenStoreError as exc:
st.error(f"Ошибка доступа к токену OpenCloud: {exc}") st.error(f"Ошибка доступа к токену OpenCloud: {exc}")
return None return None
except Exception: # noqa: BLE001 except Exception as exc: # noqa: BLE001
# Локальный тест без поднятой БД/переменных окружения — fallback. # Раньше эта ветка тихо проглатывала ЛЮБУЮ ошибку (например,
# несовпадение event loop'ов asyncpg между последовательными
# asyncio.run() — см. app/opencloud_token_store.py) и превращала
# её в безобидное "токен не найден", маскируя реальный сбой БД.
# Показываем ошибку явно, fallback на .env — только если он там
# реально есть (локальная разработка без поднятой БД).
st.error(f"Ошибка доступа к БД токенов OpenCloud: {exc}")
return os.getenv("SFM_OPENCLOUD_ACCESS_TOKEN") or None return os.getenv("SFM_OPENCLOUD_ACCESS_TOKEN") or None

View File

@ -4,7 +4,7 @@ from cryptography.fernet import Fernet, InvalidToken
from sqlalchemy import select from sqlalchemy import select
from app.core.config import settings from app.core.config import settings
from app.db.base import SessionLocal from app.db.base import SessionLocal, engine
from app.db.models.opencloud_token import OpenCloudToken from app.db.models.opencloud_token import OpenCloudToken
@ -12,6 +12,16 @@ class OpenCloudTokenStoreError(RuntimeError):
"""Ошибка при работе с таблицей токенов OpenCloud (шифрование/БД).""" """Ошибка при работе с таблицей токенов OpenCloud (шифрование/БД)."""
# engine.dispose() в finally каждой функции ниже — обязателен. Приложение
# синхронное (Streamlit), каждый вызов этого модуля оборачивается в
# отдельный asyncio.run() (см. app/app_config.py, main.py), который создаёт
# новый event loop и закрывает его по завершении. Пул соединений asyncpg
# привязан к loop'у, в котором был создан: без dispose() следующий
# asyncio.run() (уже в новом loop'е) попытается забрать из пула соединение,
# открытое в уже закрытом loop'е предыдущего вызова, и упадёт — эта ошибка
# не имеет отношения к реальной доступности БД, но выглядит как она.
def _get_fernet() -> Fernet: def _get_fernet() -> Fernet:
if not settings.FERNET_KEY: if not settings.FERNET_KEY:
raise OpenCloudTokenStoreError( raise OpenCloudTokenStoreError(
@ -35,27 +45,33 @@ async def add_token(username: str, token: str, added_by: str | None = None) -> N
encrypted = _get_fernet().encrypt(token.encode()).decode() encrypted = _get_fernet().encrypt(token.encode()).decode()
async with SessionLocal() as session: try:
obj = await session.get(OpenCloudToken, username) async with SessionLocal() as session:
if obj is not None: obj = await session.get(OpenCloudToken, username)
obj.encrypted_token = encrypted if obj is not None:
if added_by: obj.encrypted_token = encrypted
obj.added_by = added_by if added_by:
else: obj.added_by = added_by
session.add( else:
OpenCloudToken( session.add(
username=username, encrypted_token=encrypted, added_by=added_by OpenCloudToken(
username=username, encrypted_token=encrypted, added_by=added_by
)
) )
) await session.commit()
await session.commit() finally:
await engine.dispose()
async def get_token(username: str) -> str | None: async def get_token(username: str) -> str | None:
"""username -> расшифрованный App Token, либо None, если записи нет.""" """username -> расшифрованный App Token, либо None, если записи нет."""
if not username: if not username:
return None return None
async with SessionLocal() as session: try:
obj = await session.get(OpenCloudToken, username) async with SessionLocal() as session:
obj = await session.get(OpenCloudToken, username)
finally:
await engine.dispose()
if obj is None: if obj is None:
return None return None
try: try:
@ -69,17 +85,23 @@ async def get_token(username: str) -> str | None:
async def remove_token(username: str) -> bool: async def remove_token(username: str) -> bool:
"""Удаляет запись пользователя. Возвращает True, если запись была найдена.""" """Удаляет запись пользователя. Возвращает True, если запись была найдена."""
async with SessionLocal() as session: try:
obj = await session.get(OpenCloudToken, username) async with SessionLocal() as session:
if obj is None: obj = await session.get(OpenCloudToken, username)
return False if obj is None:
await session.delete(obj) return False
await session.commit() await session.delete(obj)
return True await session.commit()
return True
finally:
await engine.dispose()
async def list_usernames() -> list[str]: async def list_usernames() -> list[str]:
"""Список username, для которых сохранён токен (без расшифровки).""" """Список username, для которых сохранён токен (без расшифровки)."""
async with SessionLocal() as session: try:
result = await session.execute(select(OpenCloudToken.username)) async with SessionLocal() as session:
return list(result.scalars().all()) result = await session.execute(select(OpenCloudToken.username))
return list(result.scalars().all())
finally:
await engine.dispose()

View File

@ -6,10 +6,10 @@ black==24.4.2
isort==5.13.2 isort==5.13.2
pre-commit==3.7.1 pre-commit==3.7.1
openpyxl openpyxl
xlsxwriter==3.2.9 xlsxwriter==3.2.0
pytest pytest
pytest-asyncio pytest-asyncio
requests==2.32.3 requests==2.32.4
pydantic-settings==2.10.1 pydantic-settings==2.10.1
sqlalchemy>=2.0.32,<2.1 sqlalchemy>=2.0.32,<2.1
asyncpg==0.30.0 asyncpg==0.30.0

View File

@ -0,0 +1,161 @@
from __future__ import annotations
import asyncio
from unittest.mock import AsyncMock
import pytest
from cryptography.fernet import Fernet
from app import opencloud_token_store as store
from app.opencloud_token_store import OpenCloudTokenStoreError
class FakeResult:
def __init__(self, usernames: list[str]) -> None:
self._usernames = usernames
def scalars(self):
return self
def all(self):
return self._usernames
class FakeRow:
def __init__(self, username: str, encrypted_token: str, added_by=None) -> None:
self.username = username
self.encrypted_token = encrypted_token
self.added_by = added_by
class FakeSession:
"""Мини-модель БД: {username: FakeRow}, без реального движка."""
def __init__(self, table: dict[str, FakeRow]) -> None:
self._table = table
async def get(self, model, username): # noqa: ANN001
_ = model
return self._table.get(username)
def add(self, obj: FakeRow) -> None:
self._table[obj.username] = obj
async def delete(self, obj: FakeRow) -> None:
self._table.pop(obj.username, None)
async def commit(self) -> None:
pass
async def execute(self, stmt): # noqa: ANN001
_ = stmt
return FakeResult(sorted(self._table.keys()))
async def __aenter__(self) -> FakeSession:
return self
async def __aexit__(self, exc_type, exc, tb) -> bool: # noqa: ANN001
return False
class FakeSessionLocal:
def __init__(self, table: dict[str, FakeRow]) -> None:
self._table = table
def __call__(self) -> FakeSession:
return FakeSession(self._table)
@pytest.fixture(autouse=True)
def _fake_store(monkeypatch):
"""Подменяет SessionLocal/engine и FERNET_KEY на in-memory заглушки.
Каждый вызов add_token/get_token/remove_token/list_usernames в проде
оборачивается ОТДЕЛЬНЫМ asyncio.run() (см. app/app_config.py) — новый
event loop на каждый вызов. Тесты ниже воспроизводят это же (несколько
независимых asyncio.run() подряд в одном тесте) и проверяют, что
engine.dispose() вызывается после каждой операции.
"""
table: dict[str, FakeRow] = {}
fake_engine = AsyncMock()
monkeypatch.setattr(store, "SessionLocal", FakeSessionLocal(table))
monkeypatch.setattr(store, "engine", fake_engine)
monkeypatch.setattr(store.settings, "FERNET_KEY", Fernet.generate_key().decode())
return table, fake_engine
def test_add_token_then_get_token_across_separate_event_loops(_fake_store) -> None:
table, fake_engine = _fake_store
asyncio.run(store.add_token("raykovmse", "coherent effects plausibly apple"))
assert fake_engine.dispose.await_count == 1
token = asyncio.run(store.get_token("raykovmse"))
assert fake_engine.dispose.await_count == 2
assert token == "coherent effects plausibly apple"
def test_add_token_updates_existing_row(_fake_store) -> None:
table, fake_engine = _fake_store
asyncio.run(store.add_token("raykovmse", "first token", added_by="admin"))
asyncio.run(store.add_token("raykovmse", "second token", added_by="raykovmse"))
assert fake_engine.dispose.await_count == 2
assert len(table) == 1
assert asyncio.run(store.get_token("raykovmse")) == "second token"
def test_get_token_missing_username_disposes_engine(_fake_store) -> None:
_, fake_engine = _fake_store
result = asyncio.run(store.get_token("no-such-user"))
assert result is None
assert fake_engine.dispose.await_count == 1
def test_remove_token_disposes_engine(_fake_store) -> None:
table, fake_engine = _fake_store
asyncio.run(store.add_token("raykovmse", "token-123"))
fake_engine.dispose.reset_mock()
removed = asyncio.run(store.remove_token("raykovmse"))
assert removed is True
assert "raykovmse" not in table
assert fake_engine.dispose.await_count == 1
def test_list_usernames_disposes_engine(_fake_store) -> None:
_, fake_engine = _fake_store
asyncio.run(store.add_token("raykovmse", "token-123"))
fake_engine.dispose.reset_mock()
usernames = asyncio.run(store.list_usernames())
assert usernames == ["raykovmse"]
assert fake_engine.dispose.await_count == 1
def test_add_token_disposes_engine_even_when_commit_fails(
_fake_store, monkeypatch
) -> None:
_, fake_engine = _fake_store
async def _failing_commit(self) -> None: # noqa: ANN001
raise RuntimeError("boom")
monkeypatch.setattr(FakeSession, "commit", _failing_commit)
with pytest.raises(RuntimeError):
asyncio.run(store.add_token("raykovmse", "token-123"))
assert fake_engine.dispose.await_count == 1
def test_add_token_requires_username_and_token(_fake_store) -> None:
with pytest.raises(OpenCloudTokenStoreError):
asyncio.run(store.add_token("", "token-123"))
with pytest.raises(OpenCloudTokenStoreError):
asyncio.run(store.add_token("raykovmse", ""))