From: Thorsten Date: Sat, 18 Jul 2026 08:38:19 +0000 (+0200) Subject: Add a live MUC smoke-test tool X-Git-Url: https://git.aero2k.de/?a=commitdiff_plain;h=8af183f0d95111f94ba08d6809e22624f0f5da8f;p=urlbot-v3.git 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 --- diff --git a/tests/muc_smoke_check.py b/tests/muc_smoke_check.py new file mode 100644 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, "")) + 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()