"""Tests for the SimpleX Chat platform-plugin adapter. Loaded via the ``_plugin_adapter_loader`` helper so this lives under ``plugin_adapter_simplex`` in ``sys.modules`` and cannot collide with sibling platform-plugin tests on the same xdist worker. """ from __future__ import annotations import asyncio import json from unittest.mock import AsyncMock, MagicMock import pytest from tests.gateway._plugin_adapter_loader import load_plugin_adapter _simplex = load_plugin_adapter("simplex") SimplexAdapter = _simplex.SimplexAdapter check_requirements = _simplex.check_requirements validate_config = _simplex.validate_config is_connected = _simplex.is_connected register = _simplex.register _env_enablement = _simplex._env_enablement _standalone_send = _simplex._standalone_send _guess_extension = _simplex._guess_extension _is_image_ext = _simplex._is_image_ext _is_audio_ext = _simplex._is_audio_ext _CORR_PREFIX = _simplex._CORR_PREFIX # --------------------------------------------------------------------------- # 1. Platform enum (plugin-discovered, not bundled) # --------------------------------------------------------------------------- def test_platform_enum_resolves_via_plugin_scan(): """The plugin filesystem scan should expose Platform("simplex").""" from gateway.config import Platform p = Platform("simplex") assert p.value == "simplex" # Identity stability — repeated lookups return the same pseudo-member assert Platform("simplex") is p # --------------------------------------------------------------------------- # 2. check_requirements / validate_config / is_connected # --------------------------------------------------------------------------- def test_check_requirements_true_when_configured(monkeypatch): monkeypatch.setenv("SIMPLEX_WS_URL", "ws://127.0.0.1:5225") # websockets is a dev dep in this repo via the test plugins; the # check_requirements() gate also asserts the package imports. websockets_present = True try: import websockets # noqa: F401 except ImportError: websockets_present = False assert check_requirements() is websockets_present def test_validate_config_uses_env_or_extra(): from gateway.config import PlatformConfig # Empty extra + no env → invalid cfg = PlatformConfig(enabled=True) assert validate_config(cfg) is False # extra-only path → valid cfg2 = PlatformConfig(enabled=True, extra={"ws_url": "ws://localhost:5225"}) assert validate_config(cfg2) is True def test_is_connected_mirrors_validate(monkeypatch): from gateway.config import PlatformConfig monkeypatch.delenv("SIMPLEX_WS_URL", raising=False) cfg = PlatformConfig(enabled=True, extra={"ws_url": "ws://x"}) assert is_connected(cfg) is True assert is_connected(PlatformConfig(enabled=True)) is False # --------------------------------------------------------------------------- # 3. _env_enablement seeds PlatformConfig.extra # --------------------------------------------------------------------------- def test_env_enablement_seeds_home_channel(monkeypatch): monkeypatch.setenv("SIMPLEX_WS_URL", "ws://127.0.0.1:5225") monkeypatch.setenv("SIMPLEX_HOME_CHANNEL", "42") monkeypatch.setenv("SIMPLEX_HOME_CHANNEL_NAME", "Personal") seed = _env_enablement() assert seed["home_channel"] == {"chat_id": "42", "name": "Personal"} # --------------------------------------------------------------------------- # 4. Adapter init # --------------------------------------------------------------------------- def test_adapter_init_custom_url(): from gateway.config import PlatformConfig cfg = PlatformConfig(enabled=True, extra={"ws_url": "ws://localhost:5225"}) adapter = SimplexAdapter(cfg) assert adapter.ws_url == "ws://localhost:5225" assert adapter._running is False assert adapter._ws is None # --------------------------------------------------------------------------- # 5. Helper functions (magic-byte detection) # --------------------------------------------------------------------------- def test_guess_extension_png(): assert _guess_extension(b"\x89PNG\r\n\x1a\n") == ".png" # --------------------------------------------------------------------------- # 6. Correlation IDs # --------------------------------------------------------------------------- def test_corr_id_pending_set_self_trims(): from gateway.config import PlatformConfig cfg = PlatformConfig(enabled=True, extra={"ws_url": "ws://localhost:5225"}) adapter = SimplexAdapter(cfg) adapter._max_pending_corr = 4 for _ in range(10): adapter._make_corr_id() # After many additions, the pending set should be bounded by the trim # logic — at most one trim window above the cap. assert len(adapter._pending_corr_ids) <= adapter._max_pending_corr + 1 # --------------------------------------------------------------------------- # 7. Outbound send (mocked WS) # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_send_group(): """Groups use the structured ``/_send # json [...]`` form. The bracket chat-command form ``#[] text`` *looks* like an exact ID match in the daemon docs but is parsed as a display-name lookup — so messages to groups whose display name isn't literally the ID silently drop. The structured ``/_send`` form addresses by numeric ID and survives newlines/quoting through ``json.dumps``. """ from gateway.config import PlatformConfig cfg = PlatformConfig(enabled=True, extra={"ws_url": "ws://localhost:5225"}) adapter = SimplexAdapter(cfg) mock_ws = AsyncMock() adapter._ws = mock_ws result = await adapter.send("group:grp-99", "Hello, group!") payload = json.loads(mock_ws.send.call_args[0][0]) assert payload["cmd"].startswith("/_send #grp-99 json ") msg_content = json.loads(payload["cmd"].split(" json ", 1)[1])[0][ "msgContent" ] assert msg_content == {"type": "text", "text": "Hello, group!"} assert result.success is True # --------------------------------------------------------------------------- # 7b. Channel directory enumeration (list_channels) # --------------------------------------------------------------------------- def _adapter_with_ws(): from gateway.config import PlatformConfig cfg = PlatformConfig(enabled=True, extra={"ws_url": "ws://localhost:5225"}) adapter = SimplexAdapter(cfg) adapter._ws = AsyncMock() return adapter @pytest.mark.asyncio async def test_list_channels_contacts_and_groups(): adapter = _adapter_with_ws() async def fake_send_command(command, timeout=30.0): if command == "/contacts": return { "contacts": [ {"contactId": 1, "localDisplayName": "alice"}, {"contactId": 2, "profile": {"displayName": "bob"}}, "garbage", ] } if command == "/groups": return { "groups": [ {"groupId": 7, "localDisplayName": "friends"}, # [groupInfo, groupSummary] pair form [{"groupId": 9, "groupProfile": {"displayName": "work"}}, {}], ] } return None adapter._send_command = fake_send_command channels = await adapter.list_channels() assert {"id": "alice", "name": "alice", "type": "dm"} in channels assert {"id": "bob", "name": "bob", "type": "dm"} in channels assert {"id": "group:7", "name": "friends", "type": "group"} in channels assert {"id": "group:9", "name": "work", "type": "group"} in channels @pytest.mark.asyncio async def test_list_channels_returns_none_when_disconnected(): """None (not []) so the directory falls back to session discovery.""" from gateway.config import PlatformConfig cfg = PlatformConfig(enabled=True, extra={"ws_url": "ws://localhost:5225"}) adapter = SimplexAdapter(cfg) assert adapter._ws is None assert await adapter.list_channels() is None @pytest.mark.asyncio async def test_list_channels_returns_none_on_contacts_timeout(): adapter = _adapter_with_ws() async def fake_send_command(command, timeout=30.0): return None # daemon unresponsive adapter._send_command = fake_send_command assert await adapter.list_channels() is None # --------------------------------------------------------------------------- # 8. Inbound: filter own-echo by corrId prefix # --------------------------------------------------------------------------- # --------------------------------------------------------------------------- # 9. Standalone (out-of-process) send for cron # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_standalone_send_missing_websockets(monkeypatch): """When websockets is unimportable, return a clean error dict. Implementation detail: the standalone path does ``import websockets`` inside the function body. We simulate the package being absent by pulling it out of ``sys.modules`` and pointing the finder at None. """ import sys saved_websockets = sys.modules.pop("websockets", None) saved_meta = list(sys.meta_path) class _Blocker: @staticmethod def find_spec(name, path=None, target=None): if name == "websockets" or name.startswith("websockets."): raise ImportError("websockets blocked for test") return None sys.meta_path.insert(0, _Blocker()) try: pconfig = MagicMock() pconfig.extra = {"ws_url": "ws://localhost:5225"} result = await _standalone_send(pconfig, "contact-42", "hi") assert isinstance(result, dict) assert "error" in result assert "websockets" in result["error"] finally: sys.meta_path[:] = saved_meta if saved_websockets is not None: sys.modules["websockets"] = saved_websockets # --------------------------------------------------------------------------- # 10. register() — plugin-side metadata # --------------------------------------------------------------------------- # --------------------------------------------------------------------------- # Inbound attachment message type classification # --------------------------------------------------------------------------- def _make_file_chat_item(file_path: str, file_name: str) -> dict: """Minimal direct-chat rcvMsgContent item carrying a completed file.""" return { "chatInfo": { "type": "direct", "contact": {"contactId": 42, "localDisplayName": "tester"}, }, "chatItem": { "chatDir": {"type": "directRcv"}, "meta": {"itemTs": "2026-01-01T00:00:00Z"}, "content": { "type": "rcvMsgContent", "msgContent": {"type": "file", "text": "here you go"}, }, "file": { "fileId": 7, "fileName": file_name, "fileSource": {"filePath": file_path}, }, }, }