mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-11-15 12:50:20 +08:00
test(agent): document regression test contracts
This commit is contained in:
@@ -193,6 +193,11 @@ async def test_plugin_gateway_rejects_empty_or_redacted_replacements(
|
||||
) -> None:
|
||||
"""
|
||||
空对象和脱敏占位不能作为整份插件配置送入真实写入入口。
|
||||
|
||||
:param api_transport: 使用真实网关和内存 HTTP 传输的测试夹具
|
||||
:param entrypoint: 网关的直接、LangChain 或管理器调用入口
|
||||
:param json_encoded: 是否将请求体编码为 JSON 字符串
|
||||
:param body: 待拒绝的不安全配置对象
|
||||
"""
|
||||
gateway, request = api_transport
|
||||
result = await _invoke_gateway(entrypoint, gateway, {
|
||||
@@ -214,6 +219,9 @@ async def test_plugin_gateway_preserves_complete_unmasked_configuration(
|
||||
) -> None:
|
||||
"""
|
||||
安全完整配置按原样发送,普通文本中的星号不是脱敏占位。
|
||||
|
||||
:param api_transport: 使用真实网关和内存 HTTP 传输的测试夹具
|
||||
:param entrypoint: 网关的直接、LangChain 或管理器调用入口
|
||||
"""
|
||||
gateway, request = api_transport
|
||||
body = {
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
选择回执的时间戳保留及历史会话读取兼容回归。
|
||||
"""
|
||||
|
||||
from collections.abc import Iterator
|
||||
from copy import deepcopy
|
||||
from types import SimpleNamespace
|
||||
from uuid import uuid4
|
||||
@@ -22,7 +23,7 @@ from app.schemas.agent import AgentChatMessage
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def history_session_id():
|
||||
def history_session_id() -> Iterator[str]:
|
||||
"""
|
||||
为持久化路由测试分配唯一会话,并在结束后精确清理本次提交的记录。
|
||||
|
||||
|
||||
@@ -4,6 +4,7 @@
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
@@ -179,9 +180,11 @@ async def test_code_cannot_bypass_unloaded_skill(environment):
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_python_rpc_rejects_plugin_config_writes(environment):
|
||||
async def test_python_rpc_rejects_plugin_config_writes(environment: Any) -> None:
|
||||
"""
|
||||
Python 内的 API 助手仍是只读入口,不能提交插件配置写操作。
|
||||
|
||||
:param environment: 包含真实 Agent 图和内存 API 端口的测试环境
|
||||
"""
|
||||
scope, executor, api, tool, _, _, _, middlewares = environment
|
||||
program = (
|
||||
|
||||
@@ -635,9 +635,14 @@ async def test_mcp_read_tag_does_not_bypass_persistent_identity(invocation_runti
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_successful_write_replay_is_not_a_new_success(invocation_runtime, monkeypatch):
|
||||
async def test_successful_write_replay_is_not_a_new_success(
|
||||
invocation_runtime: Any, monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""
|
||||
同一轮命中成功回执时明确区分旧状态,本次调用没有再次执行。
|
||||
|
||||
:param invocation_runtime: 提供临时回执数据库的测试环境
|
||||
:param monkeypatch: 隔离真实 API 传输的测试夹具
|
||||
"""
|
||||
repository, records = invocation_runtime
|
||||
run = AsyncMock(return_value=json.dumps({"success": True}))
|
||||
@@ -661,9 +666,14 @@ async def test_successful_write_replay_is_not_a_new_success(invocation_runtime,
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_distinct_secret_values_do_not_share_write_receipts(invocation_runtime, monkeypatch):
|
||||
async def test_distinct_secret_values_do_not_share_write_receipts(
|
||||
invocation_runtime: Any, monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""
|
||||
不同凭据的规范参数必须区分,持久回执仅保存摘要而不保存输入。
|
||||
|
||||
:param invocation_runtime: 提供临时回执数据库的测试环境
|
||||
:param monkeypatch: 隔离真实 API 传输的测试夹具
|
||||
"""
|
||||
repository, records = invocation_runtime
|
||||
first = {"enabled": True, "taobao_cookie": "session=synthetic-first"}
|
||||
@@ -693,9 +703,14 @@ async def test_distinct_secret_values_do_not_share_write_receipts(invocation_run
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_failed_write_replay_retains_previous_failure_without_reexecuting(invocation_runtime, monkeypatch):
|
||||
async def test_failed_write_replay_retains_previous_failure_without_reexecuting(
|
||||
invocation_runtime: Any, monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""
|
||||
失败回执同样只描述旧调用,不能变成第二次调用的执行结果。
|
||||
|
||||
:param invocation_runtime: 提供临时回执数据库的测试环境
|
||||
:param monkeypatch: 隔离真实 API 传输的测试夹具
|
||||
"""
|
||||
repository, records = invocation_runtime
|
||||
run = AsyncMock(return_value=json.dumps({"success": False, "message": "操作被拒绝"}))
|
||||
@@ -715,9 +730,14 @@ async def test_failed_write_replay_retains_previous_failure_without_reexecuting(
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_lower_level_replay_does_not_settle_current_write_as_success(invocation_runtime, monkeypatch):
|
||||
async def test_lower_level_replay_does_not_settle_current_write_as_success(
|
||||
invocation_runtime: Any, monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""
|
||||
下层只返回历史回执时保留未知认领,后续请求不得据此重复副作用。
|
||||
|
||||
:param invocation_runtime: 提供临时回执数据库的测试环境
|
||||
:param monkeypatch: 隔离真实 API 传输的测试夹具
|
||||
"""
|
||||
repository, records = invocation_runtime
|
||||
run = AsyncMock(return_value=json.dumps({
|
||||
@@ -739,9 +759,14 @@ async def test_lower_level_replay_does_not_settle_current_write_as_success(invoc
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_real_agent_to_plugin_endpoint_preserves_configuration_and_services(invocation_runtime, monkeypatch):
|
||||
async def test_real_agent_to_plugin_endpoint_preserves_configuration_and_services(
|
||||
invocation_runtime: Any, monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""
|
||||
真实图经 API 网关与 HTTP 路由写配置,回放和无效写入不影响插件数据及服务。
|
||||
|
||||
:param invocation_runtime: 提供临时回执数据库的测试环境
|
||||
:param monkeypatch: 隔离真实认证和网络传输的测试夹具
|
||||
"""
|
||||
repository, records = invocation_runtime
|
||||
stored = {"enabled": True, "cookie": "session=synthetic-cookie", "keywords": list(range(60))}
|
||||
@@ -793,9 +818,12 @@ async def test_real_agent_to_plugin_endpoint_preserves_configuration_and_service
|
||||
app.dependency_overrides[get_plugin_config_command] = lambda: command
|
||||
calls = []
|
||||
with TestClient(app) as client:
|
||||
async def send(**kwargs):
|
||||
async def send(**kwargs: Any) -> SimpleNamespace:
|
||||
"""
|
||||
仅把实际 HTTP 传输换成内存 ASGI,保留真实路由与应用命令。
|
||||
|
||||
:param kwargs: API 执行器传入的请求方法、地址和 JSON 内容
|
||||
:return: 可由执行器消费的内存响应
|
||||
"""
|
||||
calls.append(kwargs)
|
||||
response = client.request(kwargs["method"], kwargs["url"], json=kwargs["json"])
|
||||
|
||||
@@ -2,6 +2,8 @@
|
||||
验证插件配置写入前对脱敏占位的识别边界。
|
||||
"""
|
||||
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
from app.agent.policy.sanitizer import contains_redacted_placeholder
|
||||
@@ -19,14 +21,17 @@ from app.agent.policy.sanitizer import contains_redacted_placeholder
|
||||
({"cookie": "session=synthetic-cookie"}, False),
|
||||
({"count": 0, "enabled": False, "empty": None}, False),
|
||||
])
|
||||
def test_redacted_placeholder_detection_preserves_nonsecret_text(value, expected):
|
||||
def test_redacted_placeholder_detection_preserves_nonsecret_text(value: Any, expected: bool) -> None:
|
||||
"""
|
||||
已遮蔽凭据不能写回,普通文本中的星号仍可作为业务内容。
|
||||
|
||||
:param value: 包含凭据或普通星号文本的配置
|
||||
:param expected: 是否应拒绝写回
|
||||
"""
|
||||
assert contains_redacted_placeholder(value) is expected
|
||||
|
||||
|
||||
def test_redacted_placeholder_detection_handles_cycles_and_shared_values():
|
||||
def test_redacted_placeholder_detection_handles_cycles_and_shared_values() -> None:
|
||||
"""
|
||||
循环结构不阻塞检查,共享对象进入敏感字段时仍须重新检查。
|
||||
"""
|
||||
|
||||
@@ -186,7 +186,7 @@ async def test_command_receipt_and_direct_manager_keep_original_payload():
|
||||
assert json.loads(await manager.call_tool(tool.name, {})) == {"success": False, "message": "目标记录不存在"}
|
||||
|
||||
|
||||
def test_command_aggregate_does_not_turn_replayed_receipt_into_success():
|
||||
def test_command_aggregate_does_not_turn_replayed_receipt_into_success() -> None:
|
||||
"""
|
||||
聚合消息时历史回执不能被旁边的成功消息覆盖成新操作成功。
|
||||
"""
|
||||
|
||||
@@ -3,6 +3,7 @@
|
||||
"""
|
||||
|
||||
from contextlib import nullcontext
|
||||
from typing import Optional
|
||||
|
||||
from app.application.plugin.config import PluginConfigCommand, PluginPurgeScope
|
||||
from app.runtime.extensions.plugin.admission import PluginMutationAdmission
|
||||
@@ -16,7 +17,7 @@ def _command(
|
||||
directory_existed: bool = True,
|
||||
is_clone: bool = False,
|
||||
purge_invalid_empty_host=None,
|
||||
existing_config: dict[str, object] | None = None,
|
||||
existing_config: Optional[dict[str, object]] = None,
|
||||
) -> PluginConfigCommand:
|
||||
"""
|
||||
构造记录端口调用顺序的插件配置用例。
|
||||
|
||||
@@ -102,9 +102,11 @@ def test_plugin_config_reads_back_through_the_config_store(config_store: PluginC
|
||||
assert config_store.read(PLUGIN_ID) == {"enable": False, "token": "probe"}
|
||||
|
||||
|
||||
def test_http_empty_config_keeps_real_instance_row_and_registered_services(config_store: PluginConfigStore):
|
||||
def test_http_empty_config_keeps_real_instance_row_and_registered_services(config_store: PluginConfigStore) -> None:
|
||||
"""
|
||||
真实 HTTP 与实例表边界拒绝空覆盖,合法完整配置仍按既有顺序生效。
|
||||
|
||||
:param config_store: 使用隔离实例表的真实配置存储
|
||||
"""
|
||||
original = {
|
||||
"enabled": True,
|
||||
|
||||
Reference in New Issue
Block a user