Spaces:
Sleeping
Sleeping
| """ | |
| Tests for Presigned URL endpoints — Direct Upload Architecture | |
| Tests: | |
| 1. GET /photos/presigned-url → 501 when boto3 not installed | |
| 2. POST /photos/confirm → 422 when missing required fields | |
| 3. POST /photos/confirm → 404 when visit_id doesn't exist | |
| 4. POST /photos/confirm → 200 when valid payload with real visit | |
| These tests hit the live database (requires running PostgreSQL). | |
| """ | |
| import pytest | |
| import httpx | |
| BASE = "http://localhost:5000/api/v1" | |
| def client(): | |
| """Create a reusable httpx client.""" | |
| with httpx.Client(base_url=BASE, timeout=10) as c: | |
| yield c | |
| def test_presigned_url_returns_501_without_boto3(client): | |
| """GET /photos/presigned-url should return 501 when boto3 is not installed.""" | |
| r = client.get("/photos/presigned-url", params={"ext": "jpg"}) | |
| assert r.status_code == 501 | |
| body = r.json() | |
| assert body["detail"]["code"] == "STORAGE_NOT_CONFIGURED" | |
| def test_confirm_requires_fields(client): | |
| """POST /photos/confirm should return 422 when missing required fields.""" | |
| r = client.post("/photos/confirm", json={}) | |
| assert r.status_code == 422 | |
| body = r.json() | |
| assert body["detail"]["code"] == "VALIDATION_ERROR" | |
| def test_confirm_rejects_unknown_visit(client): | |
| """POST /photos/confirm should return 404 for non-existent visit_id.""" | |
| r = client.post("/photos/confirm", json={ | |
| "storage_url": "https://bucket.s3.amazonaws.com/test.jpg", | |
| "visit_id": "00000000-0000-0000-0000-000000000000", | |
| "file_name": "test.jpg", | |
| }) | |
| assert r.status_code == 404 | |
| def test_confirm_succeeds_with_real_visit(client): | |
| """POST /photos/confirm should succeed with a real visit_id.""" | |
| # First, get a real visit_id from the database | |
| visits = client.get("/visits", params={"page": 1, "per_page": 1}) | |
| assert visits.status_code == 200 | |
| data = visits.json()["data"] | |
| if not data: | |
| pytest.skip("No visits in database — can't test confirm flow") | |
| visit_id = data[0]["id"] | |
| r = client.post("/photos/confirm", json={ | |
| "storage_url": "https://test-bucket.s3.amazonaws.com/photos/test_ci.jpg", | |
| "visit_id": visit_id, | |
| "file_name": "test_ci.jpg", | |
| "file_size": 12345, | |
| }) | |
| assert r.status_code == 200 | |
| body = r.json() | |
| assert body["status"] == "success" | |
| assert "id" in body["data"] | |
| assert body["data"]["storage_url"] == "https://test-bucket.s3.amazonaws.com/photos/test_ci.jpg" | |
| def test_existing_endpoints_still_work(client): | |
| """Regression: existing endpoints must not be broken by new routes.""" | |
| # Health | |
| r = client.get("/health") | |
| assert r.status_code == 200 | |
| assert r.json()["data"]["status"] == "healthy" | |
| # Visits | |
| r = client.get("/visits", params={"page": 1, "per_page": 1}) | |
| assert r.status_code == 200 | |
| assert "data" in r.json() | |
| # Outlets | |
| r = client.get("/outlets") | |
| assert r.status_code == 200 | |
| assert "data" in r.json() | |
| # Analytics | |
| r = client.get("/analytics/summary") | |
| assert r.status_code == 200 | |
| assert "data" in r.json() | |