From 791f1fe4ace5e8ab541abcd04103daa4b87bf733 Mon Sep 17 00:00:00 2001 From: InfinityPacer <160988576+InfinityPacer@users.noreply.github.com> Date: Wed, 3 Jun 2026 18:34:20 +0800 Subject: [PATCH] =?UTF-8?q?test:=20=E5=85=B1=E4=BA=AB=E6=B5=8B=E8=AF=95=20?= =?UTF-8?q?harness=20=E5=85=A5=20app/testing=EF=BC=88=E7=BD=91=E7=BB=9C?= =?UTF-8?q?=E5=AE=88=E5=8D=AB=20+=20=E5=BC=95=E5=AF=BC=EF=BC=89=E5=B9=B6?= =?UTF-8?q?=E7=BB=9F=E4=B8=80=20sys.modules=20=E6=89=93=E6=A1=A9=E5=8E=9F?= =?UTF-8?q?=E8=AF=AD=20(#5888)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/testing/__init__.py | 10 +- app/testing/bootstrap.py | 169 ++++++++++++++++++++++++ app/testing/network_guard.py | 40 ++++++ tests/conftest.py | 51 ++----- tests/test_feishu.py | 15 +-- tests/test_mediaserver_image_signing.py | 13 -- tests/test_plugin_helper.py | 11 +- tests/test_search_ai_recommend.py | 23 +--- tests/test_subscribe_chain.py | 25 ++-- tests/test_system_llm_test.py | 114 ++++++---------- tests/test_system_nettest.py | 110 ++++++--------- tests/test_testing_bootstrap.py | 62 +++++++++ 12 files changed, 393 insertions(+), 250 deletions(-) create mode 100644 app/testing/bootstrap.py create mode 100644 app/testing/network_guard.py create mode 100644 tests/test_testing_bootstrap.py diff --git a/app/testing/__init__.py b/app/testing/__init__.py index 1fddb187..58524dd1 100644 --- a/app/testing/__init__.py +++ b/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 diff --git a/app/testing/bootstrap.py b/app/testing/bootstrap.py new file mode 100644 index 00000000..79dc81cd --- /dev/null +++ b/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`` + 把 ``/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`` + 把 ``/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) diff --git a/app/testing/network_guard.py b/app/testing/network_guard.py new file mode 100644 index 00000000..9729e960 --- /dev/null +++ b/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 diff --git a/tests/conftest.py b/tests/conftest.py index ff79b30e..518b72ba 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,43 +1,14 @@ -"""pytest 全局引导：在 import 任何测试模块前把 CONFIG_DIR 指向临时目录并建表，隔离真实库。""" -import atexit -import os -import shutil -import sys -import tempfile -from types import ModuleType +"""pytest 全局引导：隔离 CONFIG_DIR、补 sites 垫片、建表、装载网络守卫。 -# 必须早于首个 import app.*：app.db 在导入时即按 CONFIG_PATH 连接 user.db -if not os.environ.get("CONFIG_DIR"): - _isolated_config_dir = tempfile.mkdtemp(prefix="mp-test-config-") - os.environ["CONFIG_DIR"] = _isolated_config_dir +引导与网络守卫均复用 ``app/testing`` 的共享 harness（与插件仓 conftest 同源）， +引导逻辑只在 ``app/testing`` 维护一处。 +""" +# 必须早于首个 import app.db（其在 import 期即按 CONFIG_PATH 连库）：prepare_backend 内部 +# 先隔离 CONFIG_DIR、补 app.helper.sites 垫片，再建表。app/testing 仅依赖标准库、import 不连库， +# 故此处先 import 再调用是安全的。 +from app.testing.bootstrap import prepare_backend - def _cleanup_isolated_config_dir(): - """进程退出时先释放 SQLite 连接池再删临时目录。 +prepare_backend() - Windows 下 Engine 若仍持有 user.db 的文件锁，直接 rmtree 会因占用而静默失败 - （ignore_errors=True）、残留临时目录；先 dispose 释放连接再删可规避。 - """ - try: - from app.db import Engine - Engine.dispose() - except Exception: - pass - shutil.rmtree(_isolated_config_dir, ignore_errors=True) - - atexit.register(_cleanup_isolated_config_dir) - -# app.helper.sites 由独立仓库动态拉取（CI / 全新环境无该模块），而众多 app.chain.* / -# app.modules.* 在 import 期依赖它。在此统一补一个最小垫片，省去各测试文件各自打桩； -# 若真实模块已存在（本地已拉取）则 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() +# 复用共享 autouse 网络守卫；同一实现亦供各插件仓 conftest import 复用，避免逐仓维护 +from app.testing.network_guard import block_real_network # noqa: E402,F401 diff --git a/tests/test_feishu.py b/tests/test_feishu.py index bfd65659..14f5eec0 100644 --- a/tests/test_feishu.py +++ b/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")) -sys.modules.setdefault("dateparser", ModuleType("dateparser")) - -if "Pinyin2Hanzi" not in sys.modules: - pinyin_module = ModuleType("Pinyin2Hanzi") - setattr(pinyin_module, "is_pinyin", lambda value: False) - sys.modules["Pinyin2Hanzi"] = pinyin_module +# 可选三方依赖在 CI / 全新环境可能未安装，补占位避免 app.modules.feishu 导入失败 +ensure_optional_stub("psutil") +ensure_optional_stub("dateparser") +ensure_optional_stub("Pinyin2Hanzi", is_pinyin=lambda value: False) from app.modules.feishu import FeishuModule from app.modules.feishu.feishu import Feishu diff --git a/tests/test_mediaserver_image_signing.py b/tests/test_mediaserver_image_signing.py index 09c39d80..97adfc73 100644 --- a/tests/test_mediaserver_image_signing.py +++ b/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 diff --git a/tests/test_plugin_helper.py b/tests/test_plugin_helper.py index 765be146..c954c6f1 100644 --- a/tests/test_plugin_helper.py +++ b/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): """ diff --git a/tests/test_search_ai_recommend.py b/tests/test_search_ai_recommend.py index f9da888c..a55b32cf 100644 --- a/tests/test_search_ai_recommend.py +++ b/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): - module = sys.modules.get(name) - if module is None: - module = ModuleType(name) - 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), -) +# 可选三方依赖在 CI / 全新环境可能未安装，补占位（带用例所需属性）避免导入失败 +ensure_optional_stub("qbittorrentapi", TorrentFilesList=list) +ensure_optional_stub("transmission_rpc", File=object) +ensure_optional_stub("psutil", __spec__=importlib.machinery.ModuleSpec("psutil", loader=None)) from app.agent.tools.factory import MoviePilotToolFactory from app.agent import ReplyMode diff --git a/tests/test_subscribe_chain.py b/tests/test_subscribe_chain.py index 70af4828..0f582179 100644 --- a/tests/test_subscribe_chain.py +++ b/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): - """临时替换模块依赖，并记录原模块以便加载完成后恢复。""" - if name not in original_modules: - original_modules[name] = sys.modules.get(name) - sys.modules[name] = module + """登记一个加载期临时替换模块；实际替换与精确还原由 stub_modules 在加载时统一处理。""" + stub_deps[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 - spec.loader.exec_module(module) - module._injected_modules = { - 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 + # 加载期用 stub_modules 精确替换依赖、退出时统一还原；module_name 非桩，缓存入 sys.modules 供复用 + with stub_modules(stub_deps): + sys.modules[module_name] = module + spec.loader.exec_module(module) + module._injected_modules = {name: sys.modules.get(name) for name in stub_deps} return module, module.SubscribeChain diff --git a/tests/test_system_llm_test.py b/tests/test_system_llm_test.py index e3b6822f..9c1668f9 100644 --- a/tests/test_system_llm_test.py +++ b/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 - -_ORIGINAL_STUBBED_MODULES = {} +from app.testing import stub_modules -def _stub_module(name: str, **attrs): - """ - 安装临时 stub 模块，并记录原模块用于导入后恢复。 - """ - if name not in _ORIGINAL_STUBBED_MODULES: - _ORIGINAL_STUBBED_MODULES[name] = sys.modules.get(name) +def _stub(name: str, **attrs) -> tuple: + """构造带指定属性的占位模块，返回 ``(模块名, 模块)`` 供 :func:`stub_modules` 使用。""" module = ModuleType(name) for key, value in attrs.items(): setattr(module, key, value) - sys.modules[name] = module - return module + return name, module class _Dummy: @@ -35,69 +28,44 @@ class _DummyError(Exception): self.duration_ms = duration_ms -for _module_name in ("pillow_avif", "aiofiles", "psutil"): - _stub_module(_module_name) +# 在 import 期用占位模块替换重依赖/外部模块，import 完由 stub_modules 精确还原，避免污染其它用例 +_STUB_MODULES = dict([ + _stub("pillow_avif"), + _stub("aiofiles"), + _stub("psutil"), + _stub("app.helper.sites", SitesHelper=_Dummy), + _stub("app.chain.mediaserver", MediaServerChain=_Dummy), + _stub("app.chain.search", SearchChain=_Dummy), + _stub("app.chain.system", SystemChain=_Dummy), + _stub("app.agent.llm", LLMHelper=_Dummy, LLMProviderManager=_Dummy, + LLMTestError=_DummyError, LLMTestTimeout=_DummyError, + render_auth_result_html=lambda success, message: message), + _stub("app.core.event", eventmanager=_Dummy(), Event=_Dummy, EventManager=_Dummy), + _stub("app.core.metainfo", MetaInfo=_Dummy), + _stub("app.core.module", ModuleManager=_Dummy), + _stub("app.core.security", verify_apitoken=_Dummy, verify_resource_token=_Dummy, verify_token=_Dummy), + _stub("app.db.models", User=_Dummy), + _stub("app.db.systemconfig_oper", SystemConfigOper=_Dummy), + _stub("app.db.user_oper", get_current_active_superuser=_Dummy, + get_current_active_superuser_async=_Dummy, get_current_active_user_async=_Dummy), + _stub("app.helper.llm", LLMHelper=_Dummy, LLMTestError=_DummyError, LLMTestTimeout=_DummyError), + _stub("app.helper.mediaserver", MediaServerHelper=_Dummy), + _stub("app.helper.message", MessageHelper=_Dummy), + _stub("app.helper.progress", ProgressHelper=_Dummy), + _stub("app.helper.rule", RuleHelper=_Dummy), + _stub("app.helper.server", MoviePilotServerHelper=_Dummy), + _stub("app.helper.system", SystemHelper=_Dummy), + _stub("app.helper.image", ImageHelper=_Dummy), + _stub("app.scheduler", Scheduler=_Dummy), + _stub("app.log", logger=_Dummy(), log_settings=_Dummy(), + LogConfigModel=type("LogConfigModel", (), {})), + _stub("app.utils.crypto", HashUtils=_Dummy), + _stub("app.utils.http", RequestUtils=_Dummy, AsyncRequestUtils=_Dummy), + _stub("version", APP_VERSION="test"), +]) -_stub_module("app.helper.sites", SitesHelper=_Dummy) -_stub_module("app.chain.mediaserver", MediaServerChain=_Dummy) -_stub_module("app.chain.search", SearchChain=_Dummy) -_stub_module("app.chain.system", SystemChain=_Dummy) -_stub_module( - "app.agent.llm", - LLMHelper=_Dummy, - LLMProviderManager=_Dummy, - LLMTestError=_DummyError, - LLMTestTimeout=_DummyError, - render_auth_result_html=lambda success, message: message, -) -_stub_module("app.core.event", eventmanager=_Dummy(), Event=_Dummy, EventManager=_Dummy) -_stub_module("app.core.metainfo", MetaInfo=_Dummy) -_stub_module("app.core.module", ModuleManager=_Dummy) -_stub_module( - "app.core.security", - verify_apitoken=_Dummy, - verify_resource_token=_Dummy, - verify_token=_Dummy, -) -_stub_module("app.db.models", User=_Dummy) -_stub_module("app.db.systemconfig_oper", SystemConfigOper=_Dummy) -_stub_module( - "app.db.user_oper", - get_current_active_superuser=_Dummy, - get_current_active_superuser_async=_Dummy, - get_current_active_user_async=_Dummy, -) -_stub_module( - "app.helper.llm", - 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") - -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 +with stub_modules(_STUB_MODULES): + from app.api.endpoints import llm as system_endpoint class LlmTestEndpointTest(unittest.TestCase): diff --git a/tests/test_system_nettest.py b/tests/test_system_nettest.py index 5e1911b2..b9fbd890 100644 --- a/tests/test_system_nettest.py +++ b/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 - -_ORIGINAL_STUBBED_MODULES = {} +from app.testing import stub_modules -def _stub_module(name: str, **attrs): - """ - 安装临时 stub 模块，并记录原模块用于导入后恢复。 - """ - if name not in _ORIGINAL_STUBBED_MODULES: - _ORIGINAL_STUBBED_MODULES[name] = sys.modules.get(name) +def _stub(name: str, **attrs) -> tuple: + """构造带指定属性的占位模块，返回 ``(模块名, 模块)`` 供 :func:`stub_modules` 使用。""" module = ModuleType(name) for key, value in attrs.items(): setattr(module, key, value) - sys.modules[name] = module - return module + return name, module class _Dummy: @@ -36,67 +29,42 @@ class _DummyError(Exception): self.duration_ms = duration_ms -for _module_name in ("pillow_avif", "aiofiles", "psutil"): - _stub_module(_module_name) +# 在 import 期用占位模块替换重依赖/外部模块，import 完由 stub_modules 精确还原，避免污染其它用例 +_STUB_MODULES = dict([ + _stub("pillow_avif"), + _stub("aiofiles"), + _stub("psutil"), + _stub("app.helper.sites", SitesHelper=_Dummy), + _stub("app.chain.media", MediaChain=_Dummy), + _stub("app.chain.mediaserver", MediaServerChain=_Dummy), + _stub("app.chain.search", SearchChain=_Dummy), + _stub("app.chain.system", SystemChain=_Dummy), + _stub("app.core.event", eventmanager=_Dummy(), Event=_Dummy, EventManager=_Dummy), + _stub("app.core.metainfo", MetaInfo=_Dummy), + _stub("app.core.module", ModuleManager=_Dummy), + _stub("app.core.security", verify_apitoken=_Dummy, verify_resource_token=_Dummy, verify_token=_Dummy), + _stub("app.db.models", User=_Dummy), + _stub("app.db.systemconfig_oper", SystemConfigOper=_Dummy), + _stub("app.db.user_oper", get_current_active_superuser=_Dummy, + get_current_active_superuser_async=_Dummy, get_current_active_user_async=_Dummy), + _stub("app.helper.llm", LLMHelper=_Dummy, LLMTestError=_DummyError, LLMTestTimeout=_DummyError), + _stub("app.helper.mediaserver", MediaServerHelper=_Dummy), + _stub("app.helper.message", MessageHelper=_Dummy), + _stub("app.helper.progress", ProgressHelper=_Dummy), + _stub("app.helper.rule", RuleHelper=_Dummy), + _stub("app.helper.server", MoviePilotServerHelper=_Dummy), + _stub("app.helper.system", SystemHelper=_Dummy), + _stub("app.helper.image", ImageHelper=_Dummy), + _stub("app.scheduler", Scheduler=_Dummy), + _stub("app.log", logger=_Dummy(), log_settings=_Dummy(), + LogConfigModel=type("LogConfigModel", (), {})), + _stub("app.utils.crypto", HashUtils=_Dummy), + _stub("app.utils.http", RequestUtils=_Dummy, AsyncRequestUtils=_Dummy), + _stub("version", APP_VERSION="test", FRONTEND_VERSION="frontend-test"), +]) -_stub_module("app.helper.sites", SitesHelper=_Dummy) -_stub_module("app.chain.media", MediaChain=_Dummy) -_stub_module("app.chain.mediaserver", MediaServerChain=_Dummy) -_stub_module("app.chain.search", SearchChain=_Dummy) -_stub_module("app.chain.system", SystemChain=_Dummy) -_stub_module( - "app.core.event", - eventmanager=_Dummy(), - Event=_Dummy, - EventManager=_Dummy, -) -_stub_module("app.core.metainfo", MetaInfo=_Dummy) -_stub_module("app.core.module", ModuleManager=_Dummy) -_stub_module( - "app.core.security", - verify_apitoken=_Dummy, - verify_resource_token=_Dummy, - verify_token=_Dummy, -) -_stub_module("app.db.models", User=_Dummy) -_stub_module("app.db.systemconfig_oper", SystemConfigOper=_Dummy) -_stub_module( - "app.db.user_oper", - get_current_active_superuser=_Dummy, - get_current_active_superuser_async=_Dummy, - get_current_active_user_async=_Dummy, -) -_stub_module( - "app.helper.llm", - 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") - -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 +with stub_modules(_STUB_MODULES): + from app.api.endpoints import system as system_endpoint class NettestSecurityTest(unittest.TestCase): diff --git a/tests/test_testing_bootstrap.py b/tests/test_testing_bootstrap.py new file mode 100644 index 00000000..9a5b6e39 --- /dev/null +++ b/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"]