226 lines
9.2 KiB
Python
226 lines
9.2 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from wf_mcp.models import BrokerConfig
|
|
from wf_mcp.server import create_server_client
|
|
|
|
from .conftest import assert_safe_tool_maps, server_config
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_server_search_mode_pins_stable_control_and_workflow_tools() -> None:
|
|
config = server_config()
|
|
|
|
client = create_server_client(config, search_tools=True)
|
|
async with client:
|
|
tools = await client.list_tools()
|
|
names = [tool.name for tool in tools]
|
|
|
|
assert "search_tools" in names
|
|
assert "call_tool" in names
|
|
assert "wf.admin.list_sources" in names
|
|
assert "wf.admin.list_connections" in names
|
|
assert "wf.admin.get_connection_statuses" in names
|
|
assert "wf.admin.reload_config" in names
|
|
assert "wf.admin.list_proxy_tools" in names
|
|
assert "wf.admin.get_proxy_tool" in names
|
|
assert "wf.workflow.list_artifacts" in names
|
|
assert "wf.workflow.list_capabilities" in names
|
|
assert "wf.workflow.inspect_capability" in names
|
|
assert "wf.workflow.call_capability" in names
|
|
assert "wf.workflow.list_draft_workspaces" in names
|
|
assert "wf.workflow.create_draft_workspace" in names
|
|
assert "wf.workflow.get_draft_workspace" in names
|
|
assert "wf.workflow.delete_draft_workspace" in names
|
|
assert "wf.workflow.patch_draft_workspace" in names
|
|
assert "wf.workflow.validate_draft_workspace" in names
|
|
assert "wf.workflow.set_draft_name" in names
|
|
assert "wf.workflow.set_draft_route" in names
|
|
assert "wf.workflow.set_step_input_map" in names
|
|
assert "wf.workflow.set_step_output_map" in names
|
|
assert "wf.workflow.set_workflow_output_map" in names
|
|
assert "wf.workflow.bind" in names
|
|
assert "wf.workflow.remove_draft_route" in names
|
|
assert "wf.workflow.remove_draft_step" in names
|
|
assert "wf.workflow.remove_draft_binding" in names
|
|
assert "wf.workflow.create_minimal_draft_workspace" in names
|
|
assert "wf.workflow.create_artifact_from_workspace" in names
|
|
assert "wf.workflow.create_wrapper_from_workspace" in names
|
|
assert "wf.workflow.inspect_artifact" in names
|
|
assert "wf.workflow.list_deployments" in names
|
|
assert "wf.workflow.inspect_deployment" in names
|
|
assert "wf.workflow.save_deployment" in names
|
|
assert "wf.workflow.delete_deployment" in names
|
|
assert "wf.workflow.validate_deployment" in names
|
|
assert "wf.workflow.run_deployment" in names
|
|
|
|
assert "wf.workflow.validate_draft" not in names
|
|
assert "wf.workflow.compile_draft" not in names
|
|
assert "wf.workflow.create_artifact_from_plan" not in names
|
|
assert "wf.workflow.create_artifact_from_draft" not in names
|
|
assert "wf.workflow.patch_draft" not in names
|
|
assert "wf.admin.call_tool" not in names
|
|
assert "fixture.personal.echo_tool" not in names
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_server_search_mode_can_use_safe_tool_names() -> None:
|
|
config = server_config()
|
|
|
|
client = create_server_client(
|
|
config,
|
|
search_tools=True,
|
|
safe_tool_names=True,
|
|
)
|
|
async with client:
|
|
tools = await client.list_tools()
|
|
names = [tool.name for tool in tools]
|
|
|
|
assert "search_tools" in names
|
|
assert "call_tool" in names
|
|
assert "wf_admin_list_sources" in names
|
|
assert "wf_workflow_call_capability" in names
|
|
assert "wf.admin.list_sources" not in names
|
|
|
|
result = await assert_safe_tool_maps(
|
|
client,
|
|
original_name="wf.admin.list_sources",
|
|
safe_name="wf_admin_list_sources",
|
|
)
|
|
source_ids = {source["id"] for source in result["sources"]}
|
|
assert "wf.std" in source_ids
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_server_safe_tool_names_adapts_dotted_runtime_names() -> None:
|
|
config = server_config()
|
|
|
|
client = create_server_client(config, safe_tool_names=True)
|
|
async with client:
|
|
artifacts = await assert_safe_tool_maps(
|
|
client,
|
|
original_name="wf.workflow.list_artifacts",
|
|
safe_name="wf_workflow_list_artifacts",
|
|
)
|
|
echo = await assert_safe_tool_maps(
|
|
client,
|
|
original_name="fixture.personal.echo_tool",
|
|
safe_name="fixture_personal_echo_tool",
|
|
arguments={"text": "hello"},
|
|
)
|
|
|
|
assert artifacts["nodes"] == []
|
|
assert artifacts["total"] == 0
|
|
assert echo["echoed"] == "hello"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_workflow_tools_have_human_metadata(tmp_path: Path) -> None:
|
|
config = BrokerConfig(
|
|
store_root=tmp_path / "unified_metadata_store",
|
|
connections=[],
|
|
)
|
|
|
|
client = create_server_client(config, admin_tools=False)
|
|
async with client:
|
|
tools = await client.list_tools()
|
|
by_name = {tool.name: tool for tool in tools}
|
|
list_artifacts = by_name["wf.workflow.list_artifacts"]
|
|
validate_deployment = by_name["wf.workflow.validate_deployment"]
|
|
run_deployment = by_name["wf.workflow.run_deployment"]
|
|
inspect_run = by_name["wf.workflow.inspect_run"]
|
|
read_run_trace = by_name["wf.workflow.read_run_trace"]
|
|
|
|
assert list_artifacts.title == "List Workflow Artifacts"
|
|
assert "saved workflow artifacts" in (list_artifacts.description or "")
|
|
assert "query" in list_artifacts.inputSchema["properties"]
|
|
assert "kind" in list_artifacts.inputSchema["properties"]
|
|
assert "cursor" in list_artifacts.inputSchema["properties"]
|
|
assert "limit" in list_artifacts.inputSchema["properties"]
|
|
live_check_schema = validate_deployment.inputSchema["properties"]["live_check"]
|
|
assert "upstream" in live_check_schema.get("description", "")
|
|
assert run_deployment.title == "Run Workflow Deployment"
|
|
assert "deployment_id" in (run_deployment.description or "")
|
|
assert "trace_range" in run_deployment.inputSchema["properties"]
|
|
trace_range_schema = run_deployment.inputSchema["properties"]["trace_range"]
|
|
assert "Debug traces" in trace_range_schema.get("description", "")
|
|
assert "null" in [option.get("type") for option in trace_range_schema["anyOf"]]
|
|
assert inspect_run.title == "Inspect Workflow Run"
|
|
assert "trace" in (inspect_run.description or "").lower()
|
|
read_trace_schema = read_run_trace.inputSchema["properties"]["trace_range"]
|
|
assert "Debug traces" in read_trace_schema.get("description", "")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_artifact_from_plan_exposes_plan_as_plain_object(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
config = BrokerConfig(
|
|
store_root=tmp_path / "unified_create_artifact_schema_store",
|
|
connections=[],
|
|
)
|
|
|
|
client = create_server_client(config, admin_tools=False)
|
|
async with client:
|
|
tools = await client.list_tools()
|
|
by_name = {tool.name: tool for tool in tools}
|
|
schema = by_name["wf.workflow.create_artifact_from_plan"].inputSchema
|
|
plan_schema = schema["properties"]["plan"]
|
|
|
|
assert plan_schema["type"] == "object"
|
|
assert plan_schema.get("additionalProperties") is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_draft_tools_expose_plain_object_and_patch_array_schemas(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
config = BrokerConfig(
|
|
store_root=tmp_path / "unified_draft_schema_store",
|
|
connections=[],
|
|
)
|
|
|
|
client = create_server_client(config, admin_tools=False)
|
|
async with client:
|
|
tools = await client.list_tools()
|
|
by_name = {tool.name: tool for tool in tools}
|
|
|
|
validate_schema = by_name["wf.workflow.validate_draft"].inputSchema
|
|
validate_draft_schema = validate_schema["properties"]["draft"]
|
|
create_schema = by_name["wf.workflow.create_artifact_from_draft"].inputSchema
|
|
create_draft_schema = create_schema["properties"]["draft"]
|
|
patch_schema = by_name["wf.workflow.patch_draft"].inputSchema
|
|
patch_draft_schema = patch_schema["properties"]["draft"]
|
|
patch_patch_schema = patch_schema["properties"]["patch"]
|
|
|
|
assert validate_draft_schema["type"] == "object"
|
|
assert validate_draft_schema.get("additionalProperties") is True
|
|
assert create_draft_schema["type"] == "object"
|
|
assert create_draft_schema.get("additionalProperties") is True
|
|
assert patch_draft_schema["type"] == "object"
|
|
assert patch_patch_schema["type"] == "array"
|
|
assert "$defs" not in patch_patch_schema
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_admin_tools_have_human_metadata(tmp_path: Path) -> None:
|
|
config = BrokerConfig(
|
|
store_root=tmp_path / "unified_admin_metadata_store",
|
|
connections=[],
|
|
)
|
|
|
|
client = create_server_client(config)
|
|
async with client:
|
|
tools = await client.list_tools()
|
|
by_name = {tool.name: tool for tool in tools}
|
|
list_connections = by_name["wf.admin.list_connections"]
|
|
reload_config = by_name["wf.admin.reload_config"]
|
|
|
|
assert list_connections.title == "List Connections"
|
|
assert "configured MCP connections" in (list_connections.description or "")
|
|
assert reload_config.title == "Reload Config"
|
|
assert "remount" in (reload_config.description or "")
|