learningcircuit--local-deep-research/tests/chat/test_chat_large_data.py

# 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("...")