fix(test): 修复test_dedup_pure收集阶段db连接副作用(P0) #894

Closed
xiaoxia wants to merge 1 commits from fix/dedup-pure-test-db-import into develop
+57 -1
View File
@@ -8,9 +8,65 @@ from unittest.mock import MagicMock
import numpy as np
import pytest
# 模块级mock cv2dedup模块import时需要
# 模块级mock有副作用的依赖(纯算法测试不需要db/celery/cv2
# 注意:必须在 import dedup 前全部 mock 完,避免链式导入触发db连接
# cv2(视频处理依赖,纯算法测试不需要)
sys.modules["cv2"] = MagicMock()
# celery 及其子模块
_mock_celery = MagicMock()
_mock_celery.Task = MagicMock
_mock_celery.Celery = MagicMock
sys.modules["celery"] = _mock_celery
# sqlalchemy 作为包结构 mock
_mock_sqla = MagicMock()
_mock_sqla.__path__ = []
_mock_sqla.__package__ = "sqlalchemy"
_mock_sqla_orm = MagicMock()
_mock_sqla_orm.__path__ = []
_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()
# 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
# 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()
from video_processing.dedup import ( # noqa: E402
VideoDeduplicator,
VideoFingerprint,