import sqlite3 import pytest import httpx from src import main from src.today_store import TodayPlanFull, TodaySessionConflict, TodayStore def test_active_session_is_durable_and_account_scoped(tmp_path): path = tmp_path / "today.sqlite3" store = TodayStore(path, clock=lambda: 1234.5) saved = store.update_session( "Timmy", base_revision=0, device_id="phone-a", identity="issue:stackchain/dashboard:42:", elapsed_ms=90_000, running=True, ) assert saved == { "revision": 1, "device_id": "phone-a", "identity": "issue:stackchain/dashboard:42:", "elapsed_ms": 90_000, "running": True, "break_deadline_at": None, "updated_at": 1234.5, } assert TodayStore(path).get_session("timmy") == saved assert store.get_session("alexander") == { "revision": 0, "device_id": "", "identity": "", "elapsed_ms": 0, "running": False, "break_deadline_at": None, "updated_at": None, } def test_timed_break_deadline_is_durable_in_the_account_session(tmp_path): path = tmp_path / "today.sqlite3" store = TodayStore(path, clock=lambda: 1234.5) saved = store.update_session( "timmy", base_revision=0, device_id="phone-a", identity="issue:r:42:", elapsed_ms=90_000, running=False, break_deadline_at=1_800_000, ) assert saved["break_deadline_at"] == 1_800_000 assert saved["running"] is False assert TodayStore(path).get_session("timmy") == saved def test_active_session_claim_rejects_a_stale_revision(tmp_path): store = TodayStore(tmp_path / "today.sqlite3") first = store.update_session( "timmy", base_revision=0, device_id="phone-a", identity="issue:r:1:", elapsed_ms=1_000, running=True, ) claimed = store.update_session( "timmy", base_revision=first["revision"], device_id="phone-b", identity=first["identity"], elapsed_ms=first["elapsed_ms"], running=True, ) with pytest.raises(TodaySessionConflict) as raised: store.update_session( "timmy", base_revision=first["revision"], device_id="phone-a", identity=first["identity"], elapsed_ms=5_000, running=True, ) assert raised.value.session == claimed assert store.get_session("timmy") == claimed @pytest.mark.anyio async def test_active_session_api_persists_a_bounded_break_deadline(monkeypatch, tmp_path): async def user(): return {"login": "timmy"} store = TodayStore(tmp_path / "today.sqlite3") monkeypatch.setattr(main, "current_user", user) monkeypatch.setattr(main, "_today_store", lambda: store) payload = main.TodaySessionUpdate( base_revision=0, device_id="phone-a", identity="issue:r:1:", elapsed_ms=5_000, running=False, break_deadline_at=1_800_000, ) saved = await main.update_today_session(payload) assert saved["break_deadline_at"] == 1_800_000 assert (await main.get_today_session()) == saved def test_active_session_api_rejects_a_running_break(): with pytest.raises(ValueError, match="break must remain paused"): main.TodaySessionUpdate( base_revision=0, device_id="phone-a", identity="issue:r:1:", elapsed_ms=5_000, running=True, break_deadline_at=1_800_000, ) @pytest.mark.anyio async def test_active_session_api_returns_current_state_on_claim_conflict(monkeypatch, tmp_path): async def user(): return {"login": "timmy"} store = TodayStore(tmp_path / "today.sqlite3") current = store.update_session( "timmy", base_revision=0, device_id="phone-a", identity="issue:r:1:", elapsed_ms=5_000, running=True, ) monkeypatch.setattr(main, "current_user", user) monkeypatch.setattr(main, "_today_store", lambda: store) payload = main.TodaySessionUpdate( base_revision=0, device_id="phone-b", identity="issue:r:1:", elapsed_ms=5_000, running=True, ) with pytest.raises(main.HTTPException) as raised: await main.update_today_session(payload) assert raised.value.status_code == 409 assert raised.value.detail == {"code": "session_changed", "session": current} @pytest.mark.anyio async def test_transient_today_store_failure_tells_clients_when_to_retry(monkeypatch): async def user(): return {"login": "timmy"} class BusyStore: def apply(self, *_args, **_kwargs): raise sqlite3.OperationalError("database is locked") monkeypatch.setattr(main, "current_user", user) monkeypatch.setattr(main, "_today_store", lambda: BusyStore()) payload = main.TodayOperation( operation_id="retry-me", action="add", item_id="issue:r:1:" ) with pytest.raises(main.HTTPException) as raised: await main.update_today_plan(payload) assert raised.value.status_code == 503 assert raised.value.headers == {"Retry-After": "1"} def test_operations_are_durable_ordered_idempotent_and_account_scoped(tmp_path): path = tmp_path / "today.sqlite3" store = TodayStore(path, limit=3) assert store.apply("timmy", "op-1", "add", "issue:stackchain/dashboard:1:") == { "revision": 1, "ids": ["issue:stackchain/dashboard:1:"], "capacity_minutes": None, "estimates": {}, } store.apply("timmy", "op-2", "add", "issue:stackchain/dashboard:2:") store.apply("timmy", "op-3", "add", "issue:stackchain/dashboard:3:") moved = store.apply( "timmy", "op-4", "move", "issue:stackchain/dashboard:3:", direction="up" ) duplicate = store.apply( "timmy", "op-4", "move", "issue:stackchain/dashboard:3:", direction="up" ) assert moved == duplicate == { "revision": 4, "ids": [ "issue:stackchain/dashboard:1:", "issue:stackchain/dashboard:3:", "issue:stackchain/dashboard:2:", ], "capacity_minutes": None, "estimates": {}, } assert TodayStore(path, limit=3).get("timmy") == moved assert store.get("alexander") == { "revision": 0, "ids": [], "capacity_minutes": None, "estimates": {}, } def test_initialized_today_reads_remain_available_during_a_planning_write(tmp_path): path = tmp_path / "today.sqlite3" store = TodayStore(path, timeout=0.05) expected = store.apply("timmy", "seed", "add", "issue:r:1:") writer = sqlite3.connect(path) writer.execute("BEGIN IMMEDIATE") try: assert TodayStore(path, timeout=0.05).get("timmy") == expected finally: writer.rollback() writer.close() def test_capacity_and_estimates_are_durable_account_scoped_and_follow_item_identity(tmp_path): path = tmp_path / "today.sqlite3" store = TodayStore(path, limit=3) store.apply("timmy", "add-1", "add", "issue:r:1:") store.apply("timmy", "add-2", "add", "issue:r:2:") configured = store.apply_batch("timmy", [{ "operation_id": "capacity-1", "action": "configure", "item_id": "plan", "capacity_minutes": 180, "estimates": {"issue:r:1:": 60, "issue:r:2:": 45, "issue:r:99:": 30}, "base_revision": 2, }]) store.apply("timmy", "move-2", "move", "issue:r:2:", direction="up") store.apply("timmy", "remove-1", "remove", "issue:r:1:") assert configured["capacity_minutes"] == 180 assert configured["estimates"] == {"issue:r:1:": 60, "issue:r:2:": 45} assert TodayStore(path, limit=3).get("timmy") == { "revision": 5, "ids": ["issue:r:2:"], "capacity_minutes": 180, "estimates": {"issue:r:2:": 45}, } assert store.get("alexander") == { "revision": 0, "ids": [], "capacity_minutes": None, "estimates": {}, } def test_limit_is_atomic_and_remove_frees_capacity(tmp_path): store = TodayStore(tmp_path / "today.sqlite3", limit=2) store.apply("timmy", "one", "add", "issue:r:1:") store.apply("timmy", "two", "add", "issue:r:2:") with pytest.raises(TodayPlanFull): store.apply("timmy", "three", "add", "issue:r:3:") assert store.get("timmy") == { "revision": 2, "ids": ["issue:r:1:", "issue:r:2:"], "capacity_minutes": None, "estimates": {}, } store.apply("timmy", "remove", "remove", "issue:r:1:") assert store.apply("timmy", "retry-three", "add", "issue:r:3:")["ids"] == [ "issue:r:2:", "issue:r:3:", ] def test_batch_applies_ordered_operations_once_and_rejects_only_capacity_conflicts(tmp_path): store = TodayStore(tmp_path / "today.sqlite3", limit=2) result = store.apply_batch( "timmy", [ {"operation_id": "one", "action": "add", "item_id": "issue:r:1:"}, {"operation_id": "two", "action": "add", "item_id": "issue:r:2:"}, {"operation_id": "full", "action": "add", "item_id": "issue:r:3:"}, {"operation_id": "remove", "action": "remove", "item_id": "issue:r:1:"}, {"operation_id": "three", "action": "add", "item_id": "issue:r:3:"}, ], ) assert result == { "revision": 4, "ids": ["issue:r:2:", "issue:r:3:"], "capacity_minutes": None, "estimates": {}, "accepted_operation_ids": ["one", "two", "remove", "three"], "duplicate_operation_ids": [], "rejected_operations": [{"operation_id": "full", "reason": "today_full"}], } replay = store.apply_batch("timmy", [ {"operation_id": "two", "action": "add", "item_id": "issue:r:2:"}, {"operation_id": "three", "action": "add", "item_id": "issue:r:3:"}, ]) assert replay["revision"] == 4 assert replay["accepted_operation_ids"] == [] assert replay["duplicate_operation_ids"] == ["two", "three"] def test_operation_receipts_are_bounded_per_account_and_existing_schema_migrates(tmp_path): path = tmp_path / "today.sqlite3" with sqlite3.connect(path) as connection: connection.execute( "CREATE TABLE today_operations (login TEXT NOT NULL, operation_id TEXT NOT NULL, " "PRIMARY KEY (login, operation_id))" ) connection.execute( "INSERT INTO today_operations(login, operation_id) VALUES ('timmy', 'legacy')" ) store = TodayStore(path, limit=20, operation_limit=3, clock=lambda: 1_000) for index in range(5): store.apply("timmy", f"op-{index}", "add", f"issue:r:{index}:") store.apply("alexander", "other", "add", "issue:r:99:") with sqlite3.connect(path) as connection: columns = {row[1] for row in connection.execute("PRAGMA table_info(today_operations)")} timmy = connection.execute( "SELECT operation_id FROM today_operations WHERE login = 'timmy' ORDER BY rowid" ).fetchall() alexander = connection.execute( "SELECT operation_id FROM today_operations WHERE login = 'alexander'" ).fetchall() assert "created_at" in columns assert timmy == [("op-2",), ("op-3",), ("op-4",)] assert alexander == [("other",)] def test_pruned_stale_today_operation_cannot_reorder_a_newer_plan(tmp_path): store = TodayStore(tmp_path / "today.sqlite3", limit=5, operation_limit=1) store.apply_batch("timmy", [ {"operation_id": "seed-1", "action": "add", "item_id": "issue:r:1:", "base_revision": 0}, {"operation_id": "seed-2", "action": "add", "item_id": "issue:r:2:", "base_revision": 0}, ]) store.apply_batch("timmy", [{ "operation_id": "offline-move", "action": "move", "item_id": "issue:r:2:", "direction": "up", "base_revision": 2, }]) newer = store.apply_batch("timmy", [{ "operation_id": "newer-move", "action": "move", "item_id": "issue:r:1:", "direction": "up", "base_revision": 3, }]) replay = store.apply_batch("timmy", [{ "operation_id": "offline-move", "action": "move", "item_id": "issue:r:2:", "direction": "up", "base_revision": 2, }]) assert replay["ids"] == newer["ids"] == ["issue:r:1:", "issue:r:2:"] assert replay["revision"] == 4 assert replay["accepted_operation_ids"] == [] assert replay["rejected_operations"] == [ {"operation_id": "offline-move", "reason": "stale_intent"} ] def test_today_receipt_age_pruning_is_account_scoped(tmp_path): clock = [1_000.0] path = tmp_path / "today.sqlite3" store = TodayStore( path, limit=5, operation_retention_seconds=60, clock=lambda: clock[0] ) store.apply("timmy", "old", "add", "issue:r:1:") store.apply("alexander", "other-old", "add", "issue:r:2:") clock[0] += 61 store.apply("timmy", "fresh", "remove", "issue:r:1:") with sqlite3.connect(path) as connection: timmy = connection.execute( "SELECT operation_id FROM today_operations WHERE login = 'timmy'" ).fetchall() alexander = connection.execute( "SELECT operation_id FROM today_operations WHERE login = 'alexander'" ).fetchall() assert timmy == [("fresh",)] assert alexander == [("other-old",)] def test_today_api_model_preserves_client_base_revision(): operation = main.TodayOperation( operation_id="offline", action="move", item_id="issue:r:2:", direction="up", base_revision=7, ) assert operation.model_dump()["base_revision"] == 7 def test_today_api_model_accepts_bounded_capacity_configuration(): operation = main.TodayOperation( operation_id="capacity", action="configure", item_id="plan", capacity_minutes=180, estimates={"issue:r:1:": 60, "issue:r:2:": 45}, base_revision=2, ) assert operation.model_dump() == { "operation_id": "capacity", "action": "configure", "item_id": "plan", "direction": None, "base_revision": 2, "capacity_minutes": 180, "estimates": {"issue:r:1:": 60, "issue:r:2:": 45}, } @pytest.mark.anyio async def test_single_capacity_configuration_uses_atomic_batch_path(monkeypatch): async def user(): return {"login": "timmy"} calls = [] class Store: def apply_batch(self, login, operations): calls.append((login, operations)) return {"revision": 1, "ids": [], "capacity_minutes": 180, "estimates": {}} monkeypatch.setattr(main, "current_user", user) monkeypatch.setattr(main, "_today_store", lambda: Store()) payload = main.TodayOperation( operation_id="capacity", action="configure", item_id="plan", capacity_minutes=180, ) result = await main.update_today_plan(payload) assert result["capacity_minutes"] == 180 assert calls == [("timmy", [payload.model_dump()])] @pytest.mark.anyio async def test_authenticated_today_api_uses_confirmed_account_and_csrf(monkeypatch, tmp_path): monkeypatch.setenv("STACKCHAIN_DASHBOARD_AUTH_MODE", "operator") monkeypatch.setenv("STACKCHAIN_DASHBOARD_ACCESS_TOKEN", "correct horse battery staple") monkeypatch.setenv( "STACKCHAIN_DASHBOARD_SESSION_SECRET", "a-separate-session-signing-secret-with-enough-entropy", ) monkeypatch.setenv("STACKCHAIN_SESSION_DB", str(tmp_path / "sessions.sqlite3")) monkeypatch.setenv("STACKCHAIN_LOGIN_ATTEMPT_DB", str(tmp_path / "login.sqlite3")) monkeypatch.setenv("STACKCHAIN_TODAY_DB", str(tmp_path / "today.sqlite3")) async def user(): return {"id": 1, "login": "Timmy"} monkeypatch.setattr(main, "current_user", user) transport = httpx.ASGITransport(app=main.app) async with httpx.AsyncClient(transport=transport, base_url="https://test") as client: await client.post( "/api/v1/session", json={"access_token": "correct horse battery staple"} ) forbidden = await client.patch( "/api/v1/today", json={ "operation_id": "mobile-1", "action": "add", "item_id": "issue:stackchain/dashboard:357:", }, ) changed = await client.patch( "/api/v1/today", json={ "operations": [ {"operation_id": "mobile-1", "action": "add", "item_id": "issue:stackchain/dashboard:357:"}, {"operation_id": "mobile-2", "action": "add", "item_id": "issue:stackchain/dashboard:359:"}, ], }, headers={ "Origin": "https://test", "X-CSRF-Token": client.cookies["stackchain_csrf"], }, ) fetched = await client.get("/api/v1/today") changed_session = await client.patch( "/api/v1/today/session", json={ "base_revision": 0, "device_id": "phone-a", "identity": "issue:stackchain/dashboard:357:", "elapsed_ms": 12_000, "running": True, }, headers={ "Origin": "https://test", "X-CSRF-Token": client.cookies["stackchain_csrf"], }, ) fetched_session = await client.get("/api/v1/today/session") assert forbidden.status_code == 403 assert changed.status_code == 200 assert changed.json() == { "revision": 2, "ids": ["issue:stackchain/dashboard:357:", "issue:stackchain/dashboard:359:"], "capacity_minutes": None, "estimates": {}, "accepted_operation_ids": ["mobile-1", "mobile-2"], "duplicate_operation_ids": [], "rejected_operations": [], } assert fetched.json() == { "revision": 2, "ids": ["issue:stackchain/dashboard:357:", "issue:stackchain/dashboard:359:"], "capacity_minutes": None, "estimates": {}, } assert fetched.headers["cache-control"] == "no-store" assert changed_session.status_code == 200 assert fetched_session.json() == changed_session.json() assert fetched_session.headers["cache-control"] == "no-store"