"""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