Compare commits

...

1 Commits

Author SHA1 Message Date
xiaoxia 24b4c6f1a0 fix(test): fix Unit Tests collection failure - dedup mock pollutes sys.modules
AI Code Review / AI Code Review (pull_request) Failing after 0s
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 39s
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 1m11s
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Successful in 1m24s
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 / PR Build Worker Image (pull_request) Successful in 56s
CI/CD Pipeline / Frontend Lint (pull_request) Successful in 2m7s
CI/CD Pipeline / Validate - Code Quality (pull_request) Successful in 2m54s
CI/CD Pipeline / PR Build Web Image (pull_request) Successful in 2m21s
Preview Deploy / Deploy Preview Environment (pull_request) Failing after 36s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 4m15s
CI/CD Pipeline / PR Build API Image (pull_request) Successful in 5m45s
CI/CD Pipeline / Frontend Unit Tests (pull_request) Has been skipped
CI/CD Pipeline / Build Production API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Deploy Production (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 / Canary Release to Production (pull_request) Has been cancelled
CI/CD Pipeline / Production Browser E2E (pull_request) Has been skipped
CI/CD Pipeline / Unit Tests (pull_request) Successful in 2m19s
CI/CD Pipeline / Integration Tests (pull_request) Successful in 2m13s
Preview Cleanup / Cleanup Preview Environment (pull_request) Successful in 20s
ACR Cleanup / ACR Image Cleanup (pull_request_target) Failing after 23s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 51m3s
2026-07-26 09:17:53 +08:00
+35 -41
View File
@@ -10,65 +10,59 @@ import pytest
# 模块级mock有副作用的依赖(纯算法测试不需要db/celery/cv2
# 注意:必须在 import dedup 前全部 mock 完,避免链式导入触发db连接
# ⚠️ 只 mock 具体叶子模块,绝不 mock 整个父包,否则会污染其他测试文件的导入
def _mock_module(**attrs):
"""创建带 __spec__ 的 mock 模块,避免导入系统 AttributeError: __spec__"""
m = MagicMock()
m.__spec__ = None
for k, v in attrs.items():
setattr(m, k, v)
return m
# cv2(视频处理依赖,纯算法测试不需要)
sys.modules["cv2"] = MagicMock()
# 模块级mock worker_app.dbdedup模块import时会触发数据库初始化,纯算法测试不需要)
sys.modules["worker_app.db"] = MagicMock()
sys.modules["worker_app.db"].SessionLocal = MagicMock()
sys.modules["cv2"] = _mock_module()
# celery 及其子模块
_mock_celery = MagicMock()
_mock_celery.Task = MagicMock
_mock_celery.Celery = MagicMock
_mock_celery.__spec__ = None
sys.modules["celery"] = _mock_celery
# sqlalchemy 作为包结构 mock
# sqlalchemydedup 导入了 Session 类型)
_mock_sqla = MagicMock()
_mock_sqla.__path__ = []
_mock_sqla.__package__ = "sqlalchemy"
_mock_sqla.__spec__ = None
_mock_sqla_orm = MagicMock()
_mock_sqla_orm.__path__ = []
_mock_sqla_orm.__spec__ = None
_mock_sqla_orm.Session = MagicMock
_mock_sqla_engine = MagicMock()
sys.modules["sqlalchemy"] = _mock_sqla
sys.modules["sqlalchemy.orm"] = _mock_sqla_orm
sys.modules["sqlalchemy.engine"] = _mock_sqla_engine
sys.modules["sqlalchemy.ext"] = MagicMock()
sys.modules["sqlalchemy.ext.declarative"] = MagicMock()
sys.modules["sqlalchemy.engine"] = _mock_module()
sys.modules["sqlalchemy.ext"] = _mock_module()
sys.modules["sqlalchemy.ext.declarative"] = _mock_module()
# worker_app 及其子模块(避免导入时触发数据库连接
_mock_worker_app = MagicMock()
_mock_worker_app.__path__ = []
_mock_worker_db = MagicMock()
_mock_worker_db.SessionLocal = MagicMock()
_mock_worker_celery = MagicMock()
_mock_worker_celery.celery_app = MagicMock()
_mock_worker_core = MagicMock()
_mock_worker_core.__path__ = []
_mock_worker_config = MagicMock()
_mock_worker_config.get_settings = MagicMock(return_value=MagicMock())
sys.modules["worker_app"] = _mock_worker_app
sys.modules["worker_app.db"] = _mock_worker_db
sys.modules["worker_app.celery_app"] = _mock_worker_celery
sys.modules["worker_app.core"] = _mock_worker_core
sys.modules["worker_app.core.config"] = _mock_worker_config
# worker_app 子模块(只 mock 具体需要的,不 mock 整个 worker_app 包
sys.modules["worker_app.db"] = _mock_module(SessionLocal=MagicMock())
sys.modules["worker_app.celery_app"] = _mock_module(celery_app=MagicMock())
sys.modules["worker_app.core.config"] = _mock_module(get_settings=MagicMock(return_value=MagicMock()))
# packages.adapters.sqlalchemy_impl(整个包mock掉)
_mock_sqla_impl = MagicMock()
_mock_sqla_impl.__path__ = []
sys.modules["packages.adapters.sqlalchemy_impl"] = _mock_sqla_impl
sys.modules["packages.adapters.sqlalchemy_impl.generated_video_repository"] = MagicMock()
sys.modules["packages.adapters.sqlalchemy_impl.schema_guard"] = MagicMock()
# packages.shared
_mock_packages_shared = MagicMock()
_mock_packages_shared.__path__ = []
_mock_shared_config = MagicMock()
_mock_shared_config.get_shared_settings = MagicMock(return_value=MagicMock())
sys.modules["packages.shared"] = _mock_packages_shared
sys.modules["packages.shared.config"] = _mock_shared_config
sys.modules["packages.shared.storage"] = MagicMock()
# packages - 只 mock 真正触发副作用的模块,不 mock 整个父包
# session 模块是触发数据库连接的元凶(ensure_database_exists),必须 mock 掉
sys.modules["packages.adapters.sqlalchemy_impl.session"] = _mock_module(
Base=MagicMock(),
build_engine=MagicMock(),
build_session_factory=MagicMock(),
ensure_database_exists=MagicMock(),
initialize_database=MagicMock(),
)
sys.modules["packages.adapters.sqlalchemy_impl.generated_video_repository"] = _mock_module()
sys.modules["packages.shared.config"] = _mock_module(get_shared_settings=MagicMock(return_value=MagicMock()))
sys.modules["packages.shared.storage"] = _mock_module()
from video_processing.dedup import ( # noqa: E402
VideoDeduplicator,