Add live integration tests for the Bot realtime receive path

tests/test_bot_integration.py exercises the event-receive side of the bot
framework against a live server:
- test_bot_receives_its_own_message: subscribe to a room, post into it, and
  assert the message comes back over the /api/realtime stream as a
  BotMessageEvent (actor == the bot).
- test_bot_receives_message_from_another_user: same, but the message is posted
  by a second identity (skipped unless CHATTO_SECOND_LOGIN/PASSWORD are set).

Gated on CHATTO_BOT_KEY (a bot key used directly as a bearer token);
CHATTO_BASE_URL overrides the target server. Uses bot.run(until=...) for a
clean stream shutdown rather than task cancellation.

tests/test_bot_integration.py

@@ -0,0 +1,162 @@
+"""Integration tests for the :class:`~chattolib.bot.Bot` realtime event path.
+
+These run against a live Chatto server and exercise the *receive* side of the
+bot framework end-to-end: open the ``/api/realtime`` WebSocket, subscribe to
+rooms, and confirm that posted messages come back as typed
+:class:`~chattolib.bot.BotMessageEvent` values.
+
+Run with::
+
+    CHATTO_BOT_KEY=cht_BK_... pytest tests/test_bot_integration.py -v
+
+Requires the ``CHATTO_BOT_KEY`` environment variable (a bot key, used
+directly as a bearer token). The optional ``CHATTO_BASE_URL`` overrides the
+target server (defaults to the production ``https://chat.chatto.run``).
+
+Unlike ``test_integration.py`` (which needs a human login/password), a bot
+key is enough here — the bot posts to itself, so no second identity is
+required. The ``test_bot_receives_message_from_another_user`` test is skipped
+unless ``CHATTO_SECOND_LOGIN``/``CHATTO_SECOND_PASSWORD`` are also set.
+"""
+
+import asyncio
+import os
+
+import pytest
+
+from chattolib.bot import Bot, BotMessageEvent
+
+pytestmark = pytest.mark.skipif(
+    not os.environ.get("CHATTO_BOT_KEY"),
+    reason="CHATTO_BOT_KEY env var required",
+)
+
+BASE_URL = os.environ.get("CHATTO_BASE_URL", "https://chat.chatto.run")
+BOT_KEY = os.environ.get("CHATTO_BOT_KEY")
+SENTINEL = "[chattolib-bot-integration] receive-path check"
+
+
+async def _login_bot() -> Bot:
+    assert BOT_KEY is not None  # guaranteed by the module-level skipif
+    return await Bot.login(BOT_KEY, base_url=BASE_URL)
+
+
+async def _pick_member_room_id(bot: Bot) -> str:
+    """Return the id of a room the bot is a member of (joining one if needed)."""
+    rooms = await bot.list_rooms()
+    target = next((r for r in rooms if r.room and r.viewer_state.is_member), None)
+    if target is not None:
+        return target.room.id
+    # No member room yet — join the first room in the first group.
+    groups = await bot.list_room_groups()
+    if not groups or not groups[0].rooms:
+        pytest.skip("bot has no rooms and no room groups to join")
+    first = groups[0].rooms[0].room
+    if first is None:
+        pytest.skip("first room group has no rooms")
+    await bot.join_room(first.id)
+    return first.id
+
+
+async def _wait_for_event(
+    bot: Bot,
+    room_id: str,
+    *,
+    trigger,
+    matches,
+    timeout: float = 30.0,
+) -> list[BotMessageEvent]:
+    """Subscribe to ``room_id``'s timeline, fire ``trigger``, await the event.
+
+    ``trigger`` is an awaitable fired *after* the stream is subscribed (so the
+    posted message is not projected before we're listening). ``matches`` is
+    called with each incoming :class:`BotMessageEvent`; the first call that
+    returns ``True`` ends the wait. Returns the list of matching events.
+
+    Uses ``bot.run(until=...)`` so the stream shuts down cleanly — cancelling
+    the run task directly would propagate into the websocket recv loop.
+    """
+    received: list[BotMessageEvent] = []
+    done = asyncio.Event()
+
+    def on_event(event: BotMessageEvent) -> None:
+        if matches(event):
+            received.append(event)
+            done.set()
+
+    bot.on("message", on_event)
+    stop = asyncio.Event()
+    run_task = asyncio.create_task(bot.run(retained_room_ids=[room_id], until=stop))
+    await asyncio.sleep(2)  # let the stream connect + subscribe
+
+    try:
+        await trigger()
+        await asyncio.wait_for(done.wait(), timeout=timeout)
+    finally:
+        stop.set()  # tell run() to stop looping
+        try:
+            await asyncio.wait_for(run_task, timeout=10)
+        except (TimeoutError, asyncio.CancelledError, Exception):
+            run_task.cancel()
+            try:
+                await run_task
+            except (asyncio.CancelledError, Exception):
+                pass
+    return received
+
+
+async def test_bot_receives_its_own_message():
+    """Post a message and confirm it comes back over the realtime stream.
+
+    This is the core receive-path check: the bot subscribes to a room, posts
+    into it, and the server projects the ``message_posted`` timeline event
+    back to the bot as a ``BotMessageEvent``.
+    """
+    bot = await _login_bot()
+    try:
+        target_id = await _pick_member_room_id(bot)
+        received = await _wait_for_event(
+            bot,
+            target_id,
+            trigger=lambda: bot.say(target_id, SENTINEL),
+            matches=lambda e: e.room_id == target_id and e.body == SENTINEL,
+        )
+        assert len(received) >= 1, "bot did not receive its own posted message"
+        # The echoed event should carry the same body and the bot as actor.
+        assert received[0].body == SENTINEL
+        assert received[0].message.actor_id == bot.user.id
+    finally:
+        await bot.close()
+
+
+async def test_bot_receives_message_from_another_user():
+    """Receive a message posted by a *different* user (needs 2nd identity).
+
+    Skipped unless ``CHATTO_SECOND_LOGIN``/``CHATTO_SECOND_PASSWORD`` are set.
+    """
+    second_login = os.environ.get("CHATTO_SECOND_LOGIN")
+    second_password = os.environ.get("CHATTO_SECOND_PASSWORD")
+    if not second_login or not second_password:
+        pytest.skip("CHATTO_SECOND_LOGIN / CHATTO_SECOND_PASSWORD not set")
+
+    from chattolib.client import ChattoClient
+
+    bot = await _login_bot()
+    try:
+        target_id = await _pick_member_room_id(bot)
+
+        async def trigger() -> None:
+            login = ChattoClient.login(second_login, second_password, base_url=BASE_URL)
+            async with await login as other:
+                await other.post_message(target_id, SENTINEL)
+
+        received = await _wait_for_event(
+            bot,
+            target_id,
+            trigger=trigger,
+            matches=lambda e: e.message.actor_id != bot.user.id and e.room_id == target_id,
+        )
+        assert len(received) >= 1, "bot did not receive the second user's message"
+        assert received[0].message.actor_id != bot.user.id
+    finally:
+        await bot.close()