diff --git a/app/services/minimum_workload.py b/app/services/minimum_workload.py index 2fa3b75..284bbea 100644 --- a/app/services/minimum_workload.py +++ b/app/services/minimum_workload.py @@ -1,15 +1,20 @@ -"""MinimumWorkload service — CRUD and validation helpers. +"""MinimumWorkload service — CRUD, workload computation and validation. -BE-CAL-004: user-level workload config read/write + future validation entry point. +BE-CAL-004: user-level workload config read/write. +BE-CAL-007: workload warning rules — compute actual scheduled minutes across + daily/weekly/monthly/yearly periods and compare against thresholds. """ from __future__ import annotations import copy +from datetime import date, timedelta from typing import Optional +from sqlalchemy import func as sa_func from sqlalchemy.orm import Session +from app.models.calendar import SlotStatus, SlotType, TimeSlot from app.models.minimum_workload import ( DEFAULT_WORKLOAD_CONFIG, CATEGORIES, @@ -21,6 +26,18 @@ from app.schemas.calendar import ( MinimumWorkloadUpdate, WorkloadWarningItem, ) +from app.services.plan_slot import get_virtual_slots_for_date + + +# Slot types that map to workload categories. "system" is excluded. +_SLOT_TYPE_TO_CATEGORY = { + SlotType.WORK: "work", + SlotType.ON_CALL: "on_call", + SlotType.ENTERTAINMENT: "entertainment", +} + +# Statuses that should NOT count towards workload (cancelled / failed slots). +_EXCLUDED_STATUSES = {SlotStatus.SKIPPED, SlotStatus.ABORTED} # --------------------------------------------------------------------------- @@ -96,7 +113,146 @@ def replace_workload_config( # --------------------------------------------------------------------------- -# Validation entry point (BE-CAL-007 will flesh this out) +# Workload computation (BE-CAL-007) +# --------------------------------------------------------------------------- + +def _date_range_for_period( + period: str, + reference_date: date, +) -> tuple[date, date]: + """Return inclusive ``(start, end)`` date bounds for *period* containing *reference_date*. + + - daily → just the reference date itself + - weekly → ISO week (Mon–Sun) containing the reference date + - monthly → calendar month containing the reference date + - yearly → calendar year containing the reference date + """ + if period == "daily": + return reference_date, reference_date + + if period == "weekly": + # ISO weekday: Monday=1 … Sunday=7 + start = reference_date - timedelta(days=reference_date.weekday()) # Monday + end = start + timedelta(days=6) # Sunday + return start, end + + if period == "monthly": + start = reference_date.replace(day=1) + # Last day of month + if reference_date.month == 12: + end = reference_date.replace(month=12, day=31) + else: + end = reference_date.replace(month=reference_date.month + 1, day=1) - timedelta(days=1) + return start, end + + if period == "yearly": + start = reference_date.replace(month=1, day=1) + end = reference_date.replace(month=12, day=31) + return start, end + + raise ValueError(f"Unknown period: {period}") + + +def _sum_real_slots( + db: Session, + user_id: int, + start_date: date, + end_date: date, +) -> dict[str, int]: + """Sum ``estimated_duration`` of real (materialized) slots by category. + + Returns ``{"work": N, "on_call": N, "entertainment": N}`` with minutes. + Slots with status in ``_EXCLUDED_STATUSES`` or ``slot_type=system`` are skipped. + """ + excluded = [s.value for s in _EXCLUDED_STATUSES] + + rows = ( + db.query( + TimeSlot.slot_type, + sa_func.coalesce(sa_func.sum(TimeSlot.estimated_duration), 0), + ) + .filter( + TimeSlot.user_id == user_id, + TimeSlot.date >= start_date, + TimeSlot.date <= end_date, + TimeSlot.status.notin_(excluded), + TimeSlot.slot_type != SlotType.SYSTEM.value, + ) + .group_by(TimeSlot.slot_type) + .all() + ) + + totals: dict[str, int] = {"work": 0, "on_call": 0, "entertainment": 0} + for slot_type_val, total in rows: + # slot_type_val may be an enum or a raw string + if hasattr(slot_type_val, "value"): + slot_type_val = slot_type_val.value + cat = _SLOT_TYPE_TO_CATEGORY.get(SlotType(slot_type_val)) + if cat: + totals[cat] += int(total) + return totals + + +def _sum_virtual_slots( + db: Session, + user_id: int, + start_date: date, + end_date: date, +) -> dict[str, int]: + """Sum ``estimated_duration`` of virtual (plan-generated, not-yet-materialized) + slots by category across a date range. + + Iterates day by day — acceptable because periods are at most a year and + the function only queries plans once per day. + """ + totals: dict[str, int] = {"work": 0, "on_call": 0, "entertainment": 0} + current = start_date + while current <= end_date: + for vs in get_virtual_slots_for_date(db, user_id, current): + slot_type = vs["slot_type"] + if hasattr(slot_type, "value"): + slot_type = slot_type.value + cat = _SLOT_TYPE_TO_CATEGORY.get(SlotType(slot_type)) + if cat: + totals[cat] += vs["estimated_duration"] + current += timedelta(days=1) + return totals + + +def compute_scheduled_minutes( + db: Session, + user_id: int, + reference_date: date, +) -> dict[str, dict[str, int]]: + """Compute total scheduled minutes for each period containing *reference_date*. + + Returns the canonical shape consumed by :func:`check_workload_warnings`:: + + { + "daily": {"work": N, "on_call": N, "entertainment": N}, + "weekly": { ... }, + "monthly": { ... }, + "yearly": { ... }, + } + + Includes both real (materialized) and virtual (plan-generated) slots. + """ + result: dict[str, dict[str, int]] = {} + + for period in PERIODS: + start, end = _date_range_for_period(period, reference_date) + real = _sum_real_slots(db, user_id, start, end) + virtual = _sum_virtual_slots(db, user_id, start, end) + result[period] = { + cat: real.get(cat, 0) + virtual.get(cat, 0) + for cat in CATEGORIES + } + + return result + + +# --------------------------------------------------------------------------- +# Warning comparison # --------------------------------------------------------------------------- def check_workload_warnings( @@ -106,14 +262,12 @@ def check_workload_warnings( ) -> list[WorkloadWarningItem]: """Compare *scheduled_minutes* against the user's configured thresholds. - ``scheduled_minutes`` has the same shape as the config: + ``scheduled_minutes`` has the same shape as the config:: + {"daily": {"work": N, ...}, "weekly": {...}, ...} Returns a list of warnings for every (period, category) where the scheduled total is below the minimum. An empty list means no warnings. - - This is the entry point that BE-CAL-007 and the calendar API endpoints - will call. """ config = get_workload_config(db, user_id) warnings: list[WorkloadWarningItem] = [] @@ -142,3 +296,23 @@ def check_workload_warnings( )) return warnings + + +# --------------------------------------------------------------------------- +# High-level convenience: compute + check in one call (BE-CAL-007) +# --------------------------------------------------------------------------- + +def get_workload_warnings_for_date( + db: Session, + user_id: int, + reference_date: date, +) -> list[WorkloadWarningItem]: + """One-shot helper: compute scheduled minutes for *reference_date* and + return any workload warnings. + + Calendar API endpoints should call this after a create/edit mutation to + include warnings in the response. Warnings are advisory — they do NOT + prevent the operation. + """ + scheduled = compute_scheduled_minutes(db, user_id, reference_date) + return check_workload_warnings(db, user_id, scheduled) diff --git a/app/services/slot_immutability.py b/app/services/slot_immutability.py new file mode 100644 index 0000000..b58146c --- /dev/null +++ b/app/services/slot_immutability.py @@ -0,0 +1,171 @@ +"""Past-slot immutability rules. + +BE-CAL-008: Prevents editing or cancelling slots whose date is in the past. +Also ensures plan-edit and plan-cancel do not retroactively affect +already-materialized past slots. + +Rules: + 1. Editing a past slot (real or virtual) → raise ImmutableSlotError + 2. Cancelling a past slot (real or virtual) → raise ImmutableSlotError + 3. Plan-edit / plan-cancel must NOT retroactively change already-materialized + slots whose date is in the past. The plan_slot.detach_slot_from_plan() + mechanism already ensures this: past materialized slots keep their data. + This module provides guard functions that Calendar API endpoints call + before performing mutations. +""" + +from __future__ import annotations + +from datetime import date +from typing import Optional + +from sqlalchemy.orm import Session + +from app.models.calendar import TimeSlot +from app.services.plan_slot import parse_virtual_slot_id + + +class ImmutableSlotError(Exception): + """Raised when an operation attempts to modify a past slot.""" + + def __init__(self, slot_date: date, operation: str, detail: str = ""): + self.slot_date = slot_date + self.operation = operation + self.detail = detail + msg = ( + f"Cannot {operation} slot on {slot_date.isoformat()}: " + f"past slots are immutable" + ) + if detail: + msg += f" ({detail})" + super().__init__(msg) + + +# --------------------------------------------------------------------------- +# Core guard: date must not be in the past +# --------------------------------------------------------------------------- + +def _assert_not_past(slot_date: date, operation: str, *, today: Optional[date] = None) -> None: + """Raise :class:`ImmutableSlotError` if *slot_date* is before *today*. + + ``today`` defaults to ``date.today()`` when not supplied (allows + deterministic testing). + """ + if today is None: + today = date.today() + if slot_date < today: + raise ImmutableSlotError(slot_date, operation) + + +# --------------------------------------------------------------------------- +# Guards for real (materialized) slots +# --------------------------------------------------------------------------- + +def guard_edit_real_slot( + db: Session, + slot: TimeSlot, + *, + today: Optional[date] = None, +) -> None: + """Raise if the real *slot* is in the past and cannot be edited.""" + _assert_not_past(slot.date, "edit", today=today) + + +def guard_cancel_real_slot( + db: Session, + slot: TimeSlot, + *, + today: Optional[date] = None, +) -> None: + """Raise if the real *slot* is in the past and cannot be cancelled.""" + _assert_not_past(slot.date, "cancel", today=today) + + +# --------------------------------------------------------------------------- +# Guards for virtual (plan-generated) slots +# --------------------------------------------------------------------------- + +def guard_edit_virtual_slot( + virtual_id: str, + *, + today: Optional[date] = None, +) -> None: + """Raise if the virtual slot identified by *virtual_id* is in the past.""" + parsed = parse_virtual_slot_id(virtual_id) + if parsed is None: + raise ValueError(f"Invalid virtual slot id: {virtual_id!r}") + _plan_id, slot_date = parsed + _assert_not_past(slot_date, "edit", today=today) + + +def guard_cancel_virtual_slot( + virtual_id: str, + *, + today: Optional[date] = None, +) -> None: + """Raise if the virtual slot identified by *virtual_id* is in the past.""" + parsed = parse_virtual_slot_id(virtual_id) + if parsed is None: + raise ValueError(f"Invalid virtual slot id: {virtual_id!r}") + _plan_id, slot_date = parsed + _assert_not_past(slot_date, "cancel", today=today) + + +# --------------------------------------------------------------------------- +# Guard for plan-edit / plan-cancel: no retroactive changes to past slots +# --------------------------------------------------------------------------- + +def get_past_materialized_slot_ids( + db: Session, + plan_id: int, + *, + today: Optional[date] = None, +) -> list[int]: + """Return IDs of materialized slots for *plan_id* whose date is in the past. + + Plan-edit and plan-cancel must NOT modify these rows. The caller can + use this list to exclude them from bulk updates, or simply to verify + that no past data was touched. + """ + if today is None: + today = date.today() + rows = ( + db.query(TimeSlot.id) + .filter( + TimeSlot.plan_id == plan_id, + TimeSlot.date < today, + ) + .all() + ) + return [r[0] for r in rows] + + +def guard_plan_edit_no_past_retroaction( + db: Session, + plan_id: int, + *, + today: Optional[date] = None, +) -> list[int]: + """Return past materialized slot IDs that must NOT be modified. + + The caller (plan-edit endpoint) should update only future materialized + slots and skip these. This function is informational — it does not + raise, because the plan itself *can* be edited; the restriction is + that past slots remain untouched. + """ + return get_past_materialized_slot_ids(db, plan_id, today=today) + + +def guard_plan_cancel_no_past_retroaction( + db: Session, + plan_id: int, + *, + today: Optional[date] = None, +) -> list[int]: + """Return past materialized slot IDs that must NOT be cancelled. + + Same semantics as :func:`guard_plan_edit_no_past_retroaction`. + When cancelling a plan, future materialized slots may be removed or + marked cancelled, but past slots remain untouched. + """ + return get_past_materialized_slot_ids(db, plan_id, today=today) diff --git a/tests/test_minimum_workload.py b/tests/test_minimum_workload.py new file mode 100644 index 0000000..d913181 --- /dev/null +++ b/tests/test_minimum_workload.py @@ -0,0 +1,451 @@ +"""Tests for MinimumWorkload warning rules (BE-CAL-007). + +Tests cover: + - _date_range_for_period computation + - _sum_real_slots aggregation + - _sum_virtual_slots aggregation + - check_workload_warnings comparison logic + - get_workload_warnings_for_date end-to-end convenience + - Warnings are advisory (non-blocking) +""" + +import pytest +from datetime import date, time + +from tests.conftest import auth_header + +from app.models.calendar import ( + SchedulePlan, + SlotStatus, + SlotType, + EventType, + TimeSlot, + DayOfWeek, +) +from app.models.minimum_workload import MinimumWorkload +from app.services.minimum_workload import ( + _date_range_for_period, + _sum_real_slots, + _sum_virtual_slots, + check_workload_warnings, + compute_scheduled_minutes, + get_workload_warnings_for_date, + get_workload_config, +) +from app.schemas.calendar import WorkloadWarningItem + + +# --------------------------------------------------------------------------- +# Unit: _date_range_for_period +# --------------------------------------------------------------------------- + +class TestDateRangeForPeriod: + def test_daily(self): + d = date(2026, 3, 15) # Sunday + start, end = _date_range_for_period("daily", d) + assert start == end == d + + def test_weekly_midweek(self): + d = date(2026, 3, 18) # Wednesday + start, end = _date_range_for_period("weekly", d) + assert start == date(2026, 3, 16) # Monday + assert end == date(2026, 3, 22) # Sunday + + def test_weekly_monday(self): + d = date(2026, 3, 16) # Monday + start, end = _date_range_for_period("weekly", d) + assert start == date(2026, 3, 16) + assert end == date(2026, 3, 22) + + def test_weekly_sunday(self): + d = date(2026, 3, 22) # Sunday + start, end = _date_range_for_period("weekly", d) + assert start == date(2026, 3, 16) + assert end == date(2026, 3, 22) + + def test_monthly(self): + d = date(2026, 3, 15) + start, end = _date_range_for_period("monthly", d) + assert start == date(2026, 3, 1) + assert end == date(2026, 3, 31) + + def test_monthly_february(self): + d = date(2026, 2, 10) + start, end = _date_range_for_period("monthly", d) + assert start == date(2026, 2, 1) + assert end == date(2026, 2, 28) + + def test_monthly_december(self): + d = date(2026, 12, 25) + start, end = _date_range_for_period("monthly", d) + assert start == date(2026, 12, 1) + assert end == date(2026, 12, 31) + + def test_yearly(self): + d = date(2026, 6, 15) + start, end = _date_range_for_period("yearly", d) + assert start == date(2026, 1, 1) + assert end == date(2026, 12, 31) + + def test_unknown_period_raises(self): + with pytest.raises(ValueError, match="Unknown period"): + _date_range_for_period("hourly", date(2026, 1, 1)) + + +# --------------------------------------------------------------------------- +# Unit: check_workload_warnings (pure comparison, no DB) +# --------------------------------------------------------------------------- + +class TestCheckWorkloadWarnings: + """Test the comparison logic with pre-computed scheduled_minutes.""" + + def test_no_warnings_when_all_zero_config(self, db, seed): + """Default config (all zeros) never triggers warnings.""" + scheduled = { + "daily": {"work": 0, "on_call": 0, "entertainment": 0}, + "weekly": {"work": 0, "on_call": 0, "entertainment": 0}, + "monthly": {"work": 0, "on_call": 0, "entertainment": 0}, + "yearly": {"work": 0, "on_call": 0, "entertainment": 0}, + } + warnings = check_workload_warnings(db, seed["admin_user"].id, scheduled) + assert warnings == [] + + def test_warning_when_below_threshold(self, db, seed): + """Setting a threshold higher than scheduled triggers a warning.""" + # Set daily work minimum to 60 min + cfg = MinimumWorkload( + user_id=seed["admin_user"].id, + config={ + "daily": {"work": 60, "on_call": 0, "entertainment": 0}, + "weekly": {"work": 0, "on_call": 0, "entertainment": 0}, + "monthly": {"work": 0, "on_call": 0, "entertainment": 0}, + "yearly": {"work": 0, "on_call": 0, "entertainment": 0}, + }, + ) + db.add(cfg) + db.commit() + + scheduled = { + "daily": {"work": 30, "on_call": 0, "entertainment": 0}, + "weekly": {"work": 100, "on_call": 0, "entertainment": 0}, + "monthly": {"work": 400, "on_call": 0, "entertainment": 0}, + "yearly": {"work": 5000, "on_call": 0, "entertainment": 0}, + } + warnings = check_workload_warnings(db, seed["admin_user"].id, scheduled) + assert len(warnings) == 1 + w = warnings[0] + assert w.period == "daily" + assert w.category == "work" + assert w.current_minutes == 30 + assert w.minimum_minutes == 60 + assert w.shortfall_minutes == 30 + + def test_no_warning_when_meeting_threshold(self, db, seed): + cfg = MinimumWorkload( + user_id=seed["admin_user"].id, + config={ + "daily": {"work": 30, "on_call": 0, "entertainment": 0}, + "weekly": {"work": 0, "on_call": 0, "entertainment": 0}, + "monthly": {"work": 0, "on_call": 0, "entertainment": 0}, + "yearly": {"work": 0, "on_call": 0, "entertainment": 0}, + }, + ) + db.add(cfg) + db.commit() + + scheduled = { + "daily": {"work": 30, "on_call": 0, "entertainment": 0}, + "weekly": {"work": 100, "on_call": 0, "entertainment": 0}, + "monthly": {"work": 400, "on_call": 0, "entertainment": 0}, + "yearly": {"work": 5000, "on_call": 0, "entertainment": 0}, + } + warnings = check_workload_warnings(db, seed["admin_user"].id, scheduled) + assert warnings == [] + + def test_multiple_warnings_across_periods_and_categories(self, db, seed): + cfg = MinimumWorkload( + user_id=seed["admin_user"].id, + config={ + "daily": {"work": 50, "on_call": 20, "entertainment": 0}, + "weekly": {"work": 300, "on_call": 0, "entertainment": 0}, + "monthly": {"work": 0, "on_call": 0, "entertainment": 0}, + "yearly": {"work": 0, "on_call": 0, "entertainment": 0}, + }, + ) + db.add(cfg) + db.commit() + + scheduled = { + "daily": {"work": 10, "on_call": 5, "entertainment": 0}, + "weekly": {"work": 100, "on_call": 0, "entertainment": 0}, + "monthly": {"work": 0, "on_call": 0, "entertainment": 0}, + "yearly": {"work": 0, "on_call": 0, "entertainment": 0}, + } + warnings = check_workload_warnings(db, seed["admin_user"].id, scheduled) + assert len(warnings) == 3 + periods_cats = {(w.period, w.category) for w in warnings} + assert ("daily", "work") in periods_cats + assert ("daily", "on_call") in periods_cats + assert ("weekly", "work") in periods_cats + + +# --------------------------------------------------------------------------- +# Integration: _sum_real_slots +# --------------------------------------------------------------------------- + +class TestSumRealSlots: + def test_sums_work_slots(self, db, seed): + """Real work slots are summed correctly.""" + user_id = seed["admin_user"].id + db.add(TimeSlot( + user_id=user_id, date=date(2026, 3, 15), + slot_type=SlotType.WORK, estimated_duration=30, + scheduled_at=time(9, 0), status=SlotStatus.NOT_STARTED, + )) + db.add(TimeSlot( + user_id=user_id, date=date(2026, 3, 15), + slot_type=SlotType.WORK, estimated_duration=20, + scheduled_at=time(10, 0), status=SlotStatus.FINISHED, + )) + db.commit() + + totals = _sum_real_slots(db, user_id, date(2026, 3, 15), date(2026, 3, 15)) + assert totals["work"] == 50 + assert totals["on_call"] == 0 + assert totals["entertainment"] == 0 + + def test_excludes_skipped_and_aborted(self, db, seed): + user_id = seed["admin_user"].id + db.add(TimeSlot( + user_id=user_id, date=date(2026, 3, 15), + slot_type=SlotType.WORK, estimated_duration=30, + scheduled_at=time(9, 0), status=SlotStatus.SKIPPED, + )) + db.add(TimeSlot( + user_id=user_id, date=date(2026, 3, 15), + slot_type=SlotType.WORK, estimated_duration=20, + scheduled_at=time(10, 0), status=SlotStatus.ABORTED, + )) + db.commit() + + totals = _sum_real_slots(db, user_id, date(2026, 3, 15), date(2026, 3, 15)) + assert totals["work"] == 0 + + def test_excludes_system_slots(self, db, seed): + user_id = seed["admin_user"].id + db.add(TimeSlot( + user_id=user_id, date=date(2026, 3, 15), + slot_type=SlotType.SYSTEM, estimated_duration=10, + scheduled_at=time(8, 0), status=SlotStatus.NOT_STARTED, + )) + db.commit() + + totals = _sum_real_slots(db, user_id, date(2026, 3, 15), date(2026, 3, 15)) + assert totals == {"work": 0, "on_call": 0, "entertainment": 0} + + def test_sums_across_date_range(self, db, seed): + user_id = seed["admin_user"].id + for day in [15, 16, 17]: + db.add(TimeSlot( + user_id=user_id, date=date(2026, 3, day), + slot_type=SlotType.WORK, estimated_duration=10, + scheduled_at=time(9, 0), status=SlotStatus.NOT_STARTED, + )) + db.commit() + + totals = _sum_real_slots(db, user_id, date(2026, 3, 15), date(2026, 3, 17)) + assert totals["work"] == 30 + + def test_multiple_categories(self, db, seed): + user_id = seed["admin_user"].id + db.add(TimeSlot( + user_id=user_id, date=date(2026, 3, 15), + slot_type=SlotType.WORK, estimated_duration=25, + scheduled_at=time(9, 0), status=SlotStatus.NOT_STARTED, + )) + db.add(TimeSlot( + user_id=user_id, date=date(2026, 3, 15), + slot_type=SlotType.ON_CALL, estimated_duration=15, + scheduled_at=time(10, 0), status=SlotStatus.NOT_STARTED, + )) + db.add(TimeSlot( + user_id=user_id, date=date(2026, 3, 15), + slot_type=SlotType.ENTERTAINMENT, estimated_duration=10, + scheduled_at=time(11, 0), status=SlotStatus.NOT_STARTED, + )) + db.commit() + + totals = _sum_real_slots(db, user_id, date(2026, 3, 15), date(2026, 3, 15)) + assert totals == {"work": 25, "on_call": 15, "entertainment": 10} + + +# --------------------------------------------------------------------------- +# Integration: _sum_virtual_slots +# --------------------------------------------------------------------------- + +class TestSumVirtualSlots: + def test_sums_virtual_plan_slots(self, db, seed): + """Virtual slots from an active plan are counted.""" + user_id = seed["admin_user"].id + plan = SchedulePlan( + user_id=user_id, + slot_type=SlotType.WORK, + estimated_duration=40, + at_time=time(9, 0), + on_day=DayOfWeek.SUN, # 2026-03-15 is a Sunday + is_active=True, + ) + db.add(plan) + db.commit() + + totals = _sum_virtual_slots(db, user_id, date(2026, 3, 15), date(2026, 3, 15)) + assert totals["work"] == 40 + + def test_skips_materialized_plan_slots(self, db, seed): + """If a plan slot is already materialized, it shouldn't be double-counted.""" + user_id = seed["admin_user"].id + plan = SchedulePlan( + user_id=user_id, + slot_type=SlotType.WORK, + estimated_duration=40, + at_time=time(9, 0), + on_day=DayOfWeek.SUN, + is_active=True, + ) + db.add(plan) + db.flush() + + # Materialize it + db.add(TimeSlot( + user_id=user_id, date=date(2026, 3, 15), + slot_type=SlotType.WORK, estimated_duration=40, + scheduled_at=time(9, 0), status=SlotStatus.NOT_STARTED, + plan_id=plan.id, + )) + db.commit() + + totals = _sum_virtual_slots(db, user_id, date(2026, 3, 15), date(2026, 3, 15)) + assert totals["work"] == 0 # Already materialized, not double-counted + + +# --------------------------------------------------------------------------- +# Integration: compute_scheduled_minutes +# --------------------------------------------------------------------------- + +class TestComputeScheduledMinutes: + def test_combines_real_and_virtual(self, db, seed): + user_id = seed["admin_user"].id + + # Real slot on the 15th + db.add(TimeSlot( + user_id=user_id, date=date(2026, 3, 15), + slot_type=SlotType.WORK, estimated_duration=20, + scheduled_at=time(9, 0), status=SlotStatus.NOT_STARTED, + )) + + # Plan that fires every day + plan = SchedulePlan( + user_id=user_id, + slot_type=SlotType.ON_CALL, + estimated_duration=10, + at_time=time(14, 0), + is_active=True, + ) + db.add(plan) + db.commit() + + result = compute_scheduled_minutes(db, user_id, date(2026, 3, 15)) + + # Daily: 20 work (real) + 10 on_call (virtual) + assert result["daily"]["work"] == 20 + assert result["daily"]["on_call"] == 10 + + # Weekly: the real slot + virtual slots for every day in the week + # 2026-03-15 is Sunday → week is Mon 2026-03-09 to Sun 2026-03-15 + assert result["weekly"]["work"] == 20 + assert result["weekly"]["on_call"] >= 10 # At least the one day + + +# --------------------------------------------------------------------------- +# Integration: get_workload_warnings_for_date (end-to-end) +# --------------------------------------------------------------------------- + +class TestGetWorkloadWarningsForDate: + def test_returns_warnings_when_below_threshold(self, db, seed): + user_id = seed["admin_user"].id + + # Set daily work minimum to 60 min + db.add(MinimumWorkload( + user_id=user_id, + config={ + "daily": {"work": 60, "on_call": 0, "entertainment": 0}, + "weekly": {"work": 0, "on_call": 0, "entertainment": 0}, + "monthly": {"work": 0, "on_call": 0, "entertainment": 0}, + "yearly": {"work": 0, "on_call": 0, "entertainment": 0}, + }, + )) + + # Only 30 min of work scheduled + db.add(TimeSlot( + user_id=user_id, date=date(2026, 3, 15), + slot_type=SlotType.WORK, estimated_duration=30, + scheduled_at=time(9, 0), status=SlotStatus.NOT_STARTED, + )) + db.commit() + + warnings = get_workload_warnings_for_date(db, user_id, date(2026, 3, 15)) + assert len(warnings) >= 1 + daily_work = [w for w in warnings if w.period == "daily" and w.category == "work"] + assert len(daily_work) == 1 + assert daily_work[0].shortfall_minutes == 30 + + def test_no_warnings_when_above_threshold(self, db, seed): + user_id = seed["admin_user"].id + + db.add(MinimumWorkload( + user_id=user_id, + config={ + "daily": {"work": 30, "on_call": 0, "entertainment": 0}, + "weekly": {"work": 0, "on_call": 0, "entertainment": 0}, + "monthly": {"work": 0, "on_call": 0, "entertainment": 0}, + "yearly": {"work": 0, "on_call": 0, "entertainment": 0}, + }, + )) + + db.add(TimeSlot( + user_id=user_id, date=date(2026, 3, 15), + slot_type=SlotType.WORK, estimated_duration=45, + scheduled_at=time(9, 0), status=SlotStatus.NOT_STARTED, + )) + db.commit() + + warnings = get_workload_warnings_for_date(db, user_id, date(2026, 3, 15)) + daily_work = [w for w in warnings if w.period == "daily" and w.category == "work"] + assert len(daily_work) == 0 + + def test_warning_data_structure(self, db, seed): + """Ensure warnings contain all required fields with correct types.""" + user_id = seed["admin_user"].id + + db.add(MinimumWorkload( + user_id=user_id, + config={ + "daily": {"work": 100, "on_call": 0, "entertainment": 0}, + "weekly": {"work": 0, "on_call": 0, "entertainment": 0}, + "monthly": {"work": 0, "on_call": 0, "entertainment": 0}, + "yearly": {"work": 0, "on_call": 0, "entertainment": 0}, + }, + )) + db.commit() + + warnings = get_workload_warnings_for_date(db, user_id, date(2026, 3, 15)) + assert len(warnings) >= 1 + w = warnings[0] + assert isinstance(w, WorkloadWarningItem) + assert isinstance(w.period, str) + assert isinstance(w.category, str) + assert isinstance(w.current_minutes, int) + assert isinstance(w.minimum_minutes, int) + assert isinstance(w.shortfall_minutes, int) + assert isinstance(w.message, str) + assert w.shortfall_minutes == w.minimum_minutes - w.current_minutes diff --git a/tests/test_slot_immutability.py b/tests/test_slot_immutability.py new file mode 100644 index 0000000..a634e75 --- /dev/null +++ b/tests/test_slot_immutability.py @@ -0,0 +1,234 @@ +"""Tests for past-slot immutability rules (BE-CAL-008). + +Tests cover: + - Editing a past real slot is forbidden + - Cancelling a past real slot is forbidden + - Editing a past virtual slot is forbidden + - Cancelling a past virtual slot is forbidden + - Editing/cancelling today's slots is allowed + - Editing/cancelling future slots is allowed + - Plan-edit / plan-cancel do not retroactively affect past materialized slots +""" + +import pytest +from datetime import date, time + +from app.models.calendar import ( + SchedulePlan, + SlotStatus, + SlotType, + TimeSlot, + DayOfWeek, +) +from app.services.slot_immutability import ( + ImmutableSlotError, + guard_edit_real_slot, + guard_cancel_real_slot, + guard_edit_virtual_slot, + guard_cancel_virtual_slot, + get_past_materialized_slot_ids, + guard_plan_edit_no_past_retroaction, + guard_plan_cancel_no_past_retroaction, +) +from app.services.plan_slot import make_virtual_slot_id + + +TODAY = date(2026, 3, 31) +YESTERDAY = date(2026, 3, 30) +LAST_WEEK = date(2026, 3, 24) +TOMORROW = date(2026, 4, 1) + + +# --------------------------------------------------------------------------- +# Helper +# --------------------------------------------------------------------------- + +def _make_slot(db, seed, slot_date, plan_id=None): + """Create and return a real TimeSlot.""" + slot = TimeSlot( + user_id=seed["admin_user"].id, + date=slot_date, + slot_type=SlotType.WORK, + estimated_duration=30, + scheduled_at=time(9, 0), + status=SlotStatus.NOT_STARTED, + plan_id=plan_id, + ) + db.add(slot) + db.flush() + return slot + + +# --------------------------------------------------------------------------- +# Real slot: edit +# --------------------------------------------------------------------------- + +class TestGuardEditRealSlot: + def test_past_slot_raises(self, db, seed): + slot = _make_slot(db, seed, YESTERDAY) + db.commit() + with pytest.raises(ImmutableSlotError, match="Cannot edit"): + guard_edit_real_slot(db, slot, today=TODAY) + + def test_today_slot_allowed(self, db, seed): + slot = _make_slot(db, seed, TODAY) + db.commit() + # Should not raise + guard_edit_real_slot(db, slot, today=TODAY) + + def test_future_slot_allowed(self, db, seed): + slot = _make_slot(db, seed, TOMORROW) + db.commit() + guard_edit_real_slot(db, slot, today=TODAY) + + +# --------------------------------------------------------------------------- +# Real slot: cancel +# --------------------------------------------------------------------------- + +class TestGuardCancelRealSlot: + def test_past_slot_raises(self, db, seed): + slot = _make_slot(db, seed, YESTERDAY) + db.commit() + with pytest.raises(ImmutableSlotError, match="Cannot cancel"): + guard_cancel_real_slot(db, slot, today=TODAY) + + def test_today_slot_allowed(self, db, seed): + slot = _make_slot(db, seed, TODAY) + db.commit() + guard_cancel_real_slot(db, slot, today=TODAY) + + def test_future_slot_allowed(self, db, seed): + slot = _make_slot(db, seed, TOMORROW) + db.commit() + guard_cancel_real_slot(db, slot, today=TODAY) + + +# --------------------------------------------------------------------------- +# Virtual slot: edit +# --------------------------------------------------------------------------- + +class TestGuardEditVirtualSlot: + def test_past_virtual_raises(self): + vid = make_virtual_slot_id(1, YESTERDAY) + with pytest.raises(ImmutableSlotError, match="Cannot edit"): + guard_edit_virtual_slot(vid, today=TODAY) + + def test_today_virtual_allowed(self): + vid = make_virtual_slot_id(1, TODAY) + guard_edit_virtual_slot(vid, today=TODAY) + + def test_future_virtual_allowed(self): + vid = make_virtual_slot_id(1, TOMORROW) + guard_edit_virtual_slot(vid, today=TODAY) + + def test_invalid_virtual_id_raises_value_error(self): + with pytest.raises(ValueError, match="Invalid virtual slot id"): + guard_edit_virtual_slot("bad-id", today=TODAY) + + +# --------------------------------------------------------------------------- +# Virtual slot: cancel +# --------------------------------------------------------------------------- + +class TestGuardCancelVirtualSlot: + def test_past_virtual_raises(self): + vid = make_virtual_slot_id(1, YESTERDAY) + with pytest.raises(ImmutableSlotError, match="Cannot cancel"): + guard_cancel_virtual_slot(vid, today=TODAY) + + def test_today_virtual_allowed(self): + vid = make_virtual_slot_id(1, TODAY) + guard_cancel_virtual_slot(vid, today=TODAY) + + def test_future_virtual_allowed(self): + vid = make_virtual_slot_id(1, TOMORROW) + guard_cancel_virtual_slot(vid, today=TODAY) + + +# --------------------------------------------------------------------------- +# Plan retroaction: past materialized slots are protected +# --------------------------------------------------------------------------- + +class TestPlanNoRetroaction: + def _make_plan_with_slots(self, db, seed): + """Create a plan with materialized slots in the past, today, and future.""" + user_id = seed["admin_user"].id + plan = SchedulePlan( + user_id=user_id, + slot_type=SlotType.WORK, + estimated_duration=30, + at_time=time(9, 0), + is_active=True, + ) + db.add(plan) + db.flush() + + past_slot = _make_slot(db, seed, LAST_WEEK, plan_id=plan.id) + yesterday_slot = _make_slot(db, seed, YESTERDAY, plan_id=plan.id) + today_slot = _make_slot(db, seed, TODAY, plan_id=plan.id) + future_slot = _make_slot(db, seed, TOMORROW, plan_id=plan.id) + db.commit() + return plan, past_slot, yesterday_slot, today_slot, future_slot + + def test_get_past_materialized_slot_ids(self, db, seed): + plan, past_slot, yesterday_slot, today_slot, future_slot = ( + self._make_plan_with_slots(db, seed) + ) + past_ids = get_past_materialized_slot_ids(db, plan.id, today=TODAY) + assert set(past_ids) == {past_slot.id, yesterday_slot.id} + assert today_slot.id not in past_ids + assert future_slot.id not in past_ids + + def test_guard_plan_edit_returns_protected_ids(self, db, seed): + plan, past_slot, yesterday_slot, _, _ = ( + self._make_plan_with_slots(db, seed) + ) + protected = guard_plan_edit_no_past_retroaction(db, plan.id, today=TODAY) + assert set(protected) == {past_slot.id, yesterday_slot.id} + + def test_guard_plan_cancel_returns_protected_ids(self, db, seed): + plan, past_slot, yesterday_slot, _, _ = ( + self._make_plan_with_slots(db, seed) + ) + protected = guard_plan_cancel_no_past_retroaction(db, plan.id, today=TODAY) + assert set(protected) == {past_slot.id, yesterday_slot.id} + + def test_no_past_slots_returns_empty(self, db, seed): + """If all materialized slots are today or later, no past IDs returned.""" + user_id = seed["admin_user"].id + plan = SchedulePlan( + user_id=user_id, + slot_type=SlotType.WORK, + estimated_duration=30, + at_time=time(9, 0), + is_active=True, + ) + db.add(plan) + db.flush() + _make_slot(db, seed, TODAY, plan_id=plan.id) + _make_slot(db, seed, TOMORROW, plan_id=plan.id) + db.commit() + + past_ids = get_past_materialized_slot_ids(db, plan.id, today=TODAY) + assert past_ids == [] + + +# --------------------------------------------------------------------------- +# ImmutableSlotError attributes +# --------------------------------------------------------------------------- + +class TestImmutableSlotError: + def test_error_attributes(self): + err = ImmutableSlotError(YESTERDAY, "edit", detail="test detail") + assert err.slot_date == YESTERDAY + assert err.operation == "edit" + assert err.detail == "test detail" + assert "Cannot edit" in str(err) + assert "2026-03-30" in str(err) + assert "test detail" in str(err) + + def test_error_without_detail(self): + err = ImmutableSlotError(YESTERDAY, "cancel") + assert "Cannot cancel" in str(err) + assert "test detail" not in str(err)