q792602257andClaude Opus 4.6
dd03abbbb0
Opens the skill-management-console openspec change (cloud/local skill split with local override) with proposal, design (D1-D11), four delta specs, and tasks. Implements the agent-side persistent local skill store (storage/local_skills.py): authored local skills + cloud-skill overrides in a physically separate SQLite file, with fork-on-revocation. 10 tests pass. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
134 lines
4.6 KiB
Python
134 lines
4.6 KiB
Python
"""Tests for storage/local_skills.py.
|
|
|
|
Covers local-skill CRUD, persistence across reopen, override upsert/remove,
|
|
and fork-on-revocation (override promoted to a standalone local skill).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from skills_learning.models import (
|
|
FlowStep,
|
|
FlowTemplateSkill,
|
|
KnowledgeSkill,
|
|
SkillMetadata,
|
|
)
|
|
from storage.local_skills import LOCAL_SOURCE, LocalSkillStore
|
|
|
|
|
|
@pytest.fixture
|
|
def store(tmp_path):
|
|
return LocalSkillStore(db_path=tmp_path / "local_skills.sqlite3")
|
|
|
|
|
|
def _knowledge(name: str, content: str, *, tags=None) -> KnowledgeSkill:
|
|
return KnowledgeSkill(
|
|
metadata=SkillMetadata(id="ignored", name=name, kind="knowledge", tags=tags or []),
|
|
content=content,
|
|
)
|
|
|
|
|
|
def _flow(name: str, steps, *, parameters=None, tags=None) -> FlowTemplateSkill:
|
|
return FlowTemplateSkill(
|
|
metadata=SkillMetadata(id="ignored", name=name, kind="flow_template", tags=tags or []),
|
|
steps=[FlowStep(tool_name=t, args=a) for t, a in steps],
|
|
parameters=parameters or {},
|
|
)
|
|
|
|
|
|
# --- local skills -----------------------------------------------------------
|
|
|
|
|
|
def test_create_local_assigns_fresh_id_and_source(store):
|
|
created = store.create_local(_knowledge("Brew", "Boil water"))
|
|
assert created.id and created.id != "ignored"
|
|
assert created.metadata.source == LOCAL_SOURCE
|
|
assert store.get_local(created.id).content == "Boil water"
|
|
|
|
|
|
def test_local_skill_persists_across_reopen(tmp_path):
|
|
path = tmp_path / "local_skills.sqlite3"
|
|
s1 = LocalSkillStore(db_path=path)
|
|
created = s1.create_local(_knowledge("Brew", "Boil water"))
|
|
del s1
|
|
s2 = LocalSkillStore(db_path=path)
|
|
assert s2.get_local(created.id).content == "Boil water"
|
|
|
|
|
|
def test_update_local_and_delete_local(store):
|
|
created = store.create_local(_knowledge("Brew", "Boil water"))
|
|
store.update_local(KnowledgeSkill(metadata=created.metadata, content="Boil, then steep"))
|
|
assert store.get_local(created.id).content == "Boil, then steep"
|
|
assert store.delete_local(created.id) is True
|
|
assert store.get_local(created.id) is None
|
|
assert store.delete_local(created.id) is False
|
|
|
|
|
|
def test_update_unknown_local_raises(store):
|
|
with pytest.raises(KeyError):
|
|
store.update_local(_knowledge("ghost", "x"))
|
|
|
|
|
|
def test_list_and_search_local(store):
|
|
a = store.create_local(_knowledge("Alpha", "a", tags=["x"]))
|
|
store.create_local(_knowledge("Beta search", "b"))
|
|
names = [m.name for m in store.list_local()]
|
|
assert names == ["Alpha", "Beta search"]
|
|
hits = {m.id for m in store.search_local("search")}
|
|
assert a.id in hits or True # Beta matches; ensure no crash
|
|
assert any(m.name == "Beta search" for m in store.search_local("search"))
|
|
|
|
|
|
# --- overrides --------------------------------------------------------------
|
|
|
|
|
|
def test_override_upsert_get_remove(store):
|
|
cloud_id = "cloud-1"
|
|
assert store.has_override(cloud_id) is False
|
|
store.upsert_override(cloud_id, _knowledge("Cloud", "overridden content"))
|
|
assert store.has_override(cloud_id) is True
|
|
ov = store.get_override(cloud_id)
|
|
assert ov.metadata.kind == "knowledge"
|
|
assert ov.content == "overridden content"
|
|
assert cloud_id in store.list_override_cloud_ids()
|
|
assert store.remove_override(cloud_id) is True
|
|
assert store.has_override(cloud_id) is False
|
|
assert store.remove_override(cloud_id) is False
|
|
|
|
|
|
def test_override_replaces(store):
|
|
store.upsert_override("c1", _knowledge("C", "first"))
|
|
store.upsert_override("c1", _knowledge("C", "second"))
|
|
assert store.get_override("c1").content == "second"
|
|
|
|
|
|
def test_override_round_trips_flow_template(store):
|
|
store.upsert_override(
|
|
"c2",
|
|
_flow("Flow", [("tap", {"x": 1})], parameters={"x": {"type": "int"}}),
|
|
)
|
|
ov = store.get_override("c2")
|
|
assert isinstance(ov, FlowTemplateSkill)
|
|
assert ov.steps[0].tool_name == "tap"
|
|
assert ov.parameters["x"]["type"] == "int"
|
|
|
|
|
|
# --- fork on revocation -----------------------------------------------------
|
|
|
|
|
|
def test_fork_override_promotes_to_local_and_clears_override(store):
|
|
store.upsert_override("cloud-x", _knowledge("X", "my take"))
|
|
local_id = store.fork_override_to_local("cloud-x")
|
|
assert local_id is not None
|
|
# override gone
|
|
assert store.has_override("cloud-x") is False
|
|
# local skill present with new id, origin local
|
|
forked = store.get_local(local_id)
|
|
assert forked is not None
|
|
assert forked.metadata.source == LOCAL_SOURCE
|
|
assert forked.content == "my take"
|
|
|
|
|
|
def test_fork_with_no_override_returns_none(store):
|
|
assert store.fork_override_to_local("absent") is None
|