From ba1f2e2f787cbdd19abebe40b33029dde5fc8f45 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 18 Jul 2026 13:16:57 +0000 Subject: [PATCH] =?UTF-8?q?Sprint=204=20PR=204.6b=20=E2=80=94=20integratio?= =?UTF-8?q?n=20tests=20for=20conflicts=20router?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Summary: Continues PR 4.6b file-by-file revival of the integration tier. Adds real-HTTP coverage for /api/v1/conflicts against the session-scoped uvicorn api_server. Exercises both the pre-assignment /check endpoint and the admin-only org-wide GET / listing across the three conflict types (already_assigned, time_off, double_booked). Changed files: - tests/integration/test_conflicts.py (new) — 14 tests covering: - POST /conflicts/check: no-conflict happy path, already_assigned blocks, time_off blocks, double_booked warns but can_assign remains true, 404 on missing person, 404 on missing event - GET /conflicts/: empty envelope shape, time_off + double_booked surfacing, person_id filter narrowing, admin gate (volunteer 403), unauth 401/403, cross-org 403, pagination envelope fields Validation: - poetry run pytest tests/integration/test_conflicts.py -v — 14 passed - poetry run pytest tests/integration/ -q — 124 passed - poetry run black tests/integration/test_conflicts.py — clean - poetry run ruff check tests/integration/test_conflicts.py — clean - No OpenAPI snapshot refresh (no endpoints added or changed). Follow-ups: - Remaining integration files to revive: holidays, resources, recurring_events, assignments, audit, analytics, calendar, password_reset, solutions, solver, people, webhooks. --- tests/integration/test_conflicts.py | 343 ++++++++++++++++++++++++++++ 1 file changed, 343 insertions(+) create mode 100644 tests/integration/test_conflicts.py diff --git a/tests/integration/test_conflicts.py b/tests/integration/test_conflicts.py new file mode 100644 index 00000000..267da175 --- /dev/null +++ b/tests/integration/test_conflicts.py @@ -0,0 +1,343 @@ +#!/usr/bin/env python3 +"""Integration tests: conflicts router (Sprint 4 PR 4.6b). + +Tests the /api/v1/conflicts endpoints over real HTTP against the +session-scoped uvicorn api_server: +- POST /conflicts/check - Pre-assignment conflict check +- GET /conflicts/ - Admin-only org-wide conflict listing (ListResponse envelope) +""" + +import random +import time +from datetime import UTC, date, datetime, timedelta + +import httpx +import pytest + + +def _unique(prefix: str) -> str: + return f"{prefix}_{int(time.time() * 1000)}_{random.randint(10000, 99999)}" + + +def _future_window(days: int = 14, hours: int = 2) -> tuple[str, str]: + start = (datetime.now(UTC) + timedelta(days=days)).replace( + hour=10, minute=0, second=0, microsecond=0 + ) + end = start + timedelta(hours=hours) + return start.isoformat(), end.isoformat() + + +@pytest.fixture +def conflicts_org(api_server, api_base): + """Create an org + admin + one volunteer; return clients + ids.""" + marker = _unique("conflicts_org") + org_id = marker + admin_email = f"admin_{marker}@test.com" + vol_email = f"vol_{marker}@test.com" + + bootstrap = httpx.Client() + + org_resp = bootstrap.post( + f"{api_base}/organizations/", + json={"id": org_id, "name": f"Conflicts Setup {marker}", "region": "US", "config": {}}, + ) + assert org_resp.status_code == 201, org_resp.text + + admin_resp = bootstrap.post( + f"{api_base}/auth/signup", + json={ + "org_id": org_id, + "name": "Conflict Admin", + "email": admin_email, + "password": "AdminPass123!", + }, + ) + assert admin_resp.status_code == 201, admin_resp.text + admin_data = admin_resp.json() + assert "admin" in admin_data["roles"] + + vol_resp = bootstrap.post( + f"{api_base}/auth/signup", + json={ + "org_id": org_id, + "name": "Conflict Volunteer", + "email": vol_email, + "password": "VolPass123!", + "roles": ["volunteer"], + }, + ) + assert vol_resp.status_code == 201, vol_resp.text + vol_data = vol_resp.json() + + bootstrap.close() + + admin_client = httpx.Client() + admin_client.headers["Authorization"] = f"Bearer {admin_data['token']}" + + vol_client = httpx.Client() + vol_client.headers["Authorization"] = f"Bearer {vol_data['token']}" + + yield { + "marker": marker, + "org_id": org_id, + "admin_id": admin_data["person_id"], + "admin_client": admin_client, + "vol_id": vol_data["person_id"], + "vol_client": vol_client, + "api_base": api_base, + } + + admin_client.close() + vol_client.close() + + +def _create_event(client, api_base, org_id, *, days: int = 14, event_type: str = "worship") -> dict: + """Helper: create an event via the admin client and return its body.""" + event_id = _unique("evt") + start, end = _future_window(days=days) + resp = client.post( + f"{api_base}/events/", + json={ + "id": event_id, + "org_id": org_id, + "type": event_type, + "start_time": start, + "end_time": end, + "extra_data": {}, + }, + ) + assert resp.status_code == 201, resp.text + return resp.json() + + +def _assign(client, api_base, event_id: str, person_id: str, role: str | None = None) -> dict: + """Helper: admin assigns a person to an event via POST /events/{id}/assignments.""" + resp = client.post( + f"{api_base}/events/{event_id}/assignments", + json={"person_id": person_id, "action": "assign", "role": role}, + ) + assert resp.status_code == 200, resp.text + return resp.json() + + +def _add_timeoff(api_base: str, person_id: str, start: date, end: date) -> None: + """Helper: post a time-off period for a person (unauthenticated endpoint).""" + anon = httpx.Client() + try: + resp = anon.post( + f"{api_base}/availability/{person_id}/timeoff", + json={ + "start_date": start.isoformat(), + "end_date": end.isoformat(), + "reason": "integration test", + }, + ) + assert resp.status_code == 201, resp.text + finally: + anon.close() + + +class TestCheckConflicts: + """POST /conflicts/check — pre-assignment conflict detection.""" + + def test_no_conflict_returns_can_assign(self, conflicts_org): + data = conflicts_org + event = _create_event(data["admin_client"], data["api_base"], data["org_id"]) + + # Unauthenticated for this endpoint (no auth dependency in the router). + anon = httpx.Client() + try: + resp = anon.post( + f"{data['api_base']}/conflicts/check", + json={"person_id": data["vol_id"], "event_id": event["id"]}, + ) + finally: + anon.close() + assert resp.status_code == 200, resp.text + body = resp.json() + assert body["has_conflicts"] is False + assert body["can_assign"] is True + assert body["conflicts"] == [] + + def test_already_assigned_blocks(self, conflicts_org): + data = conflicts_org + event = _create_event(data["admin_client"], data["api_base"], data["org_id"]) + _assign(data["admin_client"], data["api_base"], event["id"], data["vol_id"]) + + resp = httpx.post( + f"{data['api_base']}/conflicts/check", + json={"person_id": data["vol_id"], "event_id": event["id"]}, + ) + assert resp.status_code == 200, resp.text + body = resp.json() + assert body["has_conflicts"] is True + assert body["can_assign"] is False + types = [c["type"] for c in body["conflicts"]] + assert "already_assigned" in types + + def test_time_off_blocks(self, conflicts_org): + data = conflicts_org + # Event 20 days out; time-off spanning that day + event = _create_event(data["admin_client"], data["api_base"], data["org_id"], days=20) + event_start = datetime.fromisoformat(event["start_time"]).date() + _add_timeoff( + data["api_base"], + data["vol_id"], + event_start - timedelta(days=1), + event_start + timedelta(days=1), + ) + + resp = httpx.post( + f"{data['api_base']}/conflicts/check", + json={"person_id": data["vol_id"], "event_id": event["id"]}, + ) + assert resp.status_code == 200, resp.text + body = resp.json() + assert body["has_conflicts"] is True + assert body["can_assign"] is False + types = [c["type"] for c in body["conflicts"]] + assert "time_off" in types + + def test_double_booked_warns_but_allows(self, conflicts_org): + data = conflicts_org + event_a = _create_event(data["admin_client"], data["api_base"], data["org_id"], days=25) + event_b = _create_event(data["admin_client"], data["api_base"], data["org_id"], days=25) + # Both events use the identical future window: overlap guaranteed + _assign(data["admin_client"], data["api_base"], event_a["id"], data["vol_id"]) + + resp = httpx.post( + f"{data['api_base']}/conflicts/check", + json={"person_id": data["vol_id"], "event_id": event_b["id"]}, + ) + assert resp.status_code == 200, resp.text + body = resp.json() + types = [c["type"] for c in body["conflicts"]] + assert "double_booked" in types + # Double-booked is non-blocking per the router contract + assert body["can_assign"] is True + + def test_missing_person_returns_404(self, conflicts_org): + data = conflicts_org + event = _create_event(data["admin_client"], data["api_base"], data["org_id"]) + resp = httpx.post( + f"{data['api_base']}/conflicts/check", + json={"person_id": f"ghost_{int(time.time())}", "event_id": event["id"]}, + ) + assert resp.status_code == 404 + + def test_missing_event_returns_404(self, conflicts_org): + data = conflicts_org + resp = httpx.post( + f"{data['api_base']}/conflicts/check", + json={"person_id": data["vol_id"], "event_id": f"nope_{int(time.time())}"}, + ) + assert resp.status_code == 404 + + +class TestListConflicts: + """GET /conflicts/ — admin-only org-wide conflict listing.""" + + def test_list_empty_when_no_conflicts(self, conflicts_org): + data = conflicts_org + resp = data["admin_client"].get( + f"{data['api_base']}/conflicts/", + params={"org_id": data["org_id"]}, + ) + assert resp.status_code == 200, resp.text + body = resp.json() + assert set(body.keys()) >= {"items", "total", "limit", "offset"} + assert body["total"] == 0 + assert body["items"] == [] + + def test_list_surfaces_time_off_conflict(self, conflicts_org): + data = conflicts_org + event = _create_event(data["admin_client"], data["api_base"], data["org_id"], days=30) + _assign(data["admin_client"], data["api_base"], event["id"], data["vol_id"]) + event_start = datetime.fromisoformat(event["start_time"]).date() + _add_timeoff( + data["api_base"], + data["vol_id"], + event_start - timedelta(days=1), + event_start + timedelta(days=1), + ) + + resp = data["admin_client"].get( + f"{data['api_base']}/conflicts/", + params={"org_id": data["org_id"]}, + ) + assert resp.status_code == 200, resp.text + body = resp.json() + assert body["total"] >= 1 + assert any(c["type"] == "time_off" for c in body["items"]) + + def test_list_surfaces_double_booked(self, conflicts_org): + data = conflicts_org + # Two overlapping events, volunteer assigned to both + event_a = _create_event(data["admin_client"], data["api_base"], data["org_id"], days=35) + event_b = _create_event(data["admin_client"], data["api_base"], data["org_id"], days=35) + _assign(data["admin_client"], data["api_base"], event_a["id"], data["vol_id"]) + _assign(data["admin_client"], data["api_base"], event_b["id"], data["vol_id"]) + + resp = data["admin_client"].get( + f"{data['api_base']}/conflicts/", + params={"org_id": data["org_id"], "person_id": data["vol_id"]}, + ) + assert resp.status_code == 200, resp.text + body = resp.json() + assert body["total"] >= 1 + assert any(c["type"] == "double_booked" for c in body["items"]) + + def test_list_person_filter_narrows_scope(self, conflicts_org): + data = conflicts_org + # Admin's own account has no conflicts; person_id filter yields empty + resp = data["admin_client"].get( + f"{data['api_base']}/conflicts/", + params={"org_id": data["org_id"], "person_id": data["admin_id"]}, + ) + assert resp.status_code == 200, resp.text + body = resp.json() + assert body["total"] == 0 + + def test_list_requires_admin(self, conflicts_org): + data = conflicts_org + resp = data["vol_client"].get( + f"{data['api_base']}/conflicts/", + params={"org_id": data["org_id"]}, + ) + assert resp.status_code == 403 + + def test_list_unauthenticated_rejected(self, conflicts_org): + data = conflicts_org + anon = httpx.Client() + try: + resp = anon.get( + f"{data['api_base']}/conflicts/", + params={"org_id": data["org_id"]}, + ) + finally: + anon.close() + assert resp.status_code in (401, 403) + + def test_list_cross_org_rejected(self, conflicts_org): + data = conflicts_org + # Admin is not a member of some other sentinel org + resp = data["admin_client"].get( + f"{data['api_base']}/conflicts/", + params={"org_id": f"other_{int(time.time())}"}, + ) + assert resp.status_code == 403 + + def test_list_pagination_envelope(self, conflicts_org): + data = conflicts_org + resp = data["admin_client"].get( + f"{data['api_base']}/conflicts/", + params={"org_id": data["org_id"], "limit": 5, "offset": 0}, + ) + assert resp.status_code == 200, resp.text + body = resp.json() + assert body["limit"] == 5 + assert body["offset"] == 0 + + +if __name__ == "__main__": + pytest.main([__file__, "-v"])