diff --git a/app/adapters/cache/redis.py b/app/adapters/cache/redis.py index 115c7397e..2166fd130 100644 --- a/app/adapters/cache/redis.py +++ b/app/adapters/cache/redis.py @@ -91,6 +91,7 @@ class RedisHelper(ConfigReloadMixin, metaclass=Singleton): "CACHE_REDIS_MAXMEMORY", "CACHE_REDIS_MAX_CONNECTIONS", "CACHE_REDIS_POOL_TIMEOUT", + "BIG_MEMORY_MODE", } def __init__(self): @@ -370,6 +371,7 @@ class AsyncRedisHelper(ConfigReloadMixin, metaclass=Singleton): "CACHE_REDIS_MAXMEMORY", "CACHE_REDIS_MAX_CONNECTIONS", "CACHE_REDIS_POOL_TIMEOUT", + "BIG_MEMORY_MODE", } def __init__(self): diff --git a/app/application/chain/context.py b/app/application/chain/context.py index 46ebc8f98..4bfcab71d 100644 --- a/app/application/chain/context.py +++ b/app/application/chain/context.py @@ -84,6 +84,7 @@ class ChainRuntimeContext: configuration: ChainRuntimeConfig = field( default_factory=lambda: ChainRuntimeConfig(media_extensions=()) ) + configuration_provider: Optional[Callable[[], ChainRuntimeConfig]] = None stop_state: StopState = field(default_factory=lambda: runtime_stop_state) system_service: Optional[SystemService] = None diff --git a/app/application/configuration.py b/app/application/configuration.py index a3f8dc24f..14a49bddd 100644 --- a/app/application/configuration.py +++ b/app/application/configuration.py @@ -2,7 +2,7 @@ from __future__ import annotations -from collections.abc import Callable +from collections.abc import Awaitable, Callable from dataclasses import dataclass from functools import partial from pathlib import Path @@ -15,6 +15,9 @@ from app.schemas.types import MediaType, SystemConfigKey SystemConfigValueNormalizer = Callable[[Any, Any], Any] """系统配置值在进入持久化端口前使用的规范化函数。""" +SystemConfigChangePublisher = Callable[[Any, Any], Awaitable[None]] +"""系统配置提交成功后的运行时变更发布函数。""" + T = TypeVar("T") @@ -350,8 +353,9 @@ class SystemConfigService: writer: SystemConfigWriter | None = None, async_executor: AsyncDatabaseExecutor | None = None, value_normalizer: SystemConfigValueNormalizer | None = None, + change_publisher: SystemConfigChangePublisher | None = None, ) -> None: - """注入读写端口、异步事务执行能力及可选值规范化边界。""" + """注入读写端口、异步事务执行能力、规范化边界和变更发布端口。""" resolved_reader = reader or repository resolved_writer = writer or repository if resolved_reader is None or resolved_writer is None: @@ -360,6 +364,24 @@ class SystemConfigService: self._writer = resolved_writer self._async_executor = async_executor self._value_normalizer = value_normalizer + self._change_publisher = change_publisher + + def configure_change_publisher( + self, + publisher: SystemConfigChangePublisher | None, + ) -> None: + """登记或清除异步配置变更发布器，供组合根绑定宿主事件总线。""" + self._change_publisher = publisher + + async def _publish_async_change( + self, + key: Any, + value: Any, + changed: bool | None, + ) -> None: + """仅在异步写入确实改变持久化值后发布运行时变更。""" + if changed is True and self._change_publisher is not None: + await self._change_publisher(key, value) def get(self, key: Any = None) -> Any: """读取配置。""" @@ -392,8 +414,10 @@ class SystemConfigService: return self._writer.increment(key, step) async def async_set(self, key: Any, value: Any) -> bool | None: - """异步写入配置，并等待数据库提交或回滚完成。""" - return (await self.async_set_with_normalized_value(key, value)).changed + """异步写入配置，并在返回前等待已登记的运行时重载完成。""" + result = await self.async_set_with_normalized_value(key, value) + await self._publish_async_change(key, result.normalized_value, result.changed) + return result.changed async def async_set_with_normalized_value( self, diff --git a/app/application/filtering.py b/app/application/filtering.py index 83dd3a9e4..d7d8dee72 100644 --- a/app/application/filtering.py +++ b/app/application/filtering.py @@ -389,18 +389,14 @@ def normalize_rule_group( async def save_system_config( key: SystemConfigKey, value: Any, - publish_config_changed: RuleConfigPublisher, ) -> Optional[bool]: - """通过统一入口保存配置并补发 ConfigChanged 事件。""" + """通过统一异步配置入口保存配置并等待运行时重载。""" normalized_value = value if isinstance(normalized_value, list): normalized_value = [item for item in normalized_value if item is not None and item != ""] normalized_value = normalized_value or None - success = await get_configured_system_config().async_set(key, normalized_value) - if success: - await publish_config_changed(key, normalized_value) - return success + return await get_configured_system_config().async_set(key, normalized_value) def replace_rule_id_in_rule_string(rule_string: str, old_rule_id: str, new_rule_id: str) -> str: @@ -499,7 +495,6 @@ class FilterRuleService: await save_system_config( SystemConfigKey.CustomFilterRules, [rule.model_dump(exclude_none=True) for rule in rules], - self._publish_config_changed, ) return { "message": f"已新增自定义过滤规则 {new_rule.id}", @@ -607,7 +602,6 @@ class FilterRuleService: await save_system_config( SystemConfigKey.CustomFilterRules, rule_definitions, - self._publish_config_changed, ) if updated.id is None: raise ValueError("更新后的自定义过滤规则缺少 id") @@ -630,7 +624,6 @@ class FilterRuleService: await save_system_config( SystemConfigKey.CustomFilterRules, [rule.model_dump(exclude_none=True) for rule in remaining], - self._publish_config_changed, ) return { "message": f"已删除自定义过滤规则 {rule_id}", diff --git a/app/application/system.py b/app/application/system.py index a22489d03..958476735 100644 --- a/app/application/system.py +++ b/app/application/system.py @@ -93,7 +93,7 @@ class ConfigurationEventPort(Protocol): """发布系统配置变更事实的端口。""" async def publish(self, key: Any, value: JsonData = None) -> None: - """发布一个已经持久化完成的配置变更。""" + """发布已持久化的配置变更，并在返回前完成同步重载处理器。""" ... @@ -235,7 +235,7 @@ class SystemService: return await self._logs.collect(name) async def update_environment(self, env: dict[str, Any]) -> SystemOperationResult: - """校验并原子执行一批可变部署设置更新。""" + """校验并执行一批可变部署设置更新。""" validation_error = self._llm.validate(env, self.settings) if validation_error: return SystemOperationResult(False, validation_error) @@ -246,6 +246,9 @@ class SystemService: success_updates = {key: value for key, value in result.items() if value[0]} failed_updates = {key: value for key, value in result.items() if value[0] is False} if failed_updates: + if success_updates: + # update_many 已经逐项提交，部分失败不能吞掉已成功配置的实时重载。 + await self._events.publish(success_updates.keys()) return SystemOperationResult( False, ", ".join(value[1] for value in failed_updates.values()), diff --git a/app/chain/base.py b/app/chain/base.py index 954e93ff0..f64729e8d 100644 --- a/app/chain/base.py +++ b/app/chain/base.py @@ -72,7 +72,8 @@ class ChainBase(RecognitionMixin, MessageProcessingMixin, NotificationMixin, met self.download_failure_repository = context.download_failure_repository self.user_repository = context.user_repository self.classification_service = context.classification_service - self.runtime_config = context.configuration + self._runtime_config_provider = context.configuration_provider + self._runtime_config = context.configuration self.stop_state = context.stop_state self.system_service = context.system_service self.durable_event_writer = context.durable_event_writer @@ -88,15 +89,25 @@ class ChainBase(RecognitionMixin, MessageProcessingMixin, NotificationMixin, met @property def runtime_config(self) -> ChainRuntimeConfig: - """返回实例快照；兼容绕过构造器的旧调用并按需取得当前快照。""" - configuration = getattr(self, "_runtime_config", None) + """返回当前运行配置；显式注入实例仍保留稳定快照语义。""" + provider = cast( + Optional[Callable[[], ChainRuntimeConfig]], + getattr(self, "_runtime_config_provider", None), + ) + if provider is not None: + return provider() + configuration = cast( + Optional[ChainRuntimeConfig], + getattr(self, "_runtime_config", None), + ) if configuration is None: return get_chain_runtime_config_snapshot() return configuration @runtime_config.setter def runtime_config(self, configuration: ChainRuntimeConfig) -> None: - """保存显式注入的 Chain 配置快照。""" + """保存显式注入的 Chain 配置快照并停止自动配置读取。""" + self._runtime_config_provider = None self._runtime_config = configuration def load_cache(self, filename: str) -> Any: diff --git a/app/chain/transfer/facade.py b/app/chain/transfer/facade.py index d83eaa870..0d4688288 100644 --- a/app/chain/transfer/facade.py +++ b/app/chain/transfer/facade.py @@ -61,6 +61,9 @@ class TransferChain( CONFIG_WATCH = { "TRANSFER_THREADS", + "RMT_MEDIAEXT", + "RMT_SUBEXT", + "RMT_AUDIOEXT", } _WORKER_RESTART_TIMEOUT_SECONDS = 30.0 diff --git a/app/chain/transfer/queue.py b/app/chain/transfer/queue.py index 4e19d6f30..ba92b19a2 100644 --- a/app/chain/transfer/queue.py +++ b/app/chain/transfer/queue.py @@ -71,16 +71,7 @@ class TransferQueueOwner(_TransferOwnerBase): ) -> None: """初始化文件整理处理链。""" super().__init__(runtime_context=runtime_context) - # 主要媒体文件后缀 - self._media_exts = self.runtime_config.video_extensions - # 字幕文件后缀 - self._subtitle_exts = self.runtime_config.subtitle_extensions - # 音频文件后缀 - self._audio_exts = self.runtime_config.audio_extensions - # 可处理的文件后缀（视频文件、字幕、音频文件和音乐歌词） - self._allowed_exts = self._media_exts + self._audio_exts + self._subtitle_exts + ( - ".lrc", ".txt", ".yaml", - ) + self._refresh_file_extensions() # 待整理任务队列 self._queue = queue.Queue() # 文件整理线程 @@ -129,6 +120,15 @@ class TransferQueueOwner(_TransferOwnerBase): # 启动整理任务 self._TransferChain__init() + def _refresh_file_extensions(self) -> None: + """按最新运行配置更新整理允许的媒体、字幕和音频后缀。""" + self._media_exts = self.runtime_config.video_extensions + self._subtitle_exts = self.runtime_config.subtitle_extensions + self._audio_exts = self.runtime_config.audio_extensions + self._allowed_exts = self._media_exts + self._audio_exts + self._subtitle_exts + ( + ".lrc", ".txt", ".yaml", + ) + def _TransferChain__init(self) -> bool: """启动一代文件整理线程，并返回是否成功取得 worker 所有权。""" with self._worker_lifecycle_lock: @@ -332,6 +332,7 @@ class TransferQueueOwner(_TransferOwnerBase): if self._closing: logger.info("文件整理链正在关闭，忽略 worker 配置热更新") return + self._refresh_file_extensions() if not self._TransferChain__stop( timeout_seconds=self._WORKER_RESTART_TIMEOUT_SECONDS ): diff --git a/app/modules/discord/capability.toml b/app/modules/discord/capability.toml index 33d7008cb..f6e9a6586 100644 --- a/app/modules/discord/capability.toml +++ b/app/modules/discord/capability.toml @@ -12,7 +12,7 @@ priority = 4 [activation] policy = "when_configured" -watch = ["Notifications"] +watch = ["Notifications", "PROXY_HOST"] [activation.selector] kind = "system_config_item" diff --git a/app/modules/emby/capability.toml b/app/modules/emby/capability.toml index e61047eaf..17bb8c599 100644 --- a/app/modules/emby/capability.toml +++ b/app/modules/emby/capability.toml @@ -12,7 +12,7 @@ priority = 1 [activation] policy = "when_configured" -watch = ["MediaServers"] +watch = ["MediaServers", "SUPERUSER"] [activation.selector] kind = "system_config_item" diff --git a/app/modules/fanart/__init__.py b/app/modules/fanart/__init__.py index e540cd828..e352d63b1 100644 --- a/app/modules/fanart/__init__.py +++ b/app/modules/fanart/__init__.py @@ -2,15 +2,14 @@ import asyncio from pathlib import Path from typing import Optional, Tuple, Union -from app.runtime.cache import cached +from app.adapters.network.http import AsyncRequestUtils, RequestUtils from app.domain.context import MediaInfo -from app.runtime.settings import get_runtime_setting - -from app.runtime.log import logger -from app.runtime.tasks import get_task_registry from app.modules import _ModuleBase +from app.runtime.cache import cached +from app.runtime.log import logger +from app.runtime.settings import get_runtime_setting +from app.runtime.tasks import get_task_registry from app.schemas.types import MediaType, ModuleType, OtherModulesType -from app.adapters.network.http import RequestUtils, AsyncRequestUtils class FanartModule(_ModuleBase): @@ -307,22 +306,12 @@ class FanartModule(_ModuleBase): } """ - # 代理 - _proxies: dict = get_runtime_setting('PROXY') - - # Fanart Api - _movie_url: str = ( - f"https://webservice.fanart.tv/v3/movies/%s?api_key={get_runtime_setting('FANART_API_KEY')}" - ) - _tv_url: str = ( - f"https://webservice.fanart.tv/v3/tv/%s?api_key={get_runtime_setting('FANART_API_KEY')}" - ) - def init_module(self) -> None: - pass + """初始化 Fanart 模块并清理旧配置 generation 的请求缓存。""" + self.clear_cache() def stop(self): - pass + """停止 Fanart 模块；请求资源由每次调用按当前配置创建。""" def test(self) -> Tuple[bool, str]: """ @@ -587,7 +576,7 @@ class FanartModule(_ModuleBase): ) -> Optional[dict]: image_url = cls.__fanart_url(media_type=media_type, queryid=queryid) try: - ret = RequestUtils(proxies=cls._proxies, timeout=10).get_res( + ret = RequestUtils(proxies=get_runtime_setting("PROXY"), timeout=10).get_res( image_url, raise_exception=True ) if ret: @@ -606,7 +595,9 @@ class FanartModule(_ModuleBase): ) -> Optional[dict]: image_url = cls.__fanart_url(media_type=media_type, queryid=queryid) try: - ret = await AsyncRequestUtils(proxies=cls._proxies, timeout=10).get_json( + ret = await AsyncRequestUtils( + proxies=get_runtime_setting("PROXY"), timeout=10 + ).get_json( image_url ) if ret: @@ -622,9 +613,10 @@ class FanartModule(_ModuleBase): """ 生成 Fanart 请求地址 """ + api_key = get_runtime_setting("FANART_API_KEY") if media_type == MediaType.MOVIE: - return cls._movie_url % queryid - return cls._tv_url % queryid + return f"https://webservice.fanart.tv/v3/movies/{queryid}?api_key={api_key}" + return f"https://webservice.fanart.tv/v3/tv/{queryid}?api_key={api_key}" def clear_cache(self) -> None: """清理同步缓存，并由宿主登记运行中事件循环的异步清理。""" diff --git a/app/modules/fanart/capability.toml b/app/modules/fanart/capability.toml index 94439c182..1c6c2c6d3 100644 --- a/app/modules/fanart/capability.toml +++ b/app/modules/fanart/capability.toml @@ -12,7 +12,7 @@ priority = 0 [activation] policy = "when_configured" -watch = ["FANART_API_KEY"] +watch = ["FANART_API_KEY", "PROXY_HOST"] [activation.selector] kind = "setting_truthy" diff --git a/app/modules/filemanager/storages/__init__.py b/app/modules/filemanager/storages/__init__.py index 6314285b9..c8d1b7904 100644 --- a/app/modules/filemanager/storages/__init__.py +++ b/app/modules/filemanager/storages/__init__.py @@ -45,7 +45,11 @@ class StorageBase(metaclass=ABCMeta): """ schema = None transtype = {} - snapshot_check_folder_modtime = True + + @property + def snapshot_check_folder_modtime(self) -> bool: + """返回存储快照目录时间检查的默认开关。""" + return True def __init__(self): self.storagehelper = StorageHelper() diff --git a/app/modules/filemanager/storages/alipan.py b/app/modules/filemanager/storages/alipan.py index 8c1bb72b6..98fc18ad1 100644 --- a/app/modules/filemanager/storages/alipan.py +++ b/app/modules/filemanager/storages/alipan.py @@ -43,9 +43,6 @@ class AliPan(StorageBase, metaclass=WeakSingleton): # 基础url base_url = "https://openapi.alipan.com" - # 阿里云盘目录时间不随子文件变更而更新，默认关闭目录修改时间检查 - snapshot_check_folder_modtime = get_runtime_setting('ALIPAN_SNAPSHOT_CHECK_FOLDER_MODTIME') - # 文件块大小，默认10MB chunk_size = 10 * 1024 * 1024 @@ -59,6 +56,11 @@ class AliPan(StorageBase, metaclass=WeakSingleton): ) self._init_session() + @property + def snapshot_check_folder_modtime(self) -> bool: + """读取当前 AliPan 目录时间检查开关，支持配置保存后立即生效。""" + return bool(get_runtime_setting('ALIPAN_SNAPSHOT_CHECK_FOLDER_MODTIME')) + def _init_session(self): """ 初始化带速率限制的会话 diff --git a/app/modules/filemanager/storages/alist.py b/app/modules/filemanager/storages/alist.py index 195eb0332..2163491bc 100644 --- a/app/modules/filemanager/storages/alist.py +++ b/app/modules/filemanager/storages/alist.py @@ -38,12 +38,14 @@ class Alist(StorageBase, metaclass=WeakSingleton): "move": "移动", } - # 快照检查目录修改时间 - snapshot_check_folder_modtime = get_runtime_setting('OPENLIST_SNAPSHOT_CHECK_FOLDER_MODTIME') - def __init__(self): super().__init__() + @property + def snapshot_check_folder_modtime(self) -> bool: + """读取当前 OpenList 目录时间检查开关，支持配置保存后立即生效。""" + return bool(get_runtime_setting('OPENLIST_SNAPSHOT_CHECK_FOLDER_MODTIME')) + def init_storage(self): """ 初始化 diff --git a/app/modules/filemanager/storages/rclone.py b/app/modules/filemanager/storages/rclone.py index ef7a18687..47057be40 100644 --- a/app/modules/filemanager/storages/rclone.py +++ b/app/modules/filemanager/storages/rclone.py @@ -48,7 +48,10 @@ class Rclone(StorageBase): "copy": "复制" } - snapshot_check_folder_modtime = get_runtime_setting('RCLONE_SNAPSHOT_CHECK_FOLDER_MODTIME') + @property + def snapshot_check_folder_modtime(self) -> bool: + """读取当前 Rclone 目录时间检查开关，支持配置保存后立即生效。""" + return bool(get_runtime_setting('RCLONE_SNAPSHOT_CHECK_FOLDER_MODTIME')) def init_storage(self): """ diff --git a/app/modules/jellyfin/capability.toml b/app/modules/jellyfin/capability.toml index 2d420fb76..ae2360d64 100644 --- a/app/modules/jellyfin/capability.toml +++ b/app/modules/jellyfin/capability.toml @@ -12,7 +12,7 @@ priority = 2 [activation] policy = "when_configured" -watch = ["MediaServers"] +watch = ["MediaServers", "SUPERUSER"] [activation.selector] kind = "system_config_item" diff --git a/app/modules/telegram/capability.toml b/app/modules/telegram/capability.toml index aae376504..7ecc4d380 100644 --- a/app/modules/telegram/capability.toml +++ b/app/modules/telegram/capability.toml @@ -12,7 +12,7 @@ priority = 0 [activation] policy = "when_configured" -watch = ["Notifications"] +watch = ["Notifications", "PROXY_HOST"] [activation.selector] kind = "system_config_item" diff --git a/app/modules/thetvdb/__init__.py b/app/modules/thetvdb/__init__.py index 36f65d3f8..e357bfcd8 100644 --- a/app/modules/thetvdb/__init__.py +++ b/app/modules/thetvdb/__init__.py @@ -36,7 +36,9 @@ class TheTvDbModule(_ModuleBase): __auth_lock = Lock() def init_module(self) -> None: - pass + """重置 TVDB 会话，使下一次请求按最新密钥、PIN 和代理建立连接。""" + with self.__auth_lock: + self.tvdb = None def _initialize_tvdb_session(self, is_retry: bool = False) -> None: """ @@ -132,7 +134,8 @@ class TheTvDbModule(_ModuleBase): """ return 4 - def stop(self): + def stop(self) -> None: + """停止当前 TVDB 会话并释放其运行时引用。""" with self.__auth_lock: self.tvdb = None @@ -147,6 +150,7 @@ class TheTvDbModule(_ModuleBase): return False, str(e) def init_setting(self) -> Tuple[str, Union[str, bool]]: + """TVDB 随宿主启动，无需额外的模块开关。""" pass def tvdb_info(self, tvdbid: int) -> Optional[dict]: diff --git a/app/modules/thetvdb/capability.toml b/app/modules/thetvdb/capability.toml index bba680f9f..b4b61a11e 100644 --- a/app/modules/thetvdb/capability.toml +++ b/app/modules/thetvdb/capability.toml @@ -12,4 +12,4 @@ priority = 4 [activation] policy = "bootstrap" -watch = [] +watch = ["PROXY_HOST", "TVDB_V4_API_KEY", "TVDB_V4_API_PIN"] diff --git a/app/monitor/dispatcher.py b/app/monitor/dispatcher.py index d830bd6ea..50e9a4dc5 100644 --- a/app/monitor/dispatcher.py +++ b/app/monitor/dispatcher.py @@ -2,7 +2,7 @@ import re import traceback from pathlib import Path from threading import Lock -from typing import Any, Callable, Dict, List, Optional, Tuple +from typing import Any, Callable, Dict, List, Optional, Tuple, cast from app.adapters.system.fsproxy import fsproxy from app.application.directory import DirectoryHelper @@ -44,11 +44,8 @@ class TransferDispatcher: :param cache: 去重缓存，默认使用 10 秒 TTL 缓存 :param retry_abandoned_callback: 自动重试最终放弃时的用户告警回调 """ - self.all_exts = all_exts if all_exts is not None else ( - get_runtime_setting('RMT_MEDIAEXT') - + get_runtime_setting('RMT_SUBEXT') - + get_runtime_setting('RMT_AUDIOEXT') - ) + # 测试和调用方显式传入的扩展名保持固定；生产实例按使用时读取当前配置。 + self._all_exts_override = all_exts self._cache = cache if cache is not None else TTLCache(region="monitor", maxsize=1024, ttl=10) self._lock = Lock() # 历史查询失败待重试的文件 @@ -56,6 +53,28 @@ class TransferDispatcher: self._pending_guard = Lock() self._retry_abandoned_callback = retry_abandoned_callback + @property + def all_exts(self) -> List[str]: + """ + 获取当前监控扩展名。 + + 监控分发器是长生命周期对象，不能在构造时固定用户后来修改的整理后缀； + 显式注入的扩展名仍作为测试和兼容调用方的固定覆盖值。 + """ + if self._all_exts_override is not None: + return self._all_exts_override + media_extensions = cast(List[str], get_runtime_setting('RMT_MEDIAEXT')) + subtitle_extensions = cast(List[str], get_runtime_setting('RMT_SUBEXT')) + audio_extensions = cast(List[str], get_runtime_setting('RMT_AUDIOEXT')) + return media_extensions + subtitle_extensions + audio_extensions + + @all_exts.setter + def all_exts(self, value: Optional[List[str]]) -> None: + """ + 设置显式监控扩展名覆盖值。 + """ + self._all_exts_override = value + @staticmethod def _is_bluray_sub(_path: Path) -> bool: """ diff --git a/app/runtime/event/dispatch.py b/app/runtime/event/dispatch.py index 26ae0d101..ea714b78a 100644 --- a/app/runtime/event/dispatch.py +++ b/app/runtime/event/dispatch.py @@ -180,6 +180,51 @@ class EventDispatcher: else: self.invoke_sync_strict(handler, isolated) + async def dispatch_broadcast_async_strict(self, event: Any) -> None: + """异步串行等待广播处理器完成，供实时配置重载入口使用。""" + handlers = self._registry.broadcast_snapshot(event.event_type) + target_plugin_id = None + if event.event_type == EventType.MessageAction and isinstance( + event.event_data, + dict, + ): + target_plugin_id = event.event_data.get("__mp_target_plugin_id") + for handler_id, handler in handlers: + if not self._registry.is_handler_enabled(handler): + continue + if target_plugin_id and not self.should_dispatch_to_target_plugin( + handler, + handler_id, + str(target_plugin_id), + ): + continue + if isinstance(event.event_data, dict): + event_data = event.event_data.copy() + event_data.pop("__mp_target_plugin_id", None) + else: + event_data = event.event_data + isolated = self._event_factory( + event_type=event.event_type, + event_data=event_data, + priority=event.priority, + correlation_id=event.correlation_id, + ) + if inspect.iscoroutinefunction(handler): + await self.invoke_async_strict( + handler, + isolated, + skip_unresolved=True, + run_sync_in_threadpool=True, + ) + else: + # 复用统一严格调用路径，保持 owner 的线程池声明和错误策略一致。 + await self.invoke_async_strict( + handler, + isolated, + skip_unresolved=True, + run_sync_in_threadpool=True, + ) + def safe_invoke_sync(self, handler: Callable, event: Any) -> None: """仅在处理器启用时执行同步调用。""" if self._registry.is_handler_enabled(handler): @@ -217,10 +262,14 @@ class EventDispatcher: self, handler: Callable[..., object], event: Any, + *, + skip_unresolved: bool = False, ) -> None: - """解析并执行同步处理器，记录错误后向 durable 调用方传播。""" + """解析并执行同步处理器，按需跳过未激活 owner，实际错误向调用方传播。""" resolved = self._binding_resolver.resolve(handler) if not resolved: + if skip_unresolved: + return raise RuntimeError("事件处理器实例不可用") method, binding, class_name, method_name = resolved with correlation_scope(event.correlation_id): @@ -273,10 +322,15 @@ class EventDispatcher: self, handler: Callable[..., object], event: Any, + *, + skip_unresolved: bool = False, + run_sync_in_threadpool: bool = False, ) -> None: - """解析并等待处理器完成，记录错误后向 durable 调用方传播。""" + """解析并等待处理器完成，按需跳过未激活 owner 或移交同步处理器线程池。""" resolved = self._binding_resolver.resolve(handler) if not resolved: + if skip_unresolved: + return raise RuntimeError("事件处理器实例不可用") method, binding, class_name, method_name = resolved with correlation_scope(event.correlation_id): @@ -288,7 +342,7 @@ class EventDispatcher: ): if inspect.iscoroutinefunction(method): await method(event) - elif binding.run_sync_in_threadpool or not class_name: + elif run_sync_in_threadpool or binding.run_sync_in_threadpool or not class_name: await run_in_threadpool(method, event) else: method(event) diff --git a/app/runtime/events.py b/app/runtime/events.py index 017fb5673..6316c0904 100644 --- a/app/runtime/events.py +++ b/app/runtime/events.py @@ -562,6 +562,20 @@ class EventManager(metaclass=Singleton): ) return event + async def async_send_event_strict( + self, + etype: EventType, + data: Optional[Union[dict[str, object], ChainEventData]] = None, + priority: Optional[int] = DEFAULT_EVENT_PRIORITY, + ) -> Event: + """异步发送广播事件并等待全部处理器完成，保证配置写入返回时已完成重载。""" + event = Event(etype, data, priority) + with self.__lifecycle_lock: + if self.__lifecycle_state != "running": + raise RuntimeError(f"事件处理处于 {self.__lifecycle_state} 状态") + await self.__dispatcher.dispatch_broadcast_async_strict(event) + return event + @staticmethod def __wait_strict_async_handler(coroutine: Any) -> Any: """在主事件循环等待异步处理器，禁止循环线程同步等待自身。""" diff --git a/app/scheduler/facade.py b/app/scheduler/facade.py index d612b8975..c9e38e51e 100644 --- a/app/scheduler/facade.py +++ b/app/scheduler/facade.py @@ -48,6 +48,7 @@ class Scheduler( CONFIG_WATCH = { "DEV", + "TZ", "COOKIECLOUD_INTERVAL", "MEDIASERVER_SYNC_INTERVAL", SystemConfigKey.MediaServers.value, diff --git a/app/startup/composition/chain.py b/app/startup/composition/chain.py index 0f1705a77..229b9aff8 100644 --- a/app/startup/composition/chain.py +++ b/app/startup/composition/chain.py @@ -125,6 +125,7 @@ def build_chain_runtime_context( classification_service=classification_service, legacy_transfer_command=execute_legacy_transfer_command, configuration=configuration(), + configuration_provider=configuration, durable_event_writer=TransactionalChainDurableEventWriter(SessionFactory), stop_state=runtime_stop_state, system_service=system_service, diff --git a/app/startup/composition/configuration.py b/app/startup/composition/configuration.py index 90794dd37..7f8e05d78 100644 --- a/app/startup/composition/configuration.py +++ b/app/startup/composition/configuration.py @@ -101,6 +101,12 @@ def publish_configuration( ) -> None: """发布 HostRuntime 使用的同一配置对象及兼容设置入口。""" configure_system_config(composition.system_service) + # 统一接管直接 async_set() 写入，避免绕过管理 API 的配置保存没有运行时事件。 + from app.startup.composition.system import _ConfigurationEventAdapter + + composition.system_service.configure_change_publisher( + _ConfigurationEventAdapter().publish + ) configure_user_configuration(composition.user_service) configure_runtime_configuration(composition.runtime) configure_runtime_settings(composition.settings) diff --git a/app/startup/composition/system.py b/app/startup/composition/system.py index 354bd83f8..4bf277cbb 100644 --- a/app/startup/composition/system.py +++ b/app/startup/composition/system.py @@ -190,12 +190,12 @@ class _ConfigurationEventAdapter(ConfigurationEventPort): """把已提交设置变更发布到宿主事件总线。""" async def publish(self, key: Any, value: Any = None) -> None: - """同步失效派生缓存后发布兼容 ConfigChanged 事件。""" + """同步失效派生缓存并等待全部配置重载处理器完成。""" payload = ConfigChangeEventData(key=key, value=value, change_type="update") if payload.key.intersection(_RUST_METAINFO_OPTION_SETTING_KEYS): - # 广播事件异步消费，必须在写入入口返回前失效缓存，避免下一次识别读取旧配置。 + # 必须在写入入口返回前失效缓存，避免下一次识别读取旧配置。 clear_rust_parse_options_cache() - await eventmanager.async_send_event( + await eventmanager.async_send_event_strict( etype=EventType.ConfigChanged, data=payload, ) diff --git a/app/startup/initializers/agent.py b/app/startup/initializers/agent.py index ca5bea557..d039ff768 100644 --- a/app/startup/initializers/agent.py +++ b/app/startup/initializers/agent.py @@ -232,7 +232,7 @@ class AgentInitializer: except Exception as error: self._manager = None self._initialized = False - logger.debug(f"配置变更协调AI智能体失败: {error}") + logger.warning(f"配置变更协调AI智能体失败，当前保持未启用状态: {error}") async def cleanup(self) -> bool: """清理 initializer 引用；未收敛的显式注入对象继续由本实例持有。""" diff --git a/docs/architecture-overview.md b/docs/architecture-overview.md index c983935ac..0720101fe 100644 --- a/docs/architecture-overview.md +++ b/docs/architecture-overview.md @@ -757,7 +757,7 @@ flowchart LR | 指标 | 当前值 | |---|---:| | Python 模块 | 1001 | -| 内部导入边 | 8,490 | +| 内部导入边 | 8,493 | | 非平凡 SCC | 1（精确 containment 的 TMDB 移植包环） | | Application / Chain 具体 Adapter 直连 | 0 / 0 | | Direct egress | 53（债务已清零，53 条精确 containment） | diff --git a/docs/architecture/optimization-checklist.md b/docs/architecture/optimization-checklist.md index 8d14e7237..0cb1d0452 100644 --- a/docs/architecture/optimization-checklist.md +++ b/docs/architecture/optimization-checklist.md @@ -94,7 +94,7 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁 | 指标 | 当前值 | 解释 | |---|---:|---| -| 宿主 Python 模块 / 内部依赖边 | 1001 / 8,490 | `dependency-baseline.json` 当前快照；分类、下载资源归类、订阅搜索、整理恢复、Agent 计划、工具视觉、终端生命周期与 AMLL 歌词模块的受控依赖 | +| 宿主 Python 模块 / 内部依赖边 | 1001 / 8,493 | `dependency-baseline.json` 当前快照；分类、下载资源归类、订阅搜索、整理恢复、Agent 计划、工具视觉、终端生命周期与 AMLL 歌词模块的受控依赖 | | 非平凡 SCC | 1 | 仅保留精确 containment 的 29 模块 TMDB 移植包环 | | 跨层 DB 边界债务 | 0 | Application、Chain、API、Agent、Runtime、Workflow 到 DB 的受控债务均为零 | | Model/Oper 事务债务 | 0 | 自建 Session、自动事务装饰器、直接 commit/rollback 等基线均为零 | @@ -102,8 +102,8 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁 | Event Contract | 53 | 均已有 payload model，但当前全部是 diagnostic enforcement | | Python 源码量 | 305,884 行 | 排除 `app/plugins/**`；61 个文件超过 1,000 行，11 个超过 2,000 行 | | 长方法 | 290 个超过 80 行 | AST 统计排除 `app/plugins/**`；65 个超过 150 行，21 个超过 250 行 | -| 全量 mypy 历史债务 | 9,400 / 509 文件 | Agent API 重构后的现状基线；canonical Facade 与 endpoint 类型边界已补齐，低水位只允许继续下降 | -| Ruff 历史诊断 | 529 | 低水位门禁通过，但规则集只覆盖 `E4/E7/E9/F/I` | +| 全量 mypy 历史债务 | 9,397 / 509 文件 | Agent API 重构后的现状基线；canonical Facade 与 endpoint 类型边界已补齐，低水位只允许继续下降 | +| Ruff 历史诊断 | 527 | 低水位门禁通过，但规则集只覆盖 `E4/E7/E9/F/I` | | 覆盖率固定基线 | Application 80.00%，Domain 80.00% | Chain、Runtime、Agent、Adapter、Startup 未进入包级覆盖率门禁 | ### 3.3 热点文件 diff --git a/scripts/architecture/event_facts.py b/scripts/architecture/event_facts.py index 86c1170fa..7a8b450a9 100644 --- a/scripts/architecture/event_facts.py +++ b/scripts/architecture/event_facts.py @@ -11,7 +11,12 @@ from typing import Any, Literal, TypeAlias _DEFAULT_IDENTITY = "" _DYNAMIC_IDENTITY = "" _EVENT_MANAGER_METHODS = {"add_event_listener", "register"} -_PRODUCER_METHODS = {"async_send_event", "send_event", "send_event_strict"} +_PRODUCER_METHODS = { + "async_send_event", + "async_send_event_strict", + "send_event", + "send_event_strict", +} _COMPREHENSION_SCOPES = ( ast.ListComp, ast.SetComp, @@ -63,6 +68,7 @@ class _BoundEventMethod: "send_event", "send_event_strict", "async_send_event", + "async_send_event_strict", ] receiver_kind: str diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index 37fd92b35..579971238 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -1074,8 +1074,8 @@ "runtime_only": true } }, - "edge_count": 8490, - "edge_sha256": "14a35db36cee8f0f6266a112086efb547f7f3cf5abb67e0f632612654ed3a05c", + "edge_count": 8493, + "edge_sha256": "cf90c1e6dbf33bbb0f9c6fee9bc62f7cf5be26f70658da0cab394627913ee69d", "edges": [ "app -> app.foundation", "app -> app.foundation.environment", @@ -8789,6 +8789,9 @@ "app.startup.composition.configuration -> app.runtime.settings", "app.startup.composition.configuration -> app.schemas", "app.startup.composition.configuration -> app.schemas.types", + "app.startup.composition.configuration -> app.startup", + "app.startup.composition.configuration -> app.startup.composition", + "app.startup.composition.configuration -> app.startup.composition.system", "app.startup.composition.context -> app.application", "app.startup.composition.context -> app.application.agent", "app.startup.composition.context -> app.application.classification", diff --git a/tests/fixtures/architecture/mypy-baseline.json b/tests/fixtures/architecture/mypy-baseline.json index 9c6fbd410..ba0669e9a 100644 --- a/tests/fixtures/architecture/mypy-baseline.json +++ b/tests/fixtures/architecture/mypy-baseline.json @@ -848,7 +848,7 @@ "app/chain/base.py": { "assignment": 12, "attr-defined": 1, - "no-any-return": 47, + "no-any-return": 46, "no-untyped-def": 7, "type-arg": 30 }, @@ -1505,7 +1505,7 @@ "misc": 2, "no-any-return": 1, "no-untyped-def": 2, - "type-arg": 10, + "type-arg": 9, "var-annotated": 1 }, "app/modules/feishu/__init__.py": { @@ -2239,7 +2239,7 @@ "empty-body": 1, "no-any-return": 2, "no-untyped-call": 1, - "no-untyped-def": 4, + "no-untyped-def": 3, "type-arg": 2 }, "app/modules/thetvdb/client.py": { diff --git a/tests/fixtures/architecture/ruff-baseline.json b/tests/fixtures/architecture/ruff-baseline.json index e7c7593ca..7021b80b8 100644 --- a/tests/fixtures/architecture/ruff-baseline.json +++ b/tests/fixtures/architecture/ruff-baseline.json @@ -209,9 +209,6 @@ "F541": 33, "I001": 1 }, - "app/modules/fanart/__init__.py": { - "I001": 1 - }, "app/modules/feishu/__init__.py": { "F401": 1, "I001": 1 @@ -661,9 +658,6 @@ "F841": 1, "I001": 1 }, - "tests/test_event_dispatch_snapshot.py": { - "I001": 1 - }, "tests/test_extensible_media_source_migration.py": { "I001": 1 }, diff --git a/tests/fixtures/architecture/runtime-contract-baseline.json b/tests/fixtures/architecture/runtime-contract-baseline.json index ca78d9446..79eb6d092 100644 --- a/tests/fixtures/architecture/runtime-contract-baseline.json +++ b/tests/fixtures/architecture/runtime-contract-baseline.json @@ -1763,7 +1763,7 @@ "c53927bd5752ec004ac403f6dc07fca7b253f0db152fc773643972d7c3bf5d34" ], "producer_fingerprints": [ - "36d3234c5f3cd7317c34d6ee71bc065946ea0e981dd2377589f73e54ebaf8703" + "10cb1814f97ef9511346d9b79e13255d603f1d7f63fdc3126138d07d74eb8881" ] }, "EventType.DownloadAdded": { @@ -2955,9 +2955,9 @@ "events": [ "EventType.ConfigChanged" ], - "fingerprint": "36d3234c5f3cd7317c34d6ee71bc065946ea0e981dd2377589f73e54ebaf8703", + "fingerprint": "10cb1814f97ef9511346d9b79e13255d603f1d7f63fdc3126138d07d74eb8881", "invalid": false, - "method": "async_send_event", + "method": "async_send_event_strict", "qualname": "_ConfigurationEventAdapter.publish", "receiver_kind": "canonical_singleton" }, diff --git a/tests/test_agent_application_services.py b/tests/test_agent_application_services.py index 2f111636a..ceaa6f36a 100644 --- a/tests/test_agent_application_services.py +++ b/tests/test_agent_application_services.py @@ -213,7 +213,7 @@ async def test_save_system_config_and_settings_service(monkeypatch): monkeypatch.setattr(filtering, "get_configured_system_config", lambda: filter_config) monkeypatch.setattr(settings_module, "plugin_system_config_mutation", lambda _key: nullcontext()) service = settings_module.SystemSettingsService(runtime, system, publish) - await filtering.save_system_config(SystemConfigKey.CustomFilterRules, [None, ""], publish) + await filtering.save_system_config(SystemConfigKey.CustomFilterRules, [None, ""]) assert filter_config.async_set.await_count == 1 secret = service.query(setting_key=SystemConfigKey.Downloaders.value, include_values=True) assert secret["settings"][0]["value"][0]["token"] == "***" diff --git a/tests/test_cache_system.py b/tests/test_cache_system.py index 55fd1f8dd..474267887 100644 --- a/tests/test_cache_system.py +++ b/tests/test_cache_system.py @@ -886,6 +886,8 @@ def test_redis_helpers_watch_pool_settings(): assert "CACHE_REDIS_POOL_TIMEOUT" in RedisHelper.CONFIG_WATCH assert "CACHE_REDIS_MAX_CONNECTIONS" in AsyncRedisHelper.CONFIG_WATCH assert "CACHE_REDIS_POOL_TIMEOUT" in AsyncRedisHelper.CONFIG_WATCH + assert "BIG_MEMORY_MODE" in RedisHelper.CONFIG_WATCH + assert "BIG_MEMORY_MODE" in AsyncRedisHelper.CONFIG_WATCH def test_async_file_backend_missing_region_has_no_items(tmp_path): """ diff --git a/tests/test_chain_runtime_context.py b/tests/test_chain_runtime_context.py index b56b023e4..11507bd6d 100644 --- a/tests/test_chain_runtime_context.py +++ b/tests/test_chain_runtime_context.py @@ -1,6 +1,7 @@ """Chain 运行上下文注入和无参兼容 provider 测试。""" import sys +from dataclasses import replace from types import ModuleType from unittest.mock import Mock, call @@ -121,6 +122,33 @@ def test_chain_keeps_explicit_typed_repositories() -> None: assert chain.user_repository is context.user_repository +def test_chain_runtime_config_provider_refreshes_compatibility_singleton() -> None: + """兼容单例 Chain 应在配置保存后读取新的运行时配置。""" + context = _context() + state = {"configuration": context.configuration} + context = replace( + context, + configuration_provider=lambda: state["configuration"], + ) + + chain = ChainBase(context) + assert chain.runtime_config.media_extensions == (".mkv",) + + state["configuration"] = replace( + context.configuration, + media_extensions=(".mp4",), + ) + assert chain.runtime_config.media_extensions == (".mp4",) + + pinned = replace(chain.runtime_config, media_extensions=(".avi",)) + chain.runtime_config = pinned + state["configuration"] = replace( + context.configuration, + media_extensions=(".mov",), + ) + assert chain.runtime_config.media_extensions == (".avi",) + + def test_chain_composition_registers_lazy_compatibility_provider(monkeypatch) -> None: """组合 API 只登记 provider，首次无参 Chain 请求时才构造完整上下文。""" from app.startup.composition import chain as chain_composition diff --git a/tests/test_configuration_initializer.py b/tests/test_configuration_initializer.py index adfa1bec4..ada89d53a 100644 --- a/tests/test_configuration_initializer.py +++ b/tests/test_configuration_initializer.py @@ -243,7 +243,7 @@ async def test_configuration_load_failure_does_not_publish_partial_service( def test_publish_configuration_reuses_composed_runtime_and_settings(monkeypatch) -> None: """正式运行时与兼容入口必须共享同一组配置对象。""" events = {} - system_service = object() + system_service = SimpleNamespace(configure_change_publisher=MagicMock()) user_service = object() runtime = object() settings_service = SimpleNamespace(update=lambda key, value: (key, value)) @@ -295,6 +295,8 @@ def test_publish_configuration_reuses_composed_runtime_and_settings(monkeypatch) configuration_composition.publish_configuration(composition, legacy_settings) assert events["system"] is system_service + system_service.configure_change_publisher.assert_called_once() + assert callable(system_service.configure_change_publisher.call_args.args[0]) assert events["user"] is user_service assert events["runtime"] is runtime assert events["settings"] is settings_service @@ -518,7 +520,7 @@ async def test_startup_failure_revokes_published_database_services(monkeypatch) from app.application.workflow import get_configured_workflow_query configuration = SimpleNamespace( - system_service=object(), + system_service=SimpleNamespace(configure_change_publisher=MagicMock()), user_service=object(), runtime=object(), settings=SimpleNamespace(update=lambda _key, _value: (True, "")), diff --git a/tests/test_configuration_ports.py b/tests/test_configuration_ports.py index 53b39104c..22396df7d 100644 --- a/tests/test_configuration_ports.py +++ b/tests/test_configuration_ports.py @@ -2,7 +2,7 @@ import asyncio from dataclasses import FrozenInstanceError -from unittest.mock import MagicMock +from unittest.mock import AsyncMock, MagicMock import pytest @@ -174,6 +174,27 @@ def test_system_config_service_normalizes_sync_and_async_writes() -> None: assert normalizer.call_count == 2 +def test_system_config_service_async_write_waits_for_runtime_reload() -> None: + """直接异步写入系统配置时也必须等待统一运行时重载端口。""" + reader = MagicMock() + writer = MagicMock() + writer.set.return_value = True + publisher = AsyncMock() + service = SystemConfigService( + reader=reader, + writer=writer, + async_executor=_InlineDatabaseExecutor(), + change_publisher=publisher, + ) + + assert asyncio.run(service.async_set(SystemConfigKey.Notifications, [{"name": "demo"}])) is True + + publisher.assert_awaited_once_with( + SystemConfigKey.Notifications, + [{"name": "demo"}], + ) + + def test_user_configuration_service_supports_sync_and_async_writes() -> None: """用户配置服务的同步与异步入口执行同一仓储方法。""" repository = MagicMock() diff --git a/tests/test_event_dispatch_snapshot.py b/tests/test_event_dispatch_snapshot.py index bd9ff7b7d..772092358 100644 --- a/tests/test_event_dispatch_snapshot.py +++ b/tests/test_event_dispatch_snapshot.py @@ -7,8 +7,8 @@ from queue import PriorityQueue import pytest -from app.runtime.config import global_vars from app.runtime import events as events_module +from app.runtime.config import global_vars from app.runtime.events import Event, eventmanager from app.schemas.types import ChainEventType, EventType @@ -53,6 +53,14 @@ class _DaemonThreadExecutor: return handle +class _UnboundHandler: + """未创建的宿主 owner，用于验证配置重载的惰性绑定语义。""" + + async def handle(self, _event): + """不应在 owner 尚未激活时执行。""" + raise AssertionError("未激活 owner 不应收到严格配置事件") + + @pytest.fixture def isolated_eventmanager(monkeypatch): """隔离全局事件总线的订阅表和广播执行器。""" @@ -194,6 +202,44 @@ def test_strict_broadcast_waits_and_propagates_handler_failure( assert isolated_eventmanager._EventManager__event_queue.empty() +@pytest.mark.anyio +async def test_async_strict_broadcast_waits_for_async_handler(isolated_eventmanager): + """异步严格广播必须在调用方返回前等待异步 handler 完成。""" + isolated_eventmanager._EventManager__lifecycle_state = "running" + calls = [] + + async def handler(_event): + """模拟需要让出事件循环的配置重载。""" + calls.append("start") + await asyncio.sleep(0) + calls.append("done") + + isolated_eventmanager.add_event_listener(EventType.ConfigChanged, handler) + + await isolated_eventmanager.async_send_event_strict( + EventType.ConfigChanged, + {"key": "AI_AGENT_ENABLE"}, + ) + + assert calls == ["start", "done"] + assert isolated_eventmanager._EventManager__event_queue.empty() + + +@pytest.mark.anyio +async def test_async_strict_broadcast_skips_unbound_optional_owner(isolated_eventmanager): + """严格配置广播应跳过尚未创建的可选 owner，而不是阻断配置保存。""" + isolated_eventmanager._EventManager__lifecycle_state = "running" + isolated_eventmanager.add_event_listener( + EventType.ConfigChanged, + _UnboundHandler.handle, + ) + + await isolated_eventmanager.async_send_event_strict( + EventType.ConfigChanged, + {"key": "CACHE_BACKEND_URL"}, + ) + + def test_sync_chain_dispatch_uses_subscription_snapshot(isolated_eventmanager): """同步链式事件中的订阅变更不影响当前处理器序列。""" calls = [] diff --git a/tests/test_monitor_runtime_settings.py b/tests/test_monitor_runtime_settings.py new file mode 100644 index 000000000..febf56670 --- /dev/null +++ b/tests/test_monitor_runtime_settings.py @@ -0,0 +1,27 @@ +from pathlib import Path +from unittest.mock import patch + +from app.monitor.dispatcher import TransferDispatcher + + +def test_transfer_dispatcher_reads_runtime_extensions_at_use_time() -> None: + """验证已创建的监控分发器会读取最新的整理后缀配置。""" + runtime_settings = { + "DOWNLOAD_TMPEXT": [".part"], + "RMT_MEDIAEXT": [".mkv"], + "RMT_SUBEXT": [".srt"], + "RMT_AUDIOEXT": [".flac"], + } + + with patch( + "app.monitor.dispatcher.get_runtime_setting", + side_effect=runtime_settings.__getitem__, + ): + dispatcher = TransferDispatcher(cache={}) + + assert dispatcher.is_transfer_candidate_path(Path("movie.mkv")) + + runtime_settings["RMT_MEDIAEXT"] = [".mp4"] + + assert not dispatcher.is_transfer_candidate_path(Path("movie.mkv")) + assert dispatcher.is_transfer_candidate_path(Path("movie.mp4")) diff --git a/tests/test_storage_runtime_settings.py b/tests/test_storage_runtime_settings.py new file mode 100644 index 000000000..3f52304f5 --- /dev/null +++ b/tests/test_storage_runtime_settings.py @@ -0,0 +1,30 @@ +"""文件存储适配器运行时设置读取回归测试。""" + +from unittest.mock import patch + +import pytest + +from app.modules.filemanager.storages.alipan import AliPan +from app.modules.filemanager.storages.alist import Alist +from app.modules.filemanager.storages.rclone import Rclone + + +@pytest.mark.parametrize( + ("storage_type", "setting_key"), + [ + (Alist, "OPENLIST_SNAPSHOT_CHECK_FOLDER_MODTIME"), + (Rclone, "RCLONE_SNAPSHOT_CHECK_FOLDER_MODTIME"), + (AliPan, "ALIPAN_SNAPSHOT_CHECK_FOLDER_MODTIME"), + ], +) +def test_storage_snapshot_setting_is_read_at_use_time(storage_type, setting_key) -> None: + """已创建的存储实例也必须读取配置变更后的最新目录检查开关。""" + state = {setting_key: False} + with patch( + f"{storage_type.__module__}.get_runtime_setting", + side_effect=lambda key: state[key], + ): + storage = storage_type.__new__(storage_type) + assert storage.snapshot_check_folder_modtime is False + state[setting_key] = True + assert storage.snapshot_check_folder_modtime is True diff --git a/tests/test_system_application.py b/tests/test_system_application.py index ca153d09c..67c576d61 100644 --- a/tests/test_system_application.py +++ b/tests/test_system_application.py @@ -139,7 +139,8 @@ async def test_update_environment_rejects_validation_and_partial_failure() -> No assert failed.success is False assert failed.message == "invalid host" assert failed.data["success_updates"] == {"PORT": (True, "ok")} - dependencies.events.publish.assert_not_awaited() + published_keys = dependencies.events.publish.await_args.args[0] + assert list(published_keys) == ["PORT"] @pytest.mark.anyio diff --git a/tests/test_system_configuration_event.py b/tests/test_system_configuration_event.py index c6eadaf75..17c673cab 100644 --- a/tests/test_system_configuration_event.py +++ b/tests/test_system_configuration_event.py @@ -14,12 +14,12 @@ from app.startup.composition import system as system_composition async def test_recognition_config_clears_rust_options_before_event_queue( monkeypatch: pytest.MonkeyPatch, ) -> None: - """识别配置写入返回前必须清缓存，再把异步广播事件加入队列。""" + """识别配置写入返回前必须清缓存，再等待配置重载事件完成。""" order: list[str] = [] clear_cache = Mock(side_effect=lambda: order.append("clear")) send_event = AsyncMock(side_effect=lambda **_kwargs: order.append("event")) monkeypatch.setattr(system_composition, "clear_rust_parse_options_cache", clear_cache) - monkeypatch.setattr(system_composition.eventmanager, "async_send_event", send_event) + monkeypatch.setattr(system_composition.eventmanager, "async_send_event_strict", send_event) await system_composition._ConfigurationEventAdapter().publish( SystemConfigKey.CustomIdentifiers, @@ -41,7 +41,7 @@ async def test_unrelated_config_does_not_clear_rust_options( clear_cache = Mock() send_event = AsyncMock() monkeypatch.setattr(system_composition, "clear_rust_parse_options_cache", clear_cache) - monkeypatch.setattr(system_composition.eventmanager, "async_send_event", send_event) + monkeypatch.setattr(system_composition.eventmanager, "async_send_event_strict", send_event) await system_composition._ConfigurationEventAdapter().publish("PORT", 3001) @@ -73,7 +73,7 @@ async def test_recognition_config_rebuilds_cached_rust_options( ) monkeypatch.setattr( system_composition.eventmanager, - "async_send_event", + "async_send_event_strict", AsyncMock(), ) metainfo_module.clear_rust_parse_options_cache() diff --git a/tests/test_transfer_worker_lifecycle.py b/tests/test_transfer_worker_lifecycle.py index 4b3fb8932..f9580bef2 100644 --- a/tests/test_transfer_worker_lifecycle.py +++ b/tests/test_transfer_worker_lifecycle.py @@ -41,6 +41,9 @@ def _build_chain(*, transfer_threads: int = 0) -> TransferChain: chain.runtime_config = SimpleNamespace( transfer_threads=transfer_threads, transfer_task_timeout=0, + video_extensions=(".mkv",), + subtitle_extensions=(".srt",), + audio_extensions=(".flac",), ) chain._queue = queue.Queue() chain._transfer_interval = 0.1