From b09a23f011738c6511956c4d67b30b6e7e1d64c7 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 18 Jul 2026 13:10:53 +0000 Subject: [PATCH] =?UTF-8?q?Sprint=204=20PR=204.6b=20=E2=80=94=20integratio?= =?UTF-8?q?n=20tests=20for=20people=20router?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Summary: extend the resurrected tests/integration/ tier with an integration file that drives the /people router over real HTTP against the session-scoped uvicorn api_server fixture. Covers /me (get, update, unauthenticated 403), admin-only create (success, forbidden for volunteer, duplicate id 409), admin-only bulk import (success, missing items 400, forbidden for volunteer), list envelope + q= search across name/email + role filter, get one + 404, PUT with self-vs-admin tenancy and no-role-escalation guard, and admin-only delete. Changed files: tests/integration/test_people.py (new, 19 tests) Validation: poetry run pytest tests/integration/ -q # 129 passed poetry run pytest tests/integration/test_people.py -v # 19 passed poetry run pytest tests/unit -q # 342 passed, 21 skipped poetry run black tests/integration/test_people.py poetry run ruff check tests/integration/test_people.py Follow-ups: continue file-by-file revival with constraints, conflicts, calendar, and solutions routers. --- tests/integration/test_people.py | 322 +++++++++++++++++++++++++++++++ 1 file changed, 322 insertions(+) create mode 100644 tests/integration/test_people.py diff --git a/tests/integration/test_people.py b/tests/integration/test_people.py new file mode 100644 index 00000000..e786d59b --- /dev/null +++ b/tests/integration/test_people.py @@ -0,0 +1,322 @@ +#!/usr/bin/env python3 +"""Integration tests: people router (Sprint 4 PR 4.6b). + +Tests the /api/v1/people endpoints over real HTTP against the +session-scoped uvicorn api_server: +- GET /people/me - Current user profile +- PUT /people/me - Update own profile +- POST /people/ - Create person (admin) +- POST /people/bulk - Bulk import (admin, JSON-array) +- GET /people/ - List (envelope, q=, role filter, org) +- GET /people/{person_id} - Get one +- PUT /people/{person_id} - Update (self or admin, no role escalation) +- DELETE /people/{person_id} - Delete (admin) +""" + +import time + +import httpx +import pytest + + +def _unique(prefix: str) -> str: + return f"{prefix}_{int(time.time() * 1000)}" + + +@pytest.fixture +def people_org(api_server, api_base): + """Create an org + admin + one volunteer; return authed clients + ids.""" + marker = _unique("people_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"People 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": "People 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": "People Volunteer", + "email": vol_email, + "password": "VolPass123!", + "roles": ["volunteer"], + }, + ) + assert vol_resp.status_code == 201, vol_resp.text + vol_data = vol_resp.json() + assert "admin" not in vol_data["roles"] + + 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_email": admin_email, + "admin_client": admin_client, + "vol_id": vol_data["person_id"], + "vol_email": vol_email, + "vol_client": vol_client, + "api_base": api_base, + } + + admin_client.close() + vol_client.close() + + +class TestMeEndpoints: + """GET + PUT /people/me — self-profile flows.""" + + def test_me_returns_current_user(self, people_org): + data = people_org + resp = data["admin_client"].get(f"{data['api_base']}/people/me") + assert resp.status_code == 200 + body = resp.json() + assert body["id"] == data["admin_id"] + assert body["email"] == data["admin_email"] + + def test_me_unauthenticated_rejected(self, api_server, api_base): + resp = httpx.Client().get(f"{api_base}/people/me") + assert resp.status_code == 403 + + def test_update_me_changes_name_and_timezone(self, people_org): + data = people_org + resp = data["vol_client"].put( + f"{data['api_base']}/people/me", + json={"name": "Renamed Volunteer", "timezone": "America/Los_Angeles"}, + ) + assert resp.status_code == 200, resp.text + body = resp.json() + assert body["name"] == "Renamed Volunteer" + assert body["timezone"] == "America/Los_Angeles" + + +class TestCreatePerson: + """POST /people/ — admin-only creation.""" + + def test_create_success(self, people_org): + data = people_org + new_id = _unique("newp") + resp = data["admin_client"].post( + f"{data['api_base']}/people/", + json={ + "id": new_id, + "org_id": data["org_id"], + "name": "Fresh Person", + "email": f"{new_id}@test.com", + "roles": ["volunteer"], + }, + ) + assert resp.status_code == 201, resp.text + assert resp.json()["id"] == new_id + assert resp.json()["org_id"] == data["org_id"] + + def test_create_requires_admin(self, people_org): + data = people_org + new_id = _unique("forbiddenp") + resp = data["vol_client"].post( + f"{data['api_base']}/people/", + json={ + "id": new_id, + "org_id": data["org_id"], + "name": "Not Allowed", + "email": f"{new_id}@test.com", + }, + ) + assert resp.status_code == 403 + + def test_create_duplicate_id_conflict(self, people_org): + data = people_org + resp = data["admin_client"].post( + f"{data['api_base']}/people/", + json={ + "id": data["vol_id"], + "org_id": data["org_id"], + "name": "Duplicate", + "email": f"dup_{int(time.time())}@test.com", + }, + ) + assert resp.status_code == 409 + + +class TestBulkImport: + """POST /people/bulk — admin-only JSON-array bulk import.""" + + def test_bulk_import_success(self, people_org): + data = people_org + marker = _unique("bulkp") + items = [ + { + "id": f"{marker}_{i}", + "org_id": data["org_id"], + "name": f"Bulk Person {i}", + "email": f"{marker}_{i}@test.com", + } + for i in range(3) + ] + + resp = data["admin_client"].post( + f"{data['api_base']}/people/bulk", + params={"org_id": data["org_id"]}, + json={"items": items}, + ) + assert resp.status_code == 200, resp.text + body = resp.json() + assert body["created"] == 3 + assert body["skipped"] == 0 + assert body["errors"] == [] + + def test_bulk_import_missing_items_400(self, people_org): + data = people_org + resp = data["admin_client"].post( + f"{data['api_base']}/people/bulk", + params={"org_id": data["org_id"]}, + json={}, + ) + assert resp.status_code == 400 + + def test_bulk_import_requires_admin(self, people_org): + data = people_org + resp = data["vol_client"].post( + f"{data['api_base']}/people/bulk", + params={"org_id": data["org_id"]}, + json={"items": []}, + ) + assert resp.status_code == 403 + + +class TestListAndGetPeople: + """GET /people/ and GET /people/{id}.""" + + def test_list_envelope(self, people_org): + data = people_org + resp = data["admin_client"].get(f"{data['api_base']}/people/") + assert resp.status_code == 200 + body = resp.json() + assert set(body.keys()) >= {"items", "total", "limit", "offset"} + ids = {p["id"] for p in body["items"]} + assert data["admin_id"] in ids + assert data["vol_id"] in ids + + def test_list_q_filter_by_email(self, people_org): + data = people_org + # Search for a term unique to admin_email; volunteer must not match. + term = data["admin_email"].split("@")[0] + resp = data["admin_client"].get( + f"{data['api_base']}/people/", + params={"org_id": data["org_id"], "q": term}, + ) + assert resp.status_code == 200 + returned = {p["id"] for p in resp.json()["items"]} + assert data["admin_id"] in returned + assert data["vol_id"] not in returned + + def test_list_role_filter(self, people_org): + data = people_org + resp = data["admin_client"].get( + f"{data['api_base']}/people/", + params={"org_id": data["org_id"], "role": "volunteer"}, + ) + assert resp.status_code == 200 + assert all("volunteer" in (p.get("roles") or []) for p in resp.json()["items"]) + + def test_get_existing(self, people_org): + data = people_org + resp = data["admin_client"].get(f"{data['api_base']}/people/{data['vol_id']}") + assert resp.status_code == 200 + assert resp.json()["id"] == data["vol_id"] + + def test_get_missing_returns_404(self, people_org): + data = people_org + resp = data["admin_client"].get(f"{data['api_base']}/people/ghost_{int(time.time())}") + assert resp.status_code == 404 + + +class TestUpdatePerson: + """PUT /people/{id} — self or admin, no volunteer role-escalation.""" + + def test_admin_can_update_other_in_org(self, people_org): + data = people_org + resp = data["admin_client"].put( + f"{data['api_base']}/people/{data['vol_id']}", + json={"name": "Volunteer-Renamed-By-Admin"}, + ) + assert resp.status_code == 200, resp.text + assert resp.json()["name"] == "Volunteer-Renamed-By-Admin" + + def test_volunteer_cannot_edit_others(self, people_org): + data = people_org + resp = data["vol_client"].put( + f"{data['api_base']}/people/{data['admin_id']}", + json={"name": "Hostile Takeover"}, + ) + assert resp.status_code == 403 + + def test_volunteer_cannot_escalate_roles(self, people_org): + data = people_org + # Self-edit but attempting to grant admin + resp = data["vol_client"].put( + f"{data['api_base']}/people/{data['vol_id']}", + json={"roles": ["admin", "volunteer"]}, + ) + assert resp.status_code == 403 + + +class TestDeletePerson: + """DELETE /people/{id} — admin only.""" + + def test_delete_removes_person(self, people_org): + data = people_org + # Create a fresh person to delete so we don't disturb fixture state. + pid = _unique("delp") + create_resp = data["admin_client"].post( + f"{data['api_base']}/people/", + json={ + "id": pid, + "org_id": data["org_id"], + "name": "Delete Me", + "email": f"{pid}@test.com", + }, + ) + assert create_resp.status_code == 201 + + resp = data["admin_client"].delete(f"{data['api_base']}/people/{pid}") + assert resp.status_code == 204 + + follow = data["admin_client"].get(f"{data['api_base']}/people/{pid}") + assert follow.status_code == 404 + + def test_delete_requires_admin(self, people_org): + data = people_org + resp = data["vol_client"].delete(f"{data['api_base']}/people/{data['admin_id']}") + assert resp.status_code == 403 + + +if __name__ == "__main__": + pytest.main([__file__, "-v"])