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

# allow: no-sut-import — black-box HTTP test; drives real routes through the Flask test client
"""
Tests for chat API input validation.

Tests various edge cases and malformed inputs to ensure
proper validation and error handling.
"""

import json
import pytest


class TestMessageContentValidation:
    """Tests for message content validation."""

    def test_empty_message_content_rejected(self, authenticated_client):
        """Test that empty message content is rejected."""
        # 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 empty content
        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)
        assert "required" in data["error"].lower()

    def test_missing_content_field_rejected(self, authenticated_client):
        """Test that missing content field is rejected."""
        # 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 without content field
        response = authenticated_client.post(
            f"/api/chat/sessions/{session_id}/messages",
            json={"trigger_research": False},
            content_type="application/json",
        )
        assert response.status_code == 400
        data = json.loads(response.data)
        assert "required" in data["error"].lower()

    def test_null_content_rejected(self, authenticated_client):
        """Test that null content is rejected."""
        # 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 null content
        response = authenticated_client.post(
            f"/api/chat/sessions/{session_id}/messages",
            json={"content": None, "trigger_research": False},
            content_type="application/json",
        )
        assert response.status_code == 400

    def test_whitespace_only_content_rejected(self, authenticated_client):
        """Test that whitespace-only content is rejected."""
        # 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 whitespace-only content - should be rejected
        response = authenticated_client.post(
            f"/api/chat/sessions/{session_id}/messages",
            json={"content": "   \t\n  ", "trigger_research": False},
            content_type="application/json",
        )
        assert response.status_code == 400
        data = json.loads(response.data)
        assert "required" in data["error"].lower()

    def test_message_at_exact_length_limit(self, authenticated_client):
        """Test message at exactly 10000 characters."""
        # 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 10000 chars
        content = "A" * 10000
        response = authenticated_client.post(
            f"/api/chat/sessions/{session_id}/messages",
            json={"content": content, "trigger_research": False},
            content_type="application/json",
        )
        assert response.status_code == 200

    def test_message_one_over_length_limit(self, authenticated_client):
        """Test message at 10001 characters is rejected."""
        # 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 10001 chars
        content = "A" * 10001
        response = authenticated_client.post(
            f"/api/chat/sessions/{session_id}/messages",
            json={"content": content, "trigger_research": False},
            content_type="application/json",
        )
        assert response.status_code == 400
        data = json.loads(response.data)
        assert "too long" in data["error"].lower()


class TestJSONMalformation:
    """Tests for malformed JSON handling."""

    def test_invalid_json_rejected(self, authenticated_client):
        """Test that invalid JSON is rejected."""
        # 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 invalid JSON
        response = authenticated_client.post(
            f"/api/chat/sessions/{session_id}/messages",
            data="{invalid json",
            content_type="application/json",
        )
        # Malformed JSON must be rejected with 400; 500 would be an
        # unhandled-exception regression.
        assert response.status_code == 400

    def test_non_object_json_handled(self, authenticated_client):
        """Test that non-object JSON is handled."""
        # 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 JSON array instead of object
        response = authenticated_client.post(
            f"/api/chat/sessions/{session_id}/messages",
            data='["not", "an", "object"]',
            content_type="application/json",
        )
        # Non-object JSON should be rejected with 400; 500 would be a bug.
        assert response.status_code == 400

    def test_deeply_nested_json_handled(self, authenticated_client):
        """Test that deeply nested JSON is handled."""
        # 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"]

        # Build deeply nested structure
        nested = {"content": "test", "trigger_research": False}
        for _ in range(50):
            nested = {"nested": nested}

        # Send deeply nested JSON
        response = authenticated_client.post(
            f"/api/chat/sessions/{session_id}/messages",
            json=nested,
            content_type="application/json",
        )
        # Missing top-level `content` field must produce a 400. A 500 here
        # would be an unhandled-exception regression.
        assert response.status_code == 400


class TestMessageToNonexistentSession:
    """Tests for sending messages to invalid sessions."""

    def test_message_to_nonexistent_session(self, authenticated_client):
        """Test sending a message to a nonexistent session."""
        response = authenticated_client.post(
            "/api/chat/sessions/nonexistent-session-id-12345/messages",
            json={"content": "Test message", "trigger_research": False},
            content_type="application/json",
        )
        assert response.status_code == 404
        data = json.loads(response.data)
        assert data["success"] is False

    def test_message_to_deleted_session(self, authenticated_client):
        """Test sending a message to a deleted session."""
        # 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"]

        authenticated_client.delete(f"/api/chat/sessions/{session_id}")

        # Try to send message
        response = authenticated_client.post(
            f"/api/chat/sessions/{session_id}/messages",
            json={"content": "Test message", "trigger_research": False},
            content_type="application/json",
        )
        # Session still exists (soft delete), might accept or reject
        assert response.status_code in [200, 404]


class TestTriggerResearchParameter:
    """Tests for trigger_research parameter."""

    def test_trigger_research_default_true(self, authenticated_client):
        """Test that trigger_research defaults to true."""
        # 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 without trigger_research field
        response = authenticated_client.post(
            f"/api/chat/sessions/{session_id}/messages",
            json={"content": "Test message"},
            content_type="application/json",
        )
        assert response.status_code == 200
        data = json.loads(response.data)
        # Default should trigger research (research_mode != "none")
        assert (
            data.get("research_mode") != "none"
            or data.get("research_id") is not None
        )

    def test_trigger_research_false(self, authenticated_client):
        """Test that trigger_research=false prevents research."""
        # 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 with trigger_research=false
        response = authenticated_client.post(
            f"/api/chat/sessions/{session_id}/messages",
            json={"content": "Test message", "trigger_research": False},
            content_type="application/json",
        )
        assert response.status_code == 200
        data = json.loads(response.data)
        # Should not trigger research
        assert data.get("research_id") is None

    def test_trigger_research_non_boolean(self, authenticated_client):
        """Non-boolean trigger_research is rejected with 400 (L_API1).

        Previously a truthy string like "false" was silently coerced to True,
        launching research against the caller's intent to suppress it. The
        route now requires a strict boolean.
        """
        # 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 with a non-boolean value
        response = authenticated_client.post(
            f"/api/chat/sessions/{session_id}/messages",
            json={"content": "Test", "trigger_research": "false"},
            content_type="application/json",
        )
        assert response.status_code == 400
        data = json.loads(response.data)
        assert data["success"] is False
        assert "trigger_research" in data["error"]

    def test_trigger_research_bool_and_default_accepted(
        self, authenticated_client
    ):
        """A real boolean (or an omitted value) is accepted."""
        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"]

        # Explicit boolean False suppresses research and succeeds.
        response = authenticated_client.post(
            f"/api/chat/sessions/{session_id}/messages",
            json={"content": "Test no research", "trigger_research": False},
            content_type="application/json",
        )
        assert response.status_code == 200
        assert json.loads(response.data)["research_mode"] == "none"

        # Omitting trigger_research defaults to True (research attempted).
        response = authenticated_client.post(
            f"/api/chat/sessions/{session_id}/messages",
            json={"content": "Test default"},
            content_type="application/json",
        )
        # 200 (research started) or 429 (rate limited) — both prove the
        # default boolean path is accepted rather than 400-rejected.
        assert response.status_code in (200, 429)


class TestGetMessagesValidation:
    """Tests for get messages endpoint validation."""

    def test_get_messages_invalid_limit(self, authenticated_client):
        """Test get messages with invalid limit."""
        # 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 invalid limit
        response = authenticated_client.get(
            f"/api/chat/sessions/{session_id}/messages?limit=invalid"
        )
        # Should use default (handle gracefully)
        assert response.status_code == 200

    def test_get_messages_negative_limit(self, authenticated_client):
        """Test get messages with negative limit."""
        # 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 negative limit
        response = authenticated_client.get(
            f"/api/chat/sessions/{session_id}/messages?limit=-5"
        )
        # Should clamp to minimum
        assert response.status_code == 200

    def test_get_messages_very_large_limit(self, authenticated_client):
        """Test get messages with very large limit."""
        # 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 very large limit
        response = authenticated_client.get(
            f"/api/chat/sessions/{session_id}/messages?limit=1000000"
        )
        # Should clamp to maximum (100)
        assert response.status_code == 200


class TestSpecialCharactersInFields:
    """Tests for special characters in various fields."""

    @pytest.mark.parametrize(
        "query",
        [
            "Query with 'single quotes'",
            'Query with "double quotes"',
            "Query with\nnewlines\n\n",
            "Query with\ttabs",
            "Query with emoji 🔥💯",
            "Query with unicode: 你好世界",
            "Query with <html> tags",
            "Query with & ampersand",
        ],
    )
    def test_special_chars_in_initial_query(self, query, authenticated_client):
        """Test special characters in initial query."""
        response = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": query},
            content_type="application/json",
        )
        assert response.status_code == 200

    @pytest.mark.parametrize(
        "content",
        [
            "Message with 'quotes'",
            'Message with "double"',
            "Message with\nlines",
            "Message with emoji 🎉",
            "Message with unicode: Привет",
            "Message with <tags>",
            "Message with & and <",
        ],
    )
    def test_special_chars_in_message(self, content, authenticated_client):
        """Test special characters in 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"]

        response = authenticated_client.post(
            f"/api/chat/sessions/{session_id}/messages",
            json={"content": content, "trigger_research": False},
            content_type="application/json",
        )
        assert response.status_code == 200


class TestContentTypeHandling:
    """Tests for Content-Type handling."""

    def test_wrong_content_type_for_post(self, authenticated_client):
        """Test POST with wrong Content-Type."""
        response = authenticated_client.post(
            "/api/chat/sessions",
            data="initial_query=test",
            content_type="application/x-www-form-urlencoded",
        )
        # create_session uses defaults when no JSON body is parseable, so
        # 200 (graceful: session created with defaults) is acceptable, as is
        # 400 (rejected up-front). 500 here would be an unhandled-exception
        # regression and must never be tolerated.
        assert response.status_code in [200, 400]

    def test_missing_content_type(self, authenticated_client):
        """Test POST without Content-Type."""
        response = authenticated_client.post(
            "/api/chat/sessions",
            data='{"initial_query": "test"}',
        )
        # Without an explicit Content-Type Flask treats the body as opaque
        # and request.get_json(silent=True) returns None → defaults used.
        # 200 (defaults) or 400 (rejected) are acceptable; 500 is not.
        assert response.status_code in [200, 400]