105 lines
3.3 KiB
Python
105 lines
3.3 KiB
Python
"""Tests for app/context/store.py — fact and document CRUD."""
|
|
import sqlite3
|
|
import pytest
|
|
from pathlib import Path
|
|
from app.context.store import (
|
|
add_fact, list_facts, delete_fact,
|
|
add_document, list_documents, delete_document,
|
|
ContextFact, ContextDocument,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def db(tmp_path):
|
|
db_path = tmp_path / "t.db"
|
|
conn = sqlite3.connect(str(db_path))
|
|
conn.executescript("""
|
|
CREATE TABLE context_facts (
|
|
id TEXT PRIMARY KEY, category TEXT NOT NULL, key TEXT NOT NULL,
|
|
value TEXT NOT NULL, source TEXT, created_at TEXT NOT NULL
|
|
);
|
|
CREATE TABLE context_documents (
|
|
id TEXT PRIMARY KEY, filename TEXT NOT NULL, doc_type TEXT NOT NULL,
|
|
full_text TEXT NOT NULL, file_size INTEGER, uploaded_at TEXT NOT NULL
|
|
);
|
|
CREATE TABLE context_chunks (
|
|
id TEXT PRIMARY KEY, document_id TEXT NOT NULL
|
|
REFERENCES context_documents(id) ON DELETE CASCADE,
|
|
chunk_index INTEGER NOT NULL, text TEXT NOT NULL, embedding BLOB
|
|
);
|
|
""")
|
|
conn.commit()
|
|
conn.close()
|
|
return db_path
|
|
|
|
|
|
def test_add_fact_returns_fact(db):
|
|
fact = add_fact(db, "host", "hostname", "heimdall.local", source="wizard")
|
|
assert isinstance(fact, ContextFact)
|
|
assert fact.id
|
|
assert fact.category == "host"
|
|
assert fact.key == "hostname"
|
|
assert fact.value == "heimdall.local"
|
|
assert fact.source == "wizard"
|
|
assert fact.created_at
|
|
|
|
|
|
def test_list_facts_empty(db):
|
|
assert list_facts(db) == []
|
|
|
|
|
|
def test_list_facts_all(db):
|
|
add_fact(db, "host", "hostname", "heimdall.local")
|
|
add_fact(db, "service", "plex", "port:32400")
|
|
facts = list_facts(db)
|
|
assert len(facts) == 2
|
|
|
|
|
|
def test_list_facts_by_category(db):
|
|
add_fact(db, "host", "hostname", "heimdall.local")
|
|
add_fact(db, "service", "plex", "port:32400")
|
|
add_fact(db, "service", "sonarr", "port:8989")
|
|
assert len(list_facts(db, category="host")) == 1
|
|
assert len(list_facts(db, category="service")) == 2
|
|
|
|
|
|
def test_delete_fact_returns_true(db):
|
|
fact = add_fact(db, "note", "k", "v")
|
|
assert delete_fact(db, fact.id) is True
|
|
assert list_facts(db) == []
|
|
|
|
|
|
def test_delete_fact_missing_returns_false(db):
|
|
assert delete_fact(db, "nonexistent") is False
|
|
|
|
|
|
def test_add_document_returns_document(db):
|
|
doc = add_document(db, "runbook.md", "markdown", "# Plex\nRestart with systemctl", file_size=100)
|
|
assert isinstance(doc, ContextDocument)
|
|
assert doc.id
|
|
assert doc.filename == "runbook.md"
|
|
assert doc.doc_type == "markdown"
|
|
|
|
|
|
def test_list_documents_empty(db):
|
|
assert list_documents(db) == []
|
|
|
|
|
|
def test_delete_document_cascades_chunks(db):
|
|
doc = add_document(db, "test.md", "markdown", "content")
|
|
conn = sqlite3.connect(str(db))
|
|
conn.execute(
|
|
"INSERT INTO context_chunks(id, document_id, chunk_index, text) VALUES (?,?,0,?)",
|
|
("c1", doc.id, "chunk text"),
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
assert delete_document(db, doc.id) is True
|
|
conn = sqlite3.connect(str(db))
|
|
chunks = conn.execute("SELECT * FROM context_chunks WHERE document_id=?", (doc.id,)).fetchall()
|
|
conn.close()
|
|
assert chunks == []
|
|
|
|
|
|
def test_delete_document_missing_returns_false(db):
|
|
assert delete_document(db, "nonexistent") is False
|