q792602257andClaude Opus 4.6
b94abde92a
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>
433 lines
15 KiB
Python
433 lines
15 KiB
Python
"""Tests for storage/skill_catalog.py.
|
|
|
|
Covers task 2.7: visibility filtering, not-found indistinguishability,
|
|
dangling-tool-reference invalidation, stable sort, round-trip serialization
|
|
for both knowledge and flow_template kinds.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
|
|
import pytest
|
|
|
|
from core.models import utc_now
|
|
from skills_learning.models import (
|
|
FlowStep,
|
|
FlowTemplateSkill,
|
|
KnowledgeSkill,
|
|
SkillMetadata,
|
|
)
|
|
from storage.skill_catalog import (
|
|
SkillCatalogStore,
|
|
validate_flow_template_tools,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def store(tmp_path):
|
|
return SkillCatalogStore(db_path=tmp_path / "skills.sqlite3")
|
|
|
|
|
|
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 {},
|
|
)
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Task 2.1: schema + sync-only write contract
|
|
# ----------------------------------------------------------------------
|
|
|
|
|
|
def test_store_creates_db_file(tmp_path):
|
|
db = tmp_path / "nested" / "skills.sqlite3"
|
|
SkillCatalogStore(db_path=db)
|
|
assert db.exists()
|
|
|
|
|
|
def test_register_subscription_is_idempotent(store):
|
|
store._register_subscription("sub-a")
|
|
store._register_subscription("sub-a")
|
|
# No error, single row, active by default
|
|
assert store._effective_active({"sub-a"}) == {"sub-a"}
|
|
|
|
|
|
def test_apply_sync_replace_all_clears_prior_skills(store):
|
|
store._register_subscription("sub-a")
|
|
store._apply_sync_replace_all(
|
|
"sub-a",
|
|
[_knowledge("k1", "keeps", "c"), _flow("f1", "flowy", steps=[("tap", {})])],
|
|
latest_version=5,
|
|
)
|
|
assert len(store.list_skills({"sub-a"})) == 2
|
|
# Replace with a single skill → previous two must be gone
|
|
store._apply_sync_replace_all("sub-a", [_knowledge("k2", "new", "c")])
|
|
ids = {m.id for m in store.list_skills({"sub-a"})}
|
|
assert ids == {"k2"}
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Task 2.2 / 2.3 / 2.4: visibility filtering
|
|
# ----------------------------------------------------------------------
|
|
|
|
|
|
def _seed_two_subscriptions(store):
|
|
store._register_subscription("sub-a")
|
|
store._register_subscription("sub-b")
|
|
store._apply_sync_replace_all(
|
|
"sub-a",
|
|
[
|
|
_knowledge("k-a1", "Alpha Knowledge", "c-a1", tags=["alpha"]),
|
|
_flow(
|
|
"f-a2",
|
|
"Alpha Flow",
|
|
steps=[("tap", {"x": 1})],
|
|
tags=["alpha"],
|
|
),
|
|
],
|
|
)
|
|
store._apply_sync_replace_all(
|
|
"sub-b",
|
|
[
|
|
_knowledge("k-b1", "Beta Knowledge", "c-b1", tags=["beta"]),
|
|
],
|
|
)
|
|
|
|
|
|
def test_list_skills_filters_by_active_subscription(store):
|
|
_seed_two_subscriptions(store)
|
|
a = {m.id for m in store.list_skills({"sub-a"})}
|
|
b = {m.id for m in store.list_skills({"sub-b"})}
|
|
both = {m.id for m in store.list_skills({"sub-a", "sub-b"})}
|
|
assert a == {"k-a1", "f-a2"}
|
|
assert b == {"k-b1"}
|
|
assert both == {"k-a1", "f-a2", "k-b1"}
|
|
|
|
|
|
def test_list_skills_empty_active_set_returns_empty(store):
|
|
_seed_two_subscriptions(store)
|
|
assert store.list_skills(set()) == []
|
|
|
|
|
|
def test_list_skills_with_inactive_subscription(store):
|
|
_seed_two_subscriptions(store)
|
|
# Mark sub-a inactive in storage
|
|
store._set_subscription_state("sub-a", active=False)
|
|
assert store.list_skills({"sub-a"}) == []
|
|
# sub-b still visible
|
|
assert {m.id for m in store.list_skills({"sub-a", "sub-b"})} == {"k-b1"}
|
|
|
|
|
|
def test_list_skills_unknown_subscription_returns_empty(store):
|
|
_seed_two_subscriptions(store)
|
|
assert store.list_skills({"unknown-sub"}) == []
|
|
|
|
|
|
def test_search_skills_matches_name_tag_description(store):
|
|
_seed_two_subscriptions(store)
|
|
# Name match
|
|
assert {m.id for m in store.search_skills("Alpha", {"sub-a", "sub-b"})} == {
|
|
"k-a1",
|
|
"f-a2",
|
|
}
|
|
# Tag match
|
|
assert {m.id for m in store.search_skills("beta", {"sub-a", "sub-b"})} == {"k-b1"}
|
|
# Description match would work the same way; verify no-match returns empty
|
|
assert store.search_skills("zzz", {"sub-a", "sub-b"}) == []
|
|
|
|
|
|
def test_search_skills_respects_visibility(store):
|
|
_seed_two_subscriptions(store)
|
|
# sub-a inactive → its skills don't surface even if query matches
|
|
store._set_subscription_state("sub-a", active=False)
|
|
assert {m.id for m in store.search_skills("Alpha", {"sub-a", "sub-b"})} == set()
|
|
|
|
|
|
def test_get_skill_returns_full_content_for_knowledge(store):
|
|
_seed_two_subscriptions(store)
|
|
skill = store.get_skill("k-a1", {"sub-a"})
|
|
assert isinstance(skill, KnowledgeSkill)
|
|
assert skill.content == "c-a1"
|
|
assert skill.metadata.name == "Alpha Knowledge"
|
|
|
|
|
|
def test_get_skill_returns_full_steps_for_flow_template(store):
|
|
_seed_two_subscriptions(store)
|
|
skill = store.get_skill("f-a2", {"sub-a"})
|
|
assert isinstance(skill, FlowTemplateSkill)
|
|
assert len(skill.steps) == 1
|
|
assert skill.steps[0].tool_name == "tap"
|
|
assert skill.steps[0].args == {"x": 1}
|
|
|
|
|
|
def test_get_skill_unknown_id_returns_none(store):
|
|
_seed_two_subscriptions(store)
|
|
assert store.get_skill("does-not-exist", {"sub-a"}) is None
|
|
|
|
|
|
def test_get_skill_not_visible_indistinguishable_from_not_found(store):
|
|
"""Existence must not leak: invisible skill returns None, same as unknown."""
|
|
_seed_two_subscriptions(store)
|
|
# Skill exists under sub-a but caller asks with only sub-b
|
|
invisible = store.get_skill("k-a1", {"sub-b"})
|
|
unknown = store.get_skill("never-existed", {"sub-b"})
|
|
assert invisible is None
|
|
assert unknown is None
|
|
|
|
|
|
def test_get_skill_with_inactive_subscription_returns_none(store):
|
|
_seed_two_subscriptions(store)
|
|
store._set_subscription_state("sub-a", active=False)
|
|
assert store.get_skill("k-a1", {"sub-a"}) is None
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Task 2.5: sync-only write contract (no public writes)
|
|
# ----------------------------------------------------------------------
|
|
|
|
|
|
def test_no_public_write_methods_on_store():
|
|
"""SkillCatalogStore must not expose create/update/remove publicly."""
|
|
public_methods = {
|
|
name
|
|
for name in dir(SkillCatalogStore)
|
|
if not name.startswith("_")
|
|
and callable(getattr(SkillCatalogStore, name))
|
|
}
|
|
forbidden = {"create", "update", "remove", "delete", "save", "put", "post"}
|
|
assert not (public_methods & forbidden), (
|
|
f"Public write methods found: {public_methods & forbidden}"
|
|
)
|
|
|
|
|
|
def test_apply_sync_upsert_round_trips_both_kinds(store):
|
|
store._register_subscription("sub-a")
|
|
original_k = _knowledge("k1", "Knowledge One", "content body", tags=["t1"])
|
|
original_f = _flow(
|
|
"f1",
|
|
"Flow One",
|
|
steps=[("tap", {"x": 1}), ("input_text", {"text": "hi"})],
|
|
parameters={"text": {"type": "string", "required": True}},
|
|
tags=["t2"],
|
|
)
|
|
store._apply_sync_upsert(original_k, "sub-a")
|
|
store._apply_sync_upsert(original_f, "sub-a")
|
|
|
|
fetched_k = store.get_skill("k1", {"sub-a"})
|
|
fetched_f = store.get_skill("f1", {"sub-a"})
|
|
assert isinstance(fetched_k, KnowledgeSkill)
|
|
assert isinstance(fetched_f, FlowTemplateSkill)
|
|
assert fetched_k.content == "content body"
|
|
assert fetched_k.metadata.tags == ["t1"]
|
|
assert [s.tool_name for s in fetched_f.steps] == ["tap", "input_text"]
|
|
assert fetched_f.steps[0].args == {"x": 1}
|
|
assert fetched_f.parameters == {"text": {"type": "string", "required": True}}
|
|
|
|
|
|
def test_apply_sync_upsert_overwrites_on_conflict(store):
|
|
store._register_subscription("sub-a")
|
|
store._apply_sync_upsert(_knowledge("k1", "v1", "old"), "sub-a")
|
|
store._apply_sync_upsert(_knowledge("k1", "v1-updated", "new"), "sub-a")
|
|
skill = store.get_skill("k1", {"sub-a"})
|
|
assert isinstance(skill, KnowledgeSkill)
|
|
assert skill.content == "new"
|
|
assert skill.metadata.name == "v1-updated"
|
|
|
|
|
|
def test_apply_sync_remove_deletes_skill(store):
|
|
store._register_subscription("sub-a")
|
|
store._apply_sync_upsert(_knowledge("k1", "v1", "c"), "sub-a")
|
|
store._apply_sync_remove("k1")
|
|
assert store.get_skill("k1", {"sub-a"}) is None
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Task 2.6: flow-template tool-reference validation
|
|
# ----------------------------------------------------------------------
|
|
|
|
|
|
def test_validate_flow_template_tools_all_known_returns_true():
|
|
skill = _flow("f", "f", steps=[("tap", {}), ("input_text", {"text": "x"})])
|
|
assert validate_flow_template_tools(skill, {"tap", "input_text"}) is True
|
|
|
|
|
|
def test_validate_flow_template_tools_dangling_returns_false():
|
|
skill = _flow("f", "f", steps=[("tap", {}), ("ghost_tool", {})])
|
|
assert validate_flow_template_tools(skill, {"tap"}) is False
|
|
|
|
|
|
def test_validate_flow_template_tools_empty_steps_returns_true():
|
|
skill = _flow("f", "f", steps=[])
|
|
assert validate_flow_template_tools(skill, set()) is True
|
|
|
|
|
|
def test_get_skill_returns_none_when_flow_template_has_dangling_reference(store):
|
|
store._register_subscription("sub-a")
|
|
skill = _flow("f1", "Flow", steps=[("tap", {}), ("ghost_tool", {})])
|
|
store._apply_sync_upsert(skill, "sub-a")
|
|
# Without registered_tools: skill is returned as-is
|
|
assert store.get_skill("f1", {"sub-a"}) is not None
|
|
# With registered_tools missing ghost_tool: returns None (unavailable)
|
|
result = store.get_skill("f1", {"sub-a"}, registered_tools={"tap"})
|
|
assert result is None
|
|
# With all tools registered: returns the skill
|
|
result = store.get_skill("f1", {"sub-a"}, registered_tools={"tap", "ghost_tool"})
|
|
assert isinstance(result, FlowTemplateSkill)
|
|
|
|
|
|
def test_get_skill_validation_skipped_for_knowledge(store):
|
|
"""registered_tools must not affect knowledge skills."""
|
|
store._register_subscription("sub-a")
|
|
store._apply_sync_upsert(_knowledge("k1", "K", "c"), "sub-a")
|
|
result = store.get_skill("k1", {"sub-a"}, registered_tools=set())
|
|
assert isinstance(result, KnowledgeSkill)
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Stable sort order
|
|
# ----------------------------------------------------------------------
|
|
|
|
|
|
def test_list_skills_is_sorted_by_name(store):
|
|
store._register_subscription("sub-a")
|
|
store._apply_sync_replace_all(
|
|
"sub-a",
|
|
[
|
|
_knowledge("z", "Zebra", "c"),
|
|
_knowledge("a", "Apple", "c"),
|
|
_knowledge("m", "Mango", "c"),
|
|
],
|
|
)
|
|
names = [m.name for m in store.list_skills({"sub-a"})]
|
|
assert names == ["Apple", "Mango", "Zebra"]
|
|
|
|
|
|
def test_search_skills_ranks_name_prefix_above_contains(store):
|
|
store._register_subscription("sub-a")
|
|
store._apply_sync_replace_all(
|
|
"sub-a",
|
|
[
|
|
_knowledge("k1", "Find Search", "c"), # contains only
|
|
_knowledge("k2", "Search Tips", "c"), # prefix match
|
|
],
|
|
)
|
|
results = store.search_skills("Search", {"sub-a"})
|
|
names = [m.name for m in results]
|
|
# Prefix match "Search Tips" must rank above contains-only "Find Search"
|
|
assert names[0] == "Search Tips"
|
|
assert names[1] == "Find Search"
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Round-trip serialization for both kinds
|
|
# ----------------------------------------------------------------------
|
|
|
|
|
|
def test_round_trip_knowledge_preserves_all_metadata(store):
|
|
store._register_subscription("sub-a")
|
|
original = KnowledgeSkill(
|
|
metadata=SkillMetadata(
|
|
id="rt-k",
|
|
name="Round",
|
|
description="desc",
|
|
kind="knowledge",
|
|
tags=["a", "b"],
|
|
source="subscription",
|
|
version=3,
|
|
created_at=utc_now(),
|
|
updated_at=utc_now(),
|
|
),
|
|
content="body",
|
|
)
|
|
store._apply_sync_upsert(original, "sub-a")
|
|
fetched = store.get_skill("rt-k", {"sub-a"})
|
|
assert isinstance(fetched, KnowledgeSkill)
|
|
assert fetched.metadata.id == "rt-k"
|
|
assert fetched.metadata.name == "Round"
|
|
assert fetched.metadata.description == "desc"
|
|
assert fetched.metadata.kind == "knowledge"
|
|
assert fetched.metadata.tags == ["a", "b"]
|
|
assert fetched.metadata.source == "subscription"
|
|
assert fetched.metadata.version == 3
|
|
assert fetched.content == "body"
|
|
|
|
|
|
def test_round_trip_flow_template_preserves_steps_and_parameters(store):
|
|
store._register_subscription("sub-a")
|
|
original = FlowTemplateSkill(
|
|
metadata=SkillMetadata(
|
|
id="rt-f",
|
|
name="RT Flow",
|
|
description="d",
|
|
kind="flow_template",
|
|
tags=["x"],
|
|
source="subscription",
|
|
version=2,
|
|
),
|
|
steps=[
|
|
FlowStep(tool_name="tap", args={"x": 10, "y": 20}),
|
|
FlowStep(tool_name="input_text", args={"text": "hello"}),
|
|
],
|
|
parameters={"text": {"type": "string", "required": True}},
|
|
)
|
|
store._apply_sync_upsert(original, "sub-a")
|
|
fetched = store.get_skill("rt-f", {"sub-a"})
|
|
assert isinstance(fetched, FlowTemplateSkill)
|
|
assert [s.tool_name for s in fetched.steps] == ["tap", "input_text"]
|
|
assert fetched.steps[0].args == {"x": 10, "y": 20}
|
|
assert fetched.parameters == {"text": {"type": "string", "required": True}}
|
|
assert fetched.metadata.version == 2
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Subscription state observability (used by sync failure path)
|
|
# ----------------------------------------------------------------------
|
|
|
|
|
|
def test_set_subscription_state_records_error_and_version(store):
|
|
store._register_subscription("sub-a")
|
|
store._set_subscription_state(
|
|
"sub-a",
|
|
active=False,
|
|
last_synced_version=7,
|
|
last_error="401 Unauthorized",
|
|
)
|
|
with store._connect() as conn:
|
|
row = conn.execute(
|
|
"SELECT active, last_synced_version, last_error FROM subscriptions WHERE id = ?",
|
|
("sub-a",),
|
|
).fetchone()
|
|
assert dict(row) == {
|
|
"active": 0,
|
|
"last_synced_version": 7,
|
|
"last_error": "401 Unauthorized",
|
|
}
|