"""Tests for `PlanningService.apply_proposed_days` (POST /api/plan/apply). Covers the contract described in PLANNING_ENGINE_SPEC.md §4: idempotency per (workspace, child, parent, date), collision handling (refuse vs `replace`), validate-the-whole-batch-before-writing, and the balance/eligibility guards. The "nothing written on refusal" assertions check actual repository state (`planned_leave_repo.list(...)`), not just the returned report — that is the property this test file cares about most. """ from datetime import date, datetime, timedelta from uuid import uuid4 from app.entities.leave_wish import LeaveWish from app.entities.parental_leave import ParentalLeaveFetch, TierData 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.planned_leave_repository import SQLitePlannedLeaveRepository from app.services.planning_service import PlanningService, ProposedDayInput from tests.conftest import SeededIds def _make_wish(workspace_id, child_id=None, **overrides) -> LeaveWish: """A `leave_wish` row for FK-constrained `planned_leave_day.wish_id` tests.""" base = dict( id=uuid4(), workspace_id=workspace_id, child_id=child_id, name="Test wish", window_start_month=6, window_start_day=1, window_end_month=6, window_end_day=30, target_weeks=1, recurrence="once", first_year=2026, last_year=None, priority=1, parent_mode="either", parent_personal_data_uuid=None, created_at=datetime.now(), updated_at=datetime.now(), ) base.update(overrides) return LeaveWish(**base) def _make_fetch( personal_data_uuid, workspace_id, child_id, sjukpenning_left: int = 390, lagsta_left: int = 90, ) -> ParentalLeaveFetch: now = datetime.now() tier = TierData( sjukpenning_total=390, sjukpenning_used=390 - sjukpenning_left, sjukpenning_left=sjukpenning_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=now, total_days=480, days_used=480 - sjukpenning_left - lagsta_left, days_left=sjukpenning_left + lagsta_left, reserved_days_total=90, reserved_days_left=90, transferable_days_left=sjukpenning_left + lagsta_left, dubbeldagar_left=60, tier_data=tier, created_at=now, updated_at=now, ) class TestApplyProposedDaysHappyPath: async def test_apply_fresh_plan_creates_rows( self, planning_service: PlanningService, planned_leave_repo: SQLitePlannedLeaveRepository, seeded_ids: SeededIds, ) -> None: ws, child, parent = ( seeded_ids.workspace_id, seeded_ids.child_id, seeded_ids.personal_data_uuid, ) days = [ ProposedDayInput(date(2026, 6, 15), parent, child, "vacation"), ProposedDayInput(date(2026, 6, 16), parent, child, "lagsta"), ProposedDayInput(date(2026, 6, 17), parent, child, "sjukpenning"), ] result = await planning_service.apply_proposed_days(ws, days) assert result.accepted is True assert result.created_count == 3 assert result.skipped_identical_count == 0 assert result.rejections == [] rows = await planned_leave_repo.list_by_workspace_child( workspace_id=ws, child_id=child, from_date=date(2026, 6, 1), to_date=date(2026, 6, 30) ) assert {r.date for r in rows} == {date(2026, 6, 15), date(2026, 6, 16), date(2026, 6, 17)} async def test_apply_skips_free_days( self, planning_service: PlanningService, planned_leave_repo: SQLitePlannedLeaveRepository, seeded_ids: SeededIds, ) -> None: ws, child, parent = ( seeded_ids.workspace_id, seeded_ids.child_id, seeded_ids.personal_data_uuid, ) days = [ ProposedDayInput(date(2026, 6, 15), parent, child, "vacation"), ProposedDayInput(date(2026, 6, 20), parent, child, "free"), # weekend, not persisted ] result = await planning_service.apply_proposed_days(ws, days) assert result.accepted is True assert result.created_count == 1 assert result.skipped_free_count == 1 rows = await planned_leave_repo.list_by_workspace_child( workspace_id=ws, child_id=child, from_date=date(2026, 6, 1), to_date=date(2026, 6, 30) ) assert len(rows) == 1 assert rows[0].date == date(2026, 6, 15) async def test_apply_is_idempotent( self, planning_service: PlanningService, planned_leave_repo: SQLitePlannedLeaveRepository, seeded_ids: SeededIds, ) -> None: ws, child, parent = ( seeded_ids.workspace_id, seeded_ids.child_id, seeded_ids.personal_data_uuid, ) days = [ ProposedDayInput(date(2026, 6, 15), parent, child, "vacation"), ProposedDayInput(date(2026, 6, 16), parent, child, "lagsta"), ] first = await planning_service.apply_proposed_days(ws, days) assert first.accepted is True assert first.created_count == 2 second = await planning_service.apply_proposed_days(ws, days) assert second.accepted is True assert second.created_count == 0 assert second.skipped_identical_count == 2 rows = await planned_leave_repo.list_by_workspace_child( workspace_id=ws, child_id=child, from_date=date(2026, 6, 1), to_date=date(2026, 6, 30) ) assert len(rows) == 2 # no duplicates class TestApplyProposedDaysCollisions: async def test_collision_without_replace_refused_and_writes_nothing( self, planning_service: PlanningService, planned_leave_repo: SQLitePlannedLeaveRepository, seeded_ids: SeededIds, ) -> None: ws, child, parent = ( seeded_ids.workspace_id, seeded_ids.child_id, seeded_ids.personal_data_uuid, ) existing = await planning_service.add_planned_day( workspace_id=ws, child_id=child, parent_personal_data_uuid=parent, date=date(2026, 6, 15), day_type="lagsta", ) days = [ ProposedDayInput(date(2026, 6, 15), parent, child, "vacation"), # collides ProposedDayInput(date(2026, 6, 16), parent, child, "vacation"), # would be new ] result = await planning_service.apply_proposed_days(ws, days, replace=False) assert result.accepted is False assert len(result.collisions) == 1 assert result.collisions[0].date == date(2026, 6, 15) assert result.collisions[0].existing_day_type == "lagsta" assert result.collisions[0].proposed_day_type == "vacation" assert result.rejections # named clearly # Nothing written at all — not even the non-colliding 6/16 day. rows = await planned_leave_repo.list_by_workspace_child( workspace_id=ws, child_id=child, from_date=date(2026, 6, 1), to_date=date(2026, 6, 30) ) assert len(rows) == 1 assert rows[0].id == existing.id assert rows[0].day_type == "lagsta" async def test_collision_with_replace_true_overwrites( self, planning_service: PlanningService, planned_leave_repo: SQLitePlannedLeaveRepository, seeded_ids: SeededIds, ) -> None: ws, child, parent = ( seeded_ids.workspace_id, seeded_ids.child_id, seeded_ids.personal_data_uuid, ) existing = await planning_service.add_planned_day( workspace_id=ws, child_id=child, parent_personal_data_uuid=parent, date=date(2026, 6, 15), day_type="lagsta", ) days = [ ProposedDayInput(date(2026, 6, 15), parent, child, "vacation"), ProposedDayInput(date(2026, 6, 16), parent, child, "vacation"), ] result = await planning_service.apply_proposed_days(ws, days, replace=True) assert result.accepted is True assert result.replaced_count == 1 assert result.created_count == 1 rows = await planned_leave_repo.list_by_workspace_child( workspace_id=ws, child_id=child, from_date=date(2026, 6, 1), to_date=date(2026, 6, 30) ) assert len(rows) == 2 by_date = {r.date: r for r in rows} assert by_date[date(2026, 6, 15)].day_type == "vacation" assert by_date[date(2026, 6, 15)].id != existing.id # old row gone, new one in its place assert by_date[date(2026, 6, 16)].day_type == "vacation" class TestApplyProposedDaysWishAttribution: """`wish_id`/`is_pinned` are new fields on `ProposedDayInput` — the keystone that lets the UI later click a multi-day band on the calendar and edit it as one period. This covers them surviving the write path and the "metadata-only update" carve-out from the collision rules. """ async def test_wish_id_and_is_pinned_survive_create( self, planning_service: PlanningService, 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, ) wish = await leave_wish_repo.create(_make_wish(ws, child)) wish_id = wish.id days = [ ProposedDayInput( date(2026, 6, 15), parent, child, "vacation", wish_id=wish_id, is_pinned=True, ), ] result = await planning_service.apply_proposed_days(ws, days) assert result.accepted is True assert result.created_count == 1 assert result.created[0].wish_id == wish_id assert result.created[0].is_pinned is True rows = await planned_leave_repo.list_by_workspace_child( workspace_id=ws, child_id=child, from_date=date(2026, 6, 1), to_date=date(2026, 6, 30) ) assert len(rows) == 1 assert rows[0].wish_id == wish_id assert rows[0].is_pinned is True async def test_ad_hoc_day_has_no_wish_id( self, planning_service: PlanningService, planned_leave_repo: SQLitePlannedLeaveRepository, seeded_ids: SeededIds, ) -> None: ws, child, parent = ( seeded_ids.workspace_id, seeded_ids.child_id, seeded_ids.personal_data_uuid, ) days = [ProposedDayInput(date(2026, 6, 15), parent, child, "vacation")] result = await planning_service.apply_proposed_days(ws, days) assert result.created[0].wish_id is None assert result.created[0].is_pinned is False async def test_reapply_with_different_wish_id_updates_attribution_not_collision( self, planning_service: PlanningService, planned_leave_repo: SQLitePlannedLeaveRepository, leave_wish_repo: SQLiteLeaveWishRepository, seeded_ids: SeededIds, ) -> None: """Same day_type/is_dubbeldag but a different `wish_id` must be a plain metadata update — not a collision requiring `replace=True`, and it must not be silently dropped as a no-op skip either. """ ws, child, parent = ( seeded_ids.workspace_id, seeded_ids.child_id, seeded_ids.personal_data_uuid, ) wish_a = (await leave_wish_repo.create(_make_wish(ws, child, name="A"))).id wish_b = (await leave_wish_repo.create(_make_wish(ws, child, name="B"))).id first = await planning_service.apply_proposed_days( ws, [ProposedDayInput(date(2026, 6, 15), parent, child, "vacation", wish_id=wish_a)] ) assert first.created[0].wish_id == wish_a original_id = first.created[0].id second = await planning_service.apply_proposed_days( ws, [ProposedDayInput(date(2026, 6, 15), parent, child, "vacation", wish_id=wish_b)], replace=False, ) assert second.accepted is True assert second.collisions == [] assert second.skipped_identical_count == 1 rows = await planned_leave_repo.list_by_workspace_child( workspace_id=ws, child_id=child, from_date=date(2026, 6, 1), to_date=date(2026, 6, 30) ) assert len(rows) == 1 assert rows[0].id == original_id # same row, not replaced assert rows[0].wish_id == wish_b # attribution updated async def test_batch_internal_duplicate_with_different_wish_id_last_one_wins( self, planning_service: PlanningService, planned_leave_repo: SQLitePlannedLeaveRepository, leave_wish_repo: SQLiteLeaveWishRepository, seeded_ids: SeededIds, ) -> None: """Two entries for the same (child, parent, date) that agree on day_type/is_dubbeldag but differ only in `wish_id` must not trip the conflicting-duplicate structural rejection. """ ws, child, parent = ( seeded_ids.workspace_id, seeded_ids.child_id, seeded_ids.personal_data_uuid, ) wish_a = (await leave_wish_repo.create(_make_wish(ws, child, name="A"))).id wish_b = (await leave_wish_repo.create(_make_wish(ws, child, name="B"))).id days = [ ProposedDayInput(date(2026, 6, 15), parent, child, "vacation", wish_id=wish_a), ProposedDayInput(date(2026, 6, 15), parent, child, "vacation", wish_id=wish_b), ] result = await planning_service.apply_proposed_days(ws, days) assert result.accepted is True assert result.rejections == [] rows = await planned_leave_repo.list_by_workspace_child( workspace_id=ws, child_id=child, from_date=date(2026, 6, 1), to_date=date(2026, 6, 30) ) assert len(rows) == 1 assert rows[0].wish_id == wish_b # later entry wins class TestApplyProposedDaysValidation: async def test_invalid_day_type_rejected_before_any_write( self, planning_service: PlanningService, planned_leave_repo: SQLitePlannedLeaveRepository, seeded_ids: SeededIds, ) -> None: ws, child, parent = ( seeded_ids.workspace_id, seeded_ids.child_id, seeded_ids.personal_data_uuid, ) days = [ ProposedDayInput(date(2026, 6, 15), parent, child, "vacation"), # would be valid alone ProposedDayInput(date(2026, 6, 16), parent, child, "annual_leave"), # invalid ] result = await planning_service.apply_proposed_days(ws, days) assert result.accepted is False assert any(r.message_key == "planApply.rejection.invalidDayType" for r in result.rejections) rows = await planned_leave_repo.list_by_workspace_child( workspace_id=ws, child_id=child, from_date=date(2026, 6, 1), to_date=date(2026, 6, 30) ) assert rows == [] async def test_no_snapshot_default_balance_matches_allocation_engine( self, planning_service: PlanningService, planned_leave_repo: SQLitePlannedLeaveRepository, seeded_ids: SeededIds, ) -> None: """Bug A regression, third copy: with no FK snapshot at all, this parent's default tier balance must be their per-parent SHARE (195 sjukpenning / 45 lagsta) of `rules.sjukpenning_days`/`lagsta_niva_days` (390/90, the WHOLE CHILD's combined total across both parents) — not the full 390/90, which `AllocationInputLoader.load` also used to wrongly assume for a single parent before the fix. A batch of exactly 195 sjukpenning days must be accepted; one more must be rejected. """ ws, child, parent = ( seeded_ids.workspace_id, seeded_ids.child_id, seeded_ids.personal_data_uuid, ) # No ParentalLeaveFetch created -> exercises the default fallback. one_too_many = [ ProposedDayInput(date(2026, 1, 1) + timedelta(days=i), parent, child, "sjukpenning") for i in range(196) # 195 default share + 1 over ] result = await planning_service.apply_proposed_days(ws, one_too_many) assert result.accepted is False assert any( r.message_key == "planApply.rejection.insufficientBalance" and r.data["tier"] == "sjukpenning" for r in result.rejections ) rows = await planned_leave_repo.list_by_workspace_child( workspace_id=ws, child_id=child, from_date=date(2026, 1, 1), to_date=date(2027, 1, 1) ) assert rows == [] exactly_the_share = one_too_many[:195] result_ok = await planning_service.apply_proposed_days(ws, exactly_the_share) assert result_ok.accepted is True async def test_batch_exceeding_remaining_balance_rejected_whole( self, planning_service: PlanningService, planned_leave_repo: SQLitePlannedLeaveRepository, parental_leave_repo: SQLiteParentalLeaveRepository, seeded_ids: SeededIds, ) -> None: ws, child, parent = ( seeded_ids.workspace_id, seeded_ids.child_id, seeded_ids.personal_data_uuid, ) await parental_leave_repo.create( _make_fetch(parent, ws, child, sjukpenning_left=1, lagsta_left=90) ) days = [ ProposedDayInput(date(2026, 6, 15), parent, child, "sjukpenning"), ProposedDayInput( date(2026, 6, 16), parent, child, "sjukpenning" ), # only 1 left, asks 2 ] result = await planning_service.apply_proposed_days(ws, days) assert result.accepted is False assert any( r.message_key == "planApply.rejection.insufficientBalance" and r.data["tier"] == "sjukpenning" for r in result.rejections ) rows = await planned_leave_repo.list_by_workspace_child( workspace_id=ws, child_id=child, from_date=date(2026, 6, 1), to_date=date(2026, 6, 30) ) assert rows == []