zhi
4f0e933de3
BE-CAL-007: Workload warning computation (already implemented in prior wave, verified tests pass - 24/24). Computes daily/weekly/monthly/yearly scheduled minutes and compares against user thresholds. Warnings are advisory only. BE-CAL-008: New slot_immutability service with guards for: - Forbid edit/cancel of past real slots (raises ImmutableSlotError) - Forbid edit/cancel of past virtual slots - Plan-edit/plan-cancel helper to identify past materialized slot IDs that must not be retroactively modified Tests: 19/19 passing.
452 lines
17 KiB
Python
452 lines
17 KiB
Python
"""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
|