diff --git a/plugins/platforms/mattermost/adapter.py b/plugins/platforms/mattermost/adapter.py index bb6dc9b81f24..3cc1593fe991 100644 --- a/plugins/platforms/mattermost/adapter.py +++ b/plugins/platforms/mattermost/adapter.py @@ -50,6 +50,24 @@ _RECONNECT_JITTER = 0.2 +class _MMApiError(Exception): + """Internal Mattermost API failure carrying transient/permanent class. + + ``retryable`` is True for network failures, request timeouts, HTTP 429 + rate-limits, and 5xx server errors — conditions that may succeed on a + retry. It is False for genuine 4xx client errors (400/401/403/404), + which will not. ``send()`` / ``edit_message()`` translate this into + ``SendResult.retryable`` so ``BasePlatformAdapter._send_with_retry`` + (network retry with backoff) and the streaming progress-edit consumer + (which keeps ``can_edit`` alive across transient edit failures) make the + right call instead of treating every failure as permanent. + """ + + def __init__(self, message: str, *, retryable: bool): + super().__init__(message) + self.retryable = retryable + + def check_mattermost_requirements() -> bool: """Return True if the Mattermost adapter can be used.""" token = os.getenv("MATTERMOST_TOKEN", "") @@ -109,58 +127,74 @@ def _headers(self) -> Dict[str, str]: "Content-Type": "application/json", } - async def _api_get(self, path: str) -> Dict[str, Any]: - """GET /api/v4/{path}.""" + async def _request_json( + self, method: str, path: str, *, payload: Optional[Dict[str, Any]] = None + ) -> Dict[str, Any]: + """Issue an /api/v4 request, raising :class:`_MMApiError` on failure. + + Distinguishes transient failures (network errors, timeouts, HTTP 429 + and 5xx) from permanent 4xx client errors so callers that surface a + ``SendResult`` can set ``retryable`` correctly. Callers that want the + legacy empty-dict sentinel use :meth:`_api_get` / :meth:`_api_post` / + :meth:`_api_put`, which wrap this. + """ import aiohttp url = f"{self._base_url}/api/v4/{path.lstrip('/')}" + kwargs: Dict[str, Any] = { + "headers": self._headers(), + "timeout": aiohttp.ClientTimeout(total=30), + } + if payload is not None: + kwargs["json"] = payload + verb = method.upper() + sender = { + "GET": self._session.get, + "POST": self._session.post, + "PUT": self._session.put, + }[verb] try: - async with self._session.get(url, headers=self._headers(), timeout=aiohttp.ClientTimeout(total=30)) as resp: + async with sender(url, **kwargs) as resp: if resp.status >= 400: body = await resp.text() - logger.error("MM API GET %s → %s: %s", path, resp.status, body[:200]) - return {} + logger.error("MM API %s %s → %s: %s", verb, path, resp.status, body[:200]) + # 429 (rate-limit) and 5xx (server) are transient; retrying + # may succeed. 4xx client errors are permanent. + retryable = resp.status == 429 or resp.status >= 500 + raise _MMApiError( + f"Mattermost API {verb} {path} returned HTTP {resp.status}", + retryable=retryable, + ) return await resp.json() - except aiohttp.ClientError as exc: - logger.error("MM API GET %s network error: %s", path, exc) + except (aiohttp.ClientError, asyncio.TimeoutError) as exc: + logger.error("MM API %s %s network error: %s", verb, path, exc) + raise _MMApiError( + f"Mattermost API {verb} {path} network error: {exc}", + retryable=True, + ) from exc + + async def _api_get(self, path: str) -> Dict[str, Any]: + """GET /api/v4/{path}; returns {} on any failure (legacy sentinel).""" + try: + return await self._request_json("GET", path) + except _MMApiError: return {} async def _api_post( self, path: str, payload: Dict[str, Any] ) -> Dict[str, Any]: - """POST /api/v4/{path} with JSON body.""" - import aiohttp - url = f"{self._base_url}/api/v4/{path.lstrip('/')}" + """POST /api/v4/{path}; returns {} on any failure (legacy sentinel).""" try: - async with self._session.post( - url, headers=self._headers(), json=payload, - timeout=aiohttp.ClientTimeout(total=30) - ) as resp: - if resp.status >= 400: - body = await resp.text() - logger.error("MM API POST %s → %s: %s", path, resp.status, body[:200]) - return {} - return await resp.json() - except aiohttp.ClientError as exc: - logger.error("MM API POST %s network error: %s", path, exc) + return await self._request_json("POST", path, payload=payload) + except _MMApiError: return {} async def _api_put( self, path: str, payload: Dict[str, Any] ) -> Dict[str, Any]: - """PUT /api/v4/{path} with JSON body.""" - import aiohttp - url = f"{self._base_url}/api/v4/{path.lstrip('/')}" + """PUT /api/v4/{path}; returns {} on any failure (legacy sentinel).""" try: - async with self._session.put( - url, headers=self._headers(), json=payload - ) as resp: - if resp.status >= 400: - body = await resp.text() - logger.error("MM API PUT %s → %s: %s", path, resp.status, body[:200]) - return {} - return await resp.json() - except aiohttp.ClientError as exc: - logger.error("MM API PUT %s network error: %s", path, exc) + return await self._request_json("PUT", path, payload=payload) + except _MMApiError: return {} async def _upload_file( @@ -198,6 +232,29 @@ async def connect(self) -> bool: if not self._base_url or not self._token: logger.error("Mattermost: URL or token not configured") + # Missing config never succeeds on retry — escalate as a + # non-retryable fatal error so the gateway drops the platform + # from its reconnect queue instead of looping forever. + self._set_fatal_error( + "config_missing", + "MATTERMOST_URL and MATTERMOST_TOKEN must be set", + retryable=False, + ) + return False + + # Single-instance guard: without a scoped lock, two gateway processes + # configured with the same token each open their own /api/v4/websocket + # listener and both receive the same 'posted' events. The per-process + # MessageDeduplicator (self._dedup) is in-memory and cannot dedup + # across processes, so every inbound post would be handled twice + # (duplicate agent runs/replies). Key the lock on URL+token like the + # Discord/Signal/Slack adapters; _acquire_platform_lock records a + # non-retryable fatal error on conflict. + if not self._acquire_platform_lock( + "mattermost-token", + f"{self._base_url}|{self._token}", + "Mattermost bot token", + ): return False self._session = aiohttp.ClientSession( @@ -205,11 +262,38 @@ async def connect(self) -> bool: ) self._closing = False - # Verify credentials and fetch bot identity. - me = await self._api_get("users/me") + # Verify credentials and fetch bot identity. Distinguish a permanent + # auth/permission failure (bad token, wrong URL) from a transient + # network blip reaching users/me so the gateway can stop retrying a + # revoked token but keep retrying a flaky connection. + try: + me = await self._request_json("GET", "users/me") + except _MMApiError as exc: + await self._session.close() + # Release the lock acquired above so a permanent auth failure + # (or a transient blip) does not leak it; mirrors Discord's + # _release_platform_lock() on its post-acquire failure exits. + self._release_platform_lock() + if exc.retryable: + logger.error("Mattermost: transient error reaching server — will retry: %s", exc) + self._set_fatal_error("connect_failed", str(exc), retryable=True) + else: + logger.error("Mattermost: authentication failed — check MATTERMOST_TOKEN and MATTERMOST_URL") + self._set_fatal_error( + "auth_failed", + "Mattermost authentication failed — check MATTERMOST_TOKEN/MATTERMOST_URL", + retryable=False, + ) + return False if not me or "id" not in me: logger.error("Mattermost: failed to authenticate — check MATTERMOST_TOKEN and MATTERMOST_URL") await self._session.close() + self._release_platform_lock() + self._set_fatal_error( + "auth_failed", + "Mattermost authentication failed — check MATTERMOST_TOKEN/MATTERMOST_URL", + retryable=False, + ) return False self._bot_user_id = me["id"] @@ -247,6 +331,14 @@ async def disconnect(self) -> None: if self._session and not self._session.closed: await self._session.close() + # Release the single-instance lock acquired in connect() so another + # gateway process can take over the token (no-op if never acquired). + self._release_platform_lock() + + # Flip _running=False and write the 'disconnected' runtime status so + # is_connected and status tooling reflect the shutdown. No-op-safe + # when a fatal error is already recorded (base guards on that). + self._mark_disconnected() logger.info("Mattermost: disconnected") @@ -293,7 +385,13 @@ async def send( resolved_root = await self._resolve_root_id(reply_to) payload["root_id"] = resolved_root - data = await self._api_post("posts", payload) + try: + data = await self._request_json("POST", "posts", payload=payload) + except _MMApiError as exc: + # Propagate transient/permanent class so _send_with_retry + # retries network/5xx/429 failures with backoff instead of + # silently dropping the response after one plain-text attempt. + return SendResult(success=False, error=str(exc), retryable=exc.retryable) if not data or "id" not in data: return SendResult(success=False, error="Failed to create post") last_id = data["id"] @@ -328,10 +426,17 @@ async def edit_message( ) -> SendResult: """Edit an existing post.""" formatted = self.format_message(content) - data = await self._api_put( - f"posts/{message_id}/patch", - {"message": formatted}, - ) + try: + data = await self._request_json( + "PUT", + f"posts/{message_id}/patch", + payload={"message": formatted}, + ) + except _MMApiError as exc: + # A transient edit failure must not permanently disable in-place + # progress editing for the rest of a streamed response: surface + # retryable so the consumer keeps can_edit=True. + return SendResult(success=False, error=str(exc), retryable=exc.retryable) if not data or "id" not in data: return SendResult(success=False, error="Failed to edit post") return SendResult(success=True, message_id=data["id"]) @@ -615,6 +720,19 @@ async def send_multiple_images( # WebSocket # ------------------------------------------------------------------ + async def _escalate_ws_fatal(self, message: str) -> None: + """Record a non-retryable fatal error and notify the gateway. + + When the WebSocket listener gives up on a permanent auth/permission + failure it must bridge that state back to the gateway: otherwise + ``_ws_task`` ends while ``_running`` stays True, ``is_connected`` + keeps returning True, and the bot is a zombie that the gateway never + reconnects. Mirrors IRC's receive-loop ``finally`` which calls + ``_set_fatal_error`` + ``_notify_fatal_error``. + """ + self._set_fatal_error("ws_auth_failed", message, retryable=False) + await self._notify_fatal_error() + async def _ws_loop(self) -> None: """Connect to the WebSocket and listen for events, reconnecting on failure.""" delay = _RECONNECT_BASE_DELAY @@ -634,9 +752,11 @@ async def _ws_loop(self) -> None: err_str = str(exc).lower() if isinstance(exc, aiohttp.WSServerHandshakeError) and exc.status in {401, 403}: logger.error("Mattermost WS auth failed (HTTP %d) — stopping reconnect", exc.status) + await self._escalate_ws_fatal(f"Mattermost WebSocket auth failed (HTTP {exc.status})") return if "401" in err_str or "403" in err_str or "unauthorized" in err_str: logger.error("Mattermost WS permanent error: %s — stopping reconnect", exc) + await self._escalate_ws_fatal(f"Mattermost WebSocket permanent error: {exc}") return logger.warning("Mattermost WS error: %s — reconnecting in %.0fs", exc, delay) @@ -834,12 +954,23 @@ async def _handle_ws_event(self, event: Dict[str, Any]) -> None: except Exception as exc: logger.warning("Mattermost: error downloading file %s: %s", fid, exc) - # Set message type based on downloaded media types. - if media_types and msg_type == MessageType.TEXT: + # Set message type based on downloaded media types. An attachment + # whose caption happens to start with '/' was provisionally tagged + # COMMAND above; the media type wins so downstream document/image/ + # audio surfacing (which keys on the message type) still fires. + if media_types and msg_type in {MessageType.TEXT, MessageType.COMMAND}: if any(m.startswith("image/") for m in media_types): msg_type = MessageType.PHOTO elif any(m.startswith("audio/") for m in media_types): - msg_type = MessageType.VOICE + # Mattermost has no distinct 'voice note' concept — an uploaded + # audio file is just a file. Classify it AUDIO, not VOICE, so + # run.py surfaces the cached path to the agent as a referenceable + # file (the audio_file_paths branch) instead of force-running STT + # on what may be music, a podcast, or a non-speech clip and never + # handing the agent the actual file. Mirrors Discord, which only + # tags true voice-message attachments VOICE and ordinary audio + # files AUDIO. + msg_type = MessageType.AUDIO elif media_types: msg_type = MessageType.DOCUMENT diff --git a/tests/gateway/test_mattermost.py b/tests/gateway/test_mattermost.py index cafe5ad68a49..c794e3cf95e4 100644 --- a/tests/gateway/test_mattermost.py +++ b/tests/gateway/test_mattermost.py @@ -750,3 +750,377 @@ async def test_document_media_type_is_full_mime(self): assert msg.media_types == ["application/pdf"] assert not msg.media_types[0].startswith("image/") assert not msg.media_types[0].startswith("audio/") + + +# --------------------------------------------------------------------------- +# Inbound audio classification (AUDIO vs VOICE) +# --------------------------------------------------------------------------- + +class TestMattermostAudioClassification: + """An uploaded audio file must be classified MessageType.AUDIO, not VOICE. + + Mattermost has no distinct 'voice note' concept, so every audio attachment + is an ordinary file. run.py only surfaces the cached file path to the + agent for MessageType.AUDIO (the audio_file_paths branch); MessageType.VOICE + is force-routed to STT. Tagging audio as VOICE would transcribe music / + podcasts / non-speech clips and never hand the agent the actual file. + """ + + def setup_method(self): + self.adapter = _make_adapter() + self.adapter._bot_user_id = "bot_user_id" + self.adapter.handle_message = AsyncMock() + + def _make_event(self, file_ids): + post_data = { + "id": "post_audio", + "user_id": "user_123", + "channel_id": "chan_456", + "message": "@bot_user_id audio attached", + "file_ids": file_ids, + } + return { + "event": "posted", + "data": { + "post": json.dumps(post_data), + "channel_type": "O", + "sender_name": "@alice", + }, + } + + @pytest.mark.asyncio + async def test_audio_attachment_classified_audio_not_voice(self): + from gateway.platforms.base import MessageType + + file_info = {"name": "podcast.mp3", "mime_type": "audio/mpeg"} + self.adapter._api_get = AsyncMock(return_value=file_info) + + mock_resp = AsyncMock() + mock_resp.status = 200 + mock_resp.read = AsyncMock(return_value=b"MP3 fake") + mock_resp.__aenter__ = AsyncMock(return_value=mock_resp) + mock_resp.__aexit__ = AsyncMock(return_value=False) + self.adapter._session = MagicMock() + self.adapter._session.get = MagicMock(return_value=mock_resp) + + with patch("gateway.platforms.base.cache_audio_from_bytes", return_value="/tmp/podcast.mp3"), \ + patch("gateway.platforms.base.cache_image_from_bytes"), \ + patch("gateway.platforms.base.cache_document_from_bytes"): + await self.adapter._handle_ws_event(self._make_event(["file_audio"])) + + msg = self.adapter.handle_message.call_args[0][0] + # The fix: audio surfaces as a referenceable file (AUDIO), not force-STT (VOICE). + assert msg.message_type == MessageType.AUDIO + assert msg.message_type != MessageType.VOICE + + +# --------------------------------------------------------------------------- +# Single-instance platform lock +# --------------------------------------------------------------------------- + +class TestMattermostPlatformLock: + """connect() must acquire a scoped lock so two gateway processes sharing + the same token don't both open a WebSocket and double-process every post + (the in-memory MessageDeduplicator cannot dedup across processes).""" + + def _auth_ok(self, adapter): + """Stub REST auth (GET users/me) so connect() reaches the lock/WS path.""" + async def fake_request_json(method, path, *, payload=None): + if path == "users/me": + return {"id": "bot_id", "username": "bot"} + return {} + adapter._request_json = fake_request_json + + @pytest.mark.asyncio + async def test_connect_acquires_scoped_lock(self, monkeypatch): + adapter = _make_adapter() + self._auth_ok(adapter) + # Don't open a real websocket. + monkeypatch.setattr(adapter, "_ws_loop", AsyncMock()) + + calls = {} + + def fake_acquire(scope, identity, metadata=None): + calls["scope"] = scope + calls["identity"] = identity + return (True, None) + + monkeypatch.setattr("gateway.status.acquire_scoped_lock", fake_acquire) + monkeypatch.setattr("gateway.status.release_scoped_lock", lambda scope, identity: None) + + ok = await adapter.connect() + try: + assert ok is True + # The lock must actually have been taken on connect. + assert calls.get("scope") == "mattermost-token" + assert adapter.has_fatal_error is False + finally: + await adapter.disconnect() + + @pytest.mark.asyncio + async def test_connect_refuses_on_lock_conflict(self, monkeypatch): + adapter = _make_adapter() + self._auth_ok(adapter) + + # Another process already owns the token. + monkeypatch.setattr( + "gateway.status.acquire_scoped_lock", + lambda scope, identity, metadata=None: (False, {"pid": 4242}), + ) + monkeypatch.setattr("gateway.status.release_scoped_lock", lambda scope, identity: None) + + ws_started = {"flag": False} + + async def fake_ws(): + ws_started["flag"] = True + + monkeypatch.setattr(adapter, "_ws_loop", fake_ws) + + ok = await adapter.connect() + + assert ok is False + # Conflict is permanent — must escalate non-retryable so the gateway + # drops the platform instead of retrying forever. + assert adapter.has_fatal_error is True + assert adapter.fatal_error_retryable is False + # No WebSocket listener should have been started on the loser. + assert ws_started["flag"] is False + + +# --------------------------------------------------------------------------- +# Regression tests for independently-verified bug fixes +# --------------------------------------------------------------------------- + +def _mock_response(status, *, json_value=None, text_value=""): + """Build an async-context-manager mock aiohttp response.""" + resp = AsyncMock() + resp.status = status + resp.json = AsyncMock(return_value=json_value if json_value is not None else {}) + resp.text = AsyncMock(return_value=text_value) + resp.__aenter__ = AsyncMock(return_value=resp) + resp.__aexit__ = AsyncMock(return_value=False) + return resp + + +class TestMattermostSlashCaptionedAttachment: + """Fix: an attachment whose caption starts with '/' is mis-typed COMMAND, + which suppressed the media-type override (guarded on TEXT only), so the + file was never surfaced as a DOCUMENT to the agent (run.py keys document + surfacing off message_type == DOCUMENT).""" + + def setup_method(self): + self.adapter = _make_adapter() + self.adapter._bot_user_id = "bot_user_id" + self.adapter._bot_username = "hermes-bot" + self.adapter.handle_message = AsyncMock() + + @pytest.mark.asyncio + async def test_slash_caption_document_is_typed_document(self): + from plugins.platforms.mattermost.adapter import MessageType + # DM (channel_type=D) so the '/notes' caption survives unmodified. + post_data = { + "id": "post_slashdoc", + "user_id": "user_123", + "channel_id": "chan_dm", + "message": "/notes for review", + "file_ids": ["file_doc"], + } + event = { + "event": "posted", + "data": { + "post": json.dumps(post_data), + "channel_type": "D", + "sender_name": "@alice", + }, + } + + self.adapter._api_get = AsyncMock( + return_value={"name": "report.pdf", "mime_type": "application/pdf"} + ) + self.adapter._session = MagicMock() + self.adapter._session.get = MagicMock( + return_value=_mock_response(200) + ) + # _mock_response's .read isn't set by default; add it. + dl = self.adapter._session.get.return_value + dl.read = AsyncMock(return_value=b"PDF fake") + + with patch("gateway.platforms.base.cache_document_from_bytes", return_value="/tmp/report.pdf"), \ + patch("gateway.platforms.base.cache_image_from_bytes"): + await self.adapter._handle_ws_event(event) + + msg = self.adapter.handle_message.call_args[0][0] + # The attachment must win over the slash-caption COMMAND tag so the + # cached document is described to the agent downstream. + assert msg.message_type == MessageType.DOCUMENT + assert msg.media_urls == ["/tmp/report.pdf"] + + +class TestMattermostSendRetryable: + """Fix: send() collapsed all failures into a static, non-retryable error, + so _send_with_retry never retried transient 5xx/429/network failures.""" + + def setup_method(self): + self.adapter = _make_adapter() + self.adapter._session = MagicMock() + + @pytest.mark.asyncio + async def test_send_5xx_is_retryable(self): + self.adapter._session.post = MagicMock( + return_value=_mock_response(503, text_value="Service Unavailable") + ) + result = await self.adapter.send("chan_1", "hello") + assert result.success is False + assert result.retryable is True + + @pytest.mark.asyncio + async def test_send_429_is_retryable(self): + self.adapter._session.post = MagicMock( + return_value=_mock_response(429, text_value="Too Many Requests") + ) + result = await self.adapter.send("chan_1", "hello") + assert result.success is False + assert result.retryable is True + + @pytest.mark.asyncio + async def test_send_network_error_is_retryable(self): + import aiohttp + + def _raise(*_a, **_k): + raise aiohttp.ClientConnectionError("connection reset") + + self.adapter._session.post = MagicMock(side_effect=_raise) + result = await self.adapter.send("chan_1", "hello") + assert result.success is False + assert result.retryable is True + + @pytest.mark.asyncio + async def test_send_4xx_is_not_retryable(self): + self.adapter._session.post = MagicMock( + return_value=_mock_response(403, text_value="Forbidden") + ) + result = await self.adapter.send("chan_1", "hello") + assert result.success is False + assert result.retryable is False + + +class TestMattermostEditRetryable: + """Fix: edit_message() never set retryable, so a single transient blip + permanently disabled progress-message editing for a streamed response.""" + + def setup_method(self): + self.adapter = _make_adapter() + self.adapter._session = MagicMock() + + @pytest.mark.asyncio + async def test_edit_5xx_is_retryable(self): + self.adapter._session.put = MagicMock( + return_value=_mock_response(502, text_value="Bad Gateway") + ) + result = await self.adapter.edit_message("chan_1", "post_1", "edited") + assert result.success is False + assert result.retryable is True + + @pytest.mark.asyncio + async def test_edit_4xx_is_not_retryable(self): + self.adapter._session.put = MagicMock( + return_value=_mock_response(404, text_value="Not Found") + ) + result = await self.adapter.edit_message("chan_1", "post_1", "edited") + assert result.success is False + assert result.retryable is False + + +class TestMattermostConnectFatalEscalation: + """Fix: connect() returned False on missing config / auth failure without + recording a fatal error, so the gateway retried the platform forever.""" + + @pytest.mark.asyncio + async def test_missing_config_sets_nonretryable_fatal(self): + from plugins.platforms.mattermost.adapter import MattermostAdapter + cfg = PlatformConfig(enabled=True, token="", extra={"url": ""}) + adapter = MattermostAdapter(cfg) + ok = await adapter.connect() + assert ok is False + assert adapter.has_fatal_error is True + assert adapter.fatal_error_retryable is False + + @pytest.mark.asyncio + async def test_auth_failure_sets_nonretryable_fatal(self): + adapter = _make_adapter() + # users/me returns a 401 → permanent auth failure. + with patch("aiohttp.ClientSession") as mock_session_cls: + session = MagicMock() + session.get = MagicMock( + return_value=_mock_response(401, text_value="Unauthorized") + ) + session.close = AsyncMock() + session.closed = False + mock_session_cls.return_value = session + ok = await adapter.connect() + assert ok is False + assert adapter.has_fatal_error is True + assert adapter.fatal_error_retryable is False + assert session.close.called + + @pytest.mark.asyncio + async def test_transient_connect_failure_is_retryable_fatal(self): + import aiohttp + adapter = _make_adapter() + with patch("aiohttp.ClientSession") as mock_session_cls: + session = MagicMock() + + def _raise(*_a, **_k): + raise aiohttp.ClientConnectionError("network unreachable") + + session.get = MagicMock(side_effect=_raise) + session.close = AsyncMock() + session.closed = False + mock_session_cls.return_value = session + ok = await adapter.connect() + assert ok is False + assert adapter.has_fatal_error is True + assert adapter.fatal_error_retryable is True + + +class TestMattermostDisconnectMarks: + """Fix: disconnect() never called _mark_disconnected(), so is_connected + stayed True and runtime status reported 'connected' after shutdown.""" + + @pytest.mark.asyncio + async def test_disconnect_marks_disconnected(self): + adapter = _make_adapter() + adapter._mark_connected() + assert adapter.is_connected is True + adapter._session = MagicMock() + adapter._session.closed = False + adapter._session.close = AsyncMock() + await adapter.disconnect() + assert adapter.is_connected is False + + +class TestMattermostWsAuthEscalation: + """Fix: on a permanent WS auth failure the listener returned silently + without escalating, leaving a zombie adapter the gateway never reconnects.""" + + @pytest.mark.asyncio + async def test_ws_permanent_auth_failure_escalates(self): + adapter = _make_adapter() + adapter._mark_connected() + notified = {"count": 0} + + async def _handler(_a): + notified["count"] += 1 + + adapter.set_fatal_error_handler(_handler) + + async def _raise_auth(*_a, **_k): + raise RuntimeError("server rejected: 401 unauthorized") + + adapter._ws_connect_and_listen = _raise_auth + await adapter._ws_loop() + + assert adapter.has_fatal_error is True + assert adapter.fatal_error_retryable is False + assert notified["count"] == 1 + assert adapter.is_connected is False