48 lines
1.6 KiB
Python
48 lines
1.6 KiB
Python
import sqlite3
|
|
|
|
from src.recent_work_store import RecentWorkStore
|
|
|
|
|
|
def item(number: int, *, title: str | None = None) -> dict:
|
|
return {
|
|
"kind": "issue",
|
|
"repository": "stackchain/dashboard",
|
|
"number": number,
|
|
"title": title or f"Issue {number}",
|
|
"route": f"#/my-work/issue/stackchain/dashboard/{number}",
|
|
}
|
|
|
|
|
|
def test_recent_work_is_encrypted_account_scoped_deduplicated_and_bounded(tmp_path):
|
|
database = tmp_path / "recent-work.sqlite3"
|
|
store = RecentWorkStore(database, encryption_key=b"r" * 32, limit=5)
|
|
|
|
for number in range(1, 7):
|
|
store.record(" Timmy ", item(number))
|
|
expected = store.record("timmy", item(3, title="Issue 3 updated"))
|
|
|
|
assert [entry["number"] for entry in expected["items"]] == [3, 6, 5, 4, 2]
|
|
assert RecentWorkStore(database, encryption_key=b"r" * 32).get("timmy") == expected
|
|
assert store.get("alexander") == {"items": []}
|
|
with sqlite3.connect(database) as connection:
|
|
payload = connection.execute(
|
|
"SELECT items FROM recent_work WHERE login = 'timmy'"
|
|
).fetchone()[0]
|
|
assert payload.startswith("v1:")
|
|
assert "Issue 3 updated" not in payload
|
|
assert "#/my-work/issue" not in payload
|
|
|
|
|
|
def test_recent_work_rejects_noncanonical_or_unsupported_items(tmp_path):
|
|
store = RecentWorkStore(tmp_path / "recent-work.sqlite3", encryption_key=b"r" * 32)
|
|
|
|
invalid = item(1)
|
|
invalid["route"] = "https://attacker.example/"
|
|
|
|
try:
|
|
store.record("timmy", invalid)
|
|
except ValueError as error:
|
|
assert str(error) == "recent work item is invalid"
|
|
else:
|
|
raise AssertionError("invalid route was accepted")
|