mirror of
https://github.com/HKUDS/nanobot.git
synced 2026-05-07 10:16:02 +00:00
test(telegram): cover silent allowlist rejection
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
parent
5aa61e08d3
commit
1813fc5021
@ -1309,6 +1309,58 @@ async def test_on_help_includes_restart_command() -> None:
|
||||
assert "/dream-restore" in help_text
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_on_start_ignores_unauthorized_user_silently() -> None:
|
||||
channel = TelegramChannel(
|
||||
TelegramConfig(enabled=True, token="123:abc", allow_from=["999"], group_policy="open"),
|
||||
MessageBus(),
|
||||
)
|
||||
update = _make_telegram_update(text="/start", chat_type="private")
|
||||
update.message.reply_text = AsyncMock()
|
||||
|
||||
await channel._on_start(update, None)
|
||||
|
||||
update.message.reply_text.assert_not_awaited()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_on_help_ignores_unauthorized_user_silently() -> None:
|
||||
channel = TelegramChannel(
|
||||
TelegramConfig(enabled=True, token="123:abc", allow_from=["999"], group_policy="open"),
|
||||
MessageBus(),
|
||||
)
|
||||
update = _make_telegram_update(text="/help", chat_type="private")
|
||||
update.message.reply_text = AsyncMock()
|
||||
|
||||
await channel._on_help(update, None)
|
||||
|
||||
update.message.reply_text.assert_not_awaited()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_on_message_ignores_unauthorized_user_before_side_effects() -> None:
|
||||
channel = TelegramChannel(
|
||||
TelegramConfig(enabled=True, token="123:abc", allow_from=["999"], group_policy="open"),
|
||||
MessageBus(),
|
||||
)
|
||||
channel._app = _FakeApp(lambda: None)
|
||||
started_typing: list[str] = []
|
||||
handled: list[dict] = []
|
||||
channel._start_typing = lambda chat_id: started_typing.append(chat_id)
|
||||
channel._add_reaction = AsyncMock(return_value=None)
|
||||
|
||||
async def capture_handle(**kwargs) -> None:
|
||||
handled.append(kwargs)
|
||||
|
||||
channel._handle_message = capture_handle
|
||||
|
||||
await channel._on_message(_make_telegram_update(text="hello", chat_type="private"), None)
|
||||
|
||||
assert started_typing == []
|
||||
channel._add_reaction.assert_not_awaited()
|
||||
assert handled == []
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_on_message_location_content() -> None:
|
||||
"""Location messages are forwarded as [location: lat, lon] content."""
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user