test(calendar): 7 integration flow tests (lifecycle, recurrence, tags, types, invitations, authz, ICS)
This commit is contained in:
chiguyong
parent
d4bc79e409
commit
5b5bd44ac4
|
|
@ -0,0 +1,297 @@
|
|||
"""Integration flow tests for calendar API (Layer 3).
|
||||
|
||||
Tests multi-step API flows via TestClient: event lifecycle, recurrence,
|
||||
tags, event types, invitations, and authz isolation.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import aiosqlite
|
||||
import pytest
|
||||
from fastapi import FastAPI
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from agentkit.calendar.db import init_calendar_db
|
||||
from agentkit.calendar.service import CalendarService
|
||||
from agentkit.server.auth.dependencies import require_authenticated
|
||||
from agentkit.server.auth.models import init_auth_db
|
||||
from agentkit.server.routes import calendar as calendar_routes
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fixtures
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _make_user(user_id: str, username: str = "user") -> dict[str, Any]:
|
||||
return {"user_id": user_id, "username": username, "role": "member"}
|
||||
|
||||
|
||||
async def _seed_user(auth_db_path: Path, user_id: str, username: str, email: str) -> None:
|
||||
async with aiosqlite.connect(str(auth_db_path)) as db:
|
||||
await db.execute(
|
||||
"INSERT INTO users (id, username, email, password_hash, role, is_active, "
|
||||
"is_terminal_authorized, is_server_terminal_authorized, created_at, updated_at) "
|
||||
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
|
||||
(user_id, username, email, "hash", "member", 1, 0, 0,
|
||||
"2026-01-01T00:00:00+00:00", "2026-01-01T00:00:00+00:00"),
|
||||
)
|
||||
await db.commit()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def calendar_db_path(tmp_path: Path) -> Path:
|
||||
path = tmp_path / "test_calendar.db"
|
||||
asyncio.run(init_calendar_db(path))
|
||||
return path
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def auth_db_path(tmp_path: Path) -> Path:
|
||||
path = tmp_path / "test_auth.db"
|
||||
asyncio.run(init_auth_db(path))
|
||||
asyncio.run(_seed_user(path, "user-a", "alice", "alice@example.com"))
|
||||
asyncio.run(_seed_user(path, "user-b", "bob", "bob@example.com"))
|
||||
return path
|
||||
|
||||
|
||||
def _make_app(calendar_db_path: Path, auth_db_path: Path, user: dict[str, Any]) -> FastAPI:
|
||||
service = CalendarService(db_path=calendar_db_path, auth_db_path=auth_db_path)
|
||||
app = FastAPI()
|
||||
app.state.calendar_service = service
|
||||
app.state.auth_db_path = str(auth_db_path)
|
||||
app.include_router(calendar_routes.router, prefix="/api/v1")
|
||||
app.dependency_overrides[require_authenticated] = lambda: user
|
||||
return app
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client_a(calendar_db_path: Path, auth_db_path: Path) -> TestClient:
|
||||
"""Client authenticated as user-a."""
|
||||
return TestClient(_make_app(calendar_db_path, auth_db_path, _make_user("user-a", "alice")))
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client_b(calendar_db_path: Path, auth_db_path: Path) -> TestClient:
|
||||
"""Client authenticated as user-b (shares same DB for authz isolation tests)."""
|
||||
return TestClient(_make_app(calendar_db_path, auth_db_path, _make_user("user-b", "bob")))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 1. Event lifecycle: create -> query -> update -> delete
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_event_lifecycle_create_query_update_delete(client_a: TestClient) -> None:
|
||||
# Create
|
||||
resp = client_a.post("/api/v1/calendar/events", json={
|
||||
"title": "Sprint Planning",
|
||||
"start_time": "2026-07-01T10:00:00+00:00",
|
||||
"end_time": "2026-07-01T11:00:00+00:00",
|
||||
"description": "Quarterly sprint",
|
||||
})
|
||||
assert resp.status_code == 200
|
||||
eid = resp.json()["event"]["id"]
|
||||
assert resp.json()["event"]["title"] == "Sprint Planning"
|
||||
|
||||
# Query single
|
||||
resp = client_a.get(f"/api/v1/calendar/events/{eid}")
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["event"]["title"] == "Sprint Planning"
|
||||
|
||||
# Update
|
||||
resp = client_a.patch(f"/api/v1/calendar/events/{eid}", json={"title": "Sprint Review"})
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["event"]["title"] == "Sprint Review"
|
||||
|
||||
# Delete
|
||||
resp = client_a.delete(f"/api/v1/calendar/events/{eid}")
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["deleted"] is True
|
||||
|
||||
# Verify gone
|
||||
resp = client_a.get(f"/api/v1/calendar/events/{eid}")
|
||||
assert resp.status_code == 404
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 2. Recurring event: RRULE expansion in list
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_recurring_event_expanded_in_list(client_a: TestClient) -> None:
|
||||
resp = client_a.post("/api/v1/calendar/events", json={
|
||||
"title": "Weekly Standup",
|
||||
"start_time": "2026-07-06T09:00:00+00:00", # Monday
|
||||
"end_time": "2026-07-06T09:30:00+00:00",
|
||||
"rrule": "FREQ=WEEKLY;BYDAY=MO;COUNT=4",
|
||||
})
|
||||
assert resp.status_code == 200
|
||||
eid = resp.json()["event"]["id"]
|
||||
|
||||
# List over a 4-week range — should see 4 occurrences
|
||||
# Note: %2B is URL-encoded "+" for timezone offset (+00:00)
|
||||
resp = client_a.get(
|
||||
"/api/v1/calendar/events?start=2026-07-01T00:00:00%2B00:00&end=2026-07-29T00:00:00%2B00:00"
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
occurrences = [e for e in resp.json()["events"] if e["id"] == eid]
|
||||
assert len(occurrences) == 4, f"Expected 4 occurrences, got {len(occurrences)}"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 3. Tag management: create tag -> tag event -> filter by tag
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_tag_create_tag_event_filter(client_a: TestClient) -> None:
|
||||
# Create tag
|
||||
resp = client_a.post("/api/v1/calendar/tags", json={"name": "important", "color": "#FF0000"})
|
||||
assert resp.status_code == 200
|
||||
tag_id = resp.json()["tag"]["id"]
|
||||
|
||||
# Create event with tag
|
||||
resp = client_a.post("/api/v1/calendar/events", json={
|
||||
"title": "Tagged Event",
|
||||
"start_time": "2026-07-15T14:00:00+00:00",
|
||||
"end_time": "2026-07-15T15:00:00+00:00",
|
||||
"tag_ids": [tag_id],
|
||||
})
|
||||
assert resp.status_code == 200
|
||||
eid = resp.json()["event"]["id"]
|
||||
|
||||
# List with tag filter
|
||||
resp = client_a.get(f"/api/v1/calendar/events?tag_id={tag_id}")
|
||||
assert resp.status_code == 200
|
||||
events = resp.json()["events"]
|
||||
assert len(events) == 1
|
||||
assert events[0]["id"] == eid
|
||||
|
||||
# List without filter — should also include
|
||||
resp = client_a.get("/api/v1/calendar/events")
|
||||
assert resp.status_code == 200
|
||||
assert any(e["id"] == eid for e in resp.json()["events"])
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 4. Event type: create type -> create event with type
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_event_type_create_and_use(client_a: TestClient) -> None:
|
||||
# Create event type
|
||||
resp = client_a.post("/api/v1/calendar/event-types", json={"name": "Meeting", "color": "#4A90D9"})
|
||||
assert resp.status_code == 200
|
||||
type_id = resp.json()["event_type"]["id"]
|
||||
|
||||
# Create event with type
|
||||
resp = client_a.post("/api/v1/calendar/events", json={
|
||||
"title": "Team Sync",
|
||||
"start_time": "2026-07-10T10:00:00+00:00",
|
||||
"end_time": "2026-07-10T11:00:00+00:00",
|
||||
"event_type_id": type_id,
|
||||
})
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["event"]["event_type_id"] == type_id
|
||||
|
||||
# List event types
|
||||
resp = client_a.get("/api/v1/calendar/event-types")
|
||||
assert resp.status_code == 200
|
||||
assert any(t["id"] == type_id for t in resp.json()["event_types"])
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 5. Invitation flow: create event -> invite -> respond -> list
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_invitation_flow(client_a: TestClient, client_b: TestClient) -> None:
|
||||
# Create event as user-a (alice)
|
||||
resp = client_a.post("/api/v1/calendar/events", json={
|
||||
"title": "Project Kickoff",
|
||||
"start_time": "2026-08-01T10:00:00+00:00",
|
||||
"end_time": "2026-08-01T11:00:00+00:00",
|
||||
})
|
||||
assert resp.status_code == 200
|
||||
eid = resp.json()["event"]["id"]
|
||||
|
||||
# Alice invites bob@example.com
|
||||
resp = client_a.post(f"/api/v1/calendar/events/{eid}/invitations", json={
|
||||
"invitee_email": "bob@example.com",
|
||||
})
|
||||
assert resp.status_code == 200
|
||||
inv_id = resp.json()["invitation"]["id"]
|
||||
assert resp.json()["invitation"]["status"] == "pending"
|
||||
|
||||
# Bob (invitee) lists his invitations — should see the invite
|
||||
resp = client_b.get("/api/v1/calendar/invitations")
|
||||
assert resp.status_code == 200
|
||||
invitations = resp.json()["invitations"]
|
||||
assert len(invitations) == 1
|
||||
assert invitations[0]["id"] == inv_id
|
||||
|
||||
# Bob responds to the invitation (accept)
|
||||
resp = client_b.post(
|
||||
f"/api/v1/calendar/invitations/{inv_id}/respond", json={"status": "accepted"}
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["status"] == "accepted"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 6. Authz isolation: user A's events invisible to user B
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_authz_isolation_user_a_invisible_to_user_b(
|
||||
client_a: TestClient, client_b: TestClient
|
||||
) -> None:
|
||||
# User A creates an event
|
||||
resp = client_a.post("/api/v1/calendar/events", json={
|
||||
"title": "Alice's Private Event",
|
||||
"start_time": "2026-07-20T10:00:00+00:00",
|
||||
"end_time": "2026-07-20T11:00:00+00:00",
|
||||
})
|
||||
assert resp.status_code == 200
|
||||
eid = resp.json()["event"]["id"]
|
||||
|
||||
# User B lists events — should NOT see Alice's event
|
||||
resp = client_b.get("/api/v1/calendar/events")
|
||||
assert resp.status_code == 200
|
||||
assert all(e["id"] != eid for e in resp.json()["events"]), \
|
||||
"User B should not see User A's event"
|
||||
|
||||
# User B tries to get Alice's event directly -> 403 (exists but not owned)
|
||||
resp = client_b.get(f"/api/v1/calendar/events/{eid}")
|
||||
assert resp.status_code == 403
|
||||
|
||||
# User B tries to delete Alice's event -> 403
|
||||
resp = client_b.delete(f"/api/v1/calendar/events/{eid}")
|
||||
assert resp.status_code == 403
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 7. ICS export
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_ics_export(client_a: TestClient) -> None:
|
||||
# Create an event
|
||||
client_a.post("/api/v1/calendar/events", json={
|
||||
"title": "Export Test",
|
||||
"start_time": "2026-09-01T10:00:00+00:00",
|
||||
"end_time": "2026-09-01T11:00:00+00:00",
|
||||
})
|
||||
|
||||
# Export to ICS
|
||||
resp = client_a.get("/api/v1/calendar/export-ics")
|
||||
assert resp.status_code == 200
|
||||
assert resp.headers["content-type"].startswith("text/calendar")
|
||||
content = resp.content.decode("utf-8")
|
||||
assert "BEGIN:VCALENDAR" in content
|
||||
assert "END:VCALENDAR" in content
|
||||
assert "Export Test" in content
|
||||
Loading…
Reference in New Issue