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

"""
Property-based fuzz tests for chat components using Hypothesis.

These tests verify that chat functions don't crash on arbitrary input
and handle edge cases gracefully.
"""

import pytest
from hypothesis import given, settings, strategies as st, assume, HealthCheck
from unittest.mock import MagicMock, patch
from contextlib import contextmanager


# =============================================================================
# Custom Strategies for Chat Testing
# =============================================================================


def message_roles():
    """Generate valid message roles."""
    return st.sampled_from(["user", "assistant"])


def message_types():
    """Generate valid message types.

    The schema removed "step" from ChatMessageType (steps live in
    chat_progress_steps with their own enum-less schema); the chat
    context layer filters out any "step" rows before they reach here.
    """
    return st.sampled_from(["query", "followup", "response"])


def safe_text(max_size=200):
    """Generate text without problematic Unicode."""
    return st.text(
        alphabet=st.characters(
            blacklist_categories=("Cs",),  # Exclude surrogates
            blacklist_characters=["\x00"],  # Exclude null bytes
        ),
        max_size=max_size,
    )


def message_dict():
    """Generate a message dictionary."""
    return st.fixed_dictionaries(
        {
            "id": st.text(
                min_size=5, max_size=20, alphabet="abcdef0123456789-"
            ),
            "role": message_roles(),
            "content": safe_text(max_size=100),
            "message_type": message_types(),
            "research_id": st.one_of(
                st.none(),
                st.text(min_size=5, max_size=20, alphabet="abcdef0123456789-"),
            ),
        }
    )


def accumulated_context_dict():
    """Generate an accumulated context dictionary."""
    return st.fixed_dictionaries(
        {
            "key_entities": st.lists(safe_text(max_size=50), max_size=20),
            "topics": st.lists(safe_text(max_size=30), max_size=10),
            "summary": safe_text(max_size=200),
        }
    )


# =============================================================================
# ChatContextManager Fuzz Tests
# =============================================================================


class TestChatContextManagerFuzz:
    """Fuzz tests for ChatContextManager."""

    @given(
        messages=st.lists(message_dict(), max_size=10),
        accumulated=st.one_of(st.none(), accumulated_context_dict()),
    )
    @settings(max_examples=50, suppress_health_check=[HealthCheck.too_slow])
    def test_build_context_handles_arbitrary_messages(
        self, messages, accumulated
    ):
        """Test that build_research_context never crashes on arbitrary input."""
        from src.local_deep_research.chat.context import ChatContextManager

        try:
            manager = ChatContextManager(
                session_id="test-session",
                messages=messages,
                accumulated_context=accumulated,
            )
            context = manager.build_research_context()

            # Verify result structure
            assert isinstance(context, dict)
            assert "session_id" in context
            assert "is_multi_turn" in context

        except Exception as e:
            pytest.fail(
                f"build_research_context crashed: {type(e).__name__}: {e}"
            )

    @given(accumulated=accumulated_context_dict())
    @settings(max_examples=50, suppress_health_check=[HealthCheck.too_slow])
    def test_accumulated_context_handles_arbitrary_data(self, accumulated):
        """Test that accumulated context with arbitrary data is handled."""
        from src.local_deep_research.chat.context import ChatContextManager

        try:
            manager = ChatContextManager(
                session_id="test-session",
                messages=[],
                accumulated_context=accumulated,
            )

            # Test key entity extraction
            entities = manager._get_key_entities()
            assert isinstance(entities, list)
            assert len(entities) <= 20  # Should be limited

            # Test topics extraction
            topics = manager._get_topics()
            assert isinstance(topics, list)
            assert len(topics) <= 10  # Should be limited

        except Exception as e:
            pytest.fail(f"Context handling crashed: {type(e).__name__}: {e}")

    @given(content=safe_text(max_size=500))
    @settings(max_examples=50, suppress_health_check=[HealthCheck.too_slow])
    def test_create_summary_handles_arbitrary_content(self, content):
        """Test that summary creation handles arbitrary content."""
        from src.local_deep_research.chat.context import ChatContextManager

        try:
            manager = ChatContextManager("test-session", [], {})
            summary = manager._create_summary(content)

            # Should return a string
            assert isinstance(summary, str)
            # Should be limited in length
            assert len(summary) <= 400  # 300 + "..."

        except Exception as e:
            pytest.fail(f"Summary creation crashed: {type(e).__name__}: {e}")

    @given(content=safe_text(max_size=200))
    @settings(max_examples=30, suppress_health_check=[HealthCheck.too_slow])
    def test_extract_context_updates_handles_arbitrary_input(self, content):
        """Test that context update extraction handles arbitrary input."""
        from src.local_deep_research.chat.context import ChatContextManager

        try:
            manager = ChatContextManager("test-session", [], {})
            updates = manager.extract_context_updates(content)

            # Verify structure
            assert isinstance(updates, dict)
            assert "new_entities" in updates
            assert "new_topics" in updates
            assert "summary_addition" in updates

        except Exception as e:
            pytest.fail(
                f"Context update extraction crashed: {type(e).__name__}: {e}"
            )


# =============================================================================
# ChatService Fuzz Tests
# =============================================================================


class TestChatServiceFuzz:
    """Fuzz tests for ChatService."""

    @given(query=st.one_of(st.none(), safe_text(max_size=200)))
    @settings(max_examples=50, suppress_health_check=[HealthCheck.too_slow])
    def test_title_generation_handles_any_string(self, query):
        """Test that title generation handles any input string."""
        from src.local_deep_research.chat.service import ChatService

        try:
            service = ChatService(username="test_user")
            title = service._generate_title(query)

            # Should return a string
            assert isinstance(title, str)
            # Should have reasonable length
            assert len(title) <= 150  # 100 + "..."

        except Exception as e:
            pytest.fail(f"Title generation crashed: {type(e).__name__}: {e}")

    @given(
        content=safe_text(max_size=100),
        role=message_roles(),
        message_type=message_types(),
    )
    @settings(max_examples=30, suppress_health_check=[HealthCheck.too_slow])
    def test_add_message_handles_unicode(self, content, role, message_type):
        """Test that add_message handles various Unicode content."""
        from src.local_deep_research.chat.service import ChatService

        # Skip empty content for valid messages
        assume(len(content.strip()) > 0)

        @contextmanager
        def mock_get_user_db_session(username, password=None):
            mock_session = MagicMock()
            # Mock the session query to return a valid session
            mock_chat_session = MagicMock()
            mock_chat_session.message_count = 0
            mock_session.query.return_value.filter_by.return_value.with_for_update.return_value.first.return_value = mock_chat_session
            mock_session.query.return_value.filter_by.return_value.first.return_value = mock_chat_session
            mock_session.add = MagicMock()
            mock_session.commit = MagicMock()
            yield mock_session

        with patch(
            "src.local_deep_research.chat.service.get_user_db_session",
            mock_get_user_db_session,
        ):
            try:
                service = ChatService(username="test_user")
                message_id = service.add_message(
                    session_id="test-session",
                    role=role,
                    content=content,
                    message_type=message_type,
                )

                # Should return a UUID string
                assert isinstance(message_id, str)
                assert len(message_id) > 0

            except ValueError:
                # Session not found is expected in some cases
                pass
            except Exception as e:
                pytest.fail(f"add_message crashed: {type(e).__name__}: {e}")


# =============================================================================
# Chat API Fuzz Tests (Parametrized for performance)
# =============================================================================


# Test data for API fuzzing - these are checked without Hypothesis
# to avoid the fixture overhead with each hypothesis iteration
SPECIAL_CONTENT_CASES = [
    "Normal message",
    "",  # Empty
    " " * 100,  # Whitespace
    "Unicode: Привет 你好 🎉",
    "<script>alert('XSS')</script>",
    "'\"<>&",  # HTML special chars
    "Line1\nLine2\rLine3",
    "A" * 9999,  # Near limit
    "\x00\x01\x02",  # Control chars
]

UNICODE_TITLE_CASES = [
    "Normal Title",
    "",
    "Привет мир",
    "你好世界",
    "🎉🔥💯",
    "<script>",
    "A" * 499,  # Near limit
]

PAGINATION_CASES = [
    (-1, 0),
    (0, 0),
    (1, 0),
    (100, 0),
    (101, 0),
    (20, -1),
    (20, 0),
    (20, 1000),
]

STATUS_CASES = [
    "active",
    "archived",
    "all",
    "invalid",
    "",
    "' OR '1'='1",
    "<script>",
]


class TestChatAPIFuzz:
    """Fuzz tests for Chat API endpoints using parametrized tests."""

    @pytest.mark.parametrize("content", SPECIAL_CONTENT_CASES)
    def test_send_message_handles_special_characters(
        self, content, authenticated_client
    ):
        """Test that send message handles special characters in content."""
        import json

        # Create a session first
        create_response = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "Test"},
            content_type="application/json",
        )

        # Session creation is the precondition, NOT the system under
        # fuzz. Skipping on a non-200 here would silently mask any
        # regression in POST /api/chat/sessions (CI would go green for
        # every parametrized case). Assert so a broken precondition
        # fails loudly.
        assert create_response.status_code == 200, (
            f"session creation precondition failed: "
            f"{create_response.status_code} {create_response.data!r}"
        )

        session_id = json.loads(create_response.data)["session_id"]

        # Send message with test content
        response = authenticated_client.post(
            f"/api/chat/sessions/{session_id}/messages",
            json={"content": content, "trigger_research": False},
            content_type="application/json",
        )

        # Should not crash. 200 = accepted, 400 = validation error.
        # 500 here would be an unhandled-exception regression on user input.
        assert response.status_code in [200, 400]

    @pytest.mark.parametrize("title", UNICODE_TITLE_CASES)
    def test_session_title_update_handles_unicode(
        self, title, authenticated_client
    ):
        """Test that session title update handles Unicode strings."""
        import json

        # Create a session
        create_response = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "Test"},
            content_type="application/json",
        )

        # Precondition assert — see test_message_send_handles_unicode
        # for the rationale (skip would mask session-create regression).
        assert create_response.status_code == 200, (
            f"session creation precondition failed: "
            f"{create_response.status_code} {create_response.data!r}"
        )

        session_id = json.loads(create_response.data)["session_id"]

        # Update with test title
        response = authenticated_client.patch(
            f"/api/chat/sessions/{session_id}",
            json={"title": title},
            content_type="application/json",
        )

        # Should not crash. 200 = accepted, 400 = validation error.
        # 500 here would be an unhandled-exception regression on user input.
        assert response.status_code in [200, 400]

    @pytest.mark.parametrize("limit,offset", PAGINATION_CASES)
    def test_pagination_handles_edge_cases(
        self, limit, offset, authenticated_client
    ):
        """Test that pagination parameters handle edge cases."""
        import json

        response = authenticated_client.get(
            f"/api/chat/sessions?limit={limit}&offset={offset}"
        )

        # Should not crash - may clamp values to valid range
        assert response.status_code in [200, 400]

        if response.status_code == 200:
            data = json.loads(response.data)
            assert "sessions" in data

    @pytest.mark.parametrize("status", STATUS_CASES)
    def test_list_sessions_handles_invalid_status(
        self, status, authenticated_client
    ):
        """Test that list sessions handles invalid status values."""
        import json

        response = authenticated_client.get(
            f"/api/chat/sessions?status={status}"
        )

        # Should not crash - invalid status should default to "active"
        assert response.status_code == 200

        data = json.loads(response.data)
        assert data["success"] is True


# =============================================================================
# Prompt Context Fuzz Tests
# =============================================================================


class TestPromptContextFuzz:
    """Fuzz tests for findings extraction."""

    @given(messages=st.lists(message_dict(), min_size=5, max_size=20))
    @settings(max_examples=20, suppress_health_check=[HealthCheck.too_slow])
    def test_findings_extraction_handles_many_messages(self, messages):
        """Test that findings extraction handles many messages."""
        from src.local_deep_research.chat.context import ChatContextManager

        try:
            manager = ChatContextManager("test-session", messages, {})
            findings = manager._extract_findings_from_history()

            # Should return a string
            assert isinstance(findings, str)

        except Exception as e:
            pytest.fail(f"Findings extraction crashed: {type(e).__name__}: {e}")


# =============================================================================
# Edge Case Tests
# =============================================================================


class TestEdgeCases:
    """Tests for specific edge cases that could cause issues."""

    def test_empty_messages_list(self):
        """Test handling of empty messages list."""
        from src.local_deep_research.chat.context import ChatContextManager

        manager = ChatContextManager("test-session", [], {})
        context = manager.build_research_context()

        assert context["is_multi_turn"] is False
        assert context["turn_count"] == 0

    def test_none_messages_list(self):
        """Test handling of None messages."""
        from src.local_deep_research.chat.context import ChatContextManager

        manager = ChatContextManager("test-session", None, {})
        context = manager.build_research_context()

        assert context["is_multi_turn"] is False

    def test_none_accumulated_context(self):
        """Test handling of None accumulated context."""
        from src.local_deep_research.chat.context import ChatContextManager

        manager = ChatContextManager("test-session", [], None)
        entities = manager._get_key_entities()
        topics = manager._get_topics()

        assert entities == []
        assert topics == []

    def test_very_long_content_truncation(self):
        """Test that very long content is properly truncated."""
        from src.local_deep_research.chat.context import ChatContextManager

        # Create message with very long content
        long_content = "A" * 10000
        messages = [
            {
                "id": "msg-1",
                "role": "assistant",
                "content": long_content,
                "message_type": "response",
                "research_id": "research-1",
            }
        ]

        manager = ChatContextManager("test-session", messages, {})
        findings = manager.build_research_context()["accumulated_findings"]

        # Should be significantly shorter than original
        assert len(findings) < len(long_content)

    def test_special_characters_in_entities(self):
        """Test handling of special characters in entities."""
        from src.local_deep_research.chat.context import ChatContextManager

        accumulated = {
            "key_entities": [
                "entity<script>",
                "entity'with'quotes",
                'entity"double',
                "entity\nnewline",
            ],
            "topics": ["topic<tag>", "topic;semicolon"],
            "summary": "Summary with <html> tags",
        }

        manager = ChatContextManager("test-session", [], accumulated)
        context = manager.build_research_context()

        # Should not crash and should include entities
        assert len(context["key_entities"]) > 0