hermes-agent/tests/test_telegram_thread_routing.py

"""Tests for Telegram thread-aware session routing.

Verifies that messages in different threads/topics get independent
conversation histories.
"""

import pytest
import sys
import os

sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..'))

from gateway.session import build_session_key, SessionSource
from gateway.platforms.base import Platform


class TestThreadSessionKey:
    """Verify session keys include thread_id for isolation."""

    def test_dm_with_thread_gets_unique_key(self):
        source = SessionSource(
            platform=Platform.TELEGRAM,
            chat_id="123456",
            chat_type="dm",
            thread_id="100",
        )
        key = build_session_key(source)
        assert "123456" in key
        assert "100" in key
        assert key == "agent:main:telegram:dm:123456:100"

    def test_dm_without_thread_uses_chat_only(self):
        source = SessionSource(
            platform=Platform.TELEGRAM,
            chat_id="123456",
            chat_type="dm",
        )
        key = build_session_key(source)
        assert key == "agent:main:telegram:dm:123456"
        assert ":100" not in key

    def test_different_threads_different_keys(self):
        source_a = SessionSource(
            platform=Platform.TELEGRAM,
            chat_id="123456",
            chat_type="dm",
            thread_id="100",
        )
        source_b = SessionSource(
            platform=Platform.TELEGRAM,
            chat_id="123456",
            chat_type="dm",
            thread_id="200",
        )
        key_a = build_session_key(source_a)
        key_b = build_session_key(source_b)
        assert key_a != key_b

    def test_same_thread_same_key(self):
        source_a = SessionSource(
            platform=Platform.TELEGRAM,
            chat_id="123456",
            chat_type="dm",
            thread_id="100",
        )
        source_b = SessionSource(
            platform=Platform.TELEGRAM,
            chat_id="123456",
            chat_type="dm",
            thread_id="100",
        )
        assert build_session_key(source_a) == build_session_key(source_b)

    def test_group_with_thread_includes_thread(self):
        source = SessionSource(
            platform=Platform.TELEGRAM,
            chat_id="789",
            chat_type="group",
            thread_id="50",
            user_id="user1",
        )
        key = build_session_key(source)
        assert "789" in key
        assert "50" in key

    def test_group_without_thread_isolates_by_user(self):
        source = SessionSource(
            platform=Platform.TELEGRAM,
            chat_id="789",
            chat_type="group",
            user_id="user1",
        )
        key = build_session_key(source, group_sessions_per_user=True)
        assert "789" in key
        assert "user1" in key

    def test_group_thread_shared_across_users(self):
        """In threads, all participants share the same session by default."""
        source_a = SessionSource(
            platform=Platform.TELEGRAM,
            chat_id="789",
            chat_type="group",
            thread_id="50",
            user_id="user1",
        )
        source_b = SessionSource(
            platform=Platform.TELEGRAM,
            chat_id="789",
            chat_type="group",
            thread_id="50",
            user_id="user2",
        )
        key_a = build_session_key(source_a, thread_sessions_per_user=False)
        key_b = build_session_key(source_b, thread_sessions_per_user=False)
        assert key_a == key_b  # Shared session in thread

    def test_group_thread_per_user_when_enabled(self):
        """With thread_sessions_per_user=True, users get isolated sessions."""
        source_a = SessionSource(
            platform=Platform.TELEGRAM,
            chat_id="789",
            chat_type="group",
            thread_id="50",
            user_id="user1",
        )
        source_b = SessionSource(
            platform=Platform.TELEGRAM,
            chat_id="789",
            chat_type="group",
            thread_id="50",
            user_id="user2",
        )
        key_a = build_session_key(source_a, thread_sessions_per_user=True)
        key_b = build_session_key(source_b, thread_sessions_per_user=True)
        assert key_a != key_b


class TestSessionSourceSerialization:
    """Verify SessionSource round-trips correctly with thread_id."""

    def test_thread_id_preserved_in_dict(self):
        source = SessionSource(
            platform=Platform.TELEGRAM,
            chat_id="123",
            chat_type="dm",
            thread_id="456",
        )
        d = source.to_dict()
        assert d["thread_id"] == "456"
        restored = SessionSource.from_dict(d)
        assert restored.thread_id == "456"

    def test_none_thread_id_preserved(self):
        source = SessionSource(
            platform=Platform.TELEGRAM,
            chat_id="123",
            chat_type="dm",
        )
        d = source.to_dict()
        assert d.get("thread_id") is None
        restored = SessionSource.from_dict(d)
        assert restored.thread_id is None