136 lines
6.0 KiB
Python
136 lines
6.0 KiB
Python
import sqlite3
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from src import main
|
|
from src.completed_filed_review_store import CompletedFiledReviewStore
|
|
|
|
|
|
def receipt(repository="stackchain/api", number=9, updated_at="2026-08-15T12:00:00Z"):
|
|
return {"repository": repository, "number": number, "updated_at": updated_at}
|
|
|
|
|
|
def test_receipts_merge_without_lost_updates_and_remain_account_scoped(tmp_path):
|
|
store = CompletedFiledReviewStore(tmp_path / "completed-filed.sqlite3")
|
|
|
|
first = store.merge(" Timmy ", [receipt()])
|
|
second = store.merge("timmy", [receipt("stackchain/web", 4)])
|
|
|
|
assert first == {"receipts": [receipt()]}
|
|
assert second == {"receipts": [receipt(), receipt("stackchain/web", 4)]}
|
|
assert store.get("TIMMY") == second
|
|
assert store.get("alexander") == {"receipts": []}
|
|
|
|
|
|
def test_receipt_history_is_encrypted_and_bound_to_its_account(tmp_path):
|
|
database = tmp_path / "completed-filed.sqlite3"
|
|
store = CompletedFiledReviewStore(database, encryption_key=b"f" * 32)
|
|
timmy = receipt("private-timmy/repository", 912)
|
|
alexander = receipt("private-alexander/repository", 427)
|
|
|
|
store.merge("timmy-private-login", [timmy])
|
|
store.merge("alexander-private-login", [alexander])
|
|
|
|
raw = database.read_bytes()
|
|
assert b"private-timmy/repository" not in raw
|
|
assert b"private-alexander/repository" not in raw
|
|
with sqlite3.connect(database) as connection:
|
|
rows = connection.execute(
|
|
"SELECT login, receipts FROM completed_filed_review_collections ORDER BY login"
|
|
).fetchall()
|
|
connection.execute(
|
|
"UPDATE completed_filed_review_collections SET receipts = ? WHERE login = ?",
|
|
(rows[1][1], rows[0][0]),
|
|
)
|
|
with pytest.raises(RuntimeError, match="private state could not be decrypted"):
|
|
store.get("alexander-private-login")
|
|
|
|
|
|
def test_receipts_keep_newest_revision_and_bound_each_account(tmp_path):
|
|
store = CompletedFiledReviewStore(tmp_path / "completed-filed.sqlite3", limit=2)
|
|
newer = receipt(updated_at="2026-08-16T12:00:00Z")
|
|
|
|
store.merge("timmy", [newer])
|
|
assert store.merge("timmy", [receipt()]) == {"receipts": [newer]}
|
|
bounded = store.merge("timmy", [receipt("stackchain/web", 2), receipt("stackchain/docs", 3)])
|
|
|
|
assert bounded == {"receipts": [receipt("stackchain/web", 2), receipt("stackchain/docs", 3)]}
|
|
|
|
|
|
def test_legacy_plaintext_receipts_migrate_for_every_account_without_loss(tmp_path):
|
|
database = tmp_path / "completed-filed.sqlite3"
|
|
with sqlite3.connect(database) as connection:
|
|
connection.execute(
|
|
"CREATE TABLE completed_filed_reviews ("
|
|
"login TEXT NOT NULL, repository TEXT NOT NULL, issue_number INTEGER NOT NULL, "
|
|
"updated_at TEXT NOT NULL, touched_at INTEGER NOT NULL, "
|
|
"PRIMARY KEY (login, repository, issue_number))"
|
|
)
|
|
connection.executemany(
|
|
"INSERT INTO completed_filed_reviews VALUES (?, ?, ?, ?, ?)",
|
|
[
|
|
("timmy", "legacy-private/first", 1, "2026-08-14T12:00:00Z", 1),
|
|
("timmy", "legacy-private/second", 2, "2026-08-15T12:00:00Z", 2),
|
|
("alexander", "legacy-private/other", 3, "2026-08-16T12:00:00Z", 1),
|
|
],
|
|
)
|
|
|
|
store = CompletedFiledReviewStore(database, encryption_key=b"m" * 32)
|
|
|
|
assert store.get("timmy") == {"receipts": [
|
|
receipt("legacy-private/first", 1, "2026-08-14T12:00:00Z"),
|
|
receipt("legacy-private/second", 2, "2026-08-15T12:00:00Z"),
|
|
]}
|
|
assert store.get("alexander") == {"receipts": [
|
|
receipt("legacy-private/other", 3, "2026-08-16T12:00:00Z")
|
|
]}
|
|
raw = database.read_bytes()
|
|
assert b"legacy-private/first" not in raw
|
|
assert b"legacy-private/other" not in raw
|
|
with sqlite3.connect(database) as connection:
|
|
assert connection.execute(
|
|
"SELECT count(*) FROM sqlite_master WHERE name = 'completed_filed_reviews'"
|
|
).fetchone()[0] == 0
|
|
|
|
|
|
@pytest.mark.parametrize("candidate", [
|
|
{"repository": "bad", "number": 1, "updated_at": "2026-08-15T12:00:00Z"},
|
|
{"repository": "stackchain/api", "number": 0, "updated_at": "2026-08-15T12:00:00Z"},
|
|
{"repository": "stackchain/api", "number": 1, "updated_at": "yesterday"},
|
|
])
|
|
def test_receipts_reject_invalid_records(tmp_path, candidate):
|
|
store = CompletedFiledReviewStore(tmp_path / "completed-filed.sqlite3")
|
|
with pytest.raises(ValueError):
|
|
store.merge("timmy", [candidate])
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_completed_filed_review_api_is_authenticated_csrf_protected_and_no_store(monkeypatch, tmp_path):
|
|
monkeypatch.setenv("STACKCHAIN_DASHBOARD_AUTH_MODE", "operator")
|
|
monkeypatch.setenv("STACKCHAIN_DASHBOARD_ACCESS_TOKEN", "correct horse battery staple")
|
|
monkeypatch.setenv("STACKCHAIN_DASHBOARD_SESSION_SECRET", "a-separate-session-signing-secret-with-enough-entropy")
|
|
monkeypatch.setenv("STACKCHAIN_SESSION_DB", str(tmp_path / "sessions.sqlite3"))
|
|
monkeypatch.setenv("STACKCHAIN_LOGIN_ATTEMPT_DB", str(tmp_path / "login.sqlite3"))
|
|
monkeypatch.setenv("STACKCHAIN_COMPLETED_FILED_REVIEW_DB", str(tmp_path / "completed.sqlite3"))
|
|
|
|
async def user():
|
|
return {"id": 1, "login": "Timmy"}
|
|
|
|
monkeypatch.setattr(main, "current_user", user)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
|
|
await client.post("/api/v1/session", json={"access_token": "correct horse battery staple"})
|
|
forbidden = await client.post("/api/v1/completed-filed-reviews", json={"receipts": [receipt()]})
|
|
headers = {"Origin": "https://test", "X-CSRF-Token": client.cookies["stackchain_csrf"]}
|
|
saved = await client.post(
|
|
"/api/v1/completed-filed-reviews", json={"receipts": [receipt()]}, headers=headers
|
|
)
|
|
fetched = await client.get("/api/v1/completed-filed-reviews")
|
|
|
|
assert forbidden.status_code == 403
|
|
assert saved.status_code == 200
|
|
assert saved.json() == {"receipts": [receipt()]}
|
|
assert fetched.json() == saved.json()
|
|
assert fetched.headers["cache-control"] == "no-store"
|