From 5bfef3e63e87c0f4279636e57ce603c0be976982 Mon Sep 17 00:00:00 2001 From: Codex Date: Mon, 24 Aug 2026 13:10:33 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=EF=BC=9A=E5=B0=81=E4=BD=8F?= =?UTF-8?q?=E7=A8=B3=E5=AE=9A=20V1=20=E6=95=B0=E6=8D=AE=E5=AE=89=E5=85=A8?= =?UTF-8?q?=E9=A3=8E=E9=99=A9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- DEVELOPMENT_LOG.md | 12 + NEXT_STEPS.md | 20 ++ STABILITY_REMEDIATION_TASK.md | 84 +++++ app/services/database_backup_service.py | 13 + app/services/storage_service.py | 229 +++++++++++++- app/services/task_lifecycle_service.py | 36 ++- scripts/repair_foreign_key_integrity.py | 347 +++++++++++++++++++++ tests/conftest.py | 26 +- tests/test_database_backup_service.py | 26 ++ tests/test_media_storage_lifecycle.py | 140 ++++++++- tests/test_repair_foreign_key_integrity.py | 263 ++++++++++++++++ tests/test_task_query_service.py | 33 ++ tests/test_test_environment_isolation.py | 19 ++ 13 files changed, 1223 insertions(+), 25 deletions(-) create mode 100644 STABILITY_REMEDIATION_TASK.md create mode 100644 scripts/repair_foreign_key_integrity.py create mode 100644 tests/test_repair_foreign_key_integrity.py create mode 100644 tests/test_test_environment_isolation.py diff --git a/DEVELOPMENT_LOG.md b/DEVELOPMENT_LOG.md index fede4c0..5c2611c 100644 --- a/DEVELOPMENT_LOG.md +++ b/DEVELOPMENT_LOG.md @@ -1,5 +1,17 @@ # Development Log +## 2026-08-24 稳定 V1 P0 数据安全整改 + +- Pytest 启动时改用每进程独立的系统临时 sandbox，并无条件隔离数据库、任务目录、上传临时目录和发布包目录；即使外部 `DATABASE_PATH` 指向活动库，测试也不会连接或清理真实数据。 +- `test_task_query_service` 的整表清理增加第二道 fail-closed 路径校验：数据库不在本次 pytest sandbox 或文件名不是 `test_workflow.sqlite3` 时立即中止。 +- 任务永久删除改为两阶段：托管媒体先原子移动到同卷隔离区并写 manifest，数据库提交失败时逆序恢复；数据库成功后才最终清除，清除失败返回 `cleanup_pending` 并保留恢复证据。外部唯一原片继续不移动、不删除。 +- 新增默认 dry-run 的 `scripts/repair_foreign_key_integrity.py`。活动库预演确认 17 条异常来自 8 个历史缺失的 `output_clip` 父记录；修复只新增 8 个 `is_active=0`、无媒体路径的 tombstone，保留 589 条发布任务、4 条字幕任务、28 条 `NEED_REVIEW` 及发布事件历史。 +- 修复前备份为 `data/backups/workflow-before-foreign-key-repair-20260824-130321-535723-35a0f972.sqlite3`；备份 `quick_check=ok` 并保留修复前 17 条外键异常，活动库修复后 `quick_check=ok`、`foreign_key_check=0`。 +- 三类 SQLite Online Backup 统一转换为 `journal_mode=DELETE` 的便携单文件快照，避免迁移、媒体清理和外键修复备份依赖 WAL/SHM sidecar。 +- 新增测试覆盖恶意活动库环境变量、清理 guard、第二个目录移动失败、数据库提交失败、最终清理失败、外键 dry-run/备份/tombstone/拒绝路径和单文件备份。 +- Windows Web 由 Alter 托管并会自动重启，未停止 Alter；数据库修复在 `BEGIN IMMEDIATE` 锁内完成备份和写入。修复后 `8001 /health=ok`，Scheduler `running=true`、`consecutive_failures=0`，Windows Worker `8765 /health=ok`；未触发真实投稿。 +- 最终验证通过：P0 定向测试 `42 passed`、完整测试 `512 passed`、Ruff、Python 编译与 Git 空白检查全部成功。全量测试故意继承活动库路径后仍使用临时 sandbox；正式 Scheduler 同期持续运行，因此活动库 mtime 会变化，但数据库大小稳定且 `foreign_key_check` 始终为 0。 + ## 2026-08-24 字幕审核、异步渲染与自动流水线整合（PR 4） - 全自动流水线在切片后创建原片/切片字幕草稿，并停在 `pending_subtitle_review`；不会继续生成文案或发送任务。 diff --git a/NEXT_STEPS.md b/NEXT_STEPS.md index 528b504..ee9c68d 100644 --- a/NEXT_STEPS.md +++ b/NEXT_STEPS.md @@ -1,5 +1,25 @@ # Next Steps +## 2026-08-24 稳定 V1 整改路线 + +- [x] P0.1：Pytest 与活动数据库、媒体目录彻底隔离，危险整表清理增加 fail-closed 校验。 +- [x] P0.2：修复活动库 17 条孤儿外键；以 8 个不可见 tombstone 保留发布、字幕和人工复核历史，最终 `foreign_key_check=0`。 +- [x] P0.3：永久删除改为“同卷隔离暂存 → 数据库提交 → 最终清理”，提交失败可恢复，最终清理失败可重试。 +- [x] P0.4：SQLite Online Backup 固定为不依赖 WAL/SHM 的单文件快照。 +- [ ] P1.1：收紧媒体读取与任务目录边界，补齐核心读写路径的 traversal / arbitrary-file 回归测试。 +- [ ] P1.2：为 Workflow Job 和 Publish Job 增加 lease owner / execution generation fencing，阻止旧 Worker 回写新执行。 +- [ ] P1.3：补齐任务状态转移约束、批处理原子性、取消/重启恢复和明确失败状态。 +- [ ] P1.4：统一第三方 AI/FFmpeg 超时、错误 JSON、429/5xx 与重试幂等边界，并避免重复计费。 +- [ ] P1.5：在不扩大个人本地项目范围的前提下处理密钥日志、输入校验和本地管理员接口门禁。 +- [ ] P2：拆分 God Service、去除查询重复、补核心集成/故障测试和可观测性；不做全面重构。 + +### 本轮人工检查 + +1. 正常打开 `http://127.0.0.1:8001/` 和发送中心，确认页面可用、Scheduler 正常、Windows Worker 正常。 +2. 不需要点击“立即发送”；本轮没有执行真实投稿，也没有改变 28 条 `NEED_REVIEW` 的人工确认边界。 +3. 若未来永久删除返回 `cleanup_pending`，不要手工移动隔离目录；保留返回信息和 manifest，使用后续安全清理入口重试。 +4. 修复前数据库备份位于 `data/backups/workflow-before-foreign-key-repair-20260824-130321-535723-35a0f972.sqlite3`，只有活动库无法通过完整性检查时才考虑恢复，不要直接覆盖当前数据库。 + ## 2026-08-23 长直播四阶段进度 - [x] PR 1：模式必选、已有文件入口、媒体/磁盘预检、持久化重型 Job、转写断点与词级时间戳。 diff --git a/STABILITY_REMEDIATION_TASK.md b/STABILITY_REMEDIATION_TASK.md new file mode 100644 index 0000000..fb277ba --- /dev/null +++ b/STABILITY_REMEDIATION_TASK.md @@ -0,0 +1,84 @@ +# 稳定 V1 整改任务书 + +## 背景 + +工程审计确认当前项目属于“可用 V1”，但存在会伤害真实数据或让失败状态不可恢复的 P0 风险。本轮目标是按小步、可测试、可回滚的方式把项目提升到“稳定 V1”，不做全面重构，也不扩大产品范围。 + +## 本轮目标 + +1. 隔离 Pytest 数据库和媒体目录，任何外部 `DATABASE_PATH` 都不能让测试连接活动库；危险清理夹具必须在删除前再次 fail-closed 校验。 +2. 为 SQLite 外键异常提供默认只读预演、应用前强制备份、事务内修复、修复后完整性复查的工具；只在验证备份后处理已确认的孤儿引用。 +3. 把永久删除改为“托管目录暂存隔离 -> 数据库提交 -> 延迟清除”；数据库失败时可把文件恢复原位，外部唯一原片始终不动。 +4. 完成独立测试与验收，并记录剩余 P1/P2 风险和下一轮顺序。 + +## 允许修改范围 + +- `tests/conftest.py` 及与本轮 P0 直接相关的测试。 +- `app/services/storage_service.py` +- `app/services/task_lifecycle_service.py` +- `app/services/database_backup_service.py`（仅复用或补充安全备份能力）。 +- `scripts/` 下新增或调整本轮修复、验证脚本。 +- `DEVELOPMENT_LOG.md`、`NEXT_STEPS.md`、本任务书和必要审计文档。 + +## 禁止修改范围 + +- 不改变 AI Provider、投稿平台、字幕和切片的正常业务语义。 +- 不更改生产 Schema，不删除任务、发布或字幕历史。 +- 不读取、输出或提交 `.env`、Token、Cookie、账号凭据。 +- 不绕过平台登录、验证码、风控或人工确认。 +- 不自动合并 PR，不强制推送，不删除分支。 + +## 已确定实现要求 + +### P0.1 测试隔离 + +- Pytest 启动时无条件使用进程级临时根目录，不继承调用者传入的活动库路径。 +- 临时数据库和媒体目录必须位于同一隔离根目录。 +- 对整表清理增加第二道路径校验；路径不在 Pytest 隔离根目录时立即中止。 +- 验证从命令行故意传入活动库路径时，测试仍不会连接或改写活动库。 + +### P0.2 外键修复 + +- 工具默认 dry-run；只有显式 `--apply` 才写入。 +- 应用前使用 SQLite Online Backup API 创建唯一备份并执行 `quick_check`。 +- 只处理当前检测到且策略明确的孤儿 `publish_jobs.output_clip_id` 和 `subtitle_jobs.output_clip_id`。 +- 修复必须单事务提交；提交前后执行 `foreign_key_check`，不允许产生新异常。 +- 尽量保留历史证据；若表约束不允许安全置空，则先归档必要字段再做最小删除，并在报告中逐条列出。 + +### P0.3 两阶段永久删除 + +- 只处理经过现有托管根目录校验的目录。 +- 文件先原子移动到同盘隔离区并写清单；任一步失败要恢复已经移动的目录。 +- 数据库提交失败时必须恢复目录；不得留下“文件没了、任务仍可见”的半成功状态。 +- 数据库提交成功后再清除隔离区；清除失败要返回明确的 `cleanup_pending`，不得把逻辑删除回滚成可见状态。 +- 重复执行必须幂等；外部原片保持不变。 + +## 验收标准 + +- 相关 P0 回归测试全部通过。 +- 全量测试、Lint/语法检查、前端语法检查通过，或对既有失败给出可复现证据。 +- 活动数据库在运行普通测试前后文件哈希、大小和外键异常计数不发生变化。 +- 外键修复应用前生成可读备份；修复后 `PRAGMA quick_check = ok` 且 `PRAGMA foreign_key_check` 为空。 +- 模拟数据库提交失败时，暂存文件恢复到原路径，任务仍可见。 +- 模拟最终清理失败时，任务保持已删除并返回可恢复的待清理状态。 +- `git diff` 只包含本轮范围，且无敏感信息、调试残留或临时产物。 + +## 测试命令 + +具体临时目录由执行者生成，不得使用 `data/workflow.sqlite3`： + +```powershell +pytest -q tests/test_task_query_service.py tests/test_media_storage_lifecycle.py tests/test_database_backup_service.py +pytest -q +ruff check app tests scripts +python -m compileall -q app scripts +node --check app/static/js/task-detail.js +``` + +## 返回格式 + +- 修改文件与关键行为。 +- 测试命令、退出码、通过/失败数量。 +- 活动数据库备份路径、修复前后外键计数和完整性结果（不含业务内容）。 +- Commit、分支、Push 和 PR 状态。 +- 未完成的 P1/P2 风险与下一轮建议。 diff --git a/app/services/database_backup_service.py b/app/services/database_backup_service.py index b0cead4..68d1c99 100644 --- a/app/services/database_backup_service.py +++ b/app/services/database_backup_service.py @@ -46,6 +46,16 @@ class BackupCleanupResult: released_bytes: int +def _finalize_portable_backup(connection: sqlite3.Connection) -> None: + """把 Online Backup 结果固定为无需 WAL/SHM sidecar 的单文件快照。""" + connection.commit() + row = connection.execute("PRAGMA journal_mode = DELETE").fetchone() + journal_mode = str(row[0]).lower() if row else "" + if journal_mode != "delete": + raise BackupSafetyError(f"备份无法切换为单文件 journal_mode：{journal_mode or 'unknown'}") + connection.commit() + + def sqlite_quick_check(database_path: Path) -> str: path = database_path.resolve() if not path.is_file(): @@ -235,6 +245,7 @@ def create_publish_migration_backup( ) backup_connection = sqlite3.connect(str(temporary_path), timeout=10) source_connection.backup(backup_connection) + _finalize_portable_backup(backup_connection) backup_connection.close() backup_connection = None source_connection.close() @@ -280,6 +291,7 @@ def create_schema_migration_backup(database_path: Path, backup_dir: Path, label: source_connection = sqlite3.connect(f"{database_path.as_uri()}?mode=ro", uri=True, timeout=10) backup_connection = sqlite3.connect(str(temporary_path), timeout=10) source_connection.backup(backup_connection) + _finalize_portable_backup(backup_connection) backup_connection.close() backup_connection = None source_connection.close() @@ -329,6 +341,7 @@ def create_media_cleanup_backup( ) backup_connection = sqlite3.connect(str(temporary_path), timeout=10) source_connection.backup(backup_connection) + _finalize_portable_backup(backup_connection) backup_connection.close() backup_connection = None source_connection.close() diff --git a/app/services/storage_service.py b/app/services/storage_service.py index 3b8b82b..8e4c6bb 100644 --- a/app/services/storage_service.py +++ b/app/services/storage_service.py @@ -1,5 +1,7 @@ from dataclasses import dataclass +from datetime import datetime, timezone from pathlib import Path, PureWindowsPath +import json import os import re import sqlite3 @@ -15,6 +17,7 @@ VIDEO_EXTENSIONS = {".mp4", ".mov", ".mkv", ".avi", ".flv", ".webm", ".m4v", ".ts"} AUDIO_EXTENSIONS = {".wav", ".mp3", ".aac", ".flac", ".ogg", ".wma", ".m4a"} TRASH_DIR_NAME = "_回收站" +DELETE_STAGING_DIR_NAME = ".niuma-delete-staging" _WINDOWS_FORBIDDEN_CHARS = re.compile(r'[<>:"/\\|?*\x00-\x1f]') _WINDOWS_RESERVED_NAMES = { "CON", @@ -55,6 +58,29 @@ class TaskMediaCleanupResult: deleted_paths: tuple[str, ...] freed_bytes: int external_source_preserved: bool + cleanup_pending: bool = False + staged_paths: tuple[str, ...] = () + + +@dataclass(frozen=True) +class StagedMediaTarget: + label: str + original_path: Path + staged_path: Path + size_bytes: int + + +@dataclass(frozen=True) +class StagedTaskMediaCleanup: + task_id: str + stage_id: str + targets: tuple[StagedMediaTarget, ...] + manifest_roots: tuple[Path, ...] + external_source_preserved: bool + + @property + def freed_bytes(self) -> int: + return sum(target.size_bytes for target in self.targets) def _ensure_writable_directory(path: Path, label: str) -> Path: @@ -530,30 +556,205 @@ def task_media_cleanup_plan_size(plan: TaskMediaCleanupPlan) -> int: ) -def apply_task_media_cleanup_plan(plan: TaskMediaCleanupPlan) -> TaskMediaCleanupResult: - deleted_paths: list[str] = [] - freed_bytes = 0 - for target in plan.targets: +def _cleanup_manifest_payload( + staged: StagedTaskMediaCleanup, + *, + status: str, + moved_paths: tuple[str, ...] = (), +) -> dict: + return { + "version": 1, + "task_id": staged.task_id, + "stage_id": staged.stage_id, + "status": status, + "updated_at": datetime.now(timezone.utc).isoformat(timespec="seconds"), + "moved_paths": list(moved_paths), + "targets": [ + { + "label": target.label, + "original_path": str(target.original_path), + "staged_path": str(target.staged_path), + "size_bytes": target.size_bytes, + } + for target in staged.targets + ], + } + + +def _write_cleanup_manifests( + staged: StagedTaskMediaCleanup, + *, + status: str, + moved_paths: tuple[str, ...] = (), +) -> None: + payload = _cleanup_manifest_payload(staged, status=status, moved_paths=moved_paths) + serialized = json.dumps(payload, ensure_ascii=False, indent=2) + for root in staged.manifest_roots: + root.mkdir(parents=True, exist_ok=True) + manifest_path = root / "manifest.json" + temporary_path = root / f"manifest.json.tmp-{uuid4().hex}" + temporary_path.write_text(serialized, encoding="utf-8") + os.replace(temporary_path, manifest_path) + + +def _cleanup_stage_roots(staged: StagedTaskMediaCleanup) -> tuple[str, ...]: + pending: list[str] = [] + for root in staged.manifest_roots: + if not root.exists(): + continue + try: + shutil.rmtree(root) + except OSError: + pending.append(str(root)) + return tuple(pending) + + +def stage_task_media_cleanup_plan(plan: TaskMediaCleanupPlan) -> StagedTaskMediaCleanup: + """把托管媒体原子移动到同卷隔离区，尚不执行永久删除。""" + stage_id = f"{plan.task_id}-{uuid4().hex}" + staged_targets: list[StagedMediaTarget] = [] + manifest_roots: list[Path] = [] + for index, target in enumerate(plan.targets): path = target.path if not path.exists(): continue if path.is_symlink() or not path.is_dir(): - raise StorageSafetyError(f"拒绝删除异常的{target.label}：{path}") - size = _directory_size_bytes(path) + raise StorageSafetyError(f"拒绝暂存异常的{target.label}：{path}") + + stage_root = _safe_managed_child( + path.parent, + (DELETE_STAGING_DIR_NAME, stage_id), + f"{target.label}删除隔离区", + ) + staged_path = _safe_managed_child( + stage_root, + (f"{index:02d}-{path.name}",), + f"{target.label}删除暂存目录", + ) + staged_targets.append( + StagedMediaTarget( + label=target.label, + original_path=path, + staged_path=staged_path, + size_bytes=_directory_size_bytes(path), + ) + ) + if stage_root not in manifest_roots: + manifest_roots.append(stage_root) + + staged = StagedTaskMediaCleanup( + task_id=plan.task_id, + stage_id=stage_id, + targets=tuple(staged_targets), + manifest_roots=tuple(manifest_roots), + external_source_preserved=plan.external_source_path is not None, + ) + if not staged.targets: + return staged + + _write_cleanup_manifests(staged, status="prepared") + moved_paths: list[str] = [] + try: + for target in staged.targets: + target.staged_path.parent.mkdir(parents=True, exist_ok=True) + shutil.move(str(target.original_path), str(target.staged_path)) + if target.original_path.exists() or not target.staged_path.exists(): + raise RuntimeError(f"暂存{target.label}后路径状态异常：{target.original_path}") + moved_paths.append(str(target.original_path)) + _write_cleanup_manifests( + staged, + status="staging", + moved_paths=tuple(moved_paths), + ) + _write_cleanup_manifests( + staged, + status="staged", + moved_paths=tuple(moved_paths), + ) + return staged + except Exception as exc: try: - shutil.rmtree(path) - except OSError as exc: - raise RuntimeError(f"删除{target.label}失败：{path}；原因：{exc}") from exc - freed_bytes += size - deleted_paths.append(str(path)) + rollback_staged_task_media_cleanup(staged) + except Exception as rollback_exc: + raise RuntimeError( + f"暂存任务媒体失败且自动恢复未完成：{exc}；恢复错误：{rollback_exc}" + ) from exc + raise RuntimeError(f"暂存任务媒体失败，已恢复原目录：{exc}") from exc + +def rollback_staged_task_media_cleanup(staged: StagedTaskMediaCleanup) -> None: + """数据库提交前失败时，把已经暂存的目录恢复到原路径。""" + if not staged.targets: + return + try: + _write_cleanup_manifests(staged, status="rolling_back") + except OSError: + pass + + errors: list[str] = [] + for target in reversed(staged.targets): + source = target.staged_path + destination = target.original_path + if source.exists(): + if destination.exists(): + errors.append(f"原路径已被占用：{destination}") + continue + try: + destination.parent.mkdir(parents=True, exist_ok=True) + shutil.move(str(source), str(destination)) + except OSError as exc: + errors.append(f"恢复{target.label}失败：{exc}") + elif not destination.exists(): + errors.append(f"原路径与暂存路径都不存在：{destination}") + + if errors: + try: + _write_cleanup_manifests(staged, status="recovery_required") + except OSError: + pass + raise RuntimeError("；".join(errors)) + + try: + _write_cleanup_manifests(staged, status="rolled_back") + except OSError: + pass + _cleanup_stage_roots(staged) + + +def finalize_staged_task_media_cleanup(staged: StagedTaskMediaCleanup) -> TaskMediaCleanupResult: + """数据库已提交后清除隔离目录；失败时保留清单供安全重试。""" + if staged.targets: + try: + _write_cleanup_manifests( + staged, + status="committed", + moved_paths=tuple(str(target.original_path) for target in staged.targets), + ) + except OSError: + return TaskMediaCleanupResult( + deleted_paths=tuple(str(target.original_path) for target in staged.targets), + freed_bytes=staged.freed_bytes, + external_source_preserved=staged.external_source_preserved, + cleanup_pending=True, + staged_paths=tuple(str(root) for root in staged.manifest_roots if root.exists()), + ) + + pending = _cleanup_stage_roots(staged) return TaskMediaCleanupResult( - deleted_paths=tuple(deleted_paths), - freed_bytes=freed_bytes, - external_source_preserved=plan.external_source_path is not None, + deleted_paths=tuple(str(target.original_path) for target in staged.targets), + freed_bytes=staged.freed_bytes, + external_source_preserved=staged.external_source_preserved, + cleanup_pending=bool(pending), + staged_paths=pending, ) +def apply_task_media_cleanup_plan(plan: TaskMediaCleanupPlan) -> TaskMediaCleanupResult: + """兼容入口：先暂存，再完成清理；需要数据库原子性的调用方应分阶段调用。""" + staged = stage_task_media_cleanup_plan(plan) + return finalize_staged_task_media_cleanup(staged) + + def move_task_directory_to_trash(task_id: str, task_name: str, task_dir_name: str | None = None) -> tuple[str, Path]: current_dir_name = resolve_task_dir_name(task_id, task_dir_name) source_dir = get_task_directory(task_id, current_dir_name) diff --git a/app/services/task_lifecycle_service.py b/app/services/task_lifecycle_service.py index 15b6e80..73e2e83 100644 --- a/app/services/task_lifecycle_service.py +++ b/app/services/task_lifecycle_service.py @@ -9,10 +9,12 @@ from app.db.database import get_connection from app.models.task import TaskCreate, TaskStatus from app.services.storage_service import ( - apply_task_media_cleanup_plan, allocate_task_dir_name, build_task_media_cleanup_plan, create_task_directory, + finalize_staged_task_media_cleanup, + rollback_staged_task_media_cleanup, + stage_task_media_cleanup_plan, validate_source_video_path, ) from app.services.task_log_service import append_task_log @@ -296,6 +298,7 @@ def delete_task_permanently(task_id: str) -> dict: cleanup_plan = build_task_media_cleanup_plan(task) existing_target_count = len(cleanup_plan.existing_targets) now = _now_iso() + staged_cleanup = None with get_connection() as connection: try: connection.execute("BEGIN IMMEDIATE") @@ -338,7 +341,7 @@ def delete_task_permanently(task_id: str) -> dict: if publishing_job: raise TaskDeletionConflictError("任务正在向平台发送视频，请等待发送结束后再删除。") - cleanup_result = apply_task_media_cleanup_plan(cleanup_plan) + staged_cleanup = stage_task_media_cleanup_plan(cleanup_plan) connection.execute( """ UPDATE workflow_jobs @@ -370,13 +373,32 @@ def delete_task_permanently(task_id: str) -> dict: (now, now, task_id), ) connection.commit() - except Exception: - connection.rollback() + except Exception as exc: + try: + connection.rollback() + finally: + if staged_cleanup is not None: + try: + rollback_staged_task_media_cleanup(staged_cleanup) + except Exception as rollback_exc: + raise RuntimeError( + "数据库删除状态提交失败，且媒体自动恢复未完成；" + f"请保留隔离清单并人工恢复。原错误：{exc}；恢复错误：{rollback_exc}" + ) from exc raise - status = "already_deleted" if task.get("is_deleted") and existing_target_count == 0 else "deleted" + cleanup_result = finalize_staged_task_media_cleanup(staged_cleanup) + if cleanup_result.cleanup_pending: + status = "cleanup_pending" + else: + status = "already_deleted" if task.get("is_deleted") and existing_target_count == 0 else "deleted" freed_mb = cleanup_result.freed_bytes / (1024 * 1024) - if status == "already_deleted": + if status == "cleanup_pending": + message = ( + "任务已从系统中永久隐藏，但隔离区文件暂时无法清除；" + "清单已保留，可安全重试清理。" + ) + elif status == "already_deleted": message = "任务已经永久删除，当前没有残留的任务视频文件。" else: message = f"任务已永久删除，共释放约 {freed_mb:.1f} MB；数据库历史记录已隐藏保留。" @@ -386,6 +408,8 @@ def delete_task_permanently(task_id: str) -> dict: "freed_bytes": cleanup_result.freed_bytes, "external_source_preserved": cleanup_result.external_source_preserved, "deleted_paths": list(cleanup_result.deleted_paths), + "cleanup_pending": cleanup_result.cleanup_pending, + "staged_paths": list(cleanup_result.staged_paths), "message": message, } diff --git a/scripts/repair_foreign_key_integrity.py b/scripts/repair_foreign_key_integrity.py new file mode 100644 index 0000000..ee4dc16 --- /dev/null +++ b/scripts/repair_foreign_key_integrity.py @@ -0,0 +1,347 @@ +"""预演并修复 output_clip 缺失造成的 SQLite 外键异常。 + +默认只读。只有显式传入 ``--apply`` 和预期异常数量时才会写入数据库。 +修复策略是不删除发布/字幕历史，而是补充不可见、无媒体路径的占位 output_clip。 +""" + +from __future__ import annotations + +import argparse +import json +import sqlite3 +import sys +from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import Path + + +PROJECT_ROOT = Path(__file__).resolve().parents[1] +if str(PROJECT_ROOT) not in sys.path: + sys.path.insert(0, str(PROJECT_ROOT)) + +from app.core.config import settings # noqa: E402 +from app.services.database_backup_service import ( # noqa: E402 + create_schema_migration_backup, + sqlite_quick_check, +) + + +SUPPORTED_CHILD_TABLES = ("publish_jobs", "subtitle_jobs") +TOMBSTONE_STATUS = "integrity_repair_tombstone" +TOMBSTONE_SOURCE = "integrity_repair_tombstone" + + +class ForeignKeyRepairSafetyError(RuntimeError): + """当前异常不满足自动修复条件。""" + + +@dataclass(frozen=True) +class RepairTombstone: + output_clip_id: str + task_id: str + source_tables: tuple[str, ...] + + +@dataclass(frozen=True) +class ForeignKeyRepairPlan: + database_path: Path + violation_count: int + tombstones: tuple[RepairTombstone, ...] + unsupported_violations: tuple[str, ...] + + @property + def tombstone_count(self) -> int: + return len(self.tombstones) + + @property + def can_apply(self) -> bool: + return not self.unsupported_violations + + +def _open_readonly(database_path: Path) -> sqlite3.Connection: + resolved = database_path.resolve() + if not resolved.is_file(): + raise ForeignKeyRepairSafetyError(f"数据库不存在：{resolved}") + connection = sqlite3.connect( + f"{resolved.as_uri()}?mode=ro", + uri=True, + timeout=10, + ) + connection.row_factory = sqlite3.Row + return connection + + +def _foreign_key_target( + connection: sqlite3.Connection, + table: str, + foreign_key_id: int, +) -> tuple[str, str, str] | None: + rows = connection.execute(f'PRAGMA foreign_key_list("{table}")').fetchall() + for row in rows: + if int(row[0]) == foreign_key_id: + return str(row[2]), str(row[3]), str(row[4]) + return None + + +def _build_plan_from_connection( + connection: sqlite3.Connection, + database_path: Path, +) -> ForeignKeyRepairPlan: + violations = connection.execute("PRAGMA foreign_key_check").fetchall() + grouped: dict[str, dict[str, set[str]]] = {} + unsupported: list[str] = [] + + for violation in violations: + table = str(violation[0]) + row_id = int(violation[1]) + parent = str(violation[2]) + foreign_key_id = int(violation[3]) + target = ( + _foreign_key_target(connection, table, foreign_key_id) + if table in SUPPORTED_CHILD_TABLES + else None + ) + if ( + table not in SUPPORTED_CHILD_TABLES + or parent != "output_clip" + or target != ("output_clip", "output_clip_id", "id") + ): + unsupported.append( + f"不支持的外键异常：table={table}, parent={parent}, fkid={foreign_key_id}" + ) + continue + + row = connection.execute( + f'SELECT output_clip_id, task_id FROM "{table}" WHERE rowid = ?', + (row_id,), + ).fetchone() + if not row or not str(row["output_clip_id"] or "").strip(): + unsupported.append(f"{table} rowid={row_id} 缺少可修复的 output_clip_id") + continue + + output_clip_id = str(row["output_clip_id"]).strip() + task_id = str(row["task_id"] or "").strip() + task_exists = connection.execute( + "SELECT 1 FROM tasks WHERE id = ?", + (task_id,), + ).fetchone() + output_exists = connection.execute( + "SELECT 1 FROM output_clip WHERE id = ?", + (output_clip_id,), + ).fetchone() + if not task_id or not task_exists or output_exists: + unsupported.append( + f"{table} rowid={row_id} 的任务或 output_clip 状态不满足占位修复条件" + ) + continue + + item = grouped.setdefault(output_clip_id, {"task_ids": set(), "tables": set()}) + item["task_ids"].add(task_id) + item["tables"].add(table) + + tombstones: list[RepairTombstone] = [] + for output_clip_id, item in sorted(grouped.items()): + task_ids = item["task_ids"] + if len(task_ids) != 1: + unsupported.append( + f"同一缺失 output_clip 被多个任务引用，拒绝猜测归属：{output_clip_id}" + ) + continue + tombstones.append( + RepairTombstone( + output_clip_id=output_clip_id, + task_id=next(iter(task_ids)), + source_tables=tuple(sorted(item["tables"])), + ) + ) + + return ForeignKeyRepairPlan( + database_path=database_path.resolve(), + violation_count=len(violations), + tombstones=tuple(tombstones), + unsupported_violations=tuple(unsupported), + ) + + +def build_repair_plan(database_path: Path) -> ForeignKeyRepairPlan: + connection = _open_readonly(database_path) + try: + return _build_plan_from_connection(connection, database_path) + finally: + connection.close() + + +def _plan_signature(plan: ForeignKeyRepairPlan) -> tuple: + return ( + plan.violation_count, + tuple( + (item.output_clip_id, item.task_id, item.source_tables) + for item in plan.tombstones + ), + plan.unsupported_violations, + ) + + +def _insert_tombstone( + connection: sqlite3.Connection, + tombstone: RepairTombstone, + now: str, +) -> None: + columns = { + str(row[1]) + for row in connection.execute("PRAGMA table_info(output_clip)").fetchall() + } + required = {"id", "task_id", "status", "created_at", "updated_at", "is_active"} + missing = sorted(required - columns) + if missing: + raise ForeignKeyRepairSafetyError( + "output_clip 缺少安全占位所需字段：" + ", ".join(missing) + ) + + values: dict[str, object] = { + "id": tombstone.output_clip_id, + "task_id": tombstone.task_id, + "status": TOMBSTONE_STATUS, + "created_at": now, + "updated_at": now, + "is_active": 0, + } + optional_values = { + "clip_candidate_id": None, + "output_file_path": "", + "output_file_name": "", + "error_message": "外键完整性修复生成的不可见占位记录；原 output_clip 已缺失", + "cut_run_id": None, + "snapshot_source": TOMBSTONE_SOURCE, + } + values.update({key: value for key, value in optional_values.items() if key in columns}) + + names = tuple(values) + placeholders = ", ".join("?" for _ in names) + quoted_names = ", ".join(f'"{name}"' for name in names) + connection.execute( + f"INSERT INTO output_clip ({quoted_names}) VALUES ({placeholders})", + tuple(values[name] for name in names), + ) + + +def apply_repair_plan( + database_path: Path, + backup_dir: Path, + expected_violation_count: int, +) -> dict: + database_path = database_path.resolve() + backup_dir = backup_dir.resolve() + initial_plan = build_repair_plan(database_path) + if initial_plan.violation_count != expected_violation_count: + raise ForeignKeyRepairSafetyError( + "外键异常数量与人工确认值不一致：" + f"expected={expected_violation_count}, actual={initial_plan.violation_count}" + ) + if initial_plan.unsupported_violations: + raise ForeignKeyRepairSafetyError("；".join(initial_plan.unsupported_violations)) + + connection = sqlite3.connect(str(database_path), timeout=10) + connection.row_factory = sqlite3.Row + connection.execute("PRAGMA foreign_keys = ON") + connection.execute("PRAGMA busy_timeout = 5000") + try: + connection.execute("BEGIN IMMEDIATE") + locked_plan = _build_plan_from_connection(connection, database_path) + if _plan_signature(locked_plan) != _plan_signature(initial_plan): + raise ForeignKeyRepairSafetyError("数据库在预演与写入之间发生变化，已中止") + + backup_path = create_schema_migration_backup( + database_path, + backup_dir, + "foreign-key-repair", + ) + backup_plan = build_repair_plan(backup_path) + if _plan_signature(backup_plan) != _plan_signature(locked_plan): + raise ForeignKeyRepairSafetyError("锁内备份与待修复状态不一致，已中止") + + now = datetime.now(timezone.utc).isoformat(timespec="seconds") + for tombstone in locked_plan.tombstones: + _insert_tombstone(connection, tombstone, now) + + remaining = connection.execute("PRAGMA foreign_key_check").fetchall() + if remaining: + raise ForeignKeyRepairSafetyError( + f"事务内复查仍有 {len(remaining)} 条外键异常，已回滚" + ) + connection.commit() + except Exception: + connection.rollback() + raise + finally: + connection.close() + + if sqlite_quick_check(database_path) != "ok": + raise ForeignKeyRepairSafetyError("修复后数据库 quick_check 失败") + final_plan = build_repair_plan(database_path) + if final_plan.violation_count != 0: + raise ForeignKeyRepairSafetyError( + f"修复后仍有 {final_plan.violation_count} 条外键异常" + ) + + return { + "mode": "apply", + "database_path": str(database_path), + "backup_path": str(backup_path), + "before_violation_count": initial_plan.violation_count, + "after_violation_count": final_plan.violation_count, + "tombstone_count": initial_plan.tombstone_count, + "quick_check": "ok", + } + + +def _dry_run_report(plan: ForeignKeyRepairPlan) -> dict: + table_counts = {table: 0 for table in SUPPORTED_CHILD_TABLES} + for tombstone in plan.tombstones: + for table in tombstone.source_tables: + table_counts[table] += 1 + return { + "mode": "dry-run", + "database_path": str(plan.database_path), + "violation_count": plan.violation_count, + "tombstone_count": plan.tombstone_count, + "tombstone_source_table_counts": table_counts, + "can_apply": plan.can_apply, + "unsupported_violations": list(plan.unsupported_violations), + } + + +def main() -> int: + parser = argparse.ArgumentParser(description="安全修复 output_clip 孤儿外键") + parser.add_argument("--database", type=Path, default=settings.database_path) + parser.add_argument("--backup-dir", type=Path, default=settings.data_dir / "backups") + parser.add_argument("--apply", action="store_true", help="实际写入；默认只预演") + parser.add_argument( + "--expected-violation-count", + type=int, + help="应用时必须提供，且必须与预演数量完全一致", + ) + args = parser.parse_args() + + try: + if args.apply: + if args.expected_violation_count is None: + raise ForeignKeyRepairSafetyError( + "--apply 必须同时提供 --expected-violation-count" + ) + report = apply_repair_plan( + args.database, + args.backup_dir, + args.expected_violation_count, + ) + else: + report = _dry_run_report(build_repair_plan(args.database)) + print(json.dumps(report, ensure_ascii=False, indent=2)) + return 0 + except (sqlite3.Error, OSError, ForeignKeyRepairSafetyError) as exc: + print(f"外键修复已中止：{exc}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/conftest.py b/tests/conftest.py index 7fd1474..320d173 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -2,13 +2,31 @@ import os import sys +import tempfile from pathlib import Path # 让测试代码可以直接导入 app 模块 PROJECT_ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(PROJECT_ROOT)) -os.environ.setdefault("STORAGE_ROOT", str(PROJECT_ROOT / "data" / "test_storage")) -os.environ.setdefault("TASKS_DIR", str(PROJECT_ROOT / "data" / "test_storage")) -os.environ.setdefault("DATA_DIR", str(PROJECT_ROOT / "data")) -os.environ.setdefault("DATABASE_PATH", str(PROJECT_ROOT / "data" / "test_workflow.sqlite3")) +# 测试进程必须拥有自己的数据库和文件根目录。这里刻意使用无条件赋值， +# 防止调用 pytest 时继承 DATABASE_PATH、STORAGE_ROOT 等活动环境变量。 +PYTEST_SANDBOX_ROOT = Path(tempfile.mkdtemp(prefix="niuma-pytest-")).resolve() +PYTEST_STORAGE_ROOT = PYTEST_SANDBOX_ROOT / "storage" +PYTEST_TASKS_DIR = PYTEST_STORAGE_ROOT / "tasks" +PYTEST_UPLOAD_TEMP_DIR = PYTEST_STORAGE_ROOT / "_临时上传" +PYTEST_DATA_DIR = PYTEST_SANDBOX_ROOT / "data" +PYTEST_DATABASE_PATH = PYTEST_DATA_DIR / "test_workflow.sqlite3" +PYTEST_PUBLISH_EXPORT_DIR = PYTEST_STORAGE_ROOT / "_发布包" + +os.environ.update( + { + "STORAGE_ROOT": str(PYTEST_STORAGE_ROOT), + "TASKS_DIR": str(PYTEST_TASKS_DIR), + "UPLOAD_TEMP_DIR": str(PYTEST_UPLOAD_TEMP_DIR), + "DATA_DIR": str(PYTEST_DATA_DIR), + "DATABASE_PATH": str(PYTEST_DATABASE_PATH), + "PUBLISH_SCHEDULER_EXPORT_DIR": str(PYTEST_PUBLISH_EXPORT_DIR), + "NIUMA_PYTEST_SANDBOX_ROOT": str(PYTEST_SANDBOX_ROOT), + } +) diff --git a/tests/test_database_backup_service.py b/tests/test_database_backup_service.py index 80f52f6..3fa63f0 100644 --- a/tests/test_database_backup_service.py +++ b/tests/test_database_backup_service.py @@ -17,6 +17,7 @@ build_cleanup_plan, create_media_cleanup_backup, create_publish_migration_backup, + create_schema_migration_backup, sqlite_quick_check, ) @@ -32,6 +33,14 @@ def _create_database(path: Path, value: str = "ok") -> None: connection.close() +def _assert_portable_backup(path: Path) -> None: + with sqlite3.connect(path) as connection: + journal_mode = connection.execute("PRAGMA journal_mode").fetchone()[0] + assert str(journal_mode).lower() == "delete" + for suffix in ("-wal", "-shm", "-journal"): + assert not Path(f"{path}{suffix}").exists() + + def _set_local_time(path: Path, value: datetime) -> None: timestamp = value.timestamp() os.utime(path, (timestamp, timestamp)) @@ -106,6 +115,22 @@ def test_repeated_backup_within_24_hours_creates_only_one_file(tmp_path): assert second is None assert backups == [first] assert sqlite_quick_check(first) == "ok" + _assert_portable_backup(first) + + +def test_schema_migration_backup_is_portable(tmp_path): + database_path = tmp_path / "workflow.sqlite3" + backup_dir = tmp_path / "backups" + _create_database(database_path) + + backup = create_schema_migration_backup( + database_path, + backup_dir, + "schema-test", + ) + + assert sqlite_quick_check(backup) == "ok" + _assert_portable_backup(backup) def test_media_cleanup_backup_is_always_created_and_valid(tmp_path): @@ -117,6 +142,7 @@ def test_media_cleanup_backup_is_always_created_and_valid(tmp_path): assert backup.name.startswith("workflow-before-media-cleanup-") assert sqlite_quick_check(backup) == "ok" + _assert_portable_backup(backup) def test_concurrent_publish_migration_creates_one_valid_backup(monkeypatch, tmp_path): diff --git a/tests/test_media_storage_lifecycle.py b/tests/test_media_storage_lifecycle.py index 4c25735..c84ff5d 100644 --- a/tests/test_media_storage_lifecycle.py +++ b/tests/test_media_storage_lifecycle.py @@ -1,8 +1,10 @@ from __future__ import annotations import os +import sqlite3 import sys import tempfile +from contextlib import contextmanager from types import SimpleNamespace from pathlib import Path @@ -18,6 +20,7 @@ from app.services import task_lifecycle_service from app.services.storage_service import ( StorageSafetyError, + build_task_media_cleanup_plan, configure_runtime_media_storage, save_uploaded_video, ) @@ -269,7 +272,7 @@ def test_delete_failure_keeps_task_visible(monkeypatch, isolated_media_settings) def fail_cleanup(_plan): raise RuntimeError("模拟文件被占用") - monkeypatch.setattr(task_lifecycle_service, "apply_task_media_cleanup_plan", fail_cleanup) + monkeypatch.setattr(task_lifecycle_service, "stage_task_media_cleanup_plan", fail_cleanup) with pytest.raises(RuntimeError, match="文件被占用"): delete_task_permanently(task_id) @@ -351,3 +354,138 @@ def test_cleanup_aborts_before_deleting_overlapping_active_directory(isolated_me apply_report(report) assert active_dir.exists() + + +def _staged_cleanup_api(): + """返回 P0.3 约定的暂存 API；生产实现尚未接入时给出明确测试失败。""" + from app.services import storage_service + + names = ( + "stage_task_media_cleanup_plan", + "rollback_staged_task_media_cleanup", + "finalize_staged_task_media_cleanup", + ) + missing = [name for name in names if not hasattr(storage_service, name)] + if missing: + pytest.fail( + "P0.3 暂存删除 API 尚未实现：" + ", ".join(missing) + ) + return tuple(getattr(storage_service, name) for name in names) + + +def _task_cleanup_plan(task_id: str): + with get_connection() as connection: + task = connection.execute( + "SELECT * FROM tasks WHERE id = ?", + (task_id,), + ).fetchone() + assert task is not None + return build_task_media_cleanup_plan(dict(task), include_legacy=False) + + +def test_staged_cleanup_rolls_back_when_second_move_fails(monkeypatch, isolated_media_settings): + """第二个托管目录移动失败时，第一个目录必须恢复到原位置。""" + stage_cleanup, _rollback_cleanup, _finalize_cleanup = _staged_cleanup_api() + task_id = "staged-delete-second-move-failure" + task_dir = settings.tasks_dir / task_id + _create_managed_task(task_id, task_dir) + export_dir = settings.publish_scheduler_export_dir / task_id + export_dir.mkdir(parents=True) + (export_dir / "clip.mp4").write_bytes(b"export") + plan = _task_cleanup_plan(task_id) + + from app.services import storage_service + + original_move = storage_service.shutil.move + move_calls = 0 + + def fail_second_move(source, destination): + nonlocal move_calls + move_calls += 1 + if move_calls == 2: + raise OSError("模拟第二个托管目录移动失败") + return original_move(source, destination) + + monkeypatch.setattr(storage_service.shutil, "move", fail_second_move) + with pytest.raises((OSError, RuntimeError), match="移动失败"): + stage_cleanup(plan) + + assert task_dir.exists() + assert export_dir.exists() + assert (task_dir / "source" / "source.mp4").read_bytes() == b"managed-video" + assert (export_dir / "clip.mp4").read_bytes() == b"export" + + +def test_database_commit_failure_restores_staged_media(monkeypatch, isolated_media_settings): + """数据库提交失败时，删除流程不得留下文件已移走、任务仍可见的状态。""" + _staged_cleanup_api() + task_id = "staged-delete-db-commit-failure" + task_dir = settings.tasks_dir / task_id + _create_managed_task(task_id, task_dir) + export_dir = settings.publish_scheduler_export_dir / task_id + export_dir.mkdir(parents=True) + (export_dir / "clip.mp4").write_bytes(b"export") + + original_get_connection = task_lifecycle_service.get_connection + + class CommitFailingConnection: + def __init__(self, connection): + self._connection = connection + + def __getattr__(self, name): + return getattr(self._connection, name) + + def commit(self): + raise sqlite3.OperationalError("模拟数据库提交失败") + + @contextmanager + def failing_connection(): + with original_get_connection() as connection: + yield CommitFailingConnection(connection) + + monkeypatch.setattr(task_lifecycle_service, "get_connection", failing_connection) + with pytest.raises(sqlite3.OperationalError, match="提交失败"): + delete_task_permanently(task_id) + + assert task_dir.exists() + assert export_dir.exists() + with get_connection() as connection: + row = connection.execute( + "SELECT is_deleted FROM tasks WHERE id = ?", + (task_id,), + ).fetchone() + assert row["is_deleted"] == 0 + + +def test_final_cleanup_failure_returns_cleanup_pending_after_db_commit( + monkeypatch, + isolated_media_settings, +): + """数据库已提交后隔离区清理失败，应保留已删除状态并返回待清理。""" + _staged_cleanup_api() + task_id = "staged-delete-final-cleanup-failure" + task_dir = settings.tasks_dir / task_id + _create_managed_task(task_id, task_dir) + + from app.services import storage_service + + original_rmtree = storage_service.shutil.rmtree + rmtree_calls = 0 + + def fail_final_cleanup(path, *args, **kwargs): + nonlocal rmtree_calls + rmtree_calls += 1 + if rmtree_calls == 1: + raise OSError("模拟最终隔离区清理失败") + return original_rmtree(path, *args, **kwargs) + + monkeypatch.setattr(storage_service.shutil, "rmtree", fail_final_cleanup) + result = delete_task_permanently(task_id) + + assert result["status"] == "cleanup_pending" + with get_connection() as connection: + row = connection.execute( + "SELECT is_deleted FROM tasks WHERE id = ?", + (task_id,), + ).fetchone() + assert row["is_deleted"] == 1 diff --git a/tests/test_repair_foreign_key_integrity.py b/tests/test_repair_foreign_key_integrity.py new file mode 100644 index 0000000..d58522d --- /dev/null +++ b/tests/test_repair_foreign_key_integrity.py @@ -0,0 +1,263 @@ +"""外键完整性修复脚本的安全回归测试。 + +这些测试使用独立的临时 SQLite 数据库，不接触项目活动数据库。 +""" + +from __future__ import annotations + +import sqlite3 +from pathlib import Path + +import pytest + +from app.db import database as database_module +from scripts.repair_foreign_key_integrity import ( + apply_repair_plan, + build_repair_plan, +) + + +def _connect(database_path: Path) -> sqlite3.Connection: + connection = sqlite3.connect(str(database_path)) + connection.row_factory = sqlite3.Row + return connection + + +def _foreign_key_violations(database_path: Path) -> list[tuple]: + with _connect(database_path) as connection: + return connection.execute("PRAGMA foreign_key_check").fetchall() + + +def _assert_portable_backup(database_path: Path) -> None: + with _connect(database_path) as connection: + journal_mode = connection.execute("PRAGMA journal_mode").fetchone()[0] + assert str(journal_mode).lower() == "delete" + for suffix in ("-wal", "-shm", "-journal"): + assert not Path(f"{database_path}{suffix}").exists() + + +def _create_database(tmp_path: Path) -> Path: + database_path = tmp_path / "workflow.sqlite3" + runtime_settings = database_module.settings + original_values = { + "database_path": runtime_settings.database_path, + "data_dir": runtime_settings.data_dir, + "tasks_dir": runtime_settings.tasks_dir, + } + try: + object.__setattr__(runtime_settings, "database_path", database_path) + object.__setattr__(runtime_settings, "data_dir", tmp_path) + object.__setattr__(runtime_settings, "tasks_dir", tmp_path / "tasks") + database_module.init_db() + finally: + for name, value in original_values.items(): + object.__setattr__(runtime_settings, name, value) + return database_path + + +def _insert_task(connection: sqlite3.Connection, task_id: str) -> None: + now = "2026-08-24T00:00:00+00:00" + connection.execute( + """ + INSERT INTO tasks (id, task_name, task_dir_name, source_type, created_at, updated_at) + VALUES (?, ?, ?, 'upload', ?, ?) + """, + (task_id, task_id, task_id, now, now), + ) + + +def _insert_orphan_references( + database_path: Path, + *, + publish_task_id: str = "task-a", + subtitle_task_id: str | None = "task-a", + orphan_output_clip_id: str = "orphan-clip-001", +) -> None: + with _connect(database_path) as connection: + connection.execute("PRAGMA foreign_keys = OFF") + _insert_task(connection, publish_task_id) + if subtitle_task_id and subtitle_task_id != publish_task_id: + _insert_task(connection, subtitle_task_id) + now = "2026-08-24T00:00:00+00:00" + connection.execute( + """ + INSERT INTO publish_jobs ( + id, task_id, output_clip_id, platform, video_file_path, video_path, + created_at, updated_at + ) VALUES (?, ?, ?, 'douyin', ?, ?, ?, ?) + """, + ( + "publish-orphan-001", + publish_task_id, + orphan_output_clip_id, + r"D:\private\source.mp4", + r"D:\private\source.mp4", + now, + now, + ), + ) + if subtitle_task_id: + connection.execute( + """ + INSERT INTO subtitle_jobs ( + id, task_id, output_clip_id, subtitle_file_path, + output_file_path, created_at, updated_at + ) VALUES (?, ?, ?, ?, ?, ?, ?) + """, + ( + "subtitle-orphan-001", + subtitle_task_id, + orphan_output_clip_id, + r"D:\private\subtitle.srt", + r"D:\private\subtitle.mp4", + now, + now, + ), + ) + connection.commit() + + +def _insert_unexpected_task_orphan(database_path: Path) -> None: + with _connect(database_path) as connection: + connection.execute("PRAGMA foreign_keys = OFF") + now = "2026-08-24T00:00:00+00:00" + connection.execute( + """ + INSERT INTO output_clip ( + id, task_id, clip_candidate_id, output_file_path, output_file_name, + status, created_at, updated_at + ) VALUES (?, ?, NULL, ?, ?, 'completed', ?, ?) + """, + ( + "output-with-missing-task", + "task-does-not-exist", + r"D:\private\original.mp4", + "original.mp4", + now, + now, + ), + ) + connection.commit() + + +def test_build_repair_plan_is_dry_run_and_counts_supported_tombstone( + tmp_path: Path, +): + database_path = _create_database(tmp_path) + _insert_orphan_references(database_path) + before = _foreign_key_violations(database_path) + + plan = build_repair_plan(database_path) + + assert plan.violation_count == 2 + assert plan.tombstone_count == 1 + assert not plan.unsupported_violations + assert len(_foreign_key_violations(database_path)) == len(before) == 2 + with _connect(database_path) as connection: + assert connection.execute( + "SELECT COUNT(*) FROM output_clip WHERE id = ?", + ("orphan-clip-001",), + ).fetchone()[0] == 0 + + +def test_apply_creates_verified_backup_and_clears_foreign_key_violations( + tmp_path: Path, +): + database_path = _create_database(tmp_path) + _insert_orphan_references(database_path) + backup_dir = tmp_path / "backups" + + result = apply_repair_plan( + database_path, + backup_dir, + expected_violation_count=2, + ) + + backup_path = Path(result["backup_path"]) + assert backup_path.exists() + _assert_portable_backup(backup_path) + assert result["before_violation_count"] == 2 + assert result["after_violation_count"] == 0 + assert result["tombstone_count"] == 1 + assert _foreign_key_violations(database_path) == [] + with _connect(database_path) as connection: + assert connection.execute("PRAGMA quick_check").fetchone()[0] == "ok" + with _connect(backup_path) as backup_connection: + assert backup_connection.execute("PRAGMA quick_check").fetchone()[0] == "ok" + assert len(backup_connection.execute("PRAGMA foreign_key_check").fetchall()) == 2 + + +def test_tombstone_is_inactive_and_does_not_copy_media_paths(tmp_path: Path): + database_path = _create_database(tmp_path) + _insert_orphan_references(database_path) + + apply_repair_plan(database_path, tmp_path / "backups", expected_violation_count=2) + + with _connect(database_path) as connection: + row = connection.execute( + """ + SELECT task_id, status, is_active, output_file_path, output_file_name + FROM output_clip WHERE id = ? + """, + ("orphan-clip-001",), + ).fetchone() + assert row["task_id"] == "task-a" + assert row["status"] == "integrity_repair_tombstone" + assert row["is_active"] == 0 + assert row["output_file_path"] in (None, "") + assert row["output_file_name"] in (None, "") + + +def test_apply_rejects_unexpected_foreign_key_violation_without_writing( + tmp_path: Path, +): + database_path = _create_database(tmp_path) + _insert_orphan_references(database_path) + _insert_unexpected_task_orphan(database_path) + before = len(_foreign_key_violations(database_path)) + backup_dir = tmp_path / "backups" + + plan = build_repair_plan(database_path) + assert plan.unsupported_violations + with pytest.raises((RuntimeError, ValueError), match="(unsupported|不支持|异常|拒绝)"): + apply_repair_plan(database_path, backup_dir, expected_violation_count=before) + + assert len(_foreign_key_violations(database_path)) == before == 3 + with _connect(database_path) as connection: + assert connection.execute( + "SELECT COUNT(*) FROM output_clip WHERE id = ?", + ("orphan-clip-001",), + ).fetchone()[0] == 0 + assert not list(backup_dir.glob("*")) if backup_dir.exists() else True + + +def test_apply_rejects_cross_task_orphan_reference_without_writing(tmp_path: Path): + database_path = _create_database(tmp_path) + _insert_orphan_references(database_path, subtitle_task_id="task-b") + before = len(_foreign_key_violations(database_path)) + backup_dir = tmp_path / "backups" + + plan = build_repair_plan(database_path) + assert plan.unsupported_violations + with pytest.raises((RuntimeError, ValueError), match="(task|任务|冲突|拒绝)"): + apply_repair_plan(database_path, backup_dir, expected_violation_count=before) + + assert len(_foreign_key_violations(database_path)) == before == 2 + assert not list(backup_dir.glob("*")) if backup_dir.exists() else True + + +def test_apply_rejects_unexpected_violation_count_without_writing(tmp_path: Path): + database_path = _create_database(tmp_path) + _insert_orphan_references(database_path) + backup_dir = tmp_path / "backups" + + with pytest.raises((RuntimeError, ValueError), match="(count|数量|expected|预期)"): + apply_repair_plan(database_path, backup_dir, expected_violation_count=99) + + assert len(_foreign_key_violations(database_path)) == 2 + with _connect(database_path) as connection: + assert connection.execute( + "SELECT COUNT(*) FROM output_clip WHERE id = ?", + ("orphan-clip-001",), + ).fetchone()[0] == 0 + assert not list(backup_dir.glob("*")) if backup_dir.exists() else True diff --git a/tests/test_task_query_service.py b/tests/test_task_query_service.py index ebdcaec..fe6a4cf 100644 --- a/tests/test_task_query_service.py +++ b/tests/test_task_query_service.py @@ -10,11 +10,14 @@ """ from datetime import datetime, timezone +import os +from pathlib import Path from uuid import uuid4 import pytest from app.db.database import get_connection, init_db +from app.core.config import settings from app.services.task_query_service import ( get_clips_overview_context, get_dashboard_context, @@ -173,6 +176,21 @@ def _insert_test_subtitle_job( def _clean_test_data() -> None: + sandbox_value = os.environ.get("NIUMA_PYTEST_SANDBOX_ROOT", "").strip() + if not sandbox_value: + raise RuntimeError("拒绝清理测试数据：缺少 NIUMA_PYTEST_SANDBOX_ROOT") + sandbox_root = Path(sandbox_value).resolve() + database_path = Path(settings.database_path).resolve() + try: + is_in_sandbox = database_path.is_relative_to(sandbox_root) + except AttributeError: # pragma: no cover - Python 3.8 兼容 + is_in_sandbox = str(database_path).lower().startswith(str(sandbox_root).lower() + os.sep) + if not is_in_sandbox or database_path.name != "test_workflow.sqlite3": + raise RuntimeError( + "拒绝清理测试数据：数据库不在 pytest sandbox 内或文件名异常；" + f"database={database_path}, sandbox={sandbox_root}" + ) + with get_connection() as connection: connection.execute("DELETE FROM publish_jobs") connection.execute("DELETE FROM subtitle_jobs") @@ -185,6 +203,21 @@ def _clean_test_data() -> None: connection.commit() +def test_clean_test_data_refuses_database_outside_pytest_sandbox(): + """危险整表清理在路径异常时必须先中止，不能连接活动库。""" + original_database_path = settings.database_path + try: + object.__setattr__( + settings, + "database_path", + Path(__file__).resolve().parents[1] / "data" / "workflow.sqlite3", + ) + with pytest.raises(RuntimeError, match="拒绝清理测试数据"): + _clean_test_data() + finally: + object.__setattr__(settings, "database_path", original_database_path) + + @pytest.fixture(autouse=True) def setup_database(): """每个测试前初始化数据库并清理旧数据""" diff --git a/tests/test_test_environment_isolation.py b/tests/test_test_environment_isolation.py new file mode 100644 index 0000000..5ef7e40 --- /dev/null +++ b/tests/test_test_environment_isolation.py @@ -0,0 +1,19 @@ +"""pytest 测试环境必须与活动数据库和媒体目录隔离。""" + +import os +from pathlib import Path + + +def test_pytest_paths_are_under_process_sandbox(): + sandbox = Path(os.environ["NIUMA_PYTEST_SANDBOX_ROOT"]).resolve() + assert sandbox.exists() + assert Path(os.environ["DATABASE_PATH"]).resolve().is_relative_to(sandbox) + assert Path(os.environ["DATABASE_PATH"]).name == "test_workflow.sqlite3" + for name in ( + "STORAGE_ROOT", + "TASKS_DIR", + "UPLOAD_TEMP_DIR", + "DATA_DIR", + "PUBLISH_SCHEDULER_EXPORT_DIR", + ): + assert Path(os.environ[name]).resolve().is_relative_to(sandbox)