diff --git a/.trivyignore.yaml b/.trivyignore.yaml index 7f9c73026..e8614a70a 100644 --- a/.trivyignore.yaml +++ b/.trivyignore.yaml @@ -4,58 +4,68 @@ vulnerabilities: - Python purls: - pkg:pypi/msgpack@1.1.2 + expired_at: 2026-11-20 statement: The finding belongs to the base image's system pip and is not imported by MoviePilot. - id: CVE-2025-47273 paths: - Python purls: - pkg:pypi/setuptools@70.3.0 + expired_at: 2026-11-20 statement: The finding belongs to the base image's system pip and is not used for dependency installation. - id: CVE-2026-33818 paths: - usr/bin/rclone purls: - pkg:golang/stdlib@v1.26.5 + expired_at: 2026-11-20 statement: The official rclone binary has no patched release for this embedded Go runtime yet. - id: CVE-2026-39821 paths: - usr/bin/rclone purls: - pkg:golang/stdlib@v1.26.5 + expired_at: 2026-11-20 statement: The official rclone binary has no patched release for this embedded Go runtime yet. - id: CVE-2026-46600 paths: - usr/bin/rclone purls: - pkg:golang/stdlib@v1.26.5 + expired_at: 2026-11-20 statement: The official rclone binary has no patched release for this embedded Go runtime yet. - id: CVE-2026-56853 paths: - usr/bin/rclone purls: - pkg:golang/stdlib@v1.26.5 + expired_at: 2026-11-20 statement: The official rclone binary has no patched release for this embedded Go runtime yet. - id: CVE-2026-56858 paths: - usr/bin/rclone purls: - pkg:golang/stdlib@v1.26.5 + expired_at: 2026-11-20 statement: The official rclone binary has no patched release for this embedded Go runtime yet. - id: CVE-2026-56859 paths: - usr/bin/rclone purls: - pkg:golang/stdlib@v1.26.5 + expired_at: 2026-11-20 statement: The official rclone binary has no patched release for this embedded Go runtime yet. - id: CVE-2026-56860 paths: - usr/bin/rclone purls: - pkg:golang/stdlib@v1.26.5 + expired_at: 2026-11-20 statement: The official rclone binary has no patched release for this embedded Go runtime yet. - id: CVE-2026-56862 paths: - usr/bin/rclone purls: - pkg:golang/stdlib@v1.26.5 + expired_at: 2026-11-20 statement: The official rclone binary has no patched release for this embedded Go runtime yet. diff --git a/app/adapters/system/backup/database.py b/app/adapters/system/backup/database.py index a4f98c294..55201a1e3 100644 --- a/app/adapters/system/backup/database.py +++ b/app/adapters/system/backup/database.py @@ -46,6 +46,73 @@ class ProcessRunner(Protocol): """执行命令并返回结果。""" +def verify_database_backup( + artifact: Path, + *, + db_type: str, + runner: ProcessRunner = subprocess.run, + tool_resolver: Callable[[str], str | None] = shutil.which, + pg_restore: str = "pg_restore", +) -> DatabaseBackupCheck: + """在不访问活动数据库的前提下校验一个受管备份文件。""" + if db_type == "sqlite": + method = "PRAGMA integrity_check" + try: + # 正式备份不会再变化，immutable 可避免只读校验创建 WAL 旁路文件。 + uri = f"{artifact.resolve().as_uri()}?mode=ro&immutable=1" + with closing(sqlite3.connect(uri, uri=True)) as connection: + rows = connection.execute("PRAGMA integrity_check").fetchall() + except sqlite3.Error as error: + return DatabaseBackupCheck(False, method, str(error)) + valid = bool(rows) and all(row[0] == "ok" for row in rows) + detail = None if valid else "; ".join(str(row[0]) for row in rows) + return DatabaseBackupCheck(valid, method, detail) + + if db_type == "postgresql": + method = "pg_restore --list" + executable = _require_tool(pg_restore, tool_resolver) + result = runner( + [executable, "--list", str(artifact)], + env=_postgres_environment(), + capture_output=True, + text=True, + check=False, + ) + valid = result.returncode == 0 and bool(result.stdout.strip()) + detail = None if valid else f"pg_restore 退出码 {result.returncode}" + return DatabaseBackupCheck(valid, method, detail) + + raise ValueError(f"不支持的数据库备份类型：{db_type}") + + +def _require_tool( + executable: str, + tool_resolver: Callable[[str], str | None], +) -> str: + resolved = tool_resolver(executable) + if resolved is None: + raise RuntimeError( + f"未找到 {executable}，请安装与服务端同主版本或更高的 " + "PostgreSQL client 并加入 PATH" + ) + return resolved + + +def _postgres_environment( + *, + password: str | None = None, + sslmode: str | None = None, +) -> dict[str, str]: + environment = dict(os.environ) + environment.pop("PGPASSWORD", None) + environment.pop("PGSSLMODE", None) + if password: + environment["PGPASSWORD"] = password + if sslmode: + environment["PGSSLMODE"] = sslmode + return environment + + class SQLiteBackupBackend: """使用 SQLite 在线备份 API 管理活动文件数据库。""" @@ -71,17 +138,7 @@ class SQLiteBackupBackend: def verify(self, artifact: Path) -> DatabaseBackupCheck: """通过 SQLite integrity_check 校验备份内容。""" - method = "PRAGMA integrity_check" - try: - # 已发布前的临时快照不会再变化；immutable 避免 WAL 模式为只读校验创建旁路文件。 - uri = f"{artifact.resolve().as_uri()}?mode=ro&immutable=1" - with closing(sqlite3.connect(uri, uri=True)) as connection: - rows = connection.execute("PRAGMA integrity_check").fetchall() - except sqlite3.Error as error: - return DatabaseBackupCheck(False, method, str(error)) - valid = bool(rows) and all(row[0] == "ok" for row in rows) - detail = None if valid else "; ".join(str(row[0]) for row in rows) - return DatabaseBackupCheck(valid, method, detail) + return verify_database_backup(artifact, db_type=self.db_type) def restore(self, artifact: Path) -> None: """在 CLI 离线进程中原子替换活动 SQLite 文件。""" @@ -137,14 +194,13 @@ class PostgreSQLBackupBackend: def verify(self, artifact: Path) -> DatabaseBackupCheck: """通过 pg_restore 目录读取校验 custom-format 归档。""" - method = "pg_restore --list" - result = self._run( - [self._require_tool(self._pg_restore), "--list", str(artifact)], - include_password=False, + return verify_database_backup( + artifact, + db_type=self.db_type, + runner=self._runner, + tool_resolver=self._tool_resolver, + pg_restore=self._pg_restore, ) - valid = result.returncode == 0 and bool(result.stdout.strip()) - detail = None if valid else f"pg_restore 退出码 {result.returncode}" - return DatabaseBackupCheck(valid, method, detail) def restore(self, artifact: Path) -> None: """在 CLI 离线进程中覆盖当前 PostgreSQL 数据库内容。""" @@ -189,21 +245,16 @@ class PostgreSQLBackupBackend: ) def _require_tool(self, executable: str) -> str: - resolved = self._tool_resolver(executable) - if resolved is None: - raise RuntimeError( - f"未找到 {executable}，请安装与服务端同主版本或更高的 " - "PostgreSQL client 并加入 PATH" - ) - return resolved + return _require_tool(executable, self._tool_resolver) def _environment(self, *, include_password: bool) -> dict[str, str]: - environment = dict(os.environ) - environment.pop("PGPASSWORD", None) - environment.pop("PGSSLMODE", None) - if include_password and self._engine.url.password: - environment["PGPASSWORD"] = str(self._engine.url.password) + password = ( + str(self._engine.url.password) + if include_password and self._engine.url.password + else None + ) sslmode = self._engine.url.query.get("sslmode") - if sslmode: - environment["PGSSLMODE"] = str(sslmode) - return environment + return _postgres_environment( + password=password, + sslmode=str(sslmode) if sslmode else None, + ) diff --git a/app/application/formatting.py b/app/application/formatting.py index 029d75fe5..26b73e0b2 100644 --- a/app/application/formatting.py +++ b/app/application/formatting.py @@ -7,7 +7,7 @@ from typing import Dict, Iterable, List, Match, Optional, Tuple, Union import anitopy -from app.runtime.config import settings +from app.application.configuration import get_chain_runtime_config_snapshot from app.domain.metainfo import MetaInfoPath from app.domain.meta.metabase import MetaBase from app.runtime.log import logger @@ -580,11 +580,12 @@ class EpisodeFormatRuleHelper: def _get_file_kind(item: FileItem) -> str: """按扩展名把样本归类为视频、字幕、音频或其他。""" extension = f".{(item.extension or '').lower().lstrip('.')}" if item.extension else "" - if extension in settings.RMT_MEDIAEXT: + config = get_chain_runtime_config_snapshot() + if extension in config.video_extensions: return "media" - if extension in settings.RMT_SUBEXT: + if extension in config.subtitle_extensions: return "subtitle" - if extension in settings.RMT_AUDIOEXT: + if extension in config.audio_extensions: return "audio" return "other" diff --git a/app/doctor/checks.py b/app/doctor/checks.py index 01f099e47..8fe531b4f 100644 --- a/app/doctor/checks.py +++ b/app/doctor/checks.py @@ -18,6 +18,8 @@ from urllib.request import Request, urlopen import psutil +from app.adapters.system.backup.database import verify_database_backup +from app.adapters.system.backup.files import BackupFiles from app.runtime.config import settings from app.runtime.topology import process_topology_issue from app.doctor.models import DoctorFinding, DoctorFindingStatus, DoctorReport, DoctorSeverity @@ -824,6 +826,116 @@ def _check_database(runner: DoctorRunnerProtocol) -> None: _check_postgresql_database(runner) else: _check_sqlite_database(runner) + _check_database_backups(runner) + + +def _check_database_backups(runner: DoctorRunnerProtocol) -> None: + """列举并离线校验与当前数据库类型匹配的受管备份。""" + db_type = "postgresql" if settings.DB_TYPE.lower() == "postgresql" else "sqlite" + try: + paths = BackupFiles(settings.DATABASE_BACKUP_PATH).list() + except OSError as error: + runner.add( + finding_id="database.backup_recovery", + severity=DoctorSeverity.Error, + status=DoctorFindingStatus.Failed, + title="数据库备份目录无法读取", + detail=str(error), + recommendation="检查数据库备份目录是否存在且当前用户具有读取权限。", + context={"db_type": db_type}, + ) + return + + matching = [path for path in paths if BackupFiles.database_type(path.name) == db_type] + mismatched = [path.name for path in paths if path not in matching] + if not paths: + runner.add( + finding_id="database.backup_recovery", + severity=DoctorSeverity.Info, + status=DoctorFindingStatus.Skipped, + title="未找到受管数据库备份", + detail=f"当前数据库类型为 {db_type}，备份目录中没有正式备份文件。", + recommendation="可执行 `moviepilot database backup` 创建一次可校验备份。", + affects_report_status=False, + context={"db_type": db_type, "backups": []}, + ) + return + + if not matching: + runner.add( + finding_id="database.backup_recovery", + severity=DoctorSeverity.Warn, + status=DoctorFindingStatus.Degraded, + title="没有匹配当前数据库类型的备份", + detail=f"当前数据库类型为 {db_type}，仅找到其他类型备份：{', '.join(mismatched)}。", + recommendation="确认数据库类型配置，或执行 `moviepilot database backup` 创建当前类型备份。", + affects_report_status=False, + context={"db_type": db_type, "backups": [], "mismatched": mismatched}, + ) + return + + backups = [] + for path in matching: + try: + size = path.stat().st_size + verification = verify_database_backup(path, db_type=db_type) + valid = verification.valid + method = verification.method + detail = verification.detail + except (OSError, RuntimeError, ValueError) as error: + size = None + valid = False + method = "unavailable" + detail = str(error) + backups.append( + { + "name": path.name, + "db_type": db_type, + "size": size, + "valid": valid, + "method": method, + "detail": detail, + } + ) + + valid_backups = [backup for backup in backups if backup["valid"]] + context = { + "db_type": db_type, + "backups": backups, + "mismatched": mismatched, + } + if valid_backups: + newest = valid_backups[0] + command = f"moviepilot database restore {newest['name']} --confirm" + runner.add( + finding_id="database.backup_recovery", + severity=DoctorSeverity.Info, + status=DoctorFindingStatus.Ok, + title="存在可还原的数据库备份", + detail=f"已校验 {len(backups)} 个 {db_type} 备份，其中 {len(valid_backups)} 个可用。", + recommendation=f"需要恢复时先停止 MoviePilot，再执行 `{command}`。", + context={**context, "restore_command": command}, + ) + return + + + failures = "; ".join( + f"{backup['name']}: {backup['detail'] or backup['method']}" + for backup in backups + ) + runner.add( + finding_id="database.backup_recovery", + severity=DoctorSeverity.Error, + status=DoctorFindingStatus.Failed, + title="数据库备份均未通过校验", + detail=( + f"已检查 {len(backups)} 个 {db_type} 备份，没有可直接还原的文件。" + f"校验结果：{failures}" + ), + recommendation="根据校验详情处理备份文件或 PostgreSQL client，再重新运行 doctor。", + affects_report_status=False, + context=context, + ) def _check_frontend_assets(runner: DoctorRunnerProtocol) -> None: diff --git a/docs/refactor/backend-architecture-next-stage.md b/docs/refactor/backend-architecture-next-stage.md index b64db5844..36669c089 100644 --- a/docs/refactor/backend-architecture-next-stage.md +++ b/docs/refactor/backend-architecture-next-stage.md @@ -989,19 +989,15 @@ Outbox adapter、DB 装饰器、Base 与 UoW，strict 清单扩大到 37 个源 **实施记录（2026-08-21）**： -- 新增 `scripts/architecture/async_blocking.py`，AST 扫描 `app/api`、`app/agent`、`app/application` 的 - async 函数，覆盖直接 `open`/Path 读写遍历、`requests`、`time.sleep`、同步 `subprocess` 和目录遍历。 -- scanner 通过局部类型流识别 `aiofiles` 与 `anyio.AsyncPath`，不会把正确异步 I/O 记成债务；baseline - 只允许调用减少/删除，新增或次数增长均在 CI architecture job 失败。 -- Web Agent 上传已从 `Path.open/write/unlink` 改为 `aiofiles` 写入和统一 `run_in_threadpool` 清理；当前仅保留 - ActivityLog 为保证 `O_EXCL` 原子创建使用的一处 `os.open` 精确债务，不泛化豁免整个文件或目录。 +- `scripts/architecture/async_blocking.py` 扫描 canonical 主程序目录和顶层运行入口中的 async 函数，覆盖 + 同步 HTTP、Oper、Path、`shutil`、`subprocess`、`os`、`time.sleep` 与 `open`。 +- scanner 按 import 来源、局部别名、互斥分支和嵌套函数定义点解析符号；`AsyncRequestUtils`、 + `anyio.Path`、延迟回调及受控 worker 内执行的同步函数不记为 async 直接阻塞。函数和 lambda 的默认值、 + decorator 等定义时表达式仍在所在 async 执行体中检查。 +- baseline 只允许调用减少或删除，新增调用及次数增长均使 CI architecture job 失败；当前记录 10 条已确认 + 存量，包括 8 条文件元数据访问、1 条目录删除和 1 条同步 Oper 读取，由后续数据库与文件 adapter 叶迁移。 - pytest 全局启用 `asyncio_debug`，专项测试验证实际 loop debug 状态；AST ratchet 与 46 个 Agent 流式回归 通过。同步第三方 Module 仍由 dispatcher 的 `app.runtime.execution.run_in_threadpool` 兼容。 -- 2026-08-22 扫描范围扩大到 `app/chain`、`app/modules`、`app/startup` 与 `app/scheduler.py`；扩大后未发现 - 新存量，仍只保留 ActivityLog 的一处原子 `os.open` 精确债务，并由测试锁定扫描根目录。 -- 扫描进一步覆盖 `adapters/db/doctor/domain/foundation/monitor/runtime/schemas/workflow` 及 CLI、Command、 - Factory、Main 顶层入口，明确排除插件源码和 SDK。AsyncPath 条件派生识别已修正；Release zip 解压读取和 - ActivityLog `O_EXCL` 独占创建移入线程池，扩围后 async 阻塞 baseline 从 1 降为 0。 ## 6. 推荐执行队列 diff --git a/scripts/architecture/async_blocking.py b/scripts/architecture/async_blocking.py index d66d5a8f1..7de5f5660 100644 --- a/scripts/architecture/async_blocking.py +++ b/scripts/architecture/async_blocking.py @@ -6,7 +6,8 @@ import argparse import ast import json from collections import Counter -from collections.abc import Iterator +from collections.abc import Iterator, Sequence +from dataclasses import dataclass from pathlib import Path PROJECT_ROOT = Path(__file__).resolve().parents[2] @@ -33,159 +34,545 @@ SCAN_ROOTS = ( "app/main.py", "app/scheduler.py", ) -BLOCKING_EXACT = { - "open", - "time.sleep", - "subprocess.call", - "subprocess.check_call", - "subprocess.check_output", - "subprocess.Popen", - "subprocess.run", - "os.listdir", - "os.scandir", - "os.walk", - "requests.delete", - "requests.get", - "requests.head", - "requests.patch", - "requests.post", - "requests.put", - "requests.request", + +_SYNC_HTTP_METHODS = { + "delete_res", + "get", + "get_json", + "get_res", + "get_stream", + "post", + "post_json", + "post_res", + "put", + "put_res", + "request", + "response_manager", } -BLOCKING_ATTRIBUTES = { +_REQUESTS_METHODS = { + "delete", + "get", + "head", + "patch", + "post", + "put", + "request", +} +_PATH_IO_METHODS = { + "exists", "glob", + "is_dir", + "is_file", "iterdir", + "mkdir", "open", "read_bytes", "read_text", + "rename", "rglob", + "stat", + "unlink", "write_bytes", "write_text", } +_SHUTIL_METHODS = { + "copy", + "copyfile", + "copytree", + "make_archive", + "move", + "rmtree", + "unpack_archive", + "which", +} +_SUBPROCESS_METHODS = { + "Popen", + "call", + "check_call", + "check_output", + "run", +} +_OS_IO_METHODS = {"listdir", "scandir", "walk"} +_SYSTEM_CONFIG_MEMORY_READS = { + "app.db.oper.SystemConfigOper.all", + "app.db.oper.SystemConfigOper.get", + "app.db.oper.systemconfig.SystemConfigOper.all", + "app.db.oper.systemconfig.SystemConfigOper.get", +} -def _call_name(node: ast.Call) -> str: - """把简单名称和属性调用还原为点分文本。""" - parts = [] - target: ast.expr = node.func - while isinstance(target, ast.Attribute): - parts.append(target.attr) - target = target.value - if isinstance(target, ast.Name): - parts.append(target.id) - return ".".join(reversed(parts)) +@dataclass(frozen=True) +class _Binding: + """描述由明确 import 或局部赋值建立的符号来源。""" + + family: str + qualified_name: str + kind: str = "module" -def _root_name(expression: ast.expr) -> str | None: - """返回属性调用最左侧的变量名。""" - while isinstance(expression, ast.Attribute): - expression = expression.value - return expression.id if isinstance(expression, ast.Name) else None +_UNKNOWN_BINDING = _Binding("unknown", "", "unknown") +_BLOCKING_FAMILIES = { + "os", + "requests", + "shutil", + "subprocess", + "sync_http", + "sync_oper", + "sync_path", + "time", +} -class _AsyncPathCollector(ast.NodeVisitor): - """用局部数据流识别 anyio AsyncPath 变量及其派生值。""" +def _binding_for_qualified(qualified_name: str) -> _Binding: + """按稳定模块路径识别门禁关心的符号族。""" + if qualified_name == "app.adapters.network.http.RequestUtils": + return _Binding("sync_http", qualified_name, "class") + if qualified_name == "app.adapters.network.http.AsyncRequestUtils": + return _Binding("async_http", qualified_name, "class") + if qualified_name in {"requests.Session", "requests.sessions.Session"}: + return _Binding("sync_http", qualified_name, "class") + if qualified_name == "pathlib.Path": + return _Binding("sync_path", qualified_name, "class") + if qualified_name == "anyio.Path": + return _Binding("async_path", qualified_name, "class") + if qualified_name.startswith("app.db.oper.") and qualified_name.endswith("Oper"): + return _Binding("sync_oper", qualified_name, "class") + if qualified_name.startswith("requests."): + return _Binding("requests", qualified_name, "callable") + if qualified_name.startswith("shutil."): + return _Binding("shutil", qualified_name, "callable") + if qualified_name.startswith("subprocess."): + return _Binding("subprocess", qualified_name, "callable") + if qualified_name.startswith("os."): + return _Binding("os", qualified_name, "callable") + if qualified_name.startswith("time."): + return _Binding("time", qualified_name, "callable") + return _Binding("module", qualified_name) - def __init__(self, function: ast.AsyncFunctionDef) -> None: - """从参数注解初始化 AsyncPath 变量集合。""" - self.paths = { - argument.arg - for argument in (*function.args.posonlyargs, *function.args.args) - if argument.annotation and "AsyncPath" in ast.unparse(argument.annotation) - } - self.path_collections: set[str] = set() + +def _resolve_binding( + expression: ast.expr, + bindings: dict[str, _Binding], +) -> _Binding | None: + """解析明确 import、构造、属性访问和简单路径派生。""" + if isinstance(expression, ast.Name): + return bindings.get(expression.id) + if isinstance(expression, ast.Attribute): + base = _resolve_binding(expression.value, bindings) + if not base or base.kind in {"collection", "unknown"}: + return None + qualified_name = f"{base.qualified_name}.{expression.attr}" + if base.kind == "module": + return _binding_for_qualified(qualified_name) + return _Binding(base.family, qualified_name, "callable") + if isinstance(expression, ast.Call): + target = _resolve_binding(expression.func, bindings) + if target and target.kind == "class": + return _Binding(target.family, target.qualified_name, "instance") + return None + if isinstance(expression, ast.BinOp) and isinstance(expression.op, ast.Div): + left = _resolve_binding(expression.left, bindings) + if left and left.family in {"async_path", "sync_path"}: + return left + if isinstance(expression, ast.IfExp): + return _merge_binding_options( + "", + ( + _resolve_binding(expression.body, bindings), + _resolve_binding(expression.orelse, bindings), + ), + ) + return None + + +def _merge_binding_options( + name: str, + options: Sequence[_Binding | None], +) -> _Binding | None: + """控制流合流时保留任一分支可能进入的同步阻塞类型。""" + if options and all(option == options[0] for option in options): + return options[0] + blocking = sorted( + ( + option + for option in options + if option and option.family in _BLOCKING_FAMILIES + ), + key=lambda item: (item.family, item.qualified_name, item.kind), + ) + if blocking: + return blocking[0] + if name == "open" and any(option is None for option in options): + return None + return _UNKNOWN_BINDING + + +def _merge_binding_states( + states: Sequence[dict[str, _Binding]], +) -> dict[str, _Binding]: + """合并互斥控制流的局部符号表。""" + names = set().union(*(state.keys() for state in states)) + merged: dict[str, _Binding] = {} + for name in names: + binding = _merge_binding_options(name, tuple(state.get(name) for state in states)) + if binding: + merged[name] = binding + return merged + + +def _load_oper_methods(root: Path) -> dict[str, set[str]]: + """从 Oper 源码建立同步方法索引，避免按方法名猜测数据库调用。""" + methods: dict[str, set[str]] = {} + oper_root = root / "app/db/oper" + if not oper_root.is_dir(): + return methods + for path in sorted(oper_root.glob("*.py")): + tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + module = ".".join(path.relative_to(root).with_suffix("").parts) + for node in tree.body: + if not isinstance(node, ast.ClassDef) or not node.name.endswith("Oper"): + continue + qualified_name = f"{module}.{node.name}" + sync_methods = { + child.name + for child in node.body + if isinstance(child, ast.FunctionDef) + and not child.name.startswith("__") + } + methods[qualified_name] = sync_methods + methods[f"app.db.oper.{node.name}"] = sync_methods + return methods + + +class _ImportCollector(ast.NodeVisitor): + """收集模块级 import，作为每个 async 函数的初始符号表。""" + + def __init__(self) -> None: + self.bindings: dict[str, _Binding] = {} + + def visit_Import(self, node: ast.Import) -> None: + for alias in node.names: + local_name = alias.asname or alias.name.split(".", 1)[0] + qualified_name = alias.name if alias.asname else local_name + self.bindings[local_name] = _binding_for_qualified(qualified_name) + + def visit_ImportFrom(self, node: ast.ImportFrom) -> None: + if node.level or not node.module: + return + for alias in node.names: + if alias.name == "*": + continue + local_name = alias.asname or alias.name + qualified_name = f"{node.module}.{alias.name}" + self.bindings[local_name] = _binding_for_qualified(qualified_name) + + def _bind_target(self, target: ast.expr, binding: _Binding | None) -> None: + if isinstance(target, ast.Name): + self.bindings[target.id] = binding or _UNKNOWN_BINDING + elif isinstance(target, (ast.Tuple, ast.List)): + for element in target.elts: + self._bind_target(element, None) def visit_Assign(self, node: ast.Assign) -> None: - """识别 AsyncPath 构造和已知路径的 `/` 派生赋值。""" - is_async_path = self._is_async_path_expression(node.value) - if is_async_path: - for target in node.targets: - if isinstance(target, ast.Name): - self.paths.add(target.id) - self.generic_visit(node) - - def _is_async_path_expression(self, expression: ast.expr) -> bool: - """识别 AsyncPath 构造及任意层级的 `/` 路径派生表达式。""" - if isinstance(expression, ast.Call): - return _call_name(expression).endswith("AsyncPath") - if isinstance(expression, ast.Name): - return expression.id in self.paths - if isinstance(expression, ast.BinOp): - return self._is_async_path_expression(expression.left) - if isinstance(expression, ast.IfExp): - return ( - self._is_async_path_expression(expression.body) - and self._is_async_path_expression(expression.orelse) - ) - return False + binding = _resolve_binding(node.value, self.bindings) + for target in node.targets: + self._bind_target(target, binding) def visit_AnnAssign(self, node: ast.AnnAssign) -> None: - """识别 `list[AsyncPath]` 等路径集合。""" - if isinstance(node.target, ast.Name): - annotation = ast.unparse(node.annotation) - if "AsyncPath" in annotation: - if "list" in annotation or "List" in annotation: - self.path_collections.add(node.target.id) - else: - self.paths.add(node.target.id) - self.generic_visit(node) - - def visit_AsyncFor(self, node: ast.AsyncFor) -> None: - """AsyncPath.iterdir 产出的元素仍是 AsyncPath。""" - if isinstance(node.target, ast.Name) and isinstance(node.iter, ast.Call): - receiver = _root_name(node.iter.func) - if receiver in self.paths: - self.paths.add(node.target.id) - self.generic_visit(node) - - def visit_For(self, node: ast.For) -> None: - """从 `list[AsyncPath]` 迭代得到的元素仍是 AsyncPath。""" - if ( - isinstance(node.target, ast.Name) - and isinstance(node.iter, ast.Name) - and node.iter.id in self.path_collections - ): - self.paths.add(node.target.id) - self.generic_visit(node) + binding = _resolve_binding(node.value, self.bindings) if node.value else None + self._bind_target(node.target, binding) def visit_FunctionDef(self, node: ast.FunctionDef) -> None: - """不分析嵌套同步函数。""" + return def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None: - """不分析嵌套异步函数。""" + return + + def visit_ClassDef(self, node: ast.ClassDef) -> None: + return class _AsyncCallVisitor(ast.NodeVisitor): - """只收集一个 async 函数本体中的阻塞调用，不进入嵌套函数定义。""" + """用局部符号传播识别一个 async 函数直接执行的阻塞调用。""" - def __init__(self, async_paths: set[str]) -> None: - """初始化违规计数器和异步文件对象白名单。""" + def __init__( + self, + function: ast.FunctionDef | ast.AsyncFunctionDef, + module_bindings: dict[str, _Binding], + oper_methods: dict[str, set[str]], + *, + record_calls: bool, + ) -> None: self.calls: Counter[str] = Counter() - self._async_paths = async_paths + self.nested_functions: list[ + tuple[ast.FunctionDef | ast.AsyncFunctionDef, dict[str, _Binding]] + ] = [] + self._bindings = dict(module_bindings) + self._oper_methods = oper_methods + self._record_calls = record_calls + self._bind_arguments(function) + + def _bind_arguments( + self, + function: ast.FunctionDef | ast.AsyncFunctionDef, + ) -> None: + arguments = ( + *function.args.posonlyargs, + *function.args.args, + *function.args.kwonlyargs, + ) + for argument in arguments: + binding = self._annotation_binding(argument.annotation) + if binding: + self._bindings[argument.arg] = binding + else: + self._bindings[argument.arg] = _UNKNOWN_BINDING + + def _annotation_binding(self, annotation: ast.expr | None) -> _Binding | None: + if annotation is None: + return None + if isinstance(annotation, ast.Subscript): + container = ast.unparse(annotation.value).rsplit(".", 1)[-1] + elements = ( + annotation.slice.elts + if isinstance(annotation.slice, ast.Tuple) + else (annotation.slice,) + ) + for element in elements: + binding = self._annotation_binding(element) + if not binding: + continue + if container in {"list", "List", "Sequence", "set", "tuple"}: + return _Binding( + binding.family, + binding.qualified_name, + "collection", + ) + if container in {"Annotated", "Optional", "Union"}: + return binding + return None + if isinstance(annotation, ast.BinOp) and isinstance(annotation.op, ast.BitOr): + return self._annotation_binding(annotation.left) or self._annotation_binding( + annotation.right + ) + binding = _resolve_binding(annotation, self._bindings) + if binding and binding.kind == "class": + return _Binding(binding.family, binding.qualified_name, "instance") + return None + + def _resolve(self, expression: ast.expr) -> _Binding | None: + return _resolve_binding(expression, self._bindings) + + @staticmethod + def _call_label(binding: _Binding) -> str: + parts = binding.qualified_name.split(".") + if binding.family == "sync_oper" and len(parts) >= 2: + return ".".join(parts[-2:]) + if binding.family in {"async_path", "sync_path"} and parts: + return f"Path.{parts[-1]}" + if binding.family == "sync_http" and len(parts) >= 2: + return ".".join(parts[-2:]) + if binding.family in {"os", "requests", "shutil", "subprocess", "time"}: + return ".".join(parts[-2:]) + return binding.qualified_name + + def _record_call(self, node: ast.Call, binding: _Binding | None) -> None: + if not self._record_calls: + return + if not binding: + if isinstance(node.func, ast.Name) and node.func.id == "open": + self.calls["open"] += 1 + return + method = binding.qualified_name.rsplit(".", 1)[-1] + blocked = False + if binding.family == "sync_http": + blocked = binding.kind == "callable" and method in _SYNC_HTTP_METHODS + elif binding.family == "requests": + blocked = method in _REQUESTS_METHODS + elif binding.family == "sync_path": + blocked = binding.kind == "callable" and method in _PATH_IO_METHODS + elif binding.family == "shutil": + blocked = method in _SHUTIL_METHODS + elif binding.family == "subprocess": + blocked = method in _SUBPROCESS_METHODS + elif binding.family == "os": + blocked = method in _OS_IO_METHODS + elif binding.family == "time": + blocked = method == "sleep" + elif binding.family == "sync_oper" and binding.kind == "callable": + class_name, method_name = binding.qualified_name.rsplit(".", 1) + blocked = ( + method_name in self._oper_methods.get(class_name, set()) + and binding.qualified_name not in _SYSTEM_CONFIG_MEMORY_READS + ) + if blocked: + self.calls[self._call_label(binding)] += 1 + + def _bind_target(self, target: ast.expr, binding: _Binding | None) -> None: + if isinstance(target, ast.Name): + self._bindings[target.id] = binding or _UNKNOWN_BINDING + elif isinstance(target, (ast.Tuple, ast.List)): + for element in target.elts: + self._bind_target(element, None) + + def visit_Import(self, node: ast.Import) -> None: + collector = _ImportCollector() + collector.visit_Import(node) + self._bindings.update(collector.bindings) + + def visit_ImportFrom(self, node: ast.ImportFrom) -> None: + collector = _ImportCollector() + collector.visit_ImportFrom(node) + self._bindings.update(collector.bindings) + + def visit_Assign(self, node: ast.Assign) -> None: + self.visit(node.value) + binding = self._resolve(node.value) + for target in node.targets: + self._bind_target(target, binding) + + def visit_AnnAssign(self, node: ast.AnnAssign) -> None: + if node.value: + self.visit(node.value) + binding = self._resolve(node.value) if node.value else None + binding = binding or self._annotation_binding(node.annotation) + self._bind_target(node.target, binding) + + def visit_AugAssign(self, node: ast.AugAssign) -> None: + self.visit(node.value) + self._bind_target(node.target, None) + + def _visit_branch( + self, + statements: Sequence[ast.stmt], + initial: dict[str, _Binding], + ) -> dict[str, _Binding]: + saved = self._bindings + self._bindings = dict(initial) + for statement in statements: + self.visit(statement) + result = self._bindings + self._bindings = saved + return result + + def visit_If(self, node: ast.If) -> None: + self.visit(node.test) + initial = dict(self._bindings) + body = self._visit_branch(node.body, initial) + alternate = self._visit_branch(node.orelse, initial) if node.orelse else initial + self._bindings = _merge_binding_states((body, alternate)) + + def visit_For(self, node: ast.For | ast.AsyncFor) -> None: + self.visit(node.iter) + initial = dict(self._bindings) + collection = self._resolve(node.iter) + element = ( + _Binding(collection.family, collection.qualified_name, "instance") + if collection and collection.kind == "collection" + else None + ) + self._bind_target(node.target, element) + for statement in node.body: + self.visit(statement) + iterated = dict(self._bindings) + completed = self._visit_branch(node.orelse, iterated) + self._bindings = _merge_binding_states((initial, completed)) + + def visit_AsyncFor(self, node: ast.AsyncFor) -> None: + self.visit_For(node) + + def _visit_comprehension( + self, + generators: Sequence[ast.comprehension], + outputs: Sequence[ast.expr], + ) -> None: + saved = self._bindings + self._bindings = dict(saved) + for generator in generators: + self.visit(generator.iter) + collection = self._resolve(generator.iter) + element = ( + _Binding(collection.family, collection.qualified_name, "instance") + if collection and collection.kind == "collection" + else None + ) + self._bind_target(generator.target, element) + for condition in generator.ifs: + self.visit(condition) + for output in outputs: + self.visit(output) + self._bindings = saved + + def visit_ListComp(self, node: ast.ListComp) -> None: + self._visit_comprehension(node.generators, (node.elt,)) + + def visit_SetComp(self, node: ast.SetComp) -> None: + self._visit_comprehension(node.generators, (node.elt,)) + + def visit_GeneratorExp(self, node: ast.GeneratorExp) -> None: + self._visit_comprehension(node.generators, (node.elt,)) + + def visit_DictComp(self, node: ast.DictComp) -> None: + self._visit_comprehension(node.generators, (node.key, node.value)) + + def visit_NamedExpr(self, node: ast.NamedExpr) -> None: + self.visit(node.value) + self._bind_target(node.target, self._resolve(node.value)) def visit_Call(self, node: ast.Call) -> None: - """记录命中阻塞名单的调用并继续遍历参数表达式。""" - name = _call_name(node) - attribute = name.rsplit(".", 1)[-1] - receiver = _root_name(node.func) - async_safe = name.startswith("aiofiles.") or receiver in self._async_paths - if not async_safe and ( - name in BLOCKING_EXACT or attribute in BLOCKING_ATTRIBUTES - ): - self.calls[name or attribute] += 1 + binding = self._resolve(node.func) + self._record_call(node, binding) + if isinstance(node.func, ast.Lambda): + self._visit_lambda_defaults(node.func) + for argument in node.args: + self.visit(argument) + for keyword in node.keywords: + self.visit(keyword.value) + self.visit(node.func.body) + return self.generic_visit(node) + def _visit_definition_expressions( + self, + node: ast.FunctionDef | ast.AsyncFunctionDef, + ) -> None: + for decorator in node.decorator_list: + self.visit(decorator) + for default in node.args.defaults: + self.visit(default) + for keyword_default in node.args.kw_defaults: + if keyword_default: + self.visit(keyword_default) + + def _visit_nested_function( + self, + node: ast.FunctionDef | ast.AsyncFunctionDef, + ) -> None: + self._visit_definition_expressions(node) + self.nested_functions.append((node, dict(self._bindings))) + self._bindings[node.name] = _UNKNOWN_BINDING + def visit_FunctionDef(self, node: ast.FunctionDef) -> None: - """嵌套同步函数不属于外层 async 的直接执行体。""" + self._visit_nested_function(node) def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None: - """嵌套异步函数由模块级收集器单独治理。""" + self._visit_nested_function(node) + + def _visit_lambda_defaults(self, node: ast.Lambda) -> None: + for default in node.args.defaults: + self.visit(default) + for keyword_default in node.args.kw_defaults: + if keyword_default: + self.visit(keyword_default) + + def visit_Lambda(self, node: ast.Lambda) -> None: + self._visit_lambda_defaults(node) -def _async_functions( +def _root_functions( tree: ast.Module, ) -> Iterator[tuple[str, ast.AsyncFunctionDef]]: - """产出模块顶层及类直接拥有的 async 函数限定名。""" + """产出模块顶层和类直接拥有的 async 入口。""" for node in tree.body: if isinstance(node, ast.AsyncFunctionDef): yield node.name, node @@ -195,24 +582,63 @@ def _async_functions( yield f"{node.name}.{method.name}", method -def collect_async_blocking(root: Path = PROJECT_ROOT) -> dict[str, int]: +def _scan_paths(root: Path, scan_roots: Sequence[str | Path]) -> Iterator[Path]: + """按稳定顺序产出存在的扫描目标。""" + for scan_root in scan_roots: + target = root / scan_root + if target.is_file(): + yield target + elif target.is_dir(): + yield from sorted(target.rglob("*.py")) + + +def collect_async_blocking( + root: Path = PROJECT_ROOT, + scan_roots: Sequence[str | Path] = SCAN_ROOTS, +) -> dict[str, int]: """扫描关键目录并以文件、函数、调用名聚合存量次数。""" debt: Counter[str] = Counter() - for scan_root in SCAN_ROOTS: - target = root / scan_root - paths = [target] if target.is_file() else sorted(target.rglob("*.py")) - for path in paths: - tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) - relative = path.relative_to(root).as_posix() - for qualname, function in _async_functions(tree): - collector = _AsyncPathCollector(function) - for statement in function.body: - collector.visit(statement) - visitor = _AsyncCallVisitor(collector.paths) - for statement in function.body: - visitor.visit(statement) + oper_methods = _load_oper_methods(root) + for path in _scan_paths(root, scan_roots): + tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + imports = _ImportCollector() + for statement in tree.body: + imports.visit(statement) + relative = path.relative_to(root).as_posix() + pending: list[ + tuple[ + str, + ast.FunctionDef | ast.AsyncFunctionDef, + dict[str, _Binding], + ] + ] = [ + (qualname, function, imports.bindings) + for qualname, function in _root_functions(tree) + ] + pending_index = 0 + while pending_index < len(pending): + qualname, function, lexical_bindings = pending[pending_index] + pending_index += 1 + is_async = isinstance(function, ast.AsyncFunctionDef) + visitor = _AsyncCallVisitor( + function, + lexical_bindings, + oper_methods, + record_calls=is_async, + ) + for statement in function.body: + visitor.visit(statement) + if is_async: for call_name, count in visitor.calls.items(): debt[f"{relative}:{qualname}:{call_name}"] += count + pending.extend( + ( + f"{qualname}.{nested.name}", + nested, + bindings, + ) + for nested, bindings in visitor.nested_functions + ) return dict(sorted(debt.items())) diff --git a/tests/fixtures/architecture/async-blocking-baseline.json b/tests/fixtures/architecture/async-blocking-baseline.json index 0967ef424..22e985d53 100644 --- a/tests/fixtures/architecture/async-blocking-baseline.json +++ b/tests/fixtures/architecture/async-blocking-baseline.json @@ -1 +1,12 @@ -{} +{ + "app/agent/tools/impl/_plugin_tool_utils.py:uninstall_plugin_runtime:shutil.rmtree": 1, + "app/agent/tools/impl/scrape_metadata.py:ScrapeMetadataTool.run:Path.exists": 1, + "app/agent/tools/impl/scrape_metadata.py:ScrapeMetadataTool.run:Path.is_dir": 1, + "app/chain/media.py:MediaChain._async_music_album_dir_fallback:Path.exists": 1, + "app/chain/media.py:MediaChain._async_music_album_dir_fallback:Path.is_file": 1, + "app/chain/media.py:MediaChain.async_recognize_music_album_directory:Path.is_dir": 1, + "app/modules/acoustid/__init__.py:AcoustIdModule.async_identify_music_by_fingerprint:Path.is_file": 1, + "app/modules/discord/discord.py:Discord._send_file:Path.exists": 1, + "app/modules/discord/discord.py:Discord._send_file:Path.is_file": 1, + "app/scheduler.py:Scheduler.execute_agent_task:AgentTaskOper.get": 1 +} diff --git a/tests/fixtures/architecture/configuration-debt-baseline.json b/tests/fixtures/architecture/configuration-debt-baseline.json index c3b8f9d2e..e424599d4 100644 --- a/tests/fixtures/architecture/configuration-debt-baseline.json +++ b/tests/fixtures/architecture/configuration-debt-baseline.json @@ -9,7 +9,7 @@ "root": "app" }, "settings_imports": { - "count": 136, + "count": 135, "files": [ "app/adapters/cache/backends.py", "app/adapters/cache/redis.py", @@ -47,7 +47,6 @@ "app/agent/tools/impl/send_voice_message.py", "app/agent/tools/impl/update_agent_task.py", "app/agent/tools/impl/update_system_settings.py", - "app/application/formatting.py", "app/application/maintenance.py", "app/application/rss.py", "app/application/security/auth.py", diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index aa37d2b01..8fa59fb8e 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -13,8 +13,8 @@ "runtime_to_db": [], "workflow_to_db": [] }, - "edge_count": 6393, - "edge_sha256": "ea375f19071a37a9c72bd01ce9fa4a070e64ad7f14a87d2e5013b5c7440680cd", + "edge_count": 6397, + "edge_sha256": "801b80e33c626aecf70e5320a2c1c81db401549faa652e718be7386bac593f43", "edges": [ "app -> app.runtime", "app -> app.runtime.compat", @@ -2494,12 +2494,13 @@ "app.application.downloader -> app.schemas", "app.application.downloader -> app.schemas.system", "app.application.downloader -> app.schemas.types", + "app.application.formatting -> app.application", + "app.application.formatting -> app.application.configuration", "app.application.formatting -> app.domain", "app.application.formatting -> app.domain.meta", "app.application.formatting -> app.domain.meta.metabase", "app.application.formatting -> app.domain.metainfo", "app.application.formatting -> app.runtime", - "app.application.formatting -> app.runtime.config", "app.application.formatting -> app.runtime.log", "app.application.formatting -> app.schemas", "app.application.formatting -> app.schemas.transfer", @@ -3663,6 +3664,9 @@ "app.db.session -> app.runtime.observability", "app.doctor.checks -> app.adapters", "app.doctor.checks -> app.adapters.system", + "app.doctor.checks -> app.adapters.system.backup", + "app.doctor.checks -> app.adapters.system.backup.database", + "app.doctor.checks -> app.adapters.system.backup.files", "app.doctor.checks -> app.adapters.system.host", "app.doctor.checks -> app.doctor", "app.doctor.checks -> app.doctor.models", diff --git a/tests/test_async_blocking_gate.py b/tests/test_async_blocking_gate.py index 2a1d468ef..bfbc10a46 100644 --- a/tests/test_async_blocking_gate.py +++ b/tests/test_async_blocking_gate.py @@ -1,10 +1,33 @@ """async 阻塞调用 ratchet 与 debug 模式测试。""" import asyncio +import textwrap +from pathlib import Path import pytest -from scripts.architecture.async_blocking import SCAN_ROOTS, compare_async_blocking +from scripts.architecture.async_blocking import ( + SCAN_ROOTS, + collect_async_blocking, + compare_async_blocking, +) + + +def _scan_source( + tmp_path: Path, + source: str, + *, + oper_sources: dict[str, str] | None = None, +) -> dict[str, int]: + """构造最小仓库并通过公开扫描入口验证源码，而非测试内部 AST 细节。""" + api_root = tmp_path / "app/api" + api_root.mkdir(parents=True) + (api_root / "sample.py").write_text(textwrap.dedent(source), encoding="utf-8") + for filename, oper_source in (oper_sources or {}).items(): + oper_path = tmp_path / "app/db/oper" / filename + oper_path.parent.mkdir(parents=True, exist_ok=True) + oper_path.write_text(textwrap.dedent(oper_source), encoding="utf-8") + return collect_async_blocking(tmp_path, scan_roots=("app/api",)) def test_async_blocking_scan_covers_runtime_entrypoints() -> None: @@ -51,6 +74,276 @@ def test_async_blocking_ratchet_allows_removal_and_rejects_growth() -> None: assert any("新增" in problem for problem in problems) +@pytest.mark.parametrize( + ("source", "expected"), + [ + ( + """ + from app.adapters.network.http import RequestUtils as RU + + async def load(): + client = RU() + alias = client + return alias.get_res("https://example.com") + """, + {"app/api/sample.py:load:RequestUtils.get_res": 1}, + ), + ( + """ + import app.adapters.network.http as http + + async def submit(): + return http.RequestUtils().post_res("https://example.com") + """, + {"app/api/sample.py:submit:RequestUtils.post_res": 1}, + ), + ( + """ + from pathlib import Path as P + + async def inspect_file(): + path = P("payload") / "item.json" + return path.exists() + """, + {"app/api/sample.py:inspect_file:Path.exists": 1}, + ), + ( + """ + import shutil as files + from subprocess import run as run_process + + async def cleanup(): + files.rmtree("payload") + run_process(["true"]) + """, + { + "app/api/sample.py:cleanup:shutil.rmtree": 1, + "app/api/sample.py:cleanup:subprocess.run": 1, + }, + ), + ( + """ + import os as operating_system + import time as clock + from requests import Session as HttpSession + + async def legacy_io(): + open("payload") + operating_system.listdir(".") + clock.sleep(0.1) + HttpSession().get("https://example.com") + """, + { + "app/api/sample.py:legacy_io:Session.get": 1, + "app/api/sample.py:legacy_io:open": 1, + "app/api/sample.py:legacy_io:os.listdir": 1, + "app/api/sample.py:legacy_io:time.sleep": 1, + }, + ), + ( + """ + from pathlib import Path + + async def outer(): + async def inner(): + return Path("payload").read_text() + return await inner() + """, + {"app/api/sample.py:outer.inner:Path.read_text": 1}, + ), + ( + """ + from pathlib import Path + + async def inspect_files(files: list[Path]): + for file in files: + if file.is_file(): + return file + return None + """, + {"app/api/sample.py:inspect_files:Path.is_file": 1}, + ), + ], +) +def test_async_blocking_scan_resolves_imports_aliases_and_nested_async( + tmp_path: Path, + source: str, + expected: dict[str, int], +) -> None: + """别名、简单局部传播和嵌套 async 都必须进入真实扫描结果。""" + assert _scan_source(tmp_path, source) == expected + + +def test_async_blocking_scan_uses_oper_source_method_kinds(tmp_path: Path) -> None: + """Oper 仅按源码中真实同步方法报告，不依赖方法名前缀猜测。""" + source = """ + from app.db.oper import SiteOper + from app.db.oper.site import SiteOper as SO + + async def load(oper: SO): + oper.list() + return await oper.get_by_id(1) + + async def load_from_facade(): + return SiteOper().list() + """ + oper_sources = { + "site.py": """ + class SiteOper: + def list(self): + return [] + + async def get_by_id(self, site_id): + return None + """, + } + + assert _scan_source(tmp_path, source, oper_sources=oper_sources) == { + "app/api/sample.py:load:SiteOper.list": 1, + "app/api/sample.py:load_from_facade:SiteOper.list": 1, + } + + +def test_async_blocking_scan_exempts_async_apis_and_memory_reads( + tmp_path: Path, +) -> None: + """异步实现、受控 worker 与内存配置读取不得成为阻塞债务。""" + source = """ + import asyncio + import anyio + import subprocess + from anyio import Path as AsyncPath + from app.agent.tools.base import run_agent_blocking + from app.adapters.network.http import AsyncRequestUtils + from app.db.oper.systemconfig import SystemConfigOper + from app.runtime.execution import run_in_threadpool as runtime_worker + from fastapi.concurrency import run_in_threadpool + + async def load(config: SystemConfigOper): + await AsyncRequestUtils().get_res("https://example.com") + await AsyncPath("payload").exists() + await run_in_threadpool(lambda: subprocess.run(["true"])) + await runtime_worker(lambda: subprocess.run(["true"])) + await run_agent_blocking("plugin", lambda: subprocess.run(["true"])) + await asyncio.to_thread(lambda: subprocess.run(["true"])) + await anyio.to_thread.run_sync(lambda: subprocess.run(["true"])) + deferred = lambda: subprocess.run(["true"]) + assert deferred + return config.get("key") + + def ordinary_sync(): + subprocess.run(["true"]) + """ + oper_sources = { + "systemconfig.py": """ + class SystemConfigOper: + def get(self, key): + return key + """, + } + + assert _scan_source(tmp_path, source, oper_sources=oper_sources) == {} + + +def test_async_blocking_scan_checks_worker_arguments_evaluated_on_loop( + tmp_path: Path, +) -> None: + """worker 调用前同步求值的普通参数仍在事件循环中执行。""" + source = """ + import asyncio + from pathlib import Path + + async def load(): + await asyncio.to_thread(print, Path("payload").read_text()) + """ + + assert _scan_source(tmp_path, source) == { + "app/api/sample.py:load:Path.read_text": 1, + } + + +def test_async_blocking_scan_merges_branch_bindings_conservatively( + tmp_path: Path, +) -> None: + """任一互斥分支可能产生同步对象时，合流调用仍属于阻塞风险。""" + source = """ + from anyio import Path as AsyncPath + from pathlib import Path + + async def load(use_sync: bool): + if use_sync: + target = Path("payload") + else: + target = AsyncPath("payload") + return target.read_text() + """ + + assert _scan_source(tmp_path, source) == { + "app/api/sample.py:load:Path.read_text": 1, + } + + +def test_nested_async_inherits_bindings_from_definition_scope(tmp_path: Path) -> None: + """嵌套 async 使用定义点已有的局部 import 和别名。""" + source = """ + async def outer(): + from pathlib import Path as LocalPath + alias = LocalPath + + async def inner(): + return alias("payload").read_text() + + return await inner() + """ + + assert _scan_source(tmp_path, source) == { + "app/api/sample.py:outer.inner:Path.read_text": 1, + } + + +def test_definition_time_expressions_remain_in_async_execution_body( + tmp_path: Path, +) -> None: + """延迟函数体不扫描，但默认值和 decorator 在定义时立即求值。""" + source = """ + import asyncio + from pathlib import Path + + def register(value): + return lambda function: function + + async def outer(): + await asyncio.to_thread( + lambda value=Path("lambda").read_text(): value + ) + + @register(Path("decorator").read_text()) + async def inner(value=Path("default").read_text()): + return value + + return await inner() + """ + + assert _scan_source(tmp_path, source) == { + "app/api/sample.py:outer:Path.read_text": 3, + } + + +def test_local_shadowing_does_not_reuse_import_or_builtin_bindings( + tmp_path: Path, +) -> None: + """参数和 comprehension target 会遮蔽同名 builtin 或导入符号。""" + source = """ + from pathlib import Path + + async def invoke(open, items): + open() + return [Path.exists() for Path in items] + """ + + assert _scan_source(tmp_path, source) == {} + + @pytest.mark.asyncio async def test_asyncio_debug_is_enabled_for_async_tests() -> None: """专项异步测试必须启用慢 callback 和阻塞诊断所需的 debug 模式。""" diff --git a/tests/test_database_backup_adapters.py b/tests/test_database_backup_adapters.py index 9e2ba3b73..517db325c 100644 --- a/tests/test_database_backup_adapters.py +++ b/tests/test_database_backup_adapters.py @@ -6,12 +6,14 @@ from pathlib import Path from types import SimpleNamespace from unittest.mock import Mock +import pytest from sqlalchemy import create_engine from sqlalchemy.engine import make_url from app.adapters.system.backup.database import ( PostgreSQLBackupBackend, SQLiteBackupBackend, + verify_database_backup, ) @@ -34,6 +36,15 @@ def test_sqlite_backup_includes_committed_wal_data(tmp_path: Path) -> None: engine.dispose() +def test_sqlite_backup_can_be_verified_without_active_engine(tmp_path: Path) -> None: + """Doctor 等离线入口不应为校验备份而构造活动数据库引擎。""" + artifact = tmp_path / "backup.db" + with sqlite3.connect(artifact) as connection: + connection.execute("CREATE TABLE entries (value TEXT NOT NULL)") + + assert verify_database_backup(artifact, db_type="sqlite").valid is True + + def test_sqlite_restore_replaces_database_and_removes_old_wal_files(tmp_path: Path) -> None: source = tmp_path / "user.db" backup = tmp_path / "backup.db" @@ -66,6 +77,11 @@ class _Runner: return subprocess.CompletedProcess(command, 0, stdout, "") +class _FailedRunner: + def __call__(self, command, **_kwargs): + return subprocess.CompletedProcess(command, 1, "", "invalid archive") + + def _postgres_backend(runner: _Runner) -> PostgreSQLBackupBackend: engine = SimpleNamespace( url=make_url( @@ -114,6 +130,55 @@ def test_postgresql_verify_and_restore_use_pg_restore(tmp_path: Path) -> None: assert restore_kwargs["env"]["PGPASSWORD"] == "secret" +def test_postgresql_backup_can_be_verified_without_active_engine(tmp_path: Path) -> None: + """PostgreSQL 归档校验只依赖 pg_restore，不连接活动数据库。""" + runner = _Runner() + artifact = tmp_path / "backup.dump" + artifact.write_bytes(b"PGDMP") + + result = verify_database_backup( + artifact, + db_type="postgresql", + runner=runner, + tool_resolver=lambda executable: executable, + ) + + assert result.valid is True + command, kwargs = runner.calls[0] + assert command == ["pg_restore", "--list", str(artifact)] + assert "PGPASSWORD" not in kwargs["env"] + assert "PGSSLMODE" not in kwargs["env"] + + +def test_postgresql_offline_verify_rejects_invalid_archive(tmp_path: Path) -> None: + """pg_restore 无法读取归档目录时备份必须判定为无效。""" + artifact = tmp_path / "backup.dump" + artifact.write_bytes(b"invalid") + + result = verify_database_backup( + artifact, + db_type="postgresql", + runner=_FailedRunner(), + tool_resolver=lambda executable: executable, + ) + + assert result.valid is False + assert result.detail == "pg_restore 退出码 1" + + +def test_postgresql_offline_verify_reports_missing_client(tmp_path: Path) -> None: + """缺少 pg_restore 时离线校验应给出可执行的安装提示。""" + artifact = tmp_path / "backup.dump" + artifact.write_bytes(b"PGDMP") + + with pytest.raises(RuntimeError, match="PostgreSQL client"): + verify_database_backup( + artifact, + db_type="postgresql", + tool_resolver=lambda _executable: None, + ) + + def test_postgresql_source_install_reports_missing_native_client() -> None: runner = _Runner() engine = SimpleNamespace( diff --git a/tests/test_doctor.py b/tests/test_doctor.py index e313f57cf..043e11b0e 100644 --- a/tests/test_doctor.py +++ b/tests/test_doctor.py @@ -1,13 +1,14 @@ from __future__ import annotations from datetime import datetime, timedelta +import sqlite3 from types import SimpleNamespace from app.runtime.config import settings -from app.doctor import checks, run_doctor +from app.doctor import checks from app.doctor.formatters import format_json_report, format_text_report from app.doctor.models import DoctorFinding, DoctorFindingStatus, DoctorSeverity -from app.doctor.runner import DoctorRunner +from app.doctor.runner import DoctorRunner, run_doctor def _current_log_timestamp() -> str: @@ -33,6 +34,98 @@ def test_doctor_report_has_stable_json_shape(tmp_path, monkeypatch): assert any(item["id"] == "runtime.paths" for item in payload["findings"]) +def test_doctor_reports_valid_backup_when_sqlite_database_is_corrupt( + tmp_path, + monkeypatch, +): + """主数据库损坏时 Doctor 仍应离线校验备份并给出还原命令。""" + monkeypatch.setattr(settings, "CONFIG_DIR", str(tmp_path)) + (tmp_path / "user.db").write_bytes(b"not a sqlite database") + backup_dir = settings.DATABASE_BACKUP_PATH + backup_dir.mkdir(parents=True) + backup = backup_dir / "sqlite_20260822_030000.db" + with sqlite3.connect(backup) as connection: + connection.execute("CREATE TABLE entries (value TEXT NOT NULL)") + (backup_dir / "sqlite_20260822_040000.db").write_bytes(b"invalid newer backup") + + runner = DoctorRunner() + checks._check_database(runner) + + assert runner.report.find("database.sqlite_open_failed") is not None + finding = runner.report.find("database.backup_recovery") + assert finding is not None + assert finding.status == DoctorFindingStatus.Ok + assert finding.context["backups"][0]["valid"] is False + assert finding.context["backups"][1]["valid"] is True + assert finding.context["restore_command"] == ( + "moviepilot database restore sqlite_20260822_030000.db --confirm" + ) + assert finding.context["restore_command"] in finding.recommendation + + +def test_doctor_distinguishes_missing_and_mismatched_backups(tmp_path, monkeypatch): + """无备份与仅存在其他数据库类型备份应生成不同诊断结论。""" + monkeypatch.setattr(settings, "CONFIG_DIR", str(tmp_path)) + runner = DoctorRunner() + checks._check_database_backups(runner) + missing = runner.report.find("database.backup_recovery") + assert missing is not None + assert missing.status == DoctorFindingStatus.Skipped + + backup_dir = settings.DATABASE_BACKUP_PATH + backup_dir.mkdir(parents=True) + (backup_dir / "postgresql_20260822_030000.dump").write_bytes(b"PGDMP") + runner = DoctorRunner() + checks._check_database_backups(runner) + mismatched = runner.report.find("database.backup_recovery") + assert mismatched is not None + assert mismatched.status == DoctorFindingStatus.Degraded + assert mismatched.context["mismatched"] == ["postgresql_20260822_030000.dump"] + + +def test_doctor_reports_invalid_backup_without_modifying_it(tmp_path, monkeypatch): + """Doctor --fix 也只校验备份，不覆盖或删除无效文件。""" + monkeypatch.setattr(settings, "CONFIG_DIR", str(tmp_path)) + backup_dir = settings.DATABASE_BACKUP_PATH + backup_dir.mkdir(parents=True) + backup = backup_dir / "sqlite_20260822_030000.db" + original = b"invalid sqlite backup" + backup.write_bytes(original) + + runner = DoctorRunner(fix=True) + checks._check_database_backups(runner) + + finding = runner.report.find("database.backup_recovery") + assert finding is not None + assert finding.status == DoctorFindingStatus.Failed + assert finding.affects_report_status is False + assert finding.context["backups"][0]["valid"] is False + assert backup.read_bytes() == original + assert runner.report.status.value == "healthy" + + +def test_doctor_exposes_missing_pg_restore_in_text_finding(tmp_path, monkeypatch): + """PostgreSQL 离线校验工具缺失时应直接告诉用户如何补齐。""" + def missing_pg_restore(*_args, **_kwargs): + raise RuntimeError("未找到 pg_restore，请安装 PostgreSQL client 并加入 PATH") + + monkeypatch.setattr(settings, "CONFIG_DIR", str(tmp_path)) + monkeypatch.setattr(settings, "DB_TYPE", "postgresql") + backup_dir = settings.DATABASE_BACKUP_PATH + backup_dir.mkdir(parents=True) + (backup_dir / "postgresql_20260822_030000.dump").write_bytes(b"PGDMP") + monkeypatch.setattr(checks, "verify_database_backup", missing_pg_restore) + + runner = DoctorRunner() + checks._check_database_backups(runner) + + finding = runner.report.find("database.backup_recovery") + assert finding is not None + assert finding.status == DoctorFindingStatus.Failed + assert "未找到 pg_restore" in finding.detail + assert "PostgreSQL client" in format_text_report(runner.report) + + def test_doctor_formatters_include_status_and_finding(tmp_path, monkeypatch): """doctor 文本和 JSON 格式化应展示状态与诊断项。""" monkeypatch.setattr(settings, "CONFIG_DIR", str(tmp_path)) diff --git a/tests/test_episode_format_helper.py b/tests/test_episode_format_helper.py index 4bb22be05..a81d9a062 100644 --- a/tests/test_episode_format_helper.py +++ b/tests/test_episode_format_helper.py @@ -1,4 +1,5 @@ from pathlib import Path +from types import SimpleNamespace import pytest @@ -22,8 +23,12 @@ def _make_file(name: str, size: int = 150 * 1024 * 1024) -> FileItem: @pytest.fixture(autouse=True) def _patch_media_exts(monkeypatch): monkeypatch.setattr( - "app.application.formatting.settings.RMT_MEDIAEXT", - [".mkv", ".mp4"], + "app.application.formatting.get_chain_runtime_config_snapshot", + lambda: SimpleNamespace( + video_extensions=(".mkv", ".mp4"), + subtitle_extensions=(".srt", ".ass"), + audio_extensions=(".flac", ".mp3"), + ), ) diff --git a/tests/test_release_supply_chain.py b/tests/test_release_supply_chain.py index 06d3a540e..542794591 100644 --- a/tests/test_release_supply_chain.py +++ b/tests/test_release_supply_chain.py @@ -1,5 +1,6 @@ """正式镜像发布的供应链门禁合同。""" +from datetime import date from pathlib import Path from ruamel.yaml import YAML @@ -8,6 +9,7 @@ from ruamel.yaml import YAML ROOT = Path(__file__).resolve().parents[1] DOCKERFILE = ROOT / "docker" / "Dockerfile" RELEASE_WORKFLOW = ROOT / ".github" / "workflows" / "build-v3.yml" +TRIVY_IGNORE = ROOT / ".trivyignore.yaml" def _load_workflow() -> dict: @@ -94,6 +96,18 @@ def test_release_scans_both_architectures_before_registry_login_and_publish() -> assert last_scan < names.index("Publish multi-architecture image") +def test_vulnerability_ignores_are_scoped_justified_and_time_bounded() -> None: + """漏洞豁免必须限定制品范围，并保留复查期限和接受理由。""" + yaml = YAML(typ="safe") + vulnerabilities = yaml.load(TRIVY_IGNORE.read_text(encoding="utf-8"))["vulnerabilities"] + + for vulnerability in vulnerabilities: + assert vulnerability["paths"] + assert vulnerability["purls"] + assert vulnerability["statement"] + assert isinstance(vulnerability["expired_at"], date) + + def test_publish_reuses_scanned_architecture_caches_without_refreshing_base() -> None: """发布构建复用已扫描候选缓存，不得在扫描后重新拉取未审计基础镜像。""" workflow = _load_workflow()