Add a live MUC smoke-test tool

Connects to a real room as a separate test account, addresses the bot by
nick with a few deterministic commands (ping/version/dice), and checks the
responses match expected patterns - a black-box way to confirm a deployed
instance is actually working end to end, not just that its systemd units
are active. Not part of the pytest suite; needs a real XMPP server/room/bot.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

diff --git a/tests/muc_smoke_check.py b/tests/muc_smoke_check.py
new file mode 100644 (file)
index 0000000..31de8f1
--- /dev/null
+++ b/tests/muc_smoke_check.py
@@ -0,0 +1,146 @@
+#!/usr/bin/env python3
+"""
+Black-box system test for a live urlbot-v3 instance.
+
+Connects to a real MUC as a separate test account, addresses the bot by
+its nickname with a handful of deterministic commands, and checks the
+responses against expected patterns. Exits non-zero on the first
+timeout/mismatch. This is a manual tool, not part of the automated pytest
+suite - it needs a real XMPP server, room, and a bot actually running.
+
+Usage:
+    python3 tests/muc_smoke_check.py \
+        --jid testbug@debianforum.de --password '...' \
+        --room schrottplatz@chat.debianforum.de --bot-nick pibug
+"""
+import argparse
+import asyncio
+import re
+import sys
+from dataclasses import dataclass
+from typing import Optional, Pattern
+
+import slixmpp
+
+
+@dataclass
+class Check:
+    command: str
+    expect: Pattern
+    description: str
+    timeout: float = 10.0
+
+
+DEFAULT_CHECKS = [
+    Check("ping", re.compile(r"^pong$"), "basic liveness"),
+    Check("version", re.compile(r"^I'm running"), "version string"),
+    Check("dice", re.compile(r"^rolling a dice for \S+:"), "dice roll (random content, fixed prefix)"),
+]
+
+
+class SmokeTestClient(slixmpp.ClientXMPP):
+    def __init__(self, jid, password, room, nick, bot_nick, checks, verbose=False):
+        super().__init__(jid, password)
+        self.room = room
+        self.nick = nick
+        self.bot_nick = bot_nick
+        self.checks = checks
+        self.verbose = verbose
+        self.results = []
+        self._pending: Optional[asyncio.Future] = None
+
+        self.register_plugin('xep_0045')
+        self.add_event_handler("session_start", self.session_start)
+        self.add_event_handler("groupchat_message", self.on_groupchat_message)
+        self.add_event_handler("message_error", self.on_message_error)
+        self.add_event_handler("groupchat_presence", self.on_groupchat_presence)
+
+    def _debug(self, text):
+        if self.verbose:
+            print(f"[debug] {text}", file=sys.stderr)
+
+    def on_message_error(self, msg):
+        # not gated by verbose - "forbidden: no permission to speak" is the
+        # most common reason this tool silently sees no responses at all.
+        print(f"[error] server rejected our message: {msg['error']['text']}", file=sys.stderr)
+
+    def on_groupchat_presence(self, presence):
+        if presence['muc']['nick'] == self.nick:
+            self._debug(f"my role/affiliation: role={presence['muc']['role']!r} "
+                        f"affiliation={presence['muc']['affiliation']!r}")
+
+    async def session_start(self, _event):
+        try:
+            self.send_presence()
+            await self.get_roster()
+            # maxstanzas=0: skip MUC history replay so old messages can't be
+            # mistaken for a fresh response.
+            result = await self.plugin['xep_0045'].join_muc_wait(self.room, self.nick, maxstanzas=0)
+            self._debug(f"joined {self.room} as {self.nick}: {result[0]}")
+            await asyncio.sleep(1)  # let the join settle before addressing the bot
+
+            for check in self.checks:
+                await self.run_check(check)
+        except Exception as e:
+            print(f"[error] session_start crashed: {e!r}", file=sys.stderr)
+            import traceback
+            traceback.print_exc()
+
+        await self.disconnect()
+        await asyncio.sleep(0.5)  # let slixmpp's internal stream tasks wind down quietly
+        self.loop.stop()
+
+    async def run_check(self, check: Check):
+        loop = asyncio.get_event_loop()
+        self._pending = loop.create_future()
+        self.send_message(mto=self.room, mbody=f"{self.bot_nick}: {check.command}", mtype='groupchat')
+        try:
+            body = await asyncio.wait_for(self._pending, timeout=check.timeout)
+            ok = bool(check.expect.search(body))
+            self.results.append((check, ok, body))
+        except asyncio.TimeoutError:
+            self.results.append((check, False, "<no response within timeout>"))
+        finally:
+            self._pending = None
+
+    def on_groupchat_message(self, msg):
+        self._debug(f"groupchat msg from {msg['mucnick']!r}: {msg['body']!r}")
+        if msg['mucnick'] in (self.nick, ""):
+            return  # ignore our own messages
+        if msg['mucnick'] != self.bot_nick:
+            return  # ignore other room members
+        if self._pending is not None and not self._pending.done():
+            self._pending.set_result(str(msg['body']))
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
+    parser.add_argument("--jid", required=True, help="test account JID, e.g. testbug@debianforum.de")
+    parser.add_argument("--password", required=True)
+    parser.add_argument("--room", required=True, help="MUC JID, e.g. schrottplatz@chat.debianforum.de")
+    parser.add_argument("--nick", default="smoketest", help="this test account's nickname in the room")
+    parser.add_argument("--bot-nick", required=True, help="the bot's nickname to address commands to")
+    parser.add_argument("--verbose", action="store_true", help="print every groupchat message and join details")
+    args = parser.parse_args()
+
+    client = SmokeTestClient(
+        args.jid, args.password, args.room, args.nick, args.bot_nick, DEFAULT_CHECKS, verbose=args.verbose,
+    )
+    client.connect()
+    client.loop.run_forever()
+
+    failures = 0
+    for check, ok, body in client.results:
+        status = "OK" if ok else "FAIL"
+        print(f"[{status}] {check.command!r} ({check.description}): {body!r}")
+        if not ok:
+            failures += 1
+
+    if failures:
+        print(f"\n{failures}/{len(client.results)} checks failed")
+        sys.exit(1)
+    print(f"\nAll {len(client.results)} checks passed")
+
+
+if __name__ == '__main__':
+    main()