mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-29 03:18:14 +08:00
961 lines
38 KiB
Python
961 lines
38 KiB
Python
"""
|
||
Agent API 网关的输入、权限、执行结果与 MCP 发现契约测试。
|
||
"""
|
||
|
||
import asyncio
|
||
import json
|
||
import re
|
||
from types import SimpleNamespace
|
||
from unittest.mock import AsyncMock, Mock, patch
|
||
|
||
import pytest
|
||
from langchain_core.utils.function_calling import convert_to_openai_tool
|
||
|
||
from app.agent.api.arguments import api_input_contract
|
||
|
||
# 策略包根通过 __getattr__ 惰性导出，Pylint 无法静态解析。
|
||
# pylint: disable=no-name-in-module
|
||
from app.agent.policy import (
|
||
DEFAULT_TOOL_POLICY_REGISTRY,
|
||
ActionEffect,
|
||
ConfirmationMode,
|
||
PrincipalRole,
|
||
RecoveryMode,
|
||
)
|
||
from app.agent.policy.api import (
|
||
API_EXTENDED_OPERATION_SPECS,
|
||
API_FIRST_BATCH_OPERATION_SPECS,
|
||
API_MUSIC_OPERATION_SPECS,
|
||
API_OPERATION_ROUTES,
|
||
API_OPERATION_SPECS,
|
||
API_PARITY_OPERATION_SPECS,
|
||
API_SYSTEM_OPERATION_SPECS,
|
||
)
|
||
from app.agent.policy.mcp import _apply_field_descriptions
|
||
from app.agent.tools.factory import MoviePilotToolFactory
|
||
from app.agent.tools.impl.agent_task import AgentTaskTool
|
||
from app.agent.tools.impl.api import MoviePilotApiInput, MoviePilotApiTool
|
||
from app.agent.tools.impl.execute_command import ExecuteCommandTool
|
||
from app.agent.tools.manager import MoviePilotToolsManager
|
||
from app.db.adapters.user import TransactionalUserRepository
|
||
from app.db.models.user import User
|
||
from app.db.session import SessionFactory, async_session_scope
|
||
from app.schemas.types import NotificationChannel
|
||
|
||
|
||
def test_api_operation_registry_matches_migration_batches() -> None:
|
||
"""
|
||
API 操作注册表必须覆盖各迁移批次且每项具有固定路由。
|
||
"""
|
||
assert len(API_FIRST_BATCH_OPERATION_SPECS) == 53
|
||
assert len(API_PARITY_OPERATION_SPECS) == 17
|
||
assert len(API_MUSIC_OPERATION_SPECS) == 10
|
||
assert len(API_SYSTEM_OPERATION_SPECS) == 7
|
||
assert len(API_EXTENDED_OPERATION_SPECS) == 145
|
||
assert len(API_OPERATION_SPECS) == 232
|
||
assert {spec.operation_id for spec in API_OPERATION_SPECS} == set(API_OPERATION_ROUTES)
|
||
assert {
|
||
"download.list",
|
||
"download.update",
|
||
"download.delete",
|
||
"downloaders.list",
|
||
}.isdisjoint(API_OPERATION_ROUTES)
|
||
assert {
|
||
"plugin.source.options",
|
||
"plugin.source.install",
|
||
"plugin.source.change",
|
||
"download.tasks.active",
|
||
"download.clients",
|
||
"download.paths",
|
||
"download.history.list",
|
||
"library.latest",
|
||
"system.versions",
|
||
"system.update.status",
|
||
"system.update.check",
|
||
"system.update.download",
|
||
"system.restart",
|
||
"system.update.install",
|
||
"system.upgrade.dev",
|
||
"dashboard.system",
|
||
"media.sources",
|
||
"search.title",
|
||
"site.add",
|
||
"subscription.get",
|
||
"storage.rename",
|
||
"transfer.manual_reviews",
|
||
"workflow.create",
|
||
"torrent.cache.get",
|
||
"database.backups.list",
|
||
"system.module.list",
|
||
"plugin.clone",
|
||
}.issubset(API_OPERATION_ROUTES)
|
||
|
||
|
||
def test_classification_operations_expose_versioned_policy_contract() -> None:
|
||
"""
|
||
媒体自动分类必须只暴露新版查询、校验、预览和版本化写入合同。
|
||
"""
|
||
expected_routes = {
|
||
"media.classification.fields": ("GET", "/api/v1/media/classification/fields"),
|
||
"media.classification.policy.get": ("GET", "/api/v1/media/classification/policy"),
|
||
"media.classification.policy.validate": ("POST", "/api/v1/media/classification/validate"),
|
||
"media.classification.policy.preview": ("POST", "/api/v1/media/classification/preview"),
|
||
"media.classification.policy.impact": ("POST", "/api/v1/media/classification/impact"),
|
||
"media.classification.policy.history": ("GET", "/api/v1/media/classification/history"),
|
||
"media.classification.policy.update": ("PUT", "/api/v1/media/classification/policy"),
|
||
"media.classification.policy.rollback": (
|
||
"POST",
|
||
"/api/v1/media/classification/rollback/{revision}",
|
||
),
|
||
}
|
||
assert {
|
||
operation_id: (route.method, route.path)
|
||
for operation_id, route in API_OPERATION_ROUTES.items()
|
||
if operation_id.startswith("media.classification.")
|
||
} == expected_routes
|
||
assert {"media.categories", "media.category.config.get"}.isdisjoint(API_OPERATION_ROUTES)
|
||
|
||
specs = {spec.operation_id: spec for spec in API_OPERATION_SPECS}
|
||
assert specs["media.classification.policy.update"].effect is ActionEffect.REVERSIBLE_WRITE
|
||
assert specs["media.classification.policy.rollback"].effect is ActionEffect.REVERSIBLE_WRITE
|
||
assert specs["media.classification.policy.update"].required_role is PrincipalRole.SYSTEM_ADMIN
|
||
assert specs["media.classification.policy.rollback"].required_role is PrincipalRole.SYSTEM_ADMIN
|
||
assert specs["media.classification.policy.update"].confirmation is ConfirmationMode.REQUIRED
|
||
assert specs["media.classification.policy.rollback"].confirmation is ConfirmationMode.REQUIRED
|
||
assert specs["media.classification.policy.update"].recovery is RecoveryMode.TRANSACTION
|
||
assert specs["media.classification.policy.rollback"].recovery is RecoveryMode.TRANSACTION
|
||
|
||
schema = MoviePilotApiTool(session_id="session", user_id="api_user").get_mcp_input_schema()
|
||
branches = {
|
||
item["properties"]["operation_id"]["const"]: item
|
||
for item in schema["oneOf"]
|
||
}
|
||
assert branches["media.classification.policy.update"]["properties"]["body"]["$ref"].endswith(
|
||
"/ClassificationPolicyPublishRequest"
|
||
)
|
||
assert branches["media.classification.policy.preview"]["properties"]["body"]["$ref"].endswith(
|
||
"/ClassificationPreviewRequest"
|
||
)
|
||
assert branches["media.classification.policy.rollback"]["properties"]["path_params"]["required"] == [
|
||
"revision"
|
||
]
|
||
|
||
|
||
def test_api_tool_message_displays_secret_safe_major_parameters() -> None:
|
||
"""
|
||
啰嗦模式提示应展示 API 主要参数，同时对设置凭据值脱敏。
|
||
"""
|
||
tool = MoviePilotApiTool(session_id="session", user_id="api_user")
|
||
|
||
list_message = tool.get_tool_message(
|
||
operation_id="subscription.list",
|
||
query={"page": 1, "count": 20},
|
||
)
|
||
secret_message = tool.get_tool_message(
|
||
operation_id="config.system.update",
|
||
body={
|
||
"setting_key": "OPENAI_API_KEY",
|
||
"value": "sk-secret-value",
|
||
"operation": "replace",
|
||
},
|
||
)
|
||
|
||
assert list_message == (
|
||
'调用 MoviePilot API：subscription.list，主要参数：'
|
||
'{"query": {"page": 1, "count": 20}}'
|
||
)
|
||
assert "OPENAI_API_KEY" in secret_message
|
||
assert '"value": "***"' in secret_message
|
||
assert "sk-secret-value" not in secret_message
|
||
|
||
|
||
def test_music_operations_expose_bidirectional_artist_album_navigation() -> None:
|
||
"""
|
||
音乐 Skill 必须完整暴露作品到作者、作者到作品及关联浏览合同。
|
||
"""
|
||
schema = MoviePilotApiTool(session_id="session", user_id="api_user").get_mcp_input_schema()
|
||
branches = {
|
||
item["properties"]["operation_id"]["const"]: item
|
||
for item in schema["oneOf"]
|
||
}
|
||
|
||
assert {
|
||
"music.recognize",
|
||
"music.explore",
|
||
"music.album.get",
|
||
"music.album.related",
|
||
"music.artist.get",
|
||
"music.artist.albums",
|
||
"music.artist.related",
|
||
"music.cache.get",
|
||
"music.cache.delete",
|
||
"music.cache.clear",
|
||
}.issubset(branches)
|
||
album_path = branches["music.album.get"]["properties"]["path_params"]
|
||
assert album_path["required"] == ["album_id"]
|
||
artist_albums = branches["music.artist.albums"]
|
||
assert artist_albums["properties"]["path_params"]["required"] == ["artist_id"]
|
||
album_type = artist_albums["properties"]["query"]["properties"]["album_type"]
|
||
assert "single" in str(album_type)
|
||
assert branches["music.cache.delete"]["properties"]["path_params"]["required"] == [
|
||
"cache_key"
|
||
]
|
||
|
||
|
||
def test_mcp_tools_list_preserves_all_moviepilot_api_operation_branches() -> None:
|
||
"""
|
||
外部 MCP tools/list 必须返回全部 operation 的精确 oneOf，而不是通用字典。
|
||
"""
|
||
manager = MoviePilotToolsManager(session_id="session", user_id="api_user")
|
||
manager.tools = [MoviePilotApiTool(session_id="session", user_id="api_user")]
|
||
|
||
definition = manager.list_tools()[0]
|
||
operation_ids = {
|
||
branch["properties"]["operation_id"]["const"]
|
||
for branch in definition.input_schema["oneOf"]
|
||
}
|
||
|
||
assert definition.name == "moviepilot_api"
|
||
assert operation_ids == set(API_OPERATION_ROUTES)
|
||
|
||
|
||
def test_local_agent_tool_restores_json_encoded_object_bodies() -> None:
|
||
"""
|
||
本地 Agent 工具 schema 不得把 JSON 对象字符串化作为合法请求体。
|
||
|
||
部分模型在 body 声明为 anyOf 联合类型时会把整个请求体编码成 JSON 字符串；
|
||
规范化阶段将其还原为原生对象，非 JSON 的非法字符串仍按合同报错。
|
||
"""
|
||
tool = MoviePilotApiTool(session_id="session", user_id="api_user")
|
||
schema = tool.tool_call_schema
|
||
assert not isinstance(schema, dict)
|
||
body_schema = schema.model_json_schema()["properties"]["body"]
|
||
body_branches = body_schema["anyOf"]
|
||
body_types = {branch.get("type") for branch in body_branches}
|
||
|
||
assert {"object", "array", "string", "null"}.issubset(body_types)
|
||
assert any(branch.get("const") == "dev" for branch in body_branches)
|
||
assert all(
|
||
branch.get("type") != "string" or branch.get("const") == "dev"
|
||
for branch in body_branches
|
||
)
|
||
|
||
body = {
|
||
"torrent_in": {
|
||
"title": "示例种子",
|
||
"enclosure": "https://example.com/test.torrent",
|
||
},
|
||
}
|
||
canonical = tool.canonical_arguments({"operation_id": "download.add", "body": body})
|
||
assert canonical["body"]["torrent_in"]["enclosure"] == body["torrent_in"]["enclosure"]
|
||
|
||
# 模型把 body 编码为 JSON 字符串时，规范化阶段还原为原生对象。
|
||
restored = tool.canonical_arguments(
|
||
{
|
||
"operation_id": "download.add",
|
||
"body": json.dumps(body, ensure_ascii=False),
|
||
}
|
||
)
|
||
assert isinstance(restored["body"], dict)
|
||
assert restored["body"]["torrent_in"]["enclosure"] == body["torrent_in"]["enclosure"]
|
||
|
||
# LangChain 入模同样必须还原，否则 canonical_arguments 通过后仍会在 Pydantic 阶段失败。
|
||
restored_model = MoviePilotApiInput.model_validate(
|
||
{
|
||
"operation_id": "download.add",
|
||
"body": json.dumps(body, ensure_ascii=False),
|
||
}
|
||
)
|
||
assert isinstance(restored_model.body, dict)
|
||
assert restored_model.body["torrent_in"]["enclosure"] == body["torrent_in"]["enclosure"]
|
||
|
||
# 固定的 system.upgrade.dev 字面量继续作为唯一合法字符串请求体。
|
||
dev_model = MoviePilotApiInput.model_validate(
|
||
{"operation_id": "system.upgrade.dev", "body": "dev"}
|
||
)
|
||
assert dev_model.body == "dev"
|
||
|
||
# 无法还原为对象/数组的非法字符串仍按合同报错。
|
||
with pytest.raises(ValueError):
|
||
tool.canonical_arguments(
|
||
{
|
||
"operation_id": "download.add",
|
||
"body": "not-a-json-object",
|
||
}
|
||
)
|
||
|
||
|
||
def test_operation_body_shapes_fit_local_agent_input_schema() -> None:
|
||
"""
|
||
所有 MCP operation 请求体必须可由内置 Agent 输入类型表达。
|
||
"""
|
||
tool = MoviePilotApiTool(session_id="session", user_id="api_user")
|
||
api_schema = tool.get_mcp_input_schema()
|
||
definitions = api_schema.get("$defs", {})
|
||
string_bodies = set()
|
||
|
||
for operation in api_schema["oneOf"]:
|
||
properties = operation.get("properties", {})
|
||
body_schema = properties.get("body")
|
||
if body_schema is None:
|
||
continue
|
||
while "$ref" in body_schema:
|
||
body_schema = definitions[body_schema["$ref"].rsplit("/", 1)[-1]]
|
||
variants = body_schema.get("anyOf", body_schema.get("oneOf", [body_schema]))
|
||
operation_id = properties["operation_id"]["const"]
|
||
for variant in variants:
|
||
while "$ref" in variant:
|
||
variant = definitions[variant["$ref"].rsplit("/", 1)[-1]]
|
||
if variant.get("type") == "string":
|
||
string_bodies.add((operation_id, variant.get("const")))
|
||
else:
|
||
assert variant.get("type") in {"object", "array", "null"}
|
||
|
||
assert string_bodies == {("system.upgrade.dev", "dev")}
|
||
|
||
|
||
def test_local_api_schema_forbids_flattened_arguments_and_examples_match_contracts() -> None:
|
||
"""
|
||
模型实际接收的 schema 禁止额外顶层字段，内联示例均通过真实 operation 合同。
|
||
"""
|
||
tool = MoviePilotApiTool(session_id="session", user_id="1")
|
||
definition = convert_to_openai_tool(tool)["function"]
|
||
schema = definition["parameters"]
|
||
assert definition["description"] == tool.description
|
||
assert schema["additionalProperties"] is False
|
||
assert set(schema["properties"]) == {"operation_id", "path_params", "query", "body"}
|
||
with pytest.raises(ValueError):
|
||
MoviePilotApiInput.model_validate({"operation_id": "site.rss", "page": 1})
|
||
|
||
examples = tool.description.split("示例中的标识须替换为此前结果中的实际标识：")[1]
|
||
decoder = json.JSONDecoder()
|
||
operations = []
|
||
for _ in range(3):
|
||
example, end = decoder.raw_decode(examples)
|
||
operations.append(tool.canonical_arguments(example)["operation_id"])
|
||
examples = examples[end:].lstrip("； ")
|
||
assert operations == ["media.detail", "subscription.execution.list", "site.rss"]
|
||
|
||
|
||
@pytest.mark.parametrize(("arguments", "expected"), [
|
||
({"operation_id": "media.detail", "media_id": "secret-value", "media_source": "tmdb"},
|
||
["media_id 应移入 path_params.media_id", "media_source 应移入 query.media_source"]),
|
||
({"operation_id": "subscription.execution.list", "limit": 2}, ["limit 应移入 query.limit"]),
|
||
({"operation_id": "site.rss", "query": {"site_id": "secret-value"}}, ["query", "site_id"]),
|
||
({"operation_id": "media.detail"}, ["path_params.media_id"]),
|
||
({"operation_id": "media.detail", "path_params": {"media_id": "27205"}}, ["query.media_source"]),
|
||
({"operation_id": "subscription.execution.list", "query": {"limit": "secret-value"}}, ["query.limit"]),
|
||
({"operation_id": "site.list", "query": {"status": "secret-value"}}, ["query.status"]),
|
||
({"operation_id": "download.add", "torrent_in": {}}, ["torrent_in 应移入 body.torrent_in"]),
|
||
({"operation_id": "config.system.update", "body": {"value": "secret-value"}}, ["body.setting_key"]),
|
||
({"operation_id": "site.rss", "https://secret-value.invalid/": "secret-value"}, ["字段名已省略"]),
|
||
])
|
||
@pytest.mark.asyncio
|
||
async def test_gateway_raw_tool_inputs_return_actionable_errors_without_execution(arguments, expected) -> None:
|
||
"""
|
||
覆盖 LangChain 和直接调用入口，错误字段不能静默丢弃、执行或泄露参数值。
|
||
"""
|
||
executor = AsyncMock()
|
||
gateway = MoviePilotApiTool(session_id="session", user_id="1", executor=executor)
|
||
message = await gateway.ainvoke({"name": gateway.name, "args": arguments, "id": "invalid", "type": "tool_call"})
|
||
assert message.status == "error"
|
||
for result in (message.content, await gateway.run(**arguments)):
|
||
payload = json.loads(result)
|
||
assert payload["error"] == "invalid_input"
|
||
assert payload["input_contract"]["operation_id"] == arguments["operation_id"]
|
||
assert all(item in payload["message"] for item in expected)
|
||
assert "secret-value" not in result
|
||
executor.execute.assert_not_awaited()
|
||
|
||
|
||
def test_mcp_collection_contract_distinguishes_exact_and_unavailable_totals() -> None:
|
||
"""
|
||
MCP 必须说明缺省全量、精确总数和外部无总数三种集合语义。
|
||
"""
|
||
schema = MoviePilotApiTool(session_id="session", user_id="api_user").get_mcp_input_schema()
|
||
branches = {
|
||
item["properties"]["operation_id"]["const"]: item
|
||
for item in schema["oneOf"]
|
||
}
|
||
|
||
subscription = branches["subscription.list"]
|
||
subscription_query = subscription["properties"]["query"]["properties"]
|
||
assert "default" not in subscription_query["page"]
|
||
assert "default" not in subscription_query["count"]
|
||
assert subscription["x-moviepilot-collection"] == {
|
||
"body_shape": "list",
|
||
"result_count_field": "collection.result_count",
|
||
"total_count_field": "collection.total_count",
|
||
"default_pagination": "unpaginated",
|
||
}
|
||
assert "page=1 和 count=1" in subscription["description"]
|
||
assert "不要仅因条目数据被截断就查询数据库" in subscription["description"]
|
||
|
||
storage = branches["storage.list"]
|
||
assert {"page", "count"}.issubset(storage["properties"]["query"]["properties"])
|
||
assert storage["x-moviepilot-collection"]["total_count_field"] == (
|
||
"collection.total_count"
|
||
)
|
||
|
||
for operation_id in ("subscription.history", "download.history.list"):
|
||
local_page = branches[operation_id]["x-moviepilot-collection"]
|
||
assert local_page["total_count_field"] == "collection.total_count"
|
||
assert local_page["default_pagination"] == "endpoint-defined"
|
||
assert "默认值继续生效" in branches[operation_id]["description"]
|
||
|
||
for operation_id in ("plugin.installed", "plugin.market"):
|
||
local_page = branches[operation_id]["x-moviepilot-collection"]
|
||
assert local_page["total_count_field"] == "collection.total_count"
|
||
assert local_page["default_pagination"] == "unpaginated"
|
||
assert "同时省略 page 和 count" in branches[operation_id]["description"]
|
||
assert "default" not in branches[operation_id]["properties"]["query"]["properties"]["max_results"]
|
||
|
||
media_search = branches["media.search"]["x-moviepilot-collection"]
|
||
assert media_search["result_count_field"] == "collection.result_count"
|
||
assert media_search["total_count_field"] is None
|
||
assert "未提供总数" in branches["media.search"]["description"]
|
||
|
||
transfer = branches["transfer.history"]["x-moviepilot-collection"]
|
||
assert transfer["body_shape"] == "page_object"
|
||
assert transfer["items_field"] == "data.list"
|
||
assert transfer["total_count_field"] == "data.total"
|
||
|
||
|
||
def test_filter_read_parameters_are_query_fields_not_get_request_bodies() -> None:
|
||
"""
|
||
规则读取筛选列表必须作为 query 参数公开，避免 Agent 构造无语义 GET body。
|
||
"""
|
||
schema = MoviePilotApiTool(session_id="session", user_id="api_user").get_mcp_input_schema()
|
||
branches = {
|
||
item["properties"]["operation_id"]["const"]: item
|
||
for item in schema["oneOf"]
|
||
}
|
||
|
||
assert "body" not in branches["filter.builtin"]["properties"]
|
||
assert "rule_ids" in branches["filter.builtin"]["properties"]["query"]["properties"]
|
||
assert "body" not in branches["filter.custom"]["properties"]
|
||
assert "rule_ids" in branches["filter.custom"]["properties"]["query"]["properties"]
|
||
assert "body" not in branches["filter.groups"]["properties"]
|
||
assert "group_names" in branches["filter.groups"]["properties"]["query"]["properties"]
|
||
|
||
|
||
def test_plugin_operations_expose_discovery_before_precise_writes() -> None:
|
||
"""
|
||
插件配置和来源操作必须给 Agent 可先发现再精确写入的完整合同。
|
||
"""
|
||
schema = MoviePilotApiTool(session_id="session", user_id="api_user").get_mcp_input_schema()
|
||
branches = {
|
||
item["properties"]["operation_id"]["const"]: item
|
||
for item in schema["oneOf"]
|
||
}
|
||
|
||
installed_query = branches["plugin.installed"]["properties"]["query"]
|
||
assert installed_query["properties"]["state"]["const"] == "installed"
|
||
assert "query" in installed_query["properties"]
|
||
max_results = installed_query["properties"]["max_results"]
|
||
integer_variant = next(item for item in max_results["anyOf"] if item.get("type") == "integer")
|
||
assert integer_variant["maximum"] == 200
|
||
|
||
config_get_path = API_OPERATION_ROUTES["plugin.config.get"].path
|
||
assert config_get_path == "/api/v1/plugin/form/{plugin_id}"
|
||
config_body = branches["plugin.config.update"]["properties"]["body"]
|
||
assert config_body["minProperties"] == 1
|
||
assert "先调用 plugin.config.get" in config_body["description"]
|
||
|
||
assert branches["plugin.source.install"]["properties"]["body"]["$ref"].endswith(
|
||
"/PluginSourceInstallRequest"
|
||
)
|
||
assert branches["plugin.source.change"]["properties"]["body"]["$ref"].endswith(
|
||
"/PluginSourceChangeRequest"
|
||
)
|
||
|
||
|
||
def test_dev_upgrade_mcp_schema_requires_exact_scalar_body() -> None:
|
||
"""
|
||
Dev 更新必须暴露精确字符串 body，不能回退为任意 JSON。
|
||
"""
|
||
schema = MoviePilotApiTool(
|
||
session_id="session",
|
||
user_id="api_user",
|
||
).get_mcp_input_schema()
|
||
branch = next(
|
||
item
|
||
for item in schema["oneOf"]
|
||
if item["properties"]["operation_id"].get("const") == "system.upgrade.dev"
|
||
)
|
||
|
||
assert branch["properties"]["body"]["const"] == "dev"
|
||
assert branch["properties"]["body"]["type"] == "string"
|
||
assert "body" in branch["required"]
|
||
|
||
|
||
def test_gateway_forwards_exact_scalar_body() -> None:
|
||
"""
|
||
网关必须原样传递少数固定 operation 声明的 JSON 标量请求体。
|
||
"""
|
||
executor = AsyncMock()
|
||
executor.execute.return_value = json.dumps({"success": True})
|
||
gateway = MoviePilotApiTool(
|
||
session_id="session",
|
||
user_id="api_user",
|
||
executor=executor,
|
||
)
|
||
gateway.set_agent_context({"is_admin": True})
|
||
|
||
result = asyncio.run(
|
||
gateway.run(
|
||
operation_id="system.upgrade.dev",
|
||
body="dev",
|
||
)
|
||
)
|
||
|
||
assert json.loads(result)["success"] is True
|
||
executor.execute.assert_awaited_once_with(
|
||
"system.upgrade.dev",
|
||
path_params=None,
|
||
query=None,
|
||
body="dev",
|
||
)
|
||
|
||
|
||
def test_system_settings_mcp_schema_explains_both_setting_sources() -> None:
|
||
"""
|
||
外部 MCP Client 应直接看到系统设置发现与精确更新参数语义。
|
||
"""
|
||
schema = MoviePilotApiTool(
|
||
session_id="session",
|
||
user_id="api_user",
|
||
).get_mcp_input_schema()
|
||
branches = {
|
||
branch["properties"]["operation_id"]["const"]: branch
|
||
for branch in schema["oneOf"]
|
||
}
|
||
query = branches["config.system.get"]["properties"]["query"]["properties"]
|
||
update_ref = branches["config.system.update"]["properties"]["body"]["$ref"]
|
||
update = schema["$defs"][update_ref.rsplit("/", 1)[-1]]["properties"]
|
||
|
||
assert "Settings 字段名" in query["setting_key"]["description"]
|
||
assert "systemconfig" in query["group"]["description"]
|
||
assert "确认" in query["show_secrets"]["description"]
|
||
assert "config.system.get" in update["setting_key"]["description"]
|
||
assert "upsert_list_item" in update["operation"]["description"]
|
||
assert "通知场景" in update["match_field"]["description"]
|
||
assert "type" in update["match_field"]["description"]
|
||
|
||
|
||
def test_api_mcp_schema_gives_every_field_concrete_chinese_guidance() -> None:
|
||
"""
|
||
MCP 契约字段必须提供具体的中文说明，不能回退为抽象占位说明。
|
||
"""
|
||
schema = MoviePilotApiTool(
|
||
session_id="session",
|
||
user_id="api_user",
|
||
).get_mcp_input_schema()
|
||
descriptions = []
|
||
|
||
def collect(node) -> None:
|
||
"""
|
||
递归收集对象字段说明并断言没有遗漏。
|
||
"""
|
||
if isinstance(node, dict):
|
||
properties = node.get("properties")
|
||
if isinstance(properties, dict):
|
||
for field_name, field_schema in properties.items():
|
||
assert isinstance(field_schema, dict), field_name
|
||
description = field_schema.get("description")
|
||
assert isinstance(description, str) and description.strip(), field_name
|
||
descriptions.append(description)
|
||
for value in node.values():
|
||
collect(value)
|
||
elif isinstance(node, list):
|
||
for value in node:
|
||
collect(value)
|
||
|
||
collect(schema)
|
||
rendered = "\n".join(descriptions)
|
||
assert "按接口模型语义传值" not in rendered
|
||
assert "declared by the selected operation" not in rendered
|
||
assert "use the matching oneOf branch for its exact type" not in rendered
|
||
assert all(re.search(r"[\u3400-\u9fff]", item) for item in descriptions)
|
||
|
||
|
||
def test_policy_classifies_api_operation_by_operation_id() -> None:
|
||
"""
|
||
网关策略必须按 operation ID 分类，而不能把整个网关视为普通读取。
|
||
"""
|
||
delete_policy = DEFAULT_TOOL_POLICY_REGISTRY.resolve(
|
||
tool_name="moviepilot_api",
|
||
arguments={"operation_id": "subscription.delete"},
|
||
requires_admin=False,
|
||
)
|
||
unknown_policy = DEFAULT_TOOL_POLICY_REGISTRY.resolve(
|
||
tool_name="moviepilot_api",
|
||
arguments={"operation_id": "unknown.operation"},
|
||
requires_admin=False,
|
||
)
|
||
|
||
assert delete_policy.effect is ActionEffect.DESTRUCTIVE_WRITE
|
||
assert delete_policy.required_role is PrincipalRole.USER
|
||
assert delete_policy.confirmation is ConfirmationMode.REQUIRED
|
||
assert unknown_policy.machine_allowed is False
|
||
|
||
|
||
def test_gateway_forwards_structured_arguments_to_api_executor() -> None:
|
||
"""
|
||
网关应只把结构化白名单调用转发给固定 API 执行器。
|
||
"""
|
||
executor = AsyncMock()
|
||
executor.execute.return_value = json.dumps({"success": True})
|
||
gateway = MoviePilotApiTool(
|
||
session_id="session",
|
||
user_id="1",
|
||
executor=executor,
|
||
)
|
||
|
||
result = asyncio.run(
|
||
gateway.run(
|
||
operation_id="media.search",
|
||
query={"page": 1, "title": "示例"},
|
||
)
|
||
)
|
||
|
||
assert json.loads(result)["success"] is True
|
||
executor.execute.assert_awaited_once_with(
|
||
"media.search",
|
||
path_params=None,
|
||
query={"page": 1, "title": "示例"},
|
||
body=None,
|
||
)
|
||
|
||
|
||
def test_gateway_rejects_unknown_operation() -> None:
|
||
"""
|
||
未知 operation ID 必须在调用固定 API 执行器前稳定失败。
|
||
"""
|
||
gateway = MoviePilotApiTool(session_id="session", user_id="user")
|
||
|
||
result = asyncio.run(gateway.run(operation_id="arbitrary.http", body={"path": "/admin"}))
|
||
|
||
assert '"success": false' in result
|
||
assert "unknown_operation" in result
|
||
|
||
|
||
def test_gateway_returns_operation_contract_when_input_is_invalid() -> None:
|
||
"""
|
||
单个 operation 的参数错误必须返回可直接纠正调用的精确合同。
|
||
"""
|
||
executor = AsyncMock()
|
||
gateway = MoviePilotApiTool(
|
||
session_id="session",
|
||
user_id="1",
|
||
executor=executor,
|
||
)
|
||
|
||
result = asyncio.run(
|
||
gateway.run(
|
||
operation_id="site.list",
|
||
query={"status": "enabled"},
|
||
)
|
||
)
|
||
payload = json.loads(result)
|
||
|
||
assert payload["error"] == "invalid_input"
|
||
assert payload["operation_id"] == "site.list"
|
||
assert payload["input_contract"]["allowed_arguments"] == ["query"]
|
||
status = payload["input_contract"]["query"]["fields"]["status"]
|
||
assert status["enum"] == ["active", "inactive", "all"]
|
||
assert "enabled" not in json.dumps(payload, ensure_ascii=False)
|
||
executor.execute.assert_not_awaited()
|
||
|
||
|
||
def test_input_contract_truncates_long_field_description_without_leaking_values() -> None:
|
||
"""
|
||
超长字段说明应保持有界，且只回传合同文本。
|
||
"""
|
||
description = "字段说明 " * 40
|
||
schema = {
|
||
"oneOf": [{
|
||
"properties": {
|
||
"operation_id": {"const": "example.read"},
|
||
"query": {"type": "object", "properties": {
|
||
"keyword": {"type": "string", "description": description}
|
||
}},
|
||
}
|
||
}]
|
||
}
|
||
|
||
contract = api_input_contract("example.read", schema)
|
||
bounded = contract["query"]["fields"]["keyword"]["description"]
|
||
|
||
assert len(bounded) <= 180
|
||
assert bounded.endswith("…")
|
||
assert bounded == description[:179].rstrip() + "…"
|
||
|
||
|
||
def test_gateway_rejects_admin_operation_for_non_admin_before_http() -> None:
|
||
"""
|
||
管理员 operation 必须在网关层拒绝普通用户，不能只依赖最终端点。
|
||
"""
|
||
executor = AsyncMock()
|
||
gateway = MoviePilotApiTool(
|
||
session_id="session",
|
||
user_id="1",
|
||
executor=executor,
|
||
)
|
||
gateway.set_agent_context({"is_admin": False})
|
||
|
||
result = asyncio.run(
|
||
gateway.run(
|
||
operation_id="config.system.get",
|
||
query={"group": "settings"},
|
||
)
|
||
)
|
||
|
||
assert json.loads(result)["error"] == "permission_denied"
|
||
executor.execute.assert_not_awaited()
|
||
|
||
|
||
def test_gateway_resolves_http_manager_to_persisted_superuser() -> None:
|
||
"""
|
||
MCP/HTTP 管理入口应绑定真实管理员，而不是伪造 api_user 身份。
|
||
"""
|
||
gateway = MoviePilotApiTool(session_id="session", user_id="api_user")
|
||
gateway.set_message_attr(channel=None, source="api", username="API Client")
|
||
gateway.set_agent_context({"is_admin": True})
|
||
|
||
with patch(
|
||
"app.application.security.auth.build_superuser_token_payload",
|
||
return_value=SimpleNamespace(
|
||
sub=7,
|
||
username="admin",
|
||
super_user=True,
|
||
),
|
||
):
|
||
identity = asyncio.run(gateway._resolve_api_identity())
|
||
|
||
assert identity == ("7", "admin", True)
|
||
|
||
|
||
def test_gateway_maps_verified_channel_admin_only_for_admin_operation() -> None:
|
||
"""
|
||
通知渠道管理员执行管理员 operation 时应延续旧工具的管理员语义。
|
||
"""
|
||
gateway = MoviePilotApiTool(session_id="session", user_id="telegram-user")
|
||
gateway.set_message_attr(
|
||
channel="telegram",
|
||
source="main-bot",
|
||
username="channel-user",
|
||
)
|
||
gateway.set_agent_context({"is_admin": True})
|
||
|
||
with patch(
|
||
"app.application.security.auth.build_superuser_token_payload",
|
||
return_value=SimpleNamespace(
|
||
sub=7,
|
||
username="admin",
|
||
super_user=True,
|
||
),
|
||
):
|
||
identity = asyncio.run(
|
||
gateway._resolve_api_identity(require_system_admin=True)
|
||
)
|
||
|
||
assert identity == ("7", "admin", True)
|
||
|
||
|
||
@pytest.mark.parametrize("channel", [NotificationChannel.QQ, NotificationChannel.Feishu])
|
||
@pytest.mark.parametrize("bound_suffix,other_suffix", [("openid", "userid"), ("userid", "openid")])
|
||
@pytest.mark.parametrize("other_value", [None, "", "different-id", "channel-id"])
|
||
def test_gateway_resolves_either_channel_binding(db, channel, bound_suffix, other_suffix, other_value) -> None:
|
||
"""
|
||
任一候选字段匹配即可绑定真实用户，另一字段缺失、为空或不同时仍可使用。
|
||
"""
|
||
prefix = channel.name.lower()
|
||
bindings = {f"{prefix}_{bound_suffix}": "channel-id"}
|
||
if other_value is not None:
|
||
bindings[f"{prefix}_{other_suffix}"] = other_value
|
||
user = db.add(User(name="channel-owner", is_active=True, is_superuser=False, settings=bindings))
|
||
users = TransactionalUserRepository(sync_session=SessionFactory, async_session=async_session_scope)
|
||
gateway = MoviePilotApiTool(
|
||
session_id="session", user_id="channel-id", data=SimpleNamespace(users=users),
|
||
)
|
||
gateway.set_message_attr(channel=channel.value, source="main-bot", username="channel-nickname")
|
||
|
||
identity = asyncio.run(gateway._resolve_api_identity())
|
||
|
||
assert identity == (str(user.id), "channel-owner", False)
|
||
|
||
|
||
@pytest.mark.parametrize("is_channel_admin", [False, True])
|
||
@pytest.mark.parametrize("user_id,username", [(11, "alice"), (12, "bob")])
|
||
@pytest.mark.parametrize(
|
||
"operation_id,arguments",
|
||
[
|
||
("subscription.add", {"body": {"name": "示例"}}),
|
||
("subscription.update", {"body": {"id": 1}}),
|
||
("subscription.delete", {"path_params": {"subscribe_id": 1}}),
|
||
],
|
||
)
|
||
def test_subscription_writes_keep_bound_channel_identity(
|
||
is_channel_admin, user_id, username, operation_id, arguments,
|
||
) -> None:
|
||
"""
|
||
不同飞书发送者的订阅写入使用各自绑定用户，不因渠道管理员资格提权。
|
||
"""
|
||
users = SimpleNamespace(
|
||
find_name_by_bindings=Mock(return_value=username),
|
||
async_get_by_name=AsyncMock(return_value=SimpleNamespace(
|
||
id=user_id, name=username, is_active=True, is_superuser=False,
|
||
)),
|
||
)
|
||
gateway = MoviePilotApiTool(
|
||
session_id="session", user_id=f"ou_{username}", data=SimpleNamespace(users=users),
|
||
)
|
||
gateway.set_message_attr(channel=NotificationChannel.Feishu.value, source="main-bot", username="")
|
||
gateway.set_agent_context({"is_admin": is_channel_admin})
|
||
policy = DEFAULT_TOOL_POLICY_REGISTRY.resolve(
|
||
tool_name="moviepilot_api", arguments={"operation_id": operation_id}, requires_admin=False,
|
||
)
|
||
assert policy.required_role is PrincipalRole.USER
|
||
assert policy.confirmation is ConfirmationMode.REQUIRED
|
||
with patch("app.agent.tools.impl.api.MoviePilotApiExecutor") as executor_type, patch(
|
||
"app.application.security.auth.build_superuser_token_payload",
|
||
) as superuser:
|
||
executor_type.return_value.execute = AsyncMock(return_value='{"success": true}')
|
||
result = asyncio.run(gateway.run(operation_id=operation_id, **arguments))
|
||
|
||
assert json.loads(result)["success"] is True
|
||
context = executor_type.call_args.kwargs["context"]
|
||
assert (context.user_id, context.username, context.is_admin) == (str(user_id), username, False)
|
||
users.find_name_by_bindings.assert_called_once_with({
|
||
"feishu_userid": f"ou_{username}", "feishu_openid": f"ou_{username}",
|
||
}, match_any=True)
|
||
superuser.assert_not_called()
|
||
|
||
|
||
@pytest.mark.parametrize("is_bound", [False, True])
|
||
def test_subscription_write_rejects_unbound_or_inactive_channel_admin(is_bound) -> None:
|
||
"""
|
||
渠道管理员未绑定有效用户时不得回退超级管理员创建订阅。
|
||
"""
|
||
users = SimpleNamespace(
|
||
find_name_by_bindings=Mock(return_value="disabled" if is_bound else None),
|
||
async_get_by_name=AsyncMock(return_value=SimpleNamespace(is_active=False)),
|
||
)
|
||
gateway = MoviePilotApiTool(
|
||
session_id="session", user_id="ou_user", data=SimpleNamespace(users=users),
|
||
)
|
||
gateway.set_message_attr(channel=NotificationChannel.Feishu.value, source="main-bot", username="")
|
||
gateway.set_agent_context({"is_admin": True})
|
||
with patch("app.agent.tools.impl.api.MoviePilotApiExecutor") as executor_type, patch(
|
||
"app.application.security.auth.build_superuser_token_payload",
|
||
) as superuser:
|
||
result = asyncio.run(gateway.run(operation_id="subscription.add", body={"name": "示例"}))
|
||
|
||
assert json.loads(result)["error"] == "operation_unavailable"
|
||
executor_type.assert_not_called()
|
||
superuser.assert_not_called()
|
||
|
||
|
||
def test_factory_uses_api_catalog_by_default(monkeypatch) -> None:
|
||
"""
|
||
统一工具工厂默认只暴露原生能力和 API 网关。
|
||
"""
|
||
monkeypatch.setattr(
|
||
MoviePilotToolFactory,
|
||
"BUILTIN_TOOL_CLASSES",
|
||
(),
|
||
)
|
||
monkeypatch.setattr(
|
||
"app.agent.tools.factory._get_plugin_agent_tools",
|
||
lambda: [],
|
||
)
|
||
|
||
tools = MoviePilotToolFactory.create_tools(
|
||
session_id="session",
|
||
user_id="user",
|
||
)
|
||
|
||
assert [tool.name for tool in tools] == [
|
||
"send_local_file",
|
||
"moviepilot_api",
|
||
]
|
||
|
||
|
||
def test_factory_keeps_native_tools_with_api_catalog(monkeypatch) -> None:
|
||
"""
|
||
统一目录保留原生工具，并追加单一 API 网关。
|
||
"""
|
||
monkeypatch.setattr(
|
||
MoviePilotToolFactory,
|
||
"BUILTIN_TOOL_CLASSES",
|
||
(AgentTaskTool, ExecuteCommandTool),
|
||
)
|
||
monkeypatch.setattr(
|
||
"app.agent.tools.factory._get_plugin_agent_tools",
|
||
lambda: [],
|
||
)
|
||
|
||
tools = MoviePilotToolFactory.create_tools(
|
||
session_id="session",
|
||
user_id="user",
|
||
)
|
||
|
||
assert [tool.name for tool in tools] == [
|
||
"agent_task",
|
||
"execute_command",
|
||
"send_local_file",
|
||
"moviepilot_api",
|
||
]
|
||
|
||
|
||
def test_mcp_field_guidance_preserves_chinese_contracts_without_mutating_inputs() -> None:
|
||
"""
|
||
保留端点中文说明，替换英文兜底，且不改变字段约束或原始对象。
|
||
"""
|
||
schema = {
|
||
"type": "object",
|
||
"properties": {
|
||
"page": {"type": "integer", "minimum": 1, "default": 1, "description": "页码从 1 开始"},
|
||
"count": {"type": "integer", "maximum": 50, "description": "Page size."},
|
||
"keyword": {"type": "string"},
|
||
},
|
||
"required": ["page"],
|
||
"additionalProperties": False,
|
||
}
|
||
original = json.loads(json.dumps(schema))
|
||
|
||
rewritten = _apply_field_descriptions(schema)
|
||
|
||
assert schema == original
|
||
assert rewritten["properties"]["page"] == schema["properties"]["page"]
|
||
assert re.search(r"[\u3400-\u9fff]", rewritten["properties"]["count"]["description"])
|
||
assert re.search(r"[\u3400-\u9fff]", rewritten["properties"]["keyword"]["description"])
|
||
for field in ("count", "keyword"):
|
||
rewritten["properties"][field].pop("description")
|
||
original["properties"][field].pop("description", None)
|
||
assert rewritten == original
|
||
|
||
|
||
def test_mcp_field_guidance_rejects_unknown_undocumented_fields() -> None:
|
||
"""
|
||
未提供中文说明且没有登记兜底的字段必须失败，不能静默遗漏文档。
|
||
"""
|
||
with pytest.raises(ValueError, match="undocumented_field"):
|
||
_apply_field_descriptions({"properties": {"undocumented_field": {"type": "string"}}})
|
||
|
||
|
||
def test_workflow_plugin_actions_preserves_optional_nullable_plugin_filter() -> None:
|
||
"""
|
||
插件动作列表的筛选参数可省略或为 null，生成契约应与既有端点签名一致。
|
||
"""
|
||
schema = MoviePilotApiTool(session_id="session", user_id="api_user").get_mcp_input_schema()
|
||
branch = next(
|
||
item for item in schema["oneOf"]
|
||
if item["properties"]["operation_id"]["const"] == "workflow.plugin.actions"
|
||
)
|
||
query = branch["properties"]["query"]
|
||
assert "query" not in branch["required"]
|
||
assert "plugin_id" not in query.get("required", [])
|
||
assert query["properties"]["plugin_id"]["anyOf"] == [{"type": "string"}, {"type": "null"}]
|