mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-10-20 05:15:31 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

test+fix: rewrite migrated-platform test imports; fix connected-check discovery, telegram extra precedence, prompt_choice import

Browse source 
- Sweep test imports gateway.platforms.{telegram,feishu,matrix,dingtalk,whatsapp}
  (+ telegram_network / feishu satellites) -> plugins.platforms.<x>.adapter
  across ~68 test files; both 'from X import' and 'from gateway.platforms import X'
  forms.
- _is_platform_connected now forces discover_plugins() before the registry
  lookup so get_connected_platforms() works on directly-constructed GatewayConfig
  (fixes dingtalk-recognised tests after its checker moved to the plugin).
- telegram _apply_yaml_config no longer re-emits generic shared-config keys
  (reply_prefix etc.) that _merge_platform_map already merges with top-level
  precedence, while still passing through telegram-specific extras (base_url).
- prompt_choice lives in hermes_cli.setup, not cli_output — fixed the import in
  feishu/dingtalk interactive_setup and the test patch target.
- test_setup_feishu rewritten to drive interactive_setup via the plugin.
This commit is contained in:
teknium1 2026-06-07 05:27:58 -07:00
parent 736ffb3bc1
commit f080fedf13
No known key found for this signature in database
66 changed files with 669 additions and 648 deletions
Download patch file Download diff file Expand all files Collapse all files

10
gateway/config.py
View file

@ -532,9 +532,17 @@ class GatewayConfig:
if checker is not None:
return checker(config)
# Plugin-registered platforms
# Plugin-registered platforms. Force plugin discovery first so this
# works even when GatewayConfig is constructed directly (e.g. in tests
# or callers that bypass load_gateway_config(), which is what triggers
# discovery in the normal path). discover_plugins() is idempotent.
try:
from gateway.platform_registry import platform_registry
try:
from hermes_cli.plugins import discover_plugins
discover_plugins()
except Exception:
pass
entry = platform_registry.get(platform.value)
if entry:
if entry.is_connected is not None:

2
plugins/platforms/dingtalk/adapter.py
View file

@ -1565,9 +1565,9 @@ def interactive_setup() -> None:
lazy-imported so the plugin's module-load surface stays minimal.
"""
from hermes_cli.config import get_env_value, save_env_value
from hermes_cli.setup import prompt_choice
from hermes_cli.cli_output import (
prompt,
prompt_choice,
prompt_yes_no,
print_header,
print_success,

2
plugins/platforms/feishu/adapter.py
View file

@ -5300,9 +5300,9 @@ def interactive_setup() -> None:
_PLATFORMS["feishu"] dict. CLI helpers are lazy-imported.
"""
from hermes_cli.config import get_env_value, save_env_value
from hermes_cli.setup import prompt_choice
from hermes_cli.cli_output import (
prompt,
prompt_choice,
prompt_yes_no,
print_header,
print_info,

15
plugins/platforms/telegram/adapter.py
View file

@ -6252,8 +6252,21 @@ def _apply_yaml_config(yaml_cfg: dict, telegram_cfg: dict) -> dict | None:
for _key in ("guest_mode", "disable_link_previews", "observe_unmentioned_group_messages"):
if _key in telegram_cfg:
extras.setdefault(_key, telegram_cfg[_key])
# Pass through telegram-specific extra keys (e.g. base_url proxy override),
# but EXCLUDE the generic shared-config keys that _merge_platform_map in
# gateway/config.py already merges with correct top-level-over-nested
# precedence. The apply_yaml_config_fn dispatch merges our return via
# dict.update() (clobber), so re-emitting those generic keys here would
# undo that precedence (top-level losing to a nested-fallback block).
_GENERIC_MERGE_KEYS = {
"reply_prefix", "reply_in_thread", "reply_to_mode",
"unauthorized_dm_behavior", "notice_delivery", "require_mention",
"channel_skill_bindings", "channel_prompts", "gateway_restart_notification",
"allow_from", "allow_admin_from", "dm_policy", "group_policy",
}
for _k, _v in _telegram_extra.items():
extras.setdefault(_k, _v)
if _k not in _GENERIC_MERGE_KEYS:
extras.setdefault(_k, _v)
return extras or None

2
tests/e2e/conftest.py
View file

@ -118,7 +118,7 @@ _ensure_discord_mock()
_ensure_slack_mock()
import discord # noqa: E402 — mocked above
from gateway.platforms.telegram import TelegramAdapter # noqa: E402
from plugins.platforms.telegram.adapter import TelegramAdapter # noqa: E402
from plugins.platforms.discord.adapter import DiscordAdapter # noqa: E402
import gateway.platforms.slack as _slack_mod # noqa: E402

2
tests/gateway/conftest.py
View file

@ -2,7 +2,7 @@
The ``_ensure_telegram_mock`` helper guarantees that a minimal mock of
the ``telegram`` package is registered in :data:`sys.modules` **before**
any test file triggers ``from gateway.platforms.telegram import ...``.
any test file triggers ``from plugins.platforms.telegram.adapter import ...``.
Without this, ``pytest-xdist`` workers that happen to collect
``test_telegram_caption_merge.py`` (bare top-level import, no per-file

2
tests/gateway/feishu_helpers.py
View file

@ -35,7 +35,7 @@ def make_adapter_skeleton(
require_mention: bool = True,
group_policy: str = "allowlist",
) -> Any:
from gateway.platforms.feishu import FeishuAdapter
from plugins.platforms.feishu.adapter import FeishuAdapter
adapter = object.__new__(FeishuAdapter)
adapter._bot_open_id = bot_open_id

6
tests/gateway/test_allowed_channels_widening.py
View file

@ -24,7 +24,7 @@ from gateway.config import Platform, PlatformConfig
# ---------------------------------------------------------------------------
def _make_telegram_adapter(*, allowed_chats=None, require_mention=None, guest_mode=False):
from gateway.platforms.telegram import TelegramAdapter
from plugins.platforms.telegram.adapter import TelegramAdapter
extra = {"guest_mode": guest_mode}
if allowed_chats is not None:
@ -162,8 +162,8 @@ class TestTelegramAllowedChats:
def _make_dingtalk_adapter(*, allowed_chats=None, require_mention=None):
# Import lazily — DingTalk SDK may not be installed.
pytest.importorskip("gateway.platforms.dingtalk", reason="DingTalk adapter not importable")
from gateway.platforms.dingtalk import DingTalkAdapter
pytest.importorskip("plugins.platforms.dingtalk.adapter", reason="DingTalk adapter not importable")
from plugins.platforms.dingtalk.adapter import DingTalkAdapter
extra = {}
if allowed_chats is not None:

2
tests/gateway/test_config.py
View file

@ -535,7 +535,7 @@ class TestLoadGatewayConfig:
telegram = config.platforms[Platform.TELEGRAM]
assert telegram.extra.get("allow_from") == ["777888999"], (
"allow_from configured under gateway.platforms.telegram must be "
"allow_from configured under plugins.platforms.telegram.adapter must be "
"bridged into PlatformConfig.extra by the shared-key loop"
)
assert telegram.extra.get("require_mention") is False

2
tests/gateway/test_config_driven_access_policy.py
View file

@ -107,7 +107,7 @@ def test_base_adapter_defaults_to_not_owning_access_policy():
("gateway.platforms.weixin", "WeixinAdapter"),
("gateway.platforms.yuanbao", "YuanbaoAdapter"),
("gateway.platforms.qqbot.adapter", "QQAdapter"),
("gateway.platforms.whatsapp", "WhatsAppAdapter"),
("plugins.platforms.whatsapp.adapter", "WhatsAppAdapter"),
],
)
def test_own_policy_adapters_declare_the_flag(module_path, class_name):

108
tests/gateway/test_dingtalk.py
View file

@ -39,7 +39,7 @@ class _FakeChatbotMessage(SimpleNamespace):
@pytest.fixture(autouse=True)
def _fake_dingtalk_optional_sdks(monkeypatch):
"""Keep DingTalk adapter tests hermetic when optional SDKs are absent."""
from gateway.platforms import dingtalk as dt
import plugins.platforms.dingtalk.adapter as dt
card_models = SimpleNamespace(**{
name: _FakeDingTalkModel
@ -94,29 +94,29 @@ class TestDingTalkRequirements:
with patch.dict("sys.modules", {"dingtalk_stream": None}), \
patch("tools.lazy_deps.ensure", side_effect=ImportError("dingtalk_stream unavailable")):
monkeypatch.setattr(
"gateway.platforms.dingtalk.DINGTALK_STREAM_AVAILABLE", False
"plugins.platforms.dingtalk.adapter.DINGTALK_STREAM_AVAILABLE", False
)
from gateway.platforms.dingtalk import check_dingtalk_requirements
from plugins.platforms.dingtalk.adapter import check_dingtalk_requirements
assert check_dingtalk_requirements() is False
def test_returns_false_when_env_vars_missing(self, monkeypatch):
monkeypatch.setattr(
"gateway.platforms.dingtalk.DINGTALK_STREAM_AVAILABLE", True
"plugins.platforms.dingtalk.adapter.DINGTALK_STREAM_AVAILABLE", True
)
monkeypatch.setattr("gateway.platforms.dingtalk.HTTPX_AVAILABLE", True)
monkeypatch.setattr("plugins.platforms.dingtalk.adapter.HTTPX_AVAILABLE", True)
monkeypatch.delenv("DINGTALK_CLIENT_ID", raising=False)
monkeypatch.delenv("DINGTALK_CLIENT_SECRET", raising=False)
from gateway.platforms.dingtalk import check_dingtalk_requirements
from plugins.platforms.dingtalk.adapter import check_dingtalk_requirements
assert check_dingtalk_requirements() is False
def test_returns_true_when_all_available(self, monkeypatch):
monkeypatch.setattr(
"gateway.platforms.dingtalk.DINGTALK_STREAM_AVAILABLE", True
"plugins.platforms.dingtalk.adapter.DINGTALK_STREAM_AVAILABLE", True
)
monkeypatch.setattr("gateway.platforms.dingtalk.HTTPX_AVAILABLE", True)
monkeypatch.setattr("plugins.platforms.dingtalk.adapter.HTTPX_AVAILABLE", True)
monkeypatch.setenv("DINGTALK_CLIENT_ID", "test-id")
monkeypatch.setenv("DINGTALK_CLIENT_SECRET", "test-secret")
from gateway.platforms.dingtalk import check_dingtalk_requirements
from plugins.platforms.dingtalk.adapter import check_dingtalk_requirements
assert check_dingtalk_requirements() is True
@ -128,7 +128,7 @@ class TestDingTalkRequirements:
class TestDingTalkAdapterInit:
def test_reads_config_from_extra(self):
from gateway.platforms.dingtalk import DingTalkAdapter
from plugins.platforms.dingtalk.adapter import DingTalkAdapter
config = PlatformConfig(
enabled=True,
extra={"client_id": "cfg-id", "client_secret": "cfg-secret"},
@ -141,7 +141,7 @@ class TestDingTalkAdapterInit:
def test_falls_back_to_env_vars(self, monkeypatch):
monkeypatch.setenv("DINGTALK_CLIENT_ID", "env-id")
monkeypatch.setenv("DINGTALK_CLIENT_SECRET", "env-secret")
from gateway.platforms.dingtalk import DingTalkAdapter
from plugins.platforms.dingtalk.adapter import DingTalkAdapter
config = PlatformConfig(enabled=True)
adapter = DingTalkAdapter(config)
assert adapter._client_id == "env-id"
@ -156,28 +156,28 @@ class TestDingTalkAdapterInit:
class TestExtractText:
def test_extracts_dict_text(self):
from gateway.platforms.dingtalk import DingTalkAdapter
from plugins.platforms.dingtalk.adapter import DingTalkAdapter
msg = MagicMock()
msg.text = {"content": " hello world "}
msg.rich_text = None
assert DingTalkAdapter._extract_text(msg) == "hello world"
def test_extracts_string_text(self):
from gateway.platforms.dingtalk import DingTalkAdapter
from plugins.platforms.dingtalk.adapter import DingTalkAdapter
msg = MagicMock()
msg.text = "plain text"
msg.rich_text = None
assert DingTalkAdapter._extract_text(msg) == "plain text"
def test_falls_back_to_rich_text(self):
from gateway.platforms.dingtalk import DingTalkAdapter
from plugins.platforms.dingtalk.adapter import DingTalkAdapter
msg = MagicMock()
msg.text = ""
msg.rich_text = [{"text": "part1"}, {"text": "part2"}, {"image": "url"}]
assert DingTalkAdapter._extract_text(msg) == "part1 part2"
def test_returns_empty_for_no_content(self):
from gateway.platforms.dingtalk import DingTalkAdapter
from plugins.platforms.dingtalk.adapter import DingTalkAdapter
msg = MagicMock()
msg.text = ""
msg.rich_text = None
@ -192,24 +192,24 @@ class TestExtractText:
class TestDeduplication:
def test_first_message_not_duplicate(self):
from gateway.platforms.dingtalk import DingTalkAdapter
from plugins.platforms.dingtalk.adapter import DingTalkAdapter
adapter = DingTalkAdapter(PlatformConfig(enabled=True))
assert adapter._dedup.is_duplicate("msg-1") is False
def test_second_same_message_is_duplicate(self):
from gateway.platforms.dingtalk import DingTalkAdapter
from plugins.platforms.dingtalk.adapter import DingTalkAdapter
adapter = DingTalkAdapter(PlatformConfig(enabled=True))
adapter._dedup.is_duplicate("msg-1")
assert adapter._dedup.is_duplicate("msg-1") is True
def test_different_messages_not_duplicate(self):
from gateway.platforms.dingtalk import DingTalkAdapter
from plugins.platforms.dingtalk.adapter import DingTalkAdapter
adapter = DingTalkAdapter(PlatformConfig(enabled=True))
adapter._dedup.is_duplicate("msg-1")
assert adapter._dedup.is_duplicate("msg-2") is False
def test_cache_cleanup_on_overflow(self):
from gateway.platforms.dingtalk import DingTalkAdapter
from plugins.platforms.dingtalk.adapter import DingTalkAdapter
adapter = DingTalkAdapter(PlatformConfig(enabled=True))
max_size = adapter._dedup._max_size
# Fill beyond max
@ -228,7 +228,7 @@ class TestSend:
@pytest.mark.asyncio
async def test_send_posts_to_webhook(self):
from gateway.platforms.dingtalk import DingTalkAdapter
from plugins.platforms.dingtalk.adapter import DingTalkAdapter
adapter = DingTalkAdapter(PlatformConfig(enabled=True))
mock_response = MagicMock()
@ -254,7 +254,7 @@ class TestSend:
@pytest.mark.asyncio
async def test_send_fails_without_webhook(self):
from gateway.platforms.dingtalk import DingTalkAdapter
from plugins.platforms.dingtalk.adapter import DingTalkAdapter
adapter = DingTalkAdapter(PlatformConfig(enabled=True))
adapter._http_client = AsyncMock()
@ -264,7 +264,7 @@ class TestSend:
@pytest.mark.asyncio
async def test_send_uses_cached_webhook(self):
from gateway.platforms.dingtalk import DingTalkAdapter
from plugins.platforms.dingtalk.adapter import DingTalkAdapter
adapter = DingTalkAdapter(PlatformConfig(enabled=True))
mock_response = MagicMock()
@ -280,7 +280,7 @@ class TestSend:
@pytest.mark.asyncio
async def test_send_handles_http_error(self):
from gateway.platforms.dingtalk import DingTalkAdapter
from plugins.platforms.dingtalk.adapter import DingTalkAdapter
adapter = DingTalkAdapter(PlatformConfig(enabled=True))
mock_response = MagicMock()
@ -299,7 +299,7 @@ class TestSend:
@pytest.mark.asyncio
async def test_send_image_renders_markdown_image(self):
from gateway.platforms.dingtalk import DingTalkAdapter
from plugins.platforms.dingtalk.adapter import DingTalkAdapter
adapter = DingTalkAdapter(PlatformConfig(enabled=True))
mock_response = MagicMock()
@ -324,7 +324,7 @@ class TestSend:
@pytest.mark.asyncio
async def test_send_image_file_returns_explicit_unsupported_error(self):
from gateway.platforms.dingtalk import DingTalkAdapter
from plugins.platforms.dingtalk.adapter import DingTalkAdapter
adapter = DingTalkAdapter(PlatformConfig(enabled=True))
result = await adapter.send_image_file("chat-123", "/tmp/demo.png")
@ -334,7 +334,7 @@ class TestSend:
@pytest.mark.asyncio
async def test_send_document_returns_explicit_unsupported_error(self):
from gateway.platforms.dingtalk import DingTalkAdapter
from plugins.platforms.dingtalk.adapter import DingTalkAdapter
adapter = DingTalkAdapter(PlatformConfig(enabled=True))
result = await adapter.send_document("chat-123", "/tmp/demo.pdf")
@ -352,7 +352,7 @@ class TestConnect:
@pytest.mark.asyncio
async def test_disconnect_closes_session_websocket(self):
from gateway.platforms.dingtalk import DingTalkAdapter
from plugins.platforms.dingtalk.adapter import DingTalkAdapter
adapter = DingTalkAdapter(PlatformConfig(enabled=True))
websocket = AsyncMock()
@ -376,16 +376,16 @@ class TestConnect:
@pytest.mark.asyncio
async def test_connect_fails_without_sdk(self, monkeypatch):
monkeypatch.setattr(
"gateway.platforms.dingtalk.DINGTALK_STREAM_AVAILABLE", False
"plugins.platforms.dingtalk.adapter.DINGTALK_STREAM_AVAILABLE", False
)
from gateway.platforms.dingtalk import DingTalkAdapter
from plugins.platforms.dingtalk.adapter import DingTalkAdapter
adapter = DingTalkAdapter(PlatformConfig(enabled=True))
result = await adapter.connect()
assert result is False
@pytest.mark.asyncio
async def test_connect_fails_without_credentials(self):
from gateway.platforms.dingtalk import DingTalkAdapter
from plugins.platforms.dingtalk.adapter import DingTalkAdapter
adapter = DingTalkAdapter(PlatformConfig(enabled=True))
adapter._client_id = ""
adapter._client_secret = ""
@ -394,7 +394,7 @@ class TestConnect:
@pytest.mark.asyncio
async def test_disconnect_cleans_up(self):
from gateway.platforms.dingtalk import DingTalkAdapter
from plugins.platforms.dingtalk.adapter import DingTalkAdapter
adapter = DingTalkAdapter(PlatformConfig(enabled=True))
adapter._session_webhooks["a"] = "http://x"
adapter._dedup._seen["b"] = 1.0
@ -410,7 +410,7 @@ class TestConnect:
async def test_disconnect_finalizes_open_streaming_cards(self):
"""Streaming cards must be finalized before HTTP client closes."""
from unittest.mock import AsyncMock, patch
from gateway.platforms.dingtalk import DingTalkAdapter
from plugins.platforms.dingtalk.adapter import DingTalkAdapter
adapter = DingTalkAdapter(PlatformConfig(enabled=True))
adapter._http_client = AsyncMock()
adapter._stream_task = None
@ -456,29 +456,29 @@ class TestWebhookDomainAllowlist:
"""
def test_api_domain_accepted(self):
from gateway.platforms.dingtalk import _DINGTALK_WEBHOOK_RE
from plugins.platforms.dingtalk.adapter import _DINGTALK_WEBHOOK_RE
assert _DINGTALK_WEBHOOK_RE.match(
"https://api.dingtalk.com/robot/send?access_token=x"
)
def test_oapi_domain_accepted(self):
from gateway.platforms.dingtalk import _DINGTALK_WEBHOOK_RE
from plugins.platforms.dingtalk.adapter import _DINGTALK_WEBHOOK_RE
assert _DINGTALK_WEBHOOK_RE.match(
"https://oapi.dingtalk.com/robot/send?access_token=x"
)
def test_http_rejected(self):
from gateway.platforms.dingtalk import _DINGTALK_WEBHOOK_RE
from plugins.platforms.dingtalk.adapter import _DINGTALK_WEBHOOK_RE
assert not _DINGTALK_WEBHOOK_RE.match("http://api.dingtalk.com/robot/send")
def test_suffix_attack_rejected(self):
from gateway.platforms.dingtalk import _DINGTALK_WEBHOOK_RE
from plugins.platforms.dingtalk.adapter import _DINGTALK_WEBHOOK_RE
assert not _DINGTALK_WEBHOOK_RE.match(
"https://api.dingtalk.com.evil.example/"
)
def test_unsanctioned_subdomain_rejected(self):
from gateway.platforms.dingtalk import _DINGTALK_WEBHOOK_RE
from plugins.platforms.dingtalk.adapter import _DINGTALK_WEBHOOK_RE
# Only api.* and oapi.* are allowed — e.g. eapi.dingtalk.com must not slip through
assert not _DINGTALK_WEBHOOK_RE.match("https://eapi.dingtalk.com/robot/send")
@ -487,7 +487,7 @@ class TestHandlerProcessIsAsync:
"""dingtalk-stream >= 0.20 requires ``process`` to be a coroutine."""
def test_process_is_coroutine_function(self):
from gateway.platforms.dingtalk import _IncomingHandler
from plugins.platforms.dingtalk.adapter import _IncomingHandler
assert asyncio.iscoroutinefunction(_IncomingHandler.process)
@ -501,7 +501,7 @@ class TestExtractText:
"""
def test_text_as_dict_legacy(self):
from gateway.platforms.dingtalk import DingTalkAdapter
from plugins.platforms.dingtalk.adapter import DingTalkAdapter
msg = MagicMock()
msg.text = {"content": "hello world"}
msg.rich_text_content = None
@ -510,7 +510,7 @@ class TestExtractText:
def test_text_as_textcontent_object(self):
"""SDK >= 0.20 shape: object with ``.content`` attribute."""
from gateway.platforms.dingtalk import DingTalkAdapter
from plugins.platforms.dingtalk.adapter import DingTalkAdapter
class FakeTextContent:
content = "hello from new sdk"
@ -527,7 +527,7 @@ class TestExtractText:
assert "TextContent(" not in result
def test_text_content_attr_with_empty_string(self):
from gateway.platforms.dingtalk import DingTalkAdapter
from plugins.platforms.dingtalk.adapter import DingTalkAdapter
class FakeTextContent:
content = ""
@ -540,7 +540,7 @@ class TestExtractText:
def test_rich_text_content_new_shape(self):
"""SDK >= 0.20 exposes rich text as ``message.rich_text_content.rich_text_list``."""
from gateway.platforms.dingtalk import DingTalkAdapter
from plugins.platforms.dingtalk.adapter import DingTalkAdapter
class FakeRichText:
rich_text_list = [{"text": "hello "}, {"text": "world"}]
@ -554,7 +554,7 @@ class TestExtractText:
def test_rich_text_legacy_shape(self):
"""Legacy ``message.rich_text`` list remains supported."""
from gateway.platforms.dingtalk import DingTalkAdapter
from plugins.platforms.dingtalk.adapter import DingTalkAdapter
msg = MagicMock()
msg.text = None
msg.rich_text_content = None
@ -563,7 +563,7 @@ class TestExtractText:
assert "legacy" in result and "rich" in result
def test_empty_message(self):
from gateway.platforms.dingtalk import DingTalkAdapter
from plugins.platforms.dingtalk.adapter import DingTalkAdapter
msg = MagicMock()
msg.text = None
msg.rich_text_content = None
@ -586,7 +586,7 @@ class TestExtractMedia:
def test_voice_rich_text_item_classified_as_voice(self):
"""Native DingTalk voice notes (type=voice) must enter the auto-STT
path via MessageType.VOICE — the gateway skips STT for AUDIO."""
from gateway.platforms.dingtalk import DingTalkAdapter
from plugins.platforms.dingtalk.adapter import DingTalkAdapter
from gateway.platforms.base import MessageType
msg = self._msg_with_rich_text(
@ -602,7 +602,7 @@ class TestExtractMedia:
def test_audio_rich_text_item_stays_audio(self):
"""Generic audio uploads (e.g. an mp3 the user attached) must NOT
be auto-transcribed — they stay MessageType.AUDIO."""
from gateway.platforms.dingtalk import DingTalkAdapter, DINGTALK_TYPE_MAPPING
from plugins.platforms.dingtalk.adapter import DingTalkAdapter, DINGTALK_TYPE_MAPPING
from gateway.platforms.base import MessageType
# Simulate a future/non-voice audio rich-text item by extending the
@ -643,7 +643,7 @@ def _make_gating_adapter(monkeypatch, *, extra=None, env=None):
monkeypatch.delenv(key, raising=False)
for key, value in (env or {}).items():
monkeypatch.setenv(key, value)
from gateway.platforms.dingtalk import DingTalkAdapter
from plugins.platforms.dingtalk.adapter import DingTalkAdapter
return DingTalkAdapter(PlatformConfig(enabled=True, extra=extra or {}))
@ -790,7 +790,7 @@ class TestIncomingHandlerProcess:
@pytest.mark.asyncio
async def test_process_extracts_session_webhook(self):
"""session_webhook must be populated from callback data."""
from gateway.platforms.dingtalk import _IncomingHandler, DingTalkAdapter
from plugins.platforms.dingtalk.adapter import _IncomingHandler, DingTalkAdapter
adapter = DingTalkAdapter(PlatformConfig(enabled=True))
adapter._on_message = AsyncMock()
@ -823,7 +823,7 @@ class TestIncomingHandlerProcess:
"""If ChatbotMessage.from_dict does not map sessionWebhook (e.g. SDK
version mismatch), the handler should fall back to extracting it
directly from the raw data dict."""
from gateway.platforms.dingtalk import _IncomingHandler, DingTalkAdapter
from plugins.platforms.dingtalk.adapter import _IncomingHandler, DingTalkAdapter
adapter = DingTalkAdapter(PlatformConfig(enabled=True))
adapter._on_message = AsyncMock()
@ -851,7 +851,7 @@ class TestIncomingHandlerProcess:
async def test_process_returns_ack_immediately(self):
"""process() must not block on _on_message — it should return
the ACK tuple before the message is fully processed."""
from gateway.platforms.dingtalk import _IncomingHandler, DingTalkAdapter
from plugins.platforms.dingtalk.adapter import _IncomingHandler, DingTalkAdapter
processing_started = asyncio.Event()
processing_gate = asyncio.Event()
@ -895,7 +895,7 @@ class TestExtractTextMentions:
Stripping all @handles collateral-damages emails, SSH URLs, and
literal references the user wrote.
"""
from gateway.platforms.dingtalk import DingTalkAdapter
from plugins.platforms.dingtalk.adapter import DingTalkAdapter
cases = [
("@bot hello", "@bot hello"),
("contact alice@example.com", "contact alice@example.com"),
@ -928,7 +928,7 @@ class TestMessageContextIsolation:
def test_contexts_keyed_by_chat_id(self):
"""Two concurrent chats must not clobber each other's context."""
from gateway.platforms.dingtalk import DingTalkAdapter
from plugins.platforms.dingtalk.adapter import DingTalkAdapter
adapter = DingTalkAdapter(PlatformConfig(enabled=True))
msg_a = MagicMock(conversation_id="chat-A", sender_staff_id="user-A")
@ -953,7 +953,7 @@ class TestCardLifecycle:
@pytest.fixture
def adapter_with_card(self):
from gateway.platforms.dingtalk import DingTalkAdapter
from plugins.platforms.dingtalk.adapter import DingTalkAdapter
a = DingTalkAdapter(PlatformConfig(
enabled=True,
extra={"card_template_id": "tmpl-1"},
@ -1144,7 +1144,7 @@ class TestDingTalkAdapterAICards:
@pytest.mark.asyncio
async def test_send_uses_ai_card_if_configured(self, config, mock_stream_client, mock_http_client, mock_message):
from gateway.platforms.dingtalk import DingTalkAdapter
from plugins.platforms.dingtalk.adapter import DingTalkAdapter
adapter = DingTalkAdapter(config)
adapter._stream_client = mock_stream_client

4
tests/gateway/test_dm_topics.py
View file

@ -40,12 +40,12 @@ def _ensure_telegram_mock():
sys.modules["telegram.request"] = telegram_mod.request
# Force reimport so the adapter picks up the mock ChatType.
sys.modules.pop("gateway.platforms.telegram", None)
sys.modules.pop("plugins.platforms.telegram.adapter", None)
_ensure_telegram_mock()
from gateway.platforms.telegram import TelegramAdapter # noqa: E402
from plugins.platforms.telegram.adapter import TelegramAdapter # noqa: E402
def _make_adapter(dm_topics_config=None, group_topics_config=None):

482
tests/gateway/test_feishu.py
View file

File diff suppressed because it is too large Load diff

4
tests/gateway/test_feishu_approval_buttons.py
View file

@ -38,8 +38,8 @@ def _ensure_feishu_mocks():
_ensure_feishu_mocks()
from gateway.config import PlatformConfig
import gateway.platforms.feishu as feishu_module
from gateway.platforms.feishu import FeishuAdapter
import plugins.platforms.feishu.adapter as feishu_module
from plugins.platforms.feishu.adapter import FeishuAdapter
# ---------------------------------------------------------------------------

32
tests/gateway/test_feishu_bot_admission.py
View file

@ -28,7 +28,7 @@ from tests.gateway.feishu_helpers import (
],
)
def test_feishu_load_settings_populates_allow_bots(monkeypatch, env_value, expected):
from gateway.platforms.feishu import FeishuAdapter
from plugins.platforms.feishu.adapter import FeishuAdapter
monkeypatch.setenv("FEISHU_APP_ID", "cli_test")
monkeypatch.setenv("FEISHU_APP_SECRET", "secret_test")
@ -39,7 +39,7 @@ def test_feishu_load_settings_populates_allow_bots(monkeypatch, env_value, expec
def test_feishu_load_settings_allow_bots_defaults_to_none(monkeypatch):
from gateway.platforms.feishu import FeishuAdapter
from plugins.platforms.feishu.adapter import FeishuAdapter
monkeypatch.setenv("FEISHU_APP_ID", "cli_test")
monkeypatch.setenv("FEISHU_APP_SECRET", "secret_test")
@ -51,7 +51,7 @@ def test_feishu_load_settings_allow_bots_defaults_to_none(monkeypatch):
def test_feishu_load_settings_ignores_extra_allow_bots(monkeypatch):
# extra is ignored — env is single source of truth (yaml is bridged to env).
from gateway.platforms.feishu import FeishuAdapter
from plugins.platforms.feishu.adapter import FeishuAdapter
monkeypatch.setenv("FEISHU_APP_ID", "cli_test")
monkeypatch.setenv("FEISHU_APP_SECRET", "secret_test")
@ -62,7 +62,7 @@ def test_feishu_load_settings_ignores_extra_allow_bots(monkeypatch):
def test_feishu_load_settings_falls_back_to_env_when_extra_missing(monkeypatch):
from gateway.platforms.feishu import FeishuAdapter
from plugins.platforms.feishu.adapter import FeishuAdapter
monkeypatch.setenv("FEISHU_APP_ID", "cli_test")
monkeypatch.setenv("FEISHU_APP_SECRET", "secret_test")
@ -75,13 +75,13 @@ def test_feishu_load_settings_falls_back_to_env_when_extra_missing(monkeypatch):
def test_feishu_load_settings_warns_on_unknown_allow_bots(monkeypatch, caplog):
import logging
from gateway.platforms.feishu import FeishuAdapter
from plugins.platforms.feishu.adapter import FeishuAdapter
monkeypatch.setenv("FEISHU_APP_ID", "cli_test")
monkeypatch.setenv("FEISHU_APP_SECRET", "secret_test")
monkeypatch.setenv("FEISHU_ALLOW_BOTS", "menton") # typo
with caplog.at_level(logging.WARNING, logger="gateway.platforms.feishu"):
with caplog.at_level(logging.WARNING, logger="plugins.platforms.feishu.adapter"):
settings = FeishuAdapter._load_settings(extra={})
assert settings.allow_bots == "none"
@ -98,7 +98,7 @@ def test_feishu_load_settings_warns_on_unknown_allow_bots(monkeypatch, caplog):
],
)
def test_feishu_load_settings_require_mention(monkeypatch, env_value, extra, expected):
from gateway.platforms.feishu import FeishuAdapter
from plugins.platforms.feishu.adapter import FeishuAdapter
monkeypatch.setenv("FEISHU_APP_ID", "cli_test")
monkeypatch.setenv("FEISHU_APP_SECRET", "secret_test")
@ -112,7 +112,7 @@ def test_feishu_load_settings_require_mention(monkeypatch, env_value, extra, exp
def test_feishu_load_settings_parses_per_group_require_mention(monkeypatch):
from gateway.platforms.feishu import FeishuAdapter
from plugins.platforms.feishu.adapter import FeishuAdapter
monkeypatch.setenv("FEISHU_APP_ID", "cli_test")
monkeypatch.setenv("FEISHU_APP_SECRET", "secret_test")
@ -133,7 +133,7 @@ def test_feishu_load_settings_parses_per_group_require_mention(monkeypatch):
def test_sender_identity_collects_every_non_empty_id_variant():
from gateway.platforms.feishu import _sender_identity
from plugins.platforms.feishu.adapter import _sender_identity
sender = SimpleNamespace(
sender_id=SimpleNamespace(open_id="ou_x", user_id="", union_id="un_x"),
@ -142,21 +142,21 @@ def test_sender_identity_collects_every_non_empty_id_variant():
def test_sender_identity_handles_missing_sender_id():
from gateway.platforms.feishu import _sender_identity
from plugins.platforms.feishu.adapter import _sender_identity
assert _sender_identity(SimpleNamespace()) == frozenset()
@pytest.mark.parametrize("sender_type", ["bot", "app"])
def test_is_bot_sender_treats_bot_and_app_as_bot_origin(sender_type):
from gateway.platforms.feishu import _is_bot_sender
from plugins.platforms.feishu.adapter import _is_bot_sender
assert _is_bot_sender(SimpleNamespace(sender_type=sender_type)) is True
@pytest.mark.parametrize("sender_type", ["user", "", None])
def test_is_bot_sender_rejects_non_bot_origin(sender_type):
from gateway.platforms.feishu import _is_bot_sender
from plugins.platforms.feishu.adapter import _is_bot_sender
assert _is_bot_sender(SimpleNamespace(sender_type=sender_type)) is False
@ -430,7 +430,7 @@ def test_admit_group_mention_checked_once_per_call():
def test_admit_per_group_require_mention_overrides_global():
from gateway.platforms.feishu import FeishuGroupRule
from plugins.platforms.feishu.adapter import FeishuGroupRule
adapter = make_adapter_skeleton(
bot_open_id="ou_self", require_mention=True, group_policy="open",
@ -454,7 +454,7 @@ def test_admit_per_group_require_mention_overrides_global():
def test_hydrate_bot_identity_populates_self_ids_from_bot_v3_info(monkeypatch):
import asyncio
from gateway.platforms import feishu as feishu_mod
import plugins.platforms.feishu.adapter as feishu_mod
FeishuAdapter = feishu_mod.FeishuAdapter
class _FakeBaseRequestBuilder:
@ -515,7 +515,7 @@ def test_hydrate_bot_identity_populates_self_ids_from_bot_v3_info(monkeypatch):
def test_resolve_sender_profile_uses_open_id_for_bot_name_lookup():
import asyncio
from gateway.platforms.feishu import FeishuAdapter
from plugins.platforms.feishu.adapter import FeishuAdapter
adapter = object.__new__(FeishuAdapter)
adapter._client = object()
@ -569,7 +569,7 @@ def _group_case(
def _group_rule(policy: str, **kwargs):
from gateway.platforms.feishu import FeishuGroupRule
from plugins.platforms.feishu.adapter import FeishuGroupRule
return FeishuGroupRule(policy=policy, **kwargs)

90
tests/gateway/test_feishu_comment.py
View file

@ -5,7 +5,7 @@ import unittest
from types import SimpleNamespace
from unittest.mock import AsyncMock, Mock, patch
from gateway.platforms.feishu_comment import (
from plugins.platforms.feishu.feishu_comment import (
parse_drive_comment_event,
_ALLOWED_NOTICE_TYPES,
_sanitize_comment_text,
@ -62,45 +62,45 @@ class TestEventFiltering(unittest.TestCase):
def _run(self, coro):
return asyncio.get_event_loop().run_until_complete(coro)
@patch("gateway.platforms.feishu_comment_rules.load_config")
@patch("gateway.platforms.feishu_comment_rules.resolve_rule")
@patch("gateway.platforms.feishu_comment_rules.is_user_allowed")
@patch("plugins.platforms.feishu.feishu_comment_rules.load_config")
@patch("plugins.platforms.feishu.feishu_comment_rules.resolve_rule")
@patch("plugins.platforms.feishu.feishu_comment_rules.is_user_allowed")
def test_self_reply_filtered(self, mock_allowed, mock_resolve, mock_load):
"""Events where from_open_id == self_open_id should be dropped."""
from gateway.platforms.feishu_comment import handle_drive_comment_event
from plugins.platforms.feishu.feishu_comment import handle_drive_comment_event
evt = _make_event(from_open_id="ou_bot", to_open_id="ou_bot")
self._run(handle_drive_comment_event(Mock(), evt, self_open_id="ou_bot"))
mock_load.assert_not_called()
@patch("gateway.platforms.feishu_comment_rules.load_config")
@patch("gateway.platforms.feishu_comment_rules.resolve_rule")
@patch("gateway.platforms.feishu_comment_rules.is_user_allowed")
@patch("plugins.platforms.feishu.feishu_comment_rules.load_config")
@patch("plugins.platforms.feishu.feishu_comment_rules.resolve_rule")
@patch("plugins.platforms.feishu.feishu_comment_rules.is_user_allowed")
def test_wrong_receiver_filtered(self, mock_allowed, mock_resolve, mock_load):
"""Events where to_open_id != self_open_id should be dropped."""
from gateway.platforms.feishu_comment import handle_drive_comment_event
from plugins.platforms.feishu.feishu_comment import handle_drive_comment_event
evt = _make_event(to_open_id="ou_other_bot")
self._run(handle_drive_comment_event(Mock(), evt, self_open_id="ou_bot"))
mock_load.assert_not_called()
@patch("gateway.platforms.feishu_comment_rules.load_config")
@patch("gateway.platforms.feishu_comment_rules.resolve_rule")
@patch("gateway.platforms.feishu_comment_rules.is_user_allowed")
@patch("plugins.platforms.feishu.feishu_comment_rules.load_config")
@patch("plugins.platforms.feishu.feishu_comment_rules.resolve_rule")
@patch("plugins.platforms.feishu.feishu_comment_rules.is_user_allowed")
def test_empty_to_open_id_filtered(self, mock_allowed, mock_resolve, mock_load):
"""Events with empty to_open_id should be dropped."""
from gateway.platforms.feishu_comment import handle_drive_comment_event
from plugins.platforms.feishu.feishu_comment import handle_drive_comment_event
evt = _make_event(to_open_id="")
self._run(handle_drive_comment_event(Mock(), evt, self_open_id="ou_bot"))
mock_load.assert_not_called()
@patch("gateway.platforms.feishu_comment_rules.load_config")
@patch("gateway.platforms.feishu_comment_rules.resolve_rule")
@patch("gateway.platforms.feishu_comment_rules.is_user_allowed")
@patch("plugins.platforms.feishu.feishu_comment_rules.load_config")
@patch("plugins.platforms.feishu.feishu_comment_rules.resolve_rule")
@patch("plugins.platforms.feishu.feishu_comment_rules.is_user_allowed")
def test_invalid_notice_type_filtered(self, mock_allowed, mock_resolve, mock_load):
"""Events with unsupported notice_type should be dropped."""
from gateway.platforms.feishu_comment import handle_drive_comment_event
from plugins.platforms.feishu.feishu_comment import handle_drive_comment_event
evt = _make_event(notice_type="resolve_comment")
self._run(handle_drive_comment_event(Mock(), evt, self_open_id="ou_bot"))
@ -116,14 +116,14 @@ class TestAccessControlIntegration(unittest.TestCase):
def _run(self, coro):
return asyncio.get_event_loop().run_until_complete(coro)
@patch("gateway.platforms.feishu_comment_rules.has_wiki_keys", return_value=False)
@patch("gateway.platforms.feishu_comment_rules.is_user_allowed", return_value=False)
@patch("gateway.platforms.feishu_comment_rules.resolve_rule")
@patch("gateway.platforms.feishu_comment_rules.load_config")
@patch("plugins.platforms.feishu.feishu_comment_rules.has_wiki_keys", return_value=False)
@patch("plugins.platforms.feishu.feishu_comment_rules.is_user_allowed", return_value=False)
@patch("plugins.platforms.feishu.feishu_comment_rules.resolve_rule")
@patch("plugins.platforms.feishu.feishu_comment_rules.load_config")
def test_denied_user_no_side_effects(self, mock_load, mock_resolve, mock_allowed, mock_wiki_keys):
"""Denied user should not trigger typing reaction or agent."""
from gateway.platforms.feishu_comment import handle_drive_comment_event
from gateway.platforms.feishu_comment_rules import ResolvedCommentRule
from plugins.platforms.feishu.feishu_comment import handle_drive_comment_event
from plugins.platforms.feishu.feishu_comment_rules import ResolvedCommentRule
mock_resolve.return_value = ResolvedCommentRule(True, "allowlist", frozenset(), "top")
mock_load.return_value = Mock()
@ -135,14 +135,14 @@ class TestAccessControlIntegration(unittest.TestCase):
# No API calls should be made for denied users
client.request.assert_not_called()
@patch("gateway.platforms.feishu_comment_rules.has_wiki_keys", return_value=False)
@patch("gateway.platforms.feishu_comment_rules.is_user_allowed", return_value=False)
@patch("gateway.platforms.feishu_comment_rules.resolve_rule")
@patch("gateway.platforms.feishu_comment_rules.load_config")
@patch("plugins.platforms.feishu.feishu_comment_rules.has_wiki_keys", return_value=False)
@patch("plugins.platforms.feishu.feishu_comment_rules.is_user_allowed", return_value=False)
@patch("plugins.platforms.feishu.feishu_comment_rules.resolve_rule")
@patch("plugins.platforms.feishu.feishu_comment_rules.load_config")
def test_disabled_comment_skipped(self, mock_load, mock_resolve, mock_allowed, mock_wiki_keys):
"""Disabled comments should return immediately."""
from gateway.platforms.feishu_comment import handle_drive_comment_event
from gateway.platforms.feishu_comment_rules import ResolvedCommentRule
from plugins.platforms.feishu.feishu_comment import handle_drive_comment_event
from plugins.platforms.feishu.feishu_comment_rules import ResolvedCommentRule
mock_resolve.return_value = ResolvedCommentRule(False, "allowlist", frozenset(), "top")
mock_load.return_value = Mock()
@ -184,9 +184,9 @@ class TestWikiReverseLookup(unittest.TestCase):
def _run(self, coro):
return asyncio.get_event_loop().run_until_complete(coro)
@patch("gateway.platforms.feishu_comment._exec_request")
@patch("plugins.platforms.feishu.feishu_comment._exec_request")
def test_reverse_lookup_success(self, mock_exec):
from gateway.platforms.feishu_comment import _reverse_lookup_wiki_token
from plugins.platforms.feishu.feishu_comment import _reverse_lookup_wiki_token
mock_exec.return_value = (0, "Success", {
"node": {"node_token": "WIKI_TOKEN_123", "obj_token": "docx_abc"},
@ -200,37 +200,37 @@ class TestWikiReverseLookup(unittest.TestCase):
self.assertEqual(query_dict["token"], "docx_abc")
self.assertEqual(query_dict["obj_type"], "docx")
@patch("gateway.platforms.feishu_comment._exec_request")
@patch("plugins.platforms.feishu.feishu_comment._exec_request")
def test_reverse_lookup_not_wiki(self, mock_exec):
from gateway.platforms.feishu_comment import _reverse_lookup_wiki_token
from plugins.platforms.feishu.feishu_comment import _reverse_lookup_wiki_token
mock_exec.return_value = (131001, "not found", {})
result = self._run(_reverse_lookup_wiki_token(Mock(), "docx", "docx_abc"))
self.assertIsNone(result)
@patch("gateway.platforms.feishu_comment._exec_request")
@patch("plugins.platforms.feishu.feishu_comment._exec_request")
def test_reverse_lookup_service_error(self, mock_exec):
from gateway.platforms.feishu_comment import _reverse_lookup_wiki_token
from plugins.platforms.feishu.feishu_comment import _reverse_lookup_wiki_token
mock_exec.return_value = (500, "internal error", {})
result = self._run(_reverse_lookup_wiki_token(Mock(), "docx", "docx_abc"))
self.assertIsNone(result)
@patch("gateway.platforms.feishu_comment._reverse_lookup_wiki_token", new_callable=AsyncMock)
@patch("gateway.platforms.feishu_comment_rules.has_wiki_keys", return_value=True)
@patch("gateway.platforms.feishu_comment_rules.is_user_allowed", return_value=True)
@patch("gateway.platforms.feishu_comment_rules.resolve_rule")
@patch("gateway.platforms.feishu_comment_rules.load_config")
@patch("gateway.platforms.feishu_comment.add_comment_reaction", new_callable=AsyncMock)
@patch("gateway.platforms.feishu_comment.batch_query_comment", new_callable=AsyncMock)
@patch("gateway.platforms.feishu_comment.query_document_meta", new_callable=AsyncMock)
@patch("plugins.platforms.feishu.feishu_comment._reverse_lookup_wiki_token", new_callable=AsyncMock)
@patch("plugins.platforms.feishu.feishu_comment_rules.has_wiki_keys", return_value=True)
@patch("plugins.platforms.feishu.feishu_comment_rules.is_user_allowed", return_value=True)
@patch("plugins.platforms.feishu.feishu_comment_rules.resolve_rule")
@patch("plugins.platforms.feishu.feishu_comment_rules.load_config")
@patch("plugins.platforms.feishu.feishu_comment.add_comment_reaction", new_callable=AsyncMock)
@patch("plugins.platforms.feishu.feishu_comment.batch_query_comment", new_callable=AsyncMock)
@patch("plugins.platforms.feishu.feishu_comment.query_document_meta", new_callable=AsyncMock)
def test_wiki_lookup_triggered_when_no_exact_match(
self, mock_meta, mock_batch, mock_reaction,
mock_load, mock_resolve, mock_allowed, mock_wiki_keys, mock_lookup,
):
"""Wiki reverse lookup should fire when rule falls to wildcard/top and wiki keys exist."""
from gateway.platforms.feishu_comment import handle_drive_comment_event
from gateway.platforms.feishu_comment_rules import ResolvedCommentRule
from plugins.platforms.feishu.feishu_comment import handle_drive_comment_event
from plugins.platforms.feishu.feishu_comment_rules import ResolvedCommentRule
# First resolve returns wildcard (no exact match), second returns exact wiki match
mock_resolve.side_effect = [

14
tests/gateway/test_feishu_comment_rules.py
View file

@ -8,7 +8,7 @@ import unittest
from pathlib import Path
from unittest.mock import patch
from gateway.platforms.feishu_comment_rules import (
from plugins.platforms.feishu.feishu_comment_rules import (
CommentsConfig,
CommentDocumentRule,
ResolvedCommentRule,
@ -195,7 +195,7 @@ class TestIsUserAllowed(unittest.TestCase):
def test_pairing_checks_store(self):
rule = ResolvedCommentRule(True, "pairing", frozenset(), "top")
with patch(
"gateway.platforms.feishu_comment_rules._load_pairing_approved",
"plugins.platforms.feishu.feishu_comment_rules._load_pairing_approved",
return_value={"ou_approved"},
):
self.assertTrue(is_user_allowed(rule, "ou_approved"))
@ -256,8 +256,8 @@ class TestLoadConfig(unittest.TestCase):
json.dump(raw, f)
path = Path(f.name)
try:
with patch("gateway.platforms.feishu_comment_rules.RULES_FILE", path):
with patch("gateway.platforms.feishu_comment_rules._rules_cache", _MtimeCache(path)):
with patch("plugins.platforms.feishu.feishu_comment_rules.RULES_FILE", path):
with patch("plugins.platforms.feishu.feishu_comment_rules._rules_cache", _MtimeCache(path)):
cfg = load_config()
self.assertTrue(cfg.enabled)
self.assertEqual(cfg.policy, "allowlist")
@ -269,7 +269,7 @@ class TestLoadConfig(unittest.TestCase):
path.unlink()
def test_load_missing_file_returns_defaults(self):
with patch("gateway.platforms.feishu_comment_rules._rules_cache", _MtimeCache(Path("/nonexistent"))):
with patch("plugins.platforms.feishu.feishu_comment_rules._rules_cache", _MtimeCache(Path("/nonexistent"))):
cfg = load_config()
self.assertTrue(cfg.enabled)
self.assertEqual(cfg.policy, "pairing")
@ -283,9 +283,9 @@ class TestPairingStore(unittest.TestCase):
self._pairing_file = Path(self._tmpdir) / "pairing.json"
with open(self._pairing_file, "w") as f:
json.dump({"approved": {}}, f)
self._patcher_file = patch("gateway.platforms.feishu_comment_rules.PAIRING_FILE", self._pairing_file)
self._patcher_file = patch("plugins.platforms.feishu.feishu_comment_rules.PAIRING_FILE", self._pairing_file)
self._patcher_cache = patch(
"gateway.platforms.feishu_comment_rules._pairing_cache",
"plugins.platforms.feishu.feishu_comment_rules._pairing_cache",
_MtimeCache(self._pairing_file),
)
self._patcher_file.start()

4
tests/gateway/test_feishu_meeting_invite.py
View file

@ -6,7 +6,7 @@ from types import SimpleNamespace
from unittest.mock import patch
from gateway.platforms.base import MessageEvent
from gateway.platforms.feishu_meeting_invite import (
from plugins.platforms.feishu.feishu_meeting_invite import (
build_meeting_invite_prompt,
handle_meeting_invited_event,
parse_meeting_invited_event,
@ -212,7 +212,7 @@ class TestMeetingInviteSendRouting(unittest.TestCase):
def test_feishu_user_id_prefix_sends_with_user_id_receive_type(self):
from gateway.config import PlatformConfig
from gateway.platforms.feishu import FeishuAdapter
from plugins.platforms.feishu.adapter import FeishuAdapter
created_requests = []

156
tests/gateway/test_feishu_onboard.py
View file

@ -1,4 +1,4 @@
"""Tests for gateway.platforms.feishu — Feishu scan-to-create registration."""
"""Tests for plugins.platforms.feishu.adapter — Feishu scan-to-create registration."""
import json
from unittest.mock import patch, MagicMock
@ -18,18 +18,18 @@ def _mock_urlopen(response_data, status=200):
class TestPostRegistration:
"""Tests for the low-level HTTP helper."""
@patch("gateway.platforms.feishu.urlopen")
@patch("plugins.platforms.feishu.adapter.urlopen")
def test_post_registration_returns_parsed_json(self, mock_urlopen_fn):
from gateway.platforms.feishu import _post_registration
from plugins.platforms.feishu.adapter import _post_registration
mock_urlopen_fn.return_value = _mock_urlopen({"nonce": "abc", "supported_auth_methods": ["client_secret"]})
result = _post_registration("https://accounts.feishu.cn", {"action": "init"})
assert result["nonce"] == "abc"
assert "client_secret" in result["supported_auth_methods"]
@patch("gateway.platforms.feishu.urlopen")
@patch("plugins.platforms.feishu.adapter.urlopen")
def test_post_registration_sends_form_encoded_body(self, mock_urlopen_fn):
from gateway.platforms.feishu import _post_registration
from plugins.platforms.feishu.adapter import _post_registration
mock_urlopen_fn.return_value = _mock_urlopen({})
_post_registration("https://accounts.feishu.cn", {"action": "init", "key": "val"})
@ -44,9 +44,9 @@ class TestPostRegistration:
class TestInitRegistration:
"""Tests for the init step."""
@patch("gateway.platforms.feishu.urlopen")
@patch("plugins.platforms.feishu.adapter.urlopen")
def test_init_succeeds_when_client_secret_supported(self, mock_urlopen_fn):
from gateway.platforms.feishu import _init_registration
from plugins.platforms.feishu.adapter import _init_registration
mock_urlopen_fn.return_value = _mock_urlopen({
"nonce": "abc",
@ -54,9 +54,9 @@ class TestInitRegistration:
})
_init_registration("feishu")
@patch("gateway.platforms.feishu.urlopen")
@patch("plugins.platforms.feishu.adapter.urlopen")
def test_init_raises_when_client_secret_not_supported(self, mock_urlopen_fn):
from gateway.platforms.feishu import _init_registration
from plugins.platforms.feishu.adapter import _init_registration
mock_urlopen_fn.return_value = _mock_urlopen({
"nonce": "abc",
@ -65,9 +65,9 @@ class TestInitRegistration:
with pytest.raises(RuntimeError, match="client_secret"):
_init_registration("feishu")
@patch("gateway.platforms.feishu.urlopen")
@patch("plugins.platforms.feishu.adapter.urlopen")
def test_init_uses_lark_url_for_lark_domain(self, mock_urlopen_fn):
from gateway.platforms.feishu import _init_registration
from plugins.platforms.feishu.adapter import _init_registration
mock_urlopen_fn.return_value = _mock_urlopen({
"nonce": "abc",
@ -82,9 +82,9 @@ class TestInitRegistration:
class TestBeginRegistration:
"""Tests for the begin step."""
@patch("gateway.platforms.feishu.urlopen")
@patch("plugins.platforms.feishu.adapter.urlopen")
def test_begin_returns_device_code_and_qr_url(self, mock_urlopen_fn):
from gateway.platforms.feishu import _begin_registration
from plugins.platforms.feishu.adapter import _begin_registration
mock_urlopen_fn.return_value = _mock_urlopen({
"device_code": "dc_123",
@ -101,9 +101,9 @@ class TestBeginRegistration:
assert result["interval"] == 5
assert result["expire_in"] == 600
@patch("gateway.platforms.feishu.urlopen")
@patch("plugins.platforms.feishu.adapter.urlopen")
def test_begin_sends_correct_archetype(self, mock_urlopen_fn):
from gateway.platforms.feishu import _begin_registration
from plugins.platforms.feishu.adapter import _begin_registration
mock_urlopen_fn.return_value = _mock_urlopen({
"device_code": "dc_123",
@ -122,10 +122,10 @@ class TestBeginRegistration:
class TestPollRegistration:
"""Tests for the poll step."""
@patch("gateway.platforms.feishu.time")
@patch("gateway.platforms.feishu.urlopen")
@patch("plugins.platforms.feishu.adapter.time")
@patch("plugins.platforms.feishu.adapter.urlopen")
def test_poll_returns_credentials_on_success(self, mock_urlopen_fn, mock_time):
from gateway.platforms.feishu import _poll_registration
from plugins.platforms.feishu.adapter import _poll_registration
mock_time.monotonic.side_effect = [0, 1]
mock_time.sleep = MagicMock()
@ -144,10 +144,10 @@ class TestPollRegistration:
assert result["domain"] == "feishu"
assert result["open_id"] == "ou_owner"
@patch("gateway.platforms.feishu.time")
@patch("gateway.platforms.feishu.urlopen")
@patch("plugins.platforms.feishu.adapter.time")
@patch("plugins.platforms.feishu.adapter.urlopen")
def test_poll_switches_domain_on_lark_tenant_brand(self, mock_urlopen_fn, mock_time):
from gateway.platforms.feishu import _poll_registration
from plugins.platforms.feishu.adapter import _poll_registration
mock_time.monotonic.side_effect = [0, 1, 2]
mock_time.sleep = MagicMock()
@ -169,11 +169,11 @@ class TestPollRegistration:
assert result is not None
assert result["domain"] == "lark"
@patch("gateway.platforms.feishu.time")
@patch("gateway.platforms.feishu.urlopen")
@patch("plugins.platforms.feishu.adapter.time")
@patch("plugins.platforms.feishu.adapter.urlopen")
def test_poll_success_with_lark_brand_in_same_response(self, mock_urlopen_fn, mock_time):
"""Credentials and lark tenant_brand in one response must not be discarded."""
from gateway.platforms.feishu import _poll_registration
from plugins.platforms.feishu.adapter import _poll_registration
mock_time.monotonic.side_effect = [0, 1]
mock_time.sleep = MagicMock()
@ -191,10 +191,10 @@ class TestPollRegistration:
assert result["domain"] == "lark"
assert result["open_id"] == "ou_lark_direct"
@patch("gateway.platforms.feishu.time")
@patch("gateway.platforms.feishu.urlopen")
@patch("plugins.platforms.feishu.adapter.time")
@patch("plugins.platforms.feishu.adapter.urlopen")
def test_poll_returns_none_on_access_denied(self, mock_urlopen_fn, mock_time):
from gateway.platforms.feishu import _poll_registration
from plugins.platforms.feishu.adapter import _poll_registration
mock_time.monotonic.side_effect = [0, 1]
mock_time.sleep = MagicMock()
@ -207,10 +207,10 @@ class TestPollRegistration:
)
assert result is None
@patch("gateway.platforms.feishu.time")
@patch("gateway.platforms.feishu.urlopen")
@patch("plugins.platforms.feishu.adapter.time")
@patch("plugins.platforms.feishu.adapter.urlopen")
def test_poll_returns_none_on_timeout(self, mock_urlopen_fn, mock_time):
from gateway.platforms.feishu import _poll_registration
from plugins.platforms.feishu.adapter import _poll_registration
mock_time.monotonic.side_effect = [0, 999]
mock_time.sleep = MagicMock()
@ -223,10 +223,10 @@ class TestPollRegistration:
)
assert result is None
@patch("gateway.platforms.feishu.time")
@patch("gateway.platforms.feishu.urlopen")
@patch("plugins.platforms.feishu.adapter.time")
@patch("plugins.platforms.feishu.adapter.urlopen")
def test_poll_timeout_uses_monotonic_clock(self, mock_urlopen_fn, mock_time):
from gateway.platforms.feishu import _poll_registration
from plugins.platforms.feishu.adapter import _poll_registration
mock_time.monotonic.side_effect = [1000, 1000.2, 1001.1]
mock_time.time.side_effect = [1000, 900, 901, 902]
@ -246,9 +246,9 @@ class TestPollRegistration:
class TestRenderQr:
"""Tests for QR code terminal rendering."""
@patch("gateway.platforms.feishu._qrcode_mod", create=True)
@patch("plugins.platforms.feishu.adapter._qrcode_mod", create=True)
def test_render_qr_returns_true_on_success(self, mock_qrcode_mod):
from gateway.platforms.feishu import _render_qr
from plugins.platforms.feishu.adapter import _render_qr
mock_qr = MagicMock()
mock_qrcode_mod.QRCode.return_value = mock_qr
@ -258,20 +258,20 @@ class TestRenderQr:
mock_qr.print_ascii.assert_called_once()
def test_render_qr_returns_false_when_qrcode_missing(self):
from gateway.platforms.feishu import _render_qr
from plugins.platforms.feishu.adapter import _render_qr
with patch("gateway.platforms.feishu._qrcode_mod", None):
with patch("plugins.platforms.feishu.adapter._qrcode_mod", None):
assert _render_qr("https://example.com/qr") is False
class TestProbeBot:
"""Tests for bot connectivity verification."""
@patch("gateway.platforms.feishu.FEISHU_AVAILABLE", True)
@patch("plugins.platforms.feishu.adapter.FEISHU_AVAILABLE", True)
def test_probe_returns_bot_info_on_success(self):
from gateway.platforms.feishu import probe_bot
from plugins.platforms.feishu.adapter import probe_bot
with patch("gateway.platforms.feishu._probe_bot_sdk") as mock_sdk:
with patch("plugins.platforms.feishu.adapter._probe_bot_sdk") as mock_sdk:
mock_sdk.return_value = {"bot_name": "TestBot", "bot_open_id": "ou_bot123"}
result = probe_bot("cli_app", "secret", "feishu")
@ -279,21 +279,21 @@ class TestProbeBot:
assert result["bot_name"] == "TestBot"
assert result["bot_open_id"] == "ou_bot123"
@patch("gateway.platforms.feishu.FEISHU_AVAILABLE", True)
@patch("plugins.platforms.feishu.adapter.FEISHU_AVAILABLE", True)
def test_probe_returns_none_on_failure(self):
from gateway.platforms.feishu import probe_bot
from plugins.platforms.feishu.adapter import probe_bot
with patch("gateway.platforms.feishu._probe_bot_sdk") as mock_sdk:
with patch("plugins.platforms.feishu.adapter._probe_bot_sdk") as mock_sdk:
mock_sdk.return_value = None
result = probe_bot("bad_id", "bad_secret", "feishu")
assert result is None
@patch("gateway.platforms.feishu.FEISHU_AVAILABLE", False)
@patch("gateway.platforms.feishu.urlopen")
@patch("plugins.platforms.feishu.adapter.FEISHU_AVAILABLE", False)
@patch("plugins.platforms.feishu.adapter.urlopen")
def test_http_fallback_when_sdk_unavailable(self, mock_urlopen_fn):
"""Without lark_oapi, probe falls back to raw HTTP."""
from gateway.platforms.feishu import probe_bot
from plugins.platforms.feishu.adapter import probe_bot
token_resp = _mock_urlopen({"code": 0, "tenant_access_token": "t-123"})
bot_resp = _mock_urlopen({"code": 0, "bot": {"bot_name": "HttpBot", "open_id": "ou_http"}})
@ -303,10 +303,10 @@ class TestProbeBot:
assert result is not None
assert result["bot_name"] == "HttpBot"
@patch("gateway.platforms.feishu.FEISHU_AVAILABLE", False)
@patch("gateway.platforms.feishu.urlopen")
@patch("plugins.platforms.feishu.adapter.FEISHU_AVAILABLE", False)
@patch("plugins.platforms.feishu.adapter.urlopen")
def test_http_fallback_returns_none_on_network_error(self, mock_urlopen_fn):
from gateway.platforms.feishu import probe_bot
from plugins.platforms.feishu.adapter import probe_bot
from urllib.error import URLError
mock_urlopen_fn.side_effect = URLError("connection refused")
@ -317,15 +317,15 @@ class TestProbeBot:
class TestQrRegister:
"""Tests for the public qr_register entry point."""
@patch("gateway.platforms.feishu.probe_bot")
@patch("gateway.platforms.feishu._render_qr")
@patch("gateway.platforms.feishu._poll_registration")
@patch("gateway.platforms.feishu._begin_registration")
@patch("gateway.platforms.feishu._init_registration")
@patch("plugins.platforms.feishu.adapter.probe_bot")
@patch("plugins.platforms.feishu.adapter._render_qr")
@patch("plugins.platforms.feishu.adapter._poll_registration")
@patch("plugins.platforms.feishu.adapter._begin_registration")
@patch("plugins.platforms.feishu.adapter._init_registration")
def test_qr_register_success_flow(
self, mock_init, mock_begin, mock_poll, mock_render, mock_probe
):
from gateway.platforms.feishu import qr_register
from plugins.platforms.feishu.adapter import qr_register
mock_begin.return_value = {
"device_code": "dc_123",
@ -350,22 +350,22 @@ class TestQrRegister:
mock_init.assert_called_once()
mock_render.assert_called_once()
@patch("gateway.platforms.feishu._init_registration")
@patch("plugins.platforms.feishu.adapter._init_registration")
def test_qr_register_returns_none_on_init_failure(self, mock_init):
from gateway.platforms.feishu import qr_register
from plugins.platforms.feishu.adapter import qr_register
mock_init.side_effect = RuntimeError("not supported")
result = qr_register()
assert result is None
@patch("gateway.platforms.feishu._render_qr")
@patch("gateway.platforms.feishu._poll_registration")
@patch("gateway.platforms.feishu._begin_registration")
@patch("gateway.platforms.feishu._init_registration")
@patch("plugins.platforms.feishu.adapter._render_qr")
@patch("plugins.platforms.feishu.adapter._poll_registration")
@patch("plugins.platforms.feishu.adapter._begin_registration")
@patch("plugins.platforms.feishu.adapter._init_registration")
def test_qr_register_returns_none_on_poll_failure(
self, mock_init, mock_begin, mock_poll, mock_render
):
from gateway.platforms.feishu import qr_register
from plugins.platforms.feishu.adapter import qr_register
mock_begin.return_value = {
"device_code": "dc_123",
@ -381,29 +381,29 @@ class TestQrRegister:
# -- Contract: expected errors → None, unexpected errors → propagate --
@patch("gateway.platforms.feishu._init_registration")
@patch("plugins.platforms.feishu.adapter._init_registration")
def test_qr_register_returns_none_on_network_error(self, mock_init):
"""URLError (network down) is an expected failure → None."""
from gateway.platforms.feishu import qr_register
from plugins.platforms.feishu.adapter import qr_register
from urllib.error import URLError
mock_init.side_effect = URLError("DNS resolution failed")
result = qr_register()
assert result is None
@patch("gateway.platforms.feishu._init_registration")
@patch("plugins.platforms.feishu.adapter._init_registration")
def test_qr_register_returns_none_on_json_error(self, mock_init):
"""Malformed server response is an expected failure → None."""
from gateway.platforms.feishu import qr_register
from plugins.platforms.feishu.adapter import qr_register
mock_init.side_effect = json.JSONDecodeError("bad json", "", 0)
result = qr_register()
assert result is None
@patch("gateway.platforms.feishu._init_registration")
@patch("plugins.platforms.feishu.adapter._init_registration")
def test_qr_register_propagates_unexpected_errors(self, mock_init):
"""Bugs (e.g. AttributeError) must not be swallowed — they propagate."""
from gateway.platforms.feishu import qr_register
from plugins.platforms.feishu.adapter import qr_register
mock_init.side_effect = AttributeError("some internal bug")
with pytest.raises(AttributeError, match="some internal bug"):
@ -411,29 +411,29 @@ class TestQrRegister:
# -- Negative paths: partial/malformed server responses --
@patch("gateway.platforms.feishu._render_qr")
@patch("gateway.platforms.feishu._begin_registration")
@patch("gateway.platforms.feishu._init_registration")
@patch("plugins.platforms.feishu.adapter._render_qr")
@patch("plugins.platforms.feishu.adapter._begin_registration")
@patch("plugins.platforms.feishu.adapter._init_registration")
def test_qr_register_returns_none_when_begin_missing_device_code(
self, mock_init, mock_begin, mock_render
):
"""Server returns begin response without device_code → RuntimeError → None."""
from gateway.platforms.feishu import qr_register
from plugins.platforms.feishu.adapter import qr_register
mock_begin.side_effect = RuntimeError("Feishu registration did not return a device_code")
result = qr_register()
assert result is None
@patch("gateway.platforms.feishu.probe_bot")
@patch("gateway.platforms.feishu._render_qr")
@patch("gateway.platforms.feishu._poll_registration")
@patch("gateway.platforms.feishu._begin_registration")
@patch("gateway.platforms.feishu._init_registration")
@patch("plugins.platforms.feishu.adapter.probe_bot")
@patch("plugins.platforms.feishu.adapter._render_qr")
@patch("plugins.platforms.feishu.adapter._poll_registration")
@patch("plugins.platforms.feishu.adapter._begin_registration")
@patch("plugins.platforms.feishu.adapter._init_registration")
def test_qr_register_succeeds_even_when_probe_fails(
self, mock_init, mock_begin, mock_poll, mock_render, mock_probe
):
"""Registration succeeds but probe fails → result with bot_name=None."""
from gateway.platforms.feishu import qr_register
from plugins.platforms.feishu.adapter import qr_register
mock_begin.return_value = {
"device_code": "dc_123",

108
tests/gateway/test_matrix.py
View file

@ -341,7 +341,7 @@ class TestMatrixConfigLoading:
def _make_adapter():
"""Create a MatrixAdapter with mocked config."""
from gateway.platforms.matrix import MatrixAdapter
from plugins.platforms.matrix.adapter import MatrixAdapter
config = PlatformConfig(
enabled=True,
token="syt_test_token",
@ -367,7 +367,7 @@ class TestMatrixTypingIndicator:
@pytest.mark.asyncio
async def test_stop_typing_clears_matrix_typing_state(self):
"""stop_typing() should send typing=false instead of waiting for timeout expiry."""
from gateway.platforms.matrix import RoomID
from plugins.platforms.matrix.adapter import RoomID
await self.adapter.stop_typing("!room:example.org")
@ -592,7 +592,7 @@ class TestMatrixBangCommandAlias:
return captured_event
def test_known_bang_command_normalizes_to_slash_command(self):
from gateway.platforms.matrix import _normalize_matrix_bang_command
from plugins.platforms.matrix.adapter import _normalize_matrix_bang_command
assert _normalize_matrix_bang_command("!model") == "/model"
assert (
@ -606,7 +606,7 @@ class TestMatrixBangCommandAlias:
assert _normalize_matrix_bang_command("!tasks") == "/tasks"
def test_unknown_bang_text_is_not_treated_as_command(self):
from gateway.platforms.matrix import _normalize_matrix_bang_command
from plugins.platforms.matrix.adapter import _normalize_matrix_bang_command
assert _normalize_matrix_bang_command("!important note") == "!important note"
assert _normalize_matrix_bang_command("! wow") == "! wow"
@ -666,7 +666,7 @@ class TestMatrixBangCommandAlias:
def test_bang_alias_underscore_resolves_to_hyphen_form(self):
"""!set_home must emit a dispatchable token even though set_home is
not itself registered — the hyphenated alias set-home is."""
from gateway.platforms.matrix import _normalize_matrix_bang_command
from plugins.platforms.matrix.adapter import _normalize_matrix_bang_command
# set_home (underscore) is NOT a registered command/alias, but
# set-home (hyphen) is. The normalizer must emit the resolvable form.
@ -686,7 +686,7 @@ class TestMatrixBangCommandAlias:
with patch.object(
skill_commands_mod, "get_skill_commands", return_value=fake_skills
):
from gateway.platforms.matrix import _normalize_matrix_bang_command
from plugins.platforms.matrix.adapter import _normalize_matrix_bang_command
# is_gateway_known_command won't know these; the skill branch must.
assert _normalize_matrix_bang_command("!arxiv") == "/arxiv"
@ -886,7 +886,7 @@ class TestMatrixDisplayName:
class TestMatrixModuleImport:
def test_module_importable_without_mautrix(self):
"""gateway.platforms.matrix must be importable even when mautrix is
"""plugins.platforms.matrix.adapter must be importable even when mautrix is
not installed — otherwise the gateway crashes for ALL platforms.
This test uses a subprocess to avoid polluting the current process's
@ -908,7 +908,7 @@ class TestMatrixModuleImport:
"for k in list(sys.modules):\n"
" if k.startswith('mautrix'): del sys.modules[k]\n"
"from unittest.mock import patch\n"
"from gateway.platforms.matrix import check_matrix_requirements\n"
"from plugins.platforms.matrix.adapter import check_matrix_requirements\n"
"with patch('tools.lazy_deps.ensure', side_effect=ImportError('blocked')):\n"
" assert not check_matrix_requirements()\n"
"print('OK')\n"
@ -925,7 +925,7 @@ class TestMatrixRequirements:
monkeypatch.setenv("MATRIX_ACCESS_TOKEN", "syt_test")
monkeypatch.setenv("MATRIX_HOMESERVER", "https://matrix.example.org")
monkeypatch.delenv("MATRIX_ENCRYPTION", raising=False)
from gateway.platforms.matrix import check_matrix_requirements
from plugins.platforms.matrix.adapter import check_matrix_requirements
try:
import mautrix # noqa: F401
assert check_matrix_requirements() is True
@ -937,13 +937,13 @@ class TestMatrixRequirements:
monkeypatch.delenv("MATRIX_ACCESS_TOKEN", raising=False)
monkeypatch.delenv("MATRIX_PASSWORD", raising=False)
monkeypatch.delenv("MATRIX_HOMESERVER", raising=False)
from gateway.platforms.matrix import check_matrix_requirements
from plugins.platforms.matrix.adapter import check_matrix_requirements
assert check_matrix_requirements() is False
def test_check_requirements_without_homeserver(self, monkeypatch):
monkeypatch.setenv("MATRIX_ACCESS_TOKEN", "syt_test")
monkeypatch.delenv("MATRIX_HOMESERVER", raising=False)
from gateway.platforms.matrix import check_matrix_requirements
from plugins.platforms.matrix.adapter import check_matrix_requirements
assert check_matrix_requirements() is False
def test_check_requirements_encryption_true_no_e2ee_deps(self, monkeypatch):
@ -952,7 +952,7 @@ class TestMatrixRequirements:
monkeypatch.setenv("MATRIX_HOMESERVER", "https://matrix.example.org")
monkeypatch.setenv("MATRIX_ENCRYPTION", "true")
from gateway.platforms import matrix as matrix_mod
import plugins.platforms.matrix.adapter as matrix_mod
with patch.object(matrix_mod, "_check_e2ee_deps", return_value=False), \
patch("tools.lazy_deps.ensure", side_effect=ImportError("mautrix unavailable")):
assert matrix_mod.check_matrix_requirements() is False
@ -963,7 +963,7 @@ class TestMatrixRequirements:
monkeypatch.setenv("MATRIX_HOMESERVER", "https://matrix.example.org")
monkeypatch.delenv("MATRIX_ENCRYPTION", raising=False)
from gateway.platforms import matrix as matrix_mod
import plugins.platforms.matrix.adapter as matrix_mod
with patch.object(matrix_mod, "_check_e2ee_deps", return_value=False):
# Still needs mautrix itself to be importable
try:
@ -979,7 +979,7 @@ class TestMatrixRequirements:
monkeypatch.setenv("MATRIX_HOMESERVER", "https://matrix.example.org")
monkeypatch.setenv("MATRIX_ENCRYPTION", "true")
from gateway.platforms import matrix as matrix_mod
import plugins.platforms.matrix.adapter as matrix_mod
with patch.object(matrix_mod, "_check_e2ee_deps", return_value=True):
try:
import mautrix # noqa: F401
@ -998,7 +998,7 @@ class TestMatrixRequirements:
a confusing ``No module named 'asyncpg'`` deep in
``MatrixAdapter.connect()``.
"""
from gateway.platforms.matrix import _check_e2ee_deps
from plugins.platforms.matrix.adapter import _check_e2ee_deps
import builtins
real_import = builtins.__import__
@ -1016,7 +1016,7 @@ class TestMatrixRequirements:
Mautrix's ``Database.create("sqlite:///...")`` driver lookup imports
aiosqlite lazily — without it, connect fails at ``crypto_db.start()``.
"""
from gateway.platforms.matrix import _check_e2ee_deps
from plugins.platforms.matrix.adapter import _check_e2ee_deps
import builtins
real_import = builtins.__import__
@ -1040,7 +1040,7 @@ class TestMatrixRequirements:
monkeypatch.setenv("MATRIX_HOMESERVER", "https://matrix.example.org")
monkeypatch.delenv("MATRIX_ENCRYPTION", raising=False)
from gateway.platforms import matrix as matrix_mod
import plugins.platforms.matrix.adapter as matrix_mod
# Simulate "mautrix installed, asyncpg missing" → feature_missing
# returns a non-empty tuple → ensure_and_bind MUST be called.
@ -1070,7 +1070,7 @@ class TestMatrixAccessTokenAuth:
@pytest.mark.asyncio
async def test_connect_with_access_token_and_encryption(self):
"""connect() should call whoami, set user_id/device_id, set up crypto."""
from gateway.platforms.matrix import MatrixAdapter
from plugins.platforms.matrix.adapter import MatrixAdapter
config = PlatformConfig(
enabled=True,
@ -1124,7 +1124,7 @@ class TestMatrixAccessTokenAuth:
fake_mautrix_mods["mautrix.client"].Client = MagicMock(return_value=mock_client)
fake_mautrix_mods["mautrix.crypto"].OlmMachine = MagicMock(return_value=mock_olm)
from gateway.platforms import matrix as matrix_mod
import plugins.platforms.matrix.adapter as matrix_mod
with patch.object(matrix_mod, "_check_e2ee_deps", return_value=True):
with patch.dict("sys.modules", fake_mautrix_mods):
with patch.object(adapter, "_refresh_dm_cache", AsyncMock()):
@ -1176,7 +1176,7 @@ class TestMatrixE2EEHardFail:
@pytest.mark.asyncio
async def test_connect_fails_when_encryption_true_but_no_e2ee_deps(self):
from gateway.platforms.matrix import MatrixAdapter
from plugins.platforms.matrix.adapter import MatrixAdapter
config = PlatformConfig(
enabled=True,
@ -1203,7 +1203,7 @@ class TestMatrixE2EEHardFail:
fake_mautrix_mods["mautrix.client"].Client = MagicMock(return_value=mock_client)
from gateway.platforms import matrix as matrix_mod
import plugins.platforms.matrix.adapter as matrix_mod
with patch.object(matrix_mod, "_check_e2ee_deps", return_value=False):
with patch.dict("sys.modules", fake_mautrix_mods):
with patch.object(adapter, "_sync_loop", AsyncMock(return_value=None)):
@ -1214,7 +1214,7 @@ class TestMatrixE2EEHardFail:
@pytest.mark.asyncio
async def test_connect_fails_when_crypto_setup_raises(self):
"""Even if _check_e2ee_deps passes, if OlmMachine raises, hard-fail."""
from gateway.platforms.matrix import MatrixAdapter
from plugins.platforms.matrix.adapter import MatrixAdapter
config = PlatformConfig(
enabled=True,
@ -1242,7 +1242,7 @@ class TestMatrixE2EEHardFail:
fake_mautrix_mods["mautrix.client"].Client = MagicMock(return_value=mock_client)
fake_mautrix_mods["mautrix.crypto"].OlmMachine = MagicMock(side_effect=Exception("olm init failed"))
from gateway.platforms import matrix as matrix_mod
import plugins.platforms.matrix.adapter as matrix_mod
with patch.object(matrix_mod, "_check_e2ee_deps", return_value=True):
with patch.dict("sys.modules", fake_mautrix_mods):
result = await adapter.connect()
@ -1254,7 +1254,7 @@ class TestMatrixDeviceId:
"""MATRIX_DEVICE_ID should be used for stable device identity."""
def test_device_id_from_config_extra(self):
from gateway.platforms.matrix import MatrixAdapter
from plugins.platforms.matrix.adapter import MatrixAdapter
config = PlatformConfig(
enabled=True,
@ -1270,7 +1270,7 @@ class TestMatrixDeviceId:
def test_device_id_from_env(self, monkeypatch):
monkeypatch.setenv("MATRIX_DEVICE_ID", "FROM_ENV")
from gateway.platforms.matrix import MatrixAdapter
from plugins.platforms.matrix.adapter import MatrixAdapter
config = PlatformConfig(
enabled=True,
@ -1285,7 +1285,7 @@ class TestMatrixDeviceId:
def test_device_id_config_takes_precedence_over_env(self, monkeypatch):
monkeypatch.setenv("MATRIX_DEVICE_ID", "FROM_ENV")
from gateway.platforms.matrix import MatrixAdapter
from plugins.platforms.matrix.adapter import MatrixAdapter
config = PlatformConfig(
enabled=True,
@ -1301,7 +1301,7 @@ class TestMatrixDeviceId:
@pytest.mark.asyncio
async def test_connect_uses_configured_device_id_over_whoami(self):
"""When MATRIX_DEVICE_ID is set, it should be used instead of whoami device_id."""
from gateway.platforms.matrix import MatrixAdapter
from plugins.platforms.matrix.adapter import MatrixAdapter
config = PlatformConfig(
enabled=True,
@ -1348,7 +1348,7 @@ class TestMatrixDeviceId:
fake_mautrix_mods["mautrix.client"].Client = MagicMock(return_value=mock_client)
fake_mautrix_mods["mautrix.crypto"].OlmMachine = MagicMock(return_value=mock_olm)
from gateway.platforms import matrix as matrix_mod
import plugins.platforms.matrix.adapter as matrix_mod
with patch.object(matrix_mod, "_check_e2ee_deps", return_value=True):
with patch.dict("sys.modules", fake_mautrix_mods):
with patch.object(adapter, "_refresh_dm_cache", AsyncMock()):
@ -1367,7 +1367,7 @@ class TestMatrixPasswordLoginDeviceId:
@pytest.mark.asyncio
async def test_password_login_uses_device_id(self):
from gateway.platforms.matrix import MatrixAdapter
from plugins.platforms.matrix.adapter import MatrixAdapter
config = PlatformConfig(
enabled=True,
@ -1591,7 +1591,7 @@ class TestMatrixEncryptedSendFallback:
class TestJoinedRoomsReference:
def test_joined_rooms_reference_preserved_after_reassignment(self):
"""_CryptoStateStore must see updates after initial sync populates rooms."""
from gateway.platforms.matrix import _CryptoStateStore
from plugins.platforms.matrix.adapter import _CryptoStateStore
joined = set()
store = _CryptoStateStore(MagicMock(), joined)
@ -1612,7 +1612,7 @@ class TestJoinedRoomsReference:
class TestMatrixEncryptedEventHandler:
@pytest.mark.asyncio
async def test_connect_registers_encrypted_event_handler_when_encryption_on(self):
from gateway.platforms.matrix import MatrixAdapter
from plugins.platforms.matrix.adapter import MatrixAdapter
config = PlatformConfig(
enabled=True,
@ -1658,7 +1658,7 @@ class TestMatrixEncryptedEventHandler:
fake_mautrix_mods["mautrix.client"].Client = MagicMock(return_value=mock_client)
fake_mautrix_mods["mautrix.crypto"].OlmMachine = MagicMock(return_value=mock_olm)
from gateway.platforms import matrix as matrix_mod
import plugins.platforms.matrix.adapter as matrix_mod
with patch.object(matrix_mod, "_check_e2ee_deps", return_value=True):
with patch.dict("sys.modules", fake_mautrix_mods):
with patch.object(adapter, "_refresh_dm_cache", AsyncMock()):
@ -1678,7 +1678,7 @@ class TestMatrixEncryptedEventHandler:
@pytest.mark.asyncio
async def test_connect_fails_on_stale_otk_conflict(self):
"""connect() must refuse E2EE when OTK upload hits 'already exists'."""
from gateway.platforms.matrix import MatrixAdapter
from plugins.platforms.matrix.adapter import MatrixAdapter
config = PlatformConfig(
enabled=True,
@ -1727,7 +1727,7 @@ class TestMatrixEncryptedEventHandler:
fake_mautrix_mods["mautrix.client"].Client = MagicMock(return_value=mock_client)
fake_mautrix_mods["mautrix.crypto"].OlmMachine = MagicMock(return_value=mock_olm)
from gateway.platforms import matrix as matrix_mod
import plugins.platforms.matrix.adapter as matrix_mod
with patch.object(matrix_mod, "_check_e2ee_deps", return_value=True):
with patch.dict("sys.modules", fake_mautrix_mods):
result = await adapter.connect()
@ -1800,7 +1800,7 @@ class TestMatrixMarkdownHtmlSecurity:
"""Tests for HTML injection prevention in _markdown_to_html_fallback."""
def setup_method(self):
from gateway.platforms.matrix import MatrixAdapter
from plugins.platforms.matrix.adapter import MatrixAdapter
self.convert = MatrixAdapter._markdown_to_html_fallback
def test_script_injection_in_header(self):
@ -1861,7 +1861,7 @@ class TestMatrixMarkdownHtmlFormatting:
"""Tests for new formatting capabilities in _markdown_to_html_fallback."""
def setup_method(self):
from gateway.platforms.matrix import MatrixAdapter
from plugins.platforms.matrix.adapter import MatrixAdapter
self.convert = MatrixAdapter._markdown_to_html_fallback
def test_fenced_code_block(self):
@ -1928,23 +1928,23 @@ class TestMatrixMarkdownHtmlFormatting:
class TestMatrixLinkSanitization:
def test_safe_https_url(self):
from gateway.platforms.matrix import MatrixAdapter
from plugins.platforms.matrix.adapter import MatrixAdapter
assert MatrixAdapter._sanitize_link_url("https://example.com") == "https://example.com"
def test_javascript_blocked(self):
from gateway.platforms.matrix import MatrixAdapter
from plugins.platforms.matrix.adapter import MatrixAdapter
assert MatrixAdapter._sanitize_link_url("javascript:alert(1)") == ""
def test_data_blocked(self):
from gateway.platforms.matrix import MatrixAdapter
from plugins.platforms.matrix.adapter import MatrixAdapter
assert MatrixAdapter._sanitize_link_url("data:text/html,bad") == ""
def test_vbscript_blocked(self):
from gateway.platforms.matrix import MatrixAdapter
from plugins.platforms.matrix.adapter import MatrixAdapter
assert MatrixAdapter._sanitize_link_url("vbscript:bad") == ""
def test_quotes_escaped(self):
from gateway.platforms.matrix import MatrixAdapter
from plugins.platforms.matrix.adapter import MatrixAdapter
result = MatrixAdapter._sanitize_link_url('http://x"y')
assert '"' not in result
assert "&quot;" in result
@ -2561,7 +2561,7 @@ class TestMatrixClockSkewWarning:
# Server events are dated 2h before startup_ts (skewed clock).
skewed_event_ts_ms = int((self.adapter._startup_ts - 7200) * 1000)
with caplog.at_level(logging.WARNING, logger="gateway.platforms.matrix"):
with caplog.at_level(logging.WARNING, logger="plugins.platforms.matrix.adapter"):
for i in range(5):
ev = self._mk_event(
sender=f"@alice{i}:example.org", ts_ms=skewed_event_ts_ms
@ -2575,7 +2575,7 @@ class TestMatrixClockSkewWarning:
# assertion.
skew_warnings = [
r for r in caplog.records
if r.name == "gateway.platforms.matrix"
if r.name == "plugins.platforms.matrix.adapter"
and r.levelname == "WARNING"
and "set-ntp" in r.getMessage()
]
@ -2600,7 +2600,7 @@ class TestMatrixClockSkewWarning:
self.adapter._startup_ts = now - 1
old_ts_ms = int((self.adapter._startup_ts - 3600) * 1000)
with caplog.at_level(logging.WARNING, logger="gateway.platforms.matrix"):
with caplog.at_level(logging.WARNING, logger="plugins.platforms.matrix.adapter"):
for i in range(5):
ev = self._mk_event(
sender=f"@alice{i}:example.org", ts_ms=old_ts_ms
@ -2611,7 +2611,7 @@ class TestMatrixClockSkewWarning:
assert self.adapter._clock_skew_warned is False
skew_warnings = [
r for r in caplog.records
if r.name == "gateway.platforms.matrix"
if r.name == "plugins.platforms.matrix.adapter"
and "set-ntp" in r.getMessage()
]
assert skew_warnings == []
@ -2626,7 +2626,7 @@ class TestMatrixClockSkewWarning:
self.adapter._startup_ts = now - 120 # extra slack vs the 30s gate
old_ts_ms = int((self.adapter._startup_ts - 3600) * 1000)
with caplog.at_level(logging.WARNING, logger="gateway.platforms.matrix"):
with caplog.at_level(logging.WARNING, logger="plugins.platforms.matrix.adapter"):
for i in range(2): # only 2 late drops — under the threshold
ev = self._mk_event(
sender=f"@alice{i}:example.org", ts_ms=old_ts_ms
@ -2652,7 +2652,7 @@ class TestMatrixClockSkewWarning:
self.adapter._startup_ts = now - 120
# Each event has a different age, ranging from 1h to 30d ago.
ages_in_hours = [1, 24, 168, 720, 4] # 1h, 1d, 1w, 30d, 4h
with caplog.at_level(logging.WARNING, logger="gateway.platforms.matrix"):
with caplog.at_level(logging.WARNING, logger="plugins.platforms.matrix.adapter"):
for i, hrs in enumerate(ages_in_hours):
ts_ms = int((self.adapter._startup_ts - hrs * 3600) * 1000)
ev = self._mk_event(
@ -2665,7 +2665,7 @@ class TestMatrixClockSkewWarning:
assert self.adapter._clock_skew_warned is False
skew_warnings = [
r for r in caplog.records
if r.name == "gateway.platforms.matrix"
if r.name == "plugins.platforms.matrix.adapter"
and "set-ntp" in r.getMessage()
]
assert skew_warnings == []
@ -2689,7 +2689,7 @@ class TestMatrixClockSkewWarning:
self.adapter._startup_ts = now - 60
skewed_ms = int((self.adapter._startup_ts - 7200) * 1000)
with caplog.at_level(logging.WARNING, logger="gateway.platforms.matrix"):
with caplog.at_level(logging.WARNING, logger="plugins.platforms.matrix.adapter"):
for i in range(3):
ev = self._mk_event(
sender=f"@alice{i}:example.org", ts_ms=skewed_ms,
@ -2715,7 +2715,7 @@ class TestMatrixClockSkewWarning:
skew_warnings = [
r for r in caplog.records
if r.name == "gateway.platforms.matrix"
if r.name == "plugins.platforms.matrix.adapter"
and "set-ntp" in r.getMessage()
]
assert len(skew_warnings) == 2, (
@ -2792,7 +2792,7 @@ class TestMatrixProxyConfig:
for k, v in proxy_env.items():
monkeypatch.setenv(k, v)
with patch.dict("sys.modules", _make_fake_mautrix()):
from gateway.platforms.matrix import MatrixAdapter
from plugins.platforms.matrix.adapter import MatrixAdapter
cfg = PlatformConfig(enabled=True, token="syt_test",
extra={"homeserver": "https://matrix.example.org",
"user_id": "@bot:example.org"})
@ -2825,7 +2825,7 @@ class TestCreateMatrixSession:
@pytest.mark.asyncio
async def test_no_proxy_returns_trust_env_session(self):
with patch.dict("sys.modules", _make_fake_mautrix()):
from gateway.platforms.matrix import _create_matrix_session
from plugins.platforms.matrix.adapter import _create_matrix_session
session = _create_matrix_session(None)
try:
assert session.trust_env is True
@ -2835,7 +2835,7 @@ class TestCreateMatrixSession:
@pytest.mark.asyncio
async def test_http_proxy_sets_default_proxy(self):
with patch.dict("sys.modules", _make_fake_mautrix()):
from gateway.platforms.matrix import _create_matrix_session
from plugins.platforms.matrix.adapter import _create_matrix_session
session = _create_matrix_session("http://proxy:8080")
try:
assert str(session._default_proxy) == "http://proxy:8080"
@ -2853,7 +2853,7 @@ class TestCreateMatrixSession:
)
),
}):
from gateway.platforms.matrix import _create_matrix_session
from plugins.platforms.matrix.adapter import _create_matrix_session
session = _create_matrix_session("socks5://proxy:1080")
try:
assert session.connector is fake_connector

4
tests/gateway/test_matrix_approval_reaction_fail_closed.py
View file

@ -17,7 +17,7 @@ import pytest
# ---------------------------------------------------------------------------
# Stub mautrix so gateway.platforms.matrix can be imported without the SDK.
# Stub mautrix so plugins.platforms.matrix.adapter can be imported without the SDK.
# ---------------------------------------------------------------------------
def _stub_mautrix():
@ -39,7 +39,7 @@ def _stub_mautrix():
_stub_mautrix()
from gateway.platforms.matrix import MatrixAdapter, _MatrixApprovalPrompt # noqa: E402
from plugins.platforms.matrix.adapter import MatrixAdapter, _MatrixApprovalPrompt # noqa: E402
# ---------------------------------------------------------------------------

4
tests/gateway/test_matrix_exec_approval.py
View file

@ -10,7 +10,7 @@ class TestMatrixExecApprovalReactions:
@pytest.mark.asyncio
async def test_send_exec_approval_registers_prompt_and_seeds_reactions(self, monkeypatch):
monkeypatch.setenv("MATRIX_ALLOWED_USERS", "@liizfq:liizfq.top")
from gateway.platforms.matrix import MatrixAdapter
from plugins.platforms.matrix.adapter import MatrixAdapter
adapter = MatrixAdapter(PlatformConfig(enabled=True, token="tok", extra={"homeserver": "https://matrix.example.org"}))
adapter._client = types.SimpleNamespace()
@ -34,7 +34,7 @@ class TestMatrixExecApprovalReactions:
@pytest.mark.asyncio
async def test_reaction_resolves_pending_approval(self, monkeypatch):
monkeypatch.setenv("MATRIX_ALLOWED_USERS", "@liizfq:liizfq.top")
from gateway.platforms.matrix import MatrixAdapter, _MatrixApprovalPrompt
from plugins.platforms.matrix.adapter import MatrixAdapter, _MatrixApprovalPrompt
adapter = MatrixAdapter(PlatformConfig(enabled=True, token="tok", extra={"homeserver": "https://matrix.example.org"}))
# Resolve user_id so _is_self_sender doesn't defensively drop all traffic (#15763).

2
tests/gateway/test_matrix_mention.py
View file

@ -17,7 +17,7 @@ from gateway.config import PlatformConfig
def _make_adapter(tmp_path=None):
"""Create a MatrixAdapter with mocked config."""
from gateway.platforms.matrix import MatrixAdapter
from plugins.platforms.matrix.adapter import MatrixAdapter
config = PlatformConfig(
enabled=True,

2
tests/gateway/test_matrix_voice.py
View file

@ -27,7 +27,7 @@ from gateway.platforms.base import MessageType
def _make_adapter():
"""Create a MatrixAdapter with mocked config."""
from gateway.platforms.matrix import MatrixAdapter
from plugins.platforms.matrix.adapter import MatrixAdapter
from gateway.config import PlatformConfig
config = PlatformConfig(

4
tests/gateway/test_platform_http_client_limits.py
View file

@ -78,7 +78,7 @@ def test_helper_is_importable_from_every_platform_that_uses_it():
# Just importing exercises the helper's import path for each adapter.
import gateway.platforms.qqbot.adapter # noqa: F401
import gateway.platforms.wecom # noqa: F401
import gateway.platforms.dingtalk # noqa: F401
import plugins.platforms.dingtalk.adapter # noqa: F401
import gateway.platforms.signal # noqa: F401
import gateway.platforms.bluebubbles # noqa: F401
import gateway.platforms.wecom_callback # noqa: F401
@ -98,7 +98,7 @@ class TestWhatsappTypingLeakFix:
def test_bare_await_removed(self):
import inspect
import gateway.platforms.whatsapp as mod
import plugins.platforms.whatsapp.adapter as mod
src = inspect.getsource(mod.WhatsAppAdapter.send_typing)
# The fix must be structural: the post() call is inside an

2
tests/gateway/test_send_image_file.py
View file

@ -82,7 +82,7 @@ def _ensure_telegram_mock():
_ensure_telegram_mock()
from gateway.platforms.telegram import TelegramAdapter # noqa: E402
from plugins.platforms.telegram.adapter import TelegramAdapter # noqa: E402
class TestTelegramSendImageFile:

2
tests/gateway/test_send_multiple_images.py
View file

@ -115,7 +115,7 @@ def _ensure_telegram_mock():
_ensure_telegram_mock()
from gateway.platforms.telegram import TelegramAdapter # noqa: E402
from plugins.platforms.telegram.adapter import TelegramAdapter # noqa: E402
class TestTelegramMultiImage:

36
tests/gateway/test_setup_feishu.py
View file

@ -39,20 +39,20 @@ def _run_setup_feishu(
def mock_get(name):
return existing_env.get(name, "")
with patch("hermes_cli.gateway.save_env_value", side_effect=mock_save), \
patch("hermes_cli.gateway.get_env_value", side_effect=mock_get), \
patch("hermes_cli.gateway.prompt_yes_no", side_effect=prompt_yes_no_responses), \
patch("hermes_cli.gateway.prompt_choice", side_effect=prompt_choice_responses), \
patch("hermes_cli.gateway.prompt", side_effect=prompt_responses), \
patch("hermes_cli.gateway.print_info"), \
patch("hermes_cli.gateway.print_success"), \
patch("hermes_cli.gateway.print_warning"), \
patch("hermes_cli.gateway.print_error"), \
patch("hermes_cli.gateway.color", side_effect=lambda t, c: t), \
patch("gateway.platforms.feishu.qr_register", return_value=qr_result):
with patch("hermes_cli.config.save_env_value", side_effect=mock_save), \
patch("hermes_cli.config.get_env_value", side_effect=mock_get), \
patch("hermes_cli.cli_output.prompt_yes_no", side_effect=prompt_yes_no_responses), \
patch("hermes_cli.setup.prompt_choice", side_effect=prompt_choice_responses), \
patch("hermes_cli.cli_output.prompt", side_effect=prompt_responses), \
patch("hermes_cli.cli_output.print_header"), \
patch("hermes_cli.cli_output.print_info"), \
patch("hermes_cli.cli_output.print_success"), \
patch("hermes_cli.cli_output.print_warning"), \
patch("hermes_cli.cli_output.print_error"), \
patch("plugins.platforms.feishu.adapter.qr_register", return_value=qr_result):
from hermes_cli.gateway import _setup_feishu
_setup_feishu()
from plugins.platforms.feishu.adapter import interactive_setup
interactive_setup()
return saved_env
@ -120,7 +120,7 @@ class TestSetupFeishuConnectionMode:
)
assert env["FEISHU_CONNECTION_MODE"] == "websocket"
@patch("gateway.platforms.feishu.probe_bot", return_value=None)
@patch("plugins.platforms.feishu.adapter.probe_bot", return_value=None)
def test_manual_path_websocket(self, _mock_probe):
env = _run_setup_feishu(
qr_result=None,
@ -129,7 +129,7 @@ class TestSetupFeishuConnectionMode:
)
assert env["FEISHU_CONNECTION_MODE"] == "websocket"
@patch("gateway.platforms.feishu.probe_bot", return_value=None)
@patch("plugins.platforms.feishu.adapter.probe_bot", return_value=None)
def test_manual_path_webhook(self, _mock_probe):
env = _run_setup_feishu(
qr_result=None,
@ -248,7 +248,7 @@ class TestSetupFeishuAdapterIntegration:
with patch.dict(os.environ, env, clear=True):
from gateway.config import PlatformConfig
from gateway.platforms.feishu import FeishuAdapter
from plugins.platforms.feishu.adapter import FeishuAdapter
adapter = FeishuAdapter(PlatformConfig())
assert adapter._app_id == "cli_test_app"
assert adapter._app_secret == "test_secret_value"
@ -261,7 +261,7 @@ class TestSetupFeishuAdapterIntegration:
env = self._make_env_from_setup(dm_idx=1)
with patch.dict(os.environ, env, clear=True):
from gateway.platforms.feishu import FeishuAdapter
from plugins.platforms.feishu.adapter import FeishuAdapter
from gateway.config import PlatformConfig
# Verify adapter initializes without error and env var is correct.
FeishuAdapter(PlatformConfig())
@ -274,6 +274,6 @@ class TestSetupFeishuAdapterIntegration:
with patch.dict(os.environ, env, clear=True):
from gateway.config import PlatformConfig
from gateway.platforms.feishu import FeishuAdapter
from plugins.platforms.feishu.adapter import FeishuAdapter
adapter = FeishuAdapter(PlatformConfig())
assert adapter._group_policy == "open"

10
tests/gateway/test_stream_consumer.py
View file

@ -148,14 +148,14 @@ class TestEditMessageFinalizeSignature:
@pytest.mark.parametrize(
"module_path,class_name",
[
("gateway.platforms.telegram", "TelegramAdapter"),
("plugins.platforms.telegram.adapter", "TelegramAdapter"),
("plugins.platforms.discord.adapter", "DiscordAdapter"),
("gateway.platforms.slack", "SlackAdapter"),
("gateway.platforms.matrix", "MatrixAdapter"),
("plugins.platforms.matrix.adapter", "MatrixAdapter"),
("plugins.platforms.mattermost.adapter", "MattermostAdapter"),
("gateway.platforms.feishu", "FeishuAdapter"),
("gateway.platforms.whatsapp", "WhatsAppAdapter"),
("gateway.platforms.dingtalk", "DingTalkAdapter"),
("plugins.platforms.feishu.adapter", "FeishuAdapter"),
("plugins.platforms.whatsapp.adapter", "WhatsAppAdapter"),
("plugins.platforms.dingtalk.adapter", "DingTalkAdapter"),
],
)
def test_edit_message_accepts_finalize(self, module_path, class_name):

2
tests/gateway/test_stream_consumer_fresh_final.py
View file

@ -391,7 +391,7 @@ class TestTelegramAdapterDeleteMessage:
"""Contract: Telegram adapter implements ``delete_message``."""
def test_delete_message_method_exists(self):
telegram = pytest.importorskip("gateway.platforms.telegram")
telegram = pytest.importorskip("plugins.platforms.telegram.adapter")
import inspect
cls = telegram.TelegramAdapter
assert hasattr(cls, "delete_message"), (

4
tests/gateway/test_stream_consumer_thread_routing.py
View file

@ -143,7 +143,7 @@ class TestFeishuFallbackThreadRouting:
async def test_create_uses_thread_id_when_available(self):
"""When reply_to=None and metadata has thread_id, message.create
should use receive_id_type='thread_id'."""
from gateway.platforms.feishu import FeishuAdapter
from plugins.platforms.feishu.adapter import FeishuAdapter
# We test the _send_raw_message method directly by mocking the client
adapter = MagicMock(spec=FeishuAdapter)
@ -200,7 +200,7 @@ class TestFeishuFallbackThreadRouting:
async def test_create_uses_chat_id_when_no_thread(self):
"""When reply_to=None and metadata has no thread_id, message.create
should use receive_id_type='chat_id' (original behavior)."""
from gateway.platforms.feishu import FeishuAdapter
from plugins.platforms.feishu.adapter import FeishuAdapter
mock_client = MagicMock()
mock_create_response = SimpleNamespace(

2
tests/gateway/test_telegram_approval_buttons.py
View file

@ -46,7 +46,7 @@ def _ensure_telegram_mock():
_ensure_telegram_mock()
from gateway.platforms.telegram import TelegramAdapter
from plugins.platforms.telegram.adapter import TelegramAdapter
from gateway.config import Platform, PlatformConfig

2
tests/gateway/test_telegram_callback_auth_fail_closed.py
View file

@ -55,7 +55,7 @@ def _inject_fake_telegram(monkeypatch):
def _make_adapter():
from gateway.platforms.telegram import TelegramAdapter
from plugins.platforms.telegram.adapter import TelegramAdapter
config = PlatformConfig(enabled=True, token="fake-token")
adapter = object.__new__(TelegramAdapter)

2
tests/gateway/test_telegram_caption_merge.py
View file

@ -1,7 +1,7 @@
"""Tests for TelegramPlatform._merge_caption caption deduplication logic."""
from gateway.platforms.telegram import TelegramAdapter
from plugins.platforms.telegram.adapter import TelegramAdapter
merge = TelegramAdapter._merge_caption

2
tests/gateway/test_telegram_channel_posts.py
View file

@ -63,7 +63,7 @@ def _build_telegram_stubs():
@pytest.fixture
def telegram_adapter_cls(monkeypatch):
"""Import TelegramAdapter without leaking temporary telegram stubs."""
module_name = "gateway.platforms.telegram"
module_name = "plugins.platforms.telegram.adapter"
existing_module = sys.modules.get(module_name)
if existing_module is not None:
yield existing_module.TelegramAdapter

2
tests/gateway/test_telegram_clarify_buttons.py
View file

@ -47,7 +47,7 @@ def _ensure_telegram_mock():
_ensure_telegram_mock()
from gateway.platforms.telegram import TelegramAdapter
from plugins.platforms.telegram.adapter import TelegramAdapter
from gateway.config import PlatformConfig

16
tests/gateway/test_telegram_conflict.py
View file

@ -34,7 +34,7 @@ def _ensure_telegram_mock():
_ensure_telegram_mock()
from gateway.platforms.telegram import TelegramAdapter # noqa: E402
from plugins.platforms.telegram.adapter import TelegramAdapter # noqa: E402
@pytest.fixture(autouse=True)
@ -42,9 +42,9 @@ def _no_auto_discovery(monkeypatch):
"""Disable DoH auto-discovery so connect() uses the plain builder chain."""
async def _noop():
return []
monkeypatch.setattr("gateway.platforms.telegram.discover_fallback_ips", _noop)
monkeypatch.setattr("plugins.platforms.telegram.adapter.discover_fallback_ips", _noop)
# Mock HTTPXRequest so the builder chain doesn't fail
monkeypatch.setattr("gateway.platforms.telegram.HTTPXRequest", lambda **kwargs: MagicMock())
monkeypatch.setattr("plugins.platforms.telegram.adapter.HTTPXRequest", lambda **kwargs: MagicMock())
@pytest.mark.asyncio
@ -103,7 +103,7 @@ async def test_polling_conflict_retries_before_fatal(monkeypatch):
builder.request.return_value = builder
builder.get_updates_request.return_value = builder
builder.build.return_value = app
monkeypatch.setattr("gateway.platforms.telegram.Application", SimpleNamespace(builder=MagicMock(return_value=builder)))
monkeypatch.setattr("plugins.platforms.telegram.adapter.Application", SimpleNamespace(builder=MagicMock(return_value=builder)))
# Speed up retries for testing
monkeypatch.setattr("asyncio.sleep", AsyncMock())
@ -179,7 +179,7 @@ async def test_polling_conflict_becomes_fatal_after_retries(monkeypatch):
builder.request.return_value = builder
builder.get_updates_request.return_value = builder
builder.build.return_value = app
monkeypatch.setattr("gateway.platforms.telegram.Application", SimpleNamespace(builder=MagicMock(return_value=builder)))
monkeypatch.setattr("plugins.platforms.telegram.adapter.Application", SimpleNamespace(builder=MagicMock(return_value=builder)))
# Speed up retries for testing
monkeypatch.setattr("asyncio.sleep", AsyncMock())
@ -232,7 +232,7 @@ async def test_connect_marks_retryable_fatal_error_for_startup_network_failure(m
start=AsyncMock(),
)
builder.build.return_value = app
monkeypatch.setattr("gateway.platforms.telegram.Application", SimpleNamespace(builder=MagicMock(return_value=builder)))
monkeypatch.setattr("plugins.platforms.telegram.adapter.Application", SimpleNamespace(builder=MagicMock(return_value=builder)))
ok = await adapter.connect()
@ -277,7 +277,7 @@ async def test_connect_clears_webhook_before_polling(monkeypatch):
builder.get_updates_request.return_value = builder
builder.build.return_value = app
monkeypatch.setattr(
"gateway.platforms.telegram.Application",
"plugins.platforms.telegram.adapter.Application",
SimpleNamespace(builder=MagicMock(return_value=builder)),
)
@ -301,7 +301,7 @@ async def test_disconnect_skips_inactive_updater_and_app(monkeypatch):
adapter._app = app
warning = MagicMock()
monkeypatch.setattr("gateway.platforms.telegram.logger.warning", warning)
monkeypatch.setattr("plugins.platforms.telegram.adapter.logger.warning", warning)
await adapter.disconnect()

12
tests/gateway/test_telegram_documents.py
View file

@ -51,7 +51,7 @@ def _ensure_telegram_mock():
_ensure_telegram_mock()
# Now we can safely import
from gateway.platforms.telegram import TelegramAdapter # noqa: E402
from plugins.platforms.telegram.adapter import TelegramAdapter # noqa: E402
# ---------------------------------------------------------------------------
@ -442,7 +442,7 @@ class TestMediaGroups:
msg1 = _make_message(caption="two images", photo=[first_photo])
msg2 = _make_message(photo=[second_photo])
with patch("gateway.platforms.telegram.cache_image_from_bytes", side_effect=["/tmp/burst-one.jpg", "/tmp/burst-two.jpg"]):
with patch("plugins.platforms.telegram.adapter.cache_image_from_bytes", side_effect=["/tmp/burst-one.jpg", "/tmp/burst-two.jpg"]):
await adapter._handle_media_message(_make_update(msg1), MagicMock())
await adapter._handle_media_message(_make_update(msg2), MagicMock())
assert adapter.handle_message.await_count == 0
@ -462,7 +462,7 @@ class TestMediaGroups:
msg1 = _make_message(caption="two images", media_group_id="album-1", photo=[first_photo])
msg2 = _make_message(media_group_id="album-1", photo=[second_photo])
with patch("gateway.platforms.telegram.cache_image_from_bytes", side_effect=["/tmp/one.jpg", "/tmp/two.jpg"]):
with patch("plugins.platforms.telegram.adapter.cache_image_from_bytes", side_effect=["/tmp/one.jpg", "/tmp/two.jpg"]):
await adapter._handle_media_message(_make_update(msg1), MagicMock())
await adapter._handle_media_message(_make_update(msg2), MagicMock())
assert adapter.handle_message.await_count == 0
@ -479,7 +479,7 @@ class TestMediaGroups:
first_photo = _make_photo(_make_file_obj(b"first"))
msg = _make_message(caption="two images", media_group_id="album-2", photo=[first_photo])
with patch("gateway.platforms.telegram.cache_image_from_bytes", return_value="/tmp/one.jpg"):
with patch("plugins.platforms.telegram.adapter.cache_image_from_bytes", return_value="/tmp/one.jpg"):
await adapter._handle_media_message(_make_update(msg), MagicMock())
assert "album-2" in adapter._media_group_events
@ -782,8 +782,8 @@ class TestTelegramPhotoBatching:
)
with (
patch("gateway.platforms.telegram.asyncio.current_task", return_value=old_task),
patch("gateway.platforms.telegram.asyncio.sleep", new=AsyncMock()),
patch("plugins.platforms.telegram.adapter.asyncio.current_task", return_value=old_task),
patch("plugins.platforms.telegram.adapter.asyncio.sleep", new=AsyncMock()),
):
await adapter._flush_photo_batch(batch_key)

2
tests/gateway/test_telegram_format.py
View file

@ -35,7 +35,7 @@ def _ensure_telegram_mock():
_ensure_telegram_mock()
from gateway.platforms.telegram import ( # noqa: E402
from plugins.platforms.telegram.adapter import ( # noqa: E402
TelegramAdapter,
_escape_mdv2,
_strip_mdv2,

2
tests/gateway/test_telegram_forum_commands.py
View file

@ -11,7 +11,7 @@ from gateway.config import Platform, PlatformConfig
def _make_test_adapter():
"""Build a TelegramAdapter without running __init__."""
from gateway.platforms.telegram import TelegramAdapter
from plugins.platforms.telegram.adapter import TelegramAdapter
adapter = object.__new__(TelegramAdapter)
adapter.platform = Platform.TELEGRAM

2
tests/gateway/test_telegram_group_gating.py
View file

@ -23,7 +23,7 @@ def _make_adapter(
observe_unmentioned_group_messages=None,
bot_username="hermes_bot",
):
from gateway.platforms.telegram import TelegramAdapter
from plugins.platforms.telegram.adapter import TelegramAdapter
extra = {}
if require_mention is not None:

2
tests/gateway/test_telegram_max_doc_bytes.py
View file

@ -29,7 +29,7 @@ def _ensure_telegram_mock():
_ensure_telegram_mock()
from gateway.platforms.telegram import TelegramAdapter # noqa: E402
from plugins.platforms.telegram.adapter import TelegramAdapter # noqa: E402
def test_max_doc_bytes_defaults_to_20mb_without_base_url():

2
tests/gateway/test_telegram_mention_boundaries.py
View file

@ -14,7 +14,7 @@ those contexts.
from types import SimpleNamespace
from gateway.config import Platform, PlatformConfig
from gateway.platforms.telegram import TelegramAdapter
from plugins.platforms.telegram.adapter import TelegramAdapter
def _make_adapter():

4
tests/gateway/test_telegram_model_picker.py
View file

@ -32,7 +32,7 @@ def _ensure_telegram_mock():
_ensure_telegram_mock()
from gateway.config import PlatformConfig
from gateway.platforms.telegram import TelegramAdapter
from plugins.platforms.telegram.adapter import TelegramAdapter
def _make_adapter():
@ -157,7 +157,7 @@ class TestTelegramModelPicker:
which is robust to whether `telegram` is the real SDK or the module
mock (the SDK markup objects don't expose a plain iterable under the
mock)."""
import gateway.platforms.telegram as tg
import plugins.platforms.telegram.adapter as tg
built: list = []

6
tests/gateway/test_telegram_network.py
View file

@ -1,4 +1,4 @@
"""Tests for gateway.platforms.telegram_network – fallback transport layer.
"""Tests for plugins.platforms.telegram.telegram_network – fallback transport layer.
Background
----------
@ -18,7 +18,7 @@ fallback IPs in order, then "stick" to whichever IP works.
import httpx
import pytest
from gateway.platforms import telegram_network as tnet
import plugins.platforms.telegram.telegram_network as tnet
# ---------------------------------------------------------------------------
@ -438,7 +438,7 @@ class TestAdapterFallbackIps:
sys.modules.setdefault(name, mod)
from gateway.config import PlatformConfig
from gateway.platforms.telegram import TelegramAdapter
from plugins.platforms.telegram.adapter import TelegramAdapter
config = PlatformConfig(enabled=True, token="test-token")
if extra:

6
tests/gateway/test_telegram_network_reconnect.py
View file

@ -33,7 +33,7 @@ def _ensure_telegram_mock():
_ensure_telegram_mock()
from gateway.platforms.telegram import TelegramAdapter # noqa: E402
from plugins.platforms.telegram.adapter import TelegramAdapter # noqa: E402
@pytest.fixture(autouse=True)
@ -41,7 +41,7 @@ def _no_auto_discovery(monkeypatch):
"""Disable DoH auto-discovery so connect() uses the plain builder chain."""
async def _noop():
return []
monkeypatch.setattr("gateway.platforms.telegram.discover_fallback_ips", _noop)
monkeypatch.setattr("plugins.platforms.telegram.adapter.discover_fallback_ips", _noop)
def _make_adapter() -> TelegramAdapter:
@ -379,7 +379,7 @@ async def test_heartbeat_probe_reenters_ladder_when_get_me_times_out():
raise asyncio.TimeoutError()
with patch("asyncio.sleep", new_callable=AsyncMock):
with patch("gateway.platforms.telegram.asyncio.wait_for", new=fast_wait_for):
with patch("plugins.platforms.telegram.adapter.asyncio.wait_for", new=fast_wait_for):
await adapter._verify_polling_after_reconnect()
adapter._handle_polling_network_error.assert_awaited_once()

2
tests/gateway/test_telegram_reactions.py
View file

@ -11,7 +11,7 @@ from gateway.session import SessionSource
def _make_adapter(**extra_env):
from gateway.platforms.telegram import TelegramAdapter
from plugins.platforms.telegram.adapter import TelegramAdapter
adapter = object.__new__(TelegramAdapter)
adapter.platform = Platform.TELEGRAM

2
tests/gateway/test_telegram_reply_mode.py
View file

@ -31,7 +31,7 @@ def _ensure_telegram_mock():
_ensure_telegram_mock()
from gateway.platforms.telegram import TelegramAdapter # noqa: E402
from plugins.platforms.telegram.adapter import TelegramAdapter # noqa: E402
@pytest.fixture()

2
tests/gateway/test_telegram_reply_quote.py
View file

@ -33,7 +33,7 @@ def _ensure_telegram_mock():
_ensure_telegram_mock()
from gateway.platforms.telegram import TelegramAdapter # noqa: E402
from plugins.platforms.telegram.adapter import TelegramAdapter # noqa: E402
def _make_adapter():

4
tests/gateway/test_telegram_send_draft_format.py
View file

@ -35,8 +35,8 @@ def _ensure_telegram_mock():
_ensure_telegram_mock()
from gateway.platforms import telegram as tg_mod # noqa: E402
from gateway.platforms.telegram import TelegramAdapter # noqa: E402
import plugins.platforms.telegram.adapter as tg_mod # noqa: E402
from plugins.platforms.telegram.adapter import TelegramAdapter # noqa: E402
def _make_adapter() -> TelegramAdapter:

6
tests/gateway/test_telegram_send_path_health.py
View file

@ -27,7 +27,7 @@ def _ensure_telegram_mock():
_ensure_telegram_mock()
from gateway.platforms.telegram import TelegramAdapter # noqa: E402
from plugins.platforms.telegram.adapter import TelegramAdapter # noqa: E402
def _make_adapter() -> TelegramAdapter:
@ -78,12 +78,12 @@ async def test_reconnect_storm_sets_and_heartbeat_clears_flag(monkeypatch):
adapter._app.bot.get_me = AsyncMock(return_value=MagicMock())
adapter._polling_error_callback_ref = AsyncMock()
monkeypatch.setattr(
"gateway.platforms.telegram.Update", MagicMock(ALL_TYPES=[])
"plugins.platforms.telegram.adapter.Update", MagicMock(ALL_TYPES=[])
)
await adapter._handle_polling_network_error(OSError("Bad Gateway"))
assert adapter._send_path_degraded is True
with patch("gateway.platforms.telegram.asyncio.sleep", new_callable=AsyncMock):
with patch("plugins.platforms.telegram.adapter.asyncio.sleep", new_callable=AsyncMock):
await adapter._verify_polling_after_reconnect()
assert adapter._send_path_degraded is False

2
tests/gateway/test_telegram_slash_confirm.py
View file

@ -34,7 +34,7 @@ def _ensure_telegram_mock():
_ensure_telegram_mock()
from gateway.platforms.telegram import TelegramAdapter
from plugins.platforms.telegram.adapter import TelegramAdapter
from gateway.config import PlatformConfig

2
tests/gateway/test_telegram_status_update.py
View file

@ -64,7 +64,7 @@ def _install_fake_telegram(monkeypatch):
@pytest.fixture
def adapter(monkeypatch):
_install_fake_telegram(monkeypatch)
from gateway.platforms.telegram import TelegramAdapter
from plugins.platforms.telegram.adapter import TelegramAdapter
a = TelegramAdapter(PlatformConfig(enabled=True, token="fake-token"))
a._bot = MagicMock()

2
tests/gateway/test_telegram_text_batch_perf.py
View file

@ -16,7 +16,7 @@ import math
import pytest
from gateway.platforms.telegram import TelegramAdapter
from plugins.platforms.telegram.adapter import TelegramAdapter
@pytest.fixture

2
tests/gateway/test_telegram_text_batching.py
View file

@ -18,7 +18,7 @@ from gateway.session import build_session_key
def _make_adapter():
"""Create a minimal TelegramAdapter for testing text batching."""
from gateway.platforms.telegram import TelegramAdapter
from plugins.platforms.telegram.adapter import TelegramAdapter
config = PlatformConfig(enabled=True, token="test-token")
adapter = object.__new__(TelegramAdapter)

8
tests/gateway/test_telegram_thread_fallback.py
View file

@ -116,7 +116,7 @@ def _inject_fake_telegram(monkeypatch):
def _make_adapter():
from gateway.platforms.telegram import TelegramAdapter
from plugins.platforms.telegram.adapter import TelegramAdapter
config = PlatformConfig(enabled=True, token="fake-token")
adapter = object.__new__(TelegramAdapter)
@ -137,7 +137,7 @@ def _make_adapter():
def test_non_forum_group_reply_thread_id_does_not_fork_session_key():
"""Reply-derived thread ids in ordinary groups must not create topic lanes."""
from gateway.platforms import telegram as telegram_mod
import plugins.platforms.telegram.adapter as telegram_mod
adapter = _make_adapter()
message = SimpleNamespace(
@ -171,7 +171,7 @@ def test_non_forum_group_reply_thread_id_does_not_fork_session_key():
def test_forum_group_topic_message_preserves_thread_session_key():
"""Real Telegram forum-topic messages should still route by topic id."""
from gateway.platforms import telegram as telegram_mod
import plugins.platforms.telegram.adapter as telegram_mod
adapter = _make_adapter()
message = SimpleNamespace(
@ -201,7 +201,7 @@ def test_forum_group_topic_message_preserves_thread_session_key():
def test_forum_general_topic_without_message_thread_id_keeps_thread_context():
"""Forum General-topic messages should keep synthetic thread context."""
from gateway.platforms import telegram as telegram_mod
import plugins.platforms.telegram.adapter as telegram_mod
adapter = _make_adapter()
message = SimpleNamespace(

6
tests/gateway/test_text_batching.py
View file

@ -218,7 +218,7 @@ class TestDiscordTextBatching:
def _make_matrix_adapter():
"""Create a minimal MatrixAdapter for testing text batching."""
from gateway.platforms.matrix import MatrixAdapter
from plugins.platforms.matrix.adapter import MatrixAdapter
config = PlatformConfig(enabled=True, token="test-token")
adapter = object.__new__(MatrixAdapter)
@ -388,7 +388,7 @@ class TestWeComTextBatching:
def _make_telegram_adapter():
"""Create a minimal TelegramAdapter for testing adaptive delay."""
from gateway.platforms.telegram import TelegramAdapter
from plugins.platforms.telegram.adapter import TelegramAdapter
config = PlatformConfig(enabled=True, token="test-token")
adapter = object.__new__(TelegramAdapter)
@ -452,7 +452,7 @@ class TestTelegramAdaptiveDelay:
def _make_feishu_adapter():
"""Create a minimal FeishuAdapter for testing adaptive delay."""
from gateway.platforms.feishu import FeishuAdapter, FeishuBatchState
from plugins.platforms.feishu.adapter import FeishuAdapter, FeishuBatchState
config = PlatformConfig(enabled=True, token="test-token")
adapter = object.__new__(FeishuAdapter)

62
tests/gateway/test_whatsapp_connect.py
View file

@ -40,7 +40,7 @@ class _AsyncCM:
def _make_adapter():
"""Create a WhatsAppAdapter with test attributes (bypass __init__)."""
from gateway.platforms.whatsapp import WhatsAppAdapter
from plugins.platforms.whatsapp.adapter import WhatsAppAdapter
adapter = WhatsAppAdapter.__new__(WhatsAppAdapter)
adapter.platform = Platform.WHATSAPP
@ -85,18 +85,18 @@ def _mock_aiohttp(status=200, json_data=None, json_side_effect=None):
def _connect_patches(mock_proc, mock_fh, mock_client_cls=None):
"""Return a dict of common patches needed to reach the health-check loop."""
patches = {
"gateway.platforms.whatsapp.check_whatsapp_requirements": True,
"gateway.platforms.whatsapp.asyncio.create_task": MagicMock(),
"plugins.platforms.whatsapp.adapter.check_whatsapp_requirements": True,
"plugins.platforms.whatsapp.adapter.asyncio.create_task": MagicMock(),
}
base = [
patch("gateway.platforms.whatsapp.check_whatsapp_requirements", return_value=True),
patch("plugins.platforms.whatsapp.adapter.check_whatsapp_requirements", return_value=True),
patch.object(Path, "exists", return_value=True),
patch.object(Path, "mkdir", return_value=None),
patch("subprocess.run", return_value=MagicMock(returncode=0)),
patch("subprocess.Popen", return_value=mock_proc),
patch("builtins.open", return_value=mock_fh),
patch("gateway.platforms.whatsapp.asyncio.sleep", new_callable=AsyncMock),
patch("gateway.platforms.whatsapp.asyncio.create_task"),
patch("plugins.platforms.whatsapp.adapter.asyncio.sleep", new_callable=AsyncMock),
patch("plugins.platforms.whatsapp.adapter.asyncio.create_task"),
]
if mock_client_cls is not None:
base.append(patch("aiohttp.ClientSession", mock_client_cls))
@ -112,7 +112,7 @@ class TestCloseBridgeLog:
@staticmethod
def _bare_adapter():
from gateway.platforms.whatsapp import WhatsAppAdapter
from plugins.platforms.whatsapp.adapter import WhatsAppAdapter
a = WhatsAppAdapter.__new__(WhatsAppAdapter)
a._bridge_log_fh = None
return a
@ -223,7 +223,7 @@ class TestConnectCleanup:
install_result = MagicMock(returncode=1, stderr="install failed")
with patch("gateway.platforms.whatsapp.check_whatsapp_requirements", return_value=True), \
with patch("plugins.platforms.whatsapp.adapter.check_whatsapp_requirements", return_value=True), \
patch.object(Path, "exists", autospec=True, side_effect=_path_exists), \
patch("subprocess.run", return_value=install_result), \
patch("gateway.status.acquire_scoped_lock", return_value=(True, None)), \
@ -402,7 +402,7 @@ class TestBridgeRuntimeFailure:
mock_fh = MagicMock()
with patch("gateway.platforms.whatsapp.check_whatsapp_requirements", return_value=True), \
with patch("plugins.platforms.whatsapp.adapter.check_whatsapp_requirements", return_value=True), \
patch.object(Path, "exists", return_value=True), \
patch.object(Path, "mkdir", return_value=None), \
patch("subprocess.run", return_value=MagicMock(returncode=0)), \
@ -423,7 +423,7 @@ class TestKillPortProcess:
"""Verify _kill_port_process uses platform-appropriate commands."""
def test_uses_netstat_and_taskkill_on_windows(self):
from gateway.platforms.whatsapp import _kill_port_process
from plugins.platforms.whatsapp.adapter import _kill_port_process
netstat_output = (
" Proto Local Address Foreign Address State PID\n"
@ -440,8 +440,8 @@ class TestKillPortProcess:
return mock_taskkill
return MagicMock()
with patch("gateway.platforms.whatsapp._IS_WINDOWS", True), \
patch("gateway.platforms.whatsapp.subprocess.run", side_effect=run_side_effect) as mock_run:
with patch("plugins.platforms.whatsapp.adapter._IS_WINDOWS", True), \
patch("plugins.platforms.whatsapp.adapter.subprocess.run", side_effect=run_side_effect) as mock_run:
_kill_port_process(3000)
# netstat called
@ -455,15 +455,15 @@ class TestKillPortProcess:
)
def test_does_not_kill_wrong_port_on_windows(self):
from gateway.platforms.whatsapp import _kill_port_process
from plugins.platforms.whatsapp.adapter import _kill_port_process
netstat_output = (
" TCP 0.0.0.0:30000 0.0.0.0:0 LISTENING 55555\n"
)
mock_netstat = MagicMock(stdout=netstat_output)
with patch("gateway.platforms.whatsapp._IS_WINDOWS", True), \
patch("gateway.platforms.whatsapp.subprocess.run", return_value=mock_netstat) as mock_run:
with patch("plugins.platforms.whatsapp.adapter._IS_WINDOWS", True), \
patch("plugins.platforms.whatsapp.adapter.subprocess.run", return_value=mock_netstat) as mock_run:
_kill_port_process(3000)
# Should NOT call taskkill because port 30000 != 3000
@ -473,12 +473,12 @@ class TestKillPortProcess:
)
def test_uses_fuser_on_linux(self):
from gateway.platforms.whatsapp import _kill_port_process
from plugins.platforms.whatsapp.adapter import _kill_port_process
mock_check = MagicMock(returncode=0)
with patch("gateway.platforms.whatsapp._IS_WINDOWS", False), \
patch("gateway.platforms.whatsapp.subprocess.run", return_value=mock_check) as mock_run:
with patch("plugins.platforms.whatsapp.adapter._IS_WINDOWS", False), \
patch("plugins.platforms.whatsapp.adapter.subprocess.run", return_value=mock_check) as mock_run:
_kill_port_process(3000)
calls = [c.args[0] for c in mock_run.call_args_list]
@ -486,12 +486,12 @@ class TestKillPortProcess:
assert ["fuser", "-k", "3000/tcp"] in calls
def test_skips_fuser_kill_when_port_free(self):
from gateway.platforms.whatsapp import _kill_port_process
from plugins.platforms.whatsapp.adapter import _kill_port_process
mock_check = MagicMock(returncode=1) # port not in use
with patch("gateway.platforms.whatsapp._IS_WINDOWS", False), \
patch("gateway.platforms.whatsapp.subprocess.run", return_value=mock_check) as mock_run:
with patch("plugins.platforms.whatsapp.adapter._IS_WINDOWS", False), \
patch("plugins.platforms.whatsapp.adapter.subprocess.run", return_value=mock_check) as mock_run:
_kill_port_process(3000)
calls = [c.args[0] for c in mock_run.call_args_list]
@ -499,10 +499,10 @@ class TestKillPortProcess:
assert ["fuser", "-k", "3000/tcp"] not in calls
def test_suppresses_exceptions(self):
from gateway.platforms.whatsapp import _kill_port_process
from plugins.platforms.whatsapp.adapter import _kill_port_process
with patch("gateway.platforms.whatsapp._IS_WINDOWS", True), \
patch("gateway.platforms.whatsapp.subprocess.run", side_effect=OSError("no netstat")):
with patch("plugins.platforms.whatsapp.adapter._IS_WINDOWS", True), \
patch("plugins.platforms.whatsapp.adapter.subprocess.run", side_effect=OSError("no netstat")):
_kill_port_process(3000) # must not raise
@ -526,9 +526,9 @@ class TestHttpSessionLifecycle:
adapter._running = True
adapter._session_lock_identity = None
with patch("gateway.platforms.whatsapp._IS_WINDOWS", True), \
patch("gateway.platforms.whatsapp.subprocess.run", return_value=MagicMock(returncode=0)) as mock_run, \
patch("gateway.platforms.whatsapp.asyncio.sleep", new_callable=AsyncMock):
with patch("plugins.platforms.whatsapp.adapter._IS_WINDOWS", True), \
patch("plugins.platforms.whatsapp.adapter.subprocess.run", return_value=MagicMock(returncode=0)) as mock_run, \
patch("plugins.platforms.whatsapp.adapter.asyncio.sleep", new_callable=AsyncMock):
await adapter.disconnect()
mock_run.assert_called_once_with(
@ -634,7 +634,7 @@ class TestNoCredsPreflight:
@pytest.mark.asyncio
async def test_connect_returns_false_when_no_creds(self, tmp_path):
from gateway.platforms.whatsapp import WhatsAppAdapter
from plugins.platforms.whatsapp.adapter import WhatsAppAdapter
adapter = WhatsAppAdapter.__new__(WhatsAppAdapter)
adapter.platform = Platform.WHATSAPP
@ -654,7 +654,7 @@ class TestNoCredsPreflight:
adapter._fatal_error_retryable = True
with patch(
"gateway.platforms.whatsapp.check_whatsapp_requirements",
"plugins.platforms.whatsapp.adapter.check_whatsapp_requirements",
return_value=True,
):
result = await adapter.connect()
@ -670,7 +670,7 @@ class TestNoCredsPreflight:
connect() proceeds to the bridge bootstrap path. We don't fully
simulate the bridge here — we just verify no fast-fail occurs.
"""
from gateway.platforms.whatsapp import WhatsAppAdapter
from plugins.platforms.whatsapp.adapter import WhatsAppAdapter
adapter = WhatsAppAdapter.__new__(WhatsAppAdapter)
adapter.platform = Platform.WHATSAPP
@ -692,7 +692,7 @@ class TestNoCredsPreflight:
adapter._acquire_platform_lock = MagicMock(return_value=False)
with patch(
"gateway.platforms.whatsapp.check_whatsapp_requirements",
"plugins.platforms.whatsapp.adapter.check_whatsapp_requirements",
return_value=True,
):
result = await adapter.connect()

4
tests/gateway/test_whatsapp_formatting.py
View file

@ -20,7 +20,7 @@ from gateway.config import Platform
def _make_adapter():
"""Create a WhatsAppAdapter with test attributes (bypass __init__)."""
from gateway.platforms.whatsapp import WhatsAppAdapter
from plugins.platforms.whatsapp.adapter import WhatsAppAdapter
adapter = WhatsAppAdapter.__new__(WhatsAppAdapter)
adapter.platform = Platform.WHATSAPP
@ -146,7 +146,7 @@ class TestMessageLimits:
"""WhatsApp message length limits."""
def test_max_message_length_is_practical(self):
from gateway.platforms.whatsapp import WhatsAppAdapter
from plugins.platforms.whatsapp.adapter import WhatsAppAdapter
assert WhatsAppAdapter.MAX_MESSAGE_LENGTH == 4096
def test_chunk_limit_reserves_default_self_chat_prefix(self, monkeypatch):

4
tests/gateway/test_whatsapp_group_gating.py
View file

@ -6,7 +6,7 @@ from gateway.config import Platform, PlatformConfig, load_gateway_config
def _make_adapter(require_mention=None, mention_patterns=None, free_response_chats=None,
dm_policy=None, allow_from=None, group_policy=None, group_allow_from=None):
from gateway.platforms.whatsapp import WhatsAppAdapter
from plugins.platforms.whatsapp.adapter import WhatsAppAdapter
extra = {}
if require_mention is not None:
@ -358,7 +358,7 @@ def test_real_dm_still_processed_after_broadcast_filter():
def test_is_broadcast_chat_helper_recognizes_common_jids():
from gateway.platforms.whatsapp import WhatsAppAdapter
from plugins.platforms.whatsapp.adapter import WhatsAppAdapter
assert WhatsAppAdapter._is_broadcast_chat("status@broadcast") is True
assert WhatsAppAdapter._is_broadcast_chat("STATUS@BROADCAST") is True

6
tests/gateway/test_whatsapp_reply_prefix.py
View file

@ -87,19 +87,19 @@ class TestAdapterInit:
"""Test that WhatsAppAdapter reads reply_prefix from config.extra."""
def test_reply_prefix_from_extra(self):
from gateway.platforms.whatsapp import WhatsAppAdapter
from plugins.platforms.whatsapp.adapter import WhatsAppAdapter
config = PlatformConfig(enabled=True, extra={"reply_prefix": "Bot\\n"})
adapter = WhatsAppAdapter(config)
assert adapter._reply_prefix == "Bot\\n"
def test_reply_prefix_default_none(self):
from gateway.platforms.whatsapp import WhatsAppAdapter
from plugins.platforms.whatsapp.adapter import WhatsAppAdapter
config = PlatformConfig(enabled=True)
adapter = WhatsAppAdapter(config)
assert adapter._reply_prefix is None
def test_reply_prefix_empty_string(self):
from gateway.platforms.whatsapp import WhatsAppAdapter
from plugins.platforms.whatsapp.adapter import WhatsAppAdapter
config = PlatformConfig(enabled=True, extra={"reply_prefix": ""})
adapter = WhatsAppAdapter(config)
assert adapter._reply_prefix == ""

2
tests/gateway/test_whatsapp_text_batching.py
View file

@ -12,7 +12,7 @@ import asyncio
from gateway.config import Platform, PlatformConfig
from gateway.platforms.base import MessageEvent, MessageType
from gateway.platforms.whatsapp import WhatsAppAdapter
from plugins.platforms.whatsapp.adapter import WhatsAppAdapter
from gateway.session import SessionSource

6
tests/gateway/test_ws_auth_retry.py
View file

@ -123,7 +123,7 @@ class TestMatrixSyncAuthRetry:
nio_mock.SyncError = SyncError
from gateway.platforms.matrix import MatrixAdapter
from plugins.platforms.matrix.adapter import MatrixAdapter
adapter = MatrixAdapter.__new__(MatrixAdapter)
adapter._closing = False
@ -154,7 +154,7 @@ class TestMatrixSyncAuthRetry:
def test_exception_with_401_stops_loop(self):
"""An exception containing '401' should stop syncing."""
from gateway.platforms.matrix import MatrixAdapter
from plugins.platforms.matrix.adapter import MatrixAdapter
adapter = MatrixAdapter.__new__(MatrixAdapter)
adapter._closing = False
@ -189,7 +189,7 @@ class TestMatrixSyncAuthRetry:
def test_transient_error_retries(self):
"""A transient error should retry (not stop immediately)."""
from gateway.platforms.matrix import MatrixAdapter
from plugins.platforms.matrix.adapter import MatrixAdapter
adapter = MatrixAdapter.__new__(MatrixAdapter)
adapter._closing = False

6
tests/hermes_cli/test_logs.py
View file

@ -87,8 +87,8 @@ class TestExtractLoggerName:
assert _extract_logger_name(line) == "gateway.run"
def test_nested_logger(self):
line = "2026-04-11 10:23:45 INFO gateway.platforms.telegram: connected"
assert _extract_logger_name(line) == "gateway.platforms.telegram"
line = "2026-04-11 10:23:45 INFO plugins.platforms.telegram.adapter: connected"
assert _extract_logger_name(line) == "plugins.platforms.telegram.adapter"
def test_warning_level(self):
line = "2026-04-11 10:23:45 WARNING tools.terminal_tool: timeout"
@ -116,7 +116,7 @@ class TestLineMatchesComponent:
assert _line_matches_component(line, ("gateway",))
def test_gateway_nested(self):
line = "2026-04-11 10:23:45 INFO gateway.platforms.telegram: msg"
line = "2026-04-11 10:23:45 INFO plugins.platforms.telegram.adapter: msg"
assert _line_matches_component(line, ("gateway",))
def test_tools_component(self):

4
tests/test_hermes_logging.py
View file

@ -305,7 +305,7 @@ class TestGatewayMode:
"""gateway.log captures records from gateway.* loggers."""
hermes_logging.setup_logging(hermes_home=hermes_home, mode="gateway")
gw_logger = logging.getLogger("gateway.platforms.telegram")
gw_logger = logging.getLogger("plugins.platforms.telegram.adapter")
gw_logger.info("telegram connected")
for h in logging.getLogger().handlers:
@ -554,7 +554,7 @@ class TestComponentFilter:
def test_passes_nested_matching_prefix(self):
f = hermes_logging._ComponentFilter(("gateway",))
record = logging.LogRecord(
"gateway.platforms.telegram", logging.INFO, "", 0, "msg", (), None
"plugins.platforms.telegram.adapter", logging.INFO, "", 0, "msg", (), None
)
assert f.filter(record) is True

2
tests/tools/test_send_message_tool.py
View file

@ -817,7 +817,7 @@ class TestSendToPlatformChunking:
fake_module = SimpleNamespace(MatrixAdapter=FakeAdapter)
with patch.dict(sys.modules, {"gateway.platforms.matrix": fake_module}):
with patch.dict(sys.modules, {"plugins.platforms.matrix.adapter": fake_module}):
result = asyncio.run(
_send_matrix_via_adapter(
SimpleNamespace(enabled=True, token="tok", extra={"homeserver": "https://matrix.example.com"}),