Compare commits

...

1 Commits

Author SHA1 Message Date
zhi
570cfee5cd BE-CAL-006: implement Calendar overlap detection service
- New overlap.py service with check_overlap(), check_overlap_for_create(),
  and check_overlap_for_edit() functions
- Detects same-day time conflicts for a user's calendar
- Checks both real (materialized) TimeSlots and virtual (plan-generated) slots
- Excludes skipped/aborted slots from conflict checks
- Edit scenario excludes the slot being edited from conflict candidates
- Returns structured SlotConflict objects with human-readable messages
- 24 passing tests covering no-conflict, conflict detection, inactive
  exclusion, edit self-exclusion, virtual slot overlap, and message content
2026-03-31 01:17:54 +00:00
2 changed files with 606 additions and 0 deletions

232
app/services/overlap.py Normal file
View File

@@ -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,
)

374
tests/test_overlap.py Normal file
View File

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