q792602257/agentic-mobile-control
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
8c99dc015a465bb39fd53ff37ad26463455e2269
agentic-mobile-control/tests/test_skill_catalog_mcp.py
T

456 lines
16 KiB
Python
Raw Blame History

"""Tests for api/skill_catalog_mcp.py.
Covers task 4.5: list/search/get via MCP-style handler calls, resolve_flow_template
success and failure cases, semantic error translation, and an explicit
assertion that no batch-execute tool is registered (design D5 / task 4.4).
"""
from __future__ import annotations
import pytest
from api.skill_catalog_mcp import (
SKILL_TOOL_NAMES,
InvalidFlowTemplateError,
MissingParameterError,
SkillAuthoringError,
SkillCatalogError,
SkillNotFoundError,
register_skill_catalog_tools,
skill_tool_handlers,
)
from skills_learning.models import (
FlowStep,
FlowTemplateSkill,
KnowledgeSkill,
SkillMetadata,
)
from storage.skill_catalog import SkillCatalogStore
# ----------------------------------------------------------------------
# Fixtures and helpers
# ----------------------------------------------------------------------
def _knowledge(skill_id: str, name: str, content: str, tags=None) -> KnowledgeSkill:
return KnowledgeSkill(
metadata=SkillMetadata(
id=skill_id,
name=name,
kind="knowledge",
tags=tags or [],
source="subscription",
),
content=content,
)
def _flow(
skill_id: str,
name: str,
*,
steps: list[tuple[str, dict]],
parameters: dict | None = None,
tags=None,
) -> FlowTemplateSkill:
return FlowTemplateSkill(
metadata=SkillMetadata(
id=skill_id,
name=name,
kind="flow_template",
tags=tags or [],
source="subscription",
),
steps=[FlowStep(tool_name=t, args=a) for t, a in steps],
parameters=parameters or {},
)
@pytest.fixture
def store(tmp_path):
return SkillCatalogStore(db_path=tmp_path / "skills.sqlite3")
@pytest.fixture
def seeded_store(store):
store._register_subscription("sub-a")
store._apply_sync_replace_all(
"sub-a",
[
_knowledge("k1", "Knowledge One", "body of knowledge", tags=["docs"]),
_flow(
"f1",
"Search Flow",
steps=[
("tap", {"x": "{x_coord}"}),
("input_text", {"text": "{query}"}),
],
parameters={
"x_coord": {"type": "number", "required": True},
"query": {"type": "string", "required": True},
},
tags=["search"],
),
],
)
return store
def _handlers(store, *, registered_tools=None, active=None, local=None):
if local is None:
import tempfile
from pathlib import Path
from storage.local_skills import LocalSkillStore
local = LocalSkillStore(db_path=Path(tempfile.mkdtemp()) / "local.sqlite3")
return skill_tool_handlers(
store=store,
local_store=local,
get_active_subscriptions=lambda: set(active if active is not None else {"sub-a"}),
get_registered_tools=(lambda: set(registered_tools or {"tap", "input_text"})),
)
# ----------------------------------------------------------------------
# list_skills / search_skills / get_skill via handlers
# ----------------------------------------------------------------------
def test_list_skills_returns_summaries_without_platform_fields(seeded_store):
handlers = _handlers(seeded_store)
result = handlers["list_skills"]()
assert result["ok"] is True
assert len(result["skills"]) == 2
# Summaries contain only id/name/description/kind/tags — no source/version
serialized = repr(result)
assert "subscription" not in serialized.lower()
assert "parent_version_id" not in serialized
assert "originating_goal" not in serialized
names = {s["name"] for s in result["skills"]}
assert names == {"Knowledge One", "Search Flow"}
def test_list_skills_empty_when_no_active_subscriptions(seeded_store):
handlers = _handlers(seeded_store, active=set())
result = handlers["list_skills"]()
assert result == {"ok": True, "skills": []}
def test_search_skills_finds_by_tag_and_name(seeded_store):
handlers = _handlers(seeded_store)
by_name = handlers["search_skills"](query="Search")
assert {s["id"] for s in by_name["skills"]} == {"f1"}
by_tag = handlers["search_skills"](query="docs")
assert {s["id"] for s in by_tag["skills"]} == {"k1"}
def test_get_skill_returns_full_knowledge_content(seeded_store):
handlers = _handlers(seeded_store)
result = handlers["get_skill"](skill_id="k1")
assert result["ok"] is True
skill = result["skill"]
assert skill["kind"] == "knowledge"
assert skill["content"] == "body of knowledge"
assert skill["name"] == "Knowledge One"
def test_get_skill_returns_full_flow_template(seeded_store):
handlers = _handlers(seeded_store)
result = handlers["get_skill"](skill_id="f1")
assert result["ok"] is True
skill = result["skill"]
assert skill["kind"] == "flow_template"
assert [s["tool_name"] for s in skill["steps"]] == ["tap", "input_text"]
assert "x_coord" in skill["parameters"]
def test_get_skill_unknown_returns_skill_not_found(seeded_store):
handlers = _handlers(seeded_store)
result = handlers["get_skill"](skill_id="does-not-exist")
assert result["ok"] is False
assert result["error"] == "skill not found"
def test_get_skill_not_visible_indistinguishable_from_not_found(seeded_store):
"""Existence must not leak: not-visible returns the same error as not-found."""
handlers = _handlers(seeded_store, active=set()) # no active subs
invisible = handlers["get_skill"](skill_id="k1")
unknown = handlers["get_skill"](skill_id="never-existed")
assert invisible == unknown == {"ok": False, "error": "skill not found"}
def test_get_skill_with_dangling_tool_reference_returns_unavailable(seeded_store):
# f1 references tap + input_text; drop input_text from the registered set
handlers = _handlers(seeded_store, registered_tools={"tap"})
result = handlers["get_skill"](skill_id="f1")
assert result["ok"] is False
# Per task 2.6, get_skill returns None for dangling refs, which surfaces
# as "skill not found" at the MCP layer (no existence leak on the cause).
assert result["error"] == "skill not found"
# ----------------------------------------------------------------------
# resolve_flow_template
# ----------------------------------------------------------------------
def test_resolve_flow_template_succeeds_with_valid_params(seeded_store):
handlers = _handlers(seeded_store)
result = handlers["resolve_flow_template"](
skill_id="f1",
params={"x_coord": 100, "query": "hello"},
)
assert result["ok"] is True
steps = result["steps"]
assert steps[0]["tool_name"] == "tap"
assert steps[0]["args"]["x"] == 100 # {x_coord} substituted
assert steps[1]["tool_name"] == "input_text"
assert steps[1]["args"]["text"] == "hello"
def test_resolve_flow_template_missing_required_param(seeded_store):
handlers = _handlers(seeded_store)
result = handlers["resolve_flow_template"](
skill_id="f1",
params={"x_coord": 100}, # missing 'query'
)
assert result["ok"] is False
assert result["error"].startswith("missing parameter:")
assert "query" in result["error"]
def test_resolve_flow_template_unknown_skill(seeded_store):
handlers = _handlers(seeded_store)
result = handlers["resolve_flow_template"](
skill_id="no-such",
params={},
)
assert result["ok"] is False
assert result["error"] == "skill not found"
def test_resolve_flow_template_on_knowledge_skill_returns_unavailable(seeded_store):
"""Cannot resolve a knowledge skill as a flow template."""
handlers = _handlers(seeded_store)
result = handlers["resolve_flow_template"](skill_id="k1", params={})
assert result["ok"] is False
assert result["error"] == "skill unavailable"
def test_resolve_flow_template_with_dangling_tool_reference(seeded_store):
"""A skill whose steps reference an unregistered tool is treated as
not-found at resolve time (consistent with get_skill's behavior)."""
handlers = _handlers(seeded_store, registered_tools={"tap"}) # missing input_text
result = handlers["resolve_flow_template"](
skill_id="f1",
params={"x_coord": 1, "query": "x"},
)
assert result["ok"] is False
assert result["error"] == "skill not found"
# ----------------------------------------------------------------------
# Registration: tools registered on FastMCP server, no batch-execute tool
# ----------------------------------------------------------------------
def _fastmcp_tool_names(server) -> set[str]:
"""Read the FastMCP tool registry directly (sync, no event loop needed)."""
# FastMCP stores tools in ToolManager._tools (mcp>=1.27). Walk the attrs
# defensively so the test doesn't break on minor internal refactors.
if hasattr(server, "_tool_manager"):
manager = server._tool_manager
if hasattr(manager, "_tools"):
return set(manager._tools.keys())
raise AssertionError(
"Could not introspect FastMCP tool registry — internal API changed"
)
def test_register_skill_catalog_tools_registers_expected_names(seeded_store, tmp_path):
from mcp.server.fastmcp import FastMCP
from storage.local_skills import LocalSkillStore
server = FastMCP("test")
register_skill_catalog_tools(
server,
store=seeded_store,
local_store=LocalSkillStore(db_path=tmp_path / "local.sqlite3"),
get_active_subscriptions=lambda: {"sub-a"},
get_registered_tools=lambda: {"tap", "input_text"},
)
names = _fastmcp_tool_names(server)
for expected in SKILL_TOOL_NAMES:
assert expected in names, f"missing tool: {expected}"
def test_no_batch_execute_tool_is_registered(seeded_store, tmp_path):
"""Task 4.4: assert no run_skill_flow / execute / batch tool is registered."""
from mcp.server.fastmcp import FastMCP
from storage.local_skills import LocalSkillStore
server = FastMCP("test")
register_skill_catalog_tools(
server,
store=seeded_store,
local_store=LocalSkillStore(db_path=tmp_path / "local.sqlite3"),
get_active_subscriptions=lambda: {"sub-a"},
get_registered_tools=lambda: {"tap", "input_text"},
)
names = _fastmcp_tool_names(server)
forbidden_fragments = ("run_", "execute", "batch", "invoke_skill")
for name in names:
lower = name.lower()
for fragment in forbidden_fragments:
assert fragment not in lower, (
f"Forbidden tool name '{name}' contains '{fragment}' — "
"design D5 forbids server-side flow execution tools"
)
def test_create_mcp_server_registers_skill_tools_when_store_provided(seeded_store):
"""Wire-up: api.mcp.create_mcp_server must register skill tools when
skill_catalog_store is provided."""
from api.mcp import create_mcp_server
from device.manager import DeviceManager
server = create_mcp_server(
manager=DeviceManager(),
skill_catalog_store=seeded_store,
skill_active_subscriptions={"sub-a"},
)
names = _fastmcp_tool_names(server)
assert "list_skills" in names
assert "tap" in names # existing device tools still present
def test_create_mcp_server_omits_skill_tools_when_no_store():
"""Wire-up must not break existing behavior when no store is provided."""
from api.mcp import create_mcp_server
from device.manager import DeviceManager
server = create_mcp_server(manager=DeviceManager())
names = _fastmcp_tool_names(server)
assert "list_skills" not in names
assert "tap" in names # existing device tools present
# ----------------------------------------------------------------------
# Error class hierarchy
# ----------------------------------------------------------------------
def test_skill_error_hierarchy():
assert issubclass(SkillNotFoundError, SkillCatalogError)
assert issubclass(InvalidFlowTemplateError, SkillCatalogError)
assert issubclass(MissingParameterError, SkillCatalogError)
assert issubclass(SkillAuthoringError, SkillCatalogError)
# ----------------------------------------------------------------------
# Authoring tools: create / update / delete with origin dispatch (D10)
# ----------------------------------------------------------------------
@pytest.fixture
def local(tmp_path):
from storage.local_skills import LocalSkillStore
return LocalSkillStore(db_path=tmp_path / "local.sqlite3")
def _authoring_handlers(seeded_store, local):
return _handlers(seeded_store, local=local)
def test_create_skill_creates_local_skill(seeded_store, local):
handlers = _authoring_handlers(seeded_store, local)
result = handlers["create_skill"](
payload={"kind": "knowledge", "name": "My Note", "content": "hello"}
)
assert result["ok"] is True
assert result["origin"] == "local"
new_id = result["skill"]["id"]
# discoverable via list with origin local
listed = handlers["list_skills"]()
entry = next(s for s in listed["skills"] if s["id"] == new_id)
assert entry["origin"] == "local"
assert entry["locally_overridden"] is False
def test_create_skill_rejects_invalid_payload(seeded_store, local):
handlers = _authoring_handlers(seeded_store, local)
result = handlers["create_skill"](payload={"kind": "knowledge", "name": ""})
assert result["ok"] is False
assert "authoring" in result["error"]
def test_update_skill_edits_local_skill(seeded_store, local):
handlers = _authoring_handlers(seeded_store, local)
created = handlers["create_skill"](
payload={"kind": "knowledge", "name": "Note", "content": "v1"}
)
local_id = created["skill"]["id"]
result = handlers["update_skill"](
skill_id=local_id,
payload={"kind": "knowledge", "name": "Note", "content": "v2"},
)
assert result["ok"] is True
assert result["origin"] == "local"
assert handlers["get_skill"](skill_id=local_id)["skill"]["content"] == "v2"
def test_update_skill_on_cloud_id_creates_override(seeded_store, local):
handlers = _authoring_handlers(seeded_store, local)
# k1 is a cloud knowledge skill in the seeded store
result = handlers["update_skill"](
skill_id="k1",
payload={"kind": "knowledge", "name": "Knowledge One", "content": "OVERRIDDEN"},
)
assert result["ok"] is True
assert result["origin"] == "cloud"
assert result["locally_overridden"] is True
# get_skill reflects the override
got = handlers["get_skill"](skill_id="k1")
assert got["skill"]["content"] == "OVERRIDDEN"
assert got["skill"]["origin"] == "cloud"
assert got["skill"]["locally_overridden"] is True
def test_delete_skill_removes_local_skill(seeded_store, local):
handlers = _authoring_handlers(seeded_store, local)
created = handlers["create_skill"](
payload={"kind": "knowledge", "name": "Temp", "content": "x"}
)
local_id = created["skill"]["id"]
result = handlers["delete_skill"](skill_id=local_id)
assert result["ok"] is True
assert result["origin"] == "local"
assert handlers["get_skill"](skill_id=local_id)["ok"] is False
def test_delete_skill_on_cloud_override_reverts(seeded_store, local):
handlers = _authoring_handlers(seeded_store, local)
handlers["update_skill"](
skill_id="k1",
payload={"kind": "knowledge", "name": "Knowledge One", "content": "OVERRIDDEN"},
)
result = handlers["delete_skill"](skill_id="k1")
assert result["ok"] is True
assert result["origin"] == "cloud"
# cloud skill resurfaces with original content
got = handlers["get_skill"](skill_id="k1")
assert got["skill"]["content"] == "body of knowledge"
assert got["skill"]["locally_overridden"] is False
def test_delete_skill_on_cloud_without_override_errors(seeded_store, local):
handlers = _authoring_handlers(seeded_store, local)
result = handlers["delete_skill"](skill_id="k1") # cloud, no override
assert result["ok"] is False
assert "authoring" in result["error"]
Reference in New Issue View Git Blame Copy Permalink