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.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Raykov-MS 2026-09-17 16:56:08 +03:00
parent cceec9c4fb
commit 4e11db322d
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:
st.error(f"Ошибка доступа к токену OpenCloud: {exc}")
return None
except Exception: # noqa: BLE001
# Локальный тест без поднятой БД/переменных окружения — fallback.
except Exception as exc: # noqa: BLE001
# Раньше эта ветка тихо проглатывала ЛЮБУЮ ошибку (например,
# несовпадение 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

View File

@ -4,7 +4,7 @@ from cryptography.fernet import Fernet, InvalidToken
from sqlalchemy import select
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
@ -12,6 +12,16 @@ class OpenCloudTokenStoreError(RuntimeError):
"""Ошибка при работе с таблицей токенов 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:
if not settings.FERNET_KEY:
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()
async with SessionLocal() as session:
obj = await session.get(OpenCloudToken, username)
if obj is not None:
obj.encrypted_token = encrypted
if added_by:
obj.added_by = added_by
else:
session.add(
OpenCloudToken(
username=username, encrypted_token=encrypted, added_by=added_by
try:
async with SessionLocal() as session:
obj = await session.get(OpenCloudToken, username)
if obj is not None:
obj.encrypted_token = encrypted
if added_by:
obj.added_by = added_by
else:
session.add(
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:
"""username -> расшифрованный App Token, либо None, если записи нет."""
if not username:
return None
async with SessionLocal() as session:
obj = await session.get(OpenCloudToken, username)
try:
async with SessionLocal() as session:
obj = await session.get(OpenCloudToken, username)
finally:
await engine.dispose()
if obj is None:
return None
try:
@ -69,17 +85,23 @@ async def get_token(username: str) -> str | None:
async def remove_token(username: str) -> bool:
"""Удаляет запись пользователя. Возвращает True, если запись была найдена."""
async with SessionLocal() as session:
obj = await session.get(OpenCloudToken, username)
if obj is None:
return False
await session.delete(obj)
await session.commit()
return True
try:
async with SessionLocal() as session:
obj = await session.get(OpenCloudToken, username)
if obj is None:
return False
await session.delete(obj)
await session.commit()
return True
finally:
await engine.dispose()
async def list_usernames() -> list[str]:
"""Список username, для которых сохранён токен (без расшифровки)."""
async with SessionLocal() as session:
result = await session.execute(select(OpenCloudToken.username))
return list(result.scalars().all())
try:
async with SessionLocal() as session:
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
pre-commit==3.7.1
openpyxl
xlsxwriter==3.2.9
xlsxwriter==3.2.0
pytest
pytest-asyncio
requests==2.32.3
requests==2.32.4
pydantic-settings==2.10.1
sqlalchemy>=2.0.32,<2.1
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", ""))