test_bluebubbles.py
1 """Tests for the BlueBubbles iMessage gateway adapter.""" 2 import pytest 3 4 from gateway.config import Platform, PlatformConfig 5 6 7 def _make_adapter(monkeypatch, **extra): 8 monkeypatch.setenv("BLUEBUBBLES_SERVER_URL", "http://localhost:1234") 9 monkeypatch.setenv("BLUEBUBBLES_PASSWORD", "secret") 10 from gateway.platforms.bluebubbles import BlueBubblesAdapter 11 12 cfg = PlatformConfig( 13 enabled=True, 14 extra={ 15 "server_url": "http://localhost:1234", 16 "password": "secret", 17 **extra, 18 }, 19 ) 20 return BlueBubblesAdapter(cfg) 21 22 23 class TestBlueBubblesConfigLoading: 24 def test_apply_env_overrides_bluebubbles(self, monkeypatch): 25 monkeypatch.setenv("BLUEBUBBLES_SERVER_URL", "http://localhost:1234") 26 monkeypatch.setenv("BLUEBUBBLES_PASSWORD", "secret") 27 monkeypatch.setenv("BLUEBUBBLES_WEBHOOK_PORT", "9999") 28 from gateway.config import GatewayConfig, _apply_env_overrides 29 30 config = GatewayConfig() 31 _apply_env_overrides(config) 32 assert Platform.BLUEBUBBLES in config.platforms 33 bc = config.platforms[Platform.BLUEBUBBLES] 34 assert bc.enabled is True 35 assert bc.extra["server_url"] == "http://localhost:1234" 36 assert bc.extra["password"] == "secret" 37 assert bc.extra["webhook_port"] == 9999 38 39 def test_home_channel_set_from_env(self, monkeypatch): 40 monkeypatch.setenv("BLUEBUBBLES_SERVER_URL", "http://localhost:1234") 41 monkeypatch.setenv("BLUEBUBBLES_PASSWORD", "secret") 42 monkeypatch.setenv("BLUEBUBBLES_HOME_CHANNEL", "user@example.com") 43 from gateway.config import GatewayConfig, _apply_env_overrides 44 45 config = GatewayConfig() 46 _apply_env_overrides(config) 47 hc = config.platforms[Platform.BLUEBUBBLES].home_channel 48 assert hc is not None 49 assert hc.chat_id == "user@example.com" 50 51 def test_not_connected_without_password(self, monkeypatch): 52 monkeypatch.setenv("BLUEBUBBLES_SERVER_URL", "http://localhost:1234") 53 monkeypatch.delenv("BLUEBUBBLES_PASSWORD", raising=False) 54 from gateway.config import GatewayConfig, _apply_env_overrides 55 56 config = GatewayConfig() 57 _apply_env_overrides(config) 58 assert Platform.BLUEBUBBLES not in config.get_connected_platforms() 59 60 61 class TestBlueBubblesHelpers: 62 def test_check_requirements(self, monkeypatch): 63 monkeypatch.setenv("BLUEBUBBLES_SERVER_URL", "http://localhost:1234") 64 monkeypatch.setenv("BLUEBUBBLES_PASSWORD", "secret") 65 from gateway.platforms.bluebubbles import check_bluebubbles_requirements 66 67 assert check_bluebubbles_requirements() is True 68 69 def test_supports_message_editing_is_false(self, monkeypatch): 70 adapter = _make_adapter(monkeypatch) 71 assert adapter.SUPPORTS_MESSAGE_EDITING is False 72 73 def test_truncate_message_omits_pagination_suffixes(self, monkeypatch): 74 adapter = _make_adapter(monkeypatch) 75 chunks = adapter.truncate_message("abcdefghij", max_length=6) 76 assert len(chunks) > 1 77 assert "".join(chunks) == "abcdefghij" 78 assert all("(" not in chunk for chunk in chunks) 79 80 @pytest.mark.asyncio 81 async def test_send_splits_paragraphs_into_multiple_bubbles(self, monkeypatch): 82 adapter = _make_adapter(monkeypatch) 83 sent = [] 84 85 async def fake_resolve_chat_guid(chat_id): 86 return "iMessage;-;user@example.com" 87 88 async def fake_api_post(path, payload): 89 sent.append(payload["message"]) 90 return {"data": {"guid": f"msg-{len(sent)}"}} 91 92 monkeypatch.setattr(adapter, "_resolve_chat_guid", fake_resolve_chat_guid) 93 monkeypatch.setattr(adapter, "_api_post", fake_api_post) 94 95 result = await adapter.send("user@example.com", "first thought\n\nsecond thought") 96 97 assert result.success is True 98 assert sent == ["first thought", "second thought"] 99 100 def test_format_message_strips_markdown(self, monkeypatch): 101 adapter = _make_adapter(monkeypatch) 102 assert adapter.format_message("**Hello** `world`") == "Hello world" 103 104 def test_strip_markdown_headers(self, monkeypatch): 105 adapter = _make_adapter(monkeypatch) 106 assert adapter.format_message("## Heading\ntext") == "Heading\ntext" 107 108 def test_strip_markdown_links(self, monkeypatch): 109 adapter = _make_adapter(monkeypatch) 110 assert adapter.format_message("[click here](http://example.com)") == "click here" 111 112 def test_init_normalizes_webhook_path(self, monkeypatch): 113 adapter = _make_adapter(monkeypatch, webhook_path="bluebubbles-webhook") 114 assert adapter.webhook_path == "/bluebubbles-webhook" 115 116 def test_init_preserves_leading_slash(self, monkeypatch): 117 adapter = _make_adapter(monkeypatch, webhook_path="/my-hook") 118 assert adapter.webhook_path == "/my-hook" 119 120 def test_server_url_normalized(self, monkeypatch): 121 adapter = _make_adapter(monkeypatch, server_url="http://localhost:1234/") 122 assert adapter.server_url == "http://localhost:1234" 123 124 def test_server_url_adds_scheme(self, monkeypatch): 125 adapter = _make_adapter(monkeypatch, server_url="localhost:1234") 126 assert adapter.server_url == "http://localhost:1234" 127 128 129 class TestBlueBubblesWebhookParsing: 130 def test_webhook_prefers_chat_guid_over_message_guid(self, monkeypatch): 131 adapter = _make_adapter(monkeypatch) 132 payload = { 133 "guid": "MESSAGE-GUID", 134 "chatGuid": "iMessage;-;user@example.com", 135 "chatIdentifier": "user@example.com", 136 } 137 record = adapter._extract_payload_record(payload) or {} 138 chat_guid = adapter._value( 139 record.get("chatGuid"), 140 payload.get("chatGuid"), 141 record.get("chat_guid"), 142 payload.get("chat_guid"), 143 payload.get("guid"), 144 ) 145 assert chat_guid == "iMessage;-;user@example.com" 146 147 def test_webhook_can_fall_back_to_sender_when_chat_fields_missing(self, monkeypatch): 148 adapter = _make_adapter(monkeypatch) 149 payload = { 150 "data": { 151 "guid": "MESSAGE-GUID", 152 "text": "hello", 153 "handle": {"address": "user@example.com"}, 154 "isFromMe": False, 155 } 156 } 157 record = adapter._extract_payload_record(payload) or {} 158 chat_guid = adapter._value( 159 record.get("chatGuid"), 160 payload.get("chatGuid"), 161 record.get("chat_guid"), 162 payload.get("chat_guid"), 163 payload.get("guid"), 164 ) 165 chat_identifier = adapter._value( 166 record.get("chatIdentifier"), 167 record.get("identifier"), 168 payload.get("chatIdentifier"), 169 payload.get("identifier"), 170 ) 171 sender = ( 172 adapter._value( 173 record.get("handle", {}).get("address") 174 if isinstance(record.get("handle"), dict) 175 else None, 176 record.get("sender"), 177 record.get("from"), 178 record.get("address"), 179 ) 180 or chat_identifier 181 or chat_guid 182 ) 183 if not (chat_guid or chat_identifier) and sender: 184 chat_identifier = sender 185 assert chat_identifier == "user@example.com" 186 187 def test_webhook_extracts_chat_guid_from_chats_array_dm(self, monkeypatch): 188 """BB v1.9+ webhook payloads omit top-level chatGuid; GUID is in chats[0].guid.""" 189 adapter = _make_adapter(monkeypatch) 190 payload = { 191 "type": "new-message", 192 "data": { 193 "guid": "MESSAGE-GUID", 194 "text": "hello", 195 "handle": {"address": "+15551234567"}, 196 "isFromMe": False, 197 "chats": [ 198 {"guid": "any;-;+15551234567", "chatIdentifier": "+15551234567"} 199 ], 200 }, 201 } 202 record = adapter._extract_payload_record(payload) or {} 203 chat_guid = adapter._value( 204 record.get("chatGuid"), 205 payload.get("chatGuid"), 206 record.get("chat_guid"), 207 payload.get("chat_guid"), 208 payload.get("guid"), 209 ) 210 if not chat_guid: 211 _chats = record.get("chats") or [] 212 if _chats and isinstance(_chats[0], dict): 213 chat_guid = _chats[0].get("guid") or _chats[0].get("chatGuid") 214 assert chat_guid == "any;-;+15551234567" 215 216 def test_webhook_extracts_chat_guid_from_chats_array_group(self, monkeypatch): 217 """Group chat GUIDs contain ;+; and must be extracted from chats array.""" 218 adapter = _make_adapter(monkeypatch) 219 payload = { 220 "type": "new-message", 221 "data": { 222 "guid": "MESSAGE-GUID", 223 "text": "hello everyone", 224 "handle": {"address": "+15551234567"}, 225 "isFromMe": False, 226 "isGroup": True, 227 "chats": [{"guid": "any;+;chat-uuid-abc123"}], 228 }, 229 } 230 record = adapter._extract_payload_record(payload) or {} 231 chat_guid = adapter._value( 232 record.get("chatGuid"), 233 payload.get("chatGuid"), 234 record.get("chat_guid"), 235 payload.get("chat_guid"), 236 payload.get("guid"), 237 ) 238 if not chat_guid: 239 _chats = record.get("chats") or [] 240 if _chats and isinstance(_chats[0], dict): 241 chat_guid = _chats[0].get("guid") or _chats[0].get("chatGuid") 242 assert chat_guid == "any;+;chat-uuid-abc123" 243 244 def test_extract_payload_record_accepts_list_data(self, monkeypatch): 245 adapter = _make_adapter(monkeypatch) 246 payload = { 247 "type": "new-message", 248 "data": [ 249 { 250 "text": "hello", 251 "chatGuid": "iMessage;-;user@example.com", 252 "chatIdentifier": "user@example.com", 253 } 254 ], 255 } 256 record = adapter._extract_payload_record(payload) 257 assert record == payload["data"][0] 258 259 def test_extract_payload_record_dict_data(self, monkeypatch): 260 adapter = _make_adapter(monkeypatch) 261 payload = {"data": {"text": "hello", "chatGuid": "iMessage;-;+1234"}} 262 record = adapter._extract_payload_record(payload) 263 assert record["text"] == "hello" 264 265 def test_extract_payload_record_fallback_to_message(self, monkeypatch): 266 adapter = _make_adapter(monkeypatch) 267 payload = {"message": {"text": "hello"}} 268 record = adapter._extract_payload_record(payload) 269 assert record["text"] == "hello" 270 271 272 class TestBlueBubblesGuidResolution: 273 def test_raw_guid_returned_as_is(self, monkeypatch): 274 """If target already contains ';' it's a raw GUID — return unchanged.""" 275 adapter = _make_adapter(monkeypatch) 276 import asyncio 277 278 result = asyncio.get_event_loop().run_until_complete( 279 adapter._resolve_chat_guid("iMessage;-;user@example.com") 280 ) 281 assert result == "iMessage;-;user@example.com" 282 283 def test_empty_target_returns_none(self, monkeypatch): 284 adapter = _make_adapter(monkeypatch) 285 import asyncio 286 287 result = asyncio.get_event_loop().run_until_complete( 288 adapter._resolve_chat_guid("") 289 ) 290 assert result is None 291 292 293 class TestBlueBubblesAttachmentDownload: 294 """Verify _download_attachment routes to the correct cache helper.""" 295 296 def test_download_image_uses_image_cache(self, monkeypatch): 297 """Image MIME routes to cache_image_from_bytes.""" 298 adapter = _make_adapter(monkeypatch) 299 import asyncio 300 import httpx 301 302 # Mock the HTTP client response 303 class MockResponse: 304 status_code = 200 305 content = b"\x89PNG\r\n\x1a\n" 306 307 def raise_for_status(self): 308 pass 309 310 async def mock_get(*args, **kwargs): 311 return MockResponse() 312 313 adapter.client = type("MockClient", (), {"get": mock_get})() 314 315 cached_path = None 316 317 def mock_cache_image(data, ext): 318 nonlocal cached_path 319 cached_path = f"/tmp/test_image{ext}" 320 return cached_path 321 322 monkeypatch.setattr( 323 "gateway.platforms.bluebubbles.cache_image_from_bytes", 324 mock_cache_image, 325 ) 326 327 att_meta = {"mimeType": "image/png", "transferName": "photo.png"} 328 result = asyncio.get_event_loop().run_until_complete( 329 adapter._download_attachment("att-guid-123", att_meta) 330 ) 331 assert result == "/tmp/test_image.png" 332 333 def test_download_audio_uses_audio_cache(self, monkeypatch): 334 """Audio MIME routes to cache_audio_from_bytes.""" 335 adapter = _make_adapter(monkeypatch) 336 import asyncio 337 338 class MockResponse: 339 status_code = 200 340 content = b"fake-audio-data" 341 342 def raise_for_status(self): 343 pass 344 345 async def mock_get(*args, **kwargs): 346 return MockResponse() 347 348 adapter.client = type("MockClient", (), {"get": mock_get})() 349 350 cached_path = None 351 352 def mock_cache_audio(data, ext): 353 nonlocal cached_path 354 cached_path = f"/tmp/test_audio{ext}" 355 return cached_path 356 357 monkeypatch.setattr( 358 "gateway.platforms.bluebubbles.cache_audio_from_bytes", 359 mock_cache_audio, 360 ) 361 362 att_meta = {"mimeType": "audio/mpeg", "transferName": "voice.mp3"} 363 result = asyncio.get_event_loop().run_until_complete( 364 adapter._download_attachment("att-guid-456", att_meta) 365 ) 366 assert result == "/tmp/test_audio.mp3" 367 368 def test_download_document_uses_document_cache(self, monkeypatch): 369 """Non-image/audio MIME routes to cache_document_from_bytes.""" 370 adapter = _make_adapter(monkeypatch) 371 import asyncio 372 373 class MockResponse: 374 status_code = 200 375 content = b"fake-doc-data" 376 377 def raise_for_status(self): 378 pass 379 380 async def mock_get(*args, **kwargs): 381 return MockResponse() 382 383 adapter.client = type("MockClient", (), {"get": mock_get})() 384 385 cached_path = None 386 387 def mock_cache_doc(data, filename): 388 nonlocal cached_path 389 cached_path = f"/tmp/{filename}" 390 return cached_path 391 392 monkeypatch.setattr( 393 "gateway.platforms.bluebubbles.cache_document_from_bytes", 394 mock_cache_doc, 395 ) 396 397 att_meta = {"mimeType": "application/pdf", "transferName": "report.pdf"} 398 result = asyncio.get_event_loop().run_until_complete( 399 adapter._download_attachment("att-guid-789", att_meta) 400 ) 401 assert result == "/tmp/report.pdf" 402 403 def test_download_returns_none_without_client(self, monkeypatch): 404 """No client → returns None gracefully.""" 405 adapter = _make_adapter(monkeypatch) 406 adapter.client = None 407 import asyncio 408 409 result = asyncio.get_event_loop().run_until_complete( 410 adapter._download_attachment("att-guid", {"mimeType": "image/png"}) 411 ) 412 assert result is None 413 414 415 # --------------------------------------------------------------------------- 416 # Webhook registration 417 # --------------------------------------------------------------------------- 418 419 420 class TestBlueBubblesWebhookUrl: 421 """_webhook_url property normalises local hosts to 'localhost'.""" 422 423 def test_default_host(self, monkeypatch): 424 adapter = _make_adapter(monkeypatch) 425 # Default webhook_host is 0.0.0.0 → normalized to localhost 426 assert "localhost" in adapter._webhook_url 427 assert str(adapter.webhook_port) in adapter._webhook_url 428 assert adapter.webhook_path in adapter._webhook_url 429 430 @pytest.mark.parametrize("host", ["0.0.0.0", "127.0.0.1", "localhost", "::"]) 431 def test_local_hosts_normalized(self, monkeypatch, host): 432 adapter = _make_adapter(monkeypatch, webhook_host=host) 433 assert adapter._webhook_url.startswith("http://localhost:") 434 435 def test_custom_host_preserved(self, monkeypatch): 436 adapter = _make_adapter(monkeypatch, webhook_host="192.168.1.50") 437 assert "192.168.1.50" in adapter._webhook_url 438 439 def test_register_url_embeds_password(self, monkeypatch): 440 """_webhook_register_url should append ?password=... for inbound auth.""" 441 adapter = _make_adapter(monkeypatch, password="secret123") 442 assert adapter._webhook_register_url.endswith("?password=secret123") 443 assert adapter._webhook_register_url.startswith(adapter._webhook_url) 444 445 def test_register_url_url_encodes_password(self, monkeypatch): 446 """Passwords with special characters must be URL-encoded.""" 447 adapter = _make_adapter(monkeypatch, password="W9fTC&L5JL*@") 448 assert "password=W9fTC%26L5JL%2A%40" in adapter._webhook_register_url 449 450 def test_register_url_omits_query_when_no_password(self, monkeypatch): 451 """If no password is configured, the register URL should be the bare URL.""" 452 monkeypatch.delenv("BLUEBUBBLES_PASSWORD", raising=False) 453 from gateway.platforms.bluebubbles import BlueBubblesAdapter 454 cfg = PlatformConfig( 455 enabled=True, 456 extra={"server_url": "http://localhost:1234", "password": ""}, 457 ) 458 adapter = BlueBubblesAdapter(cfg) 459 assert adapter._webhook_register_url == adapter._webhook_url 460 461 462 class TestBlueBubblesWebhookRegistration: 463 """Tests for _register_webhook, _unregister_webhook, _find_registered_webhooks.""" 464 465 @staticmethod 466 def _mock_client(get_response=None, post_response=None, delete_ok=True): 467 """Build a tiny mock httpx.AsyncClient.""" 468 469 async def mock_get(*args, **kwargs): 470 class R: 471 status_code = 200 472 def raise_for_status(self): 473 pass 474 def json(self): 475 return get_response or {"status": 200, "data": []} 476 return R() 477 478 async def mock_post(*args, **kwargs): 479 class R: 480 status_code = 200 481 def raise_for_status(self): 482 pass 483 def json(self): 484 return post_response or {"status": 200, "data": {}} 485 return R() 486 487 async def mock_delete(*args, **kwargs): 488 class R: 489 status_code = 200 if delete_ok else 500 490 def raise_for_status(self_inner): 491 if not delete_ok: 492 raise Exception("delete failed") 493 return R() 494 495 return type( 496 "MockClient", (), 497 {"get": mock_get, "post": mock_post, "delete": mock_delete}, 498 )() 499 500 # -- _find_registered_webhooks -- 501 502 def test_find_registered_webhooks_returns_matches(self, monkeypatch): 503 import asyncio 504 adapter = _make_adapter(monkeypatch) 505 url = adapter._webhook_url 506 adapter.client = self._mock_client( 507 get_response={"status": 200, "data": [ 508 {"id": 1, "url": url, "events": ["new-message"]}, 509 {"id": 2, "url": "http://other:9999/hook", "events": ["message"]}, 510 ]} 511 ) 512 result = asyncio.get_event_loop().run_until_complete( 513 adapter._find_registered_webhooks(url) 514 ) 515 assert len(result) == 1 516 assert result[0]["id"] == 1 517 518 def test_find_registered_webhooks_empty_when_none(self, monkeypatch): 519 import asyncio 520 adapter = _make_adapter(monkeypatch) 521 adapter.client = self._mock_client( 522 get_response={"status": 200, "data": []} 523 ) 524 result = asyncio.get_event_loop().run_until_complete( 525 adapter._find_registered_webhooks(adapter._webhook_url) 526 ) 527 assert result == [] 528 529 def test_find_registered_webhooks_handles_api_error(self, monkeypatch): 530 import asyncio 531 adapter = _make_adapter(monkeypatch) 532 adapter.client = self._mock_client() 533 534 # Override _api_get to raise 535 async def bad_get(path): 536 raise ConnectionError("server down") 537 adapter._api_get = bad_get 538 539 result = asyncio.get_event_loop().run_until_complete( 540 adapter._find_registered_webhooks(adapter._webhook_url) 541 ) 542 assert result == [] 543 544 # -- _register_webhook -- 545 546 def test_register_fresh(self, monkeypatch): 547 """No existing webhook → POST creates one.""" 548 import asyncio 549 adapter = _make_adapter(monkeypatch) 550 adapter.client = self._mock_client( 551 get_response={"status": 200, "data": []}, 552 post_response={"status": 200, "data": {"id": 42}}, 553 ) 554 ok = asyncio.get_event_loop().run_until_complete( 555 adapter._register_webhook() 556 ) 557 assert ok is True 558 559 def test_register_accepts_201(self, monkeypatch): 560 """BB might return 201 Created — must still succeed.""" 561 import asyncio 562 adapter = _make_adapter(monkeypatch) 563 adapter.client = self._mock_client( 564 get_response={"status": 200, "data": []}, 565 post_response={"status": 201, "data": {"id": 43}}, 566 ) 567 ok = asyncio.get_event_loop().run_until_complete( 568 adapter._register_webhook() 569 ) 570 assert ok is True 571 572 def test_register_reuses_existing(self, monkeypatch): 573 """Crash resilience — existing registration is reused, no POST needed.""" 574 import asyncio 575 adapter = _make_adapter(monkeypatch) 576 url = adapter._webhook_register_url 577 adapter.client = self._mock_client( 578 get_response={"status": 200, "data": [ 579 {"id": 7, "url": url, "events": ["new-message"]}, 580 ]}, 581 ) 582 583 # Track whether POST was called 584 post_called = False 585 orig_api_post = adapter._api_post 586 async def tracking_post(path, payload): 587 nonlocal post_called 588 post_called = True 589 return await orig_api_post(path, payload) 590 adapter._api_post = tracking_post 591 592 ok = asyncio.get_event_loop().run_until_complete( 593 adapter._register_webhook() 594 ) 595 assert ok is True 596 assert not post_called, "Should reuse existing, not POST again" 597 598 def test_register_returns_false_without_client(self, monkeypatch): 599 import asyncio 600 adapter = _make_adapter(monkeypatch) 601 adapter.client = None 602 ok = asyncio.get_event_loop().run_until_complete( 603 adapter._register_webhook() 604 ) 605 assert ok is False 606 607 def test_register_returns_false_on_server_error(self, monkeypatch): 608 import asyncio 609 adapter = _make_adapter(monkeypatch) 610 adapter.client = self._mock_client( 611 get_response={"status": 200, "data": []}, 612 post_response={"status": 500, "message": "internal error"}, 613 ) 614 ok = asyncio.get_event_loop().run_until_complete( 615 adapter._register_webhook() 616 ) 617 assert ok is False 618 619 # -- _unregister_webhook -- 620 621 def test_unregister_removes_matching(self, monkeypatch): 622 import asyncio 623 adapter = _make_adapter(monkeypatch) 624 url = adapter._webhook_register_url 625 adapter.client = self._mock_client( 626 get_response={"status": 200, "data": [ 627 {"id": 10, "url": url}, 628 ]}, 629 ) 630 ok = asyncio.get_event_loop().run_until_complete( 631 adapter._unregister_webhook() 632 ) 633 assert ok is True 634 635 def test_unregister_removes_all_duplicates(self, monkeypatch): 636 """Multiple orphaned registrations for same URL — all get removed.""" 637 import asyncio 638 adapter = _make_adapter(monkeypatch) 639 url = adapter._webhook_register_url 640 deleted_ids = [] 641 642 async def mock_delete(*args, **kwargs): 643 # Extract ID from URL 644 url_str = args[0] if args else "" 645 deleted_ids.append(url_str) 646 class R: 647 status_code = 200 648 def raise_for_status(self): 649 pass 650 return R() 651 652 adapter.client = self._mock_client( 653 get_response={"status": 200, "data": [ 654 {"id": 1, "url": url}, 655 {"id": 2, "url": url}, 656 {"id": 3, "url": "http://other/hook"}, 657 ]}, 658 ) 659 adapter.client.delete = mock_delete 660 661 ok = asyncio.get_event_loop().run_until_complete( 662 adapter._unregister_webhook() 663 ) 664 assert ok is True 665 assert len(deleted_ids) == 2 666 667 def test_unregister_returns_false_without_client(self, monkeypatch): 668 import asyncio 669 adapter = _make_adapter(monkeypatch) 670 adapter.client = None 671 ok = asyncio.get_event_loop().run_until_complete( 672 adapter._unregister_webhook() 673 ) 674 assert ok is False 675 676 def test_unregister_handles_api_failure_gracefully(self, monkeypatch): 677 import asyncio 678 adapter = _make_adapter(monkeypatch) 679 adapter.client = self._mock_client() 680 681 async def bad_get(path): 682 raise ConnectionError("server down") 683 adapter._api_get = bad_get 684 685 ok = asyncio.get_event_loop().run_until_complete( 686 adapter._unregister_webhook() 687 ) 688 assert ok is False