116 lines
4.0 KiB
Python
116 lines
4.0 KiB
Python
"""Tests for SQLite-backed run history storage."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
from tai.history_store import RunHistoryStore
|
|
|
|
|
|
def test_history_store_add_and_count(tmp_path) -> None: # type: ignore[no-untyped-def]
|
|
store = RunHistoryStore(tmp_path / "history.db")
|
|
assert store.count() == 0
|
|
|
|
payload = {
|
|
"schema": "tai.analysis.v1",
|
|
"generated_at": "2026-05-11T12:00:00+00:00",
|
|
"issue": "sshd failed",
|
|
"host": "ssh.archflux.net",
|
|
"model": "gemma3:4b",
|
|
"collection": {"total": 5, "failed": 1, "succeeded": 4},
|
|
"token_usage": {"prompt_tokens": 10, "completion_tokens": 20, "total_tokens": 30},
|
|
"analysis": "Root Cause...",
|
|
}
|
|
store.add_payload(payload)
|
|
|
|
assert store.count() == 1
|
|
assert store.count(host="ssh.archflux.net") == 1
|
|
assert store.count(host="other") == 0
|
|
|
|
|
|
def test_history_store_list_host_sessions(tmp_path) -> None: # type: ignore[no-untyped-def]
|
|
store = RunHistoryStore(tmp_path / "history.db")
|
|
store.add_payload(
|
|
{
|
|
"schema": "tai.analysis.v1",
|
|
"generated_at": "2026-05-11T12:00:00+00:00",
|
|
"issue": "issue one",
|
|
"host": "ssh.archflux.net",
|
|
"model": "gemma3:4b",
|
|
"collection": {"total": 1, "failed": 0, "succeeded": 1},
|
|
"token_usage": {"prompt_tokens": 1, "completion_tokens": 2, "total_tokens": 3},
|
|
"analysis": "first",
|
|
}
|
|
)
|
|
store.add_payload(
|
|
{
|
|
"schema": "tai.analysis.v1",
|
|
"generated_at": "2026-05-11T12:05:00+00:00",
|
|
"issue": "issue two",
|
|
"host": "ssh.archflux.net",
|
|
"model": "gemma3:4b",
|
|
"collection": {"total": 1, "failed": 0, "succeeded": 1},
|
|
"token_usage": {"prompt_tokens": 1, "completion_tokens": 2, "total_tokens": 3},
|
|
"analysis": "second",
|
|
}
|
|
)
|
|
|
|
sessions = store.list_host_sessions("ssh.archflux.net", limit=2)
|
|
assert len(sessions) == 2
|
|
assert sessions[0].issue == "issue two"
|
|
assert sessions[1].issue == "issue one"
|
|
|
|
|
|
def test_history_store_list_recent_and_search_keyword(tmp_path) -> None: # type: ignore[no-untyped-def]
|
|
store = RunHistoryStore(tmp_path / "history.db")
|
|
store.add_payload(
|
|
{
|
|
"schema": "tai.analysis.v1",
|
|
"generated_at": "2026-05-11T13:00:00+00:00",
|
|
"issue": "nginx failed",
|
|
"host": "web01",
|
|
"model": "gemma3:4b",
|
|
"collection": {"total": 1, "failed": 0, "succeeded": 1},
|
|
"token_usage": {"prompt_tokens": 1, "completion_tokens": 2, "total_tokens": 3},
|
|
"analysis": "nginx config typo",
|
|
}
|
|
)
|
|
store.add_payload(
|
|
{
|
|
"schema": "tai.analysis.v1",
|
|
"generated_at": "2026-05-11T13:10:00+00:00",
|
|
"issue": "sshd failed",
|
|
"host": "ssh.archflux.net",
|
|
"model": "gemma3:4b",
|
|
"collection": {"total": 1, "failed": 0, "succeeded": 1},
|
|
"token_usage": {"prompt_tokens": 1, "completion_tokens": 2, "total_tokens": 3},
|
|
"analysis": "sshd key mismatch",
|
|
}
|
|
)
|
|
|
|
recent = store.list_recent(limit=2)
|
|
assert len(recent) == 2
|
|
assert recent[0].issue == "sshd failed"
|
|
|
|
matches = store.search_keyword("key", host="ssh.archflux.net", limit=5)
|
|
assert len(matches) == 1
|
|
assert matches[0].host == "ssh.archflux.net"
|
|
|
|
|
|
def test_history_store_accepts_sqlite_url(tmp_path: Path) -> None:
|
|
db_file = tmp_path / "history-url.db"
|
|
store = RunHistoryStore(f"sqlite:///{db_file}")
|
|
store.add_payload(
|
|
{
|
|
"schema": "tai.analysis.v1",
|
|
"generated_at": "2026-05-11T13:20:00+00:00",
|
|
"issue": "test",
|
|
"host": "host1",
|
|
"model": "gemma3:4b",
|
|
"collection": {"total": 1, "failed": 0, "succeeded": 1},
|
|
"token_usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2},
|
|
"analysis": "ok",
|
|
}
|
|
)
|
|
assert store.count(host="host1") == 1
|