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

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

These tests verify complete workflows through the chat system,
from session creation through message exchange.
"""

import json


class TestChatE2EFlows:
    """End-to-end flow tests for chat feature."""

    def test_full_conversation_flow_create_send_receive(
        self, authenticated_client
    ):
        """Test complete flow: create session -> send message -> get messages."""
        # Step 1: Create session
        create_response = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "What is machine learning?"},
            content_type="application/json",
        )
        assert create_response.status_code == 200
        create_data = json.loads(create_response.data)
        assert create_data["success"] is True
        session_id = create_data["session_id"]

        # Step 2: Send a message
        send_response = authenticated_client.post(
            f"/api/chat/sessions/{session_id}/messages",
            json={
                "content": "Tell me about neural networks",
                "trigger_research": False,
            },
            content_type="application/json",
        )
        assert send_response.status_code == 200
        send_data = json.loads(send_response.data)
        assert send_data["success"] is True
        message_id = send_data["message_id"]

        # Step 3: Get messages
        get_response = authenticated_client.get(
            f"/api/chat/sessions/{session_id}/messages"
        )
        assert get_response.status_code == 200
        get_data = json.loads(get_response.data)
        assert get_data["success"] is True
        assert len(get_data["messages"]) >= 1

        # Verify our message is in the list
        message_ids = [m["id"] for m in get_data["messages"]]
        assert message_id in message_ids

    def test_multi_turn_conversation_flow(self, authenticated_client):
        """Test multiple messages in a conversation."""
        # Create session
        create_response = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "Learning about AI"},
            content_type="application/json",
        )
        session_id = json.loads(create_response.data)["session_id"]

        # Send multiple messages
        messages = [
            "What is artificial intelligence?",
            "How does deep learning work?",
            "What are transformers in NLP?",
        ]

        for i, content in enumerate(messages):
            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

        # Verify all messages are stored
        get_response = authenticated_client.get(
            f"/api/chat/sessions/{session_id}/messages"
        )
        get_data = json.loads(get_response.data)
        assert len(get_data["messages"]) == len(messages)

        # Verify session message count
        session_response = authenticated_client.get(
            f"/api/chat/sessions/{session_id}"
        )
        session_data = json.loads(session_response.data)
        assert session_data["session"]["message_count"] == len(messages)

    def test_session_update_and_archive_flow(self, authenticated_client):
        """Test session update and archive flow."""
        # Create session
        create_response = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "Test session"},
            content_type="application/json",
        )
        session_id = json.loads(create_response.data)["session_id"]

        # Update title
        update_response = authenticated_client.patch(
            f"/api/chat/sessions/{session_id}",
            json={"title": "Updated Title"},
            content_type="application/json",
        )
        assert update_response.status_code == 200
        update_data = json.loads(update_response.data)
        assert update_data["session"]["title"] == "Updated Title"

        # Archive session
        archive_response = authenticated_client.patch(
            f"/api/chat/sessions/{session_id}",
            json={"status": "archived"},
            content_type="application/json",
        )
        assert archive_response.status_code == 200
        archive_data = json.loads(archive_response.data)
        assert archive_data["session"]["status"] == "archived"

        # Verify archived session not in default list
        list_response = authenticated_client.get(
            "/api/chat/sessions?status=active"
        )
        list_data = json.loads(list_response.data)
        session_ids = [s["id"] for s in list_data["sessions"]]
        assert session_id not in session_ids

        # But appears in archived list
        archived_response = authenticated_client.get(
            "/api/chat/sessions?status=archived"
        )
        archived_data = json.loads(archived_response.data)
        archived_ids = [s["id"] for s in archived_data["sessions"]]
        assert session_id in archived_ids

    def test_concurrent_sessions_isolated(self, authenticated_client):
        """Test that multiple sessions are properly isolated."""
        # Create two sessions
        session1_response = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "Session 1 topic"},
            content_type="application/json",
        )
        session1_id = json.loads(session1_response.data)["session_id"]

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

        # Add messages to session 1
        authenticated_client.post(
            f"/api/chat/sessions/{session1_id}/messages",
            json={
                "content": "Message for session 1",
                "trigger_research": False,
            },
            content_type="application/json",
        )
        authenticated_client.post(
            f"/api/chat/sessions/{session1_id}/messages",
            json={
                "content": "Another message for session 1",
                "trigger_research": False,
            },
            content_type="application/json",
        )

        # Add different message to session 2
        authenticated_client.post(
            f"/api/chat/sessions/{session2_id}/messages",
            json={
                "content": "Message for session 2",
                "trigger_research": False,
            },
            content_type="application/json",
        )

        # Verify messages are isolated
        session1_messages = json.loads(
            authenticated_client.get(
                f"/api/chat/sessions/{session1_id}/messages"
            ).data
        )["messages"]
        session2_messages = json.loads(
            authenticated_client.get(
                f"/api/chat/sessions/{session2_id}/messages"
            ).data
        )["messages"]

        assert len(session1_messages) == 2
        assert len(session2_messages) == 1

        # Verify message content is isolated
        session1_contents = [m["content"] for m in session1_messages]
        session2_contents = [m["content"] for m in session2_messages]

        assert "Message for session 1" in session1_contents
        assert "Message for session 2" not in session1_contents
        assert "Message for session 2" in session2_contents

    def test_delete_and_list_flow(self, authenticated_client):
        """Test session deletion flow."""
        # Create session
        create_response = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "To be deleted"},
            content_type="application/json",
        )
        session_id = json.loads(create_response.data)["session_id"]

        # Verify session exists
        get_response = authenticated_client.get(
            f"/api/chat/sessions/{session_id}"
        )
        assert get_response.status_code == 200

        # Delete session
        delete_response = authenticated_client.delete(
            f"/api/chat/sessions/{session_id}"
        )
        assert delete_response.status_code == 200
        assert json.loads(delete_response.data)["success"] is True

        # Session should be permanently deleted (hard delete)
        get_deleted_response = authenticated_client.get(
            f"/api/chat/sessions/{session_id}"
        )
        assert get_deleted_response.status_code == 404

        # Should not appear in active sessions list
        list_response = authenticated_client.get(
            "/api/chat/sessions?status=active"
        )
        list_data = json.loads(list_response.data)
        active_ids = [s["id"] for s in list_data["sessions"]]
        assert session_id not in active_ids