161 lines
6.0 KiB
Python
161 lines
6.0 KiB
Python
import sqlite3
|
|
from datetime import datetime, timezone
|
|
|
|
import pytest
|
|
|
|
from src import main
|
|
from src.today_store import TodayPromotionConflict, TodayStore, WeekPlanConflict
|
|
|
|
|
|
def sample_days():
|
|
return [
|
|
{
|
|
"plan_date": "2026-08-21",
|
|
"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-23",
|
|
"ids": ["issue:secret/repo:5:"],
|
|
"capacity_minutes": 90,
|
|
"estimates": {"issue:secret/repo:5:": 30},
|
|
},
|
|
]
|
|
|
|
|
|
def test_week_plan_is_encrypted_account_scoped_revisioned_and_preserves_other_dates(tmp_path):
|
|
path = tmp_path / "today.sqlite3"
|
|
store = TodayStore(path, encryption_key=b"w" * 32)
|
|
|
|
saved = store.replace_week(
|
|
"Timmy", base_revision=0, days=sample_days(), timezone="America/Los_Angeles"
|
|
)
|
|
|
|
assert saved == {
|
|
"revision": 1,
|
|
"timezone": "America/Los_Angeles",
|
|
"days": sample_days(),
|
|
}
|
|
assert store.get_week("timmy") == saved
|
|
assert store.get_week("alexander") == {"revision": 0, "timezone": None, "days": []}
|
|
retained = path.read_bytes()
|
|
assert b"issue:secret/repo" not in retained
|
|
assert b"America/Los_Angeles" not in retained
|
|
|
|
changed = [dict(sample_days()[0], capacity_minutes=180), sample_days()[1]]
|
|
updated = store.replace_week("timmy", base_revision=1, days=changed, timezone="America/Los_Angeles")
|
|
assert updated["days"][1] == saved["days"][1]
|
|
with pytest.raises(WeekPlanConflict) as conflict:
|
|
store.replace_week("timmy", base_revision=1, days=[], timezone="UTC")
|
|
assert conflict.value.snapshot == updated
|
|
|
|
|
|
def test_week_promotion_moves_only_due_date_to_today_exactly_once(tmp_path):
|
|
store = TodayStore(
|
|
tmp_path / "today.sqlite3",
|
|
encryption_key=b"p" * 32,
|
|
clock=lambda: datetime(2026, 8, 22, 8, tzinfo=timezone.utc).timestamp(),
|
|
)
|
|
today = store.get("timmy")
|
|
week = store.replace_week("timmy", base_revision=0, days=sample_days(), timezone="UTC")
|
|
|
|
promoted = store.promote_week(
|
|
"timmy", promotion_id="week-2026-08-21-r1", week_revision=week["revision"],
|
|
plan_date="2026-08-21", today_revision=today["revision"],
|
|
)
|
|
replay = store.promote_week(
|
|
"timmy", promotion_id="week-2026-08-21-r1", week_revision=week["revision"],
|
|
plan_date="2026-08-21", today_revision=today["revision"],
|
|
)
|
|
|
|
assert replay == promoted
|
|
assert promoted["ids"] == sample_days()[0]["ids"]
|
|
remaining = store.get_week("timmy")
|
|
assert remaining["revision"] == 2
|
|
assert remaining["days"] == [sample_days()[1]]
|
|
|
|
|
|
def test_week_promotion_preserves_nonempty_today_and_the_due_week(tmp_path):
|
|
store = TodayStore(
|
|
tmp_path / "today.sqlite3",
|
|
encryption_key=b"p" * 32,
|
|
clock=lambda: datetime(2026, 8, 22, 8, tzinfo=timezone.utc).timestamp(),
|
|
)
|
|
today = store.apply("timmy", "active", "add", "issue:r:active:")
|
|
week = store.replace_week("timmy", base_revision=0, days=sample_days(), timezone="UTC")
|
|
|
|
with pytest.raises(TodayPromotionConflict) as blocked:
|
|
store.promote_week(
|
|
"timmy", promotion_id="week-2026-08-21-r1", week_revision=week["revision"],
|
|
plan_date="2026-08-21", today_revision=today["revision"],
|
|
)
|
|
|
|
assert blocked.value.today == today
|
|
assert blocked.value.week == week
|
|
assert store.get("timmy") == today
|
|
assert store.get_week("timmy") == week
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_week_promotion_api_returns_both_preserved_plans_for_review(monkeypatch, tmp_path):
|
|
async def user():
|
|
return {"login": "timmy"}
|
|
|
|
store = TodayStore(
|
|
tmp_path / "today.sqlite3",
|
|
encryption_key=b"p" * 32,
|
|
clock=lambda: datetime(2026, 8, 22, 8, tzinfo=timezone.utc).timestamp(),
|
|
)
|
|
today = store.apply("timmy", "active", "add", "issue:r:active:")
|
|
week = store.replace_week("timmy", base_revision=0, days=sample_days(), timezone="UTC")
|
|
monkeypatch.setattr(main, "current_user", user)
|
|
monkeypatch.setattr(main, "_today_store", lambda: store)
|
|
|
|
with pytest.raises(main.HTTPException) as blocked:
|
|
await main.promote_week_plan(main.WeekPromotion(
|
|
promotion_id="week-2026-08-21-r1", week_revision=week["revision"],
|
|
plan_date="2026-08-21", today_revision=today["revision"],
|
|
))
|
|
|
|
assert blocked.value.status_code == 409
|
|
assert blocked.value.detail == {
|
|
"code": "week_today_in_progress", "today": today, "week": week,
|
|
}
|
|
|
|
|
|
def test_existing_tomorrow_plan_migrates_into_week_without_losing_planning_data(tmp_path):
|
|
store = TodayStore(tmp_path / "today.sqlite3", encryption_key=b"m" * 32)
|
|
tomorrow = store.replace_tomorrow(
|
|
"timmy", base_revision=0, ids=["issue:r:9:"], capacity_minutes=75,
|
|
estimates={"issue:r:9:": 45}, plan_date="2026-08-21", timezone="UTC",
|
|
)
|
|
|
|
week = store.get_week("timmy")
|
|
|
|
assert week == {
|
|
"revision": 1, "timezone": "UTC",
|
|
"days": [{key: tomorrow[key] for key in ("plan_date", "ids", "capacity_minutes", "estimates")}],
|
|
}
|
|
assert store.get_tomorrow("timmy") == {
|
|
"revision": 2, "ids": [], "capacity_minutes": None, "estimates": {}
|
|
}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_week_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.WeekPlanUpdate(base_revision=0, days=sample_days(), timezone="UTC")
|
|
|
|
saved = await main.replace_week_plan(payload)
|
|
assert await main.get_week_plan() == saved
|
|
with pytest.raises(main.HTTPException) as raised:
|
|
await main.replace_week_plan(payload)
|
|
assert raised.value.status_code == 409
|
|
assert raised.value.detail == {"code": "week_changed", "snapshot": saved}
|