323 lines
13 KiB
Python
323 lines
13 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_start_day_plan_selects_earliest_nonempty_week_date_without_mutation(tmp_path):
|
|
store = TodayStore(
|
|
tmp_path / "today.sqlite3", encryption_key=b"r" * 32,
|
|
clock=lambda: datetime(2026, 8, 21, 13, tzinfo=timezone.utc).timestamp(),
|
|
)
|
|
days = [
|
|
{
|
|
"plan_date": "2026-08-20", "ids": [],
|
|
"capacity_minutes": 60, "estimates": {},
|
|
},
|
|
{
|
|
"plan_date": "2026-08-21", "ids": ["issue:private/repo:7:"],
|
|
"capacity_minutes": 90,
|
|
"estimates": {"issue:private/repo:7:": 45},
|
|
},
|
|
{
|
|
"plan_date": "2026-08-22", "ids": ["issue:private/repo:8:"],
|
|
"capacity_minutes": 120,
|
|
"estimates": {"issue:private/repo:8:": 30},
|
|
},
|
|
]
|
|
week = store.replace_week(
|
|
"timmy", base_revision=0, days=days, timezone="America/New_York"
|
|
)
|
|
|
|
assert store.get_start_day_plan("Timmy") == {
|
|
**days[1], "timezone": "America/New_York",
|
|
}
|
|
assert store.get_week("timmy") == week
|
|
|
|
|
|
def test_start_day_plan_falls_back_to_tomorrow_without_migrating_it(tmp_path):
|
|
store = TodayStore(tmp_path / "today.sqlite3", encryption_key=b"r" * 32)
|
|
tomorrow = store.replace_tomorrow(
|
|
"timmy", base_revision=0, ids=["issue:private/repo:9:"],
|
|
capacity_minutes=75, estimates={"issue:private/repo:9:": 45},
|
|
plan_date="2026-08-21", timezone="UTC",
|
|
)
|
|
|
|
assert store.get_start_day_plan("timmy") == tomorrow
|
|
assert store.get_tomorrow("timmy") == tomorrow
|
|
with store._connect() as connection:
|
|
assert connection.execute(
|
|
"SELECT COUNT(*) FROM week_plans WHERE login = ?", ("timmy",)
|
|
).fetchone()[0] == 0
|
|
|
|
|
|
def test_start_day_plan_uses_due_tomorrow_fallback_before_future_week_day(tmp_path):
|
|
store = TodayStore(
|
|
tmp_path / "today.sqlite3", encryption_key=b"r" * 32,
|
|
clock=lambda: datetime(2026, 8, 20, 13, tzinfo=timezone.utc).timestamp(),
|
|
)
|
|
future = [{
|
|
"plan_date": "2026-08-21", "ids": ["issue:private/repo:10:"],
|
|
"capacity_minutes": 90, "estimates": {"issue:private/repo:10:": 45},
|
|
}]
|
|
store.replace_week("timmy", base_revision=0, days=future, timezone="UTC")
|
|
tomorrow = store.replace_tomorrow(
|
|
"timmy", base_revision=0, ids=["issue:private/repo:9:"],
|
|
capacity_minutes=75, estimates={"issue:private/repo:9:": 45},
|
|
plan_date="2026-08-20", timezone="UTC",
|
|
)
|
|
|
|
assert store.get_start_day_plan("timmy") == tomorrow
|
|
|
|
|
|
def test_week_plan_rejects_the_same_work_on_multiple_dates_without_advancing_revision(tmp_path):
|
|
store = TodayStore(tmp_path / "today.sqlite3", encryption_key=b"w" * 32)
|
|
original = store.replace_week(
|
|
"timmy", base_revision=0, days=sample_days(), timezone="UTC"
|
|
)
|
|
duplicated = [
|
|
sample_days()[0],
|
|
{
|
|
"plan_date": "2026-08-22",
|
|
"ids": ["issue:secret/repo:3:"],
|
|
"capacity_minutes": 60,
|
|
"estimates": {"issue:secret/repo:3:": 30},
|
|
},
|
|
]
|
|
|
|
with pytest.raises(ValueError, match="work must be assigned to only one Week Ahead date"):
|
|
store.replace_week(
|
|
"timmy", base_revision=original["revision"], days=duplicated, timezone="UTC"
|
|
)
|
|
|
|
assert store.get_week("timmy") == original
|
|
|
|
|
|
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
|
|
|
|
|
|
def test_week_reconciliation_atomically_combines_today_and_consumes_only_due_day(tmp_path):
|
|
store = TodayStore(
|
|
tmp_path / "today.sqlite3", encryption_key=b"c" * 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")
|
|
|
|
result = store.reconcile_week(
|
|
"timmy", promotion_id="reconcile-2026-08-21-r1", week_revision=week["revision"],
|
|
plan_date="2026-08-21", today_revision=today["revision"],
|
|
ids=["issue:r:active:", "issue:secret/repo:3:"], capacity_minutes=120,
|
|
estimates={"issue:r:active:": 30, "issue:secret/repo:3:": 60},
|
|
)
|
|
replay = store.reconcile_week(
|
|
"timmy", promotion_id="reconcile-2026-08-21-r1", week_revision=week["revision"],
|
|
plan_date="2026-08-21", today_revision=today["revision"],
|
|
ids=["issue:r:active:", "issue:secret/repo:3:"], capacity_minutes=120,
|
|
estimates={"issue:r:active:": 30, "issue:secret/repo:3:": 60},
|
|
)
|
|
|
|
assert replay == result
|
|
assert result["today"]["ids"] == ["issue:r:active:", "issue:secret/repo:3:"]
|
|
assert result["week"]["days"] == [sample_days()[1]]
|
|
assert store.get("timmy") == result["today"]
|
|
assert store.get_week("timmy") == result["week"]
|
|
|
|
|
|
def test_week_reconciliation_rejects_work_outside_preserved_plans_without_partial_write(tmp_path):
|
|
store = TodayStore(
|
|
tmp_path / "today.sqlite3", encryption_key=b"c" * 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(ValueError, match="selected work must come from Today or the due Week Ahead day"):
|
|
store.reconcile_week(
|
|
"timmy", promotion_id="bad", week_revision=week["revision"],
|
|
plan_date="2026-08-21", today_revision=today["revision"],
|
|
ids=["issue:other/repo:99:"], capacity_minutes=60,
|
|
estimates={"issue:other/repo:99:": 30},
|
|
)
|
|
|
|
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,
|
|
}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_week_reconciliation_api_returns_atomic_today_and_remaining_week(monkeypatch, tmp_path):
|
|
async def user():
|
|
return {"login": "timmy"}
|
|
|
|
store = TodayStore(
|
|
tmp_path / "today.sqlite3", encryption_key=b"a" * 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)
|
|
|
|
result = await main.reconcile_week_plan(main.WeekReconciliation(
|
|
promotion_id="reconcile-2026-08-21-r1", week_revision=week["revision"],
|
|
plan_date="2026-08-21", today_revision=today["revision"],
|
|
ids=["issue:r:active:", "issue:secret/repo:3:"], capacity_minutes=120,
|
|
estimates={"issue:r:active:": 30, "issue:secret/repo:3:": 60},
|
|
))
|
|
|
|
assert result["today"]["ids"] == ["issue:r:active:", "issue:secret/repo:3:"]
|
|
assert result["week"]["days"] == [sample_days()[1]]
|
|
|
|
|
|
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}
|