import sqlite3 from datetime import datetime, timezone import pytest from src import main from src.today_store import TodayPromotionConflict, TodayStore, TomorrowPlanConflict 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, clock=lambda: datetime(2026, 8, 21, tzinfo=timezone.utc).timestamp(), ) 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, clock=lambda: datetime(2026, 8, 21, tzinfo=timezone.utc).timestamp(), ) 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_promotion_rejects_the_day_before_in_the_saved_timezone(tmp_path): store = TodayStore( tmp_path / "today.sqlite3", clock=lambda: datetime(2026, 8, 20, 6, 59, tzinfo=timezone.utc).timestamp(), ) today = store.apply("timmy", "today", "add", "issue:r:1:") tomorrow = store.replace_tomorrow( "timmy", base_revision=0, ids=["issue:r:2:"], capacity_minutes=60, estimates={}, plan_date="2026-08-20", timezone="America/Los_Angeles", ) with pytest.raises(ValueError, match="Tomorrow plan is not due"): store.promote_tomorrow( "timmy", promotion_id="tomorrow-2026-08-20-r1", tomorrow_revision=tomorrow["revision"], today_revision=today["revision"], ) assert store.get("timmy") == today assert store.get_tomorrow("timmy") == tomorrow 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} @pytest.mark.anyio async def test_tomorrow_api_reports_an_early_promotion_without_mutation(monkeypatch, tmp_path): async def user(): return {"login": "timmy"} store = TodayStore( tmp_path / "today.sqlite3", clock=lambda: datetime(2026, 8, 20, 6, 59, tzinfo=timezone.utc).timestamp(), ) monkeypatch.setattr(main, "current_user", user) monkeypatch.setattr(main, "_today_store", lambda: store) today = store.apply("timmy", "today", "add", "issue:r:1:") tomorrow = store.replace_tomorrow( "timmy", base_revision=0, ids=["issue:r:2:"], capacity_minutes=60, estimates={}, plan_date="2026-08-20", timezone="America/Los_Angeles", ) with pytest.raises(main.HTTPException) as raised: await main.promote_tomorrow_plan(main.TomorrowPromotion( promotion_id="tomorrow-2026-08-20-r1", tomorrow_revision=tomorrow["revision"], today_revision=today["revision"], )) assert raised.value.status_code == 409 assert raised.value.detail == { "code": "tomorrow_not_due", "plan_date": "2026-08-20" } assert store.get("timmy") == today assert store.get_tomorrow("timmy") == tomorrow