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

# allow: no-sut-import — black-box HTTP test; drives real routes through the Flask test client
"""
Tests for chat error handling and edge cases.

Tests error responses, edge cases, and recovery scenarios.
"""

import json


class TestErrorResponses:
    """Tests for API error response formats."""

    def test_404_response_format(self, authenticated_client):
        """Test that 404 responses have consistent format."""
        response = authenticated_client.get(
            "/api/chat/sessions/nonexistent-session-12345"
        )
        assert response.status_code == 404
        data = json.loads(response.data)
        assert data["success"] is False
        assert "error" in data

    def test_400_response_format(self, authenticated_client):
        """Test that 400 responses have consistent format."""
        # Create session first
        create_resp = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "Test"},
            content_type="application/json",
        )
        session_id = json.loads(create_resp.data)["session_id"]

        # Try invalid status update
        response = authenticated_client.patch(
            f"/api/chat/sessions/{session_id}",
            json={"status": "invalid_status"},
            content_type="application/json",
        )
        assert response.status_code == 400
        data = json.loads(response.data)
        assert data["success"] is False
        assert "error" in data

    def test_error_messages_are_descriptive(self, authenticated_client):
        """Test that error messages provide useful information."""
        # Create session
        create_resp = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "Test"},
            content_type="application/json",
        )
        session_id = json.loads(create_resp.data)["session_id"]

        # Try to send empty message
        response = authenticated_client.post(
            f"/api/chat/sessions/{session_id}/messages",
            json={"content": "", "trigger_research": False},
            content_type="application/json",
        )
        assert response.status_code == 400
        data = json.loads(response.data)
        # Error message should mention the issue
        assert len(data["error"]) > 5  # Not just empty or cryptic


class TestRobustness:
    """Tests for API robustness."""

    def test_rapid_session_creation(self, authenticated_client):
        """Test creating many sessions rapidly."""
        session_ids = []
        for i in range(10):
            response = authenticated_client.post(
                "/api/chat/sessions",
                json={"initial_query": f"Rapid test {i}"},
                content_type="application/json",
            )
            assert response.status_code == 200
            data = json.loads(response.data)
            session_ids.append(data["session_id"])

        # All should be unique
        assert len(session_ids) == len(set(session_ids))

        # All should be retrievable
        for sid in session_ids:
            get_resp = authenticated_client.get(f"/api/chat/sessions/{sid}")
            assert get_resp.status_code == 200

    def test_rapid_message_creation(self, authenticated_client):
        """Test sending many messages rapidly (some may be rate-limited)."""
        # Create session
        create_resp = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "Test"},
            content_type="application/json",
        )
        session_id = json.loads(create_resp.data)["session_id"]

        # Send 20 messages rapidly - some may be rate-limited (429)
        success_count = 0
        for i in range(20):
            response = authenticated_client.post(
                f"/api/chat/sessions/{session_id}/messages",
                json={
                    "content": f"Rapid message {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, and rate limiting should kick in
        assert success_count >= 1

        # Successfully sent messages should be retrievable
        messages_resp = authenticated_client.get(
            f"/api/chat/sessions/{session_id}/messages?limit=100"
        )
        data = json.loads(messages_resp.data)
        assert len(data["messages"]) == success_count

    def test_large_message_content(self, authenticated_client):
        """Test handling of large (but valid) message content."""
        # Create session
        create_resp = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "Test"},
            content_type="application/json",
        )
        session_id = json.loads(create_resp.data)["session_id"]

        # Send exactly at limit (10000 chars)
        large_content = "A" * 10000
        response = authenticated_client.post(
            f"/api/chat/sessions/{session_id}/messages",
            json={"content": large_content, "trigger_research": False},
            content_type="application/json",
        )
        assert response.status_code == 200

        # Verify content is stored correctly
        messages_resp = authenticated_client.get(
            f"/api/chat/sessions/{session_id}/messages"
        )
        data = json.loads(messages_resp.data)
        assert data["messages"][-1]["content"] == large_content


class TestSessionStateTransitions:
    """Tests for session state transition edge cases."""

    def test_multiple_status_changes(self, authenticated_client):
        """Test multiple sequential status changes."""
        # Create session
        create_resp = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "Test"},
            content_type="application/json",
        )
        session_id = json.loads(create_resp.data)["session_id"]

        # Change to archived
        authenticated_client.patch(
            f"/api/chat/sessions/{session_id}",
            json={"status": "archived"},
            content_type="application/json",
        )

        # Verify final state is archived
        get_resp = authenticated_client.get(f"/api/chat/sessions/{session_id}")
        data = json.loads(get_resp.data)
        assert data["session"]["status"] == "archived"

    def test_delete_already_deleted_session(self, authenticated_client):
        """Test deleting an already deleted session returns 404."""
        # Create and delete session
        create_resp = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "Test"},
            content_type="application/json",
        )
        session_id = json.loads(create_resp.data)["session_id"]

        # Delete once (hard delete)
        authenticated_client.delete(f"/api/chat/sessions/{session_id}")

        # Delete again - should return 404 (session no longer exists)
        response = authenticated_client.delete(
            f"/api/chat/sessions/{session_id}"
        )
        assert response.status_code == 404


class TestMessageEdgeCases:
    """Tests for message edge cases."""

    def test_get_messages_empty_response(self, authenticated_client):
        """Test getting messages when none exist."""
        # Create session
        create_resp = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "Test"},
            content_type="application/json",
        )
        session_id = json.loads(create_resp.data)["session_id"]

        # Get messages (should be empty)
        response = authenticated_client.get(
            f"/api/chat/sessions/{session_id}/messages"
        )
        assert response.status_code == 200
        data = json.loads(response.data)
        assert data["success"] is True
        assert data["messages"] == []

    def test_get_messages_with_large_offset(self, authenticated_client):
        """Test getting messages with offset beyond count."""
        # Create session
        create_resp = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "Test"},
            content_type="application/json",
        )
        session_id = json.loads(create_resp.data)["session_id"]

        # Send a few messages
        for i in range(3):
            authenticated_client.post(
                f"/api/chat/sessions/{session_id}/messages",
                json={"content": f"Message {i}", "trigger_research": False},
                content_type="application/json",
            )

        # Get with offset beyond message count
        response = authenticated_client.get(
            f"/api/chat/sessions/{session_id}/messages?offset=100"
        )
        assert response.status_code == 200
        data = json.loads(response.data)
        assert data["messages"] == []

    def test_message_with_null_bytes(self, authenticated_client):
        """Test message handling with null bytes."""
        # Create session
        create_resp = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "Test"},
            content_type="application/json",
        )
        session_id = json.loads(create_resp.data)["session_id"]

        # Send message with null byte (may be rejected)
        response = authenticated_client.post(
            f"/api/chat/sessions/{session_id}/messages",
            json={
                "content": "Message\x00with\x00nulls",
                "trigger_research": False,
            },
            content_type="application/json",
        )
        # Should handle gracefully - accept or reject without crashing.
        # 500 here would be an unhandled-exception regression.
        assert response.status_code in [200, 400]


class TestListSessions:
    """Tests for session listing edge cases."""

    def test_list_sessions_returns_list(self, authenticated_client):
        """Test that listing sessions returns a list."""
        response = authenticated_client.get("/api/chat/sessions")
        assert response.status_code == 200
        data = json.loads(response.data)
        assert data["success"] is True
        assert isinstance(data["sessions"], list)

    def test_list_with_zero_limit(self, authenticated_client):
        """Test listing with limit=0."""
        response = authenticated_client.get("/api/chat/sessions?limit=0")
        assert response.status_code == 200
        data = json.loads(response.data)
        # Should return empty or use default limit
        assert isinstance(data["sessions"], list)

    def test_list_ordering(self, authenticated_client):
        """Test that sessions are listed in order."""
        # Create several sessions. Assertion is inclusion-only (not strict
        # ordering), so no sleep between creates is needed.
        session_ids = []
        for i in range(3):
            create_resp = authenticated_client.post(
                "/api/chat/sessions",
                json={"initial_query": f"Order test {i}"},
                content_type="application/json",
            )
            session_ids.append(json.loads(create_resp.data)["session_id"])

        # List sessions
        response = authenticated_client.get("/api/chat/sessions?limit=10")
        data = json.loads(response.data)

        # Recent sessions should be at the top (most recently updated first)
        listed_ids = [s["id"] for s in data["sessions"]]
        # At least our sessions should be in the list
        for sid in session_ids:
            assert sid in listed_ids


class TestSessionRetrieval:
    """Tests for session retrieval edge cases."""

    def test_get_session_includes_all_fields(self, authenticated_client):
        """Test that session retrieval includes all expected fields."""
        # Create session
        create_resp = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "Field test"},
            content_type="application/json",
        )
        session_id = json.loads(create_resp.data)["session_id"]

        # Get session
        response = authenticated_client.get(f"/api/chat/sessions/{session_id}")
        assert response.status_code == 200
        data = json.loads(response.data)

        session = data["session"]
        expected_fields = [
            "id",
            "title",
            "status",
            "created_at",
            "message_count",
        ]
        for field in expected_fields:
            assert field in session, f"Missing field: {field}"

    def test_get_session_with_messages(self, authenticated_client):
        """Test getting a session that has messages."""
        # Create session
        create_resp = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "Test"},
            content_type="application/json",
        )
        session_id = json.loads(create_resp.data)["session_id"]

        # Add some messages
        for i in range(5):
            authenticated_client.post(
                f"/api/chat/sessions/{session_id}/messages",
                json={"content": f"Message {i}", "trigger_research": False},
                content_type="application/json",
            )

        # Get session
        response = authenticated_client.get(f"/api/chat/sessions/{session_id}")
        data = json.loads(response.data)

        # Message count should be updated
        assert data["session"]["message_count"] == 5