test: 共享测试 harness 入 app/testing（网络守卫 + 引导）并统一 sys.modules 打桩原语 (#5888)

app/testing/__init__.py

@@ -1,7 +1,13 @@
 """测试辅助工具（主程序与插件仓共享）。
 
-提供测试期对 ``sys.modules`` 的临时打桩能力，保证打桩在使用后还原，避免测试间
+汇集主程序与插件仓共用的测试 harness，仅供测试使用、不参与运行时逻辑：
-因残留假模块而相互污染。仅供测试使用，不参与运行时逻辑。
+
+- :mod:`app.testing.stub`：测试期对 ``sys.modules`` 的临时打桩并自动还原，避免残留假模块相互污染；
+- :mod:`app.testing.bootstrap`：隔离 CONFIG_DIR、建表、插件目录注入与 v1/v2 marker 等引导逻辑；
+- :mod:`app.testing.network_guard`：autouse 拦截测试期对非本地主机的真实出站。
+
+子模块各自按需 import（如 ``network_guard`` 依赖 pytest），故此处只 re-export 无第三方依赖的
+:func:`stub_modules`，保持 ``import app.testing`` 不引入 pytest 等测试期依赖。
 """
 from app.testing.stub import stub_modules
 

app/testing/bootstrap.py

@@ -0,0 +1,169 @@
+"""测试引导共享实现（主程序与插件仓同源）。
+
+主程序 ``tests/conftest.py`` 与各插件仓的极薄 shim（``tests/_bootstrap.py``，仅负责把
+后端定位并加入 ``sys.path``）都委托到这里，使「隔离 CONFIG_DIR / 建表 / 注入插件目录 /
+按目录打 v1·v2 marker / 退出清理」等引导逻辑只在主程序维护一处，所有消费方行为与修复一致。
+其中 :func:`isolate_config_dir` 为主程序与插件仓共用，``prepare_v1/v2_backend`` 与
+:func:`mark_plugin_generation` 为插件仓专用。
+
+本模块只依赖标准库，``import`` 期不连库、不触发 ``app.db``：调用方可安全地「先 import 本模块、
+再隔离 CONFIG_DIR」，不破坏「隔离必须早于首个 ``import app.db``」这一硬约束。
+"""
+from __future__ import annotations
+
+import atexit
+import os
+import shutil
+import sys
+import tempfile
+from pathlib import Path
+from typing import Optional
+
+# 本进程隔离出的临时 CONFIG_DIR，兼作幂等标记
+_isolated_config_dir: Optional[str] = None
+
+
+def isolate_config_dir() -> str:
+    """把 ``CONFIG_DIR`` 指向进程私有临时目录，隔离主程序真实库与配置（幂等）。
+
+    ``import app.db`` / ``import app.chain.*`` 在 import 期即按 ``settings.CONFIG_PATH`` 连接
+    ``user.db``，故本函数必须在首个 ``import app.db`` 之前调用。调用方已显式设置 ``CONFIG_DIR``
+    （如 CI 指定隔离目录）时尊重之、不覆盖。
+
+    :return: 实际生效的 CONFIG_DIR 绝对路径
+    """
+    global _isolated_config_dir
+    if _isolated_config_dir is not None:
+        return _isolated_config_dir
+    existing = os.environ.get("CONFIG_DIR")
+    if existing:
+        _isolated_config_dir = existing
+        return existing
+    tmp = tempfile.mkdtemp(prefix="mp-test-config-")
+    os.environ["CONFIG_DIR"] = tmp
+    _isolated_config_dir = tmp
+
+    def _cleanup(path: str = tmp, rmtree=shutil.rmtree, sys_mod=sys) -> None:
+        """进程退出时释放 SQLite 连接池再删临时目录。
+
+        默认参数绑定 ``rmtree``/``path``/``sys_mod``：解释器关停期标准库模块可能已被回收为 ``None``，
+        绑定后仍可安全调用。先 ``Engine.dispose`` 释放 ``user.db`` 连接，规避 Windows 下
+        文件锁导致 ``rmtree`` 静默失败（``ignore_errors``）、残留临时目录。
+        """
+        try:
+            db_mod = sys_mod.modules.get("app.db")
+            if db_mod is not None:
+                db_mod.Engine.dispose()
+        except Exception:
+            pass
+        rmtree(path, ignore_errors=True)
+
+    atexit.register(_cleanup)
+    return tmp
+
+
+def _prepend_sys_path(path: Path) -> None:
+    """把目录前置到 ``sys.path``（去重），使其内顶层包可被导入。"""
+    value = str(path)
+    if value not in sys.path:
+        sys.path.insert(0, value)
+
+
+def ensure_sites_stub() -> None:
+    """为 ``app.helper.sites`` 补最小垫片（仅在缺失时）。
+
+    ``app.helper.sites`` 由独立仓库动态拉取，CI / 全新环境无该模块，而众多 ``app.chain.*`` /
+    ``app.modules.*`` 在 import 期依赖它。统一补一个最小垫片，省去各测试文件各自打桩；若真实模块
+    已存在（本地已拉取）则用真实模块、不覆盖，不影响真实行为。须在隔离 CONFIG_DIR 之后调用，
+    以免试探性 ``import app.helper.sites`` 触发的连库落到真实库。
+    """
+    if "app.helper.sites" in sys.modules:
+        return
+    try:
+        import app.helper.sites  # noqa: F401  本地已拉取时用真实模块
+    except ModuleNotFoundError:
+        from types import ModuleType
+        stub = ModuleType("app.helper.sites")
+        stub.SitesHelper = object
+        sys.modules["app.helper.sites"] = stub
+
+
+def ensure_optional_stub(name: str, **attrs) -> None:
+    """为可选第三方依赖补占位模块（仅在缺失时），可带属性。
+
+    用例 import 的 app 代码会牵入可选三方库（如 psutil / dateparser / Pinyin2Hanzi /
+    qbittorrentapi / transmission_rpc），CI / 全新环境可能未安装。本函数在该库缺失时补一个带
+    指定属性的占位，使 import 不致失败；若已真实安装则保留真实模块、不覆盖。占位为进程级常驻
+    （与 import 生命周期一致、不作用域还原），是「让可选 import 不失败」的垫片——与
+    :func:`stub_modules`（作用域内打桩并还原）属不同用途，故不收进 stub_modules。
+
+    :param name: 可选依赖的顶层模块名
+    :param attrs: 占位模块需暴露的属性（仅在真正创建占位时设置）
+    """
+    if name in sys.modules:
+        return
+    try:
+        __import__(name)
+        return
+    except ImportError:
+        pass
+    from types import ModuleType
+    module = ModuleType(name)
+    for key, value in attrs.items():
+        setattr(module, key, value)
+    sys.modules[name] = module
+
+
+def prepare_backend() -> None:
+    """隔离 CONFIG_DIR、补 sites 垫片并建表（后端须已在 ``sys.path`` 上）。
+
+    主程序中后端即当前包；插件仓由其 ``tests/_bootstrap.py`` shim 在 import 本模块前
+    先把后端目录注入 ``sys.path``。顺序固定：先隔离 CONFIG_DIR，再补 ``app.helper.sites`` 垫片，
+    最后建表——隔离出的临时库为空，运行期查 ``systemconfig`` 等表会报 no such table，故建表；
+    ``init_db`` 仅 import models + create_all，无 alembic/网络、幂等、毫秒级。
+    """
+    isolate_config_dir()
+    ensure_sites_stub()
+    from app.db.init import init_db
+    init_db()
+
+
+def prepare_v2_backend(plugins_repo: Path) -> None:
+    """v2 插件单测引导：``prepare_backend`` + 把 ``<repo>/plugins.v2`` 注入 ``sys.path``。
+
+    与 :func:`prepare_v1_backend` 互斥：v1/v2 存在同名插件包，同一进程同时加载会相互覆盖，
+    须在各自独立的 pytest 会话中运行。
+
+    :param plugins_repo: 插件仓根目录（由调用方 shim 传入）
+    """
+    prepare_backend()
+    _prepend_sys_path(Path(plugins_repo) / "plugins.v2")
+
+
+def prepare_v1_backend(plugins_repo: Path) -> None:
+    """v1 插件单测引导：``prepare_backend`` + 把 ``<repo>/plugins`` 注入 ``sys.path``（与 v2 互斥）。
+
+    :param plugins_repo: 插件仓根目录（由调用方 shim 传入）
+    """
+    prepare_backend()
+    _prepend_sys_path(Path(plugins_repo) / "plugins")
+
+
+def mark_plugin_generation(items, pytest_module) -> None:
+    """按用例所在目录自动给其打 ``v1`` / ``v2`` marker，供按代筛选与分会话运行。
+
+    优先读取 pytest 7+ 的 ``item.path``，旧版 pytest 缺失该属性时回退到 ``item.fspath``。用
+    「不带前导斜杠」的子串匹配（``tests/v2/`` / ``tests/v1/``），兼容相对路径与绝对路径两种
+    运行方式：以 ``pytest tests/v2`` 等相对路径运行时收集路径可能不含前导斜杠。
+    ``pytest`` 模块由各仓 conftest 传入，避免本模块在非测试态强依赖 pytest。
+
+    :param items: pytest 收集到的用例集合
+    :param pytest_module: 调用方传入的 ``pytest`` 模块对象
+    """
+    for item in items:
+        item_path = getattr(item, "path", None)
+        path = str(item_path if item_path is not None else item.fspath).replace("\\", "/")
+        if "tests/v2/" in path:
+            item.add_marker(pytest_module.mark.v2)
+        elif "tests/v1/" in path:
+            item.add_marker(pytest_module.mark.v1)

app/testing/network_guard.py

@@ -0,0 +1,40 @@
+"""测试网络守卫（主程序与插件仓共享）。
+
+提供一个 autouse 的 pytest fixture，拦截测试期对非本地主机的真实出站网络。主程序
+``tests/conftest.py`` 与各插件仓 conftest 只需 ``from app.testing.network_guard import
+block_real_network`` 即复用同一道守卫——pytest 会把 conftest 命名空间内（含 import 进来的）
+fixture 一并识别，autouse 自动作用于每个用例，无需逐用例改动。
+
+仅供测试使用，不参与运行时逻辑。
+"""
+from __future__ import annotations
+
+import pytest
+
+# 本地回环/通配地址放行，其余主机一律视为真实出站；getaddrinfo 的 host 可能为 str 或 bytes
+_ALLOWED_NETWORK_HOSTS = {"127.0.0.1", "::1", "localhost", "0.0.0.0", "::", ""}
+
+
+@pytest.fixture(autouse=True)
+def block_real_network(monkeypatch):
+    """防御纵深：拦截对非本地主机的真实出站，强制测试零真实网络。
+
+    补在各用例自身 mock 之上：某用例万一漏 mock 外部依赖（TMDB / LLM 目录 / 下载器 /
+    媒体服务器 / 任意外链），其真实 DNS 解析会在此被拦并报错，而非静默发请求。本地回环放行
+    （sqlite 等）。asyncio 默认解析器经线程池调用 ``socket.getaddrinfo``，故拦此一处即覆盖
+    同步与异步出站。``monkeypatch`` 在用例结束后自动还原，不影响其他用例与进程退出。
+    """
+    import socket
+
+    _real_getaddrinfo = socket.getaddrinfo
+
+    def _guarded_getaddrinfo(host, *args, **kwargs):
+        normalized = host.decode() if isinstance(host, (bytes, bytearray)) else host
+        if normalized is not None and normalized not in _ALLOWED_NETWORK_HOSTS:
+            raise RuntimeError(
+                f"测试禁止真实出站网络：尝试解析 {normalized!r}；请 mock 对应外部依赖"
+            )
+        return _real_getaddrinfo(host, *args, **kwargs)
+
+    monkeypatch.setattr(socket, "getaddrinfo", _guarded_getaddrinfo)
+    yield

tests/conftest.py

@@ -1,43 +1,14 @@
-"""pytest 全局引导：在 import 任何测试模块前把 CONFIG_DIR 指向临时目录并建表，隔离真实库。"""
+"""pytest 全局引导：隔离 CONFIG_DIR、补 sites 垫片、建表、装载网络守卫。
-import atexit
-import os
-import shutil
-import sys
-import tempfile
-from types import ModuleType
 
-# 必须早于首个 import app.*：app.db 在导入时即按 CONFIG_PATH 连接 user.db
+引导与网络守卫均复用 ``app/testing`` 的共享 harness（与插件仓 conftest 同源），
-if not os.environ.get("CONFIG_DIR"):
+引导逻辑只在 ``app/testing`` 维护一处。
-    _isolated_config_dir = tempfile.mkdtemp(prefix="mp-test-config-")
-    os.environ["CONFIG_DIR"] = _isolated_config_dir
-
-    def _cleanup_isolated_config_dir():
-        """进程退出时先释放 SQLite 连接池再删临时目录。
-
-        Windows 下 Engine 若仍持有 user.db 的文件锁，直接 rmtree 会因占用而静默失败
-        （ignore_errors=True）、残留临时目录；先 dispose 释放连接再删可规避。
 """
-        try:
+# 必须早于首个 import app.db（其在 import 期即按 CONFIG_PATH 连库）：prepare_backend 内部
-            from app.db import Engine
+# 先隔离 CONFIG_DIR、补 app.helper.sites 垫片，再建表。app/testing 仅依赖标准库、import 不连库，
-            Engine.dispose()
+# 故此处先 import 再调用是安全的。
-        except Exception:
+from app.testing.bootstrap import prepare_backend
-            pass
-        shutil.rmtree(_isolated_config_dir, ignore_errors=True)
 
-    atexit.register(_cleanup_isolated_config_dir)
+prepare_backend()
 
-# app.helper.sites 由独立仓库动态拉取（CI / 全新环境无该模块），而众多 app.chain.* /
+# 复用共享 autouse 网络守卫；同一实现亦供各插件仓 conftest import 复用，避免逐仓维护
-# app.modules.* 在 import 期依赖它。在此统一补一个最小垫片，省去各测试文件各自打桩；
+from app.testing.network_guard import block_real_network  # noqa: E402,F401
-# 若真实模块已存在（本地已拉取）则 setdefault 不覆盖，不影响真实行为。
-if "app.helper.sites" not in sys.modules:
-    try:
-        import app.helper.sites  # noqa: F401  本地已拉取时用真实模块
-    except ModuleNotFoundError:
-        _sites_stub = ModuleType("app.helper.sites")
-        _sites_stub.SitesHelper = object
-        sys.modules["app.helper.sites"] = _sites_stub
-
-# 必须在 CONFIG_DIR 设好之后再 import；空库会让运行期查表报 no such table，故建表
-from app.db.init import init_db  # noqa: E402
-
-init_db()

tests/test_feishu.py

@@ -1,19 +1,16 @@
-import sys
 import asyncio
 import json
 import tempfile
 import unittest
-from types import ModuleType, SimpleNamespace
+from types import SimpleNamespace
 from unittest.mock import ANY, MagicMock, patch
 
+from app.testing.bootstrap import ensure_optional_stub
 
-sys.modules.setdefault("psutil", ModuleType("psutil"))
+# 可选三方依赖在 CI / 全新环境可能未安装，补占位避免 app.modules.feishu 导入失败
-sys.modules.setdefault("dateparser", ModuleType("dateparser"))
+ensure_optional_stub("psutil")
-
+ensure_optional_stub("dateparser")
-if "Pinyin2Hanzi" not in sys.modules:
+ensure_optional_stub("Pinyin2Hanzi", is_pinyin=lambda value: False)
-    pinyin_module = ModuleType("Pinyin2Hanzi")
-    setattr(pinyin_module, "is_pinyin", lambda value: False)
-    sys.modules["Pinyin2Hanzi"] = pinyin_module
 
 from app.modules.feishu import FeishuModule
 from app.modules.feishu.feishu import Feishu

tests/test_mediaserver_image_signing.py

@@ -1,19 +1,6 @@
-import sys
 import unittest
 from unittest.mock import Mock
 
-for _module_name in (
-    "app.chain.mediaserver",
-    "app.db.models",
-    "app.db.user_oper",
-    "app.helper.message",
-    "app.utils.crypto",
-):
-    if _module_name in sys.modules and not hasattr(
-        sys.modules[_module_name], "__file__"
-    ):
-        del sys.modules[_module_name]
-
 from app.chain.mediaserver import MediaServerChain
 from app.schemas import MediaServerLibrary, MediaServerPlayItem
 from app.utils.security import SecurityUtils

tests/test_plugin_helper.py

@@ -131,14 +131,15 @@ class PluginHelperTest(TestCase):
             self.skipTest(f"missing dependency: {exc}")
 
         module_names = ["app.plugins.dynamicwechat.helper", "Crypto.Cipher._mode_cbc"]
-        previous_modules = {name: sys.modules.get(name) for name in module_names}
 
         def fake_execute(_cmd):
             for module_name in module_names:
                 sys.modules[module_name] = ModuleType(module_name)
             return True, "ok"
 
-        try:
+        # patch.dict 进入时快照 sys.modules、退出时整体还原，替代手写逐项 save/restore；
+        # 保证 fake_execute 在安装窗口注入的运行态模块在用例结束后被清理、不污染其他用例
+        with patch.dict(sys.modules):
             with tempfile.TemporaryDirectory() as temp_dir:
                 requirements_file = Path(temp_dir) / "requirements.txt"
                 requirements_file.write_text("demo-package\n", encoding="utf-8")
@@ -149,12 +150,6 @@ class PluginHelperTest(TestCase):
             self.assertEqual("ok", message)
             for module_name in module_names:
                 self.assertIn(module_name, sys.modules)
-        finally:
-            for module_name, previous_module in previous_modules.items():
-                if previous_module is None:
-                    sys.modules.pop(module_name, None)
-                else:
-                    sys.modules[module_name] = previous_module
 
     def test_pip_install_serializes_concurrent_calls(self):
         """

tests/test_search_ai_recommend.py

@@ -1,28 +1,15 @@
 import asyncio
 import importlib.machinery
-import sys
 import unittest
 from types import SimpleNamespace
-from types import ModuleType
 from unittest.mock import AsyncMock, patch
 
+from app.testing.bootstrap import ensure_optional_stub
 
-def _stub_module(name: str, **attrs):
+# 可选三方依赖在 CI / 全新环境可能未安装，补占位（带用例所需属性）避免导入失败
-    module = sys.modules.get(name)
+ensure_optional_stub("qbittorrentapi", TorrentFilesList=list)
-    if module is None:
+ensure_optional_stub("transmission_rpc", File=object)
-        module = ModuleType(name)
+ensure_optional_stub("psutil", __spec__=importlib.machinery.ModuleSpec("psutil", loader=None))
-        sys.modules[name] = module
-    for key, value in attrs.items():
-        setattr(module, key, value)
-    return module
-
-
-_stub_module("qbittorrentapi", TorrentFilesList=list)
-_stub_module("transmission_rpc", File=object)
-_stub_module(
-    "psutil",
-    __spec__=importlib.machinery.ModuleSpec("psutil", loader=None),
-)
 
 from app.agent.tools.factory import MoviePilotToolFactory
 from app.agent import ReplyMode

tests/test_subscribe_chain.py

@@ -7,6 +7,7 @@ from unittest import TestCase
 from unittest.mock import patch
 
 from app.schemas.types import MediaType
+from app.testing import stub_modules
 
 
 def _load_subscribe_chain_class():
@@ -16,13 +17,11 @@ def _load_subscribe_chain_class():
         module = sys.modules[module_name]
         return module, module.SubscribeChain
 
-    original_modules = {}
+    stub_deps = {}
 
     def ensure_module(name: str, module: types.ModuleType):
-        """临时替换模块依赖，并记录原模块以便加载完成后恢复。"""
+        """登记一个加载期临时替换模块；实际替换与精确还原由 stub_modules 在加载时统一处理。"""
-        if name not in original_modules:
+        stub_deps[name] = module
-            original_modules[name] = sys.modules.get(name)
-        sys.modules[name] = module
         return module
 
     chain_module = ensure_module("app.chain", types.ModuleType("app.chain"))
@@ -298,18 +297,12 @@ def _load_subscribe_chain_class():
     subscribe_path = Path(__file__).resolve().parents[1] / "app" / "chain" / "subscribe.py"
     spec = importlib.util.spec_from_file_location(module_name, subscribe_path)
     module = importlib.util.module_from_spec(spec)
-    sys.modules[module_name] = module
     assert spec and spec.loader
+    # 加载期用 stub_modules 精确替换依赖、退出时统一还原；module_name 非桩，缓存入 sys.modules 供复用
+    with stub_modules(stub_deps):
+        sys.modules[module_name] = module
         spec.loader.exec_module(module)
-    module._injected_modules = {
+        module._injected_modules = {name: sys.modules.get(name) for name in stub_deps}
-        name: sys.modules.get(name)
-        for name in original_modules
-    }
-    for injected_name, original_module in original_modules.items():
-        if original_module is None:
-            sys.modules.pop(injected_name, None)
-        else:
-            sys.modules[injected_name] = original_module
     return module, module.SubscribeChain
 
 

tests/test_system_llm_test.py

@@ -1,24 +1,17 @@
 import asyncio
-import sys
 import unittest
 from types import ModuleType
 from unittest.mock import AsyncMock, patch
 
-
+from app.testing import stub_modules
-_ORIGINAL_STUBBED_MODULES = {}
 
 
-def _stub_module(name: str, **attrs):
+def _stub(name: str, **attrs) -> tuple:
-    """
+    """构造带指定属性的占位模块，返回 ``(模块名, 模块)`` 供 :func:`stub_modules` 使用。"""
-    安装临时 stub 模块，并记录原模块用于导入后恢复。
-    """
-    if name not in _ORIGINAL_STUBBED_MODULES:
-        _ORIGINAL_STUBBED_MODULES[name] = sys.modules.get(name)
     module = ModuleType(name)
     for key, value in attrs.items():
         setattr(module, key, value)
-    sys.modules[name] = module
+    return name, module
-    return module
 
 
 class _Dummy:
@@ -35,70 +28,45 @@ class _DummyError(Exception):
         self.duration_ms = duration_ms
 
 
-for _module_name in ("pillow_avif", "aiofiles", "psutil"):
+# 在 import 期用占位模块替换重依赖/外部模块，import 完由 stub_modules 精确还原，避免污染其它用例
-    _stub_module(_module_name)
+_STUB_MODULES = dict([
-
+    _stub("pillow_avif"),
-_stub_module("app.helper.sites", SitesHelper=_Dummy)
+    _stub("aiofiles"),
-_stub_module("app.chain.mediaserver", MediaServerChain=_Dummy)
+    _stub("psutil"),
-_stub_module("app.chain.search", SearchChain=_Dummy)
+    _stub("app.helper.sites", SitesHelper=_Dummy),
-_stub_module("app.chain.system", SystemChain=_Dummy)
+    _stub("app.chain.mediaserver", MediaServerChain=_Dummy),
-_stub_module(
+    _stub("app.chain.search", SearchChain=_Dummy),
-    "app.agent.llm",
+    _stub("app.chain.system", SystemChain=_Dummy),
-    LLMHelper=_Dummy,
+    _stub("app.agent.llm", LLMHelper=_Dummy, LLMProviderManager=_Dummy,
-    LLMProviderManager=_Dummy,
+          LLMTestError=_DummyError, LLMTestTimeout=_DummyError,
-    LLMTestError=_DummyError,
+          render_auth_result_html=lambda success, message: message),
-    LLMTestTimeout=_DummyError,
+    _stub("app.core.event", eventmanager=_Dummy(), Event=_Dummy, EventManager=_Dummy),
-    render_auth_result_html=lambda success, message: message,
+    _stub("app.core.metainfo", MetaInfo=_Dummy),
-)
+    _stub("app.core.module", ModuleManager=_Dummy),
-_stub_module("app.core.event", eventmanager=_Dummy(), Event=_Dummy, EventManager=_Dummy)
+    _stub("app.core.security", verify_apitoken=_Dummy, verify_resource_token=_Dummy, verify_token=_Dummy),
-_stub_module("app.core.metainfo", MetaInfo=_Dummy)
+    _stub("app.db.models", User=_Dummy),
-_stub_module("app.core.module", ModuleManager=_Dummy)
+    _stub("app.db.systemconfig_oper", SystemConfigOper=_Dummy),
-_stub_module(
+    _stub("app.db.user_oper", get_current_active_superuser=_Dummy,
-    "app.core.security",
+          get_current_active_superuser_async=_Dummy, get_current_active_user_async=_Dummy),
-    verify_apitoken=_Dummy,
+    _stub("app.helper.llm", LLMHelper=_Dummy, LLMTestError=_DummyError, LLMTestTimeout=_DummyError),
-    verify_resource_token=_Dummy,
+    _stub("app.helper.mediaserver", MediaServerHelper=_Dummy),
-    verify_token=_Dummy,
+    _stub("app.helper.message", MessageHelper=_Dummy),
-)
+    _stub("app.helper.progress", ProgressHelper=_Dummy),
-_stub_module("app.db.models", User=_Dummy)
+    _stub("app.helper.rule", RuleHelper=_Dummy),
-_stub_module("app.db.systemconfig_oper", SystemConfigOper=_Dummy)
+    _stub("app.helper.server", MoviePilotServerHelper=_Dummy),
-_stub_module(
+    _stub("app.helper.system", SystemHelper=_Dummy),
-    "app.db.user_oper",
+    _stub("app.helper.image", ImageHelper=_Dummy),
-    get_current_active_superuser=_Dummy,
+    _stub("app.scheduler", Scheduler=_Dummy),
-    get_current_active_superuser_async=_Dummy,
+    _stub("app.log", logger=_Dummy(), log_settings=_Dummy(),
-    get_current_active_user_async=_Dummy,
+          LogConfigModel=type("LogConfigModel", (), {})),
-)
+    _stub("app.utils.crypto", HashUtils=_Dummy),
-_stub_module(
+    _stub("app.utils.http", RequestUtils=_Dummy, AsyncRequestUtils=_Dummy),
-    "app.helper.llm",
+    _stub("version", APP_VERSION="test"),
-    LLMHelper=_Dummy,
+])
-    LLMTestError=_DummyError,
-    LLMTestTimeout=_DummyError,
-)
-_stub_module("app.helper.mediaserver", MediaServerHelper=_Dummy)
-_stub_module("app.helper.message", MessageHelper=_Dummy)
-_stub_module("app.helper.progress", ProgressHelper=_Dummy)
-_stub_module("app.helper.rule", RuleHelper=_Dummy)
-_stub_module("app.helper.server", MoviePilotServerHelper=_Dummy)
-_stub_module("app.helper.system", SystemHelper=_Dummy)
-_stub_module("app.helper.image", ImageHelper=_Dummy)
-_stub_module("app.scheduler", Scheduler=_Dummy)
-_stub_module(
-    "app.log",
-    logger=_Dummy(),
-    log_settings=_Dummy(),
-    LogConfigModel=type("LogConfigModel", (), {}),
-)
-_stub_module("app.utils.crypto", HashUtils=_Dummy)
-_stub_module("app.utils.http", RequestUtils=_Dummy, AsyncRequestUtils=_Dummy)
-_stub_module("version", APP_VERSION="test")
 
+with stub_modules(_STUB_MODULES):
     from app.api.endpoints import llm as system_endpoint
 
-for _module_name, _module in _ORIGINAL_STUBBED_MODULES.items():
-    if _module is None:
-        sys.modules.pop(_module_name, None)
-    else:
-        sys.modules[_module_name] = _module
-
 
 class LlmTestEndpointTest(unittest.TestCase):
     def test_llm_test_requires_ai_agent_enabled(self):

tests/test_system_nettest.py

@@ -1,25 +1,18 @@
 import asyncio
 import ipaddress
-import sys
 import unittest
 from types import ModuleType, SimpleNamespace
 from unittest.mock import AsyncMock, Mock, patch
 
-
+from app.testing import stub_modules
-_ORIGINAL_STUBBED_MODULES = {}
 
 
-def _stub_module(name: str, **attrs):
+def _stub(name: str, **attrs) -> tuple:
-    """
+    """构造带指定属性的占位模块，返回 ``(模块名, 模块)`` 供 :func:`stub_modules` 使用。"""
-    安装临时 stub 模块，并记录原模块用于导入后恢复。
-    """
-    if name not in _ORIGINAL_STUBBED_MODULES:
-        _ORIGINAL_STUBBED_MODULES[name] = sys.modules.get(name)
     module = ModuleType(name)
     for key, value in attrs.items():
         setattr(module, key, value)
-    sys.modules[name] = module
+    return name, module
-    return module
 
 
 class _Dummy:
@@ -36,68 +29,43 @@ class _DummyError(Exception):
         self.duration_ms = duration_ms
 
 
-for _module_name in ("pillow_avif", "aiofiles", "psutil"):
+# 在 import 期用占位模块替换重依赖/外部模块，import 完由 stub_modules 精确还原，避免污染其它用例
-    _stub_module(_module_name)
+_STUB_MODULES = dict([
-
+    _stub("pillow_avif"),
-_stub_module("app.helper.sites", SitesHelper=_Dummy)
+    _stub("aiofiles"),
-_stub_module("app.chain.media", MediaChain=_Dummy)
+    _stub("psutil"),
-_stub_module("app.chain.mediaserver", MediaServerChain=_Dummy)
+    _stub("app.helper.sites", SitesHelper=_Dummy),
-_stub_module("app.chain.search", SearchChain=_Dummy)
+    _stub("app.chain.media", MediaChain=_Dummy),
-_stub_module("app.chain.system", SystemChain=_Dummy)
+    _stub("app.chain.mediaserver", MediaServerChain=_Dummy),
-_stub_module(
+    _stub("app.chain.search", SearchChain=_Dummy),
-    "app.core.event",
+    _stub("app.chain.system", SystemChain=_Dummy),
-    eventmanager=_Dummy(),
+    _stub("app.core.event", eventmanager=_Dummy(), Event=_Dummy, EventManager=_Dummy),
-    Event=_Dummy,
+    _stub("app.core.metainfo", MetaInfo=_Dummy),
-    EventManager=_Dummy,
+    _stub("app.core.module", ModuleManager=_Dummy),
-)
+    _stub("app.core.security", verify_apitoken=_Dummy, verify_resource_token=_Dummy, verify_token=_Dummy),
-_stub_module("app.core.metainfo", MetaInfo=_Dummy)
+    _stub("app.db.models", User=_Dummy),
-_stub_module("app.core.module", ModuleManager=_Dummy)
+    _stub("app.db.systemconfig_oper", SystemConfigOper=_Dummy),
-_stub_module(
+    _stub("app.db.user_oper", get_current_active_superuser=_Dummy,
-    "app.core.security",
+          get_current_active_superuser_async=_Dummy, get_current_active_user_async=_Dummy),
-    verify_apitoken=_Dummy,
+    _stub("app.helper.llm", LLMHelper=_Dummy, LLMTestError=_DummyError, LLMTestTimeout=_DummyError),
-    verify_resource_token=_Dummy,
+    _stub("app.helper.mediaserver", MediaServerHelper=_Dummy),
-    verify_token=_Dummy,
+    _stub("app.helper.message", MessageHelper=_Dummy),
-)
+    _stub("app.helper.progress", ProgressHelper=_Dummy),
-_stub_module("app.db.models", User=_Dummy)
+    _stub("app.helper.rule", RuleHelper=_Dummy),
-_stub_module("app.db.systemconfig_oper", SystemConfigOper=_Dummy)
+    _stub("app.helper.server", MoviePilotServerHelper=_Dummy),
-_stub_module(
+    _stub("app.helper.system", SystemHelper=_Dummy),
-    "app.db.user_oper",
+    _stub("app.helper.image", ImageHelper=_Dummy),
-    get_current_active_superuser=_Dummy,
+    _stub("app.scheduler", Scheduler=_Dummy),
-    get_current_active_superuser_async=_Dummy,
+    _stub("app.log", logger=_Dummy(), log_settings=_Dummy(),
-    get_current_active_user_async=_Dummy,
+          LogConfigModel=type("LogConfigModel", (), {})),
-)
+    _stub("app.utils.crypto", HashUtils=_Dummy),
-_stub_module(
+    _stub("app.utils.http", RequestUtils=_Dummy, AsyncRequestUtils=_Dummy),
-    "app.helper.llm",
+    _stub("version", APP_VERSION="test", FRONTEND_VERSION="frontend-test"),
-    LLMHelper=_Dummy,
+])
-    LLMTestError=_DummyError,
-    LLMTestTimeout=_DummyError,
-)
-_stub_module("app.helper.mediaserver", MediaServerHelper=_Dummy)
-_stub_module("app.helper.message", MessageHelper=_Dummy)
-_stub_module("app.helper.progress", ProgressHelper=_Dummy)
-_stub_module("app.helper.rule", RuleHelper=_Dummy)
-_stub_module("app.helper.server", MoviePilotServerHelper=_Dummy)
-_stub_module("app.helper.system", SystemHelper=_Dummy)
-_stub_module("app.helper.image", ImageHelper=_Dummy)
-_stub_module("app.scheduler", Scheduler=_Dummy)
-_stub_module(
-    "app.log",
-    logger=_Dummy(),
-    log_settings=_Dummy(),
-    LogConfigModel=type("LogConfigModel", (), {}),
-)
-_stub_module("app.utils.crypto", HashUtils=_Dummy)
-_stub_module("app.utils.http", RequestUtils=_Dummy, AsyncRequestUtils=_Dummy)
-_stub_module("version", APP_VERSION="test", FRONTEND_VERSION="frontend-test")
 
+with stub_modules(_STUB_MODULES):
     from app.api.endpoints import system as system_endpoint
 
-for _module_name, _module in _ORIGINAL_STUBBED_MODULES.items():
-    if _module is None:
-        sys.modules.pop(_module_name, None)
-    else:
-        sys.modules[_module_name] = _module
-
 
 class NettestSecurityTest(unittest.TestCase):
     def test_get_env_setting_reports_rust_available_and_enabled_separately(self):

tests/test_testing_bootstrap.py

@@ -0,0 +1,62 @@
+"""共享测试引导工具的回归用例。"""
+from __future__ import annotations
+
+import builtins
+import types
+from pathlib import Path
+
+from app.testing import bootstrap
+
+
+def test_isolate_config_cleanup_uses_loaded_db_module_without_late_import(monkeypatch):
+    """清理回调只读取已加载模块，避免解释器关停期触发二次导入。"""
+    captured = {}
+    import_calls = []
+
+    def fake_import(name, *args, **kwargs):
+        """记录清理回调是否试图重新导入数据库模块。"""
+        if name == "app.db":
+            import_calls.append(name)
+        return original_import(name, *args, **kwargs)
+
+    def fake_register(func):
+        """截获 atexit 回调，便于直接验证清理行为。"""
+        captured["cleanup"] = func
+
+    monkeypatch.setattr(bootstrap, "_isolated_config_dir", None)
+    monkeypatch.delenv("CONFIG_DIR", raising=False)
+    monkeypatch.setattr(bootstrap.tempfile, "mkdtemp", lambda prefix: "/tmp/mp-test-config-demo")
+    monkeypatch.setattr(bootstrap.shutil, "rmtree", lambda *args, **kwargs: None)
+    monkeypatch.setattr(bootstrap.atexit, "register", fake_register)
+
+    original_import = builtins.__import__
+    monkeypatch.setattr(builtins, "__import__", fake_import)
+
+    bootstrap.isolate_config_dir()
+    captured["cleanup"]()
+
+    assert import_calls == []
+
+
+def test_mark_plugin_generation_prefers_pathlib_item_path():
+    """pytest 新版 item.path 可独立驱动 v1/v2 marker 标记。"""
+
+    class FakeItem:
+        """只暴露 pytest 7+ 的 path 属性，模拟新版收集对象。"""
+
+        def __init__(self, value: str):
+            self.path = Path(value)
+            self.markers = []
+
+        def add_marker(self, marker):
+            """记录被添加的 marker。"""
+            self.markers.append(marker)
+
+    pytest_module = types.SimpleNamespace(mark=types.SimpleNamespace(v1="v1", v2="v2"))
+    v2_item = FakeItem("/repo/tests/v2/test_demo.py")
+    v1_item = FakeItem("/repo/tests/v1/test_demo.py")
+
+    bootstrap.mark_plugin_generation([v2_item, v1_item], pytest_module)
+
+    assert v2_item.markers == ["v2"]
+    assert v1_item.markers == ["v1"]