"""Tests for `PlanningService.sync_plan` (POST /api/plan/sync). Covers the Task 4 contract: propose+apply in one step, a pinned `planned_leave_day` is never retyped/overwritten by a re-plan even when the fresh proposal would have chosen differently for that date, the `changes` diff surfaces each of its four kinds, sync is idempotent on an unchanged plan, and `/propose` stays read-only (persists nothing). """ from datetime import date, datetime from uuid import uuid4 from app.entities.leave_wish import LeaveWish from app.entities.parental_leave import ParentalLeaveFetch, TierData from app.entities.personal_data import PersonalData from app.entities.planned_leave import PlannedLeaveDay from app.entities.vacation_days import VacationDays from app.repositories_impl.sqlite.leave_wish_repository import SQLiteLeaveWishRepository from app.repositories_impl.sqlite.parental_leave_repository import SQLiteParentalLeaveRepository from app.repositories_impl.sqlite.personal_data_repository import SQLitePersonalDataRepository from app.repositories_impl.sqlite.planned_leave_repository import SQLitePlannedLeaveRepository from app.repositories_impl.sqlite.vacation_days_repository import SQLiteVacationDaysRepository from app.services.allocation_service import PlanningEngineService from app.services.planning_service import ( EVICTION_CHOICE_DROP_FUTURE_PL, EVICTION_CHOICE_DROP_LATEST_VACATION_THIS_YEAR, PlanningService, ) from tests.conftest import SeededIds def _make_wish(workspace_id, child_id, **overrides) -> LeaveWish: base = dict( id=uuid4(), workspace_id=workspace_id, child_id=child_id, name="Test wish", window_start_month=2, window_start_day=1, window_end_month=2, window_end_day=7, target_weeks=1, recurrence="once", first_year=2027, last_year=None, priority=1, parent_mode="either", parent_personal_data_uuid=None, window_mode="fixed", created_at=datetime.now(), updated_at=datetime.now(), ) base.update(overrides) return LeaveWish(**base) def _make_fetch(personal_data_uuid, workspace_id, child_id, fetched_at, sjuk_left, lagsta_left): tier = TierData( sjukpenning_total=390, sjukpenning_used=390 - sjuk_left, sjukpenning_left=sjuk_left, lagsta_niva_total=90, lagsta_niva_used=90 - lagsta_left, lagsta_niva_left=lagsta_left, ) return ParentalLeaveFetch( id=uuid4(), personal_data_uuid=personal_data_uuid, workspace_id=workspace_id, child_id=child_id, source="manual", fetched_at=fetched_at, total_days=480, days_used=480 - sjuk_left - lagsta_left, days_left=sjuk_left + lagsta_left, reserved_days_total=90, reserved_days_left=90, transferable_days_left=sjuk_left + lagsta_left, dubbeldagar_left=60, tier_data=tier, created_at=fetched_at, updated_at=fetched_at, ) class TestSyncNeverOverwritesAPinnedDay: async def test_pinned_day_survives_even_though_fresh_plan_would_differ( self, planning_service: PlanningService, planned_leave_repo: SQLitePlannedLeaveRepository, vacation_repo: SQLiteVacationDaysRepository, leave_wish_repo: SQLiteLeaveWishRepository, seeded_ids: SeededIds, ) -> None: ws, child, parent = ( seeded_ids.workspace_id, seeded_ids.child_id, seeded_ids.personal_data_uuid, ) # No vacation budget this year -> the monthly-salaried parent's # tier-3 policy (vacation first) falls through to FP, sjukpenning # preferred -- so a free interior day in this window would be # funded as 'sjukpenning', never 'lagsta'. await vacation_repo.create( VacationDays( id=uuid4(), personal_data_uuid=parent, workspace_id=ws, year=2027, total_days=0 ) ) await leave_wish_repo.create(_make_wish(ws, child)) pinned_date = date(2027, 2, 3) # a Wednesday inside the wish's window pinned = await planned_leave_repo.create( PlannedLeaveDay( id=uuid4(), workspace_id=ws, child_id=child, parent_personal_data_uuid=parent, date=pinned_date, day_type="lagsta", # deliberately NOT what a fresh plan would pick is_pinned=True, created_at=datetime.now(), updated_at=datetime.now(), ) ) result = await planning_service.sync_plan(ws, today=date(2027, 1, 1)) assert result.apply_result.accepted is True rows = await planned_leave_repo.list_by_workspace_child(ws, child) pinned_after = next(r for r in rows if r.date == pinned_date) assert pinned_after.id == pinned.id assert pinned_after.day_type == "lagsta" assert pinned_after.is_pinned is True # Every OTHER day in the window was free to be funded normally. other_days = [r for r in rows if r.date != pinned_date] assert other_days, "expected the rest of the window to be funded" assert all(r.day_type == "sjukpenning" for r in other_days) # The pinned date never shows up as a changed slot. assert all(c.date != pinned_date for c in result.changes) class TestSyncChangeKinds: async def test_added_then_removed_when_the_wish_shrinks( self, planning_service: PlanningService, leave_wish_repo: SQLiteLeaveWishRepository, seeded_ids: SeededIds, ) -> None: """A wish that still exists but now claims fewer dates: the dates it drops (still `wish_id`-attributed, just no longer produced by this run's batch) must be reported (and written) as 'removed' -- distinct from a deleted wish, whose rows the DB itself detaches via `ON DELETE SET NULL` (making them ad-hoc, permanently kept) before `sync_plan` ever sees them. """ ws, child, _parent = ( seeded_ids.workspace_id, seeded_ids.child_id, seeded_ids.personal_data_uuid, ) wish = await leave_wish_repo.create(_make_wish(ws, child, name="A", window_end_day=7)) # -- first sync: pure 'added' (Feb 1-7) --------------------------- result_1 = await planning_service.sync_plan(ws, today=date(2027, 1, 1)) assert result_1.apply_result.accepted is True assert result_1.changes, "expected at least one 'added' change" assert {c.kind for c in result_1.changes} == {"added"} # -- idempotent re-sync: no changes ------------------------------- result_2 = await planning_service.sync_plan(ws, today=date(2027, 1, 1)) assert result_2.apply_result.accepted is True assert result_2.changes == [] # -- shrink the window (Feb 1-4) -> the dropped tail is 'removed' - wish.window_end_day = 4 await leave_wish_repo.update(wish.id, wish) result_3 = await planning_service.sync_plan(ws, today=date(2027, 1, 1)) assert result_3.apply_result.accepted is True assert result_3.changes, "expected 'removed' changes once the window shrinks" assert {c.kind for c in result_3.changes} == {"removed"} assert all(c.date > date(2027, 2, 4) for c in result_3.changes) async def test_retyped_when_the_underlying_pool_changes( self, planning_service: PlanningService, personal_data_repo: SQLitePersonalDataRepository, parental_leave_repo: SQLiteParentalLeaveRepository, leave_wish_repo: SQLiteLeaveWishRepository, seeded_ids: SeededIds, ) -> None: ws, child = seeded_ids.workspace_id, seeded_ids.child_id # An hourly-salaried parent prefers FP (sjukpenning first) over # vacation for tier-3 interior days, so this is deterministic # regardless of any assumed vacation default. parent = await personal_data_repo.create( PersonalData(uuid=uuid4(), workspace_id=ws, name="Hourly Parent", salary_type="hourly") ) # `parent_mode="specific"` -- with a second (seeded) parent now also # in the workspace, an "either" wish would default to whichever # parent sorts first, not necessarily this one. await leave_wish_repo.create( _make_wish(ws, child, parent_mode="specific", parent_personal_data_uuid=parent.uuid) ) await parental_leave_repo.create( _make_fetch( parent.uuid, ws, child, datetime(2027, 1, 1), sjuk_left=100, lagsta_left=100 ) ) result_1 = await planning_service.sync_plan(ws, today=date(2027, 1, 1)) assert result_1.apply_result.accepted is True assert any(c.new_day_type == "sjukpenning" for c in result_1.changes) # sjukpenning now exhausted -> the SAME dates must fall back to lagsta. await parental_leave_repo.create( _make_fetch(parent.uuid, ws, child, datetime(2027, 1, 2), sjuk_left=0, lagsta_left=100) ) result_2 = await planning_service.sync_plan(ws, today=date(2027, 1, 1)) assert result_2.apply_result.accepted is True assert result_2.changes, "expected 'retyped' changes once sjukpenning ran out" assert {c.kind for c in result_2.changes} == {"retyped"} assert all( c.old_day_type == "sjukpenning" and c.new_day_type == "lagsta" for c in result_2.changes ) async def test_reattributed_when_a_day_moves_to_a_different_wish( self, planning_service: PlanningService, planned_leave_repo: SQLitePlannedLeaveRepository, vacation_repo: SQLiteVacationDaysRepository, leave_wish_repo: SQLiteLeaveWishRepository, seeded_ids: SeededIds, ) -> None: ws, child, parent = ( seeded_ids.workspace_id, seeded_ids.child_id, seeded_ids.personal_data_uuid, ) await vacation_repo.create( VacationDays( id=uuid4(), personal_data_uuid=parent, workspace_id=ws, year=2027, total_days=0 ) ) # `old_wish`'s own window doesn't cover `target_date` at all, so it # never competes for that key -- it just needs to be a real, still- # existing wish so the persisted row's `wish_id` stays non-null # (unlike deleting it, which `ON DELETE SET NULL` would turn into an # ad-hoc, permanently-kept row before `sync_plan` ever sees it). old_wish = await leave_wish_repo.create( _make_wish( ws, child, name="Old", window_start_month=3, window_start_day=1, window_end_month=3, window_end_day=7, ) ) new_wish = await leave_wish_repo.create( _make_wish( ws, child, name="New", window_start_month=2, window_start_day=3, window_end_month=2, window_end_day=3, target_days=1, ) ) target_date = date(2027, 2, 3) # A row identical to what `new_wish` would compute for this date # (same date/type), but still attributed to `old_wish`. await planned_leave_repo.create( PlannedLeaveDay( id=uuid4(), workspace_id=ws, child_id=child, parent_personal_data_uuid=parent, date=target_date, day_type="sjukpenning", wish_id=old_wish.id, is_pinned=False, created_at=datetime.now(), updated_at=datetime.now(), ) ) result = await planning_service.sync_plan(ws, today=date(2027, 1, 1)) assert result.apply_result.accepted is True by_date = {c.date: c for c in result.changes} assert target_date in by_date changed = by_date[target_date] assert changed.kind == "reattributed" assert changed.old_wish_id == old_wish.id assert changed.new_wish_id == new_wish.id assert changed.old_day_type == changed.new_day_type == "sjukpenning" class TestSyncIdempotent: async def test_unchanged_plan_produces_no_changes( self, planning_service: PlanningService, leave_wish_repo: SQLiteLeaveWishRepository, seeded_ids: SeededIds, ) -> None: ws, child, _parent = ( seeded_ids.workspace_id, seeded_ids.child_id, seeded_ids.personal_data_uuid, ) await leave_wish_repo.create(_make_wish(ws, child)) first = await planning_service.sync_plan(ws, today=date(2027, 1, 1)) assert first.apply_result.accepted is True second = await planning_service.sync_plan(ws, today=date(2027, 1, 1)) assert second.apply_result.accepted is True assert second.changes == [] class TestSyncCreditsFreedCapacityWithinSameCall: """Regression for the bug this change fixes: `sync_plan` used to delete stale (shrunk-away) rows only AFTER `apply_proposed_days` had already validated and written the fresh batch, so a wish shrinking never credited the capacity it freed to a same-call growing wish -- a family near their sjukpenning cap who shortened one window and lengthened another would have the whole sync rejected with `insufficientBalance` even though the freed days more than covered the new demand. """ async def test_shrinking_one_window_frees_capacity_for_a_new_one_in_the_same_sync( self, planning_service: PlanningService, planned_leave_repo: SQLitePlannedLeaveRepository, vacation_repo: SQLiteVacationDaysRepository, parental_leave_repo: SQLiteParentalLeaveRepository, leave_wish_repo: SQLiteLeaveWishRepository, seeded_ids: SeededIds, ) -> None: ws, child, parent = ( seeded_ids.workspace_id, seeded_ids.child_id, seeded_ids.personal_data_uuid, ) # No vacation budget -> forces sjukpenning/lagsta funding, so this # test only has to reason about the sjukpenning pool. await vacation_repo.create( VacationDays( id=uuid4(), personal_data_uuid=parent, workspace_id=ws, year=2027, total_days=0 ) ) await parental_leave_repo.create( _make_fetch(parent, ws, child, datetime(2027, 1, 1), sjuk_left=390, lagsta_left=90) ) summer = await leave_wish_repo.create( _make_wish( ws, child, name="Summer", window_start_month=6, window_start_day=1, window_end_month=6, window_end_day=30, target_weeks=4, ) ) first = await planning_service.sync_plan(ws, today=date(2027, 1, 1)) assert first.apply_result.accepted is True summer_rows = await planned_leave_repo.list_by_workspace_child( ws, child, from_date=date(2027, 6, 1), to_date=date(2027, 6, 30) ) sjuk_used = sum(1 for r in summer_rows if r.day_type == "sjukpenning") assert sjuk_used > 10, "expected the 4-week summer window to fund several sjukpenning days" # Pin the balance to exactly what's already spent -> zero slack left, # simulating a family sitting right at their sjukpenning cap. await parental_leave_repo.create( _make_fetch( parent, ws, child, datetime(2027, 1, 2), sjuk_left=sjuk_used, lagsta_left=90 ) ) # Shrink summer to one week and add a smaller winter wish elsewhere. # The capacity freed by shrinking summer comfortably covers winter's # demand, but only if the freed days are credited within this call. summer.window_end_day = 7 summer.target_weeks = 1 await leave_wish_repo.update(summer.id, summer) await leave_wish_repo.create( _make_wish( ws, child, name="Winter", window_start_month=12, window_start_day=1, window_end_month=12, window_end_day=7, target_weeks=1, ) ) second = await planning_service.sync_plan(ws, today=date(2027, 1, 1)) assert second.apply_result.accepted is True, second.apply_result.rejections assert not any( r.message_key == "planApply.rejection.insufficientBalance" for r in second.apply_result.rejections ) winter_rows = await planned_leave_repo.list_by_workspace_child( ws, child, from_date=date(2027, 12, 1), to_date=date(2027, 12, 7) ) assert any(r.day_type == "sjukpenning" for r in winter_rows) class TestSyncEvictionPreview: """`sync_plan`'s eviction preview/resolution — the fix for the data-loss bug where editing one wish silently deleted another wish's rows because the shared pool went to the edited wish first (see this file's module docstring / PLANNING_ENGINE_SPEC.md). Both scenarios below use two wishes over the EXACT SAME window (Mon 11 - Fri 15 Jan 2027 -- five plain workdays, no weekend, no Swedish red day in range) so a higher-priority wish's fresh placement claims literally the same dates the lower-priority wish already had persisted, guaranteeing full contention regardless of the exact day-count the allocator computes. """ async def test_would_evict_returns_preview_and_mutates_nothing( self, planning_service: PlanningService, planned_leave_repo: SQLitePlannedLeaveRepository, vacation_repo: SQLiteVacationDaysRepository, parental_leave_repo: SQLiteParentalLeaveRepository, leave_wish_repo: SQLiteLeaveWishRepository, seeded_ids: SeededIds, ) -> None: ws, child, parent = ( seeded_ids.workspace_id, seeded_ids.child_id, seeded_ids.personal_data_uuid, ) # No vacation this year, no lagsta -> the only lever is sjukpenning, # so a shortfall there can never be silently backstopped by anything # else (it must show up as a genuine shortfall/eviction). await vacation_repo.create( VacationDays( id=uuid4(), personal_data_uuid=parent, workspace_id=ws, year=2027, total_days=0 ) ) await parental_leave_repo.create( _make_fetch(parent, ws, child, datetime(2027, 1, 1), sjuk_left=390, lagsta_left=0) ) existing = await leave_wish_repo.create( _make_wish( ws, child, name="Existing", window_start_month=1, window_start_day=11, window_end_month=1, window_end_day=15, priority=5, ) ) first = await planning_service.sync_plan(ws, today=date(2027, 1, 1)) assert first.apply_result.accepted is True rows_after_first = await planned_leave_repo.list_by_workspace_child(ws, child) existing_count = len(rows_after_first) assert existing_count > 0 assert all(r.wish_id == existing.id for r in rows_after_first) assert first.eviction_preview.would_evict is False # Shrink the shared pool to exactly what `existing` already spent (no # slack), then a higher-priority wish over the SAME window steals it. await parental_leave_repo.create( _make_fetch( parent, ws, child, datetime(2027, 1, 2), sjuk_left=existing_count, lagsta_left=0 ) ) await leave_wish_repo.create( _make_wish( ws, child, name="New", window_start_month=2, window_start_day=1, window_end_month=2, window_end_day=5, priority=1, ) ) second = await planning_service.sync_plan(ws, today=date(2027, 1, 1)) # Nothing accepted, nothing changed, and the caller must be told why. assert second.apply_result.accepted is False assert second.changes == [] assert second.apply_result.rejections assert ( second.apply_result.rejections[0].message_key == "planSync.rejection.evictionChoiceRequired" ) preview = second.eviction_preview assert preview.would_evict is True assert preview.candidates, "expected the evicted 'Existing' days to be listed" assert all(c.wish_id == existing.id for c in preview.candidates) assert all(c.wish_name == "Existing" for c in preview.candidates) assert all(c.day_type == "sjukpenning" for c in preview.candidates) assert all( c.child_id == child and c.parent_personal_data_uuid == parent for c in preview.candidates ) assert len(preview.shortages) == 1 shortage = preview.shortages[0] assert shortage.resource_kind == "parental_leave" assert shortage.parent_personal_data_uuid == parent assert shortage.tier == "sjukpenning" assert shortage.child_id == child assert shortage.year is None assert shortage.days_evicted == len(preview.candidates) # Requirement: mutates NOTHING, not even the harmless parts. rows_after_second = await planned_leave_repo.list_by_workspace_child(ws, child) assert {r.id for r in rows_after_second} == {r.id for r in rows_after_first} async def test_shortfall_days_is_not_masked_when_a_higher_priority_wish_claims_the_same_dates( self, planning_service: PlanningService, planned_leave_repo: SQLitePlannedLeaveRepository, vacation_repo: SQLiteVacationDaysRepository, parental_leave_repo: SQLiteParentalLeaveRepository, leave_wish_repo: SQLiteLeaveWishRepository, seeded_ids: SeededIds, ) -> None: """Regression for a real bug this eviction-preview work uncovered in `_Ledger._fund_solo` (`app/services/allocation_service.py`): when a higher-priority wish's placement lands on the EXACT SAME calendar dates a lower-priority wish already occupied (as opposed to merely exhausting a shared pool on different dates — the scenario the test above covers), `_fund_solo` used to find those dates already in `_existing_lookup` (via `record_days()`, folded in by the earlier- processed wish this same run) and credit them toward the LATER wish's own `cost`/`shortfall_days` as if THAT wish had gotten them too — so a wish entirely crowded out on every one of its dates could still report `shortfall_days == 0`. Fixed by only crediting cost for dates present in `_existing_lookup` BEFORE this run's own `record_days()` calls started (`_pre_run_occupied`); a same-run claim by a different wish now correctly falls through unfunded. NOTE — known, separately-scoped gap this test does NOT claim to fix: because the two wishes here land on the identical (child, parent, date) keys, `sync_plan`'s `stale_keys`/`_genuinely_evicted_rows` (which only look at keys ABSENT from the fresh batch) never fire — the row survives and is silently metadata-updated to the new wish's `wish_id` (a 'reattributed' `PlanChange`, not a deletion), bypassing `eviction_preview` entirely. That is a real, distinct hole in the eviction-preview feature (same-date collisions between two DIFFERENT wishes are not gated the way stale-key drops are) — flagged in this session's handoff report as follow-up work, not fixed here. This test only pins down that `shortfall_days`/`weeks_funded` — the signal both the UI's "window funded" display AND `_genuinely_evicted_rows` depend on — are no longer silently wrong. """ ws, child, parent = ( seeded_ids.workspace_id, seeded_ids.child_id, seeded_ids.personal_data_uuid, ) await vacation_repo.create( VacationDays( id=uuid4(), personal_data_uuid=parent, workspace_id=ws, year=2027, total_days=0 ) ) await parental_leave_repo.create( _make_fetch(parent, ws, child, datetime(2027, 1, 1), sjuk_left=390, lagsta_left=0) ) existing = await leave_wish_repo.create( _make_wish( ws, child, name="Existing", window_start_month=1, window_start_day=11, window_end_month=1, window_end_day=15, priority=5, ) ) first = await planning_service.sync_plan(ws, today=date(2027, 1, 1)) assert first.apply_result.accepted is True existing_count = len(await planned_leave_repo.list_by_workspace_child(ws, child)) await parental_leave_repo.create( _make_fetch( parent, ws, child, datetime(2027, 1, 2), sjuk_left=existing_count, lagsta_left=0 ) ) # SAME window as `existing` -- the higher-priority wish's placement # lands on the literal same calendar dates, not merely a shared pool. await leave_wish_repo.create( _make_wish( ws, child, name="New", window_start_month=1, window_start_day=11, window_end_month=1, window_end_day=15, priority=1, ) ) second = await planning_service.sync_plan(ws, today=date(2027, 1, 1)) existing_outcome = next(o for o in second.proposal.outcomes if o.wish_id == existing.id) assert existing_outcome.shortfall_days == existing_count, ( "the crowded-out wish's own outcome must report the real shortfall, " "not silently claim the other wish's days as its own funded total" ) assert existing_outcome.weeks_funded == 0.0 async def test_drop_future_pl_choice_frees_the_days_and_lets_the_edit_land( self, planning_service: PlanningService, planned_leave_repo: SQLitePlannedLeaveRepository, vacation_repo: SQLiteVacationDaysRepository, parental_leave_repo: SQLiteParentalLeaveRepository, leave_wish_repo: SQLiteLeaveWishRepository, seeded_ids: SeededIds, ) -> None: ws, child, parent = ( seeded_ids.workspace_id, seeded_ids.child_id, seeded_ids.personal_data_uuid, ) await vacation_repo.create( VacationDays( id=uuid4(), personal_data_uuid=parent, workspace_id=ws, year=2027, total_days=0 ) ) await parental_leave_repo.create( _make_fetch(parent, ws, child, datetime(2027, 1, 1), sjuk_left=390, lagsta_left=0) ) existing = await leave_wish_repo.create( _make_wish( ws, child, name="Existing", window_start_month=1, window_start_day=11, window_end_month=1, window_end_day=15, priority=5, ) ) first = await planning_service.sync_plan(ws, today=date(2027, 1, 1)) assert first.apply_result.accepted is True existing_count = len(await planned_leave_repo.list_by_workspace_child(ws, child)) await parental_leave_repo.create( _make_fetch( parent, ws, child, datetime(2027, 1, 2), sjuk_left=existing_count, lagsta_left=0 ) ) new_wish = await leave_wish_repo.create( _make_wish( ws, child, name="New", window_start_month=2, window_start_day=1, window_end_month=2, window_end_day=5, priority=1, ) ) preview = await planning_service.sync_plan(ws, today=date(2027, 1, 1)) assert preview.eviction_preview.would_evict is True resolved = await planning_service.sync_plan( ws, today=date(2027, 1, 1), eviction_choice=EVICTION_CHOICE_DROP_FUTURE_PL ) assert resolved.apply_result.accepted is True, resolved.apply_result.rejections rows_after = await planned_leave_repo.list_by_workspace_child(ws, child) assert rows_after, "expected the new wish's days to have been written" assert all(r.wish_id == new_wish.id for r in rows_after) assert not any(r.wish_id == existing.id for r in rows_after) async def test_wrong_choice_still_rejects_and_mutates_nothing( self, planning_service: PlanningService, planned_leave_repo: SQLitePlannedLeaveRepository, vacation_repo: SQLiteVacationDaysRepository, parental_leave_repo: SQLiteParentalLeaveRepository, leave_wish_repo: SQLiteLeaveWishRepository, seeded_ids: SeededIds, ) -> None: """The vacation-eviction option can't free a parental-leave shortage -- `apply_proposed_days`'s own balance validation rejects the whole batch exactly as it would with no choice at all, atomically. """ ws, child, parent = ( seeded_ids.workspace_id, seeded_ids.child_id, seeded_ids.personal_data_uuid, ) await vacation_repo.create( VacationDays( id=uuid4(), personal_data_uuid=parent, workspace_id=ws, year=2027, total_days=0 ) ) await parental_leave_repo.create( _make_fetch(parent, ws, child, datetime(2027, 1, 1), sjuk_left=390, lagsta_left=0) ) await leave_wish_repo.create( _make_wish( ws, child, name="Existing", window_start_month=1, window_start_day=11, window_end_month=1, window_end_day=15, priority=5, ) ) first = await planning_service.sync_plan(ws, today=date(2027, 1, 1)) assert first.apply_result.accepted is True rows_after_first = await planned_leave_repo.list_by_workspace_child(ws, child) existing_count = len(rows_after_first) await parental_leave_repo.create( _make_fetch( parent, ws, child, datetime(2027, 1, 2), sjuk_left=existing_count, lagsta_left=0 ) ) await leave_wish_repo.create( _make_wish( ws, child, name="New", window_start_month=2, window_start_day=1, window_end_month=2, window_end_day=5, priority=1, ) ) resolved = await planning_service.sync_plan( ws, today=date(2027, 1, 1), eviction_choice=EVICTION_CHOICE_DROP_LATEST_VACATION_THIS_YEAR, ) assert resolved.apply_result.accepted is False rows_after = await planned_leave_repo.list_by_workspace_child(ws, child) assert {r.id for r in rows_after} == {r.id for r in rows_after_first} async def test_drop_latest_vacation_choice_frees_vacation_days( self, planning_service: PlanningService, planned_leave_repo: SQLitePlannedLeaveRepository, vacation_repo: SQLiteVacationDaysRepository, parental_leave_repo: SQLiteParentalLeaveRepository, leave_wish_repo: SQLiteLeaveWishRepository, seeded_ids: SeededIds, ) -> None: ws, child, parent = ( seeded_ids.workspace_id, seeded_ids.child_id, seeded_ids.personal_data_uuid, ) # No FP at all -> the ONLY way either wish's window can be funded is # vacation, so a shortfall here can only ever be a vacation shortage. await parental_leave_repo.create( _make_fetch(parent, ws, child, datetime(2027, 1, 1), sjuk_left=0, lagsta_left=0) ) vacation_row = await vacation_repo.create( VacationDays( id=uuid4(), personal_data_uuid=parent, workspace_id=ws, year=2027, total_days=25 ) ) existing = await leave_wish_repo.create( _make_wish( ws, child, name="Existing", window_start_month=1, window_start_day=11, window_end_month=1, window_end_day=15, priority=5, ) ) first = await planning_service.sync_plan(ws, today=date(2027, 1, 1)) assert first.apply_result.accepted is True rows_after_first = await planned_leave_repo.list_by_workspace_child(ws, child) existing_count = len(rows_after_first) assert existing_count > 0 assert all(r.day_type == "vacation" for r in rows_after_first) # Shrink the shared vacation budget to exactly what `existing` spent. vacation_row.total_days = existing_count await vacation_repo.update(vacation_row.id, vacation_row) new_wish = await leave_wish_repo.create( _make_wish( ws, child, name="New", window_start_month=2, window_start_day=1, window_end_month=2, window_end_day=5, priority=1, ) ) preview = await planning_service.sync_plan(ws, today=date(2027, 1, 1)) assert preview.eviction_preview.would_evict is True assert all(c.day_type == "vacation" for c in preview.eviction_preview.candidates) assert len(preview.eviction_preview.shortages) == 1 shortage = preview.eviction_preview.shortages[0] assert shortage.resource_kind == "vacation" assert shortage.year == 2027 assert shortage.tier is None assert shortage.child_id is None # The wrong option (PL) can't touch a vacation shortage. wrong = await planning_service.sync_plan( ws, today=date(2027, 1, 1), eviction_choice=EVICTION_CHOICE_DROP_FUTURE_PL ) assert wrong.apply_result.accepted is False rows_still_original = await planned_leave_repo.list_by_workspace_child(ws, child) assert {r.id for r in rows_still_original} == {r.id for r in rows_after_first} resolved = await planning_service.sync_plan( ws, today=date(2027, 1, 1), eviction_choice=EVICTION_CHOICE_DROP_LATEST_VACATION_THIS_YEAR, ) assert resolved.apply_result.accepted is True, resolved.apply_result.rejections rows_after = await planned_leave_repo.list_by_workspace_child(ws, child) assert rows_after assert all(r.wish_id == new_wish.id for r in rows_after) assert not any(r.wish_id == existing.id for r in rows_after) async def test_pinned_day_never_evicted_under_any_choice( self, planning_service: PlanningService, planned_leave_repo: SQLitePlannedLeaveRepository, vacation_repo: SQLiteVacationDaysRepository, parental_leave_repo: SQLiteParentalLeaveRepository, leave_wish_repo: SQLiteLeaveWishRepository, seeded_ids: SeededIds, ) -> None: ws, child, parent = ( seeded_ids.workspace_id, seeded_ids.child_id, seeded_ids.personal_data_uuid, ) await vacation_repo.create( VacationDays( id=uuid4(), personal_data_uuid=parent, workspace_id=ws, year=2027, total_days=0 ) ) await parental_leave_repo.create( _make_fetch(parent, ws, child, datetime(2027, 1, 1), sjuk_left=390, lagsta_left=0) ) await leave_wish_repo.create( _make_wish( ws, child, name="Existing", window_start_month=1, window_start_day=11, window_end_month=1, window_end_day=15, priority=5, ) ) first = await planning_service.sync_plan(ws, today=date(2027, 1, 1)) assert first.apply_result.accepted is True rows = await planned_leave_repo.list_by_workspace_child(ws, child) assert rows to_pin = rows[0] pinned = await planning_service.update(to_pin.id, is_pinned=True) assert pinned.is_pinned is True await parental_leave_repo.create( _make_fetch(parent, ws, child, datetime(2027, 1, 2), sjuk_left=len(rows), lagsta_left=0) ) await leave_wish_repo.create( _make_wish( ws, child, name="New", window_start_month=2, window_start_day=1, window_end_month=2, window_end_day=5, priority=1, ) ) preview = await planning_service.sync_plan(ws, today=date(2027, 1, 1)) assert preview.eviction_preview.would_evict is True assert all(c.date != pinned.date for c in preview.eviction_preview.candidates) resolved = await planning_service.sync_plan( ws, today=date(2027, 1, 1), eviction_choice=EVICTION_CHOICE_DROP_FUTURE_PL ) assert resolved.apply_result.accepted is True, resolved.apply_result.rejections rows_after = await planned_leave_repo.list_by_workspace_child(ws, child) survivor = next((r for r in rows_after if r.id == pinned.id), None) assert survivor is not None, "the pinned day must never be evicted" assert survivor.is_pinned is True assert survivor.date == pinned.date assert survivor.day_type == pinned.day_type class TestSyncEvictionPreviewAbsentWhenWithinBudget: async def test_plain_sync_has_no_preview_friction( self, planning_service: PlanningService, leave_wish_repo: SQLiteLeaveWishRepository, seeded_ids: SeededIds, ) -> None: """Requirement 5 — a save that fits within budget applies straight through: no extra round trip, no empty-but-present friction either. """ ws, child, _parent = ( seeded_ids.workspace_id, seeded_ids.child_id, seeded_ids.personal_data_uuid, ) await leave_wish_repo.create(_make_wish(ws, child)) result = await planning_service.sync_plan(ws, today=date(2027, 1, 1)) assert result.apply_result.accepted is True assert result.eviction_preview.would_evict is False assert result.eviction_preview.candidates == [] assert result.eviction_preview.shortages == [] class TestProposeStaysReadOnly: async def test_propose_never_writes_a_row( self, planning_engine_service: PlanningEngineService, planned_leave_repo: SQLitePlannedLeaveRepository, leave_wish_repo: SQLiteLeaveWishRepository, seeded_ids: SeededIds, ) -> None: ws, child, _parent = ( seeded_ids.workspace_id, seeded_ids.child_id, seeded_ids.personal_data_uuid, ) await leave_wish_repo.create(_make_wish(ws, child)) proposal = await planning_engine_service.propose(ws, today=date(2027, 1, 1)) assert proposal.days, "expected the proposal to compute something" rows = await planned_leave_repo.list_by_workspace_child(ws, child) assert rows == []