From 5303580107d5e4458225f14ea3581774ded1d955 Mon Sep 17 00:00:00 2001 From: ilia Date: Tue, 14 Jul 2026 18:23:15 -0400 Subject: [PATCH] test: add cross-endpoint integration and error-handling suites Adds test_api_integration.py (person identification, tag+search, favorites+search, bulk favorites, and role-permission-grant workflows chained across multiple endpoints) and test_api_errors.py (404/401/403/422 consistency sweeps, SQL-injection/XSS-style input handling, large-payload handling) per the two suites the existing tests/API_TEST_PLAN.md already documented but never implemented. Along the way, found and documented (via a passing test, not a fix) that GET /api/v1/faces/unidentified has no auth dependency, unlike its sibling routes. --- tests/test_api_errors.py | 195 ++++++++++++++++++++++++++++++++++ tests/test_api_integration.py | 171 +++++++++++++++++++++++++++++ 2 files changed, 366 insertions(+) create mode 100644 tests/test_api_errors.py create mode 100644 tests/test_api_integration.py diff --git a/tests/test_api_errors.py b/tests/test_api_errors.py new file mode 100644 index 0000000..ab38814 --- /dev/null +++ b/tests/test_api_errors.py @@ -0,0 +1,195 @@ +"""Cross-endpoint error-handling and edge-case sweeps. + +Complements the per-endpoint success/failure tests in the other +`test_api_*.py` files by checking that error *shapes* (status codes, no +crashes, no data corruption) are consistent across the API surface, and by +probing a few classic edge cases (injection-style input, large payloads). +""" + +from __future__ import annotations + + +class TestNotFoundResponses: + """A made-up ID should be a clean 404 everywhere, never a 500.""" + + def test_person_not_found(self, test_client, admin_user, auth_headers): + response = test_client.get("/api/v1/people/999999", headers=auth_headers) + assert response.status_code == 404 + + def test_photo_not_found(self, test_client, admin_user, auth_headers): + response = test_client.get("/api/v1/photos/999999", headers=auth_headers) + assert response.status_code == 404 + + def test_face_similar_not_found(self, test_client, admin_user, auth_headers): + response = test_client.get("/api/v1/faces/999999/similar", headers=auth_headers) + assert response.status_code == 404 + + def test_user_not_found(self, test_client, admin_user, auth_headers): + response = test_client.get("/api/v1/users/999999", headers=auth_headers) + assert response.status_code == 404 + + def test_video_people_not_found(self, test_client): + response = test_client.get("/api/v1/videos/999999/people") + assert response.status_code == 404 + + def test_tag_update_not_found(self, test_client, admin_user, auth_headers): + response = test_client.put( + "/api/v1/tags/999999", json={"tag_name": "renamed"}, headers=auth_headers + ) + assert response.status_code == 400 # ValueError from the service, not a 404 tag lookup + + +class TestUnauthorizedResponses: + """Protected endpoints should uniformly return 401 when no token is sent.""" + + @staticmethod + def _cases(): + return [ + ("get", "/api/v1/photos", {"params": {"search_type": "no_faces"}}), + ("get", "/api/v1/users", {}), + ("get", "/api/v1/role-permissions", {}), + ("get", "/api/v1/videos", {}), + ("post", "/api/v1/log/click", {"json": {"page": "/x", "element_type": "button"}}), + ] + + def test_all_protected_endpoints_reject_missing_token(self, test_client): + for method, path, kwargs in self._cases(): + response = getattr(test_client, method)(path, **kwargs) + assert response.status_code == 401, f"{method.upper()} {path} did not return 401" + + def test_malformed_bearer_token_rejected(self, test_client): + response = test_client.get( + "/api/v1/users", + headers={"Authorization": "Bearer not-a-real-jwt"}, + ) + assert response.status_code == 401 + + def test_unidentified_faces_endpoint_has_no_auth_dependency(self, test_client): + """Documents current behavior: unlike almost every other /api/v1/faces/* + route, this one has no `Depends(get_current_user)`, so it's reachable + without a token. Flagging via a passing test (not asserting 401) so a + future auth addition here is a deliberate, visible change.""" + response = test_client.get("/api/v1/faces/unidentified") + assert response.status_code == 200 + + +class TestForbiddenResponses: + """Admin-only endpoints should uniformly return 403 for a regular user.""" + + @staticmethod + def _cases(): + return [ + ("get", "/api/v1/users", {}), + ("get", "/api/v1/role-permissions", {}), + ("post", "/api/v1/pending-identifications/clear-denied", {}), + ("post", "/api/v1/reported-photos/cleanup", {}), + ] + + def test_all_admin_only_endpoints_reject_regular_user( + self, test_client, admin_user, regular_auth_headers + ): + for method, path, kwargs in self._cases(): + kwargs["headers"] = regular_auth_headers + response = getattr(test_client, method)(path, **kwargs) + assert response.status_code == 403, f"{method.upper()} {path} did not return 403" + + +class TestValidationResponses: + def test_malformed_json_types_rejected_with_422( + self, test_client, admin_user, auth_headers + ): + """photo_ids should be a list of ints, not a bare string.""" + response = test_client.post( + "/api/v1/photos/bulk-add-favorites", + json={"photo_ids": "not-a-list"}, + headers=auth_headers, + ) + assert response.status_code == 422 + + def test_missing_required_body_field_rejected_with_422( + self, test_client, admin_user, auth_headers + ): + response = test_client.post("/api/v1/tags", json={}, headers=auth_headers) + assert response.status_code == 422 + + def test_empty_bulk_delete_list_rejected_with_400( + self, test_client, admin_user, auth_headers + ): + response = test_client.post( + "/api/v1/photos/bulk-delete", json={"photo_ids": []}, headers=auth_headers + ) + assert response.status_code == 400 + + +class TestInjectionAndUnsafeInputHandling: + """The API uses parameterized SQL/ORM everywhere; these are smoke checks + that hostile-looking strings are treated as plain data, not executed. + """ + + def test_sql_injection_style_tag_name_is_stored_literally( + self, test_client, admin_user, auth_headers + ): + payload = "'; DROP TABLE tags; --" + create_resp = test_client.post( + "/api/v1/tags", json={"tag_name": payload}, headers=auth_headers + ) + assert create_resp.status_code == 200 + assert create_resp.json()["tag_name"] == payload + + # Table must still exist and be queryable. + list_resp = test_client.get("/api/v1/tags", headers=auth_headers) + assert list_resp.status_code == 200 + assert any(t["tag_name"] == payload for t in list_resp.json()["items"]) + + def test_script_tag_in_person_name_is_stored_literally_not_executed( + self, test_client, admin_user, auth_headers + ): + payload = "" + response = test_client.post( + "/api/v1/people", + json={"first_name": payload, "last_name": "XSS"}, + headers=auth_headers, + ) + assert response.status_code == 201 + assert response.json()["first_name"] == payload + + def test_sql_injection_style_search_param_returns_normal_response( + self, test_client, admin_user, auth_headers + ): + response = test_client.get( + "/api/v1/photos", + params={"search_type": "name", "person_name": "'; DROP TABLE photos; --"}, + headers=auth_headers, + ) + # No matches expected, but must not error out or corrupt data. + assert response.status_code == 200 + assert response.json()["items"] == [] + + +class TestLargePayloadHandling: + def test_large_bulk_favorites_request_handled_gracefully( + self, test_client, admin_user, auth_headers + ): + """A large list of nonexistent photo IDs should 404 cleanly, not hang or crash.""" + bogus_ids = list(range(900000, 900500)) + response = test_client.post( + "/api/v1/photos/bulk-add-favorites", + json={"photo_ids": bogus_ids}, + headers=auth_headers, + ) + assert response.status_code == 404 + + def test_many_decisions_in_one_review_request( + self, test_client, admin_user, auth_headers + ): + """Reviewing a batch of nonexistent pending-linkage IDs shouldn't crash; + every decision should surface as an individual error entry.""" + decisions = [{"id": i, "decision": "deny"} for i in range(900000, 900200)] + response = test_client.post( + "/api/v1/pending-linkages/review", + json={"decisions": decisions}, + headers=auth_headers, + ) + assert response.status_code == 200 + body = response.json() + assert len(body["errors"]) == len(decisions) diff --git a/tests/test_api_integration.py b/tests/test_api_integration.py new file mode 100644 index 0000000..5b1c63e --- /dev/null +++ b/tests/test_api_integration.py @@ -0,0 +1,171 @@ +"""Cross-endpoint workflow ("integration") tests. + +Unlike the other `test_api_*.py` files (one endpoint's success/failure paths +per test), these chain several endpoints together the way a real client +would, to catch regressions that only show up when features interact. + +Workflows that need Redis/RQ (photo import, face processing, auto-match) or +DeepFace (similarity) are intentionally out of scope here — this sandbox has +neither running, and those are already covered where feasible in +`test_api_faces.py`/`test_api_jobs.py`. +""" + +from __future__ import annotations + + +class TestPersonIdentificationWorkflow: + def test_create_person_then_identify_face_removes_it_from_unidentified( + self, test_client, admin_user, auth_headers, test_face + ): + create_resp = test_client.post( + "/api/v1/people", + json={"first_name": "Alice", "last_name": "Wonder"}, + headers=auth_headers, + ) + assert create_resp.status_code == 201 + person_id = create_resp.json()["id"] + + before = test_client.get("/api/v1/faces/unidentified", headers=auth_headers) + assert test_face.id in {item["id"] for item in before.json()["items"]} + + identify_resp = test_client.post( + f"/api/v1/faces/{test_face.id}/identify", + json={"person_id": person_id}, + headers=auth_headers, + ) + assert identify_resp.status_code == 200 + + after = test_client.get("/api/v1/faces/unidentified", headers=auth_headers) + assert test_face.id not in {item["id"] for item in after.json()["items"]} + + person_faces = test_client.get( + f"/api/v1/people/{person_id}/faces", headers=auth_headers + ) + assert person_faces.status_code == 200 + assert any(f["id"] == test_face.id for f in person_faces.json()["items"]) + + +class TestTagAndSearchWorkflow: + def test_tag_photo_then_search_by_tag_then_untag( + self, test_client, admin_user, auth_headers, test_photo + ): + tag_resp = test_client.post( + "/api/v1/tags", json={"tag_name": "workflow-tag"}, headers=auth_headers + ) + assert tag_resp.status_code == 200 + + add_resp = test_client.post( + "/api/v1/tags/photos/add", + json={"photo_ids": [test_photo.id], "tag_names": ["workflow-tag"]}, + headers=auth_headers, + ) + assert add_resp.status_code == 200 + assert add_resp.json()["photos_updated"] == 1 + + search_resp = test_client.get( + "/api/v1/photos", + params={"search_type": "tags", "tag_names": "workflow-tag"}, + headers=auth_headers, + ) + assert search_resp.status_code == 200 + assert test_photo.id in {item["id"] for item in search_resp.json()["items"]} + + remove_resp = test_client.post( + "/api/v1/tags/photos/remove", + json={"photo_ids": [test_photo.id], "tag_names": ["workflow-tag"]}, + headers=auth_headers, + ) + assert remove_resp.status_code == 200 + + after = test_client.get( + "/api/v1/photos", + params={"search_type": "tags", "tag_names": "workflow-tag"}, + headers=auth_headers, + ) + assert test_photo.id not in {item["id"] for item in after.json()["items"]} + + +class TestFavoritesWorkflow: + def test_favorite_photo_then_search_favorites_then_unfavorite( + self, test_client, admin_user, auth_headers, test_photo + ): + toggle_on = test_client.post( + f"/api/v1/photos/{test_photo.id}/toggle-favorite", headers=auth_headers + ) + assert toggle_on.status_code == 200 + + search_resp = test_client.get( + "/api/v1/photos", params={"search_type": "favorites"}, headers=auth_headers + ) + assert search_resp.status_code == 200 + assert test_photo.id in {item["id"] for item in search_resp.json()["items"]} + + toggle_off = test_client.post( + f"/api/v1/photos/{test_photo.id}/toggle-favorite", headers=auth_headers + ) + assert toggle_off.status_code == 200 + + after = test_client.get( + "/api/v1/photos", params={"search_type": "favorites"}, headers=auth_headers + ) + assert test_photo.id not in {item["id"] for item in after.json()["items"]} + + def test_favorites_search_requires_auth(self, test_client): + response = test_client.get("/api/v1/photos", params={"search_type": "favorites"}) + assert response.status_code == 401 + + +class TestBulkFavoritesWorkflow: + def test_bulk_add_then_bulk_remove( + self, test_client, admin_user, auth_headers, test_photo, test_photo_2 + ): + add_resp = test_client.post( + "/api/v1/photos/bulk-add-favorites", + json={"photo_ids": [test_photo.id, test_photo_2.id]}, + headers=auth_headers, + ) + assert add_resp.status_code == 200 + assert add_resp.json()["added_count"] == 2 + + # Re-adding the same photos should be a no-op, not a duplicate. + add_again = test_client.post( + "/api/v1/photos/bulk-add-favorites", + json={"photo_ids": [test_photo.id, test_photo_2.id]}, + headers=auth_headers, + ) + assert add_again.json()["added_count"] == 0 + assert add_again.json()["already_favorite_count"] == 2 + + remove_resp = test_client.post( + "/api/v1/photos/bulk-remove-favorites", + json={"photo_ids": [test_photo.id, test_photo_2.id]}, + headers=auth_headers, + ) + assert remove_resp.status_code == 200 + assert remove_resp.json()["removed_count"] == 2 + + +class TestUserRolePermissionsWorkflow: + def test_new_viewer_user_gains_access_after_role_permission_grant( + self, test_client, admin_user, auth_headers, regular_user, regular_auth_headers + ): + """`regular_user` is created with the default VIEWER role, which does + not include `user_uploaded` out of the box (see + backend/constants/role_features.py). Granting it should immediately + unlock the pending-photos endpoint for that user, without re-login. + """ + before = test_client.get( + "/api/v1/pending-photos", headers=regular_auth_headers + ) + assert before.status_code == 403 + + grant_resp = test_client.put( + "/api/v1/role-permissions", + json={"permissions": {"viewer": {"user_uploaded": True}}}, + headers=auth_headers, + ) + assert grant_resp.status_code == 200 + assert grant_resp.json()["permissions"]["viewer"]["user_uploaded"] is True + + after = test_client.get("/api/v1/pending-photos", headers=regular_auth_headers) + assert after.status_code == 200 -- 2.49.1