Spaces:
Sleeping
Sleeping
File size: 5,926 Bytes
67af2df | 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 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 | """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:
@pytest.mark.parametrize("path", READ_ROUTES)
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
@pytest.mark.parametrize("bad", ["short", "has spaces here!", "x" * 65, ""])
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
|