"""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"]