120 lines
4.3 KiB
Python
120 lines
4.3 KiB
Python
import sqlite3
|
|
|
|
from src.following_store import FollowingStore
|
|
|
|
|
|
KEY = b"f" * 32
|
|
ITEM = {
|
|
"repository": "stackchain/api",
|
|
"number": 42,
|
|
"title": "Make mobile review useful",
|
|
"state": "open",
|
|
"updated_at": "2026-08-23T03:00:00Z",
|
|
"url": "https://forge.example/stackchain/api/issues/42",
|
|
}
|
|
|
|
|
|
def test_confirmed_watch_is_account_scoped_idempotent_and_encrypted(tmp_path):
|
|
path = tmp_path / "following.sqlite3"
|
|
store = FollowingStore(path, encryption_key=KEY, limit=50)
|
|
|
|
first = store.set_watching("Timmy", ITEM, True)
|
|
repeated = store.set_watching("timmy", ITEM, True)
|
|
|
|
assert first == repeated == {
|
|
"revision": 1,
|
|
"items": [{**ITEM, "kind": "issue", "has_unseen_change": False}],
|
|
}
|
|
assert store.get("other") == {"revision": 0, "items": []}
|
|
stored = sqlite3.connect(path).execute(
|
|
"SELECT items FROM following_issues WHERE login = ?", ("timmy",)
|
|
).fetchone()[0]
|
|
assert ITEM["title"] not in stored
|
|
assert ITEM["repository"] not in stored
|
|
|
|
|
|
def test_refresh_marks_changed_items_unseen_and_sorts_them_first(tmp_path):
|
|
store = FollowingStore(tmp_path / "following.sqlite3", encryption_key=KEY)
|
|
quiet = {**ITEM, "number": 7, "title": "Quiet", "updated_at": "2026-08-23T02:00:00Z"}
|
|
store.set_watching("timmy", quiet, True)
|
|
store.set_watching("timmy", ITEM, True)
|
|
|
|
snapshot = store.refresh("timmy", [{
|
|
**quiet, "title": "Changed upstream", "state": "closed",
|
|
"updated_at": "2026-08-23T04:00:00Z",
|
|
}])
|
|
|
|
assert [(item["number"], item["has_unseen_change"]) for item in snapshot["items"]] == [
|
|
(7, True), (42, False)
|
|
]
|
|
assert snapshot["items"][0]["title"] == "Changed upstream"
|
|
assert snapshot["items"][0]["state"] == "closed"
|
|
|
|
|
|
def test_acknowledgement_is_revision_conditional_and_later_change_is_unseen(tmp_path):
|
|
store = FollowingStore(tmp_path / "following.sqlite3", encryption_key=KEY)
|
|
store.set_watching("timmy", ITEM, True)
|
|
changed = {**ITEM, "updated_at": "2026-08-23T04:00:00Z"}
|
|
store.refresh("timmy", [changed])
|
|
|
|
stale = store.acknowledge("timmy", ITEM["repository"], ITEM["number"], ITEM["updated_at"])
|
|
assert stale["items"][0]["has_unseen_change"] is True
|
|
seen = store.acknowledge("timmy", ITEM["repository"], ITEM["number"], changed["updated_at"])
|
|
assert seen["items"][0]["has_unseen_change"] is False
|
|
|
|
store.refresh("timmy", [{**ITEM, "updated_at": "2026-08-23T05:00:00Z"}])
|
|
assert store.get("timmy")["items"][0]["has_unseen_change"] is True
|
|
|
|
|
|
def test_reconfirmed_watch_does_not_mark_an_unseen_change_as_seen(tmp_path):
|
|
store = FollowingStore(tmp_path / "following.sqlite3", encryption_key=KEY)
|
|
store.set_watching("timmy", ITEM, True)
|
|
changed = {**ITEM, "updated_at": "2026-08-23T04:00:00Z"}
|
|
store.refresh("timmy", [changed])
|
|
|
|
snapshot = store.set_watching("timmy", changed, True)
|
|
|
|
assert snapshot["items"][0]["has_unseen_change"] is True
|
|
|
|
|
|
def test_following_preserves_kind_and_uses_it_in_item_identity(tmp_path):
|
|
store = FollowingStore(tmp_path / "following.sqlite3", encryption_key=KEY)
|
|
issue = {**ITEM, "kind": "issue"}
|
|
pull = {
|
|
**ITEM,
|
|
"kind": "pull",
|
|
"title": "Review the mobile flow",
|
|
"url": "https://forge.example/stackchain/api/pulls/42",
|
|
}
|
|
|
|
store.set_watching("timmy", issue, True)
|
|
snapshot = store.set_watching("timmy", pull, True)
|
|
|
|
assert [(item["kind"], item["number"]) for item in snapshot["items"]] == [
|
|
("pull", 42),
|
|
("issue", 42),
|
|
]
|
|
|
|
|
|
def test_following_reads_legacy_items_as_issues(tmp_path):
|
|
store = FollowingStore(tmp_path / "following.sqlite3", encryption_key=KEY)
|
|
store.set_watching("timmy", ITEM, True)
|
|
|
|
assert store.get("timmy")["items"][0]["kind"] == "issue"
|
|
|
|
|
|
def test_acknowledgement_only_marks_the_requested_item_kind_seen(tmp_path):
|
|
store = FollowingStore(tmp_path / "following.sqlite3", encryption_key=KEY)
|
|
for kind in ("issue", "pull"):
|
|
store.set_watching("timmy", {**ITEM, "kind": kind}, True)
|
|
store.refresh("timmy", [{**ITEM, "kind": kind, "updated_at": "2026-08-23T06:00:00Z"}])
|
|
|
|
snapshot = store.acknowledge(
|
|
"timmy", ITEM["repository"], ITEM["number"], "2026-08-23T06:00:00Z", kind="pull"
|
|
)
|
|
|
|
assert {item["kind"]: item["has_unseen_change"] for item in snapshot["items"]} == {
|
|
"issue": True,
|
|
"pull": False,
|
|
}
|