mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-08-29 20:17:13 +08:00
58 lines
1.8 KiB
Python
58 lines
1.8 KiB
Python
"""async 阻塞调用 ratchet 与 debug 模式测试。"""
|
|
|
|
import asyncio
|
|
|
|
import pytest
|
|
|
|
from scripts.architecture.async_blocking import SCAN_ROOTS, compare_async_blocking
|
|
|
|
|
|
def test_async_blocking_scan_covers_runtime_entrypoints() -> None:
|
|
"""扫描范围必须覆盖全部 canonical 宿主目录和顶层运行入口。"""
|
|
assert {
|
|
"app/adapters",
|
|
"app/api",
|
|
"app/agent",
|
|
"app/application",
|
|
"app/chain",
|
|
"app/db",
|
|
"app/doctor",
|
|
"app/domain",
|
|
"app/foundation",
|
|
"app/monitor",
|
|
"app/modules",
|
|
"app/runtime",
|
|
"app/schemas",
|
|
"app/startup",
|
|
"app/workflow",
|
|
"app/cli.py",
|
|
"app/command.py",
|
|
"app/factory.py",
|
|
"app/main.py",
|
|
"app/scheduler.py",
|
|
}.issubset({str(path) for path in SCAN_ROOTS})
|
|
|
|
|
|
def test_async_blocking_ratchet_allows_removal_and_rejects_growth() -> None:
|
|
"""存量减少合法，新增或增加直接阻塞调用必须失败。"""
|
|
baseline = {
|
|
"app/api/a.py:stream:Path.read_text": 2,
|
|
"app/agent/a.py:run:time.sleep": 1,
|
|
}
|
|
reduced = {"app/api/a.py:stream:Path.read_text": 1}
|
|
increased = {
|
|
"app/api/a.py:stream:Path.read_text": 3,
|
|
"app/application/a.py:load:requests.get": 1,
|
|
}
|
|
|
|
assert compare_async_blocking(baseline, reduced) == []
|
|
problems = compare_async_blocking(baseline, increased)
|
|
assert any("增长" in problem for problem in problems)
|
|
assert any("新增" in problem for problem in problems)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_asyncio_debug_is_enabled_for_async_tests() -> None:
|
|
"""专项异步测试必须启用慢 callback 和阻塞诊断所需的 debug 模式。"""
|
|
assert asyncio.get_running_loop().get_debug() is True
|