diff --git a/tests/unit/test_ingest_hevc_transcode_task.py b/tests/unit/test_ingest_hevc_transcode_task.py index db1159220..cbcb8f13a 100644 --- a/tests/unit/test_ingest_hevc_transcode_task.py +++ b/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 纯逻辑用例 -# 断言到 ,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"),