tests/unit/test_dedup_pure.py

@@ -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.db（dedup模块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
+# sqlalchemy（dedup 导入了 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,