Merge pull request 'test: add cross-endpoint integration and error-handling suites' (#62) from test/backend-integration-error-suites into dev
This commit is contained in:
commit
05b908241d
195
tests/test_api_errors.py
Normal file
195
tests/test_api_errors.py
Normal file
@ -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 = "<script>alert(1)</script>"
|
||||
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)
|
||||
171
tests/test_api_integration.py
Normal file
171
tests/test_api_integration.py
Normal file
@ -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
|
||||
Loading…
x
Reference in New Issue
Block a user