243 lines
9.6 KiB
Python
243 lines
9.6 KiB
Python
import json
|
|
import sqlite3
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from src import main, unfiled_draft_store
|
|
from src.unfiled_draft_store import UnfiledDraftConflict, UnfiledDraftStore
|
|
|
|
|
|
ENCRYPTION_KEY = b"d" * 32
|
|
|
|
|
|
def draft(draft_id="phone-capture", *, title="Broken checkout", evidence=None):
|
|
return {
|
|
"id": draft_id,
|
|
"title": title,
|
|
"body": "Steps from the field",
|
|
"saved_at": 1_723_600_000_000,
|
|
"blockers": [{"repository": "stackchain/api", "number": 7, "title": "API rollout"}],
|
|
"evidence": (evidence if evidence is not None else [
|
|
{
|
|
"filename": "checkout.png",
|
|
"content_type": "image/png",
|
|
"note": "Error after tapping Pay",
|
|
"data": "cG5nLWJ5dGVz",
|
|
}
|
|
]),
|
|
}
|
|
|
|
|
|
def test_unfiled_drafts_encrypt_private_content_and_authenticate_the_account(tmp_path):
|
|
path = tmp_path / "unfiled.sqlite3"
|
|
store = UnfiledDraftStore(path, encryption_key=ENCRYPTION_KEY)
|
|
|
|
created = store.replace("timmy", 0, [draft()])
|
|
|
|
with sqlite3.connect(path) as connection:
|
|
payload = connection.execute(
|
|
"SELECT drafts FROM unfiled_drafts WHERE login = 'timmy'"
|
|
).fetchone()[0]
|
|
connection.execute(
|
|
"INSERT INTO unfiled_drafts(login, revision, drafts) VALUES (?, ?, ?)",
|
|
("alexander", 1, payload),
|
|
)
|
|
assert payload.startswith("v1:")
|
|
assert "Broken checkout" not in payload
|
|
assert "Steps from the field" not in payload
|
|
assert "cG5nLWJ5dGVz" not in payload
|
|
assert store.get("timmy") == created
|
|
with pytest.raises(
|
|
unfiled_draft_store.UnfiledDraftEncryptionError,
|
|
match="could not be decrypted",
|
|
):
|
|
store.get("alexander")
|
|
|
|
|
|
def test_unfiled_drafts_migrate_plaintext_without_changing_revision_or_order(tmp_path):
|
|
path = tmp_path / "unfiled.sqlite3"
|
|
store = UnfiledDraftStore(path, encryption_key=ENCRYPTION_KEY)
|
|
legacy = [draft(), draft("second", title="Second")]
|
|
with sqlite3.connect(path) as connection:
|
|
connection.execute(
|
|
"INSERT INTO unfiled_drafts(login, revision, drafts) VALUES (?, ?, ?)",
|
|
("timmy", 7, json.dumps(legacy, separators=(",", ":"))),
|
|
)
|
|
|
|
assert store.get("timmy") == {"revision": 7, "drafts": legacy}
|
|
with sqlite3.connect(path) as connection:
|
|
migrated = connection.execute(
|
|
"SELECT revision, drafts FROM unfiled_drafts WHERE login = 'timmy'"
|
|
).fetchone()
|
|
assert migrated[0] == 7
|
|
assert migrated[1].startswith("v1:")
|
|
assert "Broken checkout" not in migrated[1]
|
|
|
|
|
|
def test_unfiled_drafts_reject_tampered_ciphertext(tmp_path):
|
|
path = tmp_path / "unfiled.sqlite3"
|
|
store = UnfiledDraftStore(path, encryption_key=ENCRYPTION_KEY)
|
|
store.replace("timmy", 0, [draft()])
|
|
with sqlite3.connect(path) as connection:
|
|
payload = connection.execute(
|
|
"SELECT drafts FROM unfiled_drafts WHERE login = 'timmy'"
|
|
).fetchone()[0]
|
|
replacement = "A" if payload[-1] != "A" else "B"
|
|
connection.execute(
|
|
"UPDATE unfiled_drafts SET drafts = ? WHERE login = 'timmy'",
|
|
(payload[:-1] + replacement,),
|
|
)
|
|
|
|
with pytest.raises(
|
|
unfiled_draft_store.UnfiledDraftEncryptionError,
|
|
match="could not be decrypted",
|
|
):
|
|
store.get("timmy")
|
|
|
|
|
|
def test_unfiled_drafts_are_revisioned_ordered_and_account_scoped(tmp_path):
|
|
store = UnfiledDraftStore(
|
|
tmp_path / "unfiled.sqlite3", encryption_key=ENCRYPTION_KEY
|
|
)
|
|
|
|
created = store.replace(" Timmy ", 0, [draft(), draft("second", title="Second")])
|
|
|
|
assert created == {"revision": 1, "drafts": [draft(), draft("second", title="Second")]}
|
|
assert store.get("timmy") == created
|
|
assert store.get("alexander") == {"revision": 0, "drafts": []}
|
|
with pytest.raises(UnfiledDraftConflict) as conflict:
|
|
store.replace("timmy", 0, [draft(title="Stale overwrite")])
|
|
assert conflict.value.snapshot == created
|
|
|
|
|
|
def test_unfiled_drafts_bound_collection_and_decoded_evidence(tmp_path):
|
|
store = UnfiledDraftStore(
|
|
tmp_path / "unfiled.sqlite3",
|
|
encryption_key=ENCRYPTION_KEY,
|
|
limit=2,
|
|
max_total_bytes=12,
|
|
)
|
|
|
|
with pytest.raises(ValueError, match="limited to 2"):
|
|
store.replace("timmy", 0, [draft("one", evidence=[]), draft("two", evidence=[]), draft("three", evidence=[])])
|
|
with pytest.raises(ValueError, match="evidence is too large"):
|
|
store.replace("timmy", 0, [draft(evidence=[{
|
|
"filename": "large.png", "content_type": "image/png", "data": "eHh4eHh4eHh4eHh4eHh4eHh4eHg=",
|
|
}])])
|
|
with pytest.raises(ValueError, match="valid base64"):
|
|
store.replace("timmy", 0, [draft(evidence=[{
|
|
"filename": "bad.png", "content_type": "image/png", "data": "not base64!",
|
|
}])])
|
|
|
|
|
|
def test_unfiled_drafts_allow_untitled_photo_evidence_but_reject_empty_records(tmp_path):
|
|
store = UnfiledDraftStore(
|
|
tmp_path / "unfiled.sqlite3", encryption_key=ENCRYPTION_KEY
|
|
)
|
|
photo_only = draft(title="")
|
|
|
|
validated = main.UnfiledDraft.model_validate(photo_only).model_dump()
|
|
assert store.replace("timmy", 0, [validated])["drafts"][0]["title"] == ""
|
|
|
|
empty = draft(title="", evidence=[])
|
|
with pytest.raises(ValueError, match="title or evidence is required"):
|
|
main.UnfiledDraft.model_validate(empty)
|
|
with pytest.raises(ValueError, match="title or evidence is required"):
|
|
store.replace("timmy", 1, [empty])
|
|
|
|
|
|
def test_unfiled_drafts_validate_and_round_trip_complete_filing_plan(tmp_path):
|
|
store = UnfiledDraftStore(
|
|
tmp_path / "unfiled.sqlite3", encryption_key=ENCRYPTION_KEY
|
|
)
|
|
planned = draft(evidence=[])
|
|
planned.pop("evidence")
|
|
planned["filing_plan"] = {
|
|
"repository": "stackchain/dashboard",
|
|
"label_ids": [7, 3],
|
|
"milestone_id": 12,
|
|
"due_date": "2026-08-21",
|
|
"template_name": "Bug report",
|
|
"template_id": "bug.yml",
|
|
"captured_body": "Original field notes",
|
|
"assignee": "alexander",
|
|
"assignee_name": "Alexander",
|
|
"estimate_minutes": 45,
|
|
"completion_intent": "create-and-start",
|
|
}
|
|
|
|
assert store.replace("timmy", 0, [planned])["drafts"] == [planned]
|
|
|
|
invalid = draft(evidence=[])
|
|
invalid["filing_plan"] = {**planned["filing_plan"], "repository": "not-a-repository"}
|
|
with pytest.raises(ValueError, match="filing repository is invalid"):
|
|
store.replace("timmy", 1, [invalid])
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_unfiled_draft_api_is_account_scoped_csrf_protected_and_conflict_safe(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_UNFILED_DRAFT_DB", str(tmp_path / "drafts.sqlite3"))
|
|
monkeypatch.setenv(
|
|
"STACKCHAIN_UNFILED_DRAFT_ENCRYPTION_KEY",
|
|
"ZGRkZGRkZGRkZGRkZGRkZGRkZGRkZGRkZGRkZGRkZGQ=",
|
|
)
|
|
identity = {"login": "Timmy"}
|
|
|
|
async def user():
|
|
return {"id": 1, "login": identity["login"]}
|
|
|
|
monkeypatch.setattr(main, "current_user", user)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
planned = draft(evidence=[])
|
|
planned["filing_plan"] = {
|
|
"repository": "stackchain/dashboard",
|
|
"label_ids": [7, 3],
|
|
"milestone_id": 12,
|
|
"due_date": "2026-08-21",
|
|
"template_name": "Bug report",
|
|
"template_id": "bug.yml",
|
|
"captured_body": "Original field notes",
|
|
"assignee": "alexander",
|
|
"assignee_name": "Alexander",
|
|
"estimate_minutes": 45,
|
|
"completion_intent": "create-and-start",
|
|
}
|
|
payload = {"revision": 0, "drafts": [planned]}
|
|
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.put("/api/v1/unfiled-drafts", json=payload)
|
|
headers = {"Origin": "https://test", "X-CSRF-Token": client.cookies["stackchain_csrf"]}
|
|
saved = await client.put("/api/v1/unfiled-drafts", json=payload, headers=headers)
|
|
stale = await client.put("/api/v1/unfiled-drafts", json=payload, headers=headers)
|
|
fetched = await client.get("/api/v1/unfiled-drafts")
|
|
identity["login"] = "Alexander"
|
|
isolated = await client.get("/api/v1/unfiled-drafts")
|
|
monkeypatch.delenv("STACKCHAIN_UNFILED_DRAFT_ENCRYPTION_KEY")
|
|
unavailable = await client.get("/api/v1/unfiled-drafts")
|
|
monkeypatch.setenv("STACKCHAIN_UNFILED_DRAFT_ENCRYPTION_KEY", "not-base64")
|
|
malformed = await client.get("/api/v1/unfiled-drafts")
|
|
|
|
assert forbidden.status_code == 403
|
|
assert saved.status_code == 200
|
|
expected = planned.copy()
|
|
expected.pop("evidence")
|
|
assert saved.json() == {"revision": 1, "drafts": [expected]}
|
|
assert stale.status_code == 409
|
|
assert stale.json()["detail"]["snapshot"] == saved.json()
|
|
assert fetched.json() == saved.json()
|
|
assert fetched.headers["cache-control"] == "no-store"
|
|
assert isolated.json() == {"revision": 0, "drafts": []}
|
|
assert unavailable.status_code == 503
|
|
assert unavailable.json()["detail"] == "Draft synchronization is unavailable"
|
|
assert malformed.status_code == 503
|
|
assert malformed.json()["detail"] == "Draft synchronization is unavailable"
|