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

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

Tests message creation, retrieval, and ordering.
"""

import json


class TestMessageCreation:
    """Tests for message creation."""

    def test_create_user_message(self, authenticated_client):
        """Test creating a user message."""
        # 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
        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)
        assert "message_id" in data
        assert data["message_id"] is not None

    def test_create_multiple_messages(self, authenticated_client):
        """Test creating multiple messages in a session."""
        # 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 multiple messages
        message_ids = []
        for i in range(5):
            response = authenticated_client.post(
                f"/api/chat/sessions/{session_id}/messages",
                json={"content": f"Message {i}", "trigger_research": False},
                content_type="application/json",
            )
            assert response.status_code == 200
            data = json.loads(response.data)
            message_ids.append(data["message_id"])

        # All message IDs should be unique
        assert len(message_ids) == len(set(message_ids))

    def test_message_updates_session_message_count(self, authenticated_client):
        """Test that sending messages updates session message 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"]

        # Get initial count
        get_resp = authenticated_client.get(f"/api/chat/sessions/{session_id}")
        initial_count = json.loads(get_resp.data)["session"]["message_count"]

        # Send 3 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",
            )

        # Check updated count
        get_resp = authenticated_client.get(f"/api/chat/sessions/{session_id}")
        new_count = json.loads(get_resp.data)["session"]["message_count"]
        assert new_count == initial_count + 3


class TestMessageRetrieval:
    """Tests for message retrieval."""

    def test_get_messages_empty_session(self, authenticated_client):
        """Test getting messages from empty session."""
        # 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
        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 isinstance(data["messages"], list)
        assert len(data["messages"]) == 0

    def test_get_messages_returns_all(self, authenticated_client):
        """Test getting messages returns all created 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"]

        # Send 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 messages
        response = authenticated_client.get(
            f"/api/chat/sessions/{session_id}/messages"
        )
        data = json.loads(response.data)
        assert len(data["messages"]) == 5

    def test_get_messages_in_order(self, authenticated_client):
        """Test that messages are returned in correct order."""
        # 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 messages with identifiable content
        for i in range(5):
            authenticated_client.post(
                f"/api/chat/sessions/{session_id}/messages",
                json={
                    "content": f"Message number {i}",
                    "trigger_research": False,
                },
                content_type="application/json",
            )

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

        # Verify order by sequence_number
        for i, msg in enumerate(messages):
            assert msg["sequence_number"] == i + 1

    def test_get_messages_with_pagination(self, authenticated_client):
        """Test message pagination."""
        # 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 10 messages
        for i in range(10):
            authenticated_client.post(
                f"/api/chat/sessions/{session_id}/messages",
                json={"content": f"Message {i}", "trigger_research": False},
                content_type="application/json",
            )

        # Get first page
        page1 = authenticated_client.get(
            f"/api/chat/sessions/{session_id}/messages?limit=5&offset=0"
        )
        page1_data = json.loads(page1.data)
        assert len(page1_data["messages"]) == 5

        # Get second page
        page2 = authenticated_client.get(
            f"/api/chat/sessions/{session_id}/messages?limit=5&offset=5"
        )
        page2_data = json.loads(page2.data)
        assert len(page2_data["messages"]) == 5

        # Messages should be different
        page1_ids = [m["id"] for m in page1_data["messages"]]
        page2_ids = [m["id"] for m in page2_data["messages"]]
        assert set(page1_ids).isdisjoint(set(page2_ids))


class TestMessageFields:
    """Tests for message field content."""

    def test_message_has_required_fields(self, authenticated_client):
        """Test that messages have all required fields."""
        # 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
        authenticated_client.post(
            f"/api/chat/sessions/{session_id}/messages",
            json={"content": "Test message", "trigger_research": False},
            content_type="application/json",
        )

        # Get messages
        response = authenticated_client.get(
            f"/api/chat/sessions/{session_id}/messages"
        )
        data = json.loads(response.data)
        msg = data["messages"][0]

        required_fields = [
            "id",
            "session_id",
            "role",
            "content",
            "sequence_number",
            "created_at",
        ]
        for field in required_fields:
            assert field in msg, f"Missing field: {field}"

    def test_user_message_has_correct_role(self, authenticated_client):
        """Test that user messages have role='user'."""
        # 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
        authenticated_client.post(
            f"/api/chat/sessions/{session_id}/messages",
            json={"content": "Test message", "trigger_research": False},
            content_type="application/json",
        )

        # Get messages
        response = authenticated_client.get(
            f"/api/chat/sessions/{session_id}/messages"
        )
        data = json.loads(response.data)
        msg = data["messages"][0]

        assert msg["role"] == "user"

    def test_message_content_preserved(self, authenticated_client):
        """Test that message content is preserved exactly."""
        # 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 specific content
        content = "Test message with special chars: <>&"
        authenticated_client.post(
            f"/api/chat/sessions/{session_id}/messages",
            json={"content": content, "trigger_research": False},
            content_type="application/json",
        )

        # Get messages
        response = authenticated_client.get(
            f"/api/chat/sessions/{session_id}/messages"
        )
        data = json.loads(response.data)
        msg = data["messages"][0]

        assert msg["content"] == content


class TestMessageSequencing:
    """Tests for message sequence numbering."""

    def test_sequence_numbers_increment(self, authenticated_client):
        """Test that sequence numbers increment correctly."""
        # 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 multiple 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 messages
        response = authenticated_client.get(
            f"/api/chat/sessions/{session_id}/messages"
        )
        data = json.loads(response.data)

        # Verify sequence numbers
        for i, msg in enumerate(data["messages"]):
            assert msg["sequence_number"] == i + 1

    def test_sequence_numbers_unique_per_session(self, authenticated_client):
        """Test that sequence numbers are per-session."""
        # Create two sessions
        session_ids = []
        for _ in range(2):
            create_resp = authenticated_client.post(
                "/api/chat/sessions",
                json={"initial_query": "Test"},
                content_type="application/json",
            )
            session_ids.append(json.loads(create_resp.data)["session_id"])

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

        # Both sessions should have messages with sequence 1, 2, 3
        for sid in session_ids:
            response = authenticated_client.get(
                f"/api/chat/sessions/{sid}/messages"
            )
            data = json.loads(response.data)
            sequences = [m["sequence_number"] for m in data["messages"]]
            assert sequences == [1, 2, 3]


class TestMessageIsolation:
    """Tests for message isolation between sessions."""

    def test_messages_isolated_between_sessions(self, authenticated_client):
        """Test that messages from one session don't appear in another."""
        # Create two sessions
        create_resp1 = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "Session 1"},
            content_type="application/json",
        )
        session1_id = json.loads(create_resp1.data)["session_id"]

        create_resp2 = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "Session 2"},
            content_type="application/json",
        )
        session2_id = json.loads(create_resp2.data)["session_id"]

        # Send different messages to each
        authenticated_client.post(
            f"/api/chat/sessions/{session1_id}/messages",
            json={"content": "Session 1 message", "trigger_research": False},
            content_type="application/json",
        )

        authenticated_client.post(
            f"/api/chat/sessions/{session2_id}/messages",
            json={"content": "Session 2 message", "trigger_research": False},
            content_type="application/json",
        )

        # Get messages from session 1
        response1 = authenticated_client.get(
            f"/api/chat/sessions/{session1_id}/messages"
        )
        data1 = json.loads(response1.data)

        # Get messages from session 2
        response2 = authenticated_client.get(
            f"/api/chat/sessions/{session2_id}/messages"
        )
        data2 = json.loads(response2.data)

        # Each should only have its own message
        assert len(data1["messages"]) == 1
        assert data1["messages"][0]["content"] == "Session 1 message"

        assert len(data2["messages"]) == 1
        assert data2["messages"][0]["content"] == "Session 2 message"