"""Tests for navi.skills.store — the catalog's SQL against a fake pool."""
import json
import asyncpg
import pytest
from navi.skills.models import SkillError
from navi.skills.store import SkillRef, SkillStore
from tests.conftest_factory import FakeConnection, FakePool, FakeRecord
SKILL = """---
kind: skill
name: Логи
description: Как искать в логах.
area: workflows
tags: [logs]
---
Тело.
"""
PROTOCOL = """---
kind: protocol
name: Выключение
steps: [Раз, Два]
post_checks: [Готово]
on_deviation: Стоп.
on_blocked: Стоп и доложить.
on_already_done: Проверить.
---
"""
# A model often writes the front matter without the kind, and the proposal is
# parsed with kind='protocol' when it is stored; accepting it must not then
# fail on the very document the store already read.
PROTOCOL_NO_KIND = """---
name: Выключение
steps: [Раз, Два]
post_checks: [Готово]
on_deviation: Стоп.
on_blocked: Стоп и доложить.
on_already_done: Проверить.
---
"""
def _store(conn: FakeConnection | None = None) -> tuple[SkillStore, FakeConnection]:
conn = conn or FakeConnection()
store = SkillStore(FakePool(conn))
# Skip the DDL round trip: it would eat a queued result.
store._ready = True
return store, conn
def _row(**overrides) -> FakeRecord:
base = {
"id": "sk1", "kind": "skill", "name": "Логи",
"description": "Как искать в логах.", "triggers": [], "tags": ["logs"],
"area": "workflows", "visibility": "personal", "owner_id": "u1",
"content": SKILL, "version": 1, "content_hash": "h",
"created_at": None, "updated_at": None, "updated_by": "u1",
}
base.update(overrides)
return FakeRecord(**base)
class TestReads:
async def test_get_maps_the_row(self):
store, conn = _store()
conn.enqueue(_row())
entry = await store.get("sk1")
assert entry.name == "Логи"
assert entry.tags == ["logs"]
assert entry.area == "workflows"
assert conn.calls[-1][2] == ("sk1",)
async def test_get_missing_is_none(self):
store, _ = _store()
assert await store.get("nope") is None
async def test_find_prefers_the_user_own_entry(self):
store, conn = _store()
conn.enqueue(_row())
await store.find(user_id="u1", name="Логи")
sql, args = conn.calls[-1][1], conn.calls[-1][2]
assert "(owner_id = $1 OR visibility = 'shared')" in sql
assert "lower(name) = lower($2)" in sql
assert "ORDER BY (owner_id = $1) DESC" in sql
assert args == ("u1", "Логи")
async def test_find_by_id_and_kind(self):
store, conn = _store()
await store.find(user_id="u1", skill_id="sk1", kind="protocol")
assert conn.calls[-1][2] == ("u1", "sk1", "protocol")
assert "kind = $3" in conn.calls[-1][1]
async def test_list_filters_by_query_tag_and_area(self):
store, conn = _store()
conn.enqueue([_row()])
entries = await store.list(
user_id="u1", kind="skill", query="лог", tag="logs", area="workflows",
)
assert len(entries) == 1
sql, args = conn.calls[-1][1], conn.calls[-1][2]
assert "name ILIKE $3" in sql
assert "array_to_string(tags, ' ')" in sql
assert "$4 = ANY(tags)" in sql
assert "area = $5" in sql
assert args == ("u1", "skill", "%лог%", "logs", "workflows", 100)
async def test_list_clamps_the_limit(self):
store, conn = _store()
await store.list(user_id="u1", limit=5000)
assert conn.calls[-1][2] == ("u1", 500)
store, conn = _store()
await store.list(user_id="u1", limit=0)
assert conn.calls[-1][2] == ("u1", 1)
async def test_list_versions_serialises_dates(self):
store, conn = _store()
conn.enqueue([FakeRecord(
version=2, content=SKILL, edited_by="u1", reason="правка",
created_at=None, content_hash="h2",
)])
items = await store.list_versions("sk1")
assert items == [{
"version": 2, "content": SKILL, "edited_by": "u1", "reason": "правка",
"created_at": None, "content_hash": "h2",
}]
class TestCreate:
async def test_writes_the_row_and_the_first_version(self):
store, conn = _store()
conn.enqueue("OK") # INSERT INTO skills
conn.enqueue("OK") # INSERT INTO skill_versions
conn.enqueue(_row()) # the read-back
entry = await store.create(owner_id="u1", content=SKILL, edited_by="u1")
skills_insert = next(c for c in conn.calls if "INSERT INTO skills" in c[1])
versions_insert = next(c for c in conn.calls if "INSERT INTO skill_versions" in c[1])
args = skills_insert[2]
assert args[1] == "skill"
assert args[2] == "Логи"
assert args[3] == "Как искать в логах."
assert args[5] == ["logs"]
assert args[6] == "workflows"
assert args[7] == "personal"
assert args[8] == "u1"
assert json.loads(args[10])["tags"] == ["logs"]
assert "VALUES ($1, 1, $2" in versions_insert[1]
assert versions_insert[2][1] == SKILL
assert entry.id == "sk1"
async def test_shared_entries_have_no_owner(self):
store, conn = _store()
conn.enqueue("OK")
conn.enqueue("OK")
conn.enqueue(_row(owner_id="", visibility="shared"))
await store.create(owner_id="u1", content=SKILL, visibility="shared")
assert conn.calls[0][2][8] == ""
assert conn.calls[0][2][7] == "shared"
async def test_a_protocol_must_declare_its_escape_hatches(self):
store, conn = _store()
with pytest.raises(SkillError, match="post_checks"):
await store.create(owner_id="u1", content="---\nkind: protocol\nname: X\nsteps: [a]\n---\n")
assert conn.calls == [] # nothing reached the database
async def test_duplicate_name_is_a_skill_error(self):
store, conn = _store()
conn.enqueue(asyncpg.UniqueViolationError("duplicate key"))
with pytest.raises(SkillError, match="already exists"):
await store.create(owner_id="u1", content=SKILL)
async def test_unknown_visibility_is_refused(self):
store, conn = _store()
with pytest.raises(SkillError, match="visibility"):
await store.create(owner_id="u1", content=SKILL, visibility="public")
assert conn.calls == []
class TestUpdate:
async def test_bumps_the_version_and_appends_history(self):
store, conn = _store()
conn.enqueue(_row(version=1))
conn.enqueue("OK")
conn.enqueue("OK")
conn.enqueue(_row(version=2, content=SKILL + "\nПравка.\n"))
await store.update("sk1", content=SKILL + "\nПравка.\n", edited_by="u2", reason="fix")
version_insert = next(c for c in conn.calls if "INSERT INTO skill_versions" in c[1])
assert version_insert[2][1] == 2
assert version_insert[2][5] == "u2"
assert version_insert[2][6] == "fix"
async def test_missing_entry_returns_none(self):
store, _ = _store()
assert await store.update("nope", content=SKILL) is None
async def test_the_kind_is_fixed_by_the_row(self):
store, conn = _store()
conn.enqueue(_row(kind="skill"))
# The row is a skill; a document that calls itself a protocol is
# refused rather than silently changing what the entry is.
with pytest.raises(SkillError, match="request says 'skill'"):
await store.update("sk1", content=PROTOCOL)
assert [c[0] for c in conn.calls] == ["fetchrow"]
class TestDeleteAndVisibility:
async def test_delete_reports_what_happened(self):
store, conn = _store()
conn.enqueue("DELETE 1")
assert await store.delete("sk1") is True
store, conn = _store()
conn.enqueue("DELETE 0")
assert await store.delete("sk1") is False
async def test_promoting_to_shared_drops_the_owner(self):
store, conn = _store()
conn.enqueue("UPDATE 1")
conn.enqueue(_row(owner_id="", visibility="shared"))
await store.set_visibility("sk1", "shared", edited_by="u1")
assert conn.calls[0][2] == ("sk1", "shared", "", "u1")
async def test_demoting_takes_the_editor_as_owner(self):
store, conn = _store()
conn.enqueue("UPDATE 1")
conn.enqueue(_row(owner_id="u2", visibility="personal"))
await store.set_visibility("sk1", "personal", edited_by="u2")
assert conn.calls[0][2] == ("sk1", "personal", "u2", "u2")
class TestProposals:
def _proposal_row(self, **overrides) -> FakeRecord:
base = {
"id": "pr1", "owner_id": "u1", "session_id": "s1", "name": "Выключение",
"content": PROTOCOL, "frontmatter": {"name": "Выключение", "steps": ["Раз", "Два"]},
"replaces_id": None, "status": "pending", "skill_id": None, "error": "",
"created_at": None, "decided_at": None, "decided_by": "",
}
base.update(overrides)
return FakeRecord(**base)
async def test_propose_stores_the_document_without_a_catalog_row(self):
store, conn = _store()
conn.enqueue("OK") # INSERT INTO skill_proposals
conn.enqueue(self._proposal_row())
proposal = await store.create_proposal(
owner_id="u1", content=PROTOCOL, session_id="s1",
)
insert = conn.calls[0]
assert "INSERT INTO skill_proposals" in insert[1]
assert insert[2][1] == "u1"
assert insert[2][2] == "s1"
assert insert[2][3] == "Выключение"
assert proposal.status == "pending"
assert proposal.frontmatter["steps"] == ["Раз", "Два"]
assert not any("INSERT INTO skills " in c[1] for c in conn.calls)
async def test_a_skill_is_not_proposed(self):
store, conn = _store()
with pytest.raises(SkillError):
await store.create_proposal(owner_id="u1", content=SKILL)
assert conn.calls == []
async def test_a_replacement_must_point_at_your_protocol(self):
store, conn = _store()
conn.enqueue(_row(kind="protocol", owner_id="u2"))
with pytest.raises(SkillError, match="protocol you own"):
await store.create_proposal(owner_id="u1", content=PROTOCOL, replaces_id="sk1")
store, conn = _store()
with pytest.raises(SkillError, match="does not exist"):
await store.create_proposal(owner_id="u1", content=PROTOCOL, replaces_id="gone")
async def test_accepting_writes_the_catalog_row(self):
store, conn = _store()
conn.enqueue(self._proposal_row()) # get_proposal
conn.enqueue("OK") # INSERT INTO skills
conn.enqueue("OK") # INSERT INTO skill_versions
conn.enqueue(_row(kind="protocol", name="Выключение")) # the read-back
conn.enqueue("OK") # mark accepted
conn.enqueue(self._proposal_row(status="accepted", skill_id="sk1"))
proposal, entry, error = await store.decide_proposal(
"pr1", accepted=True, user_id="u1",
)
assert error == ""
assert entry is not None and entry.id == "sk1"
assert proposal.status == "accepted"
assert proposal.skill_id == "sk1"
skills_insert = next(c for c in conn.calls if "INSERT INTO skills " in c[1])
assert skills_insert[2][8] == "u1" # owner_id
update = next(c for c in conn.calls if "UPDATE skill_proposals" in c[1])
assert update[2][1] == "accepted"
assert update[2][3] == "u1"
async def test_accepting_a_proposal_whose_document_omits_the_kind(self):
store, conn = _store()
conn.enqueue(self._proposal_row(content=PROTOCOL_NO_KIND))
conn.enqueue("OK") # INSERT INTO skills
conn.enqueue("OK") # INSERT INTO skill_versions
conn.enqueue(_row(kind="protocol", content=PROTOCOL_NO_KIND, name="Выключение"))
conn.enqueue("OK") # mark accepted
conn.enqueue(self._proposal_row(
content=PROTOCOL_NO_KIND, status="accepted", skill_id="sk1",
))
proposal, entry, error = await store.decide_proposal(
"pr1", accepted=True, user_id="u1",
)
assert error == ""
assert entry is not None and entry.kind == "protocol"
assert proposal.status == "accepted"
assert not any("UPDATE skill_proposals SET error" in c[1] for c in conn.calls)
async def test_accepting_a_replacement_bumps_that_protocol(self):
store, conn = _store()
conn.enqueue(self._proposal_row(replaces_id="sk1"))
conn.enqueue(_row(kind="protocol", version=1)) # current
conn.enqueue("OK") # history row
conn.enqueue("OK") # update skills
conn.enqueue(_row(kind="protocol", version=2)) # read-back
conn.enqueue("OK") # mark accepted
conn.enqueue(self._proposal_row(status="accepted", replaces_id="sk1", skill_id="sk1"))
_, entry, error = await store.decide_proposal("pr1", accepted=True, user_id="u1")
assert error == ""
assert entry.version == 2
assert any("UPDATE skills SET" in c[1] for c in conn.calls)
async def test_rejecting_leaves_the_catalog_untouched(self):
store, conn = _store()
conn.enqueue(self._proposal_row())
conn.enqueue("OK")
conn.enqueue(self._proposal_row(status="rejected", decided_by="u1"))
proposal, entry, error = await store.decide_proposal("pr1", accepted=False, user_id="u1")
assert entry is None and error == ""
assert proposal.status == "rejected"
assert not any("INSERT INTO skills " in c[1] for c in conn.calls)
async def test_an_already_decided_proposal_is_reported_not_rewritten(self):
store, conn = _store()
conn.enqueue(self._proposal_row(status="rejected"))
proposal, entry, error = await store.decide_proposal("pr1", accepted=True, user_id="u1")
assert proposal.status == "rejected"
assert entry is None and error == ""
assert [c[0] for c in conn.calls] == ["fetchrow"]
async def test_an_unknown_proposal_says_so(self):
store, _ = _store()
assert await store.decide_proposal("nope", accepted=True) == (
None, None, "no such proposal",
)
async def test_a_refused_document_stays_pending_with_the_reason(self):
store, conn = _store()
conn.enqueue(self._proposal_row())
conn.enqueue(asyncpg.UniqueViolationError("duplicate key"))
conn.enqueue("OK") # store the error text
conn.enqueue(self._proposal_row(error="already exists"))
proposal, entry, error = await store.decide_proposal("pr1", accepted=True, user_id="u1")
assert entry is None
assert "already exists" in error
assert proposal.status == "pending"
assert proposal.error == "already exists"
class TestSkillRef:
def test_line_names_the_scope_and_the_id(self):
ref = SkillRef(id="sk1", kind="protocol", name="Выключение", description="Порядок.",
tags=["power"], area="infrastructure", visibility="shared", version=2)
line = ref.line()
assert line.startswith("[protocol] Выключение v2 (infrastructure/shared)")
assert "id=sk1" in line
assert "[tags: power]" in line
def test_a_first_version_is_not_spelled_out(self):
ref = SkillRef(id="sk2", kind="skill", name="Логи", description="")
assert "v1" not in ref.line()
