1081 lines
35 KiB
Python
1081 lines
35 KiB
Python
|
"""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
|