jacobh2
6d92b87bf3
A repeating period was a single row that the planner expanded into one window per year at read time, so no year had anywhere to differ from the others. Editing one year rewrote them all, and removing a year in the middle had to split the wish in two. Neither survives contact with the thing this app is for: Easter moves, and the cheapest week is not the same week every year. Each year is now its own row, linked by a series id. Editing one year touches one row; editing the series re-projects the rest from it. Removing a year is one delete. The weekday snap is worked out once, when the rows are made, and stored, so a row now says what it means rather than being re-derived on every read. Recurrence stops being a stored field and becomes what it always was in practice — an instruction at create and update time; whether a period repeats is whether it belongs to a series. Migration 004 rebuilds the table and expands the existing rows, keeping each original row as its first year so the days already planned against it still point somewhere. A Feb 29 anchor recurring into a common year used to raise out of the allocator; it clamps now. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Ei7gCZzdd751uZtPzb1Pep
932 lines
36 KiB
Python
932 lines
36 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,
|
|
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 agree with
|
|
`AllocationInputLoader.load`'s (both go through the shared
|
|
`default_tier_pool_share` helper — see its docstring). `seeded_ids`'s
|
|
workspace has exactly ONE member, so the default is the FULL
|
|
`rules.sjukpenning_days` (390) — the pool follows actual parent
|
|
count, not an unconditional halving (see `app.utils.tip_predicates.
|
|
default_tier_pool_share`: a sole parent gets the whole pool, a
|
|
partnered one gets half). A batch of exactly 390 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(391) # 390 default share (sole parent) + 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[:390]
|
|
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"
|