ayurishchevandClaude Opus 5.5
e1f197ccff
Пункт 12 ревью 2026-09-28 17:35 (docs/changes/026): tests/test_app.py (936 строк) разнесён на test_security, test_devices, test_operations, test_backups, test_events, test_ids_migrations, test_architecture; общие хелперы — tests/helpers.py. Тела тестов не менялись (AST совпадает). security.reset_all_failures() — фикстура сбрасывает счётчики попыток по всем ключам, тесты не зависят от порядка. Тесты: 36 из 36 полным прогоном, каждый модуль отдельно и в обратном порядке; тест-линтер после переноса проверен мутацией. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
69 lines
3.1 KiB
Python
69 lines
3.1 KiB
Python
import ast
|
|
from pathlib import Path
|
|
|
|
|
|
def test_no_sync_db_calls_in_async_functions():
|
|
"""Регресс п.9 ревью: в async-функциях app/ не должно быть прямых вызовов синхронных функций,
|
|
обращающихся к БД (session_scope) — напрямую или через один уровень вызовов, — это блокирует event loop.
|
|
Правильный способ — asyncio.to_thread(fn, ...). Ссылка на функцию, переданная в asyncio.to_thread(fn, ...),
|
|
вызовом не считается; тела вложенных def/async def в область видимости внешней функции не входят.
|
|
Единственное исключение — app.main.lifespan (выполняется до приёма запросов)."""
|
|
app_dir = Path(__file__).resolve().parent.parent / "app"
|
|
allowed = {"lifespan"}
|
|
|
|
class _OwnScope(ast.NodeVisitor):
|
|
"""Вызовы прямо в теле функции, не заходя в тела вложенных def/async def."""
|
|
|
|
def __init__(self):
|
|
self.calls: list[ast.Call] = []
|
|
|
|
def visit_FunctionDef(self, node): # не спускаемся во вложенную функцию
|
|
pass
|
|
|
|
def visit_AsyncFunctionDef(self, node):
|
|
pass
|
|
|
|
def visit_Call(self, node):
|
|
self.calls.append(node)
|
|
self.generic_visit(node)
|
|
|
|
def own_calls(func) -> list[ast.Call]:
|
|
v = _OwnScope()
|
|
for stmt in func.body:
|
|
v.visit(stmt)
|
|
return v.calls
|
|
|
|
def call_name(call: ast.Call) -> str | None:
|
|
f = call.func
|
|
if isinstance(f, ast.Name):
|
|
return f.id
|
|
if isinstance(f, ast.Attribute):
|
|
return f.attr
|
|
return None
|
|
|
|
sync_funcs, async_funcs = [], []
|
|
for path in sorted(app_dir.rglob("*.py")):
|
|
tree = ast.parse(path.read_text(), filename=str(path))
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, ast.AsyncFunctionDef):
|
|
async_funcs.append((path, node))
|
|
elif isinstance(node, ast.FunctionDef):
|
|
sync_funcs.append((path, node))
|
|
|
|
# функции, обращающиеся к БД напрямую, и функции, вызывающие их (один уровень)
|
|
level0 = {f.name for _, f in sync_funcs if any(call_name(c) == "session_scope" for c in own_calls(f))}
|
|
level1 = {f.name for _, f in sync_funcs
|
|
if f.name not in level0 and any(call_name(c) in level0 for c in own_calls(f))}
|
|
dangerous = level0 | level1
|
|
|
|
violations = []
|
|
for path, func in async_funcs:
|
|
if func.name in allowed:
|
|
continue
|
|
for c in own_calls(func):
|
|
name = call_name(c)
|
|
if name in dangerous:
|
|
violations.append(f"{path.relative_to(app_dir.parent)}::{func.name} вызывает {name}() напрямую, "
|
|
"в обход asyncio.to_thread — обращение к БД блокирует event loop")
|
|
assert not violations, "\n".join(violations)
|