tests/unit/classes/test_session_conversation_store.py

import pytest
from classes.session_conversation_store import SessionConversationStore


class TestSessionConversationStore:
    """Unit tests for SessionConversationStore class"""

    @pytest.fixture
    def store(self):
        """Fresh store instance for each test"""
        return SessionConversationStore()

    @pytest.fixture
    def sample_session_id(self):
        return "test-session-123"

    @pytest.fixture
    def sample_conversation_id(self):
        return "conversation-abc"

    # ==================== add_human_message Tests ====================

    def test_add_human_message_new_session(
        self, store, sample_session_id, sample_conversation_id
    ):
        """Test adding a human message to a new session"""
        conversation = store.add_human_message(
            sample_session_id, sample_conversation_id, "Hello"
        )

        assert sample_session_id in store.session_conversation_map
        assert (
            sample_conversation_id in store.session_conversation_map[sample_session_id]
        )

        assert len(conversation) == 1
        assert conversation[0].role == "user"
        assert conversation[0].content == "Hello"

    def test_add_human_message_existing_session_new_conversation(
        self, store, sample_session_id
    ):
        """Test adding a human message to a new conversation in existing session"""
        conversation_id_1 = "conversation-1"
        conversation_id_2 = "conversation-2"

        conv1 = store.add_human_message(
            sample_session_id, conversation_id_1, "First conversation"
        )
        conv2 = store.add_human_message(
            sample_session_id, conversation_id_2, "Second conversation"
        )

        assert len(store.session_conversation_map[sample_session_id]) == 2
        assert conversation_id_1 in store.session_conversation_map[sample_session_id]
        assert conversation_id_2 in store.session_conversation_map[sample_session_id]

        # Verify return values
        assert len(conv1) == 1
        assert len(conv2) == 1
        assert conv1[0].content == "First conversation"
        assert conv2[0].content == "Second conversation"

    def test_add_human_message_existing_conversation(
        self, store, sample_session_id, sample_conversation_id
    ):
        """Test adding multiple human messages to the same conversation"""
        conv1 = store.add_human_message(
            sample_session_id, sample_conversation_id, "First message"
        )
        conv2 = store.add_human_message(
            sample_session_id, sample_conversation_id, "Second message"
        )

        assert len(conv2) == 2
        assert conv2[0].content == "First message"
        assert conv2[1].content == "Second message"

    def test_add_human_message_empty_string(
        self, store, sample_session_id, sample_conversation_id
    ):
        """Test adding an empty human message"""
        conversation = store.add_human_message(
            sample_session_id, sample_conversation_id, ""
        )

        assert len(conversation) == 1
        assert conversation[0].content == ""
        assert conversation[0].role == "user"

    def test_add_human_message_with_special_characters(
        self, store, sample_session_id, sample_conversation_id
    ):
        """Test adding a human message with special characters"""
        special_message = "Hello! @#$%^&*() 你好 🎉\n\tNew line"
        conversation = store.add_human_message(
            sample_session_id, sample_conversation_id, special_message
        )

        assert conversation[0].content == special_message

    def test_add_human_message_very_long_text(
        self, store, sample_session_id, sample_conversation_id
    ):
        """Test adding a very long human message"""
        long_message = "x" * 10000
        conversation = store.add_human_message(
            sample_session_id, sample_conversation_id, long_message
        )

        assert conversation[0].content == long_message

    # ==================== add_assistant_reply Tests ====================

    def test_add_assistant_reply_new_session(
        self, store, sample_session_id, sample_conversation_id
    ):
        """Test adding an assistant reply to a new session"""
        conversation = store.add_assistant_reply(
            sample_session_id, sample_conversation_id, "Hello!"
        )

        assert sample_session_id in store.session_conversation_map
        assert (
            sample_conversation_id in store.session_conversation_map[sample_session_id]
        )

        assert len(conversation) == 1
        assert conversation[0].role == "assistant"
        assert conversation[0].content == "Hello!"

    def test_add_assistant_reply_existing_session_new_conversation(
        self, store, sample_session_id
    ):
        """Test adding an assistant reply to a new conversation in existing session"""
        conversation_id_1 = "conversation-1"
        conversation_id_2 = "conversation-2"

        conv1 = store.add_assistant_reply(
            sample_session_id, conversation_id_1, "First reply"
        )
        conv2 = store.add_assistant_reply(
            sample_session_id, conversation_id_2, "Second reply"
        )

        assert len(store.session_conversation_map[sample_session_id]) == 2
        assert len(conv1) == 1
        assert len(conv2) == 1

    def test_add_assistant_reply_existing_conversation(
        self, store, sample_session_id, sample_conversation_id
    ):
        """Test adding multiple assistant replies to the same conversation"""
        conv1 = store.add_assistant_reply(
            sample_session_id, sample_conversation_id, "First reply"
        )
        conv2 = store.add_assistant_reply(
            sample_session_id, sample_conversation_id, "Second reply"
        )

        assert len(conv2) == 2
        assert conv2[0].content == "First reply"
        assert conv2[1].content == "Second reply"

    def test_add_assistant_reply_empty_string(
        self, store, sample_session_id, sample_conversation_id
    ):
        """Test adding an empty assistant reply"""
        conversation = store.add_assistant_reply(
            sample_session_id, sample_conversation_id, ""
        )

        assert len(conversation) == 1
        assert conversation[0].content == ""
        assert conversation[0].role == "assistant"

    def test_add_methods_return_same_reference(
        self, store, sample_session_id, sample_conversation_id
    ):
        """Test that returned conversation is the actual stored list (same reference)"""
        conv1 = store.add_human_message(
            sample_session_id, sample_conversation_id, "Message 1"
        )
        conv2 = store.add_assistant_reply(
            sample_session_id, sample_conversation_id, "Reply 1"
        )

        # Should be the same list object
        assert conv1 is conv2
        assert (
            conv2
            is store.session_conversation_map[sample_session_id][sample_conversation_id]
        )

        # Changes to returned list affect stored data
        assert len(conv2) == 2

    # ==================== delete_session_conversations Tests ====================

    def test_delete_session_conversations_existing_session(
        self, store, sample_session_id
    ):
        """Test deleting all conversations in a session"""
        store.add_human_message(sample_session_id, "conversation-1", "Message 1")
        store.add_human_message(sample_session_id, "conversation-2", "Message 2")

        store.delete_session_conversations(sample_session_id)

        assert sample_session_id not in store.session_conversation_map

    def test_delete_session_conversations_nonexistent_session(self, store):
        """Test deleting a nonexistent session (should not raise error)"""
        # Should not raise any exception
        store.delete_session_conversations("nonexistent-session")

        assert "nonexistent-session" not in store.session_conversation_map

    def test_delete_session_conversations_does_not_affect_other_sessions(self, store):
        """Test that deleting one session doesn't affect others"""
        session_1 = "session-1"
        session_2 = "session-2"

        store.add_human_message(session_1, "conversation-1", "Message 1")
        store.add_human_message(session_2, "conversation-2", "Message 2")

        store.delete_session_conversations(session_1)

        assert session_1 not in store.session_conversation_map
        assert session_2 in store.session_conversation_map
        assert "conversation-2" in store.session_conversation_map[session_2]

    def test_delete_session_conversations_with_multiple_conversations(
        self, store, sample_session_id
    ):
        """Test deleting a session with multiple conversations"""
        store.add_human_message(sample_session_id, "conversation-1", "Message 1")
        store.add_human_message(sample_session_id, "conversation-2", "Message 2")
        store.add_human_message(sample_session_id, "conversation-3", "Message 3")

        store.delete_session_conversations(sample_session_id)

        assert sample_session_id not in store.session_conversation_map

    # ==================== Integration Tests ====================

    def test_conversation_flow_user_assistant_alternating(
        self, store, sample_session_id, sample_conversation_id
    ):
        """Test realistic conversation flow with alternating messages"""
        store.add_human_message(
            sample_session_id, sample_conversation_id, "What's the weather?"
        )
        store.add_assistant_reply(
            sample_session_id, sample_conversation_id, "It's sunny today!"
        )
        store.add_human_message(sample_session_id, sample_conversation_id, "Thanks!")
        conversation = store.add_assistant_reply(
            sample_session_id, sample_conversation_id, "You're welcome!"
        )

        assert len(conversation) == 4
        assert conversation[0].role == "user"
        assert conversation[1].role == "assistant"
        assert conversation[2].role == "user"
        assert conversation[3].role == "assistant"

    def test_multiple_conversations_in_same_session(self, store, sample_session_id):
        """Test multiple independent conversations in the same session"""
        conv1_id = "conversation-1"
        conv2_id = "conversation-2"

        # Conversation 1
        store.add_human_message(sample_session_id, conv1_id, "Hello in conv 1")
        conv1 = store.add_assistant_reply(
            sample_session_id, conv1_id, "Reply in conv 1"
        )

        # Conversation 2
        store.add_human_message(sample_session_id, conv2_id, "Hello in conv 2")
        conv2 = store.add_assistant_reply(
            sample_session_id, conv2_id, "Reply in conv 2"
        )

        assert len(conv1) == 2
        assert len(conv2) == 2
        assert conv1[0].content == "Hello in conv 1"
        assert conv2[0].content == "Hello in conv 2"

    def test_multiple_sessions_isolation(self, store):
        """Test that sessions are completely isolated"""
        session1 = "session-1"
        session2 = "session-2"
        conversation_id = "conversation-1"

        conv1 = store.add_human_message(
            session1, conversation_id, "Message in session 1"
        )
        conv2 = store.add_human_message(
            session2, conversation_id, "Message in session 2"
        )

        assert conv1[0].content == "Message in session 1"
        assert conv2[0].content == "Message in session 2"

    def test_conversation_order_preserved(
        self, store, sample_session_id, sample_conversation_id
    ):
        """Test that message order is preserved in the conversation"""
        messages = [f"Message {i}" for i in range(10)]

        conversation = None
        for i, msg in enumerate(messages):
            if i % 2 == 0:
                conversation = store.add_human_message(
                    sample_session_id, sample_conversation_id, msg
                )
            else:
                conversation = store.add_assistant_reply(
                    sample_session_id, sample_conversation_id, msg
                )

        assert len(conversation) == 10
        for i, msg in enumerate(messages):
            assert conversation[i].content == msg

    def test_empty_session_after_deletion(
        self, store, sample_session_id, sample_conversation_id
    ):
        """Test that a session can be recreated after deletion"""
        # Create and delete
        store.add_human_message(
            sample_session_id, sample_conversation_id, "First message"
        )
        store.delete_session_conversations(sample_session_id)

        # Recreate
        conversation = store.add_human_message(
            sample_session_id, sample_conversation_id, "New message"
        )

        assert len(conversation) == 1
        assert conversation[0].content == "New message"

    def test_chat_message_structure(
        self, store, sample_session_id, sample_conversation_id
    ):
        """Test that ChatMessage objects have correct structure"""
        conversation = store.add_human_message(
            sample_session_id, sample_conversation_id, "Test"
        )

        message = conversation[0]

        assert hasattr(message, "role")
        assert hasattr(message, "content")
        assert message.role == "user"
        assert message.content == "Test"

    def test_consecutive_human_messages(
        self, store, sample_session_id, sample_conversation_id
    ):
        """Test adding multiple human messages without assistant replies"""
        store.add_human_message(sample_session_id, sample_conversation_id, "Message 1")
        store.add_human_message(sample_session_id, sample_conversation_id, "Message 2")
        conversation = store.add_human_message(
            sample_session_id, sample_conversation_id, "Message 3"
        )

        assert len(conversation) == 3
        assert all(msg.role == "user" for msg in conversation)

    def test_consecutive_assistant_messages(
        self, store, sample_session_id, sample_conversation_id
    ):
        """Test adding multiple assistant messages without human messages"""
        store.add_assistant_reply(sample_session_id, sample_conversation_id, "Reply 1")
        store.add_assistant_reply(sample_session_id, sample_conversation_id, "Reply 2")
        conversation = store.add_assistant_reply(
            sample_session_id, sample_conversation_id, "Reply 3"
        )

        assert len(conversation) == 3
        assert all(msg.role == "assistant" for msg in conversation)

    def test_large_conversation_history(
        self, store, sample_session_id, sample_conversation_id
    ):
        """Test handling a large conversation with many messages"""
        num_messages = 1000

        conversation = None
        for i in range(num_messages):
            if i % 2 == 0:
                conversation = store.add_human_message(
                    sample_session_id, sample_conversation_id, f"User message {i}"
                )
            else:
                conversation = store.add_assistant_reply(
                    sample_session_id, sample_conversation_id, f"Assistant message {i}"
                )

        assert len(conversation) == num_messages

    def test_special_conversation_and_session_ids(self, store):
        """Test using special characters in IDs"""
        special_ids = [
            ("session-with-dashes", "conversation-with-dashes"),
            ("session_with_underscores", "conversation_with_underscores"),
            ("session123", "conversation456"),
            ("a", "b"),  # Single character IDs
        ]

        for session_id, conversation_id in special_ids:
            conversation = store.add_human_message(
                session_id, conversation_id, "Test message"
            )
            assert len(conversation) == 1

    def test_dict_structure_integrity(self, store):
        """Test that internal data structure maintains correct nesting"""
        session_id = "test-session"
        conv_id_1 = "conversation-1"
        conv_id_2 = "conversation-2"

        store.add_human_message(session_id, conv_id_1, "Message 1")
        store.add_human_message(session_id, conv_id_2, "Message 2")

        # Check structure
        assert isinstance(store.session_conversation_map, dict)
        assert isinstance(store.session_conversation_map[session_id], dict)
        assert isinstance(store.session_conversation_map[session_id][conv_id_1], list)