Spaces:
Running
Running
| # 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" | |