q792602257/agentic-mobile-control
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
ffc3df6c5b50808d9523c3264231497a4588af23
agentic-mobile-control/tests/test_skill_catalog_e2e.py
T
q792602257andClaude Opus 4.6 b94abde92a feat(skill-catalog-subscription): synced catalog + HTTP sync + MCP tools 
Consumes the external Subscription Platform as source of truth for skill
content; reuses skills_learning domain models (extended with KnowledgeSkill)
and workflow.skill_exec resolver. HTTP/MCP deps land in api/ per
CONSTITUTION.md; synced skills use a physically separate SQLite file
(tasks/skills.sqlite3) to preserve the skill-authoring capability boundary.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-07-07 10:33:23 +08:00

370 lines
13 KiB
Python
Raw Blame History

"""End-to-end validation for skill-catalog-subscription (tasks 5.1-5.3).
Combines: FakeSubscriptionClient → SkillSyncRunner → SkillCatalogStore →
api.skill_catalog_mcp handlers. Verifies the full sync+query+resolve loop
and that resolved flow templates drive through the existing device-capability
tools one step at a time (no batched execution; Observe-Think-Act preserved).
"""
from __future__ import annotations
from typing import Any
import pytest
from api.mcp import tool_handlers
from api.skill_catalog_mcp import skill_tool_handlers
from api.skill_sync import SkillSyncRunner, SyncDelta
from device.manager import DeviceManager
from runtime.executor import Executor, ExecutorConfig
from runtime.planner import PlannedStep
from skills_learning.models import (
FlowStep,
FlowTemplateSkill,
KnowledgeSkill,
SkillMetadata,
)
from storage.skill_catalog import SkillCatalogStore
from tests.fakes import FakeDriver
# ----------------------------------------------------------------------
# Shared fake client
# ----------------------------------------------------------------------
class E2EFakeClient:
"""In-process SubscriptionClient with revocable entitlements."""
def __init__(self) -> None:
self.skills_by_sub: dict[str, list[Any]] = {}
def set_skills(self, sub_id: str, skills: list[Any]) -> None:
self.skills_by_sub[sub_id] = list(skills)
def revoke(self, sub_id: str) -> None:
self.skills_by_sub.pop(sub_id, None)
def fetch_entitled_skills(self, subscription_id, since_version=None):
skills = self.skills_by_sub.get(subscription_id, [])
return SyncDelta(
skills=list(skills),
removed_ids=[],
latest_version=1,
is_full_replace=True,
)
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 / "e2e_skills.sqlite3")
@pytest.fixture
def fake_client():
return E2EFakeClient()
@pytest.fixture
def runner(store, fake_client):
return SkillSyncRunner(
store=store,
client=fake_client,
subscriptions=["sub-a"],
poll_interval=999.0, # no auto-poll
)
def _mcp_handlers(store, *, registered_tools, active=None):
return skill_tool_handlers(
store=store,
get_active_subscriptions=lambda: set(active if active is not None else {"sub-a"}),
get_registered_tools=lambda: set(registered_tools),
)
# ----------------------------------------------------------------------
# Task 5.1: seed + verify listable / searchable / fetchable via MCP
# ----------------------------------------------------------------------
def test_e2e_synced_skills_are_listable_searchable_fetchable_via_mcp(
store, fake_client, runner
):
# Seed the fake Subscription Platform with one of each kind
fake_client.set_skills(
"sub-a",
[
_knowledge(
"xhs-tips",
"Xiaohongshu Search Tips",
"Use specific keywords and filter by recent posts.",
tags=["social", "search"],
),
_flow(
"xhs-search-flow",
"Xiaohongshu Search Flow",
steps=[
("tap", {"x": "{search_x}", "y": "{search_y}"}),
("input_text", {"text": "{query}"}),
("tap", {"x": "{submit_x}", "y": "{submit_y}"}),
],
parameters={
"search_x": {"type": "number", "required": True},
"search_y": {"type": "number", "required": True},
"query": {"type": "string", "required": True},
"submit_x": {"type": "number", "required": True},
"submit_y": {"type": "number", "required": True},
},
tags=["social", "automation"],
),
],
)
# Run one sync tick
outcomes = runner.tick()
assert outcomes["sub-a"].success is True
assert outcomes["sub-a"].fetched == 2
# All device-capability tools are registered (used as the validator set)
registered = {"tap", "input_text", "swipe", "launch_app"}
handlers = _mcp_handlers(store, registered_tools=registered)
# list_skills returns both
listed = handlers["list_skills"]()
assert listed["ok"] is True
ids = {s["id"] for s in listed["skills"]}
assert ids == {"xhs-tips", "xhs-search-flow"}
# search_skills finds the knowledge skill by tag and the flow by name
by_tag = handlers["search_skills"](query="social")
assert {s["id"] for s in by_tag["skills"]} == {"xhs-tips", "xhs-search-flow"}
by_name = handlers["search_skills"](query="Xiaohongshu Search Flow")
assert {s["id"] for s in by_name["skills"]} == {"xhs-search-flow"}
# get_skill returns full content for each kind
knowledge = handlers["get_skill"](skill_id="xhs-tips")
assert knowledge["ok"] is True
assert knowledge["skill"]["kind"] == "knowledge"
assert "specific keywords" in knowledge["skill"]["content"]
flow = handlers["get_skill"](skill_id="xhs-search-flow")
assert flow["ok"] is True
assert flow["skill"]["kind"] == "flow_template"
assert [s["tool_name"] for s in flow["skill"]["steps"]] == [
"tap",
"input_text",
"tap",
]
# ----------------------------------------------------------------------
# Task 5.2: revocation — sync-time removal + query-time re-check
# ----------------------------------------------------------------------
def test_e2e_revocation_via_sync_cycle_removes_skill(
store, fake_client, runner
):
"""When the Subscription Platform no longer includes a skill in the
entitled set, the next sync cycle removes it locally."""
fake_client.set_skills(
"sub-a",
[_knowledge("k1", "Keep Me", "c"), _knowledge("k2", "Revoke Me", "c")],
)
runner.tick()
handlers = _mcp_handlers(store, registered_tools=set())
assert {s["id"] for s in handlers["list_skills"]()["skills"]} == {"k1", "k2"}
# Subscription Platform revokes k2
fake_client.set_skills("sub-a", [_knowledge("k1", "Keep Me", "c")])
runner.tick()
after = handlers["list_skills"]()
ids = {s["id"] for s in after["skills"]}
assert ids == {"k1"}
assert handlers["get_skill"](skill_id="k2")["ok"] is False
def test_e2e_revocation_via_query_time_recheck_hides_skill_before_next_sync(
store, fake_client, runner
):
"""Marking a subscription inactive hides its skills on the next query,
without waiting for the next sync tick to remove the rows."""
fake_client.set_skills("sub-a", [_knowledge("k1", "K", "c")])
runner.tick()
handlers_before = _mcp_handlers(store, registered_tools=set())
assert handlers_before["get_skill"](skill_id="k1")["ok"] is True
# Revoke the subscription locally (e.g., via push notification of revocation)
store._set_subscription_state("sub-a", active=False)
# Same handler config — query re-check excludes sub-a's skills immediately
handlers_after = _mcp_handlers(store, registered_tools=set())
assert handlers_after["list_skills"]()["skills"] == []
assert handlers_after["get_skill"](skill_id="k1")["ok"] is False
# ----------------------------------------------------------------------
# Task 5.3: resolve via MCP, then drive through device-capability tools
# ----------------------------------------------------------------------
def test_e2e_resolved_flow_template_drives_existing_device_tools_one_step_at_a_time(
store, fake_client, runner
):
"""Resolve a flow-template skill via MCP, then execute each resulting
step through the existing device-capability tools (Executor + FakeDriver).
Confirms the Observe-Think-Act loop is preserved: no batched
server-side execution primitive is used. Each step is issued
individually, exactly as the LLM would.
"""
# 1. Seed a flow template that uses tap + input_text (real device tools)
fake_client.set_skills(
"sub-a",
[
_flow(
"login-flow",
"Login Flow",
steps=[
("tap", {"x": 50, "y": 100}),
("input_text", {"text": "user@example.com"}),
("tap", {"x": 50, "y": 200}),
],
parameters={},
)
],
)
runner.tick()
# 2. Set up the existing device-capability tool surface (apex-agent-mvp)
fake_driver = FakeDriver()
manager = DeviceManager()
manager.register_device("device-1", lambda: fake_driver)
manager.connect("device-1", max_retries=1)
device_handlers = tool_handlers(manager=manager)
registered_tool_names = set(device_handlers.keys())
# 3. Resolve the flow template via the MCP handler
skill_handlers = _mcp_handlers(store, registered_tools=registered_tool_names)
resolved = skill_handlers["resolve_flow_template"](
skill_id="login-flow", params={}
)
assert resolved["ok"] is True
steps = resolved["steps"]
assert len(steps) == 3
# 4. Build an Executor with the device-capability tools — the same
# Executor the Agent Runtime uses for the Act phase. Bind device_id
# via closures so each step executes against the connected device.
# Each step is issued individually through executor.execute(step),
# exactly as the LLM-driven Observe-Think-Act loop does.
device_id = "device-1"
executor = Executor(
tools={
"tap": lambda **kw: device_handlers["tap"](device_id=device_id, **kw),
"input_text": lambda **kw: device_handlers["input_text"](
device_id=device_id, **kw
),
},
config=ExecutorConfig(max_retries=1, backoff_seconds=0),
)
results = []
for index, step in enumerate(steps, start=1):
planned = PlannedStep(
action=step["tool_name"],
description=f"login-flow step {index}",
args=step["args"],
)
result = executor.execute(planned)
results.append(result)
# 5. Every step must succeed through the normal Act loop
assert all(r.success for r in results), (
f"step failed: {[r.error for r in results if not r.success]}"
)
assert len(results) == 3
# 6. The FakeDriver must have received the calls in order — proving the
# resolved template actually reached the device via the existing
# capability surface (not via a new batch primitive).
tool_calls = [c for c in fake_driver.calls if c[0] in ("tap", "input")]
# 3 steps → exactly 3 device-level calls (tap, input, tap)
assert len(tool_calls) == 3
tap_calls = [c for c in tool_calls if c[0] == "tap"]
assert tap_calls[0][1] == (50, 100)
assert tap_calls[1][1] == (50, 200)
input_calls = [c for c in tool_calls if c[0] == "input"]
assert input_calls[0][1] == ("user@example.com",)
def test_e2e_resolved_flow_template_uses_parameter_substitution(
store, fake_client, runner
):
"""Resolution substitutes {param} placeholders so the LLM-driven Act loop
receives concrete argument values, not templates."""
fake_client.set_skills(
"sub-a",
[
_flow(
"param-flow",
"Parameterized",
steps=[
("tap", {"x": "{coord_x}", "y": "{coord_y}"}),
("input_text", {"text": "fixed-text"}),
],
parameters={
"coord_x": {"type": "number", "required": True},
"coord_y": {"type": "number", "required": True},
},
)
],
)
runner.tick()
handlers = _mcp_handlers(store, registered_tools={"tap", "input_text"})
resolved = handlers["resolve_flow_template"](
skill_id="param-flow",
params={"coord_x": 42, "coord_y": 99},
)
assert resolved["ok"] is True
steps = resolved["steps"]
# Placeholder substituted with provided param values
assert steps[0]["args"] == {"x": 42, "y": 99}
# Non-parameterized args pass through unchanged
assert steps[1]["args"] == {"text": "fixed-text"}
Reference in New Issue View Git Blame Copy Permalink