File size: 3,310 Bytes
dc4e6da | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 | """
Functional tests — GET /jobs/{request_id}/status
==================================================
Tests for the job-status polling endpoint.
Key behaviours:
• 404 for a request_id that is not in Supabase
• 500/404 for a malformed (non-UUID) request_id
• Response contract when a real job exists (structural)
• Status field is constrained to known values
"""
import pytest
import requests
from tests.conftest import BASE_URL, TIMEOUT, NONEXISTENT_REQUEST_ID
ENDPOINT_TPL = f"{BASE_URL}/jobs/{{request_id}}/status"
VALID_STATUSES = {
"pending", "processing", "downloading", "generating",
"zipping", "uploading", "completed", "completed_no_gdrive",
"completed_gdrive_failed", "failed", "error",
}
class TestJobStatusEndpoint:
"""GET /jobs/{request_id}/status"""
def test_unknown_uuid_returns_non_200(self, http):
url = ENDPOINT_TPL.format(request_id=NONEXISTENT_REQUEST_ID)
r = http.get(url, timeout=TIMEOUT)
# Supabase lookup fails → 404 expected; some deployments may raise 500
assert r.status_code in (404, 500), (
f"Expected 404/500 for unknown request_id, got {r.status_code}"
)
def test_unknown_uuid_response_is_json(self, http):
url = ENDPOINT_TPL.format(request_id=NONEXISTENT_REQUEST_ID)
r = http.get(url, timeout=TIMEOUT)
assert "application/json" in r.headers.get("Content-Type", "")
def test_unknown_uuid_has_detail(self, http):
url = ENDPOINT_TPL.format(request_id=NONEXISTENT_REQUEST_ID)
r = http.get(url, timeout=TIMEOUT)
body = r.json()
assert "detail" in body, f"Error response must contain 'detail'. Got: {body}"
def test_garbage_request_id_returns_error(self, http):
"""A completely non-UUID string should still return a meaningful error."""
url = ENDPOINT_TPL.format(request_id="not-a-real-id")
r = http.get(url, timeout=TIMEOUT)
assert r.status_code in (404, 500)
def test_endpoint_is_get_only(self, http):
url = ENDPOINT_TPL.format(request_id=NONEXISTENT_REQUEST_ID)
r = http.post(url, json={}, timeout=TIMEOUT)
assert r.status_code == 405, (
f"POST to a GET-only endpoint should be 405, got {r.status_code}"
)
def test_status_field_in_known_values_if_200(self, http):
"""
If we ever get a 200 (e.g., a real job in the DB), the status must
be one of the known values documented in the API.
"""
url = ENDPOINT_TPL.format(request_id=NONEXISTENT_REQUEST_ID)
r = http.get(url, timeout=TIMEOUT)
if r.status_code == 200:
body = r.json()
assert body.get("status") in VALID_STATUSES, (
f"Unexpected status value: {body.get('status')}"
)
def test_200_response_contract_if_present(self, http):
"""
If a 200 is returned, the body must contain the required fields.
"""
url = ENDPOINT_TPL.format(request_id=NONEXISTENT_REQUEST_ID)
r = http.get(url, timeout=TIMEOUT)
if r.status_code == 200:
body = r.json()
for field in ("request_id", "status", "created_at", "updated_at"):
assert field in body, f"Missing required field '{field}' in {body}"
|