113 lines
4.4 KiB
Python
113 lines
4.4 KiB
Python
import sqlite3
|
|
|
|
import pytest
|
|
|
|
from src import main
|
|
from src.today_store import TodayStore, TomorrowPlanConflict, TodayPromotionConflict
|
|
|
|
|
|
def test_tomorrow_plan_is_encrypted_scoped_and_promotes_atomically_exactly_once(tmp_path):
|
|
path = tmp_path / "today.sqlite3"
|
|
key = b"n" * 32
|
|
store = TodayStore(path, encryption_key=key)
|
|
store.apply("timmy", "today-1", "add", "issue:r:1:")
|
|
|
|
tomorrow = store.replace_tomorrow(
|
|
"Timmy", base_revision=0,
|
|
ids=["issue:secret/repo:3:", "issue:secret/repo:2:"],
|
|
capacity_minutes=150,
|
|
estimates={"issue:secret/repo:3:": 60, "issue:secret/repo:2:": 45},
|
|
plan_date="2026-08-20", timezone="America/Los_Angeles",
|
|
)
|
|
|
|
assert tomorrow == {
|
|
"revision": 1,
|
|
"ids": ["issue:secret/repo:3:", "issue:secret/repo:2:"],
|
|
"capacity_minutes": 150,
|
|
"estimates": {"issue:secret/repo:3:": 60, "issue:secret/repo:2:": 45},
|
|
"plan_date": "2026-08-20",
|
|
"timezone": "America/Los_Angeles",
|
|
}
|
|
assert store.get("timmy")["ids"] == ["issue:r:1:"]
|
|
assert store.get_tomorrow("alexander")["ids"] == []
|
|
retained = path.read_bytes()
|
|
assert b"issue:secret/repo" not in retained
|
|
assert b"America/Los_Angeles" not in retained
|
|
|
|
promoted = store.promote_tomorrow(
|
|
"timmy", promotion_id="rollover-2026-08-20",
|
|
tomorrow_revision=1, today_revision=1,
|
|
)
|
|
replay = store.promote_tomorrow(
|
|
"timmy", promotion_id="rollover-2026-08-20",
|
|
tomorrow_revision=1, today_revision=1,
|
|
)
|
|
|
|
assert replay == promoted
|
|
assert promoted["ids"] == tomorrow["ids"]
|
|
assert promoted["capacity_minutes"] == 150
|
|
assert promoted["plan_date"] == "2026-08-20"
|
|
assert store.get_tomorrow("timmy") == {
|
|
"revision": 2, "ids": [], "capacity_minutes": None, "estimates": {}
|
|
}
|
|
|
|
|
|
def test_tomorrow_edit_and_promotion_reject_stale_revisions_without_touching_today(tmp_path):
|
|
store = TodayStore(tmp_path / "today.sqlite3", encryption_key=b"s" * 32)
|
|
original_today = store.apply("timmy", "today", "add", "issue:r:1:")
|
|
first = store.replace_tomorrow(
|
|
"timmy", base_revision=0, ids=["issue:r:2:"], capacity_minutes=60,
|
|
estimates={"issue:r:2:": 30}, plan_date="2026-08-20", timezone="UTC",
|
|
)
|
|
|
|
with pytest.raises(TomorrowPlanConflict) as stale_edit:
|
|
store.replace_tomorrow(
|
|
"timmy", base_revision=0, ids=["issue:r:9:"], capacity_minutes=None,
|
|
estimates={}, plan_date="2026-08-20", timezone="UTC",
|
|
)
|
|
assert stale_edit.value.snapshot == first
|
|
|
|
store.apply("timmy", "today-changed", "add", "issue:r:3:")
|
|
with pytest.raises(TodayPromotionConflict) as stale_today:
|
|
store.promote_tomorrow(
|
|
"timmy", promotion_id="rollover", tomorrow_revision=1,
|
|
today_revision=original_today["revision"],
|
|
)
|
|
assert stale_today.value.today["ids"] == ["issue:r:1:", "issue:r:3:"]
|
|
assert store.get_tomorrow("timmy") == first
|
|
|
|
|
|
def test_tomorrow_validation_is_bounded_and_date_specific(tmp_path):
|
|
store = TodayStore(tmp_path / "today.sqlite3")
|
|
with pytest.raises(ValueError, match="ISO calendar date"):
|
|
store.replace_tomorrow(
|
|
"timmy", base_revision=0, ids=[], capacity_minutes=None, estimates={},
|
|
plan_date="tomorrow", timezone="UTC",
|
|
)
|
|
with pytest.raises(ValueError, match="unique"):
|
|
store.replace_tomorrow(
|
|
"timmy", base_revision=0, ids=["same", "same"], capacity_minutes=None,
|
|
estimates={}, plan_date="2026-08-20", timezone="UTC",
|
|
)
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_tomorrow_api_round_trip_and_conflict_contract(monkeypatch, tmp_path):
|
|
async def user():
|
|
return {"login": "timmy"}
|
|
|
|
store = TodayStore(tmp_path / "today.sqlite3", encryption_key=b"a" * 32)
|
|
monkeypatch.setattr(main, "current_user", user)
|
|
monkeypatch.setattr(main, "_today_store", lambda: store)
|
|
payload = main.TomorrowPlanUpdate(
|
|
base_revision=0, ids=["issue:r:2:"], capacity_minutes=90,
|
|
estimates={"issue:r:2:": 45}, plan_date="2026-08-20", timezone="UTC",
|
|
)
|
|
saved = await main.replace_tomorrow_plan(payload)
|
|
assert await main.get_tomorrow_plan() == saved
|
|
|
|
with pytest.raises(main.HTTPException) as raised:
|
|
await main.replace_tomorrow_plan(payload)
|
|
assert raised.value.status_code == 409
|
|
assert raised.value.detail == {"code": "tomorrow_changed", "snapshot": saved}
|