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

# allow: no-sut-import — black-box HTTP test; drives real routes through the Flask test client
"""
Tests for Chat API endpoints and progress handling.
"""

import json
import time


class TestChatSessionAPI:
    """Tests for chat session endpoints."""

    def test_create_chat_session(self, authenticated_client):
        """Test creating a new chat session."""
        response = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "Test research question"},
            content_type="application/json",
        )
        assert response.status_code == 200

        data = json.loads(response.data)
        assert data["success"] is True
        assert "session_id" in data
        assert data["session_id"] is not None

    def test_get_chat_sessions(self, authenticated_client):
        """Test listing chat sessions."""
        # First create a session
        authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "Test query"},
            content_type="application/json",
        )

        # Then list sessions
        response = authenticated_client.get("/api/chat/sessions")
        assert response.status_code == 200

        data = json.loads(response.data)
        assert data["success"] is True
        assert "sessions" in data
        assert isinstance(data["sessions"], list)

    def test_get_chat_session_by_id(self, authenticated_client):
        """Test getting a specific chat session."""
        # Create a session
        create_response = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "Test query"},
            content_type="application/json",
        )
        session_id = json.loads(create_response.data)["session_id"]

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

        data = json.loads(response.data)
        assert data["success"] is True
        assert "session" in data
        assert data["session"]["id"] == session_id


class TestGenerateTitleEndpoint:
    """Tests for POST /api/chat/sessions/<id>/generate-title."""

    def test_generate_title_requires_query(self, authenticated_client):
        """Missing query → 400."""
        create_response = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "Test query"},
            content_type="application/json",
        )
        session_id = json.loads(create_response.data)["session_id"]

        response = authenticated_client.post(
            f"/api/chat/sessions/{session_id}/generate-title",
            json={},
            content_type="application/json",
        )
        assert response.status_code == 400
        data = json.loads(response.data)
        assert data["success"] is False

    def test_generate_title_returns_404_for_missing_session(
        self, authenticated_client
    ):
        """Unknown session → 404."""
        response = authenticated_client.post(
            "/api/chat/sessions/does-not-exist/generate-title",
            json={"query": "what is x"},
            content_type="application/json",
        )
        assert response.status_code == 404

    def test_generate_title_no_op_when_llm_disabled(self, authenticated_client):
        """When chat.llm_title_generation is disabled, endpoint returns
        success=False and title=None so the frontend keeps the fallback title.

        This is the default settings state in test fixtures.
        """
        create_response = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "Test query"},
            content_type="application/json",
        )
        session_id = json.loads(create_response.data)["session_id"]

        response = authenticated_client.post(
            f"/api/chat/sessions/{session_id}/generate-title",
            json={"query": "does quantum entanglement violate locality"},
            content_type="application/json",
        )
        # 200 regardless of LLM outcome — caller treats title=null as "no change"
        assert response.status_code == 200


class TestChatMessageAPI:
    """Tests for chat message endpoints."""

    def test_send_message_without_research(self, authenticated_client):
        """Test sending a message without triggering research."""
        # Create a session first
        create_response = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "Initial query"},
            content_type="application/json",
        )
        session_id = json.loads(create_response.data)["session_id"]

        # Send a message without research
        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 data["success"] is True
        assert "research_id" not in data or data.get("research_id") is None

    def test_get_session_messages(self, authenticated_client):
        """Test getting messages for a session."""
        # Create a session
        create_response = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "Test query"},
            content_type="application/json",
        )
        session_id = json.loads(create_response.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 "messages" in data
        assert isinstance(data["messages"], list)
        # Server tells client whether research is currently running so
        # chat.js loadSession can restore the "thinking" indicator
        # without inferring it from message metadata. None when no
        # research is in flight.
        assert "in_progress_research_id" in data
        assert data["in_progress_research_id"] is None

    def test_message_role_and_type_serialize_as_plain_strings(
        self, authenticated_client
    ):
        """JSON response has role/message_type as plain strings, not 'ChatRole.USER'.

        Guards against the concern that `(str, enum.Enum)` might leak
        `"ChatRole.USER"` through Flask's jsonify. `(str, enum.Enum)`
        members are str subclasses, so json.dumps hits the isinstance(str)
        fast path and emits the value only.
        """
        create_response = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "Serialization test"},
            content_type="application/json",
        )
        session_id = json.loads(create_response.data)["session_id"]

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

        response = authenticated_client.get(
            f"/api/chat/sessions/{session_id}/messages"
        )
        data = json.loads(response.data)
        assert data["success"] is True
        assert data["messages"], "expected at least one message"
        for m in data["messages"]:
            # Plain string, not 'ChatRole.X' or 'ChatMessageType.X'
            assert m["role"] in {"user", "assistant"}, m["role"]
            assert m["message_type"] in {
                "query",
                "followup",
                "response",
                "step",
            }, m["message_type"]


class TestChatSessionUpdate:
    """Tests for chat session update operations."""

    def test_update_session_title(self, authenticated_client):
        """Test updating a chat session title."""
        # Create a session
        create_response = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "Test query"},
            content_type="application/json",
        )
        session_id = json.loads(create_response.data)["session_id"]

        # Update the title
        response = authenticated_client.patch(
            f"/api/chat/sessions/{session_id}",
            json={"title": "Updated Title"},
            content_type="application/json",
        )
        assert response.status_code == 200

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

        # Verify the title was updated
        get_response = authenticated_client.get(
            f"/api/chat/sessions/{session_id}"
        )
        session_data = json.loads(get_response.data)
        assert session_data["session"]["title"] == "Updated Title"

    def test_update_session_with_empty_title(self, authenticated_client):
        """Test that empty title is rejected or handled."""
        # Create a session
        create_response = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "Test query"},
            content_type="application/json",
        )
        session_id = json.loads(create_response.data)["session_id"]

        # Try to update with empty title
        response = authenticated_client.patch(
            f"/api/chat/sessions/{session_id}",
            json={"title": ""},
            content_type="application/json",
        )
        # Should either reject or keep old title
        assert response.status_code in [200, 400]

    def test_update_nonexistent_session(self, authenticated_client):
        """Test updating a session that doesn't exist."""
        response = authenticated_client.patch(
            "/api/chat/sessions/nonexistent-id",
            json={"title": "New Title"},
            content_type="application/json",
        )
        # Missing session must return 404; a 500 here would be a real
        # unhandled-exception regression.
        assert response.status_code == 404
        data = json.loads(response.data)
        assert "success" in data


class TestChatSessionDelete:
    """Tests for chat session deletion."""

    def test_delete_session(self, authenticated_client):
        """Test deleting a chat session."""
        # Create a session
        create_response = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "Test query"},
            content_type="application/json",
        )
        session_id = json.loads(create_response.data)["session_id"]

        # Delete the session
        response = authenticated_client.delete(
            f"/api/chat/sessions/{session_id}"
        )
        assert response.status_code == 200

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

        # After a hard delete the row is gone; subsequent GET must be 404.
        # 500 here would be an unhandled-exception regression.
        get_response = authenticated_client.get(
            f"/api/chat/sessions/{session_id}"
        )
        assert get_response.status_code == 404

    def test_delete_nonexistent_session(self, authenticated_client):
        """Test deleting a session that doesn't exist returns 404."""
        response = authenticated_client.delete(
            "/api/chat/sessions/nonexistent-id"
        )
        # Deleting a non-existent session must return 404, not 500.
        assert response.status_code == 404


class TestChatMessageValidation:
    """Tests for chat message validation."""

    def test_send_empty_message(self, authenticated_client):
        """Test sending an empty message."""
        # Create a session
        create_response = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "Test query"},
            content_type="application/json",
        )
        session_id = json.loads(create_response.data)["session_id"]

        # Send empty message
        response = authenticated_client.post(
            f"/api/chat/sessions/{session_id}/messages",
            json={"content": "", "trigger_research": False},
            content_type="application/json",
        )
        # Should be rejected
        assert response.status_code == 400

    def test_send_message_to_nonexistent_session(self, authenticated_client):
        """Test sending a message to a session that doesn't exist."""
        response = authenticated_client.post(
            "/api/chat/sessions/nonexistent-id/messages",
            json={"content": "Test message", "trigger_research": False},
            content_type="application/json",
        )
        assert response.status_code == 404

    def test_message_with_special_characters(self, authenticated_client):
        """Test sending a message with special characters."""
        # Create a session
        create_response = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "Test query"},
            content_type="application/json",
        )
        session_id = json.loads(create_response.data)["session_id"]

        # Send message with special characters
        special_message = 'Test with émojis 🚀 and spëcial chars <>&"'
        response = authenticated_client.post(
            f"/api/chat/sessions/{session_id}/messages",
            json={"content": special_message, "trigger_research": False},
            content_type="application/json",
        )
        assert response.status_code == 200

        # Verify message is stored correctly
        messages_response = authenticated_client.get(
            f"/api/chat/sessions/{session_id}/messages"
        )
        messages = json.loads(messages_response.data)["messages"]
        assert any(m["content"] == special_message for m in messages)


class TestChatSessionPagination:
    """Tests for chat session listing and pagination."""

    def test_list_sessions_with_limit(self, authenticated_client):
        """Test listing sessions with a limit."""
        # Create multiple sessions
        for i in range(5):
            authenticated_client.post(
                "/api/chat/sessions",
                json={"initial_query": f"Query {i}"},
                content_type="application/json",
            )

        # List with limit
        response = authenticated_client.get("/api/chat/sessions?limit=3")
        assert response.status_code == 200

        data = json.loads(response.data)
        assert len(data["sessions"]) <= 3

    def test_list_sessions_order(self, authenticated_client):
        """Test that sessions are listed in order (most recent first)."""
        # ChatSession.created_at is set by ``utcnow()`` at INSERT time.
        # SQLite's CURRENT_TIMESTAMP can have second-only resolution on some
        # platforms, so a small sleep guarantees strictly-increasing
        # timestamps between sessions and avoids tie-breaking surprises in
        # the listing query's ORDER BY created_at DESC.
        session_ids = []
        for i in range(3):
            response = authenticated_client.post(
                "/api/chat/sessions",
                json={"initial_query": f"Query {i}"},
                content_type="application/json",
            )
            session_ids.append(json.loads(response.data)["session_id"])
            time.sleep(0.05)

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

        # Precondition (assert, don't skip): the listing endpoint must
        # return at least the three sessions we just created — anything
        # less is itself a failure we want to surface, not silently
        # bypass with a conditional that masks the ordering assertion.
        assert len(data["sessions"]) >= 3, (
            f"Expected ≥3 sessions, got {len(data['sessions'])}"
        )
        listed_ids = [s["id"] for s in data["sessions"]]
        # Most recent first: last created session should appear first.
        assert listed_ids[0] == session_ids[-1]


class TestChatResearchIntegration:
    """Tests for chat research integration."""

    def test_message_triggers_research(self, authenticated_client):
        """Test that sending a message with trigger_research=True returns research_id."""
        # Create a session
        create_response = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "What is quantum computing?"},
            content_type="application/json",
        )
        session_id = json.loads(create_response.data)["session_id"]

        # Send message with research trigger
        response = authenticated_client.post(
            f"/api/chat/sessions/{session_id}/messages",
            json={
                "content": "Explain quantum entanglement",
                "trigger_research": True,
            },
            content_type="application/json",
        )
        assert response.status_code == 200

        data = json.loads(response.data)
        assert data["success"] is True
        # Research should be triggered
        assert "research_id" in data

    def test_multiple_messages_in_session(self, authenticated_client):
        """Test sending multiple messages in the same session."""
        # Create a session
        create_response = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "Initial query"},
            content_type="application/json",
        )
        session_id = json.loads(create_response.data)["session_id"]

        # Send multiple messages
        messages = ["First message", "Second message", "Third message"]
        for msg in messages:
            response = authenticated_client.post(
                f"/api/chat/sessions/{session_id}/messages",
                json={"content": msg, "trigger_research": False},
                content_type="application/json",
            )
            assert response.status_code == 200

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

        assert len(data["messages"]) >= len(messages)