|
5 | 5 | from unittest.mock import AsyncMock, MagicMock, patch |
6 | 6 |
|
7 | 7 | import pytest |
8 | | -from telegram import Message |
| 8 | +from telegram import Message, MessageEntity, TextQuote |
9 | 9 |
|
10 | 10 | import astrbot.api.message_components as Comp |
11 | 11 | from astrbot.api.platform import Group |
@@ -496,6 +496,120 @@ async def test_telegram_reply_without_quote_text_uses_full_message(quote_text): |
496 | 496 | assert reply.chain[0].text == original_text |
497 | 497 |
|
498 | 498 |
|
| 499 | +@pytest.mark.asyncio |
| 500 | +@pytest.mark.parametrize("in_caption", [False, True], ids=["text", "caption"]) |
| 501 | +async def test_telegram_hidden_links_survive_conversion_and_reply(in_caption): |
| 502 | + """Preserve linked text and URLs in messages and full replies. |
| 503 | +
|
| 504 | + Args: |
| 505 | + in_caption: Whether the linked text is a photo caption. |
| 506 | + """ |
| 507 | + adapter = _load_telegram_adapter()( |
| 508 | + make_platform_config("telegram"), {}, asyncio.Queue() |
| 509 | + ) |
| 510 | + text = "😀 查看文档 和 示例" |
| 511 | + entities = [ |
| 512 | + MessageEntity("bold", 3, 4), |
| 513 | + MessageEntity("text_link", 3, 4, url="https://example.com/docs"), |
| 514 | + MessageEntity("text_link", 10, 2, url="https://example.com/demo"), |
| 515 | + ] |
| 516 | + update = create_mock_update( |
| 517 | + message_text=None if in_caption else text, |
| 518 | + entities=None if in_caption else entities, |
| 519 | + caption=text if in_caption else None, |
| 520 | + caption_entities=entities if in_caption else None, |
| 521 | + photo=[create_mock_file()] if in_caption else None, |
| 522 | + ) |
| 523 | + expected = ( |
| 524 | + "😀 查看文档 (https://example.com/docs) 和 示例 (https://example.com/demo)" |
| 525 | + ) |
| 526 | + |
| 527 | + result = await adapter.convert_message(update, _build_context()) |
| 528 | + assert result.message_str == expected |
| 529 | + assert [part.text for part in result.message if isinstance(part, Comp.Plain)] == [ |
| 530 | + expected |
| 531 | + ] |
| 532 | + |
| 533 | + reply_update = create_mock_update( |
| 534 | + message_text="Which link?", reply_to_message=update.message |
| 535 | + ) |
| 536 | + result = await adapter.convert_message(reply_update, _build_context()) |
| 537 | + reply = result.message[0] |
| 538 | + assert reply.message_str == reply.text == expected |
| 539 | + assert [part.text for part in reply.chain if isinstance(part, Comp.Plain)] == [ |
| 540 | + expected |
| 541 | + ] |
| 542 | + |
| 543 | + |
| 544 | +@pytest.mark.asyncio |
| 545 | +@pytest.mark.parametrize("position", [5, 0], ids=["matched", "stale"]) |
| 546 | +async def test_telegram_partial_quote_preserves_only_overlapping_links(position): |
| 547 | + """Recover a quoted link without including links outside the selected text. |
| 548 | +
|
| 549 | + Args: |
| 550 | + position: UTF-16 quote offset, either matching the source or stale. |
| 551 | + """ |
| 552 | + adapter = _load_telegram_adapter()( |
| 553 | + make_platform_config("telegram"), {}, asyncio.Queue() |
| 554 | + ) |
| 555 | + original = create_mock_update( |
| 556 | + message_text="😀 查看文档 和 示例", |
| 557 | + entities=[ |
| 558 | + MessageEntity("text_link", 3, 4, url="https://example.com/docs"), |
| 559 | + MessageEntity("text_link", 10, 2, url="https://example.com/demo"), |
| 560 | + ], |
| 561 | + ) |
| 562 | + update = create_mock_update( |
| 563 | + reply_to_message=original.message, |
| 564 | + quote=TextQuote(text="文档", position=position), |
| 565 | + ) |
| 566 | + |
| 567 | + result = await adapter.convert_message(update, _build_context()) |
| 568 | + |
| 569 | + expected = "文档 (https://example.com/docs)" if position == 5 else "文档" |
| 570 | + reply = result.message[0] |
| 571 | + assert reply.message_str == reply.text == expected |
| 572 | + assert reply.chain == [Comp.Plain(expected)] |
| 573 | + |
| 574 | + |
| 575 | +@pytest.mark.asyncio |
| 576 | +async def test_telegram_hidden_links_and_mentions_use_original_offsets(): |
| 577 | + """Keep link positions intact when self mentions are removed.""" |
| 578 | + adapter = _load_telegram_adapter()( |
| 579 | + make_platform_config("telegram"), {}, asyncio.Queue() |
| 580 | + ) |
| 581 | + text = "😀 @test_bot 文档 @test_bot https://example.com" |
| 582 | + sdk_message = Message.de_json( |
| 583 | + { |
| 584 | + "message_id": 1, |
| 585 | + "date": 0, |
| 586 | + "chat": {"id": 123, "type": "private"}, |
| 587 | + "text": text, |
| 588 | + "entities": [ |
| 589 | + {"type": "mention", "offset": 3, "length": 9}, |
| 590 | + MessageEntity( |
| 591 | + "text_link", 13, 2, url="https://example.com/docs" |
| 592 | + ).to_dict(), |
| 593 | + {"type": "mention", "offset": 16, "length": 9}, |
| 594 | + {"type": "url", "offset": 26, "length": 19}, |
| 595 | + ], |
| 596 | + }, |
| 597 | + bot=None, |
| 598 | + ) |
| 599 | + update = create_mock_update(message_text=text, entities=sdk_message.entities) |
| 600 | + update.message.parse_entity = sdk_message.parse_entity |
| 601 | + |
| 602 | + result = await adapter.convert_message(update, _build_context()) |
| 603 | + |
| 604 | + assert ( |
| 605 | + result.message_str == "😀 文档 (https://example.com/docs) https://example.com" |
| 606 | + ) |
| 607 | + assert [part.qq for part in result.message if isinstance(part, Comp.At)] == [ |
| 608 | + "test_bot", |
| 609 | + "test_bot", |
| 610 | + ] |
| 611 | + |
| 612 | + |
499 | 613 | @pytest.mark.asyncio |
500 | 614 | @pytest.mark.parametrize( |
501 | 615 | ("text", "entities", "expected", "names"), |
|
0 commit comments