Spaces:
Running
Running
| """Each browser session sees only its own records. | |
| These are the tests that make the per-visitor dashboard trustworthy. Without | |
| them, a scoping mistake in a query would silently expose every visitor's history | |
| to every other visitor, and nothing else in the suite would notice. | |
| """ | |
| from __future__ import annotations | |
| import pytest | |
| BODY = {"dominant_emotion": "Happy", "duration_seconds": 10} | |
| READ_ROUTES = ["/api/logs", "/api/logs/stats"] | |
| def _post(client, headers, emotion="Happy"): | |
| return client.post( | |
| "/api/logs", | |
| json={"dominant_emotion": emotion, "duration_seconds": 10}, | |
| headers=headers, | |
| ) | |
| class TestSessionHeaderRequired: | |
| def test_reads_require_a_session_id(self, client, path): | |
| assert client.get(path).status_code == 400 | |
| def test_writes_require_a_session_id(self, client): | |
| response = client.post("/api/logs", json=BODY, headers={"X-API-Key": "test-ingest-key"}) | |
| assert response.status_code == 400 | |
| def test_malformed_session_ids_are_rejected(self, client, bad): | |
| response = client.get("/api/logs", headers={"X-Session-Id": bad}) | |
| assert response.status_code == 400 | |
| class TestReadIsolation: | |
| def test_a_session_sees_only_its_own_entries( | |
| self, client, ingest_headers, other_session_headers, session_headers | |
| ): | |
| _post(client, ingest_headers, "Happy") | |
| _post(client, ingest_headers, "Sad") | |
| _post(client, other_session_headers, "Angry") | |
| body = client.get("/api/logs", headers=session_headers).json() | |
| assert body["total"] == 2 | |
| assert {item["dominant_emotion"] for item in body["items"]} == {"Happy", "Sad"} | |
| def test_the_other_session_sees_only_its_own( | |
| self, client, ingest_headers, other_session_headers | |
| ): | |
| _post(client, ingest_headers, "Happy") | |
| _post(client, other_session_headers, "Angry") | |
| body = client.get( | |
| "/api/logs", headers={"X-Session-Id": other_session_headers["X-Session-Id"]} | |
| ).json() | |
| assert body["total"] == 1 | |
| assert body["items"][0]["dominant_emotion"] == "Angry" | |
| def test_stats_are_scoped_too( | |
| self, client, ingest_headers, other_session_headers, session_headers | |
| ): | |
| for _ in range(3): | |
| _post(client, ingest_headers, "Happy") | |
| _post(client, other_session_headers, "Sad") | |
| stats = client.get("/api/logs/stats", headers=session_headers).json() | |
| assert stats["total_count"] == 3 | |
| assert {e["emotion"] for e in stats["by_emotion"]} == {"Happy"} | |
| def test_an_empty_session_sees_nothing(self, client, ingest_headers): | |
| _post(client, ingest_headers, "Happy") | |
| body = client.get("/api/logs", headers={"X-Session-Id": "a-brand-new-session-99"}).json() | |
| assert body["total"] == 0 | |
| class TestCrossSessionAccessIsDenied: | |
| def test_cannot_fetch_another_sessions_entry_by_id( | |
| self, client, ingest_headers, other_session_headers | |
| ): | |
| created = _post(client, other_session_headers).json() | |
| response = client.get( | |
| f"/api/logs/{created['id']}", | |
| headers={"X-Session-Id": ingest_headers["X-Session-Id"]}, | |
| ) | |
| # 404 rather than 403: an id you do not own is indistinguishable from one | |
| # that does not exist, so the endpoint cannot be used to probe. | |
| assert response.status_code == 404 | |
| def test_cannot_delete_another_sessions_entry( | |
| self, client, ingest_headers, other_session_headers, session_headers | |
| ): | |
| created = _post(client, other_session_headers).json() | |
| response = client.delete(f"/api/logs/{created['id']}", headers=session_headers) | |
| assert response.status_code == 404 | |
| # And it is still there for its owner. | |
| owner = client.get( | |
| "/api/logs", headers={"X-Session-Id": other_session_headers["X-Session-Id"]} | |
| ).json() | |
| assert owner["total"] == 1 | |
| def test_clear_all_only_clears_the_callers_records( | |
| self, client, ingest_headers, other_session_headers, session_headers | |
| ): | |
| _post(client, ingest_headers, "Happy") | |
| _post(client, ingest_headers, "Sad") | |
| _post(client, other_session_headers, "Angry") | |
| response = client.delete("/api/logs?confirm=true", headers=session_headers) | |
| assert response.json()["deleted"] == 2 | |
| survivor = client.get( | |
| "/api/logs", headers={"X-Session-Id": other_session_headers["X-Session-Id"]} | |
| ).json() | |
| assert survivor["total"] == 1 | |
| class TestCorsAllowsTheSessionHeader: | |
| def test_preflight_permits_x_session_id(self, client): | |
| """A header missing from allow_headers is blocked before it is sent. | |
| curl ignores CORS, so this failure mode is invisible outside a browser: | |
| every request simply never leaves the page. | |
| """ | |
| response = client.options( | |
| "/api/logs", | |
| headers={ | |
| "Origin": "http://localhost:3000", | |
| "Access-Control-Request-Method": "GET", | |
| "Access-Control-Request-Headers": "x-session-id", | |
| }, | |
| ) | |
| allowed = response.headers.get("access-control-allow-headers", "").lower() | |
| assert "x-session-id" in allowed | |
| class TestNoOperatorKeyRequired: | |
| def test_reading_needs_no_api_key(self, client, ingest_headers, session_headers): | |
| _post(client, ingest_headers) | |
| # A plain session id is enough; there is no operator credential. | |
| assert "X-API-Key" not in session_headers | |
| assert client.get("/api/logs", headers=session_headers).status_code == 200 | |
| def test_response_does_not_echo_the_session_id(self, client, ingest_headers): | |
| body = _post(client, ingest_headers).json() | |
| assert "session_id" not in body | |