From f814c271cc5412c365a631bbdb8b2c9e9eab27d9 Mon Sep 17 00:00:00 2001 From: InfinityPacer <160988576+InfinityPacer@users.noreply.github.com> Date: Tue, 14 Jul 2026 16:03:29 +0800 Subject: [PATCH] refactor(runtime): tighten resource cleanup and test isolation (#6116) --- app/helper/doh.py | 55 ++++++++-- app/helper/message.py | 14 ++- app/log.py | 93 ++++++++++------- app/modules/telegram/telegram.py | 4 - app/startup/lifecycle.py | 39 ++++--- app/startup/modules_initializer.py | 2 + app/testing/network_guard.py | 42 ++++++-- app/utils/singleton.py | 9 ++ pytest.ini | 4 - tests/conftest.py | 22 ++-- tests/test_doh_helper.py | 56 ++++++++++ tests/test_emby_dashboard_links.py | 1 + tests/test_lifecycle_shutdown.py | 46 +++++++++ tests/test_log_shutdown.py | 149 +++++++++++++++++++++++++++ tests/test_media_interaction.py | 10 ++ tests/test_message_queue_shutdown.py | 30 ++++++ tests/test_network_guard.py | 22 ++++ tests/test_singleton.py | 29 ++++++ tests/test_subscribe_endpoint.py | 3 +- tests/test_telegram.py | 29 ++++-- tests/test_web_agent_stream.py | 3 +- 21 files changed, 563 insertions(+), 99 deletions(-) create mode 100644 tests/test_lifecycle_shutdown.py create mode 100644 tests/test_log_shutdown.py create mode 100644 tests/test_message_queue_shutdown.py create mode 100644 tests/test_network_guard.py create mode 100644 tests/test_singleton.py diff --git a/app/helper/doh.py b/app/helper/doh.py index 620be3934..be1e69b4c 100644 --- a/app/helper/doh.py +++ b/app/helper/doh.py @@ -18,8 +18,10 @@ from app.log import logger from app.utils.mixins import ConfigReloadMixin from app.utils.singleton import Singleton -# 定义一个全局线程池执行器 -_executor = concurrent.futures.ThreadPoolExecutor() +# DoH 关闭时需要释放线程池；保持惰性创建可避免未启用 DoH 时占用进程级资源 +_executor: Optional[concurrent.futures.ThreadPoolExecutor] = None +_executor_lock = Lock() +_doh_enabled = False # 定义默认的DoH配置 _doh_timeout = 5 @@ -29,11 +31,21 @@ _doh_lock = Lock() _orig_getaddrinfo = socket.getaddrinfo +def _get_executor_locked() -> concurrent.futures.ThreadPoolExecutor: + """在持有执行器锁时按需获取 DoH 查询线程池""" + global _executor + if _executor is None: + _executor = concurrent.futures.ThreadPoolExecutor() + return _executor + + def enable_doh(enable: bool) -> None: """ 对 socket.getaddrinfo 进行补丁 """ + global _doh_enabled + def _patched_getaddrinfo(host: str, *args, **kwargs): """ socket.getaddrinfo的补丁版本。 @@ -47,9 +59,15 @@ def enable_doh(enable: bool) -> None: logger.info(f"已解析 [{host}] 为 [{ip}] (缓存)") return _orig_getaddrinfo(ip, *args, **kwargs) # 使用DoH解析主机 - futures = [] - for resolver in settings.DOH_RESOLVERS.split(","): - futures.append(_executor.submit(_doh_query, resolver, host)) + with _executor_lock: + if not _doh_enabled: + return _orig_getaddrinfo(host, *args, **kwargs) + executor = _get_executor_locked() + # 一次解析的任务必须在同一临界区提交完，避免关闭过程中部分任务落入新线程池 + futures = [ + executor.submit(_doh_query, resolver, host) + for resolver in settings.DOH_RESOLVERS.split(",") + ] for future in concurrent.futures.as_completed(futures): ip = future.result() if ip is not None: @@ -60,11 +78,9 @@ def enable_doh(enable: bool) -> None: break return _orig_getaddrinfo(host, *args, **kwargs) - if enable: - # 替换 socket.getaddrinfo 方法 - socket.getaddrinfo = _patched_getaddrinfo - else: - socket.getaddrinfo = _orig_getaddrinfo + with _executor_lock: + _doh_enabled = enable + socket.getaddrinfo = _patched_getaddrinfo if enable else _orig_getaddrinfo class DohHelper(ConfigReloadMixin, metaclass=Singleton): @@ -77,14 +93,31 @@ class DohHelper(ConfigReloadMixin, metaclass=Singleton): enable_doh(settings.DOH_ENABLE) def on_config_changed(self) -> None: + if not settings.DOH_ENABLE: + self.shutdown() + return with _doh_lock: # DOH配置有变动的情况下，清空缓存 _doh_cache.clear() - enable_doh(settings.DOH_ENABLE) + enable_doh(True) def get_reload_name(self) -> str: return 'DoH' + def shutdown(self) -> None: + """恢复系统 DNS 并释放 DoH 查询线程池""" + global _executor, _doh_enabled + with _executor_lock: + _doh_enabled = False + socket.getaddrinfo = _orig_getaddrinfo + executor = _executor + _executor = None + with _doh_lock: + _doh_cache.clear() + if executor: + executor.shutdown(wait=True) + + def _doh_query(resolver: str, host: str) -> Optional[str]: """ 使用给定的DoH解析器查询给定主机的IP地址。 diff --git a/app/helper/message.py b/app/helper/message.py index 3fb9be12a..e185dde99 100644 --- a/app/helper/message.py +++ b/app/helper/message.py @@ -605,6 +605,7 @@ class MessageQueueManager(metaclass=SingletonClass): self.check_interval = check_interval self._running = True + self._stop_event = threading.Event() self.thread = threading.Thread(target=self._monitor_loop, daemon=True) self.thread.start() @@ -752,13 +753,15 @@ class MessageQueueManager(metaclass=SingletonClass): logger.info(f"队列剩余消息：{self.queue.qsize()}") except queue.Empty: break - time.sleep(self.check_interval) + if self._stop_event.wait(self.check_interval): + break def stop(self) -> None: """ 停止队列管理器 """ self._running = False + self._stop_event.set() logger.info("正在停止消息队列...") self.thread.join() logger.info("消息队列已停止") @@ -841,7 +844,8 @@ def stop_message(): """ 停止消息服务 """ - # 停止消息队列 - MessageQueueManager().stop() - # 关闭消息演染器 - TemplateHelper().close() + # 只关闭已启动的服务，避免清理路径反向创建后台线程和缓存 + if queue_manager := MessageQueueManager.get_existing_instance(): + queue_manager.stop() + if template_helper := TemplateHelper.get_existing_instance(): + template_helper.close() diff --git a/app/log.py b/app/log.py index 7a922dc8b..85957b57f 100644 --- a/app/log.py +++ b/app/log.py @@ -124,7 +124,7 @@ class NonBlockingFileHandler: """ _instance = None _lock = threading.Lock() - _rotating_handlers = {} + _stop_sentinel = object() def __new__(cls): if cls._instance is None: @@ -138,6 +138,9 @@ class NonBlockingFileHandler: return self._initialized = True + self._state_lock = threading.RLock() + self._handlers_lock = threading.Lock() + self._rotating_handlers = {} self._write_queue = queue.Queue(maxsize=log_settings.ASYNC_FILE_QUEUE_SIZE) self._executor = ThreadPoolExecutor(max_workers=log_settings.ASYNC_FILE_WORKERS, thread_name_prefix="LogWriter") @@ -151,27 +154,28 @@ class NonBlockingFileHandler: """ 获取或创建RotatingFileHandler实例 """ - if file_path not in self._rotating_handlers: - # 确保目录存在 - file_path.parent.mkdir(parents=True, exist_ok=True) + with self._handlers_lock: + if file_path not in self._rotating_handlers: + # 确保目录存在 + file_path.parent.mkdir(parents=True, exist_ok=True) - # 创建RotatingFileHandler - handler = RotatingFileHandler( - filename=str(file_path), - maxBytes=log_settings.LOG_MAX_FILE_SIZE_BYTES, - backupCount=log_settings.LOG_BACKUP_COUNT, - encoding='utf-8' - ) + # 创建RotatingFileHandler + handler = RotatingFileHandler( + filename=str(file_path), + maxBytes=log_settings.LOG_MAX_FILE_SIZE_BYTES, + backupCount=log_settings.LOG_BACKUP_COUNT, + encoding='utf-8' + ) - # 设置格式化器 - formatter = logging.Formatter(log_settings.LOG_FILE_FORMAT) - handler.setFormatter(formatter) + # 设置格式化器 + formatter = logging.Formatter(log_settings.LOG_FILE_FORMAT) + handler.setFormatter(formatter) - self._rotating_handlers[file_path] = handler + self._rotating_handlers[file_path] = handler - return self._rotating_handlers[file_path] + return self._rotating_handlers[file_path] - def write_log(self, level: str, message: str, file_path: Path): + def write_log(self, level: str, message: str, file_path: Path) -> None: """ 写入日志 - 自动检测协程环境并使用合适的方式 """ @@ -181,8 +185,11 @@ class NonBlockingFileHandler: if self._is_in_event_loop(): # 在协程环境中，使用非阻塞方式 self._write_non_blocking(entry) - else: - # 不在协程环境中，直接同步写入 + return + with self._state_lock: + if not self._running: + return + # 不在协程环境中，持锁同步写入，避免关闭文件处理器时仍有写操作进行 self._write_sync(entry) @staticmethod @@ -196,15 +203,19 @@ class NonBlockingFileHandler: except RuntimeError: return False - def _write_non_blocking(self, entry: LogEntry): + def _write_non_blocking(self, entry: LogEntry) -> bool: """ 非阻塞写入（用于协程环境） """ - try: - self._write_queue.put_nowait(entry) - except queue.Full: - # 队列满时，使用线程池处理 - self._executor.submit(self._write_sync, entry) + with self._state_lock: + if not self._running: + return False + try: + self._write_queue.put_nowait(entry) + except queue.Full: + # 队列满时，使用线程池处理 + self._executor.submit(self._write_sync, entry) + return True @staticmethod def _write_sync(entry: LogEntry): @@ -215,8 +226,7 @@ class NonBlockingFileHandler: # 获取RotatingFileHandler实例 handler = NonBlockingFileHandler()._get_rotating_handler(entry.file_path) - # 使用RotatingFileHandler的emit方法，只传递原始消息 - handler.emit(logging.LogRecord( + handler.handle(logging.LogRecord( name='', level=getattr(logging, entry.level.upper(), logging.INFO), pathname='', @@ -235,22 +245,28 @@ class NonBlockingFileHandler: """ 后台批量写入线程 """ - while self._running: + while True: try: # 收集一批日志条目 batch = [] + should_stop = False end_time = time.time() + log_settings.WRITE_TIMEOUT while len(batch) < log_settings.BATCH_WRITE_SIZE and time.time() < end_time: try: remaining_time = max(0, end_time - time.time()) entry = self._write_queue.get(timeout=remaining_time) + if entry is self._stop_sentinel: + should_stop = True + break batch.append(entry) except queue.Empty: break if batch: self._write_batch(batch) + if should_stop: + break except Exception as e: print(f"批量写入线程错误: {e}") @@ -275,8 +291,7 @@ class NonBlockingFileHandler: # 批量写入 for entry in entries: - # 使用RotatingFileHandler的emit方法，只传递原始消息 - handler.emit(logging.LogRecord( + handler.handle(logging.LogRecord( name='', level=getattr(logging, entry.level.upper(), logging.INFO), pathname='', @@ -294,15 +309,23 @@ class NonBlockingFileHandler: def shutdown(self): """ - 关闭文件处理器 + 排空异步日志并关闭文件处理器 """ - self._running = False - if hasattr(self, '_write_thread'): - self._write_thread.join(timeout=5) + with self._state_lock: + if not self._running: + return + self._running = False + if hasattr(self, '_write_thread') and self._write_thread.is_alive(): + # 状态锁保证停止标记之后不会再有生产者入队 + self._write_queue.put(self._stop_sentinel) + if hasattr(self, '_write_thread') and self._write_thread.is_alive(): + self._write_thread.join() if self._executor: self._executor.shutdown(wait=True) - # 清理缓存 + for handler in self._rotating_handlers.values(): + handler.flush() + handler.close() self._rotating_handlers.clear() diff --git a/app/modules/telegram/telegram.py b/app/modules/telegram/telegram.py index 4d1b2b402..5eefb533d 100644 --- a/app/modules/telegram/telegram.py +++ b/app/modules/telegram/telegram.py @@ -279,8 +279,6 @@ class Telegram: @staticmethod def _telegramify_item_text(item: Text) -> str: """将 telegramify 文本片段转换为 Telegram MarkdownV2 字符串。""" - if hasattr(item, "content"): - return item.content if entities_to_markdownv2: return entities_to_markdownv2(item.text, item.entities) return standardize(item.text) @@ -290,8 +288,6 @@ class Telegram: """将 telegramify 文本或媒体片段转换为 Telegram MarkdownV2 caption。""" if isinstance(item, Text): return Telegram._telegramify_item_text(item) - if hasattr(item, "caption"): - return item.caption if entities_to_markdownv2: return entities_to_markdownv2(item.caption_text, item.caption_entities) return standardize(item.caption_text) diff --git a/app/startup/lifecycle.py b/app/startup/lifecycle.py index 0ac7f7d1b..a40d1525b 100644 --- a/app/startup/lifecycle.py +++ b/app/startup/lifecycle.py @@ -20,6 +20,7 @@ from app.chain.system import SystemChain from app.core.config import global_vars, settings from app.helper.server import MoviePilotServerHelper from app.helper.system import SystemHelper +from app.log import LoggerManager from app.startup.command_initializer import init_command, stop_command, restart_command from app.startup.modules_initializer import init_modules, stop_modules from app.startup.monitor_initializer import stop_monitor, init_monitor @@ -97,20 +98,24 @@ async def lifespan(app: FastAPI): pass except Exception as e: print(str(e)) - if not settings.MOVIEPILOT_SAFE_MODE: - # 备份插件 - SystemChain().backup_plugins() - # 停止工作流 - stop_workflow() - # 停止命令 - stop_command() - # 停止监控器 - stop_monitor() - # 停止定时器 - stop_scheduler() - # 停止插件 - stop_plugins() - # 停止模块 - await stop_modules() - # 关闭共享的异步 HTTP 连接池，释放底层连接资源 - await aclose_shared_async_transports() + try: + if not settings.MOVIEPILOT_SAFE_MODE: + # 备份插件 + SystemChain().backup_plugins() + # 停止工作流 + stop_workflow() + # 停止命令 + stop_command() + # 停止监控器 + stop_monitor() + # 停止定时器 + stop_scheduler() + # 停止插件 + stop_plugins() + # 停止模块 + await stop_modules() + # 关闭共享的异步 HTTP 连接池，释放底层连接资源 + await aclose_shared_async_transports() + finally: + # 日志最后关闭，确保其他组件的收尾信息已写入文件 + LoggerManager.shutdown() diff --git a/app/startup/modules_initializer.py b/app/startup/modules_initializer.py index c66ff5d2b..840cfd3db 100644 --- a/app/startup/modules_initializer.py +++ b/app/startup/modules_initializer.py @@ -137,6 +137,8 @@ async def stop_modules(): EventManager().stop() # 停止虚拟显示 DisplayHelper().stop() + # 停止 DoH 服务 + DohHelper().shutdown() # 停止线程池 ThreadHelper().shutdown() # 停止消息服务 diff --git a/app/testing/network_guard.py b/app/testing/network_guard.py index 9729e9601..5e510cb7f 100644 --- a/app/testing/network_guard.py +++ b/app/testing/network_guard.py @@ -9,6 +9,8 @@ fixture 一并识别，autouse 自动作用于每个用例，无需逐用例改 """ from __future__ import annotations +import ipaddress + import pytest # 本地回环/通配地址放行，其余主机一律视为真实出站；getaddrinfo 的 host 可能为 str 或 bytes @@ -20,21 +22,45 @@ def block_real_network(monkeypatch): """防御纵深：拦截对非本地主机的真实出站，强制测试零真实网络。 补在各用例自身 mock 之上：某用例万一漏 mock 外部依赖（TMDB / LLM 目录 / 下载器 / - 媒体服务器 / 任意外链），其真实 DNS 解析会在此被拦并报错，而非静默发请求。本地回环放行 - （sqlite 等）。asyncio 默认解析器经线程池调用 ``socket.getaddrinfo``，故拦此一处即覆盖 - 同步与异步出站。``monkeypatch`` 在用例结束后自动还原，不影响其他用例与进程退出。 + 媒体服务器 / 任意外链），其 DNS 解析或 socket 连接会被拦截。本地回环放行（sqlite 等）。 + 所有拦截记录会在用例收尾再次断言，避免业务代码捕获网络异常后让漏 mock 的用例静默通过。 + ``monkeypatch`` 在用例结束后自动还原，不影响其他用例与进程退出。 """ import socket _real_getaddrinfo = socket.getaddrinfo + _real_connect = socket.socket.connect + attempts = [] + + def _is_allowed_host(host) -> bool: + normalized = host.decode() if isinstance(host, (bytes, bytearray)) else host + if normalized is None or normalized in _ALLOWED_NETWORK_HOSTS: + return True + try: + address = ipaddress.ip_address(str(normalized).split("%", 1)[0]) + return address.is_loopback or address.is_unspecified + except ValueError: + return False + + def _blocked(operation: str, host): + attempts.append((operation, host)) + raise RuntimeError( + f"测试禁止真实出站网络：尝试通过 {operation} 访问 {host!r}；请 mock 对应外部依赖" + ) 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 对应外部依赖" - ) + if not _is_allowed_host(host): + _blocked("DNS", host) return _real_getaddrinfo(host, *args, **kwargs) + def _guarded_connect(sock, address): + if isinstance(address, tuple) and address and not _is_allowed_host(address[0]): + _blocked("socket", address[0]) + return _real_connect(sock, address) + monkeypatch.setattr(socket, "getaddrinfo", _guarded_getaddrinfo) + monkeypatch.setattr(socket.socket, "connect", _guarded_connect) yield + if attempts: + details = ", ".join(f"{operation}:{host}" for operation, host in attempts) + pytest.fail(f"测试期间发生真实出站网络尝试：{details}") diff --git a/app/utils/singleton.py b/app/utils/singleton.py index 15503ab34..d98b97647 100644 --- a/app/utils/singleton.py +++ b/app/utils/singleton.py @@ -10,6 +10,11 @@ class Singleton(abc.ABCMeta, type): _instances: dict = {} + def get_existing_instance(cls, *args, **kwargs): + """按相同参数返回已创建实例，不触发初始化""" + key = (cls, args, frozenset(kwargs.items())) + return cls._instances.get(key) + def __call__(cls, *args, **kwargs): key = (cls, args, frozenset(kwargs.items())) if key not in cls._instances: @@ -31,6 +36,10 @@ class SingletonClass(abc.ABCMeta, type): _instances: dict = {} + def get_existing_instance(cls): + """返回已创建实例，不触发初始化""" + return cls._instances.get(cls) + def __call__(cls, *args, **kwargs): if cls not in cls._instances: cls._instances[cls] = super().__call__(*args, **kwargs) diff --git a/pytest.ini b/pytest.ini index 85c4473a0..4ec1fb1bd 100644 --- a/pytest.ini +++ b/pytest.ini @@ -7,9 +7,5 @@ timeout_method = thread # 让本仓自身的新告警更醒目。本仓代码引发的告警一律不在此忽略，应在源码/用例处修复。 filterwarnings = ignore:datetime.datetime.utcfromtimestamp\(\) is deprecated:DeprecationWarning - ignore:websockets.legacy is deprecated:DeprecationWarning - ignore:websockets.InvalidStatusCode is deprecated:DeprecationWarning - ignore:pkg_resources is deprecated as an API:DeprecationWarning - ignore:Deprecated call to .pkg_resources.declare_namespace:DeprecationWarning ignore:'crypt' is deprecated:DeprecationWarning ignore:'audioop' is deprecated:DeprecationWarning diff --git a/tests/conftest.py b/tests/conftest.py index 78a67f4d4..573bf9738 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -16,9 +16,11 @@ prepare_backend() from app.testing.network_guard import block_real_network # noqa: E402,F401 -def _report_session_cleanup_error(name: str, err: Exception) -> None: - """测试收尾清理失败只记录诊断，不覆盖原始 pytest 退出状态。""" +def _report_session_cleanup_error(session, name: str, err: Exception) -> None: + """记录收尾错误；原测试绿色时将会话标记为失败。""" sys.stderr.write(f"\npytest session cleanup failed: {name}: {err!r}\n") + if session.exitstatus == 0: + session.exitstatus = 1 def pytest_sessionfinish(session, exitstatus): @@ -28,21 +30,27 @@ def pytest_sessionfinish(session, exitstatus): shutdown_blocking_executors(cancel_futures=True) except Exception as err: - _report_session_cleanup_error("agent blocking executors", err) + _report_session_cleanup_error(session, "agent blocking executors", err) try: from app.helper.thread import ThreadHelper - from app.utils.singleton import Singleton - helper = Singleton._instances.get((ThreadHelper, (), frozenset())) + helper = ThreadHelper.get_existing_instance() if helper: helper.shutdown() except Exception as err: - _report_session_cleanup_error("thread helper", err) + _report_session_cleanup_error(session, "thread helper", err) + + try: + from app.helper.message import stop_message + + stop_message() + except Exception as err: + _report_session_cleanup_error(session, "message service", err) try: from app.log import LoggerManager LoggerManager.shutdown() except Exception as err: - _report_session_cleanup_error("logger manager", err) + _report_session_cleanup_error(session, "logger manager", err) diff --git a/tests/test_doh_helper.py b/tests/test_doh_helper.py index 544483c9c..07c534699 100644 --- a/tests/test_doh_helper.py +++ b/tests/test_doh_helper.py @@ -3,6 +3,61 @@ import socket from app.helper import doh +def test_doh_executor_is_lazy_and_shutdown_restores_socket(monkeypatch): + """DoH 线程池按需创建，并在模块关闭时恢复系统 DNS""" + original_getaddrinfo = socket.getaddrinfo + helper = object.__new__(doh.DohHelper) + monkeypatch.setattr(doh.settings, "DOH_DOMAINS", "example.com") + monkeypatch.setattr(doh.settings, "DOH_RESOLVERS", "resolver.test") + monkeypatch.setattr(doh, "_doh_query", lambda resolver, host: "203.0.113.7") + monkeypatch.setattr(doh, "_orig_getaddrinfo", lambda host, *args, **kwargs: []) + + try: + helper.shutdown() + assert doh._executor is None + + doh.enable_doh(True) + socket.getaddrinfo("example.com", None) + executor = doh._executor + assert executor is not None + + helper.shutdown() + + assert doh._executor is None + assert socket.getaddrinfo is doh._orig_getaddrinfo + assert getattr(executor, "_shutdown", False) + finally: + helper.shutdown() + socket.getaddrinfo = original_getaddrinfo + + +def test_doh_config_reload_disables_and_closes_executor(monkeypatch): + """热更新关闭 DoH 时恢复系统 DNS 并释放已创建的线程池""" + original_getaddrinfo = socket.getaddrinfo + helper = object.__new__(doh.DohHelper) + monkeypatch.setattr(doh.settings, "DOH_DOMAINS", "example.com") + monkeypatch.setattr(doh.settings, "DOH_RESOLVERS", "resolver.test") + monkeypatch.setattr(doh, "_doh_query", lambda resolver, host: "203.0.113.7") + monkeypatch.setattr(doh, "_orig_getaddrinfo", lambda host, *args, **kwargs: []) + + try: + helper.shutdown() + doh.enable_doh(True) + socket.getaddrinfo("example.com", None) + executor = doh._executor + assert executor is not None + monkeypatch.setattr(doh.settings, "DOH_ENABLE", False) + + helper.on_config_changed() + + assert doh._executor is None + assert getattr(executor, "_shutdown", False) + assert socket.getaddrinfo is doh._orig_getaddrinfo + finally: + helper.shutdown() + socket.getaddrinfo = original_getaddrinfo + + def test_enable_doh_reuses_cached_host_resolution(monkeypatch): """ 同一 DoH 域名第二次解析应命中缓存，避免重复请求远端解析器。 @@ -33,6 +88,7 @@ def test_enable_doh_reuses_cached_host_resolution(monkeypatch): socket.getaddrinfo("example.com", None) socket.getaddrinfo("example.com", None) finally: + object.__new__(doh.DohHelper).shutdown() socket.getaddrinfo = original_getaddrinfo with doh._doh_lock: doh._doh_cache.clear() diff --git a/tests/test_emby_dashboard_links.py b/tests/test_emby_dashboard_links.py index ba886cf70..1b65bdaa2 100644 --- a/tests/test_emby_dashboard_links.py +++ b/tests/test_emby_dashboard_links.py @@ -102,6 +102,7 @@ class EmbyDashboardLinksTest(unittest.TestCase): with ( patch.object(client, "_Emby__get_emby_librarys") as librarys, patch.object(client, "_Emby__get_local_image_by_id") as image_by_id, + patch.object(client, "get_items_count", return_value=0), ): librarys.return_value = [ { diff --git a/tests/test_lifecycle_shutdown.py b/tests/test_lifecycle_shutdown.py new file mode 100644 index 000000000..c984a3f5e --- /dev/null +++ b/tests/test_lifecycle_shutdown.py @@ -0,0 +1,46 @@ +import asyncio +from unittest.mock import AsyncMock, MagicMock + +import pytest +from fastapi import FastAPI + +from app.startup import lifecycle + + +def test_lifespan_closes_logger_when_early_shutdown_step_fails(monkeypatch): + """前置关闭步骤失败时仍应关闭 Logger""" + monkeypatch.setattr(lifecycle.settings, "MOVIEPILOT_SAFE_MODE", False) + monkeypatch.setattr(lifecycle.global_vars, "set_loop", MagicMock()) + for name in ( + "init_routers", + "init_modules", + "init_plugins", + "init_scheduler", + "init_monitor", + "init_command", + "init_workflow", + "stop_workflow", + "stop_command", + "stop_monitor", + "stop_scheduler", + "stop_plugins", + ): + monkeypatch.setattr(lifecycle, name, MagicMock()) + + system_chain = MagicMock() + system_chain.backup_plugins.side_effect = RuntimeError("backup failed") + monkeypatch.setattr(lifecycle, "SystemChain", MagicMock(return_value=system_chain)) + monkeypatch.setattr(lifecycle, "init_extra", AsyncMock()) + monkeypatch.setattr(lifecycle, "stop_modules", AsyncMock()) + monkeypatch.setattr(lifecycle, "aclose_shared_async_transports", AsyncMock()) + logger_shutdown = MagicMock() + monkeypatch.setattr(lifecycle.LoggerManager, "shutdown", logger_shutdown) + + async def run_lifespan(): + with pytest.raises(RuntimeError, match="backup failed"): + async with lifecycle.lifespan(FastAPI()): + pass + + asyncio.run(run_lifespan()) + + logger_shutdown.assert_called_once_with() diff --git a/tests/test_log_shutdown.py b/tests/test_log_shutdown.py new file mode 100644 index 000000000..8a2ddfbb5 --- /dev/null +++ b/tests/test_log_shutdown.py @@ -0,0 +1,149 @@ +import threading +import time +from unittest.mock import MagicMock + +from app.log import LogEntry, NonBlockingFileHandler, log_settings + + +def test_non_blocking_file_handler_shutdown_wakes_writer_and_closes_handlers(tmp_path): + """日志关闭应立即唤醒空闲写线程，并关闭所有已打开的文件处理器""" + original_instance = NonBlockingFileHandler._instance + NonBlockingFileHandler._instance = None + handler = NonBlockingFileHandler() + handler._rotating_handlers = {} + log_handler = handler._get_rotating_handler(tmp_path / "shutdown.log") + + try: + started_at = time.monotonic() + handler.shutdown() + elapsed = time.monotonic() - started_at + + assert elapsed < 1 + assert not handler._write_thread.is_alive() + assert log_handler.stream is None + assert handler._write_non_blocking( + LogEntry("info", "late-message", tmp_path / "shutdown.log") + ) is False + assert handler._write_queue.empty() + finally: + if handler._write_thread.is_alive(): + handler._running = False + handler._write_thread.join(timeout=5) + if log_handler.stream is not None: + log_handler.close() + NonBlockingFileHandler._instance = original_instance + + +def test_non_blocking_file_handler_shutdown_drains_queued_batches(monkeypatch, tmp_path): + """停止标记之前已进入队列的日志应跨批次全部写完""" + original_instance = NonBlockingFileHandler._instance + NonBlockingFileHandler._instance = None + monkeypatch.setattr(log_settings, "BATCH_WRITE_SIZE", 2) + handler = NonBlockingFileHandler() + handler._rotating_handlers = {} + written = [] + monkeypatch.setattr( + handler, + "_write_batch", + lambda batch: written.extend(entry.message for entry in batch), + ) + + try: + for index in range(5): + handler._write_non_blocking( + LogEntry("info", f"message-{index}", tmp_path / "drain.log") + ) + + handler.shutdown() + + assert written == [f"message-{index}" for index in range(5)] + assert not handler._write_thread.is_alive() + finally: + if handler._write_thread.is_alive(): + handler._running = False + handler._write_queue.put(handler._stop_sentinel) + handler._write_thread.join(timeout=5) + NonBlockingFileHandler._instance = original_instance + + +def test_non_blocking_file_handler_creates_one_handler_for_concurrent_first_write(monkeypatch, tmp_path): + """同一路径首次并发写入时只创建并关闭一个文件处理器""" + original_instance = NonBlockingFileHandler._instance + NonBlockingFileHandler._instance = None + handler = NonBlockingFileHandler() + handler._rotating_handlers = {} + first_created = threading.Event() + second_started = threading.Event() + release_first = threading.Event() + created_handlers = [] + results = [] + + class ProbeHandler: + def __init__(self, **kwargs): + self.closed = False + created_handlers.append(self) + if len(created_handlers) == 1: + first_created.set() + release_first.wait(timeout=2) + + @staticmethod + def setFormatter(formatter): + pass + + @staticmethod + def flush(): + pass + + def close(self): + self.closed = True + + monkeypatch.setattr("app.log.RotatingFileHandler", ProbeHandler) + file_path = tmp_path / "concurrent.log" + + def get_handler(started=None): + if started: + started.set() + results.append(handler._get_rotating_handler(file_path)) + + first = threading.Thread(target=get_handler) + second = threading.Thread(target=get_handler, args=(second_started,)) + try: + first.start() + assert first_created.wait(timeout=1) + second.start() + assert second_started.wait(timeout=1) + time.sleep(0.05) + release_first.set() + first.join(timeout=2) + second.join(timeout=2) + + assert len(created_handlers) == 1 + assert results[0] is results[1] + + handler.shutdown() + assert created_handlers[0].closed is True + finally: + release_first.set() + first.join(timeout=2) + second.join(timeout=2) + handler.shutdown() + NonBlockingFileHandler._instance = original_instance + + +def test_non_blocking_file_handler_uses_handler_lock(monkeypatch, tmp_path): + """日志写入通过 Handler 入口串行化 emit 与 rollover""" + original_instance = NonBlockingFileHandler._instance + NonBlockingFileHandler._instance = None + handler = NonBlockingFileHandler() + handler._rotating_handlers = {} + log_handler = MagicMock() + monkeypatch.setattr(handler, "_get_rotating_handler", MagicMock(return_value=log_handler)) + + try: + handler._write_sync(LogEntry("info", "message", tmp_path / "locked.log")) + + log_handler.handle.assert_called_once() + log_handler.emit.assert_not_called() + finally: + handler.shutdown() + NonBlockingFileHandler._instance = original_instance diff --git a/tests/test_media_interaction.py b/tests/test_media_interaction.py index c777ce669..3648ea311 100644 --- a/tests/test_media_interaction.py +++ b/tests/test_media_interaction.py @@ -20,6 +20,16 @@ def clear_media_interactions(): plugin_input_interaction_manager.clear() +@pytest.fixture(autouse=True) +def mock_default_media_search(): + """未显式验证搜索结果的消息路由用例不访问真实媒体元数据服务""" + with patch( + "app.chain.media.MediaChain.search", + side_effect=lambda title: (_build_meta(title), []), + ): + yield + + def _build_meta(name: str) -> MetaBase: """构造媒体识别元数据。""" meta = MetaBase(name) diff --git a/tests/test_message_queue_shutdown.py b/tests/test_message_queue_shutdown.py new file mode 100644 index 000000000..9d37a1b2a --- /dev/null +++ b/tests/test_message_queue_shutdown.py @@ -0,0 +1,30 @@ +import time + +from app.helper.message import MessageQueueManager, TemplateHelper, stop_message +from app.utils.singleton import SingletonClass + + +def test_message_queue_stop_wakes_idle_monitor(monkeypatch): + """消息队列停止时应唤醒空闲监控线程，不等待完整检查周期""" + monkeypatch.setattr(MessageQueueManager, "init_config", lambda self: None) + manager = object.__new__(MessageQueueManager) + manager.__init__(check_interval=10) + + started_at = time.monotonic() + manager.stop() + elapsed = time.monotonic() - started_at + + assert elapsed < 1 + assert not manager.thread.is_alive() + + +def test_stop_message_does_not_initialize_absent_services(monkeypatch): + """消息服务未初始化时，关闭入口不应为了清理而创建后台资源""" + monkeypatch.setattr(SingletonClass, "_instances", {}) + + assert MessageQueueManager.get_existing_instance() is None + assert TemplateHelper.get_existing_instance() is None + stop_message() + + assert MessageQueueManager not in SingletonClass._instances + assert TemplateHelper not in SingletonClass._instances diff --git a/tests/test_network_guard.py b/tests/test_network_guard.py new file mode 100644 index 000000000..cc1ed7526 --- /dev/null +++ b/tests/test_network_guard.py @@ -0,0 +1,22 @@ +import socket + +import pytest + +from app.testing.network_guard import block_real_network + + +def test_network_guard_fails_when_blocked_attempt_is_swallowed(monkeypatch): + """业务代码即使捕获网络异常，网络守卫仍应在用例收尾报告失败""" + fixture = block_real_network.__wrapped__(monkeypatch) + next(fixture) + + try: + try: + socket.getaddrinfo("external.example", 443) + except RuntimeError: + pass + + with pytest.raises(pytest.fail.Exception, match="external.example"): + next(fixture) + finally: + monkeypatch.undo() diff --git a/tests/test_singleton.py b/tests/test_singleton.py new file mode 100644 index 000000000..25e6e4809 --- /dev/null +++ b/tests/test_singleton.py @@ -0,0 +1,29 @@ +from app.utils.singleton import Singleton, SingletonClass + + +def test_singleton_class_can_read_existing_instance_without_creating(monkeypatch): + """按类单例可以只读取已存在实例""" + + class Example(metaclass=SingletonClass): + pass + + monkeypatch.setattr(SingletonClass, "_instances", {}) + + assert Example.get_existing_instance() is None + instance = Example() + assert Example.get_existing_instance() is instance + + +def test_parameterized_singleton_can_read_matching_instance_without_creating(monkeypatch): + """参数化单例按相同参数读取已存在实例""" + + class Example(metaclass=Singleton): + def __init__(self, name): + self.name = name + + monkeypatch.setattr(Singleton, "_instances", {}) + + assert Example.get_existing_instance("first") is None + instance = Example("first") + assert Example.get_existing_instance("first") is instance + assert Example.get_existing_instance("second") is None diff --git a/tests/test_subscribe_endpoint.py b/tests/test_subscribe_endpoint.py index 2e34d8b56..ae1c09606 100644 --- a/tests/test_subscribe_endpoint.py +++ b/tests/test_subscribe_endpoint.py @@ -362,7 +362,7 @@ class SubscribeEndpointTest(TestCase): with patch( "app.api.endpoints.subscribe.Subscribe.async_list_by_username", new=AsyncMock(return_value=owned), - ): + ), patch("app.api.endpoints.subscribe.Scheduler") as scheduler_cls: response = asyncio.run( search_subscribes( background_tasks=background_tasks, @@ -376,6 +376,7 @@ class SubscribeEndpointTest(TestCase): [task["kwargs"]["sid"] for task in background_tasks.tasks], [17, 18], ) + self.assertEqual(scheduler_cls.return_value.start.call_count, 0) def test_subscribe_files_hides_other_user_row(self): """ diff --git a/tests/test_telegram.py b/tests/test_telegram.py index 1d142d0ef..6b9cecb6c 100644 --- a/tests/test_telegram.py +++ b/tests/test_telegram.py @@ -3,6 +3,7 @@ Telegram 模块单元测试（pytest 原生）。 """ import json +import warnings from types import SimpleNamespace from unittest.mock import MagicMock, Mock, patch @@ -286,13 +287,29 @@ def test_send_msg_markdown_escaping(telegram): assert send_kwargs["text"].startswith("*测试标题*\n") -def test_telegramify_new_content_fields_are_used_directly(): - """新版telegramify对象应直接使用已渲染的MarkdownV2字段""" - text_item = SimpleNamespace(content="已转义\\_文本") - file_item = SimpleNamespace(caption="已转义\\_说明") +def test_telegramify_current_fields_are_used_directly(): + """telegramify 对象直接使用当前 MarkdownV2 字段""" + from telegramify_markdown.content import ContentTrace, File, Text - assert Telegram._telegramify_item_text(text_item) == "已转义\\_文本" - assert Telegram._telegramify_item_caption(file_item) == "已转义\\_说明" + text_item = Text( + text="已转义_文本", + entities=[], + content_trace=ContentTrace(source_type="test"), + ) + file_item = File( + file_name="test.txt", + file_data=b"test", + caption_text="已转义_说明", + caption_entities=[], + content_trace=ContentTrace(source_type="test"), + ) + + with warnings.catch_warnings(record=True) as warning_records: + warnings.simplefilter("always") + assert Telegram._telegramify_item_text(text_item) == "已转义\\_文本" + assert Telegram._telegramify_item_caption(file_item) == "已转义\\_说明" + + assert not warning_records def test_send_msg_with_html_parse_mode_keeps_html(telegram): diff --git a/tests/test_web_agent_stream.py b/tests/test_web_agent_stream.py index f7d033c5d..617bac334 100644 --- a/tests/test_web_agent_stream.py +++ b/tests/test_web_agent_stream.py @@ -183,7 +183,8 @@ def test_build_web_agent_command_items_returns_slash_commands(): def test_build_web_agent_command_items_includes_sites_command(): """WebAgent 命令建议应包含内建站点管理命令。""" - commands = _build_web_agent_command_items() + with patch("app.command.Scheduler"), patch("app.command.ThreadHelper"): + commands = _build_web_agent_command_items() assert any(command["command"] == "/sites" for command in commands)