mirror of
https://github.com/Sea-Haven-Industries/pr-reviewer.git
synced 2026-10-03 04:33:29 +00:00
94 lines
3.1 KiB
Python
94 lines
3.1 KiB
Python
|
"""Tests for the SQLite review cache (stdlib sqlite3, temp DB per test)."""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import json
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
from app.store import ReviewStore
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
def store(tmp_path) -> ReviewStore:
|
||
|
|
return ReviewStore(str(tmp_path / "cache.db"))
|
||
|
|
|
||
|
|
|
||
|
|
def test_upsert_insert_then_get(store: ReviewStore) -> None:
|
||
|
|
store.upsert(
|
||
|
|
"o",
|
||
|
|
"r",
|
||
|
|
1,
|
||
|
|
status="ready",
|
||
|
|
diff_hash="abc",
|
||
|
|
review_json=json.dumps({"summary": "hi"}),
|
||
|
|
pr_title="Title",
|
||
|
|
pr_author="octocat",
|
||
|
|
pr_url="https://x/1",
|
||
|
|
pr_updated_at="2026-01-01",
|
||
|
|
)
|
||
|
|
row = store.get("o", "r", 1)
|
||
|
|
assert row is not None
|
||
|
|
assert row["status"] == "ready"
|
||
|
|
assert row["diff_hash"] == "abc"
|
||
|
|
assert row["pr_title"] == "Title"
|
||
|
|
assert json.loads(row["review_json"])["summary"] == "hi"
|
||
|
|
assert row["cached_at"] > 0
|
||
|
|
|
||
|
|
|
||
|
|
def test_get_missing_returns_none(store: ReviewStore) -> None:
|
||
|
|
assert store.get("o", "r", 999) is None
|
||
|
|
|
||
|
|
|
||
|
|
def test_update_preserves_unspecified_columns(store: ReviewStore) -> None:
|
||
|
|
store.upsert("o", "r", 1, status="ready", diff_hash="h1", review_json="{}")
|
||
|
|
# Update only status; diff_hash and review_json must survive.
|
||
|
|
store.upsert("o", "r", 1, status="reviewing")
|
||
|
|
row = store.get("o", "r", 1)
|
||
|
|
assert row["status"] == "reviewing"
|
||
|
|
assert row["diff_hash"] == "h1"
|
||
|
|
assert row["review_json"] == "{}"
|
||
|
|
|
||
|
|
|
||
|
|
def test_unknown_column_raises(store: ReviewStore) -> None:
|
||
|
|
with pytest.raises(ValueError, match="unknown review columns"):
|
||
|
|
store.upsert("o", "r", 1, status="ready", bogus="x")
|
||
|
|
|
||
|
|
|
||
|
|
def test_list_orders_and_filters_closed(store: ReviewStore) -> None:
|
||
|
|
store.upsert("o", "r", 1, status="ready", pr_updated_at="2026-01-01")
|
||
|
|
store.upsert(
|
||
|
|
"o", "r", 2, status="closed", pr_updated_at="2026-02-02", closed_at=1.0
|
||
|
|
)
|
||
|
|
all_rows = store.list()
|
||
|
|
assert [r["number"] for r in all_rows] == [2, 1] # updated_at desc
|
||
|
|
open_rows = store.list(include_closed=False)
|
||
|
|
assert [r["number"] for r in open_rows] == [1]
|
||
|
|
|
||
|
|
|
||
|
|
def test_reset_stale_reviewing(store: ReviewStore) -> None:
|
||
|
|
store.upsert("o", "r", 1, status="reviewing")
|
||
|
|
store.upsert("o", "r", 2, status="ready")
|
||
|
|
n = store.reset_stale_reviewing()
|
||
|
|
assert n == 1
|
||
|
|
assert store.get("o", "r", 1)["status"] == "error"
|
||
|
|
assert store.get("o", "r", 1)["error"] == "interrupted"
|
||
|
|
assert store.get("o", "r", 2)["status"] == "ready"
|
||
|
|
|
||
|
|
|
||
|
|
def test_mark_missing_closed_and_purge(store: ReviewStore) -> None:
|
||
|
|
store.upsert("o", "r", 1, status="ready")
|
||
|
|
store.upsert("o", "r", 2, status="ready")
|
||
|
|
# Only PR 1 is still live; PR 2 should be closed.
|
||
|
|
store.mark_missing_closed({("o", "r", 1)})
|
||
|
|
assert store.get("o", "r", 1)["status"] == "ready"
|
||
|
|
closed = store.get("o", "r", 2)
|
||
|
|
assert closed["status"] == "closed"
|
||
|
|
assert closed["closed_at"] is not None
|
||
|
|
|
||
|
|
# A long grace keeps it; a zero grace purges it.
|
||
|
|
assert store.purge_closed(older_than_seconds=10_000) == 0
|
||
|
|
assert store.get("o", "r", 2) is not None
|
||
|
|
assert store.purge_closed(older_than_seconds=0) == 1
|
||
|
|
assert store.get("o", "r", 2) is None
|