598 lines
24 KiB
Python
598 lines
24 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_early_start_moves_a_future_day_to_empty_today_exactly_once(tmp_path):
|
|
store = TodayStore(
|
|
tmp_path / "today.sqlite3",
|
|
encryption_key=b"e" * 32,
|
|
clock=lambda: datetime(2026, 8, 20, 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="early-2026-08-21-r1", week_revision=week["revision"],
|
|
plan_date="2026-08-21", today_revision=today["revision"], allow_future=True,
|
|
)
|
|
replay = store.promote_week(
|
|
"timmy", promotion_id="early-2026-08-21-r1", week_revision=week["revision"],
|
|
plan_date="2026-08-21", today_revision=today["revision"], allow_future=True,
|
|
)
|
|
|
|
assert replay == promoted
|
|
assert promoted["ids"] == sample_days()[0]["ids"]
|
|
assert promoted["plan_date"] == "2026-08-21"
|
|
assert store.get_week("timmy")["days"] == [sample_days()[1]]
|
|
|
|
|
|
def test_pull_week_item_atomically_appends_to_active_today_and_replays_exactly_once(tmp_path):
|
|
store = TodayStore(tmp_path / "today.sqlite3", encryption_key=b"i" * 32)
|
|
active_id = "issue:stackchain/dashboard:1:"
|
|
pulled_id = "issue:secret/repo:3:"
|
|
today = store.apply("timmy", "seed-active", "add", active_id)
|
|
week = store.replace_week("timmy", base_revision=0, days=sample_days(), timezone="UTC")
|
|
|
|
result = store.pull_week_item(
|
|
"timmy", operation_id="pull-secret-3", identity=pulled_id,
|
|
today_revision=today["revision"], week_revision=week["revision"],
|
|
)
|
|
replay = store.pull_week_item(
|
|
"timmy", operation_id="pull-secret-3", identity=pulled_id,
|
|
today_revision=today["revision"], week_revision=week["revision"],
|
|
)
|
|
|
|
assert replay == result
|
|
assert result["today"]["ids"] == [active_id, pulled_id]
|
|
assert result["today"]["estimates"] == {pulled_id: 60}
|
|
assert result["week"]["days"][0]["ids"] == ["issue:secret/repo:2:"]
|
|
assert result["week"]["days"][0]["estimates"] == {"issue:secret/repo:2:": 45}
|
|
assert result["week"]["days"][1] == sample_days()[1]
|
|
assert store.get("timmy") == result["today"]
|
|
assert store.get_week("timmy") == result["week"]
|
|
|
|
|
|
def test_pull_week_item_rejects_full_or_changed_today_without_partial_write(tmp_path):
|
|
store = TodayStore(tmp_path / "today.sqlite3", encryption_key=b"f" * 32)
|
|
today = store.get("timmy")
|
|
for index in range(5):
|
|
today = store.apply("timmy", f"seed-{index}", "add", f"issue:r:{index + 1}:")
|
|
week = store.replace_week("timmy", base_revision=0, days=sample_days(), timezone="UTC")
|
|
|
|
with pytest.raises(main.TodayPlanFull):
|
|
store.pull_week_item(
|
|
"timmy", operation_id="full", identity="issue:secret/repo:3:",
|
|
today_revision=today["revision"], week_revision=week["revision"],
|
|
)
|
|
with pytest.raises(TodayPromotionConflict):
|
|
store.pull_week_item(
|
|
"timmy", operation_id="stale", identity="issue:secret/repo:3:",
|
|
today_revision=today["revision"] - 1, week_revision=week["revision"],
|
|
)
|
|
|
|
assert store.get("timmy") == today
|
|
assert store.get_week("timmy") == week
|
|
|
|
|
|
def test_normal_week_promotion_still_rejects_a_future_day_without_writes(tmp_path):
|
|
store = TodayStore(
|
|
tmp_path / "today.sqlite3",
|
|
encryption_key=b"e" * 32,
|
|
clock=lambda: datetime(2026, 8, 20, 8, tzinfo=timezone.utc).timestamp(),
|
|
)
|
|
today = store.get("timmy")
|
|
week = store.replace_week("timmy", base_revision=0, days=sample_days(), timezone="UTC")
|
|
|
|
with pytest.raises(main.TomorrowPlanNotDue):
|
|
store.promote_week(
|
|
"timmy", promotion_id="normal-2026-08-21-r1", week_revision=week["revision"],
|
|
plan_date="2026-08-21", today_revision=today["revision"],
|
|
)
|
|
|
|
assert store.get("timmy") == today
|
|
assert store.get_week("timmy") == week
|
|
|
|
|
|
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_early_start_api_promotes_a_future_day(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, 20, 8, tzinfo=timezone.utc).timestamp(),
|
|
)
|
|
today = store.get("timmy")
|
|
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.start_week_day_early(main.WeekPromotion(
|
|
promotion_id="early-2026-08-21-r1", week_revision=week["revision"],
|
|
plan_date="2026-08-21", today_revision=today["revision"],
|
|
))
|
|
|
|
assert result["ids"] == sample_days()[0]["ids"]
|
|
assert store.get_week("timmy")["days"] == [sample_days()[1]]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_pull_week_item_api_returns_both_atomic_plans(monkeypatch, tmp_path):
|
|
async def user():
|
|
return {"login": "timmy"}
|
|
|
|
store = TodayStore(tmp_path / "today.sqlite3", encryption_key=b"u" * 32)
|
|
today = store.apply("timmy", "seed", "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.pull_week_item_into_today(main.WeekItemPull(
|
|
operation_id="pull-api", identity="issue:secret/repo:3:",
|
|
week_revision=week["revision"], today_revision=today["revision"],
|
|
))
|
|
|
|
assert result["today"]["ids"] == ["issue:r:active:", "issue:secret/repo:3:"]
|
|
assert result["week"]["days"][0]["ids"] == ["issue:secret/repo:2:"]
|
|
|
|
|
|
@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}
|
|
|
|
|
|
def test_reschedule_today_item_atomically_moves_it_into_selected_week_day(tmp_path):
|
|
store = TodayStore(tmp_path / "today.sqlite3", encryption_key=b"s" * 32)
|
|
active_id = "issue:stackchain/dashboard:1228:"
|
|
other_id = "issue:stackchain/dashboard:1229:"
|
|
today = store.apply("timmy", "add-active", "add", active_id)
|
|
today = store.apply("timmy", "add-other", "add", other_id)
|
|
week = store.replace_week(
|
|
"timmy",
|
|
base_revision=0,
|
|
timezone="UTC",
|
|
days=[{
|
|
"plan_date": "2026-08-24",
|
|
"ids": [active_id],
|
|
"capacity_minutes": 120,
|
|
"estimates": {active_id: 30},
|
|
}, {
|
|
"plan_date": "2026-08-25",
|
|
"ids": [],
|
|
"capacity_minutes": 90,
|
|
"estimates": {},
|
|
}],
|
|
)
|
|
|
|
result = store.reschedule_today_to_week(
|
|
"timmy",
|
|
operation_id="reschedule-active-2026-08-25",
|
|
identity=active_id,
|
|
estimate_minutes=45,
|
|
plan_date="2026-08-25",
|
|
today_revision=today["revision"],
|
|
week_revision=week["revision"],
|
|
)
|
|
|
|
assert result["today"]["ids"] == [other_id]
|
|
assert result["week"]["days"] == [{
|
|
"plan_date": "2026-08-24", "ids": [], "capacity_minutes": 120, "estimates": {},
|
|
}, {
|
|
"plan_date": "2026-08-25", "ids": [active_id], "capacity_minutes": 90,
|
|
"estimates": {active_id: 45},
|
|
}]
|
|
assert store.get("timmy") == result["today"]
|
|
assert store.get_week("timmy") == result["week"]
|
|
|
|
|
|
def test_reschedule_today_item_is_idempotent_and_rejects_stale_plans_without_partial_write(tmp_path):
|
|
store = TodayStore(tmp_path / "today.sqlite3", encryption_key=b"i" * 32)
|
|
identity = "issue:stackchain/dashboard:1228:"
|
|
today = store.apply("timmy", "add-active", "add", identity)
|
|
week = store.replace_week(
|
|
"timmy", base_revision=0, timezone="UTC",
|
|
days=[{
|
|
"plan_date": "2026-08-25", "ids": [],
|
|
"capacity_minutes": 60, "estimates": {},
|
|
}],
|
|
)
|
|
arguments = {
|
|
"operation_id": "same-operation", "identity": identity,
|
|
"estimate_minutes": 30, "plan_date": "2026-08-25",
|
|
"today_revision": today["revision"], "week_revision": week["revision"],
|
|
}
|
|
|
|
moved = store.reschedule_today_to_week("timmy", **arguments)
|
|
assert store.reschedule_today_to_week("timmy", **arguments) == moved
|
|
|
|
another = store.apply("timmy", "add-another", "add", "issue:stackchain/dashboard:1230:")
|
|
with pytest.raises(TodayPromotionConflict) as conflict:
|
|
store.reschedule_today_to_week(
|
|
"timmy", **{**arguments, "operation_id": "stale-operation", "identity": another["ids"][-1]}
|
|
)
|
|
assert conflict.value.today == another
|
|
assert store.get_week("timmy") == moved["week"]
|
|
|
|
|
|
def test_reschedule_today_item_requires_explicit_capacity_overload_confirmation(tmp_path):
|
|
store = TodayStore(tmp_path / "today.sqlite3", encryption_key=b"o" * 32)
|
|
identity = "issue:stackchain/dashboard:1228:"
|
|
today = store.apply("timmy", "add-active", "add", identity)
|
|
week = store.replace_week(
|
|
"timmy", base_revision=0, timezone="UTC",
|
|
days=[{
|
|
"plan_date": "2026-08-25", "ids": ["issue:stackchain/dashboard:1:"],
|
|
"capacity_minutes": 60, "estimates": {"issue:stackchain/dashboard:1:": 45},
|
|
}],
|
|
)
|
|
arguments = {
|
|
"identity": identity, "estimate_minutes": 30, "plan_date": "2026-08-25",
|
|
"today_revision": today["revision"], "week_revision": week["revision"],
|
|
}
|
|
|
|
with pytest.raises(ValueError, match="explicit overload confirmation"):
|
|
store.reschedule_today_to_week(
|
|
"timmy", operation_id="unconfirmed", allow_over_capacity=False, **arguments
|
|
)
|
|
assert store.get("timmy") == today
|
|
assert store.get_week("timmy") == week
|
|
|
|
moved = store.reschedule_today_to_week(
|
|
"timmy", operation_id="confirmed", allow_over_capacity=True, **arguments
|
|
)
|
|
assert moved["week"]["days"][0]["ids"][-1] == identity
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_week_reschedule_api_returns_atomic_snapshots_and_conflict_recovery(monkeypatch, tmp_path):
|
|
async def user():
|
|
return {"login": "timmy"}
|
|
|
|
store = TodayStore(tmp_path / "today.sqlite3", encryption_key=b"a" * 32)
|
|
identity = "issue:stackchain/dashboard:1228:"
|
|
today = store.apply("timmy", "add-active", "add", identity)
|
|
week = store.replace_week(
|
|
"timmy", base_revision=0, timezone="UTC",
|
|
days=[{
|
|
"plan_date": "2026-08-25", "ids": [],
|
|
"capacity_minutes": 60, "estimates": {},
|
|
}],
|
|
)
|
|
monkeypatch.setattr(main, "current_user", user)
|
|
monkeypatch.setattr(main, "_today_store", lambda: store)
|
|
payload = main.WeekReschedule(
|
|
operation_id="move-active", identity=identity, estimate_minutes=30,
|
|
plan_date="2026-08-25", today_revision=today["revision"],
|
|
week_revision=week["revision"],
|
|
)
|
|
|
|
moved = await main.reschedule_today_to_week(payload)
|
|
assert moved["today"]["ids"] == []
|
|
assert moved["week"]["days"][0]["ids"] == [identity]
|
|
|
|
stale = main.WeekReschedule(
|
|
operation_id="stale", identity="issue:stackchain/dashboard:1229:", estimate_minutes=30,
|
|
plan_date="2026-08-25", today_revision=today["revision"], week_revision=week["revision"],
|
|
)
|
|
with pytest.raises(main.HTTPException) as raised:
|
|
await main.reschedule_today_to_week(stale)
|
|
assert raised.value.status_code == 409
|
|
assert raised.value.detail["code"] == "today_changed"
|
|
assert raised.value.detail["today"] == moved["today"]
|
|
assert raised.value.detail["week"] == moved["week"]
|