mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-29 11:27:43 +08:00
230 lines
8.8 KiB
Python
230 lines
8.8 KiB
Python
"""
|
||
Agent 系统设置完整合同、Schema、写入边界和并发语义测试。
|
||
"""
|
||
|
||
from contextlib import nullcontext
|
||
from unittest.mock import AsyncMock, MagicMock
|
||
|
||
import pytest
|
||
|
||
import app.application.settings as settings_module
|
||
from app.agent.policy.api import API_OPERATION_ROUTES
|
||
from app.agent.tools.impl.api import MoviePilotApiTool
|
||
from app.application.settings.contract import (
|
||
ALL_SETTING_SPECS,
|
||
CORE_SETTING_SPECS,
|
||
SETTING_GROUPS,
|
||
SYSTEMCONFIG_SETTING_SPECS,
|
||
build_value_schema,
|
||
validate_setting_value,
|
||
)
|
||
from app.runtime.config import Settings
|
||
from app.schemas.types import SystemConfigKey
|
||
|
||
|
||
def test_subscription_search_strategy_contract_declares_three_choices() -> None:
|
||
"""
|
||
订阅搜索策略 Schema 明确声明智能、完整和单页三种选项。
|
||
"""
|
||
from pydantic import ValidationError
|
||
|
||
spec = ALL_SETTING_SPECS[SystemConfigKey.SubscribeSearchStrategy.value]
|
||
schema = build_value_schema(spec)
|
||
assert schema["anyOf"][0]["enum"] == ["smart", "full", "single_page"]
|
||
for value in (None, "smart", "full", "single_page"):
|
||
validate_setting_value(spec, value)
|
||
for value in (True, "unknown", {"mode": "full"}):
|
||
with pytest.raises(ValidationError):
|
||
validate_setting_value(spec, value)
|
||
|
||
|
||
def test_settings_contract_covers_every_declared_source() -> None:
|
||
"""
|
||
合同必须覆盖全部 Settings 字段和全部 SystemConfigKey，且每项有来源说明。
|
||
"""
|
||
assert set(CORE_SETTING_SPECS) == set(Settings.model_fields)
|
||
assert set(SYSTEMCONFIG_SETTING_SPECS) == {item.value for item in SystemConfigKey}
|
||
assert len(ALL_SETTING_SPECS) == len(Settings.model_fields) + len(SystemConfigKey)
|
||
assert SETTING_GROUPS
|
||
|
||
for spec in ALL_SETTING_SPECS.values():
|
||
assert spec.description
|
||
assert spec.group
|
||
assert spec.source_file
|
||
assert spec.source_line > 0
|
||
schema = build_value_schema(spec)
|
||
assert schema.get("description") == spec.description
|
||
assert isinstance(schema, dict)
|
||
|
||
|
||
def test_settings_contract_exposes_machine_readable_boundaries() -> None:
|
||
"""
|
||
合同应公开单位、示例、依赖和复杂设置的专用 operation 边界。
|
||
"""
|
||
runtime_spec = ALL_SETTING_SPECS["ACCESS_TOKEN_EXPIRE_MINUTES"]
|
||
assert runtime_spec.unit == "minutes"
|
||
assert runtime_spec.examples
|
||
runtime_schema = build_value_schema(runtime_spec)
|
||
assert runtime_schema["x-unit"] == "minutes"
|
||
assert runtime_schema["examples"] == list(runtime_spec.examples)
|
||
assert build_value_schema(ALL_SETTING_SPECS["API_WORKERS"])["minimum"] == 1
|
||
|
||
managed = ALL_SETTING_SPECS[SystemConfigKey.CustomFilterRules.value]
|
||
assert managed.generic_write_allowed is False
|
||
assert managed.update_operations == ()
|
||
assert managed.preferred_operation_ids
|
||
assert managed.apply_mode == "dedicated_operation"
|
||
|
||
dependent = ALL_SETTING_SPECS[SystemConfigKey.SearchFilterRuleGroups.value]
|
||
assert SystemConfigKey.UserFilterRuleGroups.value in dependent.dependencies
|
||
|
||
|
||
def test_settings_catalog_is_lightweight_and_value_free() -> None:
|
||
"""
|
||
list 阶段只返回合同索引，不应读取当前配置值。
|
||
"""
|
||
runtime = MagicMock()
|
||
system = MagicMock()
|
||
service = settings_module.SystemSettingsService(runtime, system, AsyncMock())
|
||
|
||
result = service.catalog(group="ai_agent", limit=2)
|
||
|
||
assert result["returned_count"] <= 2
|
||
assert result["settings"]
|
||
assert all("value" not in item for item in result["settings"])
|
||
assert all("value_schema" not in item for item in result["settings"])
|
||
assert all("update_operations" in item for item in result["settings"])
|
||
runtime.get.assert_not_called()
|
||
system.get.assert_not_called()
|
||
|
||
|
||
def test_settings_describe_returns_schema_value_and_revision() -> None:
|
||
"""
|
||
describe 阶段应返回单项完整 schema、当前值和稳定 revision。
|
||
"""
|
||
runtime = MagicMock()
|
||
runtime.get.return_value = "gpt-test"
|
||
system = MagicMock()
|
||
service = settings_module.SystemSettingsService(runtime, system, AsyncMock())
|
||
|
||
result = service.describe(setting_key="LLM_MODEL")
|
||
|
||
assert result["setting_key"] == "LLM_MODEL"
|
||
assert result["value"] == "gpt-test"
|
||
assert result["revision"].startswith("sha256:")
|
||
assert result["definition"]["value_schema"]["type"] == "string"
|
||
assert result["definition"]["persistence"] == "app.env"
|
||
|
||
|
||
def test_structured_sensitive_settings_preserve_non_secret_match_fields() -> None:
|
||
"""
|
||
复杂敏感配置只隐藏凭据字段，保留名称和类型供 Agent 定位条目。
|
||
"""
|
||
spec = ALL_SETTING_SPECS[SystemConfigKey.Notifications.value]
|
||
redacted = settings_module.redact_setting_value(
|
||
spec,
|
||
[{"name": "telegram-main", "type": "telegram", "config": {"token": "secret"}}],
|
||
)
|
||
assert redacted[0]["name"] == "telegram-main"
|
||
assert redacted[0]["type"] == "telegram"
|
||
assert redacted[0]["config"]["token"] == "***"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_settings_update_rejects_managed_and_internal_generic_writes(monkeypatch) -> None:
|
||
"""
|
||
通用写入口必须拒绝专用 operation 和内部状态设置。
|
||
"""
|
||
runtime = MagicMock()
|
||
system = MagicMock()
|
||
publish = AsyncMock()
|
||
monkeypatch.setattr(settings_module, "plugin_system_config_mutation", lambda _key: nullcontext())
|
||
service = settings_module.SystemSettingsService(runtime, system, publish)
|
||
|
||
with pytest.raises(ValueError, match="禁止通用写入"):
|
||
await service.update(
|
||
setting_key=SystemConfigKey.CustomFilterRules.value,
|
||
value=[],
|
||
)
|
||
with pytest.raises(ValueError, match="内部状态"):
|
||
await service.update(
|
||
setting_key=SystemConfigKey.AIAgentConfig.value,
|
||
value={},
|
||
)
|
||
runtime.update.assert_not_called()
|
||
system.async_set_with_normalized_value.assert_not_called()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_runtime_settings_revision_is_optional_but_protects_conditional_writes(monkeypatch) -> None:
|
||
"""
|
||
旧调用可省略 revision，新调用提供旧 revision 时必须拒绝并发覆盖。
|
||
"""
|
||
runtime = MagicMock()
|
||
state = {"value": "old"}
|
||
runtime.get.side_effect = lambda _key: state["value"]
|
||
|
||
def update(_key, value):
|
||
"""
|
||
更新测试中的运行时设置快照。
|
||
"""
|
||
state["value"] = value
|
||
return True, "updated"
|
||
|
||
runtime.update.side_effect = update
|
||
system = MagicMock()
|
||
publish = AsyncMock()
|
||
monkeypatch.setattr(settings_module, "plugin_system_config_mutation", lambda _key: nullcontext())
|
||
service = settings_module.SystemSettingsService(runtime, system, publish)
|
||
|
||
legacy = await service.update(setting_key="LLM_MODEL", value="legacy")
|
||
assert legacy["changed"] is True
|
||
revision = legacy["revision"]
|
||
|
||
success = await service.update(
|
||
setting_key="LLM_MODEL",
|
||
value="next",
|
||
expected_revision=revision,
|
||
)
|
||
assert success["saved_value"] == "next"
|
||
|
||
with pytest.raises(settings_module.SystemSettingConflictError, match="expected_revision"):
|
||
await service.update(
|
||
setting_key="LLM_MODEL",
|
||
value="stale",
|
||
expected_revision=revision,
|
||
)
|
||
assert state["value"] == "next"
|
||
|
||
|
||
def test_setting_key_resolution_accepts_enum_names_and_values() -> None:
|
||
"""
|
||
设置解析应同时支持 SystemConfigKey 的枚举名和值。
|
||
"""
|
||
by_name = settings_module.resolve_setting_spec(SystemConfigKey.Downloaders.name)
|
||
by_value = settings_module.resolve_setting_spec(SystemConfigKey.Downloaders.value)
|
||
assert by_name is not None
|
||
assert by_name == by_value
|
||
assert by_name.systemconfig_key is SystemConfigKey.Downloaders
|
||
|
||
|
||
def test_system_setting_operations_expose_the_three_stage_contract() -> None:
|
||
"""
|
||
Agent 网关必须公开 list、describe、get 和 update 的固定路由与参数分支。
|
||
"""
|
||
assert API_OPERATION_ROUTES["config.system.list"].method == "GET"
|
||
assert API_OPERATION_ROUTES["config.system.list"].path == "/api/v1/system/settings/catalog"
|
||
assert API_OPERATION_ROUTES["config.system.describe"].method == "GET"
|
||
assert API_OPERATION_ROUTES["config.system.describe"].path == "/api/v1/system/settings/describe/{setting_key}"
|
||
|
||
schema = MoviePilotApiTool(session_id="contract", user_id="1").get_mcp_input_schema()
|
||
branches = {
|
||
item["properties"]["operation_id"]["const"]: item
|
||
for item in schema["oneOf"]
|
||
}
|
||
assert branches["config.system.list"]["properties"]["query"]["properties"]["limit"]["maximum"] == 500
|
||
assert branches["config.system.describe"]["properties"]["path_params"]["required"] == ["setting_key"]
|
||
update_body_ref = branches["config.system.update"]["properties"]["body"]["$ref"]
|
||
update_body = schema["$defs"][update_body_ref.rsplit("/", 1)[-1]]
|
||
assert "expected_revision" in update_body["properties"]
|