File size: 3,435 Bytes
bbe01fe
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
84c1ab9
 
bbe01fe
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
# backend/tests/test_chat_endpoint.py
# Tests the /chat endpoint security layer (auth, rate limit, input validation).
# The pipeline itself is mocked — these tests cover the HTTP contract, not LLM logic.

import pytest


VALID_UUID = "a1b2c3d4-e5f6-4789-8abc-def012345678"


def chat(client, message: str, session_id: str = VALID_UUID, token: str | None = None):
    headers = {"Content-Type": "application/json"}
    if token:
        headers["Authorization"] = f"Bearer {token}"
    return client.post(
        "/chat",
        json={"message": message, "session_id": session_id},
        headers=headers,
    )


class TestChatAuth:
    def test_no_token_returns_401(self, app_client):
        resp = chat(app_client, "hello")
        assert resp.status_code == 401

    def test_valid_token_accepted(self, app_client, valid_token):
        resp = chat(app_client, "Tell me about your projects", token=valid_token)
        # 200 means authentication passed; pipeline may still return streaming content
        assert resp.status_code == 200

    def test_expired_token_returns_401(self, app_client, expired_token):
        resp = chat(app_client, "hello", token=expired_token)
        assert resp.status_code == 401

    def test_wrong_secret_returns_401(self, app_client, wrong_secret_token):
        resp = chat(app_client, "hello", token=wrong_secret_token)
        assert resp.status_code == 401

    def test_malformed_token_returns_401(self, app_client):
        resp = chat(app_client, "hello", token="not.a.jwt")
        assert resp.status_code == 401


class TestChatInputValidation:
    def test_empty_message_returns_422(self, app_client, valid_token):
        resp = chat(app_client, "", token=valid_token)
        assert resp.status_code == 422

    def test_message_too_long_returns_422(self, app_client, valid_token):
        resp = chat(app_client, "x" * 501, token=valid_token)
        assert resp.status_code == 422

    def test_invalid_session_id_returns_422(self, app_client, valid_token):
        # session_id must match ^[a-zA-Z0-9_-]+$ — spaces and special chars are rejected.
        resp = chat(app_client, "hello", session_id="invalid id with spaces!", token=valid_token)
        assert resp.status_code == 422

    def test_missing_session_id_returns_422(self, app_client, valid_token):
        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {valid_token}",
        }
        resp = app_client.post("/chat", json={"message": "hello"}, headers=headers)
        assert resp.status_code == 422

    def test_missing_message_returns_422(self, app_client, valid_token):
        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {valid_token}",
        }
        resp = app_client.post(
            "/chat", json={"session_id": VALID_UUID}, headers=headers
        )
        assert resp.status_code == 422


class TestChatResponse:
    def test_successful_response_is_event_stream(self, app_client, valid_token):
        resp = chat(app_client, "What is TextOps?", token=valid_token)
        assert resp.status_code == 200
        assert "text/event-stream" in resp.headers.get("content-type", "")

    def test_response_has_no_cache_header(self, app_client, valid_token):
        resp = chat(app_client, "What is TextOps?", token=valid_token)
        assert resp.headers.get("cache-control") == "no-cache"