diff --git a/tests/test_calendar_api.py b/tests/test_calendar_api.py new file mode 100644 index 0000000..af4cf61 --- /dev/null +++ b/tests/test_calendar_api.py @@ -0,0 +1,357 @@ +"""Tests for TEST-BE-CAL-001: Calendar API coverage. + +Covers core API surfaces: + - slot create / day view / edit / cancel + - virtual slot edit / cancel materialization flows + - plan create / list / get / edit / cancel + - date-list + - workload-config user/admin endpoints +""" + +from datetime import date, time, timedelta + +from app.models.calendar import ( + SchedulePlan, + SlotStatus, + SlotType, + TimeSlot, + DayOfWeek, +) +from tests.conftest import auth_header + + +FUTURE_DATE = date.today() + timedelta(days=30) +FUTURE_DATE_2 = date.today() + timedelta(days=31) + + +def _create_plan(db, *, user_id: int, slot_type=SlotType.WORK, at_time=time(9, 0), on_day=None, on_week=None): + plan = SchedulePlan( + user_id=user_id, + slot_type=slot_type, + estimated_duration=30, + at_time=at_time, + on_day=on_day, + on_week=on_week, + is_active=True, + ) + db.add(plan) + db.commit() + db.refresh(plan) + return plan + + +def _create_slot(db, *, user_id: int, slot_date: date, scheduled_at=time(9, 0), status=SlotStatus.NOT_STARTED, plan_id=None): + slot = TimeSlot( + user_id=user_id, + date=slot_date, + slot_type=SlotType.WORK, + estimated_duration=30, + scheduled_at=scheduled_at, + status=status, + priority=0, + plan_id=plan_id, + ) + db.add(slot) + db.commit() + db.refresh(slot) + return slot + + +class TestCalendarSlotApi: + def test_create_slot_success(self, client, seed): + r = client.post( + "/calendar/slots", + json={ + "date": FUTURE_DATE.isoformat(), + "slot_type": "work", + "scheduled_at": "09:00:00", + "estimated_duration": 30, + "event_type": "job", + "event_data": {"type": "Task", "code": "TASK-42"}, + "priority": 3, + }, + headers=auth_header(seed["admin_token"]), + ) + assert r.status_code == 201, r.text + data = r.json() + assert data["slot"]["date"] == FUTURE_DATE.isoformat() + assert data["slot"]["slot_type"] == "work" + assert data["slot"]["event_type"] == "job" + assert data["slot"]["event_data"]["code"] == "TASK-42" + assert data["warnings"] == [] + + def test_day_view_returns_real_and_virtual_slots_sorted(self, client, db, seed): + # Real slots + _create_slot(db, user_id=seed["admin_user"].id, slot_date=FUTURE_DATE, scheduled_at=time(11, 0)) + skipped = _create_slot( + db, + user_id=seed["admin_user"].id, + slot_date=FUTURE_DATE, + scheduled_at=time(12, 0), + status=SlotStatus.SKIPPED, + ) + + # Virtual weekly plan matching FUTURE_DATE weekday + weekday_map = { + 0: DayOfWeek.MON, + 1: DayOfWeek.TUE, + 2: DayOfWeek.WED, + 3: DayOfWeek.THU, + 4: DayOfWeek.FRI, + 5: DayOfWeek.SAT, + 6: DayOfWeek.SUN, + } + _create_plan( + db, + user_id=seed["admin_user"].id, + at_time=time(8, 0), + on_day=weekday_map[FUTURE_DATE.weekday()], + ) + + r = client.get( + f"/calendar/day?date={FUTURE_DATE.isoformat()}", + headers=auth_header(seed["admin_token"]), + ) + assert r.status_code == 200, r.text + data = r.json() + assert data["date"] == FUTURE_DATE.isoformat() + assert len(data["slots"]) == 2 + assert [slot["scheduled_at"] for slot in data["slots"]] == ["08:00:00", "11:00:00"] + assert data["slots"][0]["virtual_id"].startswith("plan-") + assert data["slots"][1]["id"] is not None + # skipped slot hidden + assert all(slot.get("id") != skipped.id for slot in data["slots"]) + + def test_edit_real_slot_success(self, client, db, seed): + slot = _create_slot(db, user_id=seed["admin_user"].id, slot_date=FUTURE_DATE, scheduled_at=time(9, 0)) + + r = client.patch( + f"/calendar/slots/{slot.id}", + json={ + "scheduled_at": "10:30:00", + "estimated_duration": 40, + "priority": 7, + }, + headers=auth_header(seed["admin_token"]), + ) + assert r.status_code == 200, r.text + data = r.json() + assert data["slot"]["id"] == slot.id + assert data["slot"]["scheduled_at"] == "10:30:00" + assert data["slot"]["estimated_duration"] == 40 + assert data["slot"]["priority"] == 7 + + def test_edit_virtual_slot_materializes_and_detaches(self, client, db, seed): + weekday_map = { + 0: DayOfWeek.MON, + 1: DayOfWeek.TUE, + 2: DayOfWeek.WED, + 3: DayOfWeek.THU, + 4: DayOfWeek.FRI, + 5: DayOfWeek.SAT, + 6: DayOfWeek.SUN, + } + plan = _create_plan( + db, + user_id=seed["admin_user"].id, + at_time=time(8, 0), + on_day=weekday_map[FUTURE_DATE.weekday()], + ) + virtual_id = f"plan-{plan.id}-{FUTURE_DATE.isoformat()}" + + r = client.patch( + f"/calendar/slots/virtual/{virtual_id}", + json={"scheduled_at": "08:30:00", "priority": 5}, + headers=auth_header(seed["admin_token"]), + ) + assert r.status_code == 200, r.text + data = r.json() + assert data["slot"]["id"] is not None + assert data["slot"]["scheduled_at"] == "08:30:00" + assert data["slot"]["plan_id"] is None + materialized = db.query(TimeSlot).filter(TimeSlot.id == data["slot"]["id"]).first() + assert materialized is not None + assert materialized.plan_id is None + + def test_cancel_real_slot_sets_skipped(self, client, db, seed): + slot = _create_slot(db, user_id=seed["admin_user"].id, slot_date=FUTURE_DATE) + + r = client.post( + f"/calendar/slots/{slot.id}/cancel", + headers=auth_header(seed["admin_token"]), + ) + assert r.status_code == 200, r.text + data = r.json() + assert data["slot"]["status"] == "skipped" + assert data["message"] == "Slot cancelled successfully" + + def test_cancel_virtual_slot_materializes_then_skips(self, client, db, seed): + weekday_map = { + 0: DayOfWeek.MON, + 1: DayOfWeek.TUE, + 2: DayOfWeek.WED, + 3: DayOfWeek.THU, + 4: DayOfWeek.FRI, + 5: DayOfWeek.SAT, + 6: DayOfWeek.SUN, + } + plan = _create_plan( + db, + user_id=seed["admin_user"].id, + at_time=time(8, 0), + on_day=weekday_map[FUTURE_DATE.weekday()], + ) + virtual_id = f"plan-{plan.id}-{FUTURE_DATE.isoformat()}" + + r = client.post( + f"/calendar/slots/virtual/{virtual_id}/cancel", + headers=auth_header(seed["admin_token"]), + ) + assert r.status_code == 200, r.text + data = r.json() + assert data["slot"]["status"] == "skipped" + assert data["slot"]["plan_id"] is None + assert "cancelled" in data["message"].lower() + + def test_date_list_only_returns_future_materialized_dates(self, client, db, seed): + _create_slot(db, user_id=seed["admin_user"].id, slot_date=FUTURE_DATE) + _create_slot(db, user_id=seed["admin_user"].id, slot_date=FUTURE_DATE_2, status=SlotStatus.SKIPPED) + _create_plan(db, user_id=seed["admin_user"].id, at_time=time(8, 0)) # virtual-only, should not appear + + r = client.get("/calendar/dates", headers=auth_header(seed["admin_token"])) + assert r.status_code == 200, r.text + assert r.json()["dates"] == [FUTURE_DATE.isoformat()] + + +class TestCalendarPlanApi: + def test_create_list_get_plan(self, client, seed): + create = client.post( + "/calendar/plans", + json={ + "slot_type": "work", + "estimated_duration": 30, + "at_time": "09:00:00", + "on_day": "mon", + "event_type": "job", + "event_data": {"type": "Task", "code": "TASK-1"}, + }, + headers=auth_header(seed["admin_token"]), + ) + assert create.status_code == 201, create.text + plan = create.json() + assert plan["slot_type"] == "work" + assert plan["on_day"] == "mon" + + listing = client.get("/calendar/plans", headers=auth_header(seed["admin_token"])) + assert listing.status_code == 200, listing.text + assert len(listing.json()["plans"]) == 1 + assert listing.json()["plans"][0]["id"] == plan["id"] + + single = client.get(f"/calendar/plans/{plan['id']}", headers=auth_header(seed["admin_token"])) + assert single.status_code == 200, single.text + assert single.json()["id"] == plan["id"] + assert single.json()["event_data"]["code"] == "TASK-1" + + def test_edit_plan_detaches_future_materialized_slots(self, client, db, seed): + plan = _create_plan(db, user_id=seed["admin_user"].id, at_time=time(9, 0)) + future_slot = _create_slot(db, user_id=seed["admin_user"].id, slot_date=FUTURE_DATE, plan_id=plan.id) + + r = client.patch( + f"/calendar/plans/{plan.id}", + json={"at_time": "10:15:00", "estimated_duration": 25}, + headers=auth_header(seed["admin_token"]), + ) + assert r.status_code == 200, r.text + data = r.json() + assert data["at_time"] == "10:15:00" + assert data["estimated_duration"] == 25 + + db.refresh(future_slot) + assert future_slot.plan_id is None + + def test_cancel_plan_deactivates_and_preserves_past_ids_list(self, client, db, seed): + plan = _create_plan(db, user_id=seed["admin_user"].id, at_time=time(9, 0)) + future_slot = _create_slot(db, user_id=seed["admin_user"].id, slot_date=FUTURE_DATE, plan_id=plan.id) + + r = client.post( + f"/calendar/plans/{plan.id}/cancel", + headers=auth_header(seed["admin_token"]), + ) + assert r.status_code == 200, r.text + data = r.json() + assert data["plan"]["is_active"] is False + assert isinstance(data["preserved_past_slot_ids"], list) + + db.refresh(future_slot) + assert future_slot.plan_id is None + + def test_list_plans_include_inactive(self, client, db, seed): + active = _create_plan(db, user_id=seed["admin_user"].id, at_time=time(9, 0)) + inactive = _create_plan(db, user_id=seed["admin_user"].id, at_time=time(10, 0)) + inactive.is_active = False + db.commit() + + active_only = client.get("/calendar/plans", headers=auth_header(seed["admin_token"])) + assert active_only.status_code == 200 + assert [p["id"] for p in active_only.json()["plans"]] == [active.id] + + with_inactive = client.get( + "/calendar/plans?include_inactive=true", + headers=auth_header(seed["admin_token"]), + ) + assert with_inactive.status_code == 200 + ids = {p["id"] for p in with_inactive.json()["plans"]} + assert ids == {active.id, inactive.id} + + +class TestWorkloadConfigApi: + def test_user_workload_config_put_patch_get(self, client, seed): + put = client.put( + "/calendar/workload-config", + json={ + "daily": {"work": 60, "on_call": 10, "entertainment": 5}, + "weekly": {"work": 300, "on_call": 20, "entertainment": 15}, + "monthly": {"work": 900, "on_call": 60, "entertainment": 45}, + "yearly": {"work": 10000, "on_call": 200, "entertainment": 100}, + }, + headers=auth_header(seed["admin_token"]), + ) + assert put.status_code == 200, put.text + assert put.json()["config"]["daily"]["work"] == 60 + + patch = client.patch( + "/calendar/workload-config", + json={"daily": {"work": 90, "on_call": 10, "entertainment": 5}}, + headers=auth_header(seed["admin_token"]), + ) + assert patch.status_code == 200, patch.text + assert patch.json()["config"]["daily"]["work"] == 90 + assert patch.json()["config"]["weekly"]["work"] == 300 + + get = client.get("/calendar/workload-config", headers=auth_header(seed["admin_token"])) + assert get.status_code == 200, get.text + assert get.json()["config"]["daily"]["work"] == 90 + + def test_admin_can_manage_other_user_workload_config(self, client, seed): + patch = client.patch( + f"/calendar/workload-config/{seed['dev_user'].id}", + json={"daily": {"work": 45, "on_call": 0, "entertainment": 0}}, + headers=auth_header(seed["admin_token"]), + ) + assert patch.status_code == 200, patch.text + assert patch.json()["user_id"] == seed["dev_user"].id + assert patch.json()["config"]["daily"]["work"] == 45 + + get = client.get( + f"/calendar/workload-config/{seed['dev_user'].id}", + headers=auth_header(seed["admin_token"]), + ) + assert get.status_code == 200, get.text + assert get.json()["config"]["daily"]["work"] == 45 + + def test_non_admin_cannot_manage_other_user_workload_config(self, client, seed): + r = client.get( + f"/calendar/workload-config/{seed['admin_user'].id}", + headers=auth_header(seed["dev_token"]), + ) + assert r.status_code == 403, r.text diff --git a/tests/test_calendar_models.py b/tests/test_calendar_models.py new file mode 100644 index 0000000..8ff5e66 --- /dev/null +++ b/tests/test_calendar_models.py @@ -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