tests/unit/test_ingest_hevc_transcode_task.py

@@ -16,12 +16,11 @@ 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 引用持有，不影响本文件测试。
+# 注意：只 mock worker_app 的依赖（db / celery），不 mock video_processing.*——
+# 后者属于共享工具包，sys.modules 注入即使清理也会因 video_processing/__init__.py
+# 的 `from . import oss_helpers` 导致包属性残留为 MagicMock，跨文件污染
+# test_oss_helpers_pure / test_oss_upload_crash_fix / test_p02_worker_oss_fix
+# 等测试文件（39 个用例失败）。改为直接 import 真实模块，patch.object 打补丁。
 _SAVED_MODULES_KEYS = set(sys.modules.keys())
 
 _mock_db_module = MagicMock()
@@ -41,32 +40,21 @@ def _passthrough_decorator(*args, **kwargs):
 _mock_celery_module.celery_app.task = MagicMock(side_effect=_passthrough_decorator)
 sys.modules["worker_app.celery_app"] = _mock_celery_module
 
-# mock video_processing 子模块（主流程会 import 它们）
-_oss_helpers_mock = MagicMock()
-_thumbnail_mock = MagicMock()
-sys.modules["video_processing.oss_helpers"] = _oss_helpers_mock
-sys.modules["video_processing.thumbnail_generator"] = _thumbnail_mock
-
 sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "worker"))
 
 import pytest  # noqa: E402
+
+# ── 先导入真实 video_processing 子模块，再导入 ingest_mod ──
+# 不使用 sys.modules 注入 mock，避免 video_processing/__init__.py 的
+# `from . import oss_helpers` 与 mock 注入/清理产生时序冲突。
 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 曾因此漏修）。
+# ── 恢复 worker_app mock，避免泄漏到其他测试文件 ──
 for _key in list(sys.modules.keys()):
-    if _key not in _SAVED_MODULES_KEYS:
+    if _key not in _SAVED_MODULES_KEYS and not _key.startswith("video_processing"):
         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):
@@ -177,9 +165,8 @@ def task_env(tmp_path):
         "job_repo": patch.object(ingest_mod, "SQLAlchemyIngestJobRepository", return_value=job_repo),
         "asset_repo": patch.object(ingest_mod, "SQLAlchemyAssetRepository", return_value=asset_repo),
         "download": patch.object(ingest_mod, "download_asset", return_value=True),
-        "upload": patch.object(
-            _oss_helpers_real,
-            "upload_to_oss",
+        "upload": patch(
+            "video_processing.oss_helpers.upload_to_oss",
             return_value=control["upload_url"],
         ),
         "metadata": patch.object(
@@ -200,7 +187,7 @@ def task_env(tmp_path):
         ),
         "subprocess": patch.object(ingest_mod.subprocess, "run", side_effect=fake_subprocess_run),
         "ntf": patch.object(tempfile, "NamedTemporaryFile", side_effect=fake_ntf),
-        # 缩略图生成跳过
+        # 缩略图生成跳过——字符串目标，patch.start 时从 sys.modules 取真实模块再打补丁
         "thumb": patch(
             "video_processing.thumbnail_generator.extract_first_frame",
             side_effect=RuntimeError("skip thumb"),