358 lines
13 KiB
Python
358 lines
13 KiB
Python
"""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
|