# allow: no-sut-import — black-box HTTP test; drives real routes through the Flask test client """ Tests for chat large data handling. These tests verify that the chat system handles large inputs and responses gracefully without failures. """ import json class TestLargeDataHandling: """Tests for handling large data in chat.""" def test_very_long_message_10000_chars_accepted(self, authenticated_client): """Test that messages up to 10000 chars are accepted.""" # Create session create_response = authenticated_client.post( "/api/chat/sessions", json={"initial_query": "Large message test"}, content_type="application/json", ) session_id = json.loads(create_response.data)["session_id"] # Send a 10000 character message (at the limit) long_message = "A" * 10000 response = authenticated_client.post( f"/api/chat/sessions/{session_id}/messages", json={"content": long_message, "trigger_research": False}, content_type="application/json", ) assert response.status_code == 200 data = json.loads(response.data) assert data["success"] is True assert data["message_id"] is not None def test_message_exceeds_max_length_rejected(self, authenticated_client): """Test that messages over 10000 chars are rejected.""" # Create session create_response = authenticated_client.post( "/api/chat/sessions", json={"initial_query": "Large message test"}, content_type="application/json", ) session_id = json.loads(create_response.data)["session_id"] # Send a message over the limit (10001 chars) too_long_message = "A" * 10001 response = authenticated_client.post( f"/api/chat/sessions/{session_id}/messages", json={"content": too_long_message, "trigger_research": False}, content_type="application/json", ) assert response.status_code == 400 data = json.loads(response.data) assert data["success"] is False assert "too long" in data["error"].lower() def test_session_title_max_500_chars(self, authenticated_client): """Test that session titles up to 500 chars are accepted via update.""" # Create session create_response = authenticated_client.post( "/api/chat/sessions", json={"initial_query": "Title test"}, content_type="application/json", ) session_id = json.loads(create_response.data)["session_id"] # Update with 500 char title (at the limit) long_title = "T" * 500 response = authenticated_client.patch( f"/api/chat/sessions/{session_id}", json={"title": long_title}, content_type="application/json", ) assert response.status_code == 200 data = json.loads(response.data) assert data["success"] is True def test_session_title_exceeds_max_rejected(self, authenticated_client): """Test that session titles over 500 chars are rejected.""" # Create session create_response = authenticated_client.post( "/api/chat/sessions", json={"initial_query": "Title test"}, content_type="application/json", ) session_id = json.loads(create_response.data)["session_id"] # Update with title over the limit (501 chars) too_long_title = "T" * 501 response = authenticated_client.patch( f"/api/chat/sessions/{session_id}", json={"title": too_long_title}, content_type="application/json", ) assert response.status_code == 400 data = json.loads(response.data) assert data["success"] is False assert "too long" in data["error"].lower() def test_session_with_many_messages(self, authenticated_client): """Test session with many messages works correctly (some may be rate-limited).""" # Create session create_response = authenticated_client.post( "/api/chat/sessions", json={"initial_query": "Many messages test"}, content_type="application/json", ) session_id = json.loads(create_response.data)["session_id"] # Send messages - some may be rate-limited (429) num_messages = 50 success_count = 0 for i in range(num_messages): response = authenticated_client.post( f"/api/chat/sessions/{session_id}/messages", json={ "content": f"Message number {i}", "trigger_research": False, }, content_type="application/json", ) assert response.status_code in (200, 429) if response.status_code == 200: success_count += 1 # At least some messages should succeed assert success_count >= 1 # Verify we can get all successful messages with pagination all_messages = [] offset = 0 while True: response = authenticated_client.get( f"/api/chat/sessions/{session_id}/messages", query_string={"limit": 20, "offset": offset}, ) data = json.loads(response.data) messages = data["messages"] if not messages: break all_messages.extend(messages) offset += 20 assert len(all_messages) == success_count def test_pagination_limits_enforced(self, authenticated_client): """Test that pagination limits are enforced.""" # Create session create_response = authenticated_client.post( "/api/chat/sessions", json={"initial_query": "Pagination test"}, content_type="application/json", ) session_id = json.loads(create_response.data)["session_id"] # Request with limit > max (100) response = authenticated_client.get( f"/api/chat/sessions/{session_id}/messages", query_string={"limit": 500}, ) assert response.status_code == 200 # The limit should be capped at max (100). The cap is proven against a # real 100+ message corpus in # tests/chat/test_chat_pagination_cap.py (it must disable the # send-route rate limiter to seed that many rows, so it lives in its # own non-black-box file); this case only checks that an oversized # limit doesn't error out. def test_large_initial_query_title_truncation(self, authenticated_client): """Test that long initial queries are truncated for title.""" # Create session with a very long initial query long_query = "Q" * 200 # Over 100 chars create_response = authenticated_client.post( "/api/chat/sessions", json={"initial_query": long_query}, content_type="application/json", ) assert create_response.status_code == 200 data = json.loads(create_response.data) # Title should be truncated (100 chars + "...") session_id = data["session_id"] get_response = authenticated_client.get( f"/api/chat/sessions/{session_id}" ) session_data = json.loads(get_response.data)["session"] # Title should be truncated assert len(session_data["title"]) <= 104 # 100 + "..." if len(long_query) > 100: assert session_data["title"].endswith("...")