Leopoldadmin/leocrm
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
4bc11efc25dbcd5df0327d4ac33d4fe7666e15ec
leocrm/tests/test_calendar.py
T

1081 lines
35 KiB
Python
Raw Blame History

"""Tests for Calendar plugin — calendars, entries, links, subtasks, bulk, kanban, CSV, ICS, resources, recurrence, sharing.
Tests all 29 acceptance criteria.
"""
from __future__ import annotations
import pytest
from tests.conftest import ORIGIN_HEADER, login_client
# ─── AC1: List calendars ───
@pytest.mark.asyncio
async def test_ac1_list_calendars_empty(calendar_authed_client):
"""AC1: GET /api/v1/calendars → 200 + calendar list (empty)."""
client, _ = calendar_authed_client
resp = await client.get("/api/v1/calendars", headers=ORIGIN_HEADER)
assert resp.status_code == 200
assert resp.json() == []
@pytest.mark.asyncio
async def test_ac1_list_calendars(calendar_authed_client):
"""AC1: GET /api/v1/calendars → 200 + calendar list with items."""
client, _ = calendar_authed_client
resp = await client.post(
"/api/v1/calendars",
json={"name": "My Calendar", "color": "#FF0000", "type": "personal"},
headers=ORIGIN_HEADER,
)
assert resp.status_code == 201
resp = await client.get("/api/v1/calendars", headers=ORIGIN_HEADER)
assert resp.status_code == 200
data = resp.json()
assert len(data) == 1
assert data[0]["name"] == "My Calendar"
assert data[0]["color"] == "#FF0000"
# ─── AC2: Create calendar ───
@pytest.mark.asyncio
async def test_ac2_create_calendar(calendar_authed_client):
"""AC2: POST /api/v1/calendars → 201, calendar created."""
client, _ = calendar_authed_client
resp = await client.post(
"/api/v1/calendars",
json={"name": "Work Calendar", "color": "#00FF00", "type": "team"},
headers=ORIGIN_HEADER,
)
assert resp.status_code == 201
data = resp.json()
assert data["name"] == "Work Calendar"
assert data["color"] == "#00FF00"
assert data["type"] == "team"
assert "id" in data
assert "owner_id" in data
# ─── AC3: Update calendar ───
@pytest.mark.asyncio
async def test_ac3_update_calendar(calendar_authed_client):
"""AC3: PATCH /api/v1/calendars/{id} → 200, updated."""
client, _ = calendar_authed_client
resp = await client.post(
"/api/v1/calendars",
json={"name": "Old Name"},
headers=ORIGIN_HEADER,
)
cal_id = resp.json()["id"]
resp = await client.patch(
f"/api/v1/calendars/{cal_id}",
json={"name": "New Name", "color": "#0000FF"},
headers=ORIGIN_HEADER,
)
assert resp.status_code == 200
data = resp.json()
assert data["name"] == "New Name"
assert data["color"] == "#0000FF"
# ─── AC4: Delete calendar ───
@pytest.mark.asyncio
async def test_ac4_delete_calendar(calendar_authed_client):
"""AC4: DELETE /api/v1/calendars/{id} → 204, cascade delete entries."""
client, _ = calendar_authed_client
resp = await client.post(
"/api/v1/calendars",
json={"name": "To Delete"},
headers=ORIGIN_HEADER,
)
cal_id = resp.json()["id"]
# Create an entry in the calendar
resp = await client.post(
"/api/v1/calendar/entries",
json={
"calendar_id": cal_id,
"entry_type": "appointment",
"title": "Meeting",
"start_at": "2026-07-01T10:00:00",
"end_at": "2026-07-01T11:00:00",
},
headers=ORIGIN_HEADER,
)
assert resp.status_code == 201
# Delete calendar
resp = await client.delete(f"/api/v1/calendars/{cal_id}", headers=ORIGIN_HEADER)
assert resp.status_code == 204
# Verify calendar is gone
resp = await client.get("/api/v1/calendars", headers=ORIGIN_HEADER)
assert resp.status_code == 200
assert len(resp.json()) == 0
# Verify entries are also soft-deleted
resp = await client.get(
"/api/v1/calendar/entries?start=2026-06-01&end=2026-08-01",
headers=ORIGIN_HEADER,
)
assert resp.status_code == 200
assert len(resp.json()) == 0
# ─── AC5: Share calendar ───
@pytest.mark.asyncio
async def test_ac5_share_calendar(calendar_authed_client):
"""AC5: POST /api/v1/calendars/{id}/share → 200, calendar shared."""
client, seed = calendar_authed_client
resp = await client.post(
"/api/v1/calendars",
json={"name": "Shared Cal"},
headers=ORIGIN_HEADER,
)
cal_id = resp.json()["id"]
viewer_id = str(seed["viewer_a"].id)
resp = await client.post(
f"/api/v1/calendars/{cal_id}/share",
json={"user_id": viewer_id, "permission": "read"},
headers=ORIGIN_HEADER,
)
assert resp.status_code == 200
data = resp.json()
assert data["permission"] == "read"
assert data["user_id"] == viewer_id
# ─── AC6: Get permissions ───
@pytest.mark.asyncio
async def test_ac6_get_permissions(calendar_authed_client):
"""AC6: GET /api/v1/calendars/{id}/permissions → 200 + permission list."""
client, seed = calendar_authed_client
resp = await client.post(
"/api/v1/calendars",
json={"name": "Perm Cal"},
headers=ORIGIN_HEADER,
)
cal_id = resp.json()["id"]
viewer_id = str(seed["viewer_a"].id)
editor_id = str(seed["editor_a"].id)
await client.post(
f"/api/v1/calendars/{cal_id}/share",
json={"user_id": viewer_id, "permission": "read"},
headers=ORIGIN_HEADER,
)
await client.post(
f"/api/v1/calendars/{cal_id}/share",
json={"user_id": editor_id, "permission": "write"},
headers=ORIGIN_HEADER,
)
resp = await client.get(f"/api/v1/calendars/{cal_id}/permissions", headers=ORIGIN_HEADER)
assert resp.status_code == 200
data = resp.json()
assert len(data) == 2
# ─── AC7: List entries ───
@pytest.mark.asyncio
async def test_ac7_list_entries(calendar_authed_client):
"""AC7: GET /api/v1/calendar/entries?start=...&end=... → 200 + entries in range."""
client, _ = calendar_authed_client
resp = await client.post("/api/v1/calendars", json={"name": "Cal"}, headers=ORIGIN_HEADER)
cal_id = resp.json()["id"]
await client.post(
"/api/v1/calendar/entries",
json={
"calendar_id": cal_id,
"entry_type": "appointment",
"title": "June Meeting",
"start_at": "2026-06-15T10:00:00",
"end_at": "2026-06-15T11:00:00",
},
headers=ORIGIN_HEADER,
)
await client.post(
"/api/v1/calendar/entries",
json={
"calendar_id": cal_id,
"entry_type": "appointment",
"title": "August Meeting",
"start_at": "2026-08-15T10:00:00",
"end_at": "2026-08-15T11:00:00",
},
headers=ORIGIN_HEADER,
)
resp = await client.get(
"/api/v1/calendar/entries?start=2026-06-01T00:00:00&end=2026-07-01T00:00:00",
headers=ORIGIN_HEADER,
)
assert resp.status_code == 200
data = resp.json()
assert len(data) == 1
assert data[0]["title"] == "June Meeting"
# ─── AC8: Create appointment ───
@pytest.mark.asyncio
async def test_ac8_create_appointment(calendar_authed_client):
"""AC8: POST /api/v1/calendar/entries (appointment) → 201, with start_at/end_at."""
client, _ = calendar_authed_client
resp = await client.post("/api/v1/calendars", json={"name": "Cal"}, headers=ORIGIN_HEADER)
cal_id = resp.json()["id"]
resp = await client.post(
"/api/v1/calendar/entries",
json={
"calendar_id": cal_id,
"entry_type": "appointment",
"title": "Team Standup",
"start_at": "2026-07-01T09:00:00",
"end_at": "2026-07-01T09:30:00",
"location": "Room A",
},
headers=ORIGIN_HEADER,
)
assert resp.status_code == 201
data = resp.json()
assert data["entry_type"] == "appointment"
assert data["title"] == "Team Standup"
assert data["start_at"] is not None
assert data["end_at"] is not None
assert data["location"] == "Room A"
# ─── AC9: Create task ───
@pytest.mark.asyncio
async def test_ac9_create_task(calendar_authed_client):
"""AC9: POST /api/v1/calendar/entries (task) → 201, with due_date/priority/status."""
client, _ = calendar_authed_client
resp = await client.post("/api/v1/calendars", json={"name": "Cal"}, headers=ORIGIN_HEADER)
cal_id = resp.json()["id"]
resp = await client.post(
"/api/v1/calendar/entries",
json={
"calendar_id": cal_id,
"entry_type": "task",
"title": "Write report",
"due_date": "2026-07-15T17:00:00",
"priority": "high",
"status": "open",
},
headers=ORIGIN_HEADER,
)
assert resp.status_code == 201
data = resp.json()
assert data["entry_type"] == "task"
assert data["title"] == "Write report"
assert data["priority"] == "high"
assert data["status"] == "open"
assert data["due_date"] is not None
# ─── AC10: Get entry detail ───
@pytest.mark.asyncio
async def test_ac10_get_entry_detail(calendar_authed_client):
"""AC10: GET /api/v1/calendar/entries/{id} → 200 + entry detail with links+subtasks."""
client, _ = calendar_authed_client
resp = await client.post("/api/v1/calendars", json={"name": "Cal"}, headers=ORIGIN_HEADER)
cal_id = resp.json()["id"]
resp = await client.post(
"/api/v1/calendar/entries",
json={
"calendar_id": cal_id,
"entry_type": "appointment",
"title": "Detail Test",
"start_at": "2026-07-01T10:00:00",
"end_at": "2026-07-01T11:00:00",
},
headers=ORIGIN_HEADER,
)
entry_id = resp.json()["id"]
resp = await client.get(f"/api/v1/calendar/entries/{entry_id}", headers=ORIGIN_HEADER)
assert resp.status_code == 200
data = resp.json()
assert data["title"] == "Detail Test"
assert data["links"] == []
assert data["subtasks"] == []
# ─── AC11: Update entry (drag&drop) ───
@pytest.mark.asyncio
async def test_ac11_update_entry_dragdrop(calendar_authed_client):
"""AC11: PATCH /api/v1/calendar/entries/{id} → 200, updated (drag&drop)."""
client, _ = calendar_authed_client
resp = await client.post("/api/v1/calendars", json={"name": "Cal"}, headers=ORIGIN_HEADER)
cal_id = resp.json()["id"]
resp = await client.post(
"/api/v1/calendar/entries",
json={
"calendar_id": cal_id,
"entry_type": "appointment",
"title": "Move Me",
"start_at": "2026-07-01T10:00:00",
"end_at": "2026-07-01T11:00:00",
},
headers=ORIGIN_HEADER,
)
entry_id = resp.json()["id"]
resp = await client.patch(
f"/api/v1/calendar/entries/{entry_id}",
json={"start_at": "2026-07-02T14:00:00", "end_at": "2026-07-02T15:00:00"},
headers=ORIGIN_HEADER,
)
assert resp.status_code == 200
data = resp.json()
assert "2026-07-02" in data["start_at"]
assert "14:00" in data["start_at"]
# ─── AC12: Update entry status ───
@pytest.mark.asyncio
async def test_ac12_update_entry_status(calendar_authed_client):
"""AC12: PATCH /api/v1/calendar/entries/{id} status=done → 200."""
client, _ = calendar_authed_client
resp = await client.post("/api/v1/calendars", json={"name": "Cal"}, headers=ORIGIN_HEADER)
cal_id = resp.json()["id"]
resp = await client.post(
"/api/v1/calendar/entries",
json={
"calendar_id": cal_id,
"entry_type": "task",
"title": "Complete Me",
"due_date": "2026-07-15T17:00:00",
},
headers=ORIGIN_HEADER,
)
entry_id = resp.json()["id"]
resp = await client.patch(
f"/api/v1/calendar/entries/{entry_id}",
json={"status": "done"},
headers=ORIGIN_HEADER,
)
assert resp.status_code == 200
assert resp.json()["status"] == "done"
# ─── AC13: Delete entry ───
@pytest.mark.asyncio
async def test_ac13_delete_entry(calendar_authed_client):
"""AC13: DELETE /api/v1/calendar/entries/{id} → 204."""
client, _ = calendar_authed_client
resp = await client.post("/api/v1/calendars", json={"name": "Cal"}, headers=ORIGIN_HEADER)
cal_id = resp.json()["id"]
resp = await client.post(
"/api/v1/calendar/entries",
json={
"calendar_id": cal_id,
"entry_type": "appointment",
"title": "Delete Me",
"start_at": "2026-07-01T10:00:00",
"end_at": "2026-07-01T11:00:00",
},
headers=ORIGIN_HEADER,
)
entry_id = resp.json()["id"]
resp = await client.delete(f"/api/v1/calendar/entries/{entry_id}", headers=ORIGIN_HEADER)
assert resp.status_code == 204
resp = await client.get(f"/api/v1/calendar/entries/{entry_id}", headers=ORIGIN_HEADER)
assert resp.status_code == 404
# ─── AC14: Link entry ───
@pytest.mark.asyncio
async def test_ac14_link_entry(calendar_authed_client):
"""AC14: POST /api/v1/calendar/entries/{id}/link → 200, linked."""
client, seed = calendar_authed_client
resp = await client.post("/api/v1/calendars", json={"name": "Cal"}, headers=ORIGIN_HEADER)
cal_id = resp.json()["id"]
resp = await client.post(
"/api/v1/calendar/entries",
json={
"calendar_id": cal_id,
"entry_type": "appointment",
"title": "Linked Meeting",
"start_at": "2026-07-01T10:00:00",
"end_at": "2026-07-01T11:00:00",
},
headers=ORIGIN_HEADER,
)
entry_id = resp.json()["id"]
company_id = str(seed["company_a"].id)
resp = await client.post(
f"/api/v1/calendar/entries/{entry_id}/link",
json={"entity_type": "company", "entity_id": company_id},
headers=ORIGIN_HEADER,
)
assert resp.status_code == 200
data = resp.json()
assert data["entity_type"] == "company"
assert data["entity_id"] == company_id
# ─── AC15: Create subtask ───
@pytest.mark.asyncio
async def test_ac15_create_subtask(calendar_authed_client):
"""AC15: POST /api/v1/calendar/entries/{id}/subtasks → 201, subtask created."""
client, _ = calendar_authed_client
resp = await client.post("/api/v1/calendars", json={"name": "Cal"}, headers=ORIGIN_HEADER)
cal_id = resp.json()["id"]
resp = await client.post(
"/api/v1/calendar/entries",
json={
"calendar_id": cal_id,
"entry_type": "task",
"title": "Parent Task",
"due_date": "2026-07-15T17:00:00",
},
headers=ORIGIN_HEADER,
)
entry_id = resp.json()["id"]
resp = await client.post(
f"/api/v1/calendar/entries/{entry_id}/subtasks",
json={"title": "Sub item 1"},
headers=ORIGIN_HEADER,
)
assert resp.status_code == 201
data = resp.json()
assert data["title"] == "Sub item 1"
assert data["completed"] is False
# ─── AC16: Toggle subtask ───
@pytest.mark.asyncio
async def test_ac16_toggle_subtask(calendar_authed_client):
"""AC16: PATCH /api/v1/calendar/entries/{id}/subtasks/{sub_id} → 200, toggled."""
client, _ = calendar_authed_client
resp = await client.post("/api/v1/calendars", json={"name": "Cal"}, headers=ORIGIN_HEADER)
cal_id = resp.json()["id"]
resp = await client.post(
"/api/v1/calendar/entries",
json={
"calendar_id": cal_id,
"entry_type": "task",
"title": "Parent Task",
"due_date": "2026-07-15T17:00:00",
},
headers=ORIGIN_HEADER,
)
entry_id = resp.json()["id"]
resp = await client.post(
f"/api/v1/calendar/entries/{entry_id}/subtasks",
json={"title": "Toggle Me"},
headers=ORIGIN_HEADER,
)
sub_id = resp.json()["id"]
resp = await client.patch(
f"/api/v1/calendar/entries/{entry_id}/subtasks/{sub_id}",
json={"completed": True},
headers=ORIGIN_HEADER,
)
assert resp.status_code == 200
assert resp.json()["completed"] is True
# ─── AC17: Bulk action ───
@pytest.mark.asyncio
async def test_ac17_bulk_status_change(calendar_authed_client):
"""AC17: POST /api/v1/calendar/entries/bulk → 200, bulk status change."""
client, _ = calendar_authed_client
resp = await client.post("/api/v1/calendars", json={"name": "Cal"}, headers=ORIGIN_HEADER)
cal_id = resp.json()["id"]
entry_ids = []
for i in range(3):
resp = await client.post(
"/api/v1/calendar/entries",
json={
"calendar_id": cal_id,
"entry_type": "task",
"title": f"Task {i}",
"due_date": "2026-07-15T17:00:00",
},
headers=ORIGIN_HEADER,
)
entry_ids.append(resp.json()["id"])
resp = await client.post(
"/api/v1/calendar/entries/bulk",
json={"entry_ids": entry_ids, "action": "done"},
headers=ORIGIN_HEADER,
)
assert resp.status_code == 200
data = resp.json()
assert data["action"] == "done"
assert data["affected"] == 3
# ─── AC18: Kanban board ───
@pytest.mark.asyncio
async def test_ac18_kanban(calendar_authed_client):
"""AC18: GET /api/v1/calendar/kanban → 200 + tasks grouped by status."""
client, _ = calendar_authed_client
resp = await client.post("/api/v1/calendars", json={"name": "Cal"}, headers=ORIGIN_HEADER)
cal_id = resp.json()["id"]
await client.post(
"/api/v1/calendar/entries",
json={
"calendar_id": cal_id,
"entry_type": "task",
"title": "Open Task",
"due_date": "2026-07-15T17:00:00",
"status": "open",
},
headers=ORIGIN_HEADER,
)
await client.post(
"/api/v1/calendar/entries",
json={
"calendar_id": cal_id,
"entry_type": "task",
"title": "Done Task",
"due_date": "2026-07-15T17:00:00",
"status": "done",
},
headers=ORIGIN_HEADER,
)
resp = await client.get("/api/v1/calendar/kanban", headers=ORIGIN_HEADER)
assert resp.status_code == 200
data = resp.json()
assert "open" in data
assert "in_progress" in data
assert "done" in data
assert "cancelled" in data
assert len(data["open"]) == 1
assert len(data["done"]) == 1
assert len(data["in_progress"]) == 0
# ─── AC19: CSV export ───
@pytest.mark.asyncio
async def test_ac19_csv_export(calendar_authed_client):
"""AC19: GET /api/v1/calendar/entries/export?format=csv → 200 + CSV."""
client, _ = calendar_authed_client
resp = await client.post("/api/v1/calendars", json={"name": "Cal"}, headers=ORIGIN_HEADER)
cal_id = resp.json()["id"]
await client.post(
"/api/v1/calendar/entries",
json={
"calendar_id": cal_id,
"entry_type": "appointment",
"title": "Export Test",
"start_at": "2026-07-01T10:00:00",
"end_at": "2026-07-01T11:00:00",
},
headers=ORIGIN_HEADER,
)
resp = await client.get("/api/v1/calendar/entries/export?format=csv", headers=ORIGIN_HEADER)
assert resp.status_code == 200
text = resp.text
assert "id" in text
assert "Export Test" in text
# ─── AC20: ICS feed valid token ───
@pytest.mark.asyncio
async def test_ac20_ics_feed_valid_token(calendar_authed_client):
"""AC20: GET /api/v1/calendar/{calendar_id}/ics-feed?token=valid → 200 + text/calendar."""
client, _ = calendar_authed_client
resp = await client.post("/api/v1/calendars", json={"name": "Cal"}, headers=ORIGIN_HEADER)
cal_id = resp.json()["id"]
await client.post(
"/api/v1/calendar/entries",
json={
"calendar_id": cal_id,
"entry_type": "appointment",
"title": "ICS Event",
"start_at": "2026-07-01T10:00:00",
"end_at": "2026-07-01T11:00:00",
},
headers=ORIGIN_HEADER,
)
# First call without token generates token and returns 401
resp = await client.get(f"/api/v1/calendar/{cal_id}/ics-feed", headers=ORIGIN_HEADER)
assert resp.status_code == 401
# Should have no token, so 401
@pytest.mark.asyncio
async def test_ac20_ics_feed_with_token(calendar_authed_client):
"""AC20: GET /api/v1/calendar/{calendar_id}/ics-feed?token=valid → 200 + text/calendar."""
client, _ = calendar_authed_client
resp = await client.post("/api/v1/calendars", json={"name": "Cal"}, headers=ORIGIN_HEADER)
cal_id = resp.json()["id"]
await client.post(
"/api/v1/calendar/entries",
json={
"calendar_id": cal_id,
"entry_type": "appointment",
"title": "ICS Event",
"start_at": "2026-07-01T10:00:00",
"end_at": "2026-07-01T11:00:00",
},
headers=ORIGIN_HEADER,
)
# First call without token generates token and returns 401
resp = await client.get(f"/api/v1/calendar/{cal_id}/ics-feed", headers=ORIGIN_HEADER)
assert resp.status_code == 401
# Get the token from the calendar
resp = await client.get("/api/v1/calendars", headers=ORIGIN_HEADER)
cal_data = resp.json()[0]
token = cal_data["ics_token"]
assert token is not None
# Call with valid token
resp = await client.get(
f"/api/v1/calendar/{cal_id}/ics-feed?token={token}",
headers=ORIGIN_HEADER,
)
assert resp.status_code == 200
assert "text/calendar" in resp.headers.get("content-type", "")
assert "BEGIN:VCALENDAR" in resp.text
assert "ICS Event" in resp.text
# ─── AC21: ICS feed invalid token ───
@pytest.mark.asyncio
async def test_ac21_ics_feed_invalid_token(calendar_authed_client):
"""AC21: GET /api/v1/calendar/{calendar_id}/ics-feed?token=invalid → 401."""
client, _ = calendar_authed_client
resp = await client.post("/api/v1/calendars", json={"name": "Cal"}, headers=ORIGIN_HEADER)
cal_id = resp.json()["id"]
resp = await client.get(
f"/api/v1/calendar/{cal_id}/ics-feed?token=invalidtoken",
headers=ORIGIN_HEADER,
)
assert resp.status_code == 401
# ─── AC22: ICS import ───
@pytest.mark.asyncio
async def test_ac22_ics_import(calendar_authed_client):
"""AC22: POST /api/v1/calendar/import mit .ics file → 200 + import result."""
client, _ = calendar_authed_client
resp = await client.post(
"/api/v1/calendars", json={"name": "Import Cal"}, headers=ORIGIN_HEADER
)
cal_id = resp.json()["id"]
ics_content = """BEGIN:VCALENDAR
VERSION:2.0
PRODID:-//Test//EN
BEGIN:VEVENT
UID:test1@test.com
DTSTART:20260701T100000Z
DTEND:20260701T110000Z
SUMMARY:Imported Meeting
DESCRIPTION:Test description
LOCATION:Room B
END:VEVENT
BEGIN:VEVENT
UID:test2@test.com
DTSTART:20260702T140000Z
DTEND:20260702T150000Z
SUMMARY:Second Event
END:VEVENT
END:VCALENDAR
"""
resp = await client.post(
f"/api/v1/calendar/import?calendar_id={cal_id}",
files={"file": ("test.ics", ics_content, "text/calendar")},
headers=ORIGIN_HEADER,
)
assert resp.status_code == 200
data = resp.json()
assert data["entries_created"] == 2
# ─── AC23: Resource creation (admin only) ───
@pytest.mark.asyncio
async def test_ac23_create_resource_admin(calendar_authed_client):
"""AC23: POST /api/v1/resources → 201 (admin only)."""
client, _ = calendar_authed_client
resp = await client.post(
"/api/v1/resources",
json={"name": "Meeting Room A", "type": "room"},
headers=ORIGIN_HEADER,
)
assert resp.status_code == 201
data = resp.json()
assert data["name"] == "Meeting Room A"
assert data["type"] == "room"
@pytest.mark.asyncio
async def test_ac23_create_resource_non_admin(calendar_app, db_session):
"""AC23: POST /api/v1/resources → 403 for non-admin."""
from httpx import ASGITransport, AsyncClient
from tests.conftest import seed_tenant_and_users
await seed_tenant_and_users(db_session)
transport = ASGITransport(app=calendar_app)
async with AsyncClient(transport=transport, base_url="http://test") as c:
await login_client(c, "viewer@tenanta.com")
# Install + activate plugin
await c.post("/api/v1/plugins/calendar/install", headers=ORIGIN_HEADER)
await c.post("/api/v1/plugins/calendar/activate", headers=ORIGIN_HEADER)
resp = await c.post(
"/api/v1/resources",
json={"name": "Projector", "type": "equipment"},
headers=ORIGIN_HEADER,
)
assert resp.status_code == 403
# ─── AC24: Book resource ───
@pytest.mark.asyncio
async def test_ac24_book_resource(calendar_authed_client):
"""AC24: POST /api/v1/calendar/entries/{id}/book-resource → 200."""
client, _ = calendar_authed_client
resp = await client.post("/api/v1/calendars", json={"name": "Cal"}, headers=ORIGIN_HEADER)
cal_id = resp.json()["id"]
resp = await client.post(
"/api/v1/calendar/entries",
json={
"calendar_id": cal_id,
"entry_type": "appointment",
"title": "Booking Test",
"start_at": "2026-07-01T10:00:00",
"end_at": "2026-07-01T11:00:00",
},
headers=ORIGIN_HEADER,
)
entry_id = resp.json()["id"]
resp = await client.post(
"/api/v1/resources",
json={"name": "Room X", "type": "room"},
headers=ORIGIN_HEADER,
)
resource_id = resp.json()["id"]
resp = await client.post(
f"/api/v1/calendar/entries/{entry_id}/book-resource",
json={"resource_id": resource_id},
headers=ORIGIN_HEADER,
)
assert resp.status_code == 200
data = resp.json()
assert data["resource_id"] == resource_id
assert data["entry_id"] == entry_id
# ─── AC25: Book resource conflict ───
@pytest.mark.asyncio
async def test_ac25_book_resource_conflict(calendar_authed_client):
"""AC25: POST /api/v1/calendar/entries/{id}/book-resource (conflict) → 409."""
client, _ = calendar_authed_client
resp = await client.post("/api/v1/calendars", json={"name": "Cal"}, headers=ORIGIN_HEADER)
cal_id = resp.json()["id"]
resp = await client.post(
"/api/v1/resources",
json={"name": "Room Y", "type": "room"},
headers=ORIGIN_HEADER,
)
resource_id = resp.json()["id"]
# First booking
resp = await client.post(
"/api/v1/calendar/entries",
json={
"calendar_id": cal_id,
"entry_type": "appointment",
"title": "First Booking",
"start_at": "2026-07-01T10:00:00",
"end_at": "2026-07-01T11:00:00",
},
headers=ORIGIN_HEADER,
)
entry1_id = resp.json()["id"]
await client.post(
f"/api/v1/calendar/entries/{entry1_id}/book-resource",
json={"resource_id": resource_id},
headers=ORIGIN_HEADER,
)
# Second overlapping booking → 409
resp = await client.post(
"/api/v1/calendar/entries",
json={
"calendar_id": cal_id,
"entry_type": "appointment",
"title": "Conflicting Booking",
"start_at": "2026-07-01T10:30:00",
"end_at": "2026-07-01T11:30:00",
},
headers=ORIGIN_HEADER,
)
entry2_id = resp.json()["id"]
resp = await client.post(
f"/api/v1/calendar/entries/{entry2_id}/book-resource",
json={"resource_id": resource_id},
headers=ORIGIN_HEADER,
)
assert resp.status_code == 409
# ─── AC26: Recurrence weekly ───
@pytest.mark.asyncio
async def test_ac26_recurrence_weekly(calendar_authed_client):
"""AC26: Recurrence: weekly entry generates correct occurrences for date range query."""
client, _ = calendar_authed_client
resp = await client.post("/api/v1/calendars", json={"name": "Cal"}, headers=ORIGIN_HEADER)
cal_id = resp.json()["id"]
resp = await client.post(
"/api/v1/calendar/entries",
json={
"calendar_id": cal_id,
"entry_type": "appointment",
"title": "Weekly Standup",
"start_at": "2026-06-02T09:00:00", # Tuesday
"end_at": "2026-06-02T09:30:00",
"recurrence": {"pattern": "weekly"},
},
headers=ORIGIN_HEADER,
)
assert resp.status_code == 201
# Query range covering 3 weeks
resp = await client.get(
"/api/v1/calendar/entries?start=2026-06-01T00:00:00&end=2026-06-22T00:00:00",
headers=ORIGIN_HEADER,
)
assert resp.status_code == 200
data = resp.json()
# Should have occurrences for June 2, 9, 16 (3 Tuesdays)
assert len(data) == 3
# ─── AC27: Recurrence exception ───
@pytest.mark.asyncio
async def test_ac27_recurrence_exception(calendar_authed_client):
"""AC27: Recurrence: exception date excluded from occurrences."""
client, _ = calendar_authed_client
resp = await client.post("/api/v1/calendars", json={"name": "Cal"}, headers=ORIGIN_HEADER)
cal_id = resp.json()["id"]
resp = await client.post(
"/api/v1/calendar/entries",
json={
"calendar_id": cal_id,
"entry_type": "appointment",
"title": "Daily Sync",
"start_at": "2026-06-01T09:00:00",
"end_at": "2026-06-01T09:30:00",
"recurrence": {
"pattern": "daily",
"exceptions": ["2026-06-02"],
},
},
headers=ORIGIN_HEADER,
)
assert resp.status_code == 201
resp = await client.get(
"/api/v1/calendar/entries?start=2026-06-01T00:00:00&end=2026-06-05T00:00:00",
headers=ORIGIN_HEADER,
)
assert resp.status_code == 200
data = resp.json()
# June 1, 3, 4 → 3 occurrences (June 2 excluded)
assert len(data) == 3
# ─── AC28: Reminder ARQ job ───
@pytest.mark.asyncio
async def test_ac28_reminder_scheduled(calendar_authed_client):
"""AC28: Reminder: ARQ job scheduled when reminder JSONB set."""
client, _ = calendar_authed_client
resp = await client.post("/api/v1/calendars", json={"name": "Cal"}, headers=ORIGIN_HEADER)
cal_id = resp.json()["id"]
resp = await client.post(
"/api/v1/calendar/entries",
json={
"calendar_id": cal_id,
"entry_type": "appointment",
"title": "Reminded Meeting",
"start_at": "2026-07-01T10:00:00",
"end_at": "2026-07-01T11:00:00",
"reminder": {"value": 15, "unit": "minutes", "channel": "in_app"},
},
headers=ORIGIN_HEADER,
)
assert resp.status_code == 201
data = resp.json()
assert data["reminder"] is not None
assert data["reminder"]["value"] == 15
assert data["reminder"]["channel"] == "in_app"
# ─── AC29: Calendar share read-only ───
@pytest.mark.asyncio
async def test_ac29_share_readonly_cannot_edit(calendar_app, db_session):
"""AC29: Calendar share: user with read permission can view, cannot edit (403)."""
from httpx import ASGITransport, AsyncClient
from tests.conftest import seed_tenant_and_users
seed = await seed_tenant_and_users(db_session)
transport = ASGITransport(app=calendar_app)
async with AsyncClient(transport=transport, base_url="http://test") as admin_c:
await login_client(admin_c, "admin@tenanta.com")
await admin_c.post("/api/v1/plugins/calendar/install", headers=ORIGIN_HEADER)
await admin_c.post("/api/v1/plugins/calendar/activate", headers=ORIGIN_HEADER)
# Create calendar as admin
resp = await admin_c.post(
"/api/v1/calendars",
json={"name": "Shared Cal"},
headers=ORIGIN_HEADER,
)
cal_id = resp.json()["id"]
# Share with viewer as read-only
viewer_id = str(seed["viewer_a"].id)
await admin_c.post(
f"/api/v1/calendars/{cal_id}/share",
json={"user_id": viewer_id, "permission": "read"},
headers=ORIGIN_HEADER,
)
# Create an entry as admin
resp = await admin_c.post(
"/api/v1/calendar/entries",
json={
"calendar_id": cal_id,
"entry_type": "appointment",
"title": "Shared Event",
"start_at": "2026-07-01T10:00:00",
"end_at": "2026-07-01T11:00:00",
},
headers=ORIGIN_HEADER,
)
entry_id = resp.json()["id"]
# Now login as viewer
async with AsyncClient(transport=transport, base_url="http://test") as viewer_c:
await login_client(viewer_c, "viewer@tenanta.com")
# Can view entries
resp = await viewer_c.get("/api/v1/calendar/entries", headers=ORIGIN_HEADER)
assert resp.status_code == 200
assert len(resp.json()) >= 1
# Cannot edit entry (no write permission)
resp = await viewer_c.patch(
f"/api/v1/calendar/entries/{entry_id}",
json={"title": "Hacked"},
headers=ORIGIN_HEADER,
)
assert resp.status_code == 403
# ─── AC30: Private entry visibility ───
@pytest.mark.asyncio
async def test_ac30_private_entry_visibility(calendar_app, db_session):
"""AC30: Private subtype: only owner+admin can see entry."""
from httpx import ASGITransport, AsyncClient
from tests.conftest import seed_tenant_and_users
await seed_tenant_and_users(db_session)
transport = ASGITransport(app=calendar_app)
# Admin creates a private entry
async with AsyncClient(transport=transport, base_url="http://test") as admin_c:
await login_client(admin_c, "admin@tenanta.com")
await admin_c.post("/api/v1/plugins/calendar/install", headers=ORIGIN_HEADER)
await admin_c.post("/api/v1/plugins/calendar/activate", headers=ORIGIN_HEADER)
resp = await admin_c.post(
"/api/v1/calendars",
json={"name": "Cal"},
headers=ORIGIN_HEADER,
)
cal_id = resp.json()["id"]
resp = await admin_c.post(
"/api/v1/calendar/entries",
json={
"calendar_id": cal_id,
"entry_type": "appointment",
"title": "Private Meeting",
"subtype": "private",
"start_at": "2026-07-01T10:00:00",
"end_at": "2026-07-01T11:00:00",
},
headers=ORIGIN_HEADER,
)
entry_id = resp.json()["id"]
# Admin can see private entries
resp = await admin_c.get("/api/v1/calendar/entries", headers=ORIGIN_HEADER)
assert resp.status_code == 200
titles = [e["title"] for e in resp.json()]
assert "Private Meeting" in titles
# Viewer cannot see private entries
async with AsyncClient(transport=transport, base_url="http://test") as viewer_c:
await login_client(viewer_c, "viewer@tenanta.com")
resp = await viewer_c.get("/api/v1/calendar/entries", headers=ORIGIN_HEADER)
assert resp.status_code == 200
titles = [e["title"] for e in resp.json()]
assert "Private Meeting" not in titles
# Direct access to private entry → 403
resp = await viewer_c.get(f"/api/v1/calendar/entries/{entry_id}", headers=ORIGIN_HEADER)
assert resp.status_code == 403
Reference in New Issue View Git Blame Copy Permalink