90 lines
3.2 KiB
Python
90 lines
3.2 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": [], "pinned": []}
|
|
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_pinned_work_survives_recent_limit_and_unpin_keeps_recent_item(tmp_path):
|
|
database = tmp_path / "recent-work.sqlite3"
|
|
store = RecentWorkStore(database, encryption_key=b"r" * 32, limit=5)
|
|
|
|
store.record("timmy", item(1))
|
|
pinned = store.pin("timmy", item(1))
|
|
for number in range(2, 8):
|
|
store.record("timmy", item(number))
|
|
|
|
assert pinned["pinned"] == [item(1)]
|
|
assert [entry["number"] for entry in store.get("timmy")["items"]] == [7, 6, 5, 4, 3]
|
|
assert store.get("timmy")["pinned"] == [item(1)]
|
|
assert store.get("alexander") == {"items": [], "pinned": []}
|
|
|
|
store.record("timmy", item(1, title="Issue 1 current"))
|
|
unpinned = store.unpin("timmy", item(1)["route"])
|
|
|
|
assert unpinned["pinned"] == []
|
|
assert unpinned["items"][0] == item(1, title="Issue 1 current")
|
|
with sqlite3.connect(database) as connection:
|
|
payload = connection.execute(
|
|
"SELECT items FROM recent_work WHERE login = 'timmy'"
|
|
).fetchone()[0]
|
|
assert "Issue 1" not in payload
|
|
assert "#/my-work/issue" not in payload
|
|
|
|
|
|
def test_reopening_pinned_work_promotes_it_for_every_device(tmp_path):
|
|
store = RecentWorkStore(
|
|
tmp_path / "recent-work.sqlite3",
|
|
encryption_key=b"r" * 32,
|
|
)
|
|
for number in range(1, 5):
|
|
store.pin("timmy", item(number))
|
|
|
|
reopened = store.record("timmy", item(1, title="Issue 1 current"))
|
|
|
|
assert [entry["number"] for entry in reopened["pinned"]] == [1, 4, 3, 2]
|
|
assert reopened["pinned"][0]["title"] == "Issue 1 current"
|
|
assert store.get("timmy")["pinned"] == reopened["pinned"]
|
|
|
|
|
|
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")
|