"""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: Проверить.
---
"""
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 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()
