tests/unit/test_ingest_hevc_transcode_task.py

@@ -16,6 +16,14 @@ from types import SimpleNamespace
 from unittest.mock import MagicMock, patch
 
 # 在 import worker_app 模块前 mock 掉数据库连接和 celery（同 test_ingest_validation.py）
+# 注意：模块级 sys.modules 注入若不撤销，会污染同一 pytest 进程（含 xdist
+# worker）后续收集/执行的其他测试文件——它们 from video_processing.xxx
+# import 会拿到 MagicMock（表现为 test_thumbnail_generator 纯逻辑用例
+# 断言到 <MagicMock>，50 个用例失败，且与 xdist 分发顺序相关）。
+# 因此在成功 import ingest_mod 之后立即恢复 sys.modules（同 test_dedup_pure.py
+# 的做法），mock 对象仍由本文件变量/ingest_mod 引用持有，不影响本文件测试。
+_SAVED_MODULES_KEYS = set(sys.modules.keys())
+
 _mock_db_module = MagicMock()
 _mock_db_module.SessionLocal = MagicMock()
 sys.modules["worker_app.db"] = _mock_db_module
@@ -44,6 +52,15 @@ sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "worker"))
 import pytest  # noqa: E402
 from worker_app.tasks import ingest as ingest_mod  # noqa: E402
 
+# ── 立即恢复 sys.modules，避免 mock 泄漏到其他测试文件 ──
+for _key in list(sys.modules.keys()):
+    if _key not in _SAVED_MODULES_KEYS:
+        # 保留 mock 对象供本文件 patch.object 使用
+        if _key in ("video_processing.oss_helpers", "video_processing.thumbnail_generator"):
+            continue
+        del sys.modules[_key]
+del _SAVED_MODULES_KEYS
+
 
 class _FakeJobRepo:
     def __init__(self, db):