jacobh2/F-r-ldraledigplannerare
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
6d92b87bf33fd32c5f9ab27c0b51eba4ba99ff07
F-r-ldraledigplannerare/backend/tests/unit/test_services/test_allocation_input_loader.py
T
jacobh2 6d92b87bf3 Give every year of a repeating period its own row 
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
2026-09-19 15:53:25 +02:00

629 lines
24 KiB
Python
Raw Blame History

"""Tests for `AllocationInputLoader` — the only I/O-touching half of the
planning engine (see app.services.allocation_service module docstring).
Uses the same sqlite-backed fixtures as the rest of the service test suite
(tests/conftest.py) to seed a realistic workspace and verify the loader nets
out already-committed `planned_leave_day` rows, falls back to default
entitlement when no FK snapshot exists, and resolves the age-4 savable-days
cap from parent count.
"""
from datetime import date, datetime
from uuid import uuid4
import pytest
from app.entities.child import Child
from app.entities.day_override import DayOverride
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.entities.workspace import Workspace
from app.repositories_impl.sqlite.child_repository import SQLiteChildRepository
from app.repositories_impl.sqlite.day_override_repository import SQLiteDayOverrideRepository
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.repositories_impl.sqlite.workspace_repository import SQLiteWorkspaceRepository
from app.services.allocation_service import AllocationInputLoader, Allocator, PlanningEngineService
from app.utils.errors import NotFoundError, ValidationError
def _make_loader(
workspace_repo: SQLiteWorkspaceRepository,
child_repo: SQLiteChildRepository,
personal_data_repo: SQLitePersonalDataRepository,
parental_leave_repo: SQLiteParentalLeaveRepository,
vacation_repo: SQLiteVacationDaysRepository,
planned_leave_repo: SQLitePlannedLeaveRepository,
leave_wish_repo: SQLiteLeaveWishRepository,
) -> AllocationInputLoader:
return AllocationInputLoader(
workspace_repo=workspace_repo,
child_repo=child_repo,
personal_data_repo=personal_data_repo,
parental_leave_repo=parental_leave_repo,
vacation_repo=vacation_repo,
planned_leave_repo=planned_leave_repo,
leave_wish_repo=leave_wish_repo,
)
class TestAllocationInputLoader:
async def test_workspace_not_found_raises(
self,
workspace_repo: SQLiteWorkspaceRepository,
child_repo: SQLiteChildRepository,
personal_data_repo: SQLitePersonalDataRepository,
parental_leave_repo: SQLiteParentalLeaveRepository,
vacation_repo: SQLiteVacationDaysRepository,
planned_leave_repo: SQLitePlannedLeaveRepository,
leave_wish_repo: SQLiteLeaveWishRepository,
) -> None:
loader = _make_loader(
workspace_repo,
child_repo,
personal_data_repo,
parental_leave_repo,
vacation_repo,
planned_leave_repo,
leave_wish_repo,
)
with pytest.raises(NotFoundError):
await loader.load(uuid4(), today=date(2027, 1, 1))
async def test_load_nets_out_committed_days_and_defaults_entitlement(
self,
workspace_repo: SQLiteWorkspaceRepository,
child_repo: SQLiteChildRepository,
personal_data_repo: SQLitePersonalDataRepository,
parental_leave_repo: SQLiteParentalLeaveRepository,
vacation_repo: SQLiteVacationDaysRepository,
planned_leave_repo: SQLitePlannedLeaveRepository,
leave_wish_repo: SQLiteLeaveWishRepository,
) -> None:
now = datetime.now()
ws = await workspace_repo.create(
Workspace(
id=uuid4(),
name="Familj",
created_at=now,
)
)
parent_with_fetch = await personal_data_repo.create(
PersonalData(
uuid=uuid4(),
workspace_id=ws.id,
name="Parent1",
salary_type="monthly",
employment_status="employed",
created_at=now,
)
)
parent_without_fetch = await personal_data_repo.create(
PersonalData(
uuid=uuid4(),
workspace_id=ws.id,
name="Parent2",
salary_type="hourly",
employment_status="employed",
created_at=now,
)
)
child = await child_repo.create(
Child(
id=uuid4(),
workspace_id=ws.id,
name="Barn",
birth_date=date(2026, 1, 1),
created_at=now,
)
)
tier = TierData(
sjukpenning_total=390,
sjukpenning_used=100,
sjukpenning_left=290,
lagsta_niva_total=90,
lagsta_niva_used=0,
lagsta_niva_left=90,
)
await parental_leave_repo.create(
ParentalLeaveFetch(
id=uuid4(),
personal_data_uuid=parent_with_fetch.uuid,
workspace_id=ws.id,
child_id=child.id,
source="manual",
fetched_at=now,
total_days=240,
days_used=100,
days_left=140,
reserved_days_total=90,
reserved_days_left=90,
transferable_days_left=50,
dubbeldagar_left=60,
tier_data=tier,
created_at=now,
updated_at=now,
)
)
# Two already-committed FP days for parent_with_fetch that the FK
# snapshot above does not know about yet -> must be netted out.
for i in range(2):
await planned_leave_repo.create(
PlannedLeaveDay(
id=uuid4(),
workspace_id=ws.id,
child_id=child.id,
parent_personal_data_uuid=parent_with_fetch.uuid,
date=date(2027, 3, 1 + i),
day_type="sjukpenning",
)
)
# A committed vacation day for the same parent, in 2027.
await planned_leave_repo.create(
PlannedLeaveDay(
id=uuid4(),
workspace_id=ws.id,
child_id=child.id,
parent_personal_data_uuid=parent_with_fetch.uuid,
date=date(2027, 4, 1),
day_type="vacation",
)
)
await vacation_repo.create(
VacationDays(
id=uuid4(),
personal_data_uuid=parent_with_fetch.uuid,
workspace_id=ws.id,
year=2027,
total_days=25,
saved_days_in=0,
)
)
loader = _make_loader(
workspace_repo,
child_repo,
personal_data_repo,
parental_leave_repo,
vacation_repo,
planned_leave_repo,
leave_wish_repo,
)
alloc_input = await loader.load(ws.id, today=date(2027, 1, 1))
assert alloc_input.today == date(2027, 1, 1)
# Two parents -> per-parent age-4 cap (48), not the per-child figure (96).
assert alloc_input.age4_savable_cap == 48
pool_with_fetch = alloc_input.pools[(parent_with_fetch.uuid, child.id)]
assert pool_with_fetch.sjukpenning_left == 288 # 290 - 2 committed
assert pool_with_fetch.lagsta_left == 90
# No FK snapshot at all for parent_without_fetch -> default entitlement.
# `rules.sjukpenning_days`/`lagsta_niva_days` (390/90) are the WHOLE
# CHILD's combined tier totals across both parents, so a single
# parent's default share is half of each (195/45, summing to
# `rules.days_per_parent` == 240) — matching AnalyticsService's own
# no-snapshot fallback (Bug A regression: these two must agree).
pool_defaulted = alloc_input.pools[(parent_without_fetch.uuid, child.id)]
assert pool_defaulted.sjukpenning_left == 195
assert pool_defaulted.lagsta_left == 45
vac = alloc_input.vacation[(parent_with_fetch.uuid, 2027)]
assert vac.total_days == 25
assert vac.already_used == 1 # the committed vacation day
assert len(alloc_input.existing_planned) == 3
assert alloc_input.horizon_end >= date(2030, 1, 1) # at least the child's 4th birthday
assert date(2027, 6, 6) in alloc_input.holidays # Sweden's national day
# Task 1: PersonalData.min_vacation_days_per_year (default 20) must be
# plumbed onto ParentFacts so the pure Allocator can enforce it without
# touching a repository.
parent_facts_by_uuid = {p.uuid: p for p in alloc_input.parents}
assert parent_facts_by_uuid[parent_with_fetch.uuid].min_vacation_days_per_year == 20
assert parent_facts_by_uuid[parent_without_fetch.uuid].min_vacation_days_per_year == 20
async def test_only_pinned_existing_hides_unpinned_wish_rows_but_all_existing_planned_keeps_them(
self,
workspace_repo: SQLiteWorkspaceRepository,
child_repo: SQLiteChildRepository,
personal_data_repo: SQLitePersonalDataRepository,
parental_leave_repo: SQLiteParentalLeaveRepository,
vacation_repo: SQLiteVacationDaysRepository,
planned_leave_repo: SQLitePlannedLeaveRepository,
leave_wish_repo: SQLiteLeaveWishRepository,
) -> None:
"""§fix (parent-reassignment moves dates, decision B) — `sync_plan`
loads with `only_pinned_existing=True`, which must keep hiding
unpinned wish-attributed rows from `existing_planned` (unchanged
behaviour), but `all_existing_planned` must carry EVERY persisted
row regardless — that's what lets `_Ledger._persisted_by_wish` find
a flexible wish's own prior placement even when it's currently
hidden from `existing_planned`. See `AllocationInput
.all_existing_planned`'s docstring in allocation_service.py.
"""
now = datetime.now()
ws = await workspace_repo.create(Workspace(id=uuid4(), name="Familj", created_at=now))
parent = await personal_data_repo.create(
PersonalData(
uuid=uuid4(),
workspace_id=ws.id,
name="Parent",
salary_type="monthly",
employment_status="employed",
created_at=now,
)
)
child = await child_repo.create(
Child(
id=uuid4(),
workspace_id=ws.id,
name="Barn",
birth_date=date(2026, 1, 1),
created_at=now,
)
)
wish = await leave_wish_repo.create(
LeaveWish(
id=uuid4(),
workspace_id=ws.id,
child_id=child.id,
name="Höstlov",
window_start_month=10,
window_start_day=1,
window_end_month=10,
window_end_day=7,
target_weeks=1,
first_year=2027,
last_year=None,
priority=1,
parent_mode="either",
parent_personal_data_uuid=None,
window_mode="flexible",
)
)
pinned = await planned_leave_repo.create(
PlannedLeaveDay(
id=uuid4(),
workspace_id=ws.id,
child_id=child.id,
parent_personal_data_uuid=parent.uuid,
date=date(2027, 10, 1),
day_type="vacation",
is_pinned=True,
)
)
unpinned_wish_row = await planned_leave_repo.create(
PlannedLeaveDay(
id=uuid4(),
workspace_id=ws.id,
child_id=child.id,
parent_personal_data_uuid=parent.uuid,
date=date(2027, 10, 4),
day_type="vacation",
wish_id=wish.id,
is_pinned=False,
)
)
ad_hoc = await planned_leave_repo.create(
PlannedLeaveDay(
id=uuid4(),
workspace_id=ws.id,
child_id=child.id,
parent_personal_data_uuid=parent.uuid,
date=date(2027, 10, 5),
day_type="vacation",
wish_id=None,
is_pinned=False,
)
)
loader = _make_loader(
workspace_repo,
child_repo,
personal_data_repo,
parental_leave_repo,
vacation_repo,
planned_leave_repo,
leave_wish_repo,
)
alloc_input = await loader.load(ws.id, today=date(2027, 1, 1), only_pinned_existing=True)
visible_ids = {d.id for d in alloc_input.existing_planned}
assert visible_ids == {pinned.id, ad_hoc.id}
assert unpinned_wish_row.id not in visible_ids
assert alloc_input.all_existing_planned is not None
all_ids = {d.id for d in alloc_input.all_existing_planned}
assert all_ids == {pinned.id, unpinned_wish_row.id, ad_hoc.id}
async def test_load_reflects_a_custom_minimum_vacation_days(
self,
workspace_repo: SQLiteWorkspaceRepository,
child_repo: SQLiteChildRepository,
personal_data_repo: SQLitePersonalDataRepository,
parental_leave_repo: SQLiteParentalLeaveRepository,
vacation_repo: SQLiteVacationDaysRepository,
planned_leave_repo: SQLitePlannedLeaveRepository,
leave_wish_repo: SQLiteLeaveWishRepository,
) -> None:
"""A parent whose employer sets a higher-than-statutory floor (Task 1)
must have that reflected on `ParentFacts`, not the seeded default."""
now = datetime.now()
ws = await workspace_repo.create(Workspace(id=uuid4(), name="Familj", created_at=now))
parent = await personal_data_repo.create(
PersonalData(
uuid=uuid4(),
workspace_id=ws.id,
name="Parent",
min_vacation_days_per_year=25,
created_at=now,
)
)
loader = _make_loader(
workspace_repo,
child_repo,
personal_data_repo,
parental_leave_repo,
vacation_repo,
planned_leave_repo,
leave_wish_repo,
)
alloc_input = await loader.load(ws.id, today=date(2027, 1, 1))
assert len(alloc_input.parents) == 1
assert alloc_input.parents[0].uuid == parent.uuid
assert alloc_input.parents[0].min_vacation_days_per_year == 25
async def test_planning_engine_service_glues_loader_and_allocator(
self,
workspace_repo: SQLiteWorkspaceRepository,
child_repo: SQLiteChildRepository,
personal_data_repo: SQLitePersonalDataRepository,
parental_leave_repo: SQLiteParentalLeaveRepository,
vacation_repo: SQLiteVacationDaysRepository,
planned_leave_repo: SQLitePlannedLeaveRepository,
leave_wish_repo: SQLiteLeaveWishRepository,
) -> None:
now = datetime.now()
ws = await workspace_repo.create(Workspace(id=uuid4(), name="Solo", created_at=now))
await personal_data_repo.create(
PersonalData(uuid=uuid4(), workspace_id=ws.id, name="Solo Parent", created_at=now)
)
loader = _make_loader(
workspace_repo,
child_repo,
personal_data_repo,
parental_leave_repo,
vacation_repo,
planned_leave_repo,
leave_wish_repo,
)
service = PlanningEngineService(loader, Allocator())
proposal = await service.propose(ws.id, today=date(2027, 1, 1))
assert proposal.days == []
assert proposal.outcomes == []
async def test_day_overrides_default_to_empty_without_a_repo(
self,
workspace_repo: SQLiteWorkspaceRepository,
child_repo: SQLiteChildRepository,
personal_data_repo: SQLitePersonalDataRepository,
parental_leave_repo: SQLiteParentalLeaveRepository,
vacation_repo: SQLiteVacationDaysRepository,
planned_leave_repo: SQLitePlannedLeaveRepository,
leave_wish_repo: SQLiteLeaveWishRepository,
) -> None:
"""`day_override_repo` is optional (backward compatible) — omitting it
must not crash the loader, just leave `day_overrides` empty."""
ws = await workspace_repo.create(
Workspace(id=uuid4(), name="No overrides", created_at=datetime.now())
)
await personal_data_repo.create(
PersonalData(uuid=uuid4(), workspace_id=ws.id, name="Parent", created_at=datetime.now())
)
loader = _make_loader(
workspace_repo,
child_repo,
personal_data_repo,
parental_leave_repo,
vacation_repo,
planned_leave_repo,
leave_wish_repo,
)
alloc_input = await loader.load(ws.id, today=date(2027, 1, 1))
assert alloc_input.day_overrides == {}
async def test_day_overrides_loaded_from_repository(
self,
workspace_repo: SQLiteWorkspaceRepository,
child_repo: SQLiteChildRepository,
personal_data_repo: SQLitePersonalDataRepository,
parental_leave_repo: SQLiteParentalLeaveRepository,
vacation_repo: SQLiteVacationDaysRepository,
planned_leave_repo: SQLitePlannedLeaveRepository,
leave_wish_repo: SQLiteLeaveWishRepository,
day_override_repo: SQLiteDayOverrideRepository,
) -> None:
"""When a `DayOverrideRepository` is wired in, its rows land in
`AllocationInput.day_overrides` as {date: is_working_day}."""
now = datetime.now()
ws = await workspace_repo.create(
Workspace(id=uuid4(), name="With overrides", created_at=now)
)
await personal_data_repo.create(
PersonalData(uuid=uuid4(), workspace_id=ws.id, name="Parent", created_at=now)
)
await day_override_repo.create(
DayOverride(
id=uuid4(),
workspace_id=ws.id,
date=date(2027, 12, 24),
is_working_day=True,
created_at=now,
updated_at=now,
)
)
await day_override_repo.create(
DayOverride(
id=uuid4(),
workspace_id=ws.id,
date=date(2027, 2, 2),
is_working_day=False,
created_at=now,
updated_at=now,
)
)
loader = AllocationInputLoader(
workspace_repo=workspace_repo,
child_repo=child_repo,
personal_data_repo=personal_data_repo,
parental_leave_repo=parental_leave_repo,
vacation_repo=vacation_repo,
planned_leave_repo=planned_leave_repo,
leave_wish_repo=leave_wish_repo,
day_override_repo=day_override_repo,
)
alloc_input = await loader.load(ws.id, today=date(2027, 1, 1))
assert alloc_input.day_overrides == {
date(2027, 12, 24): True,
date(2027, 2, 2): False,
}
async def test_load_costs_julafton_correctly_for_a_far_future_horizon(
self,
workspace_repo: SQLiteWorkspaceRepository,
child_repo: SQLiteChildRepository,
personal_data_repo: SQLitePersonalDataRepository,
parental_leave_repo: SQLiteParentalLeaveRepository,
vacation_repo: SQLiteVacationDaysRepository,
planned_leave_repo: SQLitePlannedLeaveRepository,
leave_wish_repo: SQLiteLeaveWishRepository,
) -> None:
"""Before the cache was extended to ~30 years, a horizon reaching the
late 2040s ran past the old 10-year cache (2026-2035) and the loader
silently fell back to the live `holidays` library, which reports
Julafton/Nyårsafton/Midsommarafton as ordinary working days. A child
born in 2044 pushes the horizon (their 4th birthday) to 2048 — assert
the resulting `holidays` map still carries all three as real red
days that far out, proving the extended cache (not a silent
fallback) served it, and that the allocator would cost them
correctly rather than charging a needless semesterdag."""
now = datetime.now()
ws = await workspace_repo.create(Workspace(id=uuid4(), name="Future", created_at=now))
await personal_data_repo.create(
PersonalData(uuid=uuid4(), workspace_id=ws.id, name="Parent", created_at=now)
)
await child_repo.create(
Child(
id=uuid4(),
workspace_id=ws.id,
name="Barn",
birth_date=date(2044, 1, 1),
created_at=now,
)
)
loader = _make_loader(
workspace_repo,
child_repo,
personal_data_repo,
parental_leave_repo,
vacation_repo,
planned_leave_repo,
leave_wish_repo,
)
alloc_input = await loader.load(ws.id, today=date(2027, 1, 1))
assert alloc_input.horizon_end.year == 2048
julafton = alloc_input.holidays.get(date(2048, 12, 24))
assert julafton is not None and "julafton" in julafton.lower()
nyarsafton = alloc_input.holidays.get(date(2048, 12, 31))
assert nyarsafton is not None and "nyårsafton" in nyarsafton.lower()
midsommarafton = [
n for d, n in alloc_input.holidays.items() if "midsommarafton" in n.lower()
]
assert midsommarafton, "Midsommarafton must still be a real free day in 2048"
async def test_load_raises_validation_error_when_horizon_exceeds_the_cache(
self,
workspace_repo: SQLiteWorkspaceRepository,
child_repo: SQLiteChildRepository,
personal_data_repo: SQLitePersonalDataRepository,
parental_leave_repo: SQLiteParentalLeaveRepository,
vacation_repo: SQLiteVacationDaysRepository,
planned_leave_repo: SQLitePlannedLeaveRepository,
leave_wish_repo: SQLiteLeaveWishRepository,
) -> None:
"""A wish row anchored in a far-out year, with no children yet (so
there's no age-12 hard cap to save it), can still in principle push
the horizon past even a ~30-year cache. `LeaveWishService` clamps a
materialised series to the cached range, but a row can reach the DB
another way (a hand-written import, an older client, this test's
direct repo write), so the loader must still defend itself. Rather than silently falling
back to the live `holidays` library (the original bug) or silently
clamping the horizon (which would contradict the §6 decision to plan
past B4 and show the squeeze), the loader must raise a
`ValidationError` — the same domain error `CalendarService` uses for
an out-of-range `start_year` — so the router can turn it into a 422
the UI can explain."""
now = datetime.now()
ws = await workspace_repo.create(Workspace(id=uuid4(), name="FarFuture", created_at=now))
await personal_data_repo.create(
PersonalData(uuid=uuid4(), workspace_id=ws.id, name="Parent", created_at=now)
)
await leave_wish_repo.create(
LeaveWish(
id=uuid4(),
workspace_id=ws.id,
child_id=None,
name="Sommar",
window_start_month=6,
window_start_day=1,
window_end_month=6,
window_end_day=30,
target_weeks=4,
first_year=2200,
last_year=None,
priority=1,
parent_mode="either",
parent_personal_data_uuid=None,
)
)
loader = _make_loader(
workspace_repo,
child_repo,
personal_data_repo,
parental_leave_repo,
vacation_repo,
planned_leave_repo,
leave_wish_repo,
)
with pytest.raises(ValidationError):
await loader.load(ws.id, today=date(2027, 1, 1))
Reference in New Issue View Git Blame Copy Permalink