TEST-BE-CAL-001 add calendar backend model and API tests
This commit is contained in:
848
tests/test_calendar_models.py
Normal file
848
tests/test_calendar_models.py
Normal file
@@ -0,0 +1,848 @@
|
||||
"""Tests for BE-CAL-001: Calendar model definitions.
|
||||
|
||||
Covers:
|
||||
- TimeSlot model creation and fields
|
||||
- SchedulePlan model creation and fields
|
||||
- Enum validations
|
||||
- Model relationships
|
||||
- DB constraints (check constraints, foreign keys)
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from datetime import date, time, datetime
|
||||
from sqlalchemy.exc import IntegrityError
|
||||
|
||||
from app.models.calendar import (
|
||||
TimeSlot,
|
||||
SchedulePlan,
|
||||
SlotType,
|
||||
SlotStatus,
|
||||
EventType,
|
||||
DayOfWeek,
|
||||
MonthOfYear,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# TimeSlot Model Tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestTimeSlotModel:
|
||||
"""Tests for TimeSlot ORM model."""
|
||||
|
||||
def test_create_timeslot_basic(self, db, seed):
|
||||
"""Test creating a basic TimeSlot with required fields."""
|
||||
slot = TimeSlot(
|
||||
user_id=seed["admin_user"].id,
|
||||
date=date(2026, 4, 1),
|
||||
slot_type=SlotType.WORK,
|
||||
estimated_duration=30,
|
||||
scheduled_at=time(9, 0),
|
||||
status=SlotStatus.NOT_STARTED,
|
||||
priority=0,
|
||||
)
|
||||
db.add(slot)
|
||||
db.commit()
|
||||
db.refresh(slot)
|
||||
|
||||
assert slot.id is not None
|
||||
assert slot.user_id == seed["admin_user"].id
|
||||
assert slot.date == date(2026, 4, 1)
|
||||
assert slot.slot_type == SlotType.WORK
|
||||
assert slot.estimated_duration == 30
|
||||
assert slot.scheduled_at == time(9, 0)
|
||||
assert slot.status == SlotStatus.NOT_STARTED
|
||||
assert slot.priority == 0
|
||||
assert slot.attended is False
|
||||
assert slot.plan_id is None
|
||||
|
||||
def test_create_timeslot_all_fields(self, db, seed):
|
||||
"""Test creating a TimeSlot with all optional fields."""
|
||||
slot = TimeSlot(
|
||||
user_id=seed["dev_user"].id,
|
||||
date=date(2026, 4, 1),
|
||||
slot_type=SlotType.ON_CALL,
|
||||
estimated_duration=45,
|
||||
scheduled_at=time(14, 30),
|
||||
started_at=time(14, 35),
|
||||
attended=True,
|
||||
actual_duration=40,
|
||||
event_type=EventType.JOB,
|
||||
event_data={"type": "Task", "code": "TASK-42"},
|
||||
priority=5,
|
||||
status=SlotStatus.FINISHED,
|
||||
)
|
||||
db.add(slot)
|
||||
db.commit()
|
||||
db.refresh(slot)
|
||||
|
||||
assert slot.started_at == time(14, 35)
|
||||
assert slot.attended is True
|
||||
assert slot.actual_duration == 40
|
||||
assert slot.event_type == EventType.JOB
|
||||
assert slot.event_data == {"type": "Task", "code": "TASK-42"}
|
||||
assert slot.priority == 5
|
||||
assert slot.status == SlotStatus.FINISHED
|
||||
|
||||
def test_timeslot_slot_type_variants(self, db, seed):
|
||||
"""Test all SlotType enum variants."""
|
||||
for idx, slot_type in enumerate(SlotType):
|
||||
slot = TimeSlot(
|
||||
user_id=seed["admin_user"].id,
|
||||
date=date(2026, 4, 1),
|
||||
slot_type=slot_type,
|
||||
estimated_duration=10,
|
||||
scheduled_at=time(idx, 0),
|
||||
status=SlotStatus.NOT_STARTED,
|
||||
priority=idx,
|
||||
)
|
||||
db.add(slot)
|
||||
db.commit()
|
||||
|
||||
slots = db.query(TimeSlot).filter_by(user_id=seed["admin_user"].id).all()
|
||||
assert len(slots) == 4
|
||||
assert {s.slot_type for s in slots} == set(SlotType)
|
||||
|
||||
def test_timeslot_status_transitions(self, db, seed):
|
||||
"""Test all SlotStatus enum variants."""
|
||||
for idx, status in enumerate(SlotStatus):
|
||||
slot = TimeSlot(
|
||||
user_id=seed["admin_user"].id,
|
||||
date=date(2026, 4, 1),
|
||||
slot_type=SlotType.WORK,
|
||||
estimated_duration=10,
|
||||
scheduled_at=time(idx, 0),
|
||||
status=status,
|
||||
priority=0,
|
||||
)
|
||||
db.add(slot)
|
||||
db.commit()
|
||||
|
||||
slots = db.query(TimeSlot).filter_by(user_id=seed["admin_user"].id).all()
|
||||
assert len(slots) == 7
|
||||
assert {s.status for s in slots} == set(SlotStatus)
|
||||
|
||||
def test_timeslot_event_type_variants(self, db, seed):
|
||||
"""Test all EventType enum variants."""
|
||||
for idx, event_type in enumerate(EventType):
|
||||
slot = TimeSlot(
|
||||
user_id=seed["admin_user"].id,
|
||||
date=date(2026, 4, 1),
|
||||
slot_type=SlotType.WORK,
|
||||
estimated_duration=10,
|
||||
scheduled_at=time(idx, 0),
|
||||
status=SlotStatus.NOT_STARTED,
|
||||
event_type=event_type,
|
||||
priority=0,
|
||||
)
|
||||
db.add(slot)
|
||||
db.commit()
|
||||
|
||||
slots = db.query(TimeSlot).filter_by(user_id=seed["admin_user"].id).all()
|
||||
assert len(slots) == 3
|
||||
assert {s.event_type for s in slots} == set(EventType)
|
||||
|
||||
def test_timeslot_nullable_event_type(self, db, seed):
|
||||
"""Test that event_type can be NULL."""
|
||||
slot = TimeSlot(
|
||||
user_id=seed["admin_user"].id,
|
||||
date=date(2026, 4, 1),
|
||||
slot_type=SlotType.WORK,
|
||||
estimated_duration=30,
|
||||
scheduled_at=time(9, 0),
|
||||
status=SlotStatus.NOT_STARTED,
|
||||
event_type=None,
|
||||
priority=0,
|
||||
)
|
||||
db.add(slot)
|
||||
db.commit()
|
||||
db.refresh(slot)
|
||||
|
||||
assert slot.event_type is None
|
||||
assert slot.event_data is None
|
||||
|
||||
def test_timeslot_duration_bounds(self, db, seed):
|
||||
"""Test duration at boundary values (1-50)."""
|
||||
# Min duration
|
||||
slot_min = TimeSlot(
|
||||
user_id=seed["admin_user"].id,
|
||||
date=date(2026, 4, 1),
|
||||
slot_type=SlotType.WORK,
|
||||
estimated_duration=1,
|
||||
scheduled_at=time(8, 0),
|
||||
status=SlotStatus.NOT_STARTED,
|
||||
priority=0,
|
||||
)
|
||||
db.add(slot_min)
|
||||
|
||||
# Max duration
|
||||
slot_max = TimeSlot(
|
||||
user_id=seed["admin_user"].id,
|
||||
date=date(2026, 4, 1),
|
||||
slot_type=SlotType.WORK,
|
||||
estimated_duration=50,
|
||||
scheduled_at=time(9, 0),
|
||||
status=SlotStatus.NOT_STARTED,
|
||||
priority=0,
|
||||
)
|
||||
db.add(slot_max)
|
||||
db.commit()
|
||||
|
||||
assert slot_min.estimated_duration == 1
|
||||
assert slot_max.estimated_duration == 50
|
||||
|
||||
def test_timeslot_priority_bounds(self, db, seed):
|
||||
"""Test priority at boundary values (0-99)."""
|
||||
slot_low = TimeSlot(
|
||||
user_id=seed["admin_user"].id,
|
||||
date=date(2026, 4, 1),
|
||||
slot_type=SlotType.WORK,
|
||||
estimated_duration=10,
|
||||
scheduled_at=time(8, 0),
|
||||
status=SlotStatus.NOT_STARTED,
|
||||
priority=0,
|
||||
)
|
||||
db.add(slot_low)
|
||||
|
||||
slot_high = TimeSlot(
|
||||
user_id=seed["admin_user"].id,
|
||||
date=date(2026, 4, 1),
|
||||
slot_type=SlotType.WORK,
|
||||
estimated_duration=10,
|
||||
scheduled_at=time(9, 0),
|
||||
status=SlotStatus.NOT_STARTED,
|
||||
priority=99,
|
||||
)
|
||||
db.add(slot_high)
|
||||
db.commit()
|
||||
|
||||
assert slot_low.priority == 0
|
||||
assert slot_high.priority == 99
|
||||
|
||||
def test_timeslot_timestamps_auto_set(self, db, seed):
|
||||
"""Test that created_at and updated_at are set automatically."""
|
||||
slot = TimeSlot(
|
||||
user_id=seed["admin_user"].id,
|
||||
date=date(2026, 4, 1),
|
||||
slot_type=SlotType.WORK,
|
||||
estimated_duration=30,
|
||||
scheduled_at=time(9, 0),
|
||||
status=SlotStatus.NOT_STARTED,
|
||||
priority=0,
|
||||
)
|
||||
db.add(slot)
|
||||
db.commit()
|
||||
db.refresh(slot)
|
||||
|
||||
assert slot.created_at is not None
|
||||
assert isinstance(slot.created_at, datetime)
|
||||
|
||||
def test_timeslot_user_foreign_key(self, db):
|
||||
"""Test that invalid user_id raises IntegrityError."""
|
||||
slot = TimeSlot(
|
||||
user_id=99999, # Non-existent user
|
||||
date=date(2026, 4, 1),
|
||||
slot_type=SlotType.WORK,
|
||||
estimated_duration=30,
|
||||
scheduled_at=time(9, 0),
|
||||
status=SlotStatus.NOT_STARTED,
|
||||
priority=0,
|
||||
)
|
||||
db.add(slot)
|
||||
with pytest.raises(IntegrityError):
|
||||
db.commit()
|
||||
|
||||
def test_timeslot_plan_relationship(self, db, seed):
|
||||
"""Test relationship between TimeSlot and SchedulePlan."""
|
||||
# Create a plan first
|
||||
plan = SchedulePlan(
|
||||
user_id=seed["admin_user"].id,
|
||||
slot_type=SlotType.WORK,
|
||||
estimated_duration=30,
|
||||
at_time=time(9, 0),
|
||||
is_active=True,
|
||||
)
|
||||
db.add(plan)
|
||||
db.commit()
|
||||
db.refresh(plan)
|
||||
|
||||
# Create a slot linked to the plan
|
||||
slot = TimeSlot(
|
||||
user_id=seed["admin_user"].id,
|
||||
date=date(2026, 4, 1),
|
||||
slot_type=SlotType.WORK,
|
||||
estimated_duration=30,
|
||||
scheduled_at=time(9, 0),
|
||||
status=SlotStatus.NOT_STARTED,
|
||||
priority=0,
|
||||
plan_id=plan.id,
|
||||
)
|
||||
db.add(slot)
|
||||
db.commit()
|
||||
db.refresh(slot)
|
||||
|
||||
assert slot.plan_id == plan.id
|
||||
assert slot.plan.id == plan.id
|
||||
assert slot.plan.user_id == seed["admin_user"].id
|
||||
|
||||
def test_timeslot_query_by_date(self, db, seed):
|
||||
"""Test querying slots by date."""
|
||||
dates = [date(2026, 4, 1), date(2026, 4, 2), date(2026, 4, 1)]
|
||||
for idx, d in enumerate(dates):
|
||||
slot = TimeSlot(
|
||||
user_id=seed["admin_user"].id,
|
||||
date=d,
|
||||
slot_type=SlotType.WORK,
|
||||
estimated_duration=30,
|
||||
scheduled_at=time(9 + idx, 0),
|
||||
status=SlotStatus.NOT_STARTED,
|
||||
priority=0,
|
||||
)
|
||||
db.add(slot)
|
||||
db.commit()
|
||||
|
||||
slots_april_1 = db.query(TimeSlot).filter_by(
|
||||
user_id=seed["admin_user"].id,
|
||||
date=date(2026, 4, 1)
|
||||
).all()
|
||||
assert len(slots_april_1) == 2
|
||||
|
||||
def test_timeslot_query_by_status(self, db, seed):
|
||||
"""Test querying slots by status."""
|
||||
for idx, status in enumerate([SlotStatus.NOT_STARTED, SlotStatus.ONGOING, SlotStatus.NOT_STARTED]):
|
||||
slot = TimeSlot(
|
||||
user_id=seed["admin_user"].id,
|
||||
date=date(2026, 4, 1),
|
||||
slot_type=SlotType.WORK,
|
||||
estimated_duration=30,
|
||||
scheduled_at=time(9 + idx, 0),
|
||||
status=status,
|
||||
priority=0,
|
||||
)
|
||||
db.add(slot)
|
||||
db.commit()
|
||||
|
||||
not_started = db.query(TimeSlot).filter_by(
|
||||
user_id=seed["admin_user"].id,
|
||||
status=SlotStatus.NOT_STARTED
|
||||
).all()
|
||||
assert len(not_started) == 2
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# SchedulePlan Model Tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestSchedulePlanModel:
|
||||
"""Tests for SchedulePlan ORM model."""
|
||||
|
||||
def test_create_plan_basic(self, db, seed):
|
||||
"""Test creating a basic SchedulePlan with required fields."""
|
||||
plan = SchedulePlan(
|
||||
user_id=seed["admin_user"].id,
|
||||
slot_type=SlotType.WORK,
|
||||
estimated_duration=30,
|
||||
at_time=time(9, 0),
|
||||
is_active=True,
|
||||
)
|
||||
db.add(plan)
|
||||
db.commit()
|
||||
db.refresh(plan)
|
||||
|
||||
assert plan.id is not None
|
||||
assert plan.user_id == seed["admin_user"].id
|
||||
assert plan.slot_type == SlotType.WORK
|
||||
assert plan.estimated_duration == 30
|
||||
assert plan.at_time == time(9, 0)
|
||||
assert plan.is_active is True
|
||||
assert plan.on_day is None
|
||||
assert plan.on_week is None
|
||||
assert plan.on_month is None
|
||||
assert plan.event_type is None
|
||||
assert plan.event_data is None
|
||||
|
||||
def test_create_plan_daily(self, db, seed):
|
||||
"""Test creating a daily plan (--at only)."""
|
||||
plan = SchedulePlan(
|
||||
user_id=seed["admin_user"].id,
|
||||
slot_type=SlotType.WORK,
|
||||
estimated_duration=25,
|
||||
at_time=time(10, 0),
|
||||
is_active=True,
|
||||
)
|
||||
db.add(plan)
|
||||
db.commit()
|
||||
db.refresh(plan)
|
||||
|
||||
assert plan.at_time == time(10, 0)
|
||||
assert plan.on_day is None
|
||||
assert plan.on_week is None
|
||||
assert plan.on_month is None
|
||||
|
||||
def test_create_plan_weekly(self, db, seed):
|
||||
"""Test creating a weekly plan (--at + --on-day)."""
|
||||
plan = SchedulePlan(
|
||||
user_id=seed["admin_user"].id,
|
||||
slot_type=SlotType.ON_CALL,
|
||||
estimated_duration=45,
|
||||
at_time=time(14, 0),
|
||||
on_day=DayOfWeek.MON,
|
||||
is_active=True,
|
||||
)
|
||||
db.add(plan)
|
||||
db.commit()
|
||||
db.refresh(plan)
|
||||
|
||||
assert plan.on_day == DayOfWeek.MON
|
||||
assert plan.on_week is None
|
||||
assert plan.on_month is None
|
||||
|
||||
def test_create_plan_monthly(self, db, seed):
|
||||
"""Test creating a monthly plan (--at + --on-day + --on-week)."""
|
||||
plan = SchedulePlan(
|
||||
user_id=seed["admin_user"].id,
|
||||
slot_type=SlotType.ENTERTAINMENT,
|
||||
estimated_duration=45,
|
||||
at_time=time(19, 0),
|
||||
on_day=DayOfWeek.FRI,
|
||||
on_week=2,
|
||||
is_active=True,
|
||||
)
|
||||
db.add(plan)
|
||||
db.commit()
|
||||
db.refresh(plan)
|
||||
|
||||
assert plan.on_day == DayOfWeek.FRI
|
||||
assert plan.on_week == 2
|
||||
assert plan.on_month is None
|
||||
|
||||
def test_create_plan_yearly(self, db, seed):
|
||||
"""Test creating a yearly plan (all period params)."""
|
||||
plan = SchedulePlan(
|
||||
user_id=seed["admin_user"].id,
|
||||
slot_type=SlotType.WORK,
|
||||
estimated_duration=50,
|
||||
at_time=time(9, 0),
|
||||
on_day=DayOfWeek.SUN,
|
||||
on_week=1,
|
||||
on_month=MonthOfYear.JAN,
|
||||
is_active=True,
|
||||
)
|
||||
db.add(plan)
|
||||
db.commit()
|
||||
db.refresh(plan)
|
||||
|
||||
assert plan.on_day == DayOfWeek.SUN
|
||||
assert plan.on_week == 1
|
||||
assert plan.on_month == MonthOfYear.JAN
|
||||
|
||||
def test_create_plan_with_event(self, db, seed):
|
||||
"""Test creating a plan with event_type and event_data."""
|
||||
plan = SchedulePlan(
|
||||
user_id=seed["admin_user"].id,
|
||||
slot_type=SlotType.WORK,
|
||||
estimated_duration=30,
|
||||
at_time=time(9, 0),
|
||||
event_type=EventType.JOB,
|
||||
event_data={"type": "Meeting", "participants": ["user1", "user2"]},
|
||||
is_active=True,
|
||||
)
|
||||
db.add(plan)
|
||||
db.commit()
|
||||
db.refresh(plan)
|
||||
|
||||
assert plan.event_type == EventType.JOB
|
||||
assert plan.event_data == {"type": "Meeting", "participants": ["user1", "user2"]}
|
||||
|
||||
def test_plan_slot_type_variants(self, db, seed):
|
||||
"""Test all SlotType enum variants for SchedulePlan."""
|
||||
for idx, slot_type in enumerate(SlotType):
|
||||
plan = SchedulePlan(
|
||||
user_id=seed["admin_user"].id,
|
||||
slot_type=slot_type,
|
||||
estimated_duration=10,
|
||||
at_time=time(idx, 0),
|
||||
is_active=True,
|
||||
)
|
||||
db.add(plan)
|
||||
db.commit()
|
||||
|
||||
plans = db.query(SchedulePlan).filter_by(user_id=seed["admin_user"].id).all()
|
||||
assert len(plans) == 4
|
||||
assert {p.slot_type for p in plans} == set(SlotType)
|
||||
|
||||
def test_plan_on_week_validation(self, db, seed):
|
||||
"""Test on_week validation (must be 1-4)."""
|
||||
# Valid values
|
||||
for week in [1, 2, 3, 4]:
|
||||
plan = SchedulePlan(
|
||||
user_id=seed["admin_user"].id,
|
||||
slot_type=SlotType.WORK,
|
||||
estimated_duration=30,
|
||||
at_time=time(9, 0),
|
||||
on_day=DayOfWeek.MON,
|
||||
on_week=week,
|
||||
is_active=True,
|
||||
)
|
||||
db.add(plan)
|
||||
db.commit()
|
||||
|
||||
plans = db.query(SchedulePlan).filter_by(user_id=seed["admin_user"].id).all()
|
||||
assert len(plans) == 4
|
||||
assert {p.on_week for p in plans} == {1, 2, 3, 4}
|
||||
|
||||
def test_plan_on_week_validation_invalid(self, db, seed):
|
||||
"""Test that invalid on_week values raise ValueError."""
|
||||
for week in [0, 5, 10, -1]:
|
||||
with pytest.raises(ValueError):
|
||||
plan = SchedulePlan(
|
||||
user_id=seed["admin_user"].id,
|
||||
slot_type=SlotType.WORK,
|
||||
estimated_duration=30,
|
||||
at_time=time(9, 0),
|
||||
on_day=DayOfWeek.MON,
|
||||
on_week=week, # Invalid
|
||||
is_active=True,
|
||||
)
|
||||
db.add(plan)
|
||||
db.commit()
|
||||
db.rollback()
|
||||
|
||||
def test_plan_duration_validation(self, db, seed):
|
||||
"""Test estimated_duration validation (must be 1-50)."""
|
||||
# Valid bounds
|
||||
plan_min = SchedulePlan(
|
||||
user_id=seed["admin_user"].id,
|
||||
slot_type=SlotType.WORK,
|
||||
estimated_duration=1,
|
||||
at_time=time(8, 0),
|
||||
is_active=True,
|
||||
)
|
||||
db.add(plan_min)
|
||||
|
||||
plan_max = SchedulePlan(
|
||||
user_id=seed["admin_user"].id,
|
||||
slot_type=SlotType.WORK,
|
||||
estimated_duration=50,
|
||||
at_time=time(9, 0),
|
||||
is_active=True,
|
||||
)
|
||||
db.add(plan_max)
|
||||
db.commit()
|
||||
|
||||
assert plan_min.estimated_duration == 1
|
||||
assert plan_max.estimated_duration == 50
|
||||
|
||||
def test_plan_duration_validation_invalid(self, db, seed):
|
||||
"""Test that invalid estimated_duration raises ValueError."""
|
||||
for duration in [0, 51, 100, -10]:
|
||||
with pytest.raises(ValueError):
|
||||
plan = SchedulePlan(
|
||||
user_id=seed["admin_user"].id,
|
||||
slot_type=SlotType.WORK,
|
||||
estimated_duration=duration,
|
||||
at_time=time(9, 0),
|
||||
is_active=True,
|
||||
)
|
||||
db.add(plan)
|
||||
db.commit()
|
||||
db.rollback()
|
||||
|
||||
def test_plan_hierarchy_constraint_month_requires_week(self, db, seed):
|
||||
"""Test validation: on_month requires on_week."""
|
||||
with pytest.raises(ValueError, match="on_month requires on_week"):
|
||||
SchedulePlan(
|
||||
user_id=seed["admin_user"].id,
|
||||
slot_type=SlotType.WORK,
|
||||
estimated_duration=30,
|
||||
at_time=time(9, 0),
|
||||
on_month=MonthOfYear.JAN, # Without on_week
|
||||
is_active=True,
|
||||
)
|
||||
|
||||
def test_plan_hierarchy_constraint_week_requires_day(self, db, seed):
|
||||
"""Test DB constraint: on_week requires on_day."""
|
||||
plan = SchedulePlan(
|
||||
user_id=seed["admin_user"].id,
|
||||
slot_type=SlotType.WORK,
|
||||
estimated_duration=30,
|
||||
at_time=time(9, 0),
|
||||
on_week=1, # Without on_day
|
||||
is_active=True,
|
||||
)
|
||||
db.add(plan)
|
||||
with pytest.raises(IntegrityError):
|
||||
db.commit()
|
||||
|
||||
def test_plan_day_of_week_enum(self, db, seed):
|
||||
"""Test all DayOfWeek enum values."""
|
||||
for day in DayOfWeek:
|
||||
plan = SchedulePlan(
|
||||
user_id=seed["admin_user"].id,
|
||||
slot_type=SlotType.WORK,
|
||||
estimated_duration=10,
|
||||
at_time=time(9, 0),
|
||||
on_day=day,
|
||||
is_active=True,
|
||||
)
|
||||
db.add(plan)
|
||||
db.commit()
|
||||
|
||||
plans = db.query(SchedulePlan).filter_by(user_id=seed["admin_user"].id).all()
|
||||
assert len(plans) == 7
|
||||
assert {p.on_day for p in plans} == set(DayOfWeek)
|
||||
|
||||
def test_plan_month_of_year_enum(self, db, seed):
|
||||
"""Test all MonthOfYear enum values."""
|
||||
for month in MonthOfYear:
|
||||
plan = SchedulePlan(
|
||||
user_id=seed["admin_user"].id,
|
||||
slot_type=SlotType.WORK,
|
||||
estimated_duration=10,
|
||||
at_time=time(9, 0),
|
||||
on_day=DayOfWeek.MON,
|
||||
on_week=1,
|
||||
on_month=month,
|
||||
is_active=True,
|
||||
)
|
||||
db.add(plan)
|
||||
db.commit()
|
||||
|
||||
plans = db.query(SchedulePlan).filter_by(user_id=seed["admin_user"].id).all()
|
||||
assert len(plans) == 12
|
||||
assert {p.on_month for p in plans} == set(MonthOfYear)
|
||||
|
||||
def test_plan_materialized_slots_relationship(self, db, seed):
|
||||
"""Test relationship between SchedulePlan and TimeSlot."""
|
||||
plan = SchedulePlan(
|
||||
user_id=seed["admin_user"].id,
|
||||
slot_type=SlotType.WORK,
|
||||
estimated_duration=30,
|
||||
at_time=time(9, 0),
|
||||
is_active=True,
|
||||
)
|
||||
db.add(plan)
|
||||
db.commit()
|
||||
db.refresh(plan)
|
||||
|
||||
# Create slots linked to the plan
|
||||
for i in range(3):
|
||||
slot = TimeSlot(
|
||||
user_id=seed["admin_user"].id,
|
||||
date=date(2026, 4, 1 + i),
|
||||
slot_type=SlotType.WORK,
|
||||
estimated_duration=30,
|
||||
scheduled_at=time(9, 0),
|
||||
status=SlotStatus.NOT_STARTED,
|
||||
priority=0,
|
||||
plan_id=plan.id,
|
||||
)
|
||||
db.add(slot)
|
||||
db.commit()
|
||||
|
||||
# Refresh to get relationship
|
||||
db.refresh(plan)
|
||||
materialized = plan.materialized_slots.all()
|
||||
assert len(materialized) == 3
|
||||
assert all(s.plan_id == plan.id for s in materialized)
|
||||
|
||||
def test_plan_is_active_default_true(self, db, seed):
|
||||
"""Test that is_active defaults to True."""
|
||||
plan = SchedulePlan(
|
||||
user_id=seed["admin_user"].id,
|
||||
slot_type=SlotType.WORK,
|
||||
estimated_duration=30,
|
||||
at_time=time(9, 0),
|
||||
)
|
||||
db.add(plan)
|
||||
db.commit()
|
||||
db.refresh(plan)
|
||||
|
||||
assert plan.is_active is True
|
||||
|
||||
def test_plan_soft_delete(self, db, seed):
|
||||
"""Test soft delete by setting is_active=False."""
|
||||
plan = SchedulePlan(
|
||||
user_id=seed["admin_user"].id,
|
||||
slot_type=SlotType.WORK,
|
||||
estimated_duration=30,
|
||||
at_time=time(9, 0),
|
||||
is_active=True,
|
||||
)
|
||||
db.add(plan)
|
||||
db.commit()
|
||||
db.refresh(plan)
|
||||
|
||||
# Soft delete
|
||||
plan.is_active = False
|
||||
db.commit()
|
||||
db.refresh(plan)
|
||||
|
||||
assert plan.is_active is False
|
||||
|
||||
def test_plan_timestamps(self, db, seed):
|
||||
"""Test that created_at is set automatically."""
|
||||
plan = SchedulePlan(
|
||||
user_id=seed["admin_user"].id,
|
||||
slot_type=SlotType.WORK,
|
||||
estimated_duration=30,
|
||||
at_time=time(9, 0),
|
||||
is_active=True,
|
||||
)
|
||||
db.add(plan)
|
||||
db.commit()
|
||||
db.refresh(plan)
|
||||
|
||||
assert plan.created_at is not None
|
||||
assert isinstance(plan.created_at, datetime)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Combined Model Tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestCalendarModelsCombined:
|
||||
"""Tests for interactions between TimeSlot and SchedulePlan."""
|
||||
|
||||
def test_plan_to_slots_cascade_behavior(self, db, seed):
|
||||
"""Test that deleting a plan doesn't delete materialized slots."""
|
||||
plan = SchedulePlan(
|
||||
user_id=seed["admin_user"].id,
|
||||
slot_type=SlotType.WORK,
|
||||
estimated_duration=30,
|
||||
at_time=time(9, 0),
|
||||
is_active=True,
|
||||
)
|
||||
db.add(plan)
|
||||
db.commit()
|
||||
db.refresh(plan)
|
||||
|
||||
# Create slots linked to the plan
|
||||
for i in range(3):
|
||||
slot = TimeSlot(
|
||||
user_id=seed["admin_user"].id,
|
||||
date=date(2026, 4, 1 + i),
|
||||
slot_type=SlotType.WORK,
|
||||
estimated_duration=30,
|
||||
scheduled_at=time(9, 0),
|
||||
status=SlotStatus.NOT_STARTED,
|
||||
priority=0,
|
||||
plan_id=plan.id,
|
||||
)
|
||||
db.add(slot)
|
||||
db.commit()
|
||||
|
||||
# Delete the plan (soft delete)
|
||||
plan.is_active = False
|
||||
db.commit()
|
||||
|
||||
# Slots should still exist
|
||||
slots = db.query(TimeSlot).filter_by(user_id=seed["admin_user"].id).all()
|
||||
assert len(slots) == 3
|
||||
# plan_id should remain (not cascade deleted)
|
||||
assert all(s.plan_id == plan.id for s in slots)
|
||||
|
||||
def test_multiple_plans_per_user(self, db, seed):
|
||||
"""Test that a user can have multiple plans."""
|
||||
for i, slot_type in enumerate([SlotType.WORK, SlotType.ON_CALL, SlotType.ENTERTAINMENT]):
|
||||
plan = SchedulePlan(
|
||||
user_id=seed["admin_user"].id,
|
||||
slot_type=slot_type,
|
||||
estimated_duration=30,
|
||||
at_time=time(9 + i, 0),
|
||||
is_active=True,
|
||||
)
|
||||
db.add(plan)
|
||||
db.commit()
|
||||
|
||||
plans = db.query(SchedulePlan).filter_by(
|
||||
user_id=seed["admin_user"].id,
|
||||
is_active=True
|
||||
).all()
|
||||
assert len(plans) == 3
|
||||
|
||||
def test_multiple_slots_per_user(self, db, seed):
|
||||
"""Test that a user can have multiple slots on same day."""
|
||||
target_date = date(2026, 4, 1)
|
||||
for i in range(5):
|
||||
slot = TimeSlot(
|
||||
user_id=seed["admin_user"].id,
|
||||
date=target_date,
|
||||
slot_type=SlotType.WORK,
|
||||
estimated_duration=10,
|
||||
scheduled_at=time(9 + i, 0),
|
||||
status=SlotStatus.NOT_STARTED,
|
||||
priority=i,
|
||||
)
|
||||
db.add(slot)
|
||||
db.commit()
|
||||
|
||||
slots = db.query(TimeSlot).filter_by(
|
||||
user_id=seed["admin_user"].id,
|
||||
date=target_date
|
||||
).all()
|
||||
assert len(slots) == 5
|
||||
# Check ordering by scheduled_at
|
||||
times = [s.scheduled_at for s in sorted(slots, key=lambda x: x.scheduled_at)]
|
||||
assert times == [time(9, 0), time(10, 0), time(11, 0), time(12, 0), time(13, 0)]
|
||||
|
||||
def test_different_users_isolated(self, db, seed):
|
||||
"""Test that users cannot see each other's slots/plans."""
|
||||
# Create plan and slot for admin
|
||||
admin_plan = SchedulePlan(
|
||||
user_id=seed["admin_user"].id,
|
||||
slot_type=SlotType.WORK,
|
||||
estimated_duration=30,
|
||||
at_time=time(9, 0),
|
||||
is_active=True,
|
||||
)
|
||||
db.add(admin_plan)
|
||||
|
||||
admin_slot = TimeSlot(
|
||||
user_id=seed["admin_user"].id,
|
||||
date=date(2026, 4, 1),
|
||||
slot_type=SlotType.WORK,
|
||||
estimated_duration=30,
|
||||
scheduled_at=time(9, 0),
|
||||
status=SlotStatus.NOT_STARTED,
|
||||
priority=0,
|
||||
)
|
||||
db.add(admin_slot)
|
||||
|
||||
# Create plan and slot for dev user
|
||||
dev_plan = SchedulePlan(
|
||||
user_id=seed["dev_user"].id,
|
||||
slot_type=SlotType.ON_CALL,
|
||||
estimated_duration=45,
|
||||
at_time=time(14, 0),
|
||||
is_active=True,
|
||||
)
|
||||
db.add(dev_plan)
|
||||
|
||||
dev_slot = TimeSlot(
|
||||
user_id=seed["dev_user"].id,
|
||||
date=date(2026, 4, 1),
|
||||
slot_type=SlotType.ON_CALL,
|
||||
estimated_duration=45,
|
||||
scheduled_at=time(14, 0),
|
||||
status=SlotStatus.NOT_STARTED,
|
||||
priority=0,
|
||||
)
|
||||
db.add(dev_slot)
|
||||
|
||||
db.commit()
|
||||
|
||||
# Verify isolation
|
||||
admin_slots = db.query(TimeSlot).filter_by(user_id=seed["admin_user"].id).all()
|
||||
dev_slots = db.query(TimeSlot).filter_by(user_id=seed["dev_user"].id).all()
|
||||
|
||||
assert len(admin_slots) == 1
|
||||
assert len(dev_slots) == 1
|
||||
assert admin_slots[0].slot_type == SlotType.WORK
|
||||
assert dev_slots[0].slot_type == SlotType.ON_CALL
|
||||
|
||||
admin_plans = db.query(SchedulePlan).filter_by(user_id=seed["admin_user"].id).all()
|
||||
dev_plans = db.query(SchedulePlan).filter_by(user_id=seed["dev_user"].id).all()
|
||||
|
||||
assert len(admin_plans) == 1
|
||||
assert len(dev_plans) == 1
|
||||
Reference in New Issue
Block a user