jacobh2/F-r-ldraledigplannerare
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add a plan sync endpoint that recomputes, persists and reports the delta

Browse Source 
The calendar is becoming the plan: a user edits a period and the schedule
updates immediately, with no separate propose and apply steps. POST
/api/plan/sync does both and returns a per-date diff against a snapshot taken
before any write, so the interface can say what actually moved rather than
guessing from coarse counters.

Pinned days now mean something. A day the user hand-edited keeps its type and
its parent: the loader hides unpinned wish-attributed rows from the allocator
so a re-plan is free to move them, while pinned rows stay visible and the
existing occupancy check refuses to touch them. Their cost is debited off the
pool before any window is funded, following the precedent already set for
preschool planning days.

Propose and apply are untouched. The allocator stays pure.
This commit is contained in:
Jacob Hagstedt
2026-09-17 17:58:47 +02:00
parent 0e14b4dac8
commit 6cd60887d5
8 changed files with 863 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/dependencies.py
+12 -11
View File
@@ -271,17 +271,6 @@ def get_analytics_service(
AnalyticsServiceDep = Annotated[AnalyticsService, Depends(get_analytics_service)]
def get_planning_service(
repo: PlannedLeaveRepositoryDep,
analytics_service: AnalyticsServiceDep,
parental_leave_repo: ParentalLeaveRepositoryDep,
) -> PlanningService:
return PlanningService(repo, analytics_service, parental_leave_repo)
PlanningServiceDep = Annotated[PlanningService, Depends(get_planning_service)]
def get_allocation_input_loader(
workspace_repo: WorkspaceRepositoryDep,
child_repo: ChildRepositoryDep,
@@ -314,6 +303,18 @@ def get_planning_engine_service(loader: AllocationInputLoaderDep) -> PlanningEng
PlanningEngineServiceDep = Annotated[PlanningEngineService, Depends(get_planning_engine_service)]
def get_planning_service(
repo: PlannedLeaveRepositoryDep,
analytics_service: AnalyticsServiceDep,
parental_leave_repo: ParentalLeaveRepositoryDep,
planning_engine_service: PlanningEngineServiceDep,
) -> PlanningService:
return PlanningService(repo, analytics_service, parental_leave_repo, planning_engine_service)
PlanningServiceDep = Annotated[PlanningService, Depends(get_planning_service)]
# ---------------------------------------------------------------------------
# Current-user dependency (JWT bearer)
# ---------------------------------------------------------------------------
backend/app/models/plan.py
+60
View File
@@ -20,6 +20,8 @@ from app.services.allocation_service import PlanProposal
from app.services.planning_service import (
DayCollision,
PlanApplyResult,
PlanChange,
PlanSyncResult,
ProposedDayInput,
Rejection,
)
@@ -328,3 +330,61 @@ class PlanApplyResponse(BaseModel):
rejections=[RejectionResponse.from_domain(r) for r in result.rejections],
created_days=[PlannedLeaveDayResponse.from_entity(d) for d in result.created],
)
# ---------------------------------------------------------------------------
# POST /api/plan/sync
# ---------------------------------------------------------------------------
class PlanSyncRequest(BaseModel):
workspace_id: UUID
class PlanChangeResponse(BaseModel):
"""One (child, parent, date) slot's delta from a `sync` call — see
`PlanChange`'s docstring for the `kind` vocabulary. Structured data only;
the frontend composes the toast copy ("Sommar 2028 flyttades 2 dagar —
Ångra") itself.
"""
kind: str
date: date
parent_personal_data_uuid: UUID
child_id: UUID
old_day_type: str | None
new_day_type: str | None
old_is_dubbeldag: bool | None
new_is_dubbeldag: bool | None
old_wish_id: UUID | None
new_wish_id: UUID | None
@classmethod
def from_domain(cls, change: PlanChange) -> "PlanChangeResponse":
return cls(
kind=change.kind,
date=change.date,
parent_personal_data_uuid=change.parent_personal_data_uuid,
child_id=change.child_id,
old_day_type=change.old_day_type,
new_day_type=change.new_day_type,
old_is_dubbeldag=change.old_is_dubbeldag,
new_is_dubbeldag=change.new_is_dubbeldag,
old_wish_id=change.old_wish_id,
new_wish_id=change.new_wish_id,
)
class PlanSyncResponse(BaseModel):
proposal: PlanProposalResponse
apply_result: PlanApplyResponse
changes: list[PlanChangeResponse]
@classmethod
def from_domain(cls, result: PlanSyncResult) -> "PlanSyncResponse":
return cls(
proposal=PlanProposalResponse.from_domain(result.proposal),
apply_result=PlanApplyResponse.from_domain(result.apply_result),
changes=[PlanChangeResponse.from_domain(c) for c in result.changes],
)
backend/app/routers/plan.py
+22
View File
@@ -15,6 +15,8 @@ from app.models.plan import (
PlanApplyResponse,
PlanProposalResponse,
PlanProposeRequest,
PlanSyncRequest,
PlanSyncResponse,
)
from app.utils.errors import NotFoundError, ValidationError
@@ -55,3 +57,23 @@ async def apply_plan(payload: PlanApplyRequest, service: PlanningServiceDep) ->
except NotFoundError as exc:
raise HTTPException(status.HTTP_404_NOT_FOUND, str(exc)) from exc
return PlanApplyResponse.from_domain(result)
@router.post("/sync", response_model=PlanSyncResponse)
async def sync_plan(payload: PlanSyncRequest, service: PlanningServiceDep) -> PlanSyncResponse:
"""Recompute and persist a workspace's plan in one step.
The direct-manipulation calendar UX's one call: no separate preview
step. Runs propose then apply internally (see
`PlanningService.sync_plan`), never retyping or overwriting a day the
user has pinned, and reports exactly what changed (`changes`) so the UI
can show e.g. "Sommar 2028 flyttades 2 dagar — Ångra" instead of
silently rewriting the calendar.
"""
try:
result = await service.sync_plan(payload.workspace_id)
except NotFoundError as exc:
raise HTTPException(status.HTTP_404_NOT_FOUND, str(exc)) from exc
except ValidationError as exc:
raise HTTPException(status.HTTP_422_UNPROCESSABLE_ENTITY, str(exc)) from exc
return PlanSyncResponse.from_domain(result)
backend/app/services/allocation_service.py
+92 -5
View File
@@ -624,10 +624,38 @@ class _Ledger:
self._existing_lookup
)
self._reserve_pinned_days()
self._reserve_preschool_days()
# -- setup ----------------------------------------------------------
def _reserve_pinned_days(self) -> None:
"""§Task 4 — debit a pinned day's cost off the top, before any
window is funded, exactly like `_reserve_preschool_days` below
reserves preschool closure days. A pinned `planned_leave_day` is a
hard, already-spent fact regardless of which mode loaded this input
(see `AllocationInputLoader.load`'s `only_pinned_existing`, which
never nets a pinned row out of the pools/vacation budgets itself —
this is the one place that happens). `_existing_lookup`/
`_persisted_lookup` (built above from the same `existing_planned`)
separately keep the Allocator from ever reassigning a pinned day's
type or parent — this method only handles the resource debit.
"""
for p in self._input.existing_planned:
if not p.is_pinned:
continue
if p.day_type == "vacation":
self._spend_vacation(p.parent_personal_data_uuid, p.date)
continue
pool = self._pools.get((p.parent_personal_data_uuid, p.child_id))
if pool is None:
continue
if p.day_type == "sjukpenning" and pool.sjukpenning_left > 0:
pool.sjukpenning_left -= 1
elif p.day_type == "lagsta" and pool.lagsta_left > 0:
pool.lagsta_left -= 1
def _reserve_preschool_days(self) -> None:
for child in self._input.children:
reserve = _preschool_reserve_days(child, self._input.today, self._input.horizon_end)
@@ -1854,7 +1882,27 @@ class AllocationInputLoader:
self._leave_wish_repo = leave_wish_repo
self._day_override_repo = day_override_repo
async def load(self, workspace_id: UUID, today: date | None = None) -> AllocationInput:
async def load(
self,
workspace_id: UUID,
today: date | None = None,
*,
only_pinned_existing: bool = False,
) -> AllocationInput:
"""Assemble the pure `AllocationInput`.
`only_pinned_existing` (§Task 4, see `PlanningEngineService.propose`)
restricts `existing_planned` — and everything derived from it here
(pool/vacation netting) — to rows that must stay immovable: pinned
rows (`is_pinned`) and ad-hoc rows with no `wish_id`. Every other
(unpinned, wish-attributed) row is left out entirely, so the
allocator treats that date as open and re-derives it fresh. Pinned
rows are *never* netted out of the pools/vacation budgets here even
when `only_pinned_existing` is False — the pure `_Ledger` does that
itself (`_reserve_pinned_days`), matching how it reserves preschool
planning days off the top, so the debit happens in one place
regardless of which mode loaded the input.
"""
today = today or date.today()
ws = await self._workspace_repo.get_by_id(workspace_id)
@@ -1885,7 +1933,16 @@ class AllocationInputLoader:
planned_for_child = await self._planned_leave_repo.list_by_workspace_child(
workspace_id, child.id
)
existing_planned.extend(planned_for_child)
if only_pinned_existing:
# Hide unpinned, wish-attributed rows entirely -- see the
# docstring above. Pinned rows and ad-hoc (no wish_id) rows
# stay visible/immovable either way.
visible_for_child = [
d for d in planned_for_child if d.is_pinned or d.wish_id is None
]
else:
visible_for_child = planned_for_child
existing_planned.extend(visible_for_child)
for parent in parent_entities:
latest = await self._parental_leave_repo.get_latest_by_parent(parent.uuid, child.id)
@@ -1910,11 +1967,16 @@ class AllocationInputLoader:
# Net out days already committed locally (planned_leave_day)
# that an FK snapshot (if any) doesn't yet reflect, so the pure
# allocator never double-books or over-spends a resource.
# Pinned days are excluded here — the pure `_Ledger` debits
# those itself (`_reserve_pinned_days`), so this loader-side
# netting only ever covers non-pinned commitments (whichever
# of those `visible_for_child` still includes).
committed = [
d
for d in planned_for_child
for d in visible_for_child
if d.parent_personal_data_uuid == parent.uuid
and d.day_type in ("sjukpenning", "lagsta")
and not d.is_pinned
]
sjuk_committed = sum(1 for d in committed if d.day_type == "sjukpenning")
lagsta_committed = sum(1 for d in committed if d.day_type == "lagsta")
@@ -1928,12 +1990,16 @@ class AllocationInputLoader:
horizon_end = _compute_horizon_end(today, children, wishes)
def _already_used_vacation(parent_uuid: UUID, year: int) -> int:
# Pinned vacation days are excluded here for the same reason as
# the FP pools above: `_Ledger._reserve_pinned_days` debits them
# itself.
return sum(
1
for d in existing_planned
if d.parent_personal_data_uuid == parent_uuid
and d.day_type == "vacation"
and d.date.year == year
and not d.is_pinned
)
vacation: dict[tuple[UUID, int], VacationBudget] = {}
@@ -2027,6 +2093,27 @@ class PlanningEngineService:
self._loader = loader
self._allocator = allocator or Allocator()
async def propose(self, workspace_id: UUID, today: date | None = None) -> PlanProposal:
alloc_input = await self._loader.load(workspace_id, today=today)
async def propose(
self,
workspace_id: UUID,
today: date | None = None,
*,
only_pinned_existing: bool = False,
) -> PlanProposal:
"""Load fresh input and run the pure allocator over it.
`only_pinned_existing` (used by `PlanningService.sync_plan`, §Task 4)
makes the loader hide every *unpinned*, wish-attributed
`planned_leave_day` row from `AllocationInput.existing_planned` (and
stop netting it out of the pools/vacation budgets) so the allocator
re-derives that portion of the plan from scratch instead of treating
it as an immovable fact — this is what lets a re-plan legitimately
move/retype previously-auto-placed days. Pinned days and ad-hoc
(`wish_id is None`) days are unaffected either way: always visible,
always immovable. `POST /api/plan/propose` never sets this — it
keeps its historical, unconditional behaviour.
"""
alloc_input = await self._loader.load(
workspace_id, today=today, only_pinned_existing=only_pinned_existing
)
return self._allocator.propose(alloc_input)
backend/app/services/planning_service.py
+206
View File
@@ -10,6 +10,7 @@ from app.config import settings
from app.entities.planned_leave import PlannedLeaveDay
from app.repositories.parental_leave_repository import ParentalLeaveRepository
from app.repositories.planned_leave_repository import PlannedLeaveRepository
from app.services.allocation_service import PlanningEngineService, PlanProposal
from app.services.analytics_service import AnalyticsService
from app.utils.errors import DubbeldagError, NotFoundError, ValidationError
@@ -87,16 +88,63 @@ class PlanApplyResult:
rejections: list[Rejection] = field(default_factory=list)
# ---------------------------------------------------------------------------
# POST /api/plan/sync — propose + apply in one step, with a structured diff.
# ---------------------------------------------------------------------------
@dataclass(frozen=True)
class PlanChange:
"""One (child, parent, date) slot's delta from `sync_plan`'s pre-write
snapshot to what's actually persisted after the call. Structured data
only — house style is `message_key` + `data`, never backend-rendered
prose, so the frontend composes the toast copy ("Sommar 2028 flyttades 2
dagar") itself. `kind`:
- 'added' — no row existed for (child, parent, date); one does now.
- 'removed' — a row existed and no longer does (old_* populated, new_*
all `None`).
- 'retyped' — the row survived but `day_type`/`is_dubbeldag` changed.
- 'reattributed' — same day and type, but `wish_id` changed (the day now
belongs to a different period).
"""
kind: str
date: date
parent_personal_data_uuid: UUID
child_id: UUID
old_day_type: str | None
new_day_type: str | None
old_is_dubbeldag: bool | None
new_is_dubbeldag: bool | None
old_wish_id: UUID | None
new_wish_id: UUID | None
@dataclass(frozen=True)
class PlanSyncResult:
proposal: PlanProposal
apply_result: PlanApplyResult
changes: list[PlanChange] = field(default_factory=list)
class PlanningService:
def __init__(
self,
repo: PlannedLeaveRepository,
analytics_service: AnalyticsService,
parental_leave_repo: ParentalLeaveRepository,
planning_engine_service: PlanningEngineService | None = None,
) -> None:
self._repo = repo
self._analytics = analytics_service
self._parental_leave_repo = parental_leave_repo
# Only needed by `sync_plan` (POST /api/plan/sync) — optional so
# existing callers (scripts/seed_demo.py, older fixtures) that only
# ever use the per-day CRUD/`apply_proposed_days` paths don't need to
# wire up a `PlanningEngineService` they'll never use.
self._engine = planning_engine_service
async def add_planned_day(
self,
@@ -413,6 +461,164 @@ class PlanningService:
replaced_count=len(to_replace),
)
# -- sync (POST /api/plan/sync) ---------------------------------------
#
# Recomputes and persists a workspace's plan in one step, then reports
# what changed. Runs `PlanningEngineService.propose(..., only_pinned_
# existing=True)`, which hides every *unpinned* wish-attributed
# `planned_leave_day` row from the allocator (see that flag's docstring
# in `allocation_service.py`) so the auto-planned portion of the plan is
# re-derived from scratch each call — this is what lets a re-plan
# legitimately move or retype previously-auto-placed days, while a
# pinned day (fed to the allocator unchanged, and debited off the top by
# `_Ledger._reserve_pinned_days`) can never be reassigned or overwritten.
#
# Pinned days therefore never appear in `proposal.days` in the first
# place (the allocator never emits anything for an already-occupied
# key) -- "no collision ever arises" is a property of the allocator, not
# something this method has to special-case in the write path. The one
# extra step this method DOES own is deleting rows that fall out of the
# freshly-computed plan entirely (a wish's window moved away from a date
# it used to occupy, or the wish was deleted) -- `apply_proposed_days`
# only ever creates/replaces/updates keys present in its own batch, so
# nothing else would ever clean those up.
async def sync_plan(self, workspace_id: UUID, today: date | None = None) -> PlanSyncResult:
"""Propose, persist, and diff — the single call the direct-
manipulation calendar UX needs (no more separate propose/apply
steps). The diff (`changes`) is computed from a pre-write snapshot
read here, before any write, compared against what's actually
persisted afterwards — never reconstructed from the client side or
guessed from `apply_proposed_days`'s own (coarser) counters.
"""
if self._engine is None:
raise ValidationError("PlanningService.sync_plan requires a PlanningEngineService")
before = await self._repo.list(workspace_id=workspace_id)
before_by_key: dict[tuple[UUID, UUID, date], PlannedLeaveDay] = {
(d.child_id, d.parent_personal_data_uuid, d.date): d for d in before
}
pinned_keys = {key for key, d in before_by_key.items() if d.is_pinned}
proposal = await self._engine.propose(workspace_id, today=today, only_pinned_existing=True)
batch = [
ProposedDayInput(
date=d.date,
parent_personal_data_uuid=d.parent_personal_data_uuid,
child_id=d.child_id,
day_type=d.day_type,
is_dubbeldag=d.is_dubbeldag,
wish_id=d.wish_id,
is_pinned=False,
)
for d in proposal.days
if d.day_type != "free"
and d.child_id is not None
# Defense-in-depth only: the allocator never proposes anything
# for an already-occupied key (pinned rows stay in
# `existing_planned` — see `only_pinned_existing` above), so this
# should never actually filter anything out.
and (d.child_id, d.parent_personal_data_uuid, d.date) not in pinned_keys
]
apply_result = await self.apply_proposed_days(
workspace_id=workspace_id, days=batch, replace=True
)
changes: list[PlanChange] = []
if apply_result.accepted:
batch_keys = {(d.child_id, d.parent_personal_data_uuid, d.date) for d in batch}
# Rows the freshly-computed plan no longer accounts for at all:
# unpinned, wish-attributed, and absent from this run's batch —
# e.g. the wish was deleted, or its window moved elsewhere.
# `apply_proposed_days` never deletes a key it wasn't handed, so
# this is the one bit of write logic sync owns beyond it.
stale = [
row
for key, row in before_by_key.items()
if not row.is_pinned and row.wish_id is not None and key not in batch_keys
]
for row in stale:
await self._repo.delete(row.id)
after = await self._repo.list(workspace_id=workspace_id)
after_by_key: dict[tuple[UUID, UUID, date], PlannedLeaveDay] = {
(d.child_id, d.parent_personal_data_uuid, d.date): d for d in after
}
for key in sorted(
set(before_by_key) | set(after_by_key),
key=lambda k: (k[2], str(k[0]), str(k[1])),
):
child_id, parent_uuid, d = key
old = before_by_key.get(key)
new = after_by_key.get(key)
if old is None and new is not None:
changes.append(
PlanChange(
kind="added",
date=d,
parent_personal_data_uuid=parent_uuid,
child_id=child_id,
old_day_type=None,
new_day_type=new.day_type,
old_is_dubbeldag=None,
new_is_dubbeldag=new.is_dubbeldag,
old_wish_id=None,
new_wish_id=new.wish_id,
)
)
elif old is not None and new is None:
changes.append(
PlanChange(
kind="removed",
date=d,
parent_personal_data_uuid=parent_uuid,
child_id=child_id,
old_day_type=old.day_type,
new_day_type=None,
old_is_dubbeldag=old.is_dubbeldag,
new_is_dubbeldag=None,
old_wish_id=old.wish_id,
new_wish_id=None,
)
)
elif old is not None and new is not None:
if old.day_type != new.day_type or old.is_dubbeldag != new.is_dubbeldag:
changes.append(
PlanChange(
kind="retyped",
date=d,
parent_personal_data_uuid=parent_uuid,
child_id=child_id,
old_day_type=old.day_type,
new_day_type=new.day_type,
old_is_dubbeldag=old.is_dubbeldag,
new_is_dubbeldag=new.is_dubbeldag,
old_wish_id=old.wish_id,
new_wish_id=new.wish_id,
)
)
elif old.wish_id != new.wish_id:
changes.append(
PlanChange(
kind="reattributed",
date=d,
parent_personal_data_uuid=parent_uuid,
child_id=child_id,
old_day_type=old.day_type,
new_day_type=new.day_type,
old_is_dubbeldag=old.is_dubbeldag,
new_is_dubbeldag=new.is_dubbeldag,
old_wish_id=old.wish_id,
new_wish_id=new.wish_id,
)
)
return PlanSyncResult(proposal=proposal, apply_result=apply_result, changes=changes)
async def _validate_dubbeldagar_batch(
self,
pending: list[ProposedDayInput],
backend/tests/conftest.py
+35 -1
View File
@@ -32,6 +32,7 @@ from app.repositories_impl.sqlite.workspace_repository import (
SQLiteWorkspaceInvitationRepository,
SQLiteWorkspaceRepository,
)
from app.services.allocation_service import AllocationInputLoader, PlanningEngineService
from app.services.analytics_service import AnalyticsService
from app.services.auth_service import AuthService
from app.services.child_service import ChildService
@@ -262,10 +263,43 @@ async def analytics_service(
)
@pytest_asyncio.fixture
async def allocation_input_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,
day_override_repo: SQLiteDayOverrideRepository,
) -> 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,
day_override_repo=day_override_repo,
)
@pytest_asyncio.fixture
async def planning_engine_service(
allocation_input_loader: AllocationInputLoader,
) -> PlanningEngineService:
return PlanningEngineService(allocation_input_loader)
@pytest_asyncio.fixture
async def planning_service(
planned_leave_repo: SQLitePlannedLeaveRepository,
analytics_service: AnalyticsService,
parental_leave_repo: SQLiteParentalLeaveRepository,
planning_engine_service: PlanningEngineService,
) -> PlanningService:
return PlanningService(planned_leave_repo, analytics_service, parental_leave_repo)
return PlanningService(
planned_leave_repo, analytics_service, parental_leave_repo, planning_engine_service
)
backend/tests/unit/test_planning_service_sync.py
+342
View File
@@ -0,0 +1,342 @@
"""Tests for `PlanningService.sync_plan` (POST /api/plan/sync).
Covers the Task 4 contract: propose+apply in one step, a pinned
`planned_leave_day` is never retyped/overwritten by a re-plan even when the
fresh proposal would have chosen differently for that date, the `changes`
diff surfaces each of its four kinds, sync is idempotent on an unchanged
plan, and `/propose` stays read-only (persists nothing).
"""
from datetime import date, datetime
from uuid import uuid4
from app.entities.leave_wish import LeaveWish
from app.entities.parental_leave import ParentalLeaveFetch, TierData
from app.entities.personal_data import PersonalData
from app.entities.planned_leave import PlannedLeaveDay
from app.entities.vacation_days import VacationDays
from app.repositories_impl.sqlite.leave_wish_repository import SQLiteLeaveWishRepository
from app.repositories_impl.sqlite.parental_leave_repository import SQLiteParentalLeaveRepository
from app.repositories_impl.sqlite.personal_data_repository import SQLitePersonalDataRepository
from app.repositories_impl.sqlite.planned_leave_repository import SQLitePlannedLeaveRepository
from app.repositories_impl.sqlite.vacation_days_repository import SQLiteVacationDaysRepository
from app.services.allocation_service import PlanningEngineService
from app.services.planning_service import PlanningService
from tests.conftest import SeededIds
def _make_wish(workspace_id, child_id, **overrides) -> LeaveWish:
base = dict(
id=uuid4(),
workspace_id=workspace_id,
child_id=child_id,
name="Test wish",
window_start_month=2,
window_start_day=1,
window_end_month=2,
window_end_day=7,
target_weeks=1,
recurrence="once",
first_year=2027,
last_year=None,
priority=1,
parent_mode="either",
parent_personal_data_uuid=None,
window_mode="fixed",
created_at=datetime.now(),
updated_at=datetime.now(),
)
base.update(overrides)
return LeaveWish(**base)
def _make_fetch(personal_data_uuid, workspace_id, child_id, fetched_at, sjuk_left, lagsta_left):
tier = TierData(
sjukpenning_total=390,
sjukpenning_used=390 - sjuk_left,
sjukpenning_left=sjuk_left,
lagsta_niva_total=90,
lagsta_niva_used=90 - lagsta_left,
lagsta_niva_left=lagsta_left,
)
return ParentalLeaveFetch(
id=uuid4(),
personal_data_uuid=personal_data_uuid,
workspace_id=workspace_id,
child_id=child_id,
source="manual",
fetched_at=fetched_at,
total_days=480,
days_used=480 - sjuk_left - lagsta_left,
days_left=sjuk_left + lagsta_left,
reserved_days_total=90,
reserved_days_left=90,
transferable_days_left=sjuk_left + lagsta_left,
dubbeldagar_left=60,
tier_data=tier,
created_at=fetched_at,
updated_at=fetched_at,
)
class TestSyncNeverOverwritesAPinnedDay:
async def test_pinned_day_survives_even_though_fresh_plan_would_differ(
self,
planning_service: PlanningService,
planned_leave_repo: SQLitePlannedLeaveRepository,
vacation_repo: SQLiteVacationDaysRepository,
leave_wish_repo: SQLiteLeaveWishRepository,
seeded_ids: SeededIds,
) -> None:
ws, child, parent = (
seeded_ids.workspace_id,
seeded_ids.child_id,
seeded_ids.personal_data_uuid,
)
# No vacation budget this year -> the monthly-salaried parent's
# tier-3 policy (vacation first) falls through to FP, sjukpenning
# preferred -- so a free interior day in this window would be
# funded as 'sjukpenning', never 'lagsta'.
await vacation_repo.create(
VacationDays(
id=uuid4(), personal_data_uuid=parent, workspace_id=ws, year=2027, total_days=0
)
)
await leave_wish_repo.create(_make_wish(ws, child))
pinned_date = date(2027, 2, 3) # a Wednesday inside the wish's window
pinned = await planned_leave_repo.create(
PlannedLeaveDay(
id=uuid4(),
workspace_id=ws,
child_id=child,
parent_personal_data_uuid=parent,
date=pinned_date,
day_type="lagsta", # deliberately NOT what a fresh plan would pick
is_pinned=True,
created_at=datetime.now(),
updated_at=datetime.now(),
)
)
result = await planning_service.sync_plan(ws, today=date(2027, 1, 1))
assert result.apply_result.accepted is True
rows = await planned_leave_repo.list_by_workspace_child(ws, child)
pinned_after = next(r for r in rows if r.date == pinned_date)
assert pinned_after.id == pinned.id
assert pinned_after.day_type == "lagsta"
assert pinned_after.is_pinned is True
# Every OTHER day in the window was free to be funded normally.
other_days = [r for r in rows if r.date != pinned_date]
assert other_days, "expected the rest of the window to be funded"
assert all(r.day_type == "sjukpenning" for r in other_days)
# The pinned date never shows up as a changed slot.
assert all(c.date != pinned_date for c in result.changes)
class TestSyncChangeKinds:
async def test_added_then_removed_when_the_wish_shrinks(
self,
planning_service: PlanningService,
leave_wish_repo: SQLiteLeaveWishRepository,
seeded_ids: SeededIds,
) -> None:
"""A wish that still exists but now claims fewer dates: the dates it
drops (still `wish_id`-attributed, just no longer produced by this
run's batch) must be reported (and written) as 'removed' -- distinct
from a deleted wish, whose rows the DB itself detaches via
`ON DELETE SET NULL` (making them ad-hoc, permanently kept) before
`sync_plan` ever sees them.
"""
ws, child, _parent = (
seeded_ids.workspace_id,
seeded_ids.child_id,
seeded_ids.personal_data_uuid,
)
wish = await leave_wish_repo.create(_make_wish(ws, child, name="A", window_end_day=7))
# -- first sync: pure 'added' (Feb 1-7) ---------------------------
result_1 = await planning_service.sync_plan(ws, today=date(2027, 1, 1))
assert result_1.apply_result.accepted is True
assert result_1.changes, "expected at least one 'added' change"
assert {c.kind for c in result_1.changes} == {"added"}
# -- idempotent re-sync: no changes -------------------------------
result_2 = await planning_service.sync_plan(ws, today=date(2027, 1, 1))
assert result_2.apply_result.accepted is True
assert result_2.changes == []
# -- shrink the window (Feb 1-4) -> the dropped tail is 'removed' -
wish.window_end_day = 4
await leave_wish_repo.update(wish.id, wish)
result_3 = await planning_service.sync_plan(ws, today=date(2027, 1, 1))
assert result_3.apply_result.accepted is True
assert result_3.changes, "expected 'removed' changes once the window shrinks"
assert {c.kind for c in result_3.changes} == {"removed"}
assert all(c.date > date(2027, 2, 4) for c in result_3.changes)
async def test_retyped_when_the_underlying_pool_changes(
self,
planning_service: PlanningService,
personal_data_repo: SQLitePersonalDataRepository,
parental_leave_repo: SQLiteParentalLeaveRepository,
leave_wish_repo: SQLiteLeaveWishRepository,
seeded_ids: SeededIds,
) -> None:
ws, child = seeded_ids.workspace_id, seeded_ids.child_id
# An hourly-salaried parent prefers FP (sjukpenning first) over
# vacation for tier-3 interior days, so this is deterministic
# regardless of any assumed vacation default.
parent = await personal_data_repo.create(
PersonalData(uuid=uuid4(), workspace_id=ws, name="Hourly Parent", salary_type="hourly")
)
# `parent_mode="specific"` -- with a second (seeded) parent now also
# in the workspace, an "either" wish would default to whichever
# parent sorts first, not necessarily this one.
await leave_wish_repo.create(
_make_wish(ws, child, parent_mode="specific", parent_personal_data_uuid=parent.uuid)
)
await parental_leave_repo.create(
_make_fetch(
parent.uuid, ws, child, datetime(2027, 1, 1), sjuk_left=100, lagsta_left=100
)
)
result_1 = await planning_service.sync_plan(ws, today=date(2027, 1, 1))
assert result_1.apply_result.accepted is True
assert any(c.new_day_type == "sjukpenning" for c in result_1.changes)
# sjukpenning now exhausted -> the SAME dates must fall back to lagsta.
await parental_leave_repo.create(
_make_fetch(parent.uuid, ws, child, datetime(2027, 1, 2), sjuk_left=0, lagsta_left=100)
)
result_2 = await planning_service.sync_plan(ws, today=date(2027, 1, 1))
assert result_2.apply_result.accepted is True
assert result_2.changes, "expected 'retyped' changes once sjukpenning ran out"
assert {c.kind for c in result_2.changes} == {"retyped"}
assert all(
c.old_day_type == "sjukpenning" and c.new_day_type == "lagsta" for c in result_2.changes
)
async def test_reattributed_when_a_day_moves_to_a_different_wish(
self,
planning_service: PlanningService,
planned_leave_repo: SQLitePlannedLeaveRepository,
vacation_repo: SQLiteVacationDaysRepository,
leave_wish_repo: SQLiteLeaveWishRepository,
seeded_ids: SeededIds,
) -> None:
ws, child, parent = (
seeded_ids.workspace_id,
seeded_ids.child_id,
seeded_ids.personal_data_uuid,
)
await vacation_repo.create(
VacationDays(
id=uuid4(), personal_data_uuid=parent, workspace_id=ws, year=2027, total_days=0
)
)
# `old_wish`'s own window doesn't cover `target_date` at all, so it
# never competes for that key -- it just needs to be a real, still-
# existing wish so the persisted row's `wish_id` stays non-null
# (unlike deleting it, which `ON DELETE SET NULL` would turn into an
# ad-hoc, permanently-kept row before `sync_plan` ever sees it).
old_wish = await leave_wish_repo.create(
_make_wish(
ws,
child,
name="Old",
window_start_month=3,
window_start_day=1,
window_end_month=3,
window_end_day=7,
)
)
new_wish = await leave_wish_repo.create(
_make_wish(
ws,
child,
name="New",
window_start_month=2,
window_start_day=3,
window_end_month=2,
window_end_day=3,
target_days=1,
)
)
target_date = date(2027, 2, 3)
# A row identical to what `new_wish` would compute for this date
# (same date/type), but still attributed to `old_wish`.
await planned_leave_repo.create(
PlannedLeaveDay(
id=uuid4(),
workspace_id=ws,
child_id=child,
parent_personal_data_uuid=parent,
date=target_date,
day_type="sjukpenning",
wish_id=old_wish.id,
is_pinned=False,
created_at=datetime.now(),
updated_at=datetime.now(),
)
)
result = await planning_service.sync_plan(ws, today=date(2027, 1, 1))
assert result.apply_result.accepted is True
by_date = {c.date: c for c in result.changes}
assert target_date in by_date
changed = by_date[target_date]
assert changed.kind == "reattributed"
assert changed.old_wish_id == old_wish.id
assert changed.new_wish_id == new_wish.id
assert changed.old_day_type == changed.new_day_type == "sjukpenning"
class TestSyncIdempotent:
async def test_unchanged_plan_produces_no_changes(
self,
planning_service: PlanningService,
leave_wish_repo: SQLiteLeaveWishRepository,
seeded_ids: SeededIds,
) -> None:
ws, child, _parent = (
seeded_ids.workspace_id,
seeded_ids.child_id,
seeded_ids.personal_data_uuid,
)
await leave_wish_repo.create(_make_wish(ws, child))
first = await planning_service.sync_plan(ws, today=date(2027, 1, 1))
assert first.apply_result.accepted is True
second = await planning_service.sync_plan(ws, today=date(2027, 1, 1))
assert second.apply_result.accepted is True
assert second.changes == []
class TestProposeStaysReadOnly:
async def test_propose_never_writes_a_row(
self,
planning_engine_service: PlanningEngineService,
planned_leave_repo: SQLitePlannedLeaveRepository,
leave_wish_repo: SQLiteLeaveWishRepository,
seeded_ids: SeededIds,
) -> None:
ws, child, _parent = (
seeded_ids.workspace_id,
seeded_ids.child_id,
seeded_ids.personal_data_uuid,
)
await leave_wish_repo.create(_make_wish(ws, child))
proposal = await planning_engine_service.propose(ws, today=date(2027, 1, 1))
assert proposal.days, "expected the proposal to compute something"
rows = await planned_leave_repo.list_by_workspace_child(ws, child)
assert rows == []
backend/tests/unit/test_services/test_allocation_service.py
+94
View File
@@ -1882,3 +1882,97 @@ class TestCase28FlexibleWishStillMovesWhenItsYearIsNotAlreadySatisfied:
# window's span) but never re-claims the fixed wish's date.
assert date(2027, 3, 23) in flex_dates
assert date(2027, 3, 25) not in flex_dates
# ---------------------------------------------------------------------------
# §Task 4 — pinned days: fixed, and debited off the top before any window
# is funded (mirrors `_reserve_preschool_days`'s precedent).
# ---------------------------------------------------------------------------
class TestPinnedDayIsNeverRetypedAndDebitsThePool:
def test_pinned_day_keeps_its_type_and_the_pool_reflects_the_debit(self) -> None:
child = _child(birth_date=date(2020, 1, 1))
parent = _parent(salary_type="monthly")
pinned_date = date(2027, 2, 10) # a Wednesday inside the wish's window
pinned = PlannedLeaveDay(
id=uuid4(),
workspace_id=uuid4(),
child_id=child.id,
parent_personal_data_uuid=parent.uuid,
date=pinned_date,
day_type="sjukpenning",
is_pinned=True,
)
# A wish covering exactly this one date -- if the pinned day were
# NOT fixed, this monthly-salaried parent's tier-3 policy (vacation
# first) would fund it as 'vacation', not 'sjukpenning'.
wish = _wish(
child_id=child.id,
name="Single day",
start=(2, 10),
end=(2, 10),
weeks=1,
first_year=2027,
)
alloc_input = _build_input(
today=date(2027, 1, 1),
horizon_end=date(2027, 12, 31),
planning_mode="solo",
children=[child],
parents=[parent],
pools={(parent.uuid, child.id): DayPool(5, 5, 60)},
vacation={(parent.uuid, 2027): VacationBudget(25, 0, 0)},
wishes=[wish],
existing_planned=[pinned],
)
proposal = Allocator().propose(alloc_input)
# Never re-proposed / retyped for that date.
assert not any(
d.date == pinned_date and d.parent_personal_data_uuid == parent.uuid
for d in proposal.days
)
# Its cost was debited off the top, before any window was funded --
# the pool reflects one less sjukpenning day, not the untouched 5.
pool_after = proposal.balances_after.pools_after[(parent.uuid, child.id)]
assert pool_after.sjukpenning_left == 4
assert pool_after.lagsta_left == 5
# The wish's own outcome still reports the pinned day as funded
# (via cost_breakdown), just not as a newly-proposed day.
outcome = proposal.outcomes[0]
assert outcome.cost_breakdown["sjukpenning"] == 1
assert outcome.shortfall_days == 0
def test_pinned_vacation_day_debits_the_vacation_budget(self) -> None:
child = _child(birth_date=date(2020, 1, 1))
parent = _parent(salary_type="monthly")
pinned_date = date(2027, 2, 10)
pinned = PlannedLeaveDay(
id=uuid4(),
workspace_id=uuid4(),
child_id=child.id,
parent_personal_data_uuid=parent.uuid,
date=pinned_date,
day_type="vacation",
is_pinned=True,
)
alloc_input = _build_input(
today=date(2027, 1, 1),
horizon_end=date(2027, 12, 31),
planning_mode="solo",
children=[child],
parents=[parent],
pools={(parent.uuid, child.id): DayPool(5, 5, 60)},
vacation={(parent.uuid, 2027): VacationBudget(25, 0, 0)},
wishes=[],
existing_planned=[pinned],
)
proposal = Allocator().propose(alloc_input)
budget_after = proposal.balances_after.vacation_after[(parent.uuid, 2027)]
assert budget_after.already_used == 1
assert budget_after.available == 24