fix(test): hevc 转码测试不再注入 video_processing mock 到 sys.modules
CI/CD Pipeline / Check push changed paths (pull_request) Has been skipped
CI/CD Pipeline / Dedup Check - skip PR tests when covered by push pipeline (pull_request) Successful in 1s
CI/CD Pipeline / Build Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Retag skipped Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Retag skipped Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Retag skipped Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / ACR Image Cleanup (pull_request) Has been skipped
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 3m43s
CI/CD Pipeline / Frontend Lint (pull_request) Has been skipped
CI/CD Pipeline / Frontend Unit Tests (pull_request) Has been skipped
CI/CD Pipeline / PR Build Web Image (pull_request) Has been skipped
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 4m1s
AI Code Review / AI Code Review (pull_request) Successful in 4m38s
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 4m44s
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Successful in 4m48s
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 4m57s
CI/CD Pipeline / PR Build API Image (pull_request) Successful in 1m44s
CI/CD Pipeline / PR Build Worker Image (pull_request) Successful in 1m51s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 7m8s
CI/CD Pipeline / Validate - Code Quality (pull_request) Has been cancelled
CI/CD Pipeline / Unit Tests (pull_request) Has been cancelled
CI/CD Pipeline / Integration Tests (pull_request) Has been cancelled
CI/CD Pipeline / Build Production API Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Web Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been cancelled
CI/CD Pipeline / Deploy Production (pull_request) Has been cancelled
CI/CD Pipeline / Production Browser E2E (pull_request) Has been cancelled
CI/CD Pipeline / Canary Release to Production (pull_request) Has been cancelled
CI/CD Pipeline / CI Gate (pull_request) Has been cancelled

根因:sys.modules 注入 video_processing.oss_helpers / thumbnail_generator 的
MagicMock 即使清理,仍因 video_processing/__init__.py 的
`from . import oss_helpers` 导致包属性残留为 MagicMock,
跨文件污染 test_oss_helpers_pure / test_oss_upload_crash_fix /
test_p02_worker_oss_fix 等(39 个用例失败)。

修复:不再向 sys.modules 注入 video_processing.* mock,改为直接
import 真实模块后用 patch.object 打补丁(patch.stop 自动还原)。
worker_app.db / celery_app mock 保留(独立包树,不影响共享工具包)。
This commit is contained in:
2026-08-31 09:04:05 +08:00
parent d2098b03ab
commit e5d7fddfcf
+18 -25
View File
@@ -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,24 @@ 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"))
# ── 先导入真实 video_processing 子模块,再导入 ingest_mod ──
# 不使用 sys.modules 注入 mock,避免 video_processing/__init__.py 的
# `from . import oss_helpers` 与 mock 注入/清理产生时序冲突。
from video_processing import oss_helpers as _oss_helpers_real # noqa: E402
from video_processing import thumbnail_generator as _thumb_gen_real # noqa: E402
import pytest # noqa: E402
from worker_app.tasks import ingest as ingest_mod # noqa: E402
# ── 立即恢复 sys.modules,避免 mock 泄漏到其他测试文件 ──
# 本文件 patch.object 不依赖 mock 条目保留在 sys.modulesdownload_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:
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):
@@ -200,9 +191,10 @@ 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),
# 缩略图生成跳过
"thumb": patch(
"video_processing.thumbnail_generator.extract_first_frame",
# 缩略图生成跳过——打在真实模块属性上,patch.stop() 自动还原
"thumb": patch.object(
_thumb_gen_real,
"extract_first_frame",
side_effect=RuntimeError("skip thumb"),
),
}
@@ -312,3 +304,4 @@ class TestIngestHEVCTranscodeFlow:
for call in mocks["subprocess"].call_args_list:
cmd = call.args[0] if call.args else call.kwargs.get("cmd", [])
assert "libx264" not in cmd