"""Tests for the WhatsApp stale-bridge staleness handshake. Regression tests for the stale-bridge trap: ``connect()`` reused any already-running bridge with ``status: connected`` unconditionally, and ``disconnect()`` only kills bridges the adapter spawned itself. A long-lived bridge process therefore survived gateway restarts AND ``hermes update``, serving pre-update bridge.js behavior forever (e.g. no inbound media download → images/voice notes arrive as placeholders). The fix: bridge.js reports a hash of its own source in ``/health`` (``scriptHash``); the adapter compares it against the bridge.js on disk and restarts the bridge on mismatch. Bridges that predate the handshake report no hash and are treated as stale by definition. Also covers the npm dependency-refresh stamp: deps are reinstalled when package.json changes, not only when node_modules is missing. """ import asyncio import os from pathlib import Path from unittest.mock import AsyncMock, MagicMock, patch import pytest from gateway.config import Platform class _AsyncCM: """Minimal async context manager returning a fixed value.""" def __init__(self, value): self.value = value async def __aenter__(self): return self.value async def __aexit__(self, *exc): return False @pytest.fixture(autouse=True) def _pm_node(monkeypatch): """Stand-in for PM's Node/npm; the user's PATH copy is never picked up.""" from plugins.platforms.whatsapp import adapter as whatsapp_adapter monkeypatch.setattr(whatsapp_adapter, "find_node_executable", lambda name: f"/pm/{name}") def _make_adapter(bridge_script: str = "/tmp/test-bridge.js", session_path: Path = Path("/tmp/test-wa-session")): """Create a WhatsAppAdapter with test attributes (bypass __init__).""" from plugins.platforms.whatsapp.adapter import WhatsAppAdapter adapter = WhatsAppAdapter.__new__(WhatsAppAdapter) adapter.platform = Platform.WHATSAPP adapter.config = MagicMock() adapter._bridge_port = 19876 adapter._bridge_script = bridge_script adapter._session_path = Path(os.path.abspath(os.path.expanduser(str(session_path)))) adapter._foreign_bridge_session = None # mirror __init__; harness builds via __new__ adapter._bridge_probe_timed_out = False adapter._bridge_log_fh = None adapter._bridge_log = None adapter._bridge_process = None adapter._reply_prefix = None adapter._send_read_receipts = False adapter._dm_policy = adapter._group_policy = "pairing" adapter._allow_from = adapter._group_allow_from = set() adapter._running = False adapter._message_handler = None adapter._fatal_error_code = None adapter._fatal_error_message = None adapter._fatal_error_retryable = True adapter._fatal_error_handler = None adapter._active_sessions = {} adapter._pending_messages = {} adapter._background_tasks = set() adapter._auto_tts_disabled_chats = set() adapter._message_queue = asyncio.Queue() adapter._http_session = None return adapter def _mock_health(json_data, raise_on=None): """Mock aiohttp.ClientSession whose GET returns 200 + *json_data*; ``raise_on`` "headers"/"body" times out that phase.""" mock_resp = MagicMock() mock_resp.status = 200 mock_resp.json = AsyncMock(return_value=json_data, side_effect=asyncio.TimeoutError if raise_on == "body" else None) mock_session = MagicMock() mock_session.get = MagicMock( return_value=_AsyncCM(mock_resp), side_effect=asyncio.TimeoutError if raise_on == "headers" else None) mock_session.close = AsyncMock() return MagicMock(return_value=_AsyncCM(mock_session)) def _setup_bridge_dir(tmp_path: Path) -> Path: """Create a real bridge dir with bridge.js + package.json + creds.""" bridge_dir = tmp_path / "whatsapp-bridge" bridge_dir.mkdir() (bridge_dir / "bridge.js").write_text("// current bridge code\n", encoding="utf-8") (bridge_dir / "package.json").write_text('{"name": "bridge"}\n', encoding="utf-8") session_path = tmp_path / "session" session_path.mkdir() (session_path / "creds.json").write_text("{}", encoding="utf-8") return bridge_dir def _fresh_node_modules(bridge_dir: Path) -> None: """Create node_modules with a stamp matching the current package.json.""" from plugins.platforms.whatsapp.adapter import _file_content_hash nm = bridge_dir / "node_modules" nm.mkdir() (nm / ".hermes-pkg-hash").write_text( _file_content_hash(bridge_dir / "package.json") ) class TestStaleBridgeHandshake: @pytest.mark.asyncio async def test_restarts_bridge_when_read_receipt_config_changed(self, tmp_path): from plugins.platforms.whatsapp.adapter import _file_content_hash bridge_dir = _setup_bridge_dir(tmp_path) _fresh_node_modules(bridge_dir) adapter = _make_adapter( bridge_script=str(bridge_dir / "bridge.js"), session_path=tmp_path / "session", ) adapter._send_read_receipts = True disk_hash = _file_content_hash(bridge_dir / "bridge.js") mock_client = _mock_health( { "status": "connected", "scriptHash": disk_hash, "sendReadReceipts": False, "session": str(tmp_path / "session"), } ) mock_proc = MagicMock() mock_proc.poll.return_value = 1 mock_proc.returncode = 1 with patch("plugins.platforms.whatsapp.adapter.check_whatsapp_requirements", return_value=True), \ patch("aiohttp.ClientSession", mock_client), \ patch("plugins.platforms.whatsapp.adapter.asyncio.sleep", new_callable=AsyncMock), \ patch("plugins.platforms.whatsapp.adapter._kill_stale_bridge_by_pidfile"), \ patch("plugins.platforms.whatsapp.adapter._kill_port_process"), \ patch("subprocess.Popen", return_value=mock_proc) as mock_popen, \ patch.object(adapter, "_acquire_platform_lock", return_value=True, create=True): await adapter.connect() mock_popen.assert_called_once() @pytest.mark.asyncio @pytest.mark.parametrize("holder,status,fatal", [ ("own", "connected", None), ("other", "connected", "whatsapp_bridge_foreign_session"), ("other", "disconnected", "whatsapp_bridge_foreign_session"), ("headers-timeout", None, "whatsapp_bridge_unresponsive"), ("body-timeout", None, "whatsapp_bridge_unresponsive"), ("body-timeout-windows", None, "whatsapp_bridge_unresponsive"), ], ids=["own-session-adopted", "other-session-connected", "other-session-starting", "health-headers-timeout", "health-body-timeout", "health-body-timeout-windows"]) async def test_bridge_is_adopted_or_left_running_by_its_session(self, tmp_path, holder, status, fatal): """Two profiles default to one bridge_port; only this profile's own session may be adopted. Another profile's bridge is never killed, even while it reports ``disconnected`` (startup, reconnect, QR wait), and neither is a port holder that gives no /health answer, which proves no ownership.""" from plugins.platforms.whatsapp.adapter import _file_content_hash bridge_dir = _setup_bridge_dir(tmp_path) _fresh_node_modules(bridge_dir) adapter = _make_adapter( bridge_script=str(bridge_dir / "bridge.js"), session_path=tmp_path / "session", ) reported = tmp_path / ("session" if holder == "own" else "other-profile-session") mock_client = _mock_health( { "status": status, "scriptHash": _file_content_hash(bridge_dir / "bridge.js"), "sendReadReceipts": False, "session": str(reported), }, raise_on=holder.split("-")[0] if "timeout" in holder else None, ) windows = holder.endswith("-windows") # the listener scan must be the platform's own, or Windows still kills with ( patch("plugins.platforms.whatsapp.adapter.check_whatsapp_requirements", return_value=True), patch("aiohttp.ClientSession", mock_client), patch("plugins.platforms.whatsapp.adapter.asyncio.sleep", new_callable=AsyncMock), patch("plugins.platforms.whatsapp.adapter._kill_stale_bridge_by_pidfile") as mock_kill_pidfile, patch("plugins.platforms.whatsapp.adapter._kill_port_process") as mock_kill_port, patch("plugins.platforms.whatsapp.adapter._IS_WINDOWS", windows), patch("plugins.platforms.whatsapp.adapter._listener_pids_on_port", return_value=[] if windows else [4242]), patch("plugins.platforms.whatsapp.adapter._windows_listener_pids", return_value=[4242] if windows else []), patch("subprocess.Popen", return_value=MagicMock()) as mock_popen, patch.object(adapter, "_acquire_platform_lock", return_value=True, create=True), ): assert await adapter.connect() is (fatal is None) mock_popen.assert_not_called() mock_kill_port.assert_not_called() # Reaping by this profile's own pidfile is identity-based; only a foreign verdict skips it. assert mock_kill_pidfile.called is (fatal == "whatsapp_bridge_unresponsive") if fatal: assert adapter._fatal_error_code == fatal assert adapter._fatal_error_retryable is (fatal == "whatsapp_bridge_unresponsive") assert f"nothing on port {adapter._bridge_port} was" in adapter._fatal_error_message assert "bridge_port" in adapter._fatal_error_message if fatal == "whatsapp_bridge_foreign_session": assert str(reported) in adapter._fatal_error_message class TestDepRefreshStamp: @pytest.mark.asyncio async def test_skips_install_when_stamp_fresh(self, tmp_path): bridge_dir = _setup_bridge_dir(tmp_path) _fresh_node_modules(bridge_dir) adapter = _make_adapter( bridge_script=str(bridge_dir / "bridge.js"), session_path=tmp_path / "session", ) mock_proc = MagicMock() mock_proc.poll.return_value = 1 mock_proc.returncode = 1 with patch("plugins.platforms.whatsapp.adapter.check_whatsapp_requirements", return_value=True), \ patch("aiohttp.ClientSession", _mock_health({"status": "disconnected"})), \ patch("plugins.platforms.whatsapp.adapter.asyncio.sleep", new_callable=AsyncMock), \ patch("plugins.platforms.whatsapp.adapter._kill_stale_bridge_by_pidfile"), \ patch("plugins.platforms.whatsapp.adapter._kill_port_process"), \ patch("subprocess.run") as mock_run, \ patch("subprocess.Popen", return_value=mock_proc), \ patch.object(adapter, "_acquire_platform_lock", return_value=True, create=True): await adapter.connect() mock_run.assert_not_called() class TestCacheDirEnvPassthrough: @pytest.mark.asyncio async def test_bridge_spawn_env_has_cache_dirs(self, tmp_path): bridge_dir = _setup_bridge_dir(tmp_path) _fresh_node_modules(bridge_dir) adapter = _make_adapter( bridge_script=str(bridge_dir / "bridge.js"), session_path=tmp_path / "session", ) adapter._send_read_receipts = True mock_proc = MagicMock() mock_proc.poll.return_value = 1 mock_proc.returncode = 1 with patch("plugins.platforms.whatsapp.adapter.check_whatsapp_requirements", return_value=True), \ patch("aiohttp.ClientSession", _mock_health({"status": "disconnected"})), \ patch("plugins.platforms.whatsapp.adapter.asyncio.sleep", new_callable=AsyncMock), \ patch("plugins.platforms.whatsapp.adapter._kill_stale_bridge_by_pidfile"), \ patch("plugins.platforms.whatsapp.adapter._kill_port_process"), \ patch("subprocess.Popen", return_value=mock_proc) as mock_popen, \ patch.object(adapter, "_acquire_platform_lock", return_value=True, create=True): await adapter.connect() env = mock_popen.call_args.kwargs["env"] from gateway.platforms.base import ( get_audio_cache_dir, get_document_cache_dir, get_image_cache_dir, ) assert env["HERMES_IMAGE_CACHE_DIR"] == str(get_image_cache_dir()) assert env["HERMES_AUDIO_CACHE_DIR"] == str(get_audio_cache_dir()) assert env["HERMES_DOCUMENT_CACHE_DIR"] == str(get_document_cache_dir()) assert env["WHATSAPP_SEND_READ_RECEIPTS"] == "true"