jacobh2
0fc71101ae
Entering a parent's Försäkringskassan balances had no tests at all, despite being how every number in the planner originates and despite it becoming the first thing a new user does. The scraper endpoints are pinned per test rather than read from the ambient configuration, because a local .env sets the flag opposite to the shipped default and a test that trusts it would pass here and fail in CI. Also covered: the hard expiry at twelve, whose rejection path nothing reached and whose boundary turns out to be inclusive; and the dubbeldagar count cap, which now drives past the limit and asserts the days beyond it fall back to a single parent. Two weak assertions were strengthened, including one that derived its expectation from the same response it was checking and so could not have failed on a shared error.
930 lines
35 KiB
Python
930 lines
35 KiB
Python
"""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 dateutil.relativedelta import relativedelta
|
|
|
|
from app.config import settings
|
|
from app.entities.leave_wish import LeaveWish
|
|
from app.entities.parental_leave import ParentalLeaveFetch, TierData
|
|
from app.entities.planned_leave import PlannedLeaveDay
|
|
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 == []
|
|
|
|
|
|
class TestApplyProposedDaysEligibleRange:
|
|
"""`PlanningService._validate_eligible_range` (planning_service.py:748-775)
|
|
rejects any planned day before the child's birth date or after
|
|
`birth_date + LeaveRules.hard_expiry_age_years` (the Swedish 12-year hard
|
|
expiry on FP days).
|
|
|
|
The upper bound is INCLUSIVE: the source only rejects `d.date >
|
|
hard_expiry`, so the exact 12th-birthday date itself is still eligible
|
|
-- confirmed by `test_hard_expiry_boundary_date_itself_is_inclusive_and_accepted`
|
|
below, which would fail if that boundary were ever tightened to `>=`.
|
|
|
|
`seeded_ids`' child has a fixed `birth_date` of 2024-01-01
|
|
(`tests/conftest.py`), which these tests rely on directly.
|
|
"""
|
|
|
|
_BIRTH_DATE = date(2024, 1, 1)
|
|
|
|
async def test_day_after_the_hard_expiry_is_rejected(
|
|
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,
|
|
)
|
|
hard_expiry = self._BIRTH_DATE + relativedelta(years=settings.rules.hard_expiry_age_years)
|
|
day_after = hard_expiry + timedelta(days=1)
|
|
|
|
result = await planning_service.apply_proposed_days(
|
|
ws, [ProposedDayInput(day_after, parent, child, "sjukpenning")]
|
|
)
|
|
|
|
assert result.accepted is False
|
|
assert any(
|
|
r.message_key == "planApply.rejection.outsideEligibleWindow" for r in result.rejections
|
|
)
|
|
rows = await planned_leave_repo.list(parent_personal_data_uuid=parent)
|
|
assert rows == []
|
|
|
|
async def test_day_before_the_birth_date_is_rejected(
|
|
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,
|
|
)
|
|
day_before = self._BIRTH_DATE - timedelta(days=1)
|
|
|
|
result = await planning_service.apply_proposed_days(
|
|
ws, [ProposedDayInput(day_before, parent, child, "sjukpenning")]
|
|
)
|
|
|
|
assert result.accepted is False
|
|
assert any(
|
|
r.message_key == "planApply.rejection.outsideEligibleWindow" for r in result.rejections
|
|
)
|
|
rows = await planned_leave_repo.list(parent_personal_data_uuid=parent)
|
|
assert rows == []
|
|
|
|
async def test_hard_expiry_boundary_date_itself_is_inclusive_and_accepted(
|
|
self,
|
|
planning_service: PlanningService,
|
|
seeded_ids: SeededIds,
|
|
) -> None:
|
|
"""The 12th-birthday date exactly equal to `birth_date +
|
|
hard_expiry_age_years` is still an eligible day -- the source rejects
|
|
strictly-greater dates only, so this is the INCLUSIVE boundary."""
|
|
ws, child, parent = (
|
|
seeded_ids.workspace_id,
|
|
seeded_ids.child_id,
|
|
seeded_ids.personal_data_uuid,
|
|
)
|
|
hard_expiry = self._BIRTH_DATE + relativedelta(years=settings.rules.hard_expiry_age_years)
|
|
|
|
result = await planning_service.apply_proposed_days(
|
|
ws, [ProposedDayInput(hard_expiry, parent, child, "sjukpenning")]
|
|
)
|
|
|
|
assert result.accepted is True, result.rejections
|
|
|
|
async def test_birth_date_itself_is_inclusive_and_accepted(
|
|
self,
|
|
planning_service: PlanningService,
|
|
seeded_ids: SeededIds,
|
|
) -> None:
|
|
ws, child, parent = (
|
|
seeded_ids.workspace_id,
|
|
seeded_ids.child_id,
|
|
seeded_ids.personal_data_uuid,
|
|
)
|
|
|
|
result = await planning_service.apply_proposed_days(
|
|
ws, [ProposedDayInput(self._BIRTH_DATE, parent, child, "sjukpenning")]
|
|
)
|
|
|
|
assert result.accepted is True, result.rejections
|
|
|
|
async def test_rejected_batch_writes_nothing_even_with_a_valid_day_included(
|
|
self,
|
|
planning_service: PlanningService,
|
|
planned_leave_repo: SQLitePlannedLeaveRepository,
|
|
seeded_ids: SeededIds,
|
|
) -> None:
|
|
"""All-or-nothing: one out-of-window day in the batch must sink the
|
|
whole batch, including the otherwise-valid day alongside it."""
|
|
ws, child, parent = (
|
|
seeded_ids.workspace_id,
|
|
seeded_ids.child_id,
|
|
seeded_ids.personal_data_uuid,
|
|
)
|
|
hard_expiry = self._BIRTH_DATE + relativedelta(years=settings.rules.hard_expiry_age_years)
|
|
day_after = hard_expiry + timedelta(days=1)
|
|
valid_day = ProposedDayInput(date(2026, 6, 15), parent, child, "sjukpenning")
|
|
invalid_day = ProposedDayInput(day_after, parent, child, "sjukpenning")
|
|
|
|
result = await planning_service.apply_proposed_days(ws, [valid_day, invalid_day])
|
|
|
|
assert result.accepted is False
|
|
rows = await planned_leave_repo.list(parent_personal_data_uuid=parent)
|
|
assert rows == []
|
|
|
|
|
|
class TestApplyProposedDaysDeleteKeys:
|
|
"""`delete_keys` folds explicit deletions into the same validated batch as
|
|
|
|
the creates/replaces -- this is the fix for the bug where `sync_plan`
|
|
deleted stale rows only AFTER `apply_proposed_days` had already validated
|
|
(and written) the fresh batch, so capacity a deletion frees was never
|
|
credited back before the balance/quota check ran.
|
|
"""
|
|
|
|
async def test_delete_keys_credit_sjukpenning_tier_before_checking_balance(
|
|
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=5, lagsta_left=90)
|
|
)
|
|
old = await planning_service.add_planned_day(
|
|
workspace_id=ws,
|
|
child_id=child,
|
|
parent_personal_data_uuid=parent,
|
|
date=date(2026, 6, 1),
|
|
day_type="sjukpenning",
|
|
)
|
|
# 5 left, 1 already committed -> 4 free right now. Ask for 5 new
|
|
# sjukpenning days while deleting the 1 old one in the SAME call:
|
|
# net_new = 5 - 1 = 4, which fits -- but only if the deletion is
|
|
# credited before the check runs.
|
|
new_days = [
|
|
ProposedDayInput(date(2026, 6, i), parent, child, "sjukpenning") for i in range(2, 7)
|
|
]
|
|
delete_keys = {(child, parent, date(2026, 6, 1))}
|
|
|
|
result = await planning_service.apply_proposed_days(ws, new_days, delete_keys=delete_keys)
|
|
|
|
assert result.accepted is True, result.rejections
|
|
assert result.deleted_count == 1
|
|
assert result.created_count == 5
|
|
|
|
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, i) for i in range(2, 7)}
|
|
assert old.id not in {r.id for r in rows}
|
|
|
|
async def test_delete_keys_credit_lagsta_tier_before_checking_balance(
|
|
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=390, lagsta_left=5)
|
|
)
|
|
old = await planning_service.add_planned_day(
|
|
workspace_id=ws,
|
|
child_id=child,
|
|
parent_personal_data_uuid=parent,
|
|
date=date(2026, 6, 1),
|
|
day_type="lagsta",
|
|
)
|
|
new_days = [
|
|
ProposedDayInput(date(2026, 6, i), parent, child, "lagsta") for i in range(2, 7)
|
|
]
|
|
delete_keys = {(child, parent, date(2026, 6, 1))}
|
|
|
|
result = await planning_service.apply_proposed_days(ws, new_days, delete_keys=delete_keys)
|
|
|
|
assert result.accepted is True, result.rejections
|
|
assert result.deleted_count == 1
|
|
assert result.created_count == 5
|
|
|
|
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, i) for i in range(2, 7)}
|
|
assert old.id not in {r.id for r in rows}
|
|
|
|
async def test_delete_keys_credit_vacation_deletion_alongside_new_vacation_days(
|
|
self,
|
|
planning_service: PlanningService,
|
|
planned_leave_repo: SQLitePlannedLeaveRepository,
|
|
seeded_ids: SeededIds,
|
|
) -> None:
|
|
"""This batch nets to zero vacation days for 2026 (1 deleted, 1 added),
|
|
|
|
which fits comfortably under the statutory default entitlement (no
|
|
`VacationDays` row exists for this parent/year) -- so it exercises
|
|
the general `delete_keys` write/validate machinery for a vacation-
|
|
typed deletion, but does not by itself prove the vacation balance cap
|
|
(`_validate_remaining_vacation_balance`) is enforced or that its
|
|
`delete_keys` credit path matters. See
|
|
`test_planning_service_vacation.py::TestVacationBalanceDeleteKeysCredit`
|
|
for a tight-budget version that actually would reject without the
|
|
credit.
|
|
"""
|
|
ws, child, parent = (
|
|
seeded_ids.workspace_id,
|
|
seeded_ids.child_id,
|
|
seeded_ids.personal_data_uuid,
|
|
)
|
|
old = await planning_service.add_planned_day(
|
|
workspace_id=ws,
|
|
child_id=child,
|
|
parent_personal_data_uuid=parent,
|
|
date=date(2026, 6, 1),
|
|
day_type="vacation",
|
|
)
|
|
new_days = [ProposedDayInput(date(2026, 6, 2), parent, child, "vacation")]
|
|
delete_keys = {(child, parent, date(2026, 6, 1))}
|
|
|
|
result = await planning_service.apply_proposed_days(ws, new_days, delete_keys=delete_keys)
|
|
|
|
assert result.accepted is True, result.rejections
|
|
assert result.deleted_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 {r.date for r in rows} == {date(2026, 6, 2)}
|
|
assert old.id not in {r.id for r in rows}
|
|
|
|
async def test_delete_keys_credit_frees_dubbeldagar_quota(
|
|
self,
|
|
planning_service: PlanningService,
|
|
planned_leave_repo: SQLitePlannedLeaveRepository,
|
|
seeded_ids: SeededIds,
|
|
) -> None:
|
|
"""58 dubbeldagar already planned (below the 60 cap), 7 more asked
|
|
|
|
for while 5 are deleted in the same call: 58 - 5 + 7 = 60, exactly at
|
|
cap -> must be accepted. Without crediting the 5 deletions the
|
|
projected total reads as 58 + 7 = 65, over cap, and would be
|
|
(wrongly) refused.
|
|
"""
|
|
ws, child, parent = (
|
|
seeded_ids.workspace_id,
|
|
seeded_ids.child_id,
|
|
seeded_ids.personal_data_uuid,
|
|
)
|
|
now = datetime.now()
|
|
existing_dates = [date(2024, 2, 1) + timedelta(days=i) for i in range(58)]
|
|
for d in existing_dates:
|
|
await planned_leave_repo.create(
|
|
PlannedLeaveDay(
|
|
id=uuid4(),
|
|
workspace_id=ws,
|
|
child_id=child,
|
|
parent_personal_data_uuid=parent,
|
|
date=d,
|
|
day_type="sjukpenning",
|
|
is_dubbeldag=True,
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
)
|
|
|
|
to_delete = existing_dates[:5]
|
|
delete_keys = {(child, parent, d) for d in to_delete}
|
|
new_dates = [date(2024, 4, 1) + timedelta(days=i) for i in range(7)]
|
|
new_days = [
|
|
ProposedDayInput(d, parent, child, "sjukpenning", is_dubbeldag=True) for d in new_dates
|
|
]
|
|
|
|
result = await planning_service.apply_proposed_days(ws, new_days, delete_keys=delete_keys)
|
|
|
|
assert result.accepted is True, result.rejections
|
|
assert result.deleted_count == 5
|
|
assert result.created_count == 7
|
|
|
|
total_dubbeldagar = await planned_leave_repo.count_dubbeldagar(child)
|
|
assert total_dubbeldagar == 60
|
|
|
|
async def test_delete_keys_none_is_unchanged_behaviour(
|
|
self,
|
|
planning_service: PlanningService,
|
|
planned_leave_repo: SQLitePlannedLeaveRepository,
|
|
seeded_ids: SeededIds,
|
|
) -> None:
|
|
"""No `delete_keys` (the default) behaves exactly as before: nothing
|
|
|
|
deleted beyond ordinary `replace` semantics, `deleted_count` stays 0.
|
|
"""
|
|
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.accepted is True
|
|
assert result.deleted_count == 0
|
|
assert result.created_count == 1
|
|
|
|
async def test_rejected_batch_with_delete_keys_deletes_nothing(
|
|
self,
|
|
planning_service: PlanningService,
|
|
planned_leave_repo: SQLitePlannedLeaveRepository,
|
|
parental_leave_repo: SQLiteParentalLeaveRepository,
|
|
seeded_ids: SeededIds,
|
|
) -> None:
|
|
"""Even with `delete_keys` credited, a batch that still exceeds the
|
|
|
|
remaining balance is refused whole, and the row named by
|
|
`delete_keys` must survive untouched -- the all-or-nothing contract
|
|
applies to deletions too.
|
|
"""
|
|
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=5, lagsta_left=90)
|
|
)
|
|
old = await planning_service.add_planned_day(
|
|
workspace_id=ws,
|
|
child_id=child,
|
|
parent_personal_data_uuid=parent,
|
|
date=date(2026, 6, 1),
|
|
day_type="sjukpenning",
|
|
)
|
|
# 5 left, 1 committed -> 4 free. Deleting the 1 old row credits back
|
|
# 1, so net_new must be <= 5 to pass; ask for 7 instead (net_new = 6).
|
|
new_days = [
|
|
ProposedDayInput(date(2026, 6, i), parent, child, "sjukpenning") for i in range(2, 9)
|
|
]
|
|
delete_keys = {(child, parent, date(2026, 6, 1))}
|
|
|
|
result = await planning_service.apply_proposed_days(ws, new_days, delete_keys=delete_keys)
|
|
|
|
assert result.accepted is False
|
|
assert any(
|
|
r.message_key == "planApply.rejection.insufficientBalance" 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 len(rows) == 1
|
|
assert rows[0].id == old.id # the row `delete_keys` named was NOT deleted
|
|
|
|
async def test_delete_key_colliding_with_a_write_key_is_a_structural_rejection(
|
|
self,
|
|
planning_service: PlanningService,
|
|
planned_leave_repo: SQLitePlannedLeaveRepository,
|
|
seeded_ids: SeededIds,
|
|
) -> None:
|
|
"""A batch cannot both write and delete the same (child, parent,
|
|
|
|
date) slot -- that's a structural error caught before any DB read,
|
|
same as the other structural checks in this file.
|
|
"""
|
|
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")]
|
|
delete_keys = {(child, parent, date(2026, 6, 15))}
|
|
|
|
result = await planning_service.apply_proposed_days(ws, days, delete_keys=delete_keys)
|
|
|
|
assert result.accepted is False
|
|
assert any(
|
|
r.message_key == "planApply.rejection.deleteKeyConflict" 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 len(rows) == 1
|
|
assert rows[0].id == existing.id
|
|
assert rows[0].day_type == "lagsta"
|