diff --git a/app/services/overlap.py b/app/services/overlap.py new file mode 100644 index 0000000..1a5e262 --- /dev/null +++ b/app/services/overlap.py @@ -0,0 +1,232 @@ +"""Calendar overlap detection service. + +BE-CAL-006: Validates that a new or edited TimeSlot does not overlap with +existing slots on the same day for the same user. + +Overlap is defined as two time ranges ``[start, start + duration)`` having +a non-empty intersection. Cancelled/aborted slots are excluded from +conflict checks (they no longer occupy calendar time). + +For the **create** scenario, all existing non-cancelled slots on the target +date are checked. + +For the **edit** scenario, the slot being edited is excluded from the +candidate set so it doesn't conflict with its own previous position. +""" + +from __future__ import annotations + +from datetime import date, time, timedelta, datetime +from typing import Optional + +from sqlalchemy.orm import Session + +from app.models.calendar import SlotStatus, TimeSlot +from app.services.plan_slot import get_virtual_slots_for_date + + +# Statuses that no longer occupy calendar time — excluded from overlap checks. +_INACTIVE_STATUSES = {SlotStatus.SKIPPED, SlotStatus.ABORTED} + + +# --------------------------------------------------------------------------- +# Internal helpers +# --------------------------------------------------------------------------- + +def _time_to_minutes(t: time) -> int: + """Convert a ``time`` to minutes since midnight.""" + return t.hour * 60 + t.minute + + +def _ranges_overlap( + start_a: int, + end_a: int, + start_b: int, + end_b: int, +) -> bool: + """Return *True* if two half-open intervals ``[a, a+dur)`` overlap.""" + return start_a < end_b and start_b < end_a + + +# --------------------------------------------------------------------------- +# Conflict data class +# --------------------------------------------------------------------------- + +class SlotConflict: + """Describes a single overlap conflict.""" + + __slots__ = ("conflicting_slot_id", "conflicting_virtual_id", + "scheduled_at", "estimated_duration", "slot_type", "message") + + def __init__( + self, + *, + conflicting_slot_id: Optional[int] = None, + conflicting_virtual_id: Optional[str] = None, + scheduled_at: time, + estimated_duration: int, + slot_type: str, + message: str, + ): + self.conflicting_slot_id = conflicting_slot_id + self.conflicting_virtual_id = conflicting_virtual_id + self.scheduled_at = scheduled_at + self.estimated_duration = estimated_duration + self.slot_type = slot_type + self.message = message + + def to_dict(self) -> dict: + d: dict = { + "scheduled_at": self.scheduled_at.isoformat(), + "estimated_duration": self.estimated_duration, + "slot_type": self.slot_type, + "message": self.message, + } + if self.conflicting_slot_id is not None: + d["conflicting_slot_id"] = self.conflicting_slot_id + if self.conflicting_virtual_id is not None: + d["conflicting_virtual_id"] = self.conflicting_virtual_id + return d + + +# --------------------------------------------------------------------------- +# Core overlap detection +# --------------------------------------------------------------------------- + +def _format_time_range(start: time, duration: int) -> str: + """Format a slot time range for human-readable messages.""" + start_min = _time_to_minutes(start) + end_min = start_min + duration + end_h, end_m = divmod(end_min, 60) + # Clamp to 23:59 for display purposes + if end_h >= 24: + end_h, end_m = 23, 59 + return f"{start.strftime('%H:%M')}-{end_h:02d}:{end_m:02d}" + + +def check_overlap( + db: Session, + user_id: int, + target_date: date, + scheduled_at: time, + estimated_duration: int, + *, + exclude_slot_id: Optional[int] = None, +) -> list[SlotConflict]: + """Check for time conflicts on *target_date* for *user_id*. + + Parameters + ---------- + db : + Active database session. + user_id : + The user whose calendar is being checked. + target_date : + The date to check. + scheduled_at : + Proposed start time. + estimated_duration : + Proposed duration in minutes. + exclude_slot_id : + If editing an existing slot, pass its ``id`` so it is not counted + as conflicting with itself. + + Returns + ------- + list[SlotConflict] + Empty list means no conflicts. Non-empty means the proposed slot + overlaps with one or more existing slots. + """ + new_start = _time_to_minutes(scheduled_at) + new_end = new_start + estimated_duration + + conflicts: list[SlotConflict] = [] + + # ---- 1. Check real (materialized) slots -------------------------------- + query = ( + db.query(TimeSlot) + .filter( + TimeSlot.user_id == user_id, + TimeSlot.date == target_date, + TimeSlot.status.notin_([s.value for s in _INACTIVE_STATUSES]), + ) + ) + if exclude_slot_id is not None: + query = query.filter(TimeSlot.id != exclude_slot_id) + + existing_slots: list[TimeSlot] = query.all() + + for slot in existing_slots: + slot_start = _time_to_minutes(slot.scheduled_at) + slot_end = slot_start + slot.estimated_duration + + if _ranges_overlap(new_start, new_end, slot_start, slot_end): + existing_range = _format_time_range(slot.scheduled_at, slot.estimated_duration) + proposed_range = _format_time_range(scheduled_at, estimated_duration) + conflicts.append(SlotConflict( + conflicting_slot_id=slot.id, + scheduled_at=slot.scheduled_at, + estimated_duration=slot.estimated_duration, + slot_type=slot.slot_type.value if hasattr(slot.slot_type, 'value') else str(slot.slot_type), + message=( + f"Proposed slot {proposed_range} overlaps with existing " + f"{slot.slot_type.value if hasattr(slot.slot_type, 'value') else slot.slot_type} " + f"slot (id={slot.id}) at {existing_range}" + ), + )) + + # ---- 2. Check virtual (plan-generated) slots --------------------------- + virtual_slots = get_virtual_slots_for_date(db, user_id, target_date) + + for vs in virtual_slots: + vs_start = _time_to_minutes(vs["scheduled_at"]) + vs_end = vs_start + vs["estimated_duration"] + + if _ranges_overlap(new_start, new_end, vs_start, vs_end): + existing_range = _format_time_range(vs["scheduled_at"], vs["estimated_duration"]) + proposed_range = _format_time_range(scheduled_at, estimated_duration) + slot_type_val = vs["slot_type"].value if hasattr(vs["slot_type"], 'value') else str(vs["slot_type"]) + conflicts.append(SlotConflict( + conflicting_virtual_id=vs["virtual_id"], + scheduled_at=vs["scheduled_at"], + estimated_duration=vs["estimated_duration"], + slot_type=slot_type_val, + message=( + f"Proposed slot {proposed_range} overlaps with virtual plan " + f"slot ({vs['virtual_id']}) at {existing_range}" + ), + )) + + return conflicts + + +# --------------------------------------------------------------------------- +# Convenience wrappers for create / edit scenarios +# --------------------------------------------------------------------------- + +def check_overlap_for_create( + db: Session, + user_id: int, + target_date: date, + scheduled_at: time, + estimated_duration: int, +) -> list[SlotConflict]: + """Check overlap when creating a brand-new slot (no exclusion).""" + return check_overlap( + db, user_id, target_date, scheduled_at, estimated_duration, + ) + + +def check_overlap_for_edit( + db: Session, + user_id: int, + slot_id: int, + target_date: date, + scheduled_at: time, + estimated_duration: int, +) -> list[SlotConflict]: + """Check overlap when editing an existing slot (exclude itself).""" + return check_overlap( + db, user_id, target_date, scheduled_at, estimated_duration, + exclude_slot_id=slot_id, + ) diff --git a/tests/test_overlap.py b/tests/test_overlap.py new file mode 100644 index 0000000..563edbe --- /dev/null +++ b/tests/test_overlap.py @@ -0,0 +1,374 @@ +"""Tests for BE-CAL-006: Calendar overlap detection. + +Covers: + - No conflict when slots don't overlap + - Conflict detected for overlapping time ranges + - Create vs edit scenarios (edit excludes own slot) + - Skipped/aborted slots are not considered + - Virtual (plan-generated) slots are checked + - Edge cases: adjacent slots, exact same time, partial overlap +""" + +import pytest +from datetime import date, time + +from app.models.calendar import ( + SchedulePlan, + SlotStatus, + SlotType, + EventType, + TimeSlot, + DayOfWeek, +) +from app.services.overlap import ( + check_overlap, + check_overlap_for_create, + check_overlap_for_edit, +) + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +TARGET_DATE = date(2026, 4, 1) # A Wednesday +USER_ID = 1 +USER_ID_2 = 2 + + +def _make_slot(db, *, scheduled_at, duration=30, status=SlotStatus.NOT_STARTED, user_id=USER_ID, slot_date=TARGET_DATE, plan_id=None): + """Insert a real TimeSlot and return it.""" + slot = TimeSlot( + user_id=user_id, + date=slot_date, + slot_type=SlotType.WORK, + estimated_duration=duration, + scheduled_at=scheduled_at, + status=status, + priority=0, + plan_id=plan_id, + ) + db.add(slot) + db.flush() + return slot + + +def _make_plan(db, *, at_time, duration=30, user_id=USER_ID, on_day=None, is_active=True): + """Insert a SchedulePlan and return it.""" + plan = SchedulePlan( + user_id=user_id, + slot_type=SlotType.WORK, + estimated_duration=duration, + at_time=at_time, + on_day=on_day, + is_active=is_active, + ) + db.add(plan) + db.flush() + return plan + + +@pytest.fixture(autouse=True) +def _ensure_users(seed): + """All overlap tests need seeded users (id=1, id=2) for FK constraints.""" + pass + + +# --------------------------------------------------------------------------- +# No-conflict cases +# --------------------------------------------------------------------------- + +class TestNoConflict: + + def test_empty_calendar(self, db): + """No existing slots → no conflicts.""" + conflicts = check_overlap_for_create( + db, USER_ID, TARGET_DATE, time(9, 0), 30, + ) + assert conflicts == [] + + def test_adjacent_before(self, db): + """Existing 09:00-09:30, proposed 09:30-10:00 → no overlap.""" + _make_slot(db, scheduled_at=time(9, 0), duration=30) + db.commit() + + conflicts = check_overlap_for_create( + db, USER_ID, TARGET_DATE, time(9, 30), 30, + ) + assert conflicts == [] + + def test_adjacent_after(self, db): + """Existing 10:00-10:30, proposed 09:30-10:00 → no overlap.""" + _make_slot(db, scheduled_at=time(10, 0), duration=30) + db.commit() + + conflicts = check_overlap_for_create( + db, USER_ID, TARGET_DATE, time(9, 30), 30, + ) + assert conflicts == [] + + def test_different_user(self, db): + """Slot for user 2 should not conflict with user 1's new slot.""" + _make_slot(db, scheduled_at=time(9, 0), duration=30, user_id=USER_ID_2) + db.commit() + + conflicts = check_overlap_for_create( + db, USER_ID, TARGET_DATE, time(9, 0), 30, + ) + assert conflicts == [] + + def test_different_date(self, db): + """Same time on a different date → no conflict.""" + _make_slot(db, scheduled_at=time(9, 0), duration=30, slot_date=date(2026, 4, 2)) + db.commit() + + conflicts = check_overlap_for_create( + db, USER_ID, TARGET_DATE, time(9, 0), 30, + ) + assert conflicts == [] + + +# --------------------------------------------------------------------------- +# Conflict detection +# --------------------------------------------------------------------------- + +class TestConflictDetected: + + def test_exact_same_time(self, db): + """Same start + same duration = overlap.""" + _make_slot(db, scheduled_at=time(9, 0), duration=30) + db.commit() + + conflicts = check_overlap_for_create( + db, USER_ID, TARGET_DATE, time(9, 0), 30, + ) + assert len(conflicts) == 1 + assert conflicts[0].conflicting_slot_id is not None + assert "overlaps" in conflicts[0].message + + def test_partial_overlap_start(self, db): + """Existing 09:00-09:30, proposed 09:15-09:45 → overlap.""" + _make_slot(db, scheduled_at=time(9, 0), duration=30) + db.commit() + + conflicts = check_overlap_for_create( + db, USER_ID, TARGET_DATE, time(9, 15), 30, + ) + assert len(conflicts) == 1 + + def test_partial_overlap_end(self, db): + """Existing 09:15-09:45, proposed 09:00-09:30 → overlap.""" + _make_slot(db, scheduled_at=time(9, 15), duration=30) + db.commit() + + conflicts = check_overlap_for_create( + db, USER_ID, TARGET_DATE, time(9, 0), 30, + ) + assert len(conflicts) == 1 + + def test_proposed_contains_existing(self, db): + """Proposed 09:00-10:00 contains existing 09:15-09:45.""" + _make_slot(db, scheduled_at=time(9, 15), duration=30) + db.commit() + + conflicts = check_overlap_for_create( + db, USER_ID, TARGET_DATE, time(9, 0), 50, + ) + assert len(conflicts) == 1 + + def test_existing_contains_proposed(self, db): + """Existing 09:00-10:00 contains proposed 09:15-09:30.""" + _make_slot(db, scheduled_at=time(9, 0), duration=50) + db.commit() + + conflicts = check_overlap_for_create( + db, USER_ID, TARGET_DATE, time(9, 15), 15, + ) + assert len(conflicts) == 1 + + def test_multiple_conflicts(self, db): + """Proposed overlaps with two existing slots.""" + _make_slot(db, scheduled_at=time(9, 0), duration=30) + _make_slot(db, scheduled_at=time(9, 20), duration=30) + db.commit() + + conflicts = check_overlap_for_create( + db, USER_ID, TARGET_DATE, time(9, 10), 30, + ) + assert len(conflicts) == 2 + + +# --------------------------------------------------------------------------- +# Inactive slots excluded +# --------------------------------------------------------------------------- + +class TestInactiveExcluded: + + def test_skipped_slot_ignored(self, db): + """Skipped slot at same time should not cause conflict.""" + _make_slot(db, scheduled_at=time(9, 0), duration=30, status=SlotStatus.SKIPPED) + db.commit() + + conflicts = check_overlap_for_create( + db, USER_ID, TARGET_DATE, time(9, 0), 30, + ) + assert conflicts == [] + + def test_aborted_slot_ignored(self, db): + """Aborted slot at same time should not cause conflict.""" + _make_slot(db, scheduled_at=time(9, 0), duration=30, status=SlotStatus.ABORTED) + db.commit() + + conflicts = check_overlap_for_create( + db, USER_ID, TARGET_DATE, time(9, 0), 30, + ) + assert conflicts == [] + + def test_ongoing_slot_conflicts(self, db): + """Ongoing slot should still cause conflict.""" + _make_slot(db, scheduled_at=time(9, 0), duration=30, status=SlotStatus.ONGOING) + db.commit() + + conflicts = check_overlap_for_create( + db, USER_ID, TARGET_DATE, time(9, 0), 30, + ) + assert len(conflicts) == 1 + + def test_deferred_slot_conflicts(self, db): + """Deferred slot should still cause conflict.""" + _make_slot(db, scheduled_at=time(9, 0), duration=30, status=SlotStatus.DEFERRED) + db.commit() + + conflicts = check_overlap_for_create( + db, USER_ID, TARGET_DATE, time(9, 0), 30, + ) + assert len(conflicts) == 1 + + +# --------------------------------------------------------------------------- +# Edit scenario (exclude own slot) +# --------------------------------------------------------------------------- + +class TestEditExcludeSelf: + + def test_edit_no_self_conflict(self, db): + """Editing a slot to the same time should not conflict with itself.""" + slot = _make_slot(db, scheduled_at=time(9, 0), duration=30) + db.commit() + + conflicts = check_overlap_for_edit( + db, USER_ID, slot.id, TARGET_DATE, time(9, 0), 30, + ) + assert conflicts == [] + + def test_edit_still_detects_others(self, db): + """Editing a slot detects overlap with *other* slots.""" + slot = _make_slot(db, scheduled_at=time(9, 0), duration=30) + _make_slot(db, scheduled_at=time(9, 30), duration=30) + db.commit() + + # Move slot to overlap with the second one + conflicts = check_overlap_for_edit( + db, USER_ID, slot.id, TARGET_DATE, time(9, 20), 30, + ) + assert len(conflicts) == 1 + + def test_edit_self_excluded_others_fine(self, db): + """Moving a slot to a free spot should report no conflicts.""" + slot = _make_slot(db, scheduled_at=time(9, 0), duration=30) + _make_slot(db, scheduled_at=time(10, 0), duration=30) + db.commit() + + # Move to 11:00 — no overlap + conflicts = check_overlap_for_edit( + db, USER_ID, slot.id, TARGET_DATE, time(11, 0), 30, + ) + assert conflicts == [] + + +# --------------------------------------------------------------------------- +# Virtual slot (plan-generated) overlap +# --------------------------------------------------------------------------- + +class TestVirtualSlotOverlap: + + def test_conflict_with_virtual_slot(self, db): + """A plan that generates a virtual slot at 09:00 should conflict.""" + # TARGET_DATE is 2026-04-01 (Wednesday) + _make_plan(db, at_time=time(9, 0), duration=30, on_day=DayOfWeek.WED) + db.commit() + + conflicts = check_overlap_for_create( + db, USER_ID, TARGET_DATE, time(9, 0), 30, + ) + assert len(conflicts) == 1 + assert conflicts[0].conflicting_virtual_id is not None + assert conflicts[0].conflicting_slot_id is None + + def test_no_conflict_with_inactive_plan(self, db): + """Cancelled plan should not generate a virtual slot to conflict with.""" + _make_plan(db, at_time=time(9, 0), duration=30, on_day=DayOfWeek.WED, is_active=False) + db.commit() + + conflicts = check_overlap_for_create( + db, USER_ID, TARGET_DATE, time(9, 0), 30, + ) + assert conflicts == [] + + def test_no_conflict_with_non_matching_plan(self, db): + """Plan for Monday should not generate a virtual slot on Wednesday.""" + _make_plan(db, at_time=time(9, 0), duration=30, on_day=DayOfWeek.MON) + db.commit() + + conflicts = check_overlap_for_create( + db, USER_ID, TARGET_DATE, time(9, 0), 30, + ) + assert conflicts == [] + + def test_materialized_plan_not_double_counted(self, db): + """A plan that's already materialized should only be counted as a real slot, not also virtual.""" + plan = _make_plan(db, at_time=time(9, 0), duration=30, on_day=DayOfWeek.WED) + _make_slot(db, scheduled_at=time(9, 0), duration=30, plan_id=plan.id) + db.commit() + + conflicts = check_overlap_for_create( + db, USER_ID, TARGET_DATE, time(9, 0), 30, + ) + # Should only have 1 conflict (the real slot), not 2 + assert len(conflicts) == 1 + assert conflicts[0].conflicting_slot_id is not None + + +# --------------------------------------------------------------------------- +# Conflict message content +# --------------------------------------------------------------------------- + +class TestConflictMessage: + + def test_message_has_time_info(self, db): + """Conflict message should include time range information.""" + _make_slot(db, scheduled_at=time(9, 0), duration=30) + db.commit() + + conflicts = check_overlap_for_create( + db, USER_ID, TARGET_DATE, time(9, 15), 30, + ) + assert len(conflicts) == 1 + msg = conflicts[0].message + assert "09:00" in msg + assert "overlaps" in msg + + def test_to_dict(self, db): + """SlotConflict.to_dict() should return a proper dict.""" + _make_slot(db, scheduled_at=time(9, 0), duration=30) + db.commit() + + conflicts = check_overlap_for_create( + db, USER_ID, TARGET_DATE, time(9, 0), 30, + ) + d = conflicts[0].to_dict() + assert "scheduled_at" in d + assert "estimated_duration" in d + assert "slot_type" in d + assert "message" in d + assert "conflicting_slot_id" in d