tests/unit/test_ingest_hevc_transcode_task.py

@@ -53,14 +53,20 @@ import pytest  # noqa: E402
 from worker_app.tasks import ingest as ingest_mod  # noqa: E402
 
 # ── 立即恢复 sys.modules，避免 mock 泄漏到其他测试文件 ──
+# 本文件 patch.object 不依赖 mock 条目保留在 sys.modules：download_asset
+# 已被 ingest_mod 顶部 from import 绑定；thumb patcher 是字符串目标，start
+# 时会触发真实模块重新导入后再打补丁（真实模块 CI 可导入）。因此注入的
+# mock 条目必须全部删除——若保留 video_processing.* mock，同 xdist worker
+# 后续测试文件 import 仍会拿到 MagicMock（#1566 v1 曾因此漏修）。
 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
 
+# 导入真实模块供本文件 patch.object 打补丁（restore 后 sys.modules 中已无
+# mock）。patch 在 stop 时会自动还原模块属性，不影响其他测试文件。
+from video_processing import oss_helpers as _oss_helpers_real  # noqa: E402
+
 
 class _FakeJobRepo:
     def __init__(self, db):
@@ -172,7 +178,7 @@ def task_env(tmp_path):
         "asset_repo": patch.object(ingest_mod, "SQLAlchemyAssetRepository", return_value=asset_repo),
         "download": patch.object(ingest_mod, "download_asset", return_value=True),
         "upload": patch.object(
-            sys.modules["video_processing.oss_helpers"],
+            _oss_helpers_real,
             "upload_to_oss",
             return_value=control["upload_url"],
         ),