"""Tests for gateway/channel_directory.py — channel resolution and display.""" import asyncio import json import os from pathlib import Path from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, patch from gateway.channel_directory import ( build_channel_directory, lookup_channel_type, resolve_channel_name, format_directory_for_display, load_directory, _build_from_sessions, _build_slack, DIRECTORY_PATH, ) def _write_directory(tmp_path, platforms): """Helper to write a fake channel directory.""" data = {"updated_at": "2026-01-01T00:00:00", "platforms": platforms} cache_file = tmp_path / "channel_directory.json" cache_file.write_text(json.dumps(data)) return cache_file class TestLoadDirectory: def test_missing_file(self, tmp_path): with patch("gateway.channel_directory.DIRECTORY_PATH", tmp_path / "nope.json"): result = load_directory() assert result["updated_at"] is None assert result["platforms"] == {} def test_valid_file(self, tmp_path): cache_file = _write_directory(tmp_path, { "telegram": [{"id": "123", "name": "John", "type": "dm"}] }) with patch("gateway.channel_directory.DIRECTORY_PATH", cache_file): result = load_directory() assert result["platforms"]["telegram"][0]["name"] == "John" def test_corrupt_file(self, tmp_path): cache_file = tmp_path / "channel_directory.json" cache_file.write_text("{bad json") with patch("gateway.channel_directory.DIRECTORY_PATH", cache_file): result = load_directory() assert result["updated_at"] is None class TestBuildChannelDirectoryWrites: def test_failed_write_preserves_previous_cache(self, tmp_path, monkeypatch): cache_file = _write_directory(tmp_path, { "telegram": [{"id": "123", "name": "Alice", "type": "dm"}] }) previous = json.loads(cache_file.read_text()) def broken_dump(data, fp, *args, **kwargs): fp.write('{"updated_at":') fp.flush() raise OSError("disk full") monkeypatch.setattr(json, "dump", broken_dump) with patch("gateway.channel_directory.DIRECTORY_PATH", cache_file): asyncio.run(build_channel_directory({})) result = load_directory() assert result == previous class TestResolveChannelName: def _setup(self, tmp_path, platforms): cache_file = _write_directory(tmp_path, platforms) return patch("gateway.channel_directory.DIRECTORY_PATH", cache_file) def test_exact_match(self, tmp_path): platforms = { "discord": [ {"id": "111", "name": "bot-home", "guild": "MyServer", "type": "channel"}, {"id": "222", "name": "general", "guild": "MyServer", "type": "channel"}, ] } with self._setup(tmp_path, platforms): assert resolve_channel_name("discord", "bot-home") == "111" assert resolve_channel_name("discord", "#bot-home") == "111" def test_case_insensitive(self, tmp_path): platforms = { "slack": [{"id": "C01", "name": "Engineering", "type": "channel"}] } with self._setup(tmp_path, platforms): assert resolve_channel_name("slack", "engineering") == "C01" assert resolve_channel_name("slack", "ENGINEERING") == "C01" def test_guild_qualified_match(self, tmp_path): platforms = { "discord": [ {"id": "111", "name": "general", "guild": "ServerA", "type": "channel"}, {"id": "222", "name": "general", "guild": "ServerB", "type": "channel"}, ] } with self._setup(tmp_path, platforms): assert resolve_channel_name("discord", "ServerA/general") == "111" assert resolve_channel_name("discord", "ServerB/general") == "222" def test_prefix_match_unambiguous(self, tmp_path): platforms = { "slack": [ {"id": "C01", "name": "engineering-backend", "type": "channel"}, {"id": "C02", "name": "design-team", "type": "channel"}, ] } with self._setup(tmp_path, platforms): # "engineering" prefix matches only one channel assert resolve_channel_name("slack", "engineering") == "C01" def test_prefix_match_ambiguous_returns_none(self, tmp_path): platforms = { "slack": [ {"id": "C01", "name": "eng-backend", "type": "channel"}, {"id": "C02", "name": "eng-frontend", "type": "channel"}, ] } with self._setup(tmp_path, platforms): assert resolve_channel_name("slack", "eng") is None def test_no_channels_returns_none(self, tmp_path): with self._setup(tmp_path, {}): assert resolve_channel_name("telegram", "someone") is None def test_no_match_returns_none(self, tmp_path): platforms = { "telegram": [{"id": "123", "name": "John", "type": "dm"}] } with self._setup(tmp_path, platforms): assert resolve_channel_name("telegram", "nonexistent") is None def test_topic_name_resolves_to_composite_id(self, tmp_path): platforms = { "telegram": [{"id": "-1001:17585", "name": "Coaching Chat / topic 17585", "type": "group"}] } with self._setup(tmp_path, platforms): assert resolve_channel_name("telegram", "Coaching Chat / topic 17585") == "-1001:17585" def test_id_match_takes_precedence_over_name(self, tmp_path): """A raw channel ID resolves to itself, even when a different channel happens to be named the same string. Case-sensitive: Slack IDs are uppercase and must not be normalized away.""" platforms = { "slack": [ {"id": "C0B0QV5434G", "name": "engineering", "type": "channel"}, {"id": "C99", "name": "c0b0qv5434g", "type": "channel"}, ] } with self._setup(tmp_path, platforms): assert resolve_channel_name("slack", "C0B0QV5434G") == "C0B0QV5434G" # Lowercase still falls through to name matching (case-insensitive) assert resolve_channel_name("slack", "c0b0qv5434g") == "C99" def test_display_label_with_type_suffix_resolves(self, tmp_path): platforms = { "telegram": [ {"id": "123", "name": "Alice", "type": "dm"}, {"id": "456", "name": "Dev Group", "type": "group"}, {"id": "-1001:17585", "name": "Coaching Chat / topic 17585", "type": "group"}, ] } with self._setup(tmp_path, platforms): assert resolve_channel_name("telegram", "Alice (dm)") == "123" assert resolve_channel_name("telegram", "Dev Group (group)") == "456" assert resolve_channel_name("telegram", "Coaching Chat / topic 17585 (group)") == "-1001:17585" class TestBuildFromSessions: def _write_sessions(self, tmp_path, sessions_data): """Write sessions.json at the path _build_from_sessions expects.""" sessions_path = tmp_path / "sessions" / "sessions.json" sessions_path.parent.mkdir(parents=True) sessions_path.write_text(json.dumps(sessions_data)) def test_builds_from_sessions_json(self, tmp_path): self._write_sessions(tmp_path, { "session_1": { "origin": { "platform": "telegram", "chat_id": "12345", "chat_name": "Alice", }, "chat_type": "dm", }, "session_2": { "origin": { "platform": "telegram", "chat_id": "67890", "user_name": "Bob", }, "chat_type": "group", }, "session_3": { "origin": { "platform": "discord", "chat_id": "99999", }, }, }) with patch.dict(os.environ, {"HERMES_HOME": str(tmp_path)}): entries = _build_from_sessions("telegram") assert len(entries) == 2 names = {e["name"] for e in entries} assert "Alice" in names assert "Bob" in names def test_missing_sessions_file(self, tmp_path): with patch.dict(os.environ, {"HERMES_HOME": str(tmp_path)}): entries = _build_from_sessions("telegram") assert entries == [] def test_deduplication_by_chat_id(self, tmp_path): self._write_sessions(tmp_path, { "s1": {"origin": {"platform": "telegram", "chat_id": "123", "chat_name": "X"}}, "s2": {"origin": {"platform": "telegram", "chat_id": "123", "chat_name": "X"}}, }) with patch.dict(os.environ, {"HERMES_HOME": str(tmp_path)}): entries = _build_from_sessions("telegram") assert len(entries) == 1 def test_keeps_distinct_topics_with_same_chat_id(self, tmp_path): self._write_sessions(tmp_path, { "group_root": { "origin": {"platform": "telegram", "chat_id": "-1001", "chat_name": "Coaching Chat"}, "chat_type": "group", }, "topic_a": { "origin": { "platform": "telegram", "chat_id": "-1001", "chat_name": "Coaching Chat", "thread_id": "17585", }, "chat_type": "group", }, "topic_b": { "origin": { "platform": "telegram", "chat_id": "-1001", "chat_name": "Coaching Chat", "thread_id": "17587", }, "chat_type": "group", }, }) with patch.dict(os.environ, {"HERMES_HOME": str(tmp_path)}): entries = _build_from_sessions("telegram") ids = {entry["id"] for entry in entries} names = {entry["name"] for entry in entries} assert ids == {"-1001", "-1001:17585", "-1001:17587"} assert "Coaching Chat" in names assert "Coaching Chat / topic 17585" in names assert "Coaching Chat / topic 17587" in names class TestFormatDirectoryForDisplay: def test_empty_directory(self, tmp_path): with patch("gateway.channel_directory.DIRECTORY_PATH", tmp_path / "nope.json"): result = format_directory_for_display() assert "No messaging platforms" in result def test_telegram_display(self, tmp_path): cache_file = _write_directory(tmp_path, { "telegram": [ {"id": "123", "name": "Alice", "type": "dm"}, {"id": "456", "name": "Dev Group", "type": "group"}, {"id": "-1001:17585", "name": "Coaching Chat / topic 17585", "type": "group"}, ] }) with patch("gateway.channel_directory.DIRECTORY_PATH", cache_file): result = format_directory_for_display() assert "Telegram:" in result assert "telegram:Alice" in result assert "telegram:Dev Group" in result assert "telegram:Coaching Chat / topic 17585" in result def test_discord_grouped_by_guild(self, tmp_path): cache_file = _write_directory(tmp_path, { "discord": [ {"id": "1", "name": "general", "guild": "Server1", "type": "channel"}, {"id": "2", "name": "bot-home", "guild": "Server1", "type": "channel"}, {"id": "3", "name": "chat", "guild": "Server2", "type": "channel"}, ] }) with patch("gateway.channel_directory.DIRECTORY_PATH", cache_file): result = format_directory_for_display() assert "Discord (Server1):" in result assert "Discord (Server2):" in result assert "discord:#general" in result class TestLookupChannelType: def _setup(self, tmp_path, platforms): cache_file = _write_directory(tmp_path, platforms) return patch("gateway.channel_directory.DIRECTORY_PATH", cache_file) def test_forum_channel(self, tmp_path): platforms = { "discord": [ {"id": "100", "name": "ideas", "guild": "Server1", "type": "forum"}, ] } with self._setup(tmp_path, platforms): assert lookup_channel_type("discord", "100") == "forum" def test_regular_channel(self, tmp_path): platforms = { "discord": [ {"id": "200", "name": "general", "guild": "Server1", "type": "channel"}, ] } with self._setup(tmp_path, platforms): assert lookup_channel_type("discord", "200") == "channel" def test_unknown_chat_id_returns_none(self, tmp_path): platforms = { "discord": [ {"id": "200", "name": "general", "guild": "Server1", "type": "channel"}, ] } with self._setup(tmp_path, platforms): assert lookup_channel_type("discord", "999") is None def test_unknown_platform_returns_none(self, tmp_path): with self._setup(tmp_path, {}): assert lookup_channel_type("discord", "100") is None def test_channel_without_type_key_returns_none(self, tmp_path): platforms = { "discord": [ {"id": "300", "name": "general", "guild": "Server1"}, ] } with self._setup(tmp_path, platforms): assert lookup_channel_type("discord", "300") is None def _make_slack_adapter(team_clients): """Build a stand-in for SlackAdapter exposing only ``_team_clients``.""" return SimpleNamespace(_team_clients=team_clients) def _make_slack_client(pages): """Build an AsyncWebClient mock whose ``users_conversations`` returns pages.""" client = MagicMock() client.users_conversations = AsyncMock(side_effect=pages) return client class TestBuildSlack: """_build_slack actually calls users.conversations on each workspace client.""" def test_no_team_clients_falls_back_to_sessions(self, tmp_path): sessions_path = tmp_path / "sessions" / "sessions.json" sessions_path.parent.mkdir(parents=True) sessions_path.write_text(json.dumps({ "s1": {"origin": {"platform": "slack", "chat_id": "D123", "chat_name": "Alice"}}, })) with patch.dict(os.environ, {"HERMES_HOME": str(tmp_path)}): entries = asyncio.run(_build_slack(_make_slack_adapter({}))) assert len(entries) == 1 assert entries[0]["id"] == "D123" def test_lists_channels_from_users_conversations(self, tmp_path): client = _make_slack_client([ { "ok": True, "channels": [ {"id": "C0B0QV5434G", "name": "engineering", "is_private": False}, {"id": "G123ABCDEF", "name": "secret-chat", "is_private": True}, ], "response_metadata": {}, }, ]) with patch.dict(os.environ, {"HERMES_HOME": str(tmp_path)}): entries = asyncio.run(_build_slack(_make_slack_adapter({"T1": client}))) ids = {e["id"] for e in entries} assert ids == {"C0B0QV5434G", "G123ABCDEF"} types = {e["id"]: e["type"] for e in entries} assert types["C0B0QV5434G"] == "channel" assert types["G123ABCDEF"] == "private" client.users_conversations.assert_awaited_once() def test_paginates_via_response_metadata_cursor(self, tmp_path): client = _make_slack_client([ { "ok": True, "channels": [{"id": "C001", "name": "first", "is_private": False}], "response_metadata": {"next_cursor": "cur1"}, }, { "ok": True, "channels": [{"id": "C002", "name": "second", "is_private": False}], "response_metadata": {"next_cursor": ""}, }, ]) with patch.dict(os.environ, {"HERMES_HOME": str(tmp_path)}): entries = asyncio.run(_build_slack(_make_slack_adapter({"T1": client}))) assert {e["id"] for e in entries} == {"C001", "C002"} assert client.users_conversations.await_count == 2 def test_per_workspace_error_does_not_block_others(self, tmp_path): bad = MagicMock() bad.users_conversations = AsyncMock(side_effect=RuntimeError("boom")) good = _make_slack_client([ { "ok": True, "channels": [{"id": "C999", "name": "ok-channel", "is_private": False}], "response_metadata": {}, }, ]) with patch.dict(os.environ, {"HERMES_HOME": str(tmp_path)}): entries = asyncio.run(_build_slack(_make_slack_adapter({"BAD": bad, "GOOD": good}))) assert {e["id"] for e in entries} == {"C999"} def test_session_dms_merged_when_not_in_api_results(self, tmp_path): sessions_path = tmp_path / "sessions" / "sessions.json" sessions_path.parent.mkdir(parents=True) sessions_path.write_text(json.dumps({ "s1": {"origin": {"platform": "slack", "chat_id": "D456", "chat_name": "Bob"}}, "dup": {"origin": {"platform": "slack", "chat_id": "C001", "chat_name": "first"}}, })) client = _make_slack_client([ { "ok": True, "channels": [{"id": "C001", "name": "first", "is_private": False}], "response_metadata": {}, }, ]) with patch.dict(os.environ, {"HERMES_HOME": str(tmp_path)}): entries = asyncio.run(_build_slack(_make_slack_adapter({"T1": client}))) ids = {e["id"] for e in entries} assert "C001" in ids and "D456" in ids # Channel ID from API should not be duplicated by the session merge assert sum(1 for e in entries if e["id"] == "C001") == 1 def test_skips_channels_with_no_id_or_name(self, tmp_path): client = _make_slack_client([ { "ok": True, "channels": [ {"id": "C001", "name": "good", "is_private": False}, {"id": "", "name": "no-id"}, {"id": "C002"}, # no name (e.g. IM) ], "response_metadata": {}, }, ]) with patch.dict(os.environ, {"HERMES_HOME": str(tmp_path)}): entries = asyncio.run(_build_slack(_make_slack_adapter({"T1": client}))) assert {e["id"] for e in entries} == {"C001"} def test_response_not_ok_breaks_pagination_for_that_workspace(self, tmp_path): client = _make_slack_client([ {"ok": False, "error": "missing_scope"}, ]) with patch.dict(os.environ, {"HERMES_HOME": str(tmp_path)}): entries = asyncio.run(_build_slack(_make_slack_adapter({"T1": client}))) assert entries == []