test_mattermost.py
1 """Tests for Mattermost platform adapter.""" 2 import json 3 import os 4 import time 5 import pytest 6 from unittest.mock import MagicMock, patch, AsyncMock 7 8 from gateway.config import Platform, PlatformConfig 9 10 11 # --------------------------------------------------------------------------- 12 # Platform & Config 13 # --------------------------------------------------------------------------- 14 15 class TestMattermostConfigLoading: 16 def test_apply_env_overrides_mattermost(self, monkeypatch): 17 monkeypatch.setenv("MATTERMOST_TOKEN", "mm-tok-abc123") 18 monkeypatch.setenv("MATTERMOST_URL", "https://mm.example.com") 19 20 from gateway.config import GatewayConfig, _apply_env_overrides 21 config = GatewayConfig() 22 _apply_env_overrides(config) 23 24 assert Platform.MATTERMOST in config.platforms 25 mc = config.platforms[Platform.MATTERMOST] 26 assert mc.enabled is True 27 assert mc.token == "mm-tok-abc123" 28 assert mc.extra.get("url") == "https://mm.example.com" 29 30 def test_mattermost_not_loaded_without_token(self, monkeypatch): 31 monkeypatch.delenv("MATTERMOST_TOKEN", raising=False) 32 monkeypatch.delenv("MATTERMOST_URL", raising=False) 33 34 from gateway.config import GatewayConfig, _apply_env_overrides 35 config = GatewayConfig() 36 _apply_env_overrides(config) 37 38 assert Platform.MATTERMOST not in config.platforms 39 40 def test_mattermost_home_channel(self, monkeypatch): 41 monkeypatch.setenv("MATTERMOST_TOKEN", "mm-tok-abc123") 42 monkeypatch.setenv("MATTERMOST_URL", "https://mm.example.com") 43 monkeypatch.setenv("MATTERMOST_HOME_CHANNEL", "ch_abc123") 44 monkeypatch.setenv("MATTERMOST_HOME_CHANNEL_NAME", "General") 45 46 from gateway.config import GatewayConfig, _apply_env_overrides 47 config = GatewayConfig() 48 _apply_env_overrides(config) 49 50 home = config.get_home_channel(Platform.MATTERMOST) 51 assert home is not None 52 assert home.chat_id == "ch_abc123" 53 assert home.name == "General" 54 55 def test_mattermost_url_warning_without_url(self, monkeypatch): 56 """MATTERMOST_TOKEN set but MATTERMOST_URL missing should still load.""" 57 monkeypatch.setenv("MATTERMOST_TOKEN", "mm-tok-abc123") 58 monkeypatch.delenv("MATTERMOST_URL", raising=False) 59 60 from gateway.config import GatewayConfig, _apply_env_overrides 61 config = GatewayConfig() 62 _apply_env_overrides(config) 63 64 assert Platform.MATTERMOST in config.platforms 65 assert config.platforms[Platform.MATTERMOST].extra.get("url") == "" 66 67 68 # --------------------------------------------------------------------------- 69 # Adapter format / truncate 70 # --------------------------------------------------------------------------- 71 72 def _make_adapter(): 73 """Create a MattermostAdapter with mocked config.""" 74 from gateway.platforms.mattermost import MattermostAdapter 75 config = PlatformConfig( 76 enabled=True, 77 token="test-token", 78 extra={"url": "https://mm.example.com"}, 79 ) 80 adapter = MattermostAdapter(config) 81 return adapter 82 83 84 class TestMattermostFormatMessage: 85 def setup_method(self): 86 self.adapter = _make_adapter() 87 88 def test_image_markdown_to_url(self): 89 """ should be converted to just the URL.""" 90 result = self.adapter.format_message("") 91 assert result == "https://img.example.com/cat.png" 92 93 def test_image_markdown_strips_alt_text(self): 94 result = self.adapter.format_message("Here:  done") 95 assert "" 106 assert self.adapter.format_message(content) == content 107 108 def test_plain_text_unchanged(self): 109 content = "Hello, world!" 110 assert self.adapter.format_message(content) == content 111 112 def test_multiple_images(self): 113 content = " text " 114 result = self.adapter.format_message(content) 115 assert "![" not in result 116 assert "http://a.com/1.png" in result 117 assert "http://b.com/2.png" in result 118 119 120 class TestMattermostTruncateMessage: 121 def setup_method(self): 122 self.adapter = _make_adapter() 123 124 def test_short_message_single_chunk(self): 125 msg = "Hello, world!" 126 chunks = self.adapter.truncate_message(msg, 4000) 127 assert len(chunks) == 1 128 assert chunks[0] == msg 129 130 def test_long_message_splits(self): 131 msg = "a " * 2500 # 5000 chars 132 chunks = self.adapter.truncate_message(msg, 4000) 133 assert len(chunks) >= 2 134 for chunk in chunks: 135 assert len(chunk) <= 4000 136 137 def test_custom_max_length(self): 138 msg = "Hello " * 20 139 chunks = self.adapter.truncate_message(msg, max_length=50) 140 assert all(len(c) <= 50 for c in chunks) 141 142 def test_exactly_at_limit(self): 143 msg = "x" * 4000 144 chunks = self.adapter.truncate_message(msg, 4000) 145 assert len(chunks) == 1 146 147 148 # --------------------------------------------------------------------------- 149 # Send 150 # --------------------------------------------------------------------------- 151 152 class TestMattermostSend: 153 def setup_method(self): 154 self.adapter = _make_adapter() 155 self.adapter._session = MagicMock() 156 157 @pytest.mark.asyncio 158 async def test_send_calls_api_post(self): 159 """send() should POST to /api/v4/posts with channel_id and message.""" 160 mock_resp = AsyncMock() 161 mock_resp.status = 200 162 mock_resp.json = AsyncMock(return_value={"id": "post123"}) 163 mock_resp.text = AsyncMock(return_value="") 164 mock_resp.__aenter__ = AsyncMock(return_value=mock_resp) 165 mock_resp.__aexit__ = AsyncMock(return_value=False) 166 167 self.adapter._session.post = MagicMock(return_value=mock_resp) 168 169 result = await self.adapter.send("channel_1", "Hello!") 170 171 assert result.success is True 172 assert result.message_id == "post123" 173 174 # Verify post was called with correct URL 175 call_args = self.adapter._session.post.call_args 176 assert "/api/v4/posts" in call_args[0][0] 177 # Verify payload 178 payload = call_args[1]["json"] 179 assert payload["channel_id"] == "channel_1" 180 assert payload["message"] == "Hello!" 181 182 @pytest.mark.asyncio 183 async def test_send_empty_content_succeeds(self): 184 """Empty content should return success without calling the API.""" 185 result = await self.adapter.send("channel_1", "") 186 assert result.success is True 187 188 @pytest.mark.asyncio 189 async def test_send_with_thread_reply(self): 190 """When reply_mode is 'thread', reply_to should become root_id.""" 191 self.adapter._reply_mode = "thread" 192 193 mock_resp = AsyncMock() 194 mock_resp.status = 200 195 mock_resp.json = AsyncMock(return_value={"id": "post456"}) 196 mock_resp.text = AsyncMock(return_value="") 197 mock_resp.__aenter__ = AsyncMock(return_value=mock_resp) 198 mock_resp.__aexit__ = AsyncMock(return_value=False) 199 200 self.adapter._session.post = MagicMock(return_value=mock_resp) 201 202 result = await self.adapter.send("channel_1", "Reply!", reply_to="root_post") 203 204 assert result.success is True 205 payload = self.adapter._session.post.call_args[1]["json"] 206 assert payload["root_id"] == "root_post" 207 208 @pytest.mark.asyncio 209 async def test_send_without_thread_no_root_id(self): 210 """When reply_mode is 'off', reply_to should NOT set root_id.""" 211 self.adapter._reply_mode = "off" 212 213 mock_resp = AsyncMock() 214 mock_resp.status = 200 215 mock_resp.json = AsyncMock(return_value={"id": "post789"}) 216 mock_resp.text = AsyncMock(return_value="") 217 mock_resp.__aenter__ = AsyncMock(return_value=mock_resp) 218 mock_resp.__aexit__ = AsyncMock(return_value=False) 219 220 self.adapter._session.post = MagicMock(return_value=mock_resp) 221 222 result = await self.adapter.send("channel_1", "Reply!", reply_to="root_post") 223 224 assert result.success is True 225 payload = self.adapter._session.post.call_args[1]["json"] 226 assert "root_id" not in payload 227 228 @pytest.mark.asyncio 229 async def test_send_api_failure(self): 230 """When API returns error, send should return failure.""" 231 mock_resp = AsyncMock() 232 mock_resp.status = 500 233 mock_resp.json = AsyncMock(return_value={}) 234 mock_resp.text = AsyncMock(return_value="Internal Server Error") 235 mock_resp.__aenter__ = AsyncMock(return_value=mock_resp) 236 mock_resp.__aexit__ = AsyncMock(return_value=False) 237 238 self.adapter._session.post = MagicMock(return_value=mock_resp) 239 240 result = await self.adapter.send("channel_1", "Hello!") 241 242 assert result.success is False 243 244 245 # --------------------------------------------------------------------------- 246 # WebSocket event parsing 247 # --------------------------------------------------------------------------- 248 249 class TestMattermostWebSocketParsing: 250 def setup_method(self): 251 self.adapter = _make_adapter() 252 self.adapter._bot_user_id = "bot_user_id" 253 self.adapter._bot_username = "hermes-bot" 254 # Mock handle_message to capture the MessageEvent without processing 255 self.adapter.handle_message = AsyncMock() 256 257 @pytest.mark.asyncio 258 async def test_parse_posted_event(self): 259 """'posted' events should extract message from double-encoded post JSON.""" 260 post_data = { 261 "id": "post_abc", 262 "user_id": "user_123", 263 "channel_id": "chan_456", 264 "message": "@bot_user_id Hello from Matrix!", 265 } 266 event = { 267 "event": "posted", 268 "data": { 269 "post": json.dumps(post_data), # double-encoded JSON string 270 "channel_type": "O", 271 "sender_name": "@alice", 272 }, 273 } 274 275 await self.adapter._handle_ws_event(event) 276 assert self.adapter.handle_message.called 277 msg_event = self.adapter.handle_message.call_args[0][0] 278 # @mention is stripped from the message text 279 assert msg_event.text == "Hello from Matrix!" 280 assert msg_event.message_id == "post_abc" 281 282 @pytest.mark.asyncio 283 async def test_ignore_own_messages(self): 284 """Messages from the bot's own user_id should be ignored.""" 285 post_data = { 286 "id": "post_self", 287 "user_id": "bot_user_id", # same as bot 288 "channel_id": "chan_456", 289 "message": "Bot echo", 290 } 291 event = { 292 "event": "posted", 293 "data": { 294 "post": json.dumps(post_data), 295 "channel_type": "O", 296 }, 297 } 298 299 await self.adapter._handle_ws_event(event) 300 assert not self.adapter.handle_message.called 301 302 @pytest.mark.asyncio 303 async def test_ignore_non_posted_events(self): 304 """Non-'posted' events should be ignored.""" 305 event = { 306 "event": "typing", 307 "data": {"user_id": "user_123"}, 308 } 309 310 await self.adapter._handle_ws_event(event) 311 assert not self.adapter.handle_message.called 312 313 @pytest.mark.asyncio 314 async def test_ignore_system_posts(self): 315 """Posts with a 'type' field (system messages) should be ignored.""" 316 post_data = { 317 "id": "sys_post", 318 "user_id": "user_123", 319 "channel_id": "chan_456", 320 "message": "user joined", 321 "type": "system_join_channel", 322 } 323 event = { 324 "event": "posted", 325 "data": { 326 "post": json.dumps(post_data), 327 "channel_type": "O", 328 }, 329 } 330 331 await self.adapter._handle_ws_event(event) 332 assert not self.adapter.handle_message.called 333 334 @pytest.mark.asyncio 335 async def test_channel_type_mapping(self): 336 """channel_type 'D' should map to 'dm'.""" 337 post_data = { 338 "id": "post_dm", 339 "user_id": "user_123", 340 "channel_id": "chan_dm", 341 "message": "DM message", 342 } 343 event = { 344 "event": "posted", 345 "data": { 346 "post": json.dumps(post_data), 347 "channel_type": "D", 348 "sender_name": "@bob", 349 }, 350 } 351 352 await self.adapter._handle_ws_event(event) 353 assert self.adapter.handle_message.called 354 msg_event = self.adapter.handle_message.call_args[0][0] 355 assert msg_event.source.chat_type == "dm" 356 357 @pytest.mark.asyncio 358 async def test_thread_id_from_root_id(self): 359 """Post with root_id should have thread_id set.""" 360 post_data = { 361 "id": "post_reply", 362 "user_id": "user_123", 363 "channel_id": "chan_456", 364 "message": "@bot_user_id Thread reply", 365 "root_id": "root_post_123", 366 } 367 event = { 368 "event": "posted", 369 "data": { 370 "post": json.dumps(post_data), 371 "channel_type": "O", 372 "sender_name": "@alice", 373 }, 374 } 375 376 await self.adapter._handle_ws_event(event) 377 assert self.adapter.handle_message.called 378 msg_event = self.adapter.handle_message.call_args[0][0] 379 assert msg_event.source.thread_id == "root_post_123" 380 381 @pytest.mark.asyncio 382 async def test_invalid_post_json_ignored(self): 383 """Invalid JSON in data.post should be silently ignored.""" 384 event = { 385 "event": "posted", 386 "data": { 387 "post": "not-valid-json{{{", 388 "channel_type": "O", 389 }, 390 } 391 392 await self.adapter._handle_ws_event(event) 393 assert not self.adapter.handle_message.called 394 395 396 # --------------------------------------------------------------------------- 397 # Mention behavior (require_mention + free_response_channels) 398 # --------------------------------------------------------------------------- 399 400 class TestMattermostMentionBehavior: 401 def setup_method(self): 402 self.adapter = _make_adapter() 403 self.adapter._bot_user_id = "bot_user_id" 404 self.adapter._bot_username = "hermes-bot" 405 self.adapter.handle_message = AsyncMock() 406 407 def _make_event(self, message, channel_type="O", channel_id="chan_456"): 408 post_data = { 409 "id": "post_mention", 410 "user_id": "user_123", 411 "channel_id": channel_id, 412 "message": message, 413 } 414 return { 415 "event": "posted", 416 "data": { 417 "post": json.dumps(post_data), 418 "channel_type": channel_type, 419 "sender_name": "@alice", 420 }, 421 } 422 423 @pytest.mark.asyncio 424 async def test_require_mention_true_skips_without_mention(self): 425 """Default: messages without @mention in channels are skipped.""" 426 with patch.dict(os.environ, {}, clear=False): 427 os.environ.pop("MATTERMOST_REQUIRE_MENTION", None) 428 os.environ.pop("MATTERMOST_FREE_RESPONSE_CHANNELS", None) 429 await self.adapter._handle_ws_event(self._make_event("hello")) 430 assert not self.adapter.handle_message.called 431 432 @pytest.mark.asyncio 433 async def test_require_mention_false_responds_to_all(self): 434 """MATTERMOST_REQUIRE_MENTION=false: respond to all channel messages.""" 435 with patch.dict(os.environ, {"MATTERMOST_REQUIRE_MENTION": "false"}): 436 await self.adapter._handle_ws_event(self._make_event("hello")) 437 assert self.adapter.handle_message.called 438 439 @pytest.mark.asyncio 440 async def test_free_response_channel_responds_without_mention(self): 441 """Messages in free-response channels don't need @mention.""" 442 with patch.dict(os.environ, {"MATTERMOST_FREE_RESPONSE_CHANNELS": "chan_456,chan_789"}): 443 os.environ.pop("MATTERMOST_REQUIRE_MENTION", None) 444 await self.adapter._handle_ws_event(self._make_event("hello", channel_id="chan_456")) 445 assert self.adapter.handle_message.called 446 447 @pytest.mark.asyncio 448 async def test_non_free_channel_still_requires_mention(self): 449 """Channels NOT in free-response list still require @mention.""" 450 with patch.dict(os.environ, {"MATTERMOST_FREE_RESPONSE_CHANNELS": "chan_789"}): 451 os.environ.pop("MATTERMOST_REQUIRE_MENTION", None) 452 await self.adapter._handle_ws_event(self._make_event("hello", channel_id="chan_456")) 453 assert not self.adapter.handle_message.called 454 455 @pytest.mark.asyncio 456 async def test_dm_always_responds(self): 457 """DMs (channel_type=D) always respond regardless of mention settings.""" 458 with patch.dict(os.environ, {}, clear=False): 459 os.environ.pop("MATTERMOST_REQUIRE_MENTION", None) 460 await self.adapter._handle_ws_event(self._make_event("hello", channel_type="D")) 461 assert self.adapter.handle_message.called 462 463 @pytest.mark.asyncio 464 async def test_mention_stripped_from_text(self): 465 """@mention is stripped from message text.""" 466 with patch.dict(os.environ, {}, clear=False): 467 os.environ.pop("MATTERMOST_REQUIRE_MENTION", None) 468 await self.adapter._handle_ws_event( 469 self._make_event("@hermes-bot what is 2+2") 470 ) 471 assert self.adapter.handle_message.called 472 msg = self.adapter.handle_message.call_args[0][0] 473 assert "@hermes-bot" not in msg.text 474 assert "2+2" in msg.text 475 476 477 # --------------------------------------------------------------------------- 478 # File upload (send_image) 479 # --------------------------------------------------------------------------- 480 481 class TestMattermostFileUpload: 482 def setup_method(self): 483 self.adapter = _make_adapter() 484 self.adapter._session = MagicMock() 485 486 @pytest.mark.asyncio 487 @patch("tools.url_safety.is_safe_url", return_value=True) 488 async def test_send_image_downloads_and_uploads(self, _mock_safe): 489 """send_image should download the URL, upload via /api/v4/files, then post.""" 490 # Mock the download (GET) 491 mock_dl_resp = AsyncMock() 492 mock_dl_resp.status = 200 493 mock_dl_resp.read = AsyncMock(return_value=b"\x89PNG\x00fake-image-data") 494 mock_dl_resp.content_type = "image/png" 495 mock_dl_resp.__aenter__ = AsyncMock(return_value=mock_dl_resp) 496 mock_dl_resp.__aexit__ = AsyncMock(return_value=False) 497 498 # Mock the upload (POST to /files) 499 mock_upload_resp = AsyncMock() 500 mock_upload_resp.status = 200 501 mock_upload_resp.json = AsyncMock(return_value={ 502 "file_infos": [{"id": "file_abc123"}] 503 }) 504 mock_upload_resp.text = AsyncMock(return_value="") 505 mock_upload_resp.__aenter__ = AsyncMock(return_value=mock_upload_resp) 506 mock_upload_resp.__aexit__ = AsyncMock(return_value=False) 507 508 # Mock the post (POST to /posts) 509 mock_post_resp = AsyncMock() 510 mock_post_resp.status = 200 511 mock_post_resp.json = AsyncMock(return_value={"id": "post_with_file"}) 512 mock_post_resp.text = AsyncMock(return_value="") 513 mock_post_resp.__aenter__ = AsyncMock(return_value=mock_post_resp) 514 mock_post_resp.__aexit__ = AsyncMock(return_value=False) 515 516 # Route calls: first GET (download), then POST (upload), then POST (create post) 517 self.adapter._session.get = MagicMock(return_value=mock_dl_resp) 518 post_call_count = 0 519 original_post_returns = [mock_upload_resp, mock_post_resp] 520 521 def post_side_effect(*args, **kwargs): 522 nonlocal post_call_count 523 resp = original_post_returns[min(post_call_count, len(original_post_returns) - 1)] 524 post_call_count += 1 525 return resp 526 527 self.adapter._session.post = MagicMock(side_effect=post_side_effect) 528 529 result = await self.adapter.send_image( 530 "channel_1", "https://img.example.com/cat.png", caption="A cat" 531 ) 532 533 assert result.success is True 534 assert result.message_id == "post_with_file" 535 536 537 # --------------------------------------------------------------------------- 538 # Dedup cache 539 # --------------------------------------------------------------------------- 540 541 class TestMattermostDedup: 542 def setup_method(self): 543 self.adapter = _make_adapter() 544 self.adapter._bot_user_id = "bot_user_id" 545 # Mock handle_message to capture calls without processing 546 self.adapter.handle_message = AsyncMock() 547 548 @pytest.mark.asyncio 549 async def test_duplicate_post_ignored(self): 550 """The same post_id within the TTL window should be ignored.""" 551 post_data = { 552 "id": "post_dup", 553 "user_id": "user_123", 554 "channel_id": "chan_456", 555 "message": "@bot_user_id Hello!", 556 } 557 event = { 558 "event": "posted", 559 "data": { 560 "post": json.dumps(post_data), 561 "channel_type": "O", 562 "sender_name": "@alice", 563 }, 564 } 565 566 # First time: should process 567 await self.adapter._handle_ws_event(event) 568 assert self.adapter.handle_message.call_count == 1 569 570 # Second time (same post_id): should be deduped 571 await self.adapter._handle_ws_event(event) 572 assert self.adapter.handle_message.call_count == 1 # still 1 573 574 @pytest.mark.asyncio 575 async def test_different_post_ids_both_processed(self): 576 """Different post IDs should both be processed.""" 577 for i, pid in enumerate(["post_a", "post_b"]): 578 post_data = { 579 "id": pid, 580 "user_id": "user_123", 581 "channel_id": "chan_456", 582 "message": f"@bot_user_id Message {i}", 583 } 584 event = { 585 "event": "posted", 586 "data": { 587 "post": json.dumps(post_data), 588 "channel_type": "O", 589 "sender_name": "@alice", 590 }, 591 } 592 await self.adapter._handle_ws_event(event) 593 594 assert self.adapter.handle_message.call_count == 2 595 596 def test_prune_seen_clears_expired(self): 597 """Dedup cache should remove entries older than TTL on overflow.""" 598 now = time.time() 599 dedup = self.adapter._dedup 600 # Fill with enough expired entries to trigger pruning 601 for i in range(dedup._max_size + 10): 602 dedup._seen[f"old_{i}"] = now - 600 # 10 min ago (older than default TTL) 603 604 # Add a fresh one 605 dedup._seen["fresh"] = now 606 607 # Trigger pruning by calling is_duplicate with a new entry (over max_size) 608 dedup.is_duplicate("trigger_prune") 609 610 # Old entries should be pruned, fresh one kept 611 assert "fresh" in dedup._seen 612 assert len(dedup._seen) < dedup._max_size + 10 613 614 def test_seen_cache_tracks_post_ids(self): 615 """Posts are tracked in the dedup cache.""" 616 self.adapter._dedup._seen["test_post"] = time.time() 617 assert "test_post" in self.adapter._dedup._seen 618 619 620 # --------------------------------------------------------------------------- 621 # Requirements check 622 # --------------------------------------------------------------------------- 623 624 class TestMattermostRequirements: 625 def test_check_requirements_with_token_and_url(self, monkeypatch): 626 monkeypatch.setenv("MATTERMOST_TOKEN", "test-token") 627 monkeypatch.setenv("MATTERMOST_URL", "https://mm.example.com") 628 from gateway.platforms.mattermost import check_mattermost_requirements 629 assert check_mattermost_requirements() is True 630 631 def test_check_requirements_without_token(self, monkeypatch): 632 monkeypatch.delenv("MATTERMOST_TOKEN", raising=False) 633 monkeypatch.delenv("MATTERMOST_URL", raising=False) 634 from gateway.platforms.mattermost import check_mattermost_requirements 635 assert check_mattermost_requirements() is False 636 637 def test_check_requirements_without_url(self, monkeypatch): 638 monkeypatch.setenv("MATTERMOST_TOKEN", "test-token") 639 monkeypatch.delenv("MATTERMOST_URL", raising=False) 640 from gateway.platforms.mattermost import check_mattermost_requirements 641 assert check_mattermost_requirements() is False 642 643 644 # --------------------------------------------------------------------------- 645 # Media type propagation (MIME types, not bare strings) 646 # --------------------------------------------------------------------------- 647 648 class TestMattermostMediaTypes: 649 """Verify that media_types contains actual MIME types (e.g. 'image/png') 650 rather than bare category strings ('image'), so downstream 651 ``mtype.startswith("image/")`` checks in run.py work correctly.""" 652 653 def setup_method(self): 654 self.adapter = _make_adapter() 655 self.adapter._bot_user_id = "bot_user_id" 656 self.adapter.handle_message = AsyncMock() 657 658 def _make_event(self, file_ids): 659 post_data = { 660 "id": "post_media", 661 "user_id": "user_123", 662 "channel_id": "chan_456", 663 "message": "@bot_user_id file attached", 664 "file_ids": file_ids, 665 } 666 return { 667 "event": "posted", 668 "data": { 669 "post": json.dumps(post_data), 670 "channel_type": "O", 671 "sender_name": "@alice", 672 }, 673 } 674 675 @pytest.mark.asyncio 676 async def test_image_media_type_is_full_mime(self): 677 """An image attachment should produce 'image/png', not 'image'.""" 678 file_info = {"name": "photo.png", "mime_type": "image/png"} 679 self.adapter._api_get = AsyncMock(return_value=file_info) 680 681 mock_resp = AsyncMock() 682 mock_resp.status = 200 683 mock_resp.read = AsyncMock(return_value=b"\x89PNG fake") 684 mock_resp.__aenter__ = AsyncMock(return_value=mock_resp) 685 mock_resp.__aexit__ = AsyncMock(return_value=False) 686 self.adapter._session = MagicMock() 687 self.adapter._session.get = MagicMock(return_value=mock_resp) 688 689 with patch("gateway.platforms.base.cache_image_from_bytes", return_value="/tmp/photo.png"): 690 await self.adapter._handle_ws_event(self._make_event(["file1"])) 691 692 msg = self.adapter.handle_message.call_args[0][0] 693 assert msg.media_types == ["image/png"] 694 assert msg.media_types[0].startswith("image/") 695 696 @pytest.mark.asyncio 697 async def test_audio_media_type_is_full_mime(self): 698 """An audio attachment should produce 'audio/ogg', not 'audio'.""" 699 file_info = {"name": "voice.ogg", "mime_type": "audio/ogg"} 700 self.adapter._api_get = AsyncMock(return_value=file_info) 701 702 mock_resp = AsyncMock() 703 mock_resp.status = 200 704 mock_resp.read = AsyncMock(return_value=b"OGG fake") 705 mock_resp.__aenter__ = AsyncMock(return_value=mock_resp) 706 mock_resp.__aexit__ = AsyncMock(return_value=False) 707 self.adapter._session = MagicMock() 708 self.adapter._session.get = MagicMock(return_value=mock_resp) 709 710 with patch("gateway.platforms.base.cache_audio_from_bytes", return_value="/tmp/voice.ogg"), \ 711 patch("gateway.platforms.base.cache_image_from_bytes"), \ 712 patch("gateway.platforms.base.cache_document_from_bytes"): 713 await self.adapter._handle_ws_event(self._make_event(["file2"])) 714 715 msg = self.adapter.handle_message.call_args[0][0] 716 assert msg.media_types == ["audio/ogg"] 717 assert msg.media_types[0].startswith("audio/") 718 719 @pytest.mark.asyncio 720 async def test_document_media_type_is_full_mime(self): 721 """A document attachment should produce 'application/pdf', not 'document'.""" 722 file_info = {"name": "report.pdf", "mime_type": "application/pdf"} 723 self.adapter._api_get = AsyncMock(return_value=file_info) 724 725 mock_resp = AsyncMock() 726 mock_resp.status = 200 727 mock_resp.read = AsyncMock(return_value=b"PDF fake") 728 mock_resp.__aenter__ = AsyncMock(return_value=mock_resp) 729 mock_resp.__aexit__ = AsyncMock(return_value=False) 730 self.adapter._session = MagicMock() 731 self.adapter._session.get = MagicMock(return_value=mock_resp) 732 733 with patch("gateway.platforms.base.cache_document_from_bytes", return_value="/tmp/report.pdf"), \ 734 patch("gateway.platforms.base.cache_image_from_bytes"): 735 await self.adapter._handle_ws_event(self._make_event(["file3"])) 736 737 msg = self.adapter.handle_message.call_args[0][0] 738 assert msg.media_types == ["application/pdf"] 739 assert not msg.media_types[0].startswith("image/") 740 assert not msg.media_types[0].startswith("audio/")