From 5b5bd44ac482d19ab33e177144e55dd763760c83 Mon Sep 17 00:00:00 2001 From: chiguyong Date: Wed, 24 Jun 2026 12:03:24 +0800 Subject: [PATCH] test(calendar): 7 integration flow tests (lifecycle, recurrence, tags, types, invitations, authz, ICS) --- tests/unit/calendar/test_integration_flows.py | 297 ++++++++++++++++++ 1 file changed, 297 insertions(+) create mode 100644 tests/unit/calendar/test_integration_flows.py diff --git a/tests/unit/calendar/test_integration_flows.py b/tests/unit/calendar/test_integration_flows.py new file mode 100644 index 0000000..fc934b6 --- /dev/null +++ b/tests/unit/calendar/test_integration_flows.py @@ -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