diff --git a/app/api/endpoints/agent.py b/app/api/endpoints/agent.py index 02a7edb83..828ee40c8 100644 --- a/app/api/endpoints/agent.py +++ b/app/api/endpoints/agent.py @@ -329,7 +329,7 @@ async def run_agent_command( except ValueError: channel = None try: - data = web_agent_application.dispatch_web_agent_command( + data = web_agent_application.dispatch_command( payload.command, user_id=str(current_user.id), channel=channel, diff --git a/app/api/endpoints/plugin.py b/app/api/endpoints/plugin.py index dac2719df..535825a62 100644 --- a/app/api/endpoints/plugin.py +++ b/app/api/endpoints/plugin.py @@ -949,8 +949,6 @@ def set_plugin_config( """ result = command.update(plugin_id, conf) return _SchemaResponse(success=result.success, message=result.message) - - @router.delete("/{plugin_id}", summary="卸载插件", response_model=_SchemaResponse[None]) def uninstall_plugin(plugin_id: str, _: ApiPrincipal = Depends(get_current_active_superuser)) -> Any: """ diff --git a/app/api/endpoints/recommend.py b/app/api/endpoints/recommend.py index dac522e00..6503ecbe3 100644 --- a/app/api/endpoints/recommend.py +++ b/app/api/endpoints/recommend.py @@ -1,4 +1,5 @@ -from typing import Any, Awaitable, List, Optional +from collections.abc import Awaitable, Callable +from typing import Any, List, Optional from fastapi import Depends, HTTPException, status @@ -43,6 +44,207 @@ async def _require_tmdb_result(operation: Awaitable[List[Any]]) -> List[Any]: ) from error +def _normalize_agent_media_type(media_type: str) -> Optional[str]: + """将 Agent 媒体类型转换为推荐链使用的稳定标识。""" + if media_type == "all": + return media_type + media_type_enum = MediaType.from_agent(media_type) + return media_type_enum.to_agent() if media_type_enum else None + + +async def _fetch_listenbrainz_chart( + chain: RecommendChain, + *, + page: int, + count: int, + range_name: str, + sort_by: str, + min_listen_count: int, + with_cover: bool, + entity: Optional[str], +) -> List[Any] | _SchemaResponse: + """校验榜单参数并获取 ListenBrainz 榜单结果。""" + if range_name not in LISTENBRAINZ_CHART_RANGES: + return _SchemaResponse(success=False, message="无效的榜单周期") + if sort_by not in {"listen_count.desc", "listen_count.asc"}: + return _SchemaResponse(success=False, message="无效的榜单排序") + return await chain.async_music_chart( + range_name=range_name, + page=page, + count=count, + sort_by=sort_by, + min_listen_count=max(0, min_listen_count), + with_cover=with_cover, + entity=entity or MUSIC_ENTITY_RECORDING, + ) + + +async def _fetch_listenbrainz_fresh( + chain: RecommendChain, + *, + page: int, + count: int, + music_type: Optional[str], + days: int, + fresh_sort: str, + past: bool, + future: bool, + with_cover: bool, +) -> List[Any] | _SchemaResponse: + """校验新发行参数并获取 ListenBrainz 新发行结果。""" + if music_type not in {None, MUSIC_ENTITY_ALBUM}: + return _SchemaResponse(success=False, message="新发行结果只支持专辑") + if fresh_sort not in LISTENBRAINZ_FRESH_SORTS: + return _SchemaResponse(success=False, message="无效的新发行排序") + if not past and not future: + return _SchemaResponse( + success=False, + message="past 和 future 不能同时为 false", + ) + return await chain.async_music_fresh_releases( + days=max(1, min(days, LISTENBRAINZ_FRESH_MAX_DAYS)), + sort=fresh_sort, + past=past, + future=future, + page=page, + count=count, + with_cover=with_cover, + ) + + +async def _recommend_listenbrainz( + chain: RecommendChain, + *, + source: str, + media_type: str, + page: int, + count: int, + music_type: Optional[str], + range_name: str, + sort_by: str, + days: int, + fresh_sort: str, + past: bool, + future: bool, + min_listen_count: int, + with_cover: bool, +) -> _SchemaResponse: + """执行 ListenBrainz 推荐分支并统一转换音乐结果。""" + if media_type not in {"all", "music"}: + return _SchemaResponse( + success=False, + message="ListenBrainz 来源只支持音乐媒体类型", + ) + normalized_music_type = ( + normalize_music_type(music_type, allow_artist=False) if music_type else None + ) + if music_type and normalized_music_type is None: + return _SchemaResponse(success=False, message="无效的音乐实体类型") + if source == "listenbrainz_chart": + results = await _fetch_listenbrainz_chart( + chain, + page=page, + count=count, + range_name=range_name, + sort_by=sort_by, + min_listen_count=min_listen_count, + with_cover=with_cover, + entity=normalized_music_type, + ) + else: + results = await _fetch_listenbrainz_fresh( + chain, + page=page, + count=count, + music_type=normalized_music_type, + days=days, + fresh_sort=fresh_sort, + past=past, + future=future, + with_cover=with_cover, + ) + if isinstance(results, _SchemaResponse): + return results + return _SchemaResponse( + success=True, + data=[simplify_music_info(item) for item in results or []], + ) + + +def _media_source_operations( + chain: RecommendChain, + *, + page: int, + count: int, +) -> dict[str, Callable[[], Awaitable[List[Any]]]]: + """构造影视推荐来源到链方法的延迟调用映射。""" + return { + "tmdb_trending": lambda: chain.async_tmdb_trending(page=page), + "tmdb_movies": lambda: chain.async_tmdb_movies(page=page), + "tmdb_tvs": lambda: chain.async_tmdb_tvs(page=page), + "douban_movie_hot": lambda: chain.async_douban_movie_hot(page=page, count=count), + "douban_tv_hot": lambda: chain.async_douban_tv_hot(page=page, count=count), + "douban_movie_showing": lambda: chain.async_douban_movie_showing(page=page, count=count), + "douban_movies": lambda: chain.async_douban_movies(page=page, count=count), + "douban_tvs": lambda: chain.async_douban_tvs(page=page, count=count), + "douban_movie_top250": lambda: chain.async_douban_movie_top250(page=page, count=count), + "douban_tv_weekly_chinese": lambda: chain.async_douban_tv_weekly_chinese(page=page, count=count), + "douban_tv_weekly_global": lambda: chain.async_douban_tv_weekly_global(page=page, count=count), + "douban_tv_animation": lambda: chain.async_douban_tv_animation(page=page, count=count), + "bangumi_calendar": lambda: chain.async_bangumi_calendar(page=page, count=count), + } + + +async def _fetch_media_recommendations( + chain: RecommendChain, + *, + source: str, + media_type: str, + page: int, + count: int, +) -> tuple[bool, List[Any]]: + """获取影视推荐结果，并处理按媒体类型拆分的豆瓣热门来源。""" + if source == "douban_hot": + results: List[Any] = [] + if media_type in {"all", "movie"}: + results.extend(await chain.async_douban_movie_hot(page=page, count=count)) + if media_type in {"all", "tv"}: + results.extend(await chain.async_douban_tv_hot(page=page, count=count)) + return True, results + operation = _media_source_operations(chain, page=page, count=count).get(source) + if operation is None: + return False, [] + return True, await operation() + + +def _project_agent_recommendations(results: List[Any], count: int) -> List[dict[str, Any]]: + """将推荐链结果裁剪为 Agent 稳定响应字段。""" + projected = [] + for item in (results or [])[:count]: + if not isinstance(item, dict): + continue + projected.append( + { + "title": item.get("title"), + "en_title": item.get("en_title"), + "year": item.get("year"), + "type": media_type_to_agent(item.get("type")), + "season": item.get("season"), + "tmdb_id": item.get("tmdb_id"), + "imdb_id": item.get("imdb_id"), + "douban_id": item.get("douban_id"), + "bangumi_id": item.get("bangumi_id"), + "anilist_id": item.get("anilist_id"), + "media_source": item.get("media_source"), + "media_id": item.get("media_id"), + "vote_average": item.get("vote_average"), + "poster_path": item.get("poster_path"), + "detail_link": item.get("detail_link"), + } + ) + return projected + + @router.get( "/source", summary="获取推荐数据源", @@ -86,116 +288,45 @@ async def agent_recommendations( """按稳定来源标识返回有界影视、动画或音乐推荐结果。""" page = max(1, page) count = 20 - if media_type != "all": - media_type_enum = MediaType.from_agent(media_type) - if media_type_enum is None: - return _SchemaResponse(success=False, message="无效的媒体类型") - media_type = media_type_enum.to_agent() + normalized_media_type = _normalize_agent_media_type(media_type) + if normalized_media_type is None: + return _SchemaResponse(success=False, message="无效的媒体类型") chain = RecommendChain() if source in {"listenbrainz_chart", "listenbrainz_fresh"}: - if media_type not in {"all", "music"}: - return _SchemaResponse( - success=False, - message="ListenBrainz 来源只支持音乐媒体类型", - ) - normalized_music_type = normalize_music_type(music_type, allow_artist=False) if music_type else None - if music_type and normalized_music_type is None: - return _SchemaResponse(success=False, message="无效的音乐实体类型") - if source == "listenbrainz_chart": - if range_name not in LISTENBRAINZ_CHART_RANGES: - return _SchemaResponse(success=False, message="无效的榜单周期") - if sort_by not in {"listen_count.desc", "listen_count.asc"}: - return _SchemaResponse(success=False, message="无效的榜单排序") - music_results = await chain.async_music_chart( - range_name=range_name, - page=page, - count=count, - sort_by=sort_by, - min_listen_count=max(0, min_listen_count), - with_cover=with_cover, - entity=normalized_music_type or MUSIC_ENTITY_RECORDING, - ) - else: - if normalized_music_type not in {None, MUSIC_ENTITY_ALBUM}: - return _SchemaResponse( - success=False, - message="新发行结果只支持专辑", - ) - if fresh_sort not in LISTENBRAINZ_FRESH_SORTS: - return _SchemaResponse(success=False, message="无效的新发行排序") - if not past and not future: - return _SchemaResponse( - success=False, - message="past 和 future 不能同时为 false", - ) - music_results = await chain.async_music_fresh_releases( - days=max(1, min(days, LISTENBRAINZ_FRESH_MAX_DAYS)), - sort=fresh_sort, - past=past, - future=future, - page=page, - count=count, - with_cover=with_cover, - ) - return _SchemaResponse( - success=True, - data=[simplify_music_info(item) for item in music_results or []], + return await _recommend_listenbrainz( + chain, + source=source, + media_type=normalized_media_type, + page=page, + count=count, + music_type=music_type, + range_name=range_name, + sort_by=sort_by, + days=days, + fresh_sort=fresh_sort, + past=past, + future=future, + min_listen_count=min_listen_count, + with_cover=with_cover, ) - if media_type == "music": + if normalized_media_type == "music": return _SchemaResponse( success=False, message="音乐推荐需使用 ListenBrainz 来源", ) - source_calls = { - "tmdb_trending": lambda: chain.async_tmdb_trending(page=page), - "tmdb_movies": lambda: chain.async_tmdb_movies(page=page), - "tmdb_tvs": lambda: chain.async_tmdb_tvs(page=page), - "douban_movie_hot": lambda: chain.async_douban_movie_hot(page=page, count=count), - "douban_tv_hot": lambda: chain.async_douban_tv_hot(page=page, count=count), - "douban_movie_showing": lambda: chain.async_douban_movie_showing(page=page, count=count), - "douban_movies": lambda: chain.async_douban_movies(page=page, count=count), - "douban_tvs": lambda: chain.async_douban_tvs(page=page, count=count), - "douban_movie_top250": lambda: chain.async_douban_movie_top250(page=page, count=count), - "douban_tv_weekly_chinese": lambda: chain.async_douban_tv_weekly_chinese(page=page, count=count), - "douban_tv_weekly_global": lambda: chain.async_douban_tv_weekly_global(page=page, count=count), - "douban_tv_animation": lambda: chain.async_douban_tv_animation(page=page, count=count), - "bangumi_calendar": lambda: chain.async_bangumi_calendar(page=page, count=count), - } - if source == "douban_hot": - results = [] - if media_type in {"all", "movie"}: - results.extend(await chain.async_douban_movie_hot(page=page, count=count)) - if media_type in {"all", "tv"}: - results.extend(await chain.async_douban_tv_hot(page=page, count=count)) - else: - operation = source_calls.get(source) - if operation is None: - return _SchemaResponse(success=False, message=f"不支持的推荐来源: {source}") - results = await operation() - projected = [] - for item in (results or [])[:count]: - if not isinstance(item, dict): - continue - projected.append( - { - "title": item.get("title"), - "en_title": item.get("en_title"), - "year": item.get("year"), - "type": media_type_to_agent(item.get("type")), - "season": item.get("season"), - "tmdb_id": item.get("tmdb_id"), - "imdb_id": item.get("imdb_id"), - "douban_id": item.get("douban_id"), - "bangumi_id": item.get("bangumi_id"), - "anilist_id": item.get("anilist_id"), - "media_source": item.get("media_source"), - "media_id": item.get("media_id"), - "vote_average": item.get("vote_average"), - "poster_path": item.get("poster_path"), - "detail_link": item.get("detail_link"), - } - ) - return _SchemaResponse(success=True, data=projected) + supported, results = await _fetch_media_recommendations( + chain, + source=source, + media_type=normalized_media_type, + page=page, + count=count, + ) + if not supported: + return _SchemaResponse(success=False, message=f"不支持的推荐来源: {source}") + return _SchemaResponse( + success=True, + data=_project_agent_recommendations(results, count), + ) @router.get( diff --git a/app/application/messaging/agent.py b/app/application/messaging/agent.py index 2addfa1f6..a77ee402e 100644 --- a/app/application/messaging/agent.py +++ b/app/application/messaging/agent.py @@ -48,6 +48,7 @@ from app.runtime.tasks import get_task_registry from app.schemas.message import Message from app.schemas.types import NotificationChannel, ReplyMode +__all__ = ["dispatch_command"] # Agent 选择按钮回调前缀（新旧两种格式都必须继续兼容） AGENT_CHOICE_PREFIX = "agent_interaction:choice:" LEGACY_AGENT_CHOICE_PREFIX = "agent_choice:" @@ -1704,25 +1705,6 @@ def build_web_agent_command_items() -> list[dict[str, Any]]: ) return sorted(items, key=lambda item: (item["category"], item["command"])) - -def dispatch_web_agent_command( - command: str, - *, - user_id: str, - channel: Optional[NotificationChannel], - source: Optional[str], - publish_event: Callable[[Any, dict[str, Any]], Any], -) -> dict[str, Any]: - """经消息应用边界校验并触发一条 Agent 斜杠命令。""" - return dispatch_command( - command, - user_id=user_id, - channel=channel, - source=source, - publish_event=publish_event, - ) - - def extract_web_agent_slash_command(text: str) -> Optional[str]: """ 从 WebAgent 输入中提取斜杠命令名。 diff --git a/docs/architecture/optimization-checklist.md b/docs/architecture/optimization-checklist.md index ebe7d5946..ba2761a17 100644 --- a/docs/architecture/optimization-checklist.md +++ b/docs/architecture/optimization-checklist.md @@ -103,7 +103,7 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁 | 长方法 | 290 个超过 80 行 | AST 统计排除 `app/plugins/**`；65 个超过 150 行，21 个超过 250 行 | | 全量 mypy 历史债务 | 9,983 / 591 文件 | canonical `SearchChain` Facade 已补齐显式类型转发；strict frontier 当前覆盖 41 个文件，低水位只允许继续下降 | | Ruff 历史诊断 | 630 | 低水位门禁通过，但规则集只覆盖 `E4/E7/E9/F/I` | -| 覆盖率低水位 | Application 81.86%，Domain 81.03% | Chain、Runtime、Agent、Adapter、Startup 未进入包级覆盖率门禁 | +| 覆盖率低水位 | Application 81.88%，Domain 81.03% | Chain、Runtime、Agent、Adapter、Startup 未进入包级覆盖率门禁 | ### 3.3 热点文件 diff --git a/tests/fixtures/architecture/coverage-baseline.json b/tests/fixtures/architecture/coverage-baseline.json index 94c3fff8b..fdb53db06 100644 --- a/tests/fixtures/architecture/coverage-baseline.json +++ b/tests/fixtures/architecture/coverage-baseline.json @@ -1,8 +1,8 @@ { "application": { - "covered_lines": 12996, - "percent": 81.86, - "statements": 15875 + "covered_lines": 13687, + "percent": 81.88, + "statements": 16716 }, "domain": { "covered_lines": 3686, diff --git a/tests/test_agent_application_services.py b/tests/test_agent_application_services.py new file mode 100644 index 000000000..f49946b34 --- /dev/null +++ b/tests/test_agent_application_services.py @@ -0,0 +1,397 @@ +"""Agent 应用服务的纯逻辑、端口编排和安全投影测试。""" + +from contextlib import asynccontextmanager, nullcontext +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock + +import pytest + +import app.application.commands as command_application +import app.application.filtering as filtering +import app.application.plugin.management as plugin_management +import app.application.settings as settings_module +from app.application.download.tasks import DownloadTaskMutationService, DownloadTaskService +from app.application.music.projection import simplify_music_album, simplify_music_artist, simplify_music_info +from app.application.plugin.data import ( + DeletePluginDataCommand, + PluginDataQueryService, + build_preview_payload, + clamp_preview_chars, +) +from app.application.security.secrets import is_secret_setting_key +from app.domain.context import MusicAlbumInfo, MusicArtistInfo, MusicInfo, MusicRelease +from app.domain.projection.douban import project as project_douban +from app.schemas.rule import CustomRule, FilterRuleGroup +from app.schemas.types import EventType, MediaSource, SystemConfigKey + + +def run_async(awaitable): + """在普通 pytest 函数中执行一个短异步断言。""" + import asyncio + return asyncio.run(awaitable) + + +def test_filtering_primitives_and_projection(): + """过滤规则基础函数应统一输入、解析引用并隔离可变输出。""" + assert filtering.normalize_optional_text(None) is None + assert filtering.normalize_optional_text(" value ") == "value" + assert filtering.normalize_media_type("movie") == "电影" + assert filtering.normalize_media_type("电视剧") == "电视剧" + assert filtering.normalize_media_type("") is None + with pytest.raises(ValueError, match="media_type"): + filtering.normalize_media_type("documentary") + assert filtering.validate_numeric_range("size_range", None) is None + assert filtering.validate_numeric_range("size_range", "1000 - 5000") == "1000 - 5000" + with pytest.raises(ValueError, match="格式无效"): + filtering.validate_numeric_range("size_range", "abc") + with pytest.raises(ValueError, match="起始值"): + filtering.validate_numeric_range("size_range", "5-2") + assert filtering.validate_seeders(" 12 ") == "12" + with pytest.raises(ValueError, match="非负整数"): + filtering.validate_seeders("1.2") + builtin = filtering.get_builtin_rules() + builtin["4K"]["_test"] = "changed" + assert "_test" not in filtering.get_builtin_rules()["4K"] + custom = CustomRule(id="CUSTOM", name="Custom", include="x") + group = FilterRuleGroup(name="main", rule_string="CUSTOM & 4K") + assert filtering.build_custom_rule_map([custom, CustomRule(id=None)]) == {"CUSTOM": custom} + assert filtering.build_rule_group_map([group, FilterRuleGroup(name=None)]) == {"main": group} + assert filtering.extract_rule_tokens("CUSTOM & 4K & CUSTOM") == ["CUSTOM", "4K"] + assert filtering.extract_rule_tokens(None) == [] + assert filtering.parse_rule_string("CUSTOM & 4K > !BLU")["levels"][0]["priority"] == 1 + with pytest.raises(ValueError, match="不能为空"): + filtering.parse_rule_string(" ") + with pytest.raises(ValueError, match="空层级"): + filtering.parse_rule_string("4K > ") + assert filtering.validate_rule_string("CUSTOM & 4K", ["CUSTOM", "4K"])["levels"] + with pytest.raises(ValueError, match="不存在"): + filtering.validate_rule_string("MISSING", ["4K"]) + assert filtering.serialize_builtin_rule("4K", {"name": "4K"})["source"] == "builtin" + assert filtering.serialize_custom_rule(custom, ["main"])["referenced_by_rule_groups"] == ["main"] + assert filtering.serialize_rule_group(group)["syntax_valid"] is True + assert filtering.serialize_rule_group(FilterRuleGroup(name="bad", rule_string="4K > "))["syntax_valid"] is False + assert filtering.serialize_rule_group(FilterRuleGroup(name="empty"))["syntax_valid"] is False + assert filtering.replace_rule_id_in_rule_string("OLD & OLDER", "OLD", "NEW") == "NEW & OLDER" + + +def test_filtering_normalizers_and_usage_collection(monkeypatch): + """规则实体校验应拒绝重复项，并能汇总全局与订阅引用。""" + rule = filtering.normalize_custom_rule("NEW", "New rule", "inc", None, "1-2", "3", "2024", []) + assert rule.id == "NEW" + with pytest.raises(ValueError, match="不能为空"): + filtering.normalize_custom_rule("", "New", None, None, None, None, None, []) + with pytest.raises(ValueError, match="不能为空"): + filtering.normalize_custom_rule("NEW", "", None, None, None, None, None, []) + with pytest.raises(ValueError, match="内置"): + filtering.normalize_custom_rule("4K", "Built-in", None, None, None, None, None, []) + with pytest.raises(ValueError, match="已存在"): + filtering.normalize_custom_rule("NEW", "Other", None, None, None, None, None, [rule]) + with pytest.raises(ValueError, match="规则名称"): + filtering.normalize_custom_rule("OTHER", "New rule", None, None, None, None, None, [rule]) + with pytest.raises(ValueError, match="rule_id"): + filtering.normalize_custom_rule("bad-id", "Bad", None, None, None, None, None, []) + with pytest.raises(ValueError, match="规则组名称"): + filtering.normalize_rule_group("", "4K", None, None, [], ["4K"]) + with pytest.raises(ValueError, match="category"): + filtering.normalize_rule_group("group", "4K", None, "movie", [], ["4K"]) + normalized_group, parsed = filtering.normalize_rule_group("group", "4K", "movie", "action", [], ["4K"]) + assert normalized_group.media_type == "电影" and parsed["levels"] + with pytest.raises(ValueError, match="已存在"): + filtering.normalize_rule_group("group", "4K", None, None, [normalized_group], ["4K"]) + config = MagicMock() + config.get.side_effect = lambda key: { + SystemConfigKey.SearchFilterRuleGroups: ["search"], + SystemConfigKey.SubscribeFilterRuleGroups: ["subscribe"], + SystemConfigKey.BestVersionFilterRuleGroups: ["best"], + }.get(key, []) + monkeypatch.setattr(filtering, "get_configured_system_config", lambda: config) + + class SubscriptionPort: + """提供规则组使用统计所需的最小订阅端口。""" + + async def async_list(self): + """返回带规则组引用的订阅快照。""" + return [SimpleNamespace(id=1, name="Sub", season=1, type="电影", username="u", best_version=True, filter_groups=["search", "custom"]), SimpleNamespace(filter_groups=None)] + + usage = run_async(filtering.collect_rule_group_usages(SubscriptionPort(), ["search", "custom"])) + assert usage["search"]["used_in_global_search"] is True + assert usage["search"]["subscribes"][0]["subscribe_id"] == 1 + refs = filtering.collect_custom_rule_group_refs([FilterRuleGroup(name="main", rule_string="CUSTOM & 4K"), FilterRuleGroup(name="none")], ["CUSTOM"]) + assert refs["CUSTOM"] == ["main"] + + +@pytest.mark.asyncio +async def test_filter_rule_service_queries_and_mutations(monkeypatch): + """规则服务应覆盖查询、增删改和重命名引用的事务边界。""" + old_rule = CustomRule(id="OLD", name="Old", include="old") + old_group = FilterRuleGroup(name="group", rule_string="OLD & 4K") + monkeypatch.setattr(filtering, "get_custom_rules", lambda: [old_rule]) + monkeypatch.setattr(filtering, "get_rule_groups", lambda: [old_group]) + config = MagicMock() + config.async_set = AsyncMock(return_value=True) + monkeypatch.setattr(filtering, "get_configured_system_config", lambda: config) + publish = AsyncMock() + mutation = MagicMock() + mutation.apply = AsyncMock(return_value=SimpleNamespace(to_dict=lambda: {"changed": True})) + + @asynccontextmanager + async def mutation_scope(): + """提供可观测的规则组异步事务。""" + yield mutation + + class SubscriptionPort: + """提供空订阅列表的查询端口。""" + + async def async_list(self): + """返回空订阅集合。""" + return [] + + service = filtering.FilterRuleService(SubscriptionPort(), mutation_scope, publish) + assert filtering.FilterRuleService.query_builtin(["4K"])["count"] == 1 + assert filtering.FilterRuleService.query_custom(["OLD"])["count"] == 1 + assert (await service.query_groups(include_usage=False))["count"] == 1 + assert (await service.query_groups(["missing"], include_usage=False))["count"] == 0 + added = await service.add_custom(rule_id="NEW", name="New", include="new", exclude=None, size_range=None, seeders=None, publish_time=None) + assert added["custom_rule"]["id"] == "NEW" + updated = await service.update_custom(current_rule_id="OLD", new_rule_id="RENAMED") + assert updated["rule_groups_updated_for_rule_id_rename"] == ["group"] + monkeypatch.setattr(filtering, "get_custom_rules", lambda: [CustomRule(id="DELETE", name="Delete")]) + monkeypatch.setattr(filtering, "get_rule_groups", lambda: []) + assert (await service.delete_custom("DELETE"))["count"] == 0 + assert (await service.add_group(name="new-group", rule_string="4K"))["rule_group"]["name"] == "new-group" + monkeypatch.setattr(filtering, "get_rule_groups", lambda: [FilterRuleGroup(name="old", rule_string="4K")]) + assert (await service.update_group(current_name="old", new_name="new"))["rule_group"]["name"] == "new" + assert (await service.delete_group("old"))["count"] == 0 + monkeypatch.setattr(filtering, "get_rule_groups", lambda: []) + with pytest.raises(ValueError, match="不存在"): + await service.delete_group("old") + with pytest.raises(ValueError, match="不存在"): + await service.delete_custom("missing") + + +@pytest.mark.asyncio +async def test_save_system_config_and_settings_service(monkeypatch): + """系统设置服务应处理摘要、脱敏、合并、列表更新和两类配置源。""" + runtime = MagicMock() + runtime.get.side_effect = lambda key: {"LLM_MODEL": "model", "PLUGIN_MARKET": "a"}.get(key) + runtime.update.return_value = (True, "updated") + system = MagicMock() + system.get.side_effect = lambda key: [{"name": "qb", "token": "secret"}] if key == SystemConfigKey.Downloaders else {"a": 1} + system.async_set = AsyncMock(return_value=True) + publish = AsyncMock() + filter_config = MagicMock() + filter_config.async_set = AsyncMock(return_value=True) + 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) + 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"] == "***" + shown = service.query(setting_key=SystemConfigKey.Downloaders.value, include_values=True, show_secrets=True) + assert shown["settings"][0]["value"][0]["token"] == "secret" + assert service.query(group="ai_agent")["include_values"] is False + spec = settings_module.resolve_setting_spec(SystemConfigKey.Downloaders.value) + assert spec + assert service._prepare_next_value(spec, {"name": "old", "x": 1}, {"name": "old", "y": 2}, "merge_dict", ["x"], None, None) == {"name": "old", "y": 2} + assert service._prepare_next_value(spec, [{"name": "old"}], {"name": "old", "x": 2}, "upsert_list_item", None, None, None) == [{"name": "old", "x": 2}] + assert service._prepare_next_value(spec, [{"name": "old"}], {"name": "old"}, "remove_list_item", None, None, None) == [] + with pytest.raises(ValueError, match="不支持"): + service._prepare_next_value(spec, None, None, "bad", None, None, None) + system.get.side_effect = [[], [{"name": "new"}]] + result = await service.update(setting_key=SystemConfigKey.Downloaders.value, value={"name": "new"}, operation="upsert_list_item") + assert result["changed"] is True + runtime.get.side_effect = lambda key: "old" + assert (await service.update(setting_key="PLUGIN_MARKET", value="new"))["changed"] is True + + +def test_settings_catalog_redaction_and_projection(): + """设置目录应支持分类别名、匹配字段和递归敏感值脱敏。""" + assert settings_module.normalize_group("基础配置") == "settings" + assert settings_module.normalize_group("全部") == "all" + with pytest.raises(ValueError, match="group"): + settings_module.normalize_group("unknown") + assert settings_module.resolve_setting_spec("Downloaders") is not None + assert settings_module.list_setting_specs("downloaders")[0].group == "downloaders" + assert settings_module.list_setting_specs("ai_agent", keyword="llm") + assert settings_module.get_default_list_match_field(SystemConfigKey.Downloaders.value) == "name" + assert settings_module.redact_secret_value({"apiKey": "secret", "url": "x", "nested": ["plain"]})["apiKey"] == "***" + assert settings_module.normalize_group(None) == "all" + assert settings_module.resolve_setting_spec(None) is None + spec = settings_module.resolve_setting_spec(SystemConfigKey.UserSiteAuthParams.value) + assert spec and settings_module.should_redact_setting(spec, [{"token": "secret"}]) + assert is_secret_setting_key("accessToken") and not is_secret_setting_key("token_count") + + +@pytest.mark.asyncio +async def test_plugin_management_and_data_services(monkeypatch): + """插件管理、来源补齐和数据预览应覆盖成功与安全失败路径。""" + plugin = SimpleNamespace(id="Demo", plugin_name="Demo Plugin", plugin_desc="desc", plugin_version="1", plugin_author="author", installed=True, has_update=True, state=True, repo_url=None, add_time=1) + class SourceCandidate: + """提供插件来源检查所需的公开投影。""" + + id = "Demo" + plugin_name = "Demo Plugin" + repo_url = "https://github.com/demo/repo" + has_update = True + release = "r1" + system_version = "3" + system_version_compatible = True + system_version_message = None + + def public_dict(self): + """返回来源候选的脱敏字典。""" + return {"id": self.id, "repo_url": self.repo_url} + + source = SourceCandidate() + manager = MagicMock() + manager.get_local_plugins.return_value = [plugin] + manager.get_local_repo_plugins.return_value = [source] + manager.async_get_online_plugins = AsyncMock(return_value=[]) + manager.process_plugins_list.return_value = [source] + monkeypatch.setattr(plugin_management, "get_plugin_manager", lambda: manager) + assert plugin_management.get_plugin_snapshot("Demo")["plugin_id"] == "Demo" + assert plugin_management.summarize_plugin(plugin)["source"] == "market" + assert plugin_management.is_exact_plugin_match(plugin, "demo plugin") + assert plugin_management.search_plugin_candidates("demo", [plugin])[0]["exact"] is True + assert plugin_management.summarize_candidates(plugin_management.search_plugin_candidates("demo", [plugin]), 1)[0]["id"] == "Demo" + assert await plugin_management.enrich_installed_plugin_sources([plugin]) == [plugin] + assert plugin.repo_url == source.repo_url + assert await plugin_management.load_market_plugins() == [source] + assert plugin_management.list_installed_plugins() == [plugin] + install_service = MagicMock() + install_service.install = AsyncMock(return_value=SimpleNamespace(success=True, message="ok", refreshed_only=False)) + install_service.inspect_source = AsyncMock(return_value=SimpleNamespace(online_candidates=[source], local_candidate=None, selection=SimpleNamespace(status=SimpleNamespace(value="selected"), reason="exact"), inventory_complete=True)) + monkeypatch.setattr(plugin_management, "get_plugin_install_service", lambda: install_service) + assert await plugin_management.install_plugin_runtime("Demo", source.repo_url) == (True, "ok", False) + assert (await plugin_management.inspect_plugin_sources("Demo"))["selection_status"] == "selected" + repo = MagicMock() + unit = MagicMock() + DeletePluginDataCommand(repo, unit).execute("Demo") + unit.commit.assert_called_once() + assert clamp_preview_chars(1) == 512 + truncated, total, returned, preview = build_preview_payload({"value": "x" * 1000}, 512) + assert truncated and total > returned and "截断" in preview + query_repo = MagicMock() + query_repo.get = AsyncMock(return_value={"key": "value"}) + query_repo.list = AsyncMock(return_value={"a": 1}) + query = PluginDataQueryService(query_repo, lambda _id: {"plugin_id": "Demo"}) + assert (await query.query("Demo", key="config"))["found"] + query_repo.get.return_value = None + assert not (await query.query("Demo", key="missing"))["found"] + assert (await query.query("Demo"))["count"] == 1 + with pytest.raises(ValueError, match="不存在"): + await PluginDataQueryService(query_repo, lambda _id: None).query("Demo") + + +def test_remaining_application_guard_paths(monkeypatch): + """应用服务的空结果、回滚和匹配参数保护应保持可观测。""" + helper = MagicMock() + helper.get_custom_rules.return_value = [] + helper.get_rule_groups.return_value = [] + monkeypatch.setattr(filtering, "RuleHelper", lambda: helper) + assert filtering.get_custom_rules() == [] + assert filtering.get_rule_groups() == [] + assert settings_module.SystemSettingsService._normalize_systemconfig_value([]) is None + with pytest.raises(ValueError, match="匹配字段"): + settings_module.SystemSettingsService._resolve_list_match( + settings_module.SettingSpec("CUSTOM", "settings", "x", "x"), + "upsert_list_item", + {"value": 1}, + None, + None, + ) + empty_service = DownloadTaskService( + MagicMock(return_value=[]), + lambda _hashes: {}, + MagicMock(), + MagicMock(), + MagicMock(), + ) + assert empty_service.downloading() == [] + + +def test_command_application_facade_and_dispatch(monkeypatch): + """命令应用门面应支持注册、查询、初始化和事件派发。""" + + class CommandRegistry: + """提供命令门面所需的最小注册表实现。""" + + calls = [] + + def get_commands(self): + """返回一个可触发的命令定义。""" + return {"/demo": {"description": "Demo", "pid": "plugin"}} + + def get(self, name): + """按名称返回命令定义。""" + return self.get_commands().get(name) + + def init_commands(self, plugin_id=None): + """记录命令初始化范围。""" + self.calls.append(plugin_id) + + command_application.register_command_class(CommandRegistry) + try: + assert isinstance(command_application.get_command_object(), CommandRegistry) + assert command_application.get_commands()["/demo"]["description"] == "Demo" + assert command_application.get_command("/demo")["pid"] == "plugin" + command_application.init_commands("plugin") + published = [] + result = command_application.dispatch_command( + "demo --flag", + user_id="1", + source="test", + publish_event=lambda event_type, payload: published.append((event_type, payload)), + ) + assert result["command"] == "/demo --flag" + assert result["plugin_id"] == "plugin" + assert published == [ + ( + EventType.CommandExcute, + {"cmd": "/demo --flag", "user": "1", "channel": None, "source": "test"}, + ) + ] + with pytest.raises(ValueError, match="命令不能为空"): + command_application.dispatch_command(" ", user_id="1", publish_event=lambda *_args: None) + with pytest.raises(ValueError, match="不存在"): + command_application.dispatch_command("/missing", user_id="1", publish_event=lambda *_args: None) + finally: + command_application.reset_command_class() + + with pytest.raises(RuntimeError, match="未初始化"): + command_application.get_command_object() + + +def test_music_projection_and_domain_projection(): + """音乐和豆瓣投影应保留稳定身份并裁剪大字段。""" + track = MusicInfo(media_source=MediaSource.MusicBrainz, media_id="track", title="Track", artists=["Artist"], year=2024) + assert simplify_music_info(track)["title"] == "Track" + album = MusicAlbumInfo(media_source=MediaSource.MusicBrainz, media_id="album", title="Album", artists=["Artist"], tracks=[track] * 3, releases=[MusicRelease(media_id="release", title="Release")]) + assert simplify_music_album(album, track_limit=2)["tracks_truncated"] is True + artist = MusicArtistInfo(media_source=MediaSource.MusicBrainz, media_id="artist", name="Artist", raw_data={"secret": 1}) + assert simplify_music_artist(artist)["subscribable"] is False + projected = project_douban({}, {"id": "1", "title": "Movie", "subtype": "movie", "rating": {"value": 8.5}, "pic": {"large": "poster"}}) + assert projected["poster_path"] == "poster" + assert projected["douban_id"] == "1" + assert project_douban({}, {}) == {} + + +def test_download_task_services_validate_and_delegate(monkeypatch): + """下载任务服务应补齐历史媒体并严格校验高级修改。""" + torrent = SimpleNamespace(hash="a" * 40, downloader="qb") + history = SimpleNamespace(media_source=MediaSource.TMDB, media_id="1", type="电影", title="Movie", seasons="1", episodes="2", poster="p", image="b", torrent_site="site", userid="u", username="name") + list_torrents = MagicMock(return_value=[torrent]) + service = DownloadTaskService(list_torrents, lambda _hashes: {torrent.hash: history}, MagicMock(return_value=True), MagicMock(return_value=True), MagicMock(return_value=True)) + assert service.downloading()[0].media.title == "Movie" + assert service.set_downloading(torrent.hash, "start") is True + assert service.set_downloading(torrent.hash, "bad") is False + assert service.remove_downloading(torrent.hash) is True + mutation = DownloadTaskMutationService(list_torrents=lambda **_kwargs: [torrent], set_tags=MagicMock(return_value=True), set_downloading=MagicMock(return_value=True), update_torrent=MagicMock(return_value={"limits": True, "trackers": False})) + assert mutation.update(hash_value=torrent.hash, action="start", tags=["tag"], download_limit=1)["results"] + with pytest.raises(ValueError, match="hash"): + mutation.update(hash_value="bad", action="start") + with pytest.raises(ValueError, match="至少"): + mutation.update(hash_value=torrent.hash) + with pytest.raises(ValueError, match="action"): + mutation.update(hash_value=torrent.hash, action="pause")