Black-box system test for a live urlbot-v3 instance.
Connects to a real MUC as a separate test account, addresses the bot with
-a handful of commands (some routed the normal "nick: command" way, some
-via free-form trigger phrases), 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.
+a wide range of commands (some routed the normal "nick: command" way, some
+via free-form trigger phrases, some multi-step stateful workflows), 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.
The default checks deliberately include a few "complex edge" plugins whose
binding-key wildcard shapes are the trickiest to get right under the NATS
backend (see distbot.common.routing): a leading '#' (Doctor), a
mid-pattern '#' (Selfreaction), and a scheduled/durable action that fires
without a new incoming message at all (TeaTimer) - exactly the mechanisms
-this migration had to get right.
+this migration had to get right. It also covers the stateful multi-step
+plugins (Voting, VotePoll, Recorder) via dedicated workflows rather than
+the generic single-command Check list.
+
+Deliberately NOT covered, by design (see git history for the reasoning):
+- DidYouKnow: permanently appends to a JSON file on the live host every
+ run, with no easy cleanup.
+- Searx: single hardcoded backend with a worst-case ~17 minute retry
+ loop if it's down; not worth the risk of a hung test run.
+- Translator's "translate show" (goes out as a private message, not a
+ groupchat reply - this harness only listens to groupchat) and
+ "translate that" (needs room history and a detectlanguage API key that
+ probably isn't configured).
+- Bare "vote <n>" and "droppoll": vote casts are silent successes (no
+ reply to wait for) and droppoll is sudoers-gated - neither is
+ practically assertable, so they're exercised indirectly instead (a
+ cast vote is confirmed via the following close-voting tally).
Usage:
python3 tests/muc_smoke_check.py \
import asyncio
import re
import sys
+import time
from dataclasses import dataclass
from typing import List, Optional, Pattern
command: Optional[str] = None # sent as "{bot_nick}: {command}"
raw_message: Optional[str] = None # sent verbatim (for non-addressed triggers)
timeout: float = 10.0
- # if neither command nor raw_message is set, nothing is sent - used for
- # a scheduled follow-up (e.g. TeaTimer firing) that arrives unprompted.
+ no_response: bool = False # fire-and-forget; always reported OK
+ # if command, raw_message, and no_response are all unset/False, nothing
+ # is sent - used to wait for a scheduled follow-up (e.g. TeaTimer firing).
+
+
+ANY = re.compile(r"(?s).*") # matches anything, including empty/multiline - "bot replied at all"
def default_checks(nick: str, bot_nick: str) -> List[Check]:
nick_re = re.escape(nick)
+
return [
+ # --- basics ---
Check("basic liveness", re.compile(r"^pong$"), command="ping"),
Check("version string", re.compile(r"^I'm running"), command="version"),
- Check("dice roll (random content, fixed prefix)",
- re.compile(r"^rolling a dice for \S+:"), command="dice"),
+ Check("info text", re.compile(r"^: I'm a bot named"), command="info"),
+ Check("source url", re.compile(r"^My source code can be found at "), command="source"),
+ Check("uptime/request counter", re.compile(rf"^{nick_re}: happily serving for [\d.]+ seconds?, \d+ requests? so far$"),
+ command="uptime"),
+ Check("url blacklist listing", re.compile(r"^URLs blacklisted: "), command="blacklist"),
+
+ # --- randomness/games, still deterministic prefixes ---
+ Check("dice roll", re.compile(r"^rolling a dice for \S+:"), command="dice"),
+ Check("d20 roll", re.compile(r"^rolling.*for \S+:"), command="d20"),
+ Check("coin flip (needs a word between 'nick' and 'coin' - bare 'coin' hits the "
+ "known mid-pattern '#' gap documented in routing.py)",
+ re.compile(rf"^{nick_re}: (head|tails)$"), command="flip coin"),
+ Check("8ball", re.compile(rf"^{nick_re}: .+!$"), command="8ball will it work"),
+ Check("xchoose (single-branch groups -> deterministic result)",
+ re.compile(rf"^{nick_re}: taco vegan$"), command="xchoose taco (vegan)"),
+ Check("choose, no OpenAI key configured -> deterministic fallback phrase",
+ re.compile(rf"^{nick_re}: .+"), command="choose"),
+ Check("sudo choose, test account isn't in sudoers",
+ re.compile(r"is not in the sudoers file"), command="sudo choose"),
+ Check("pray", re.compile(r"^ok!$"), command="pray for successful tests"),
+ Check("excuse (BOFH)", re.compile(rf"^{nick_re}: .+"), command="excuse"),
+ Check("klammer", re.compile(rf"^{nick_re},"), command="klammer"),
+ Check("terminate (confirmed harmless joke reply, no actual shutdown)",
+ re.compile(r"^insufficient power supply,"), command="terminate"),
+ Check("unicode art", re.compile(rf"^{nick_re}, here's some"), command="unicode"),
+ Check("slap", re.compile(r"^/me slaps "), command="slap smoketest-target"),
+ Check("morse encode", re.compile(rf"^{nick_re}: \S"), command="morse-encode hi"),
+
+ # --- complex edges: the trickiest binding-key wildcard shapes ---
Check(
"leading '#' wildcard, zero-words-before case (Doctor: '#.doctor.#' must match "
"'doctor' as the literal first word, same as real AMQP - this exact shape was broken "
re.compile(rf"^{nick_re}: .+"),
raw_message=f"me thinks {bot_nick} is quite alright",
),
+ Check(
+ "leading '#' catch-all, no bot address at all (MentalDeficits: fires on 3+ '?'/'!' anywhere)",
+ re.compile(r"Multiple exclamation/question marks"),
+ raw_message="does this really work??? really???",
+ ),
+ Check(
+ "catch-all with dynamic content interpolated into the binding key itself "
+ "(SecurityTracker: builds its own URL from a CVE id found in free text)",
+ re.compile(r"^https://security-tracker\.debian\.org/tracker/CVE-2021-1234$"),
+ raw_message="did you see CVE-2021-1234 yet",
+ ),
+
+ # --- scheduled/durable action: no new message prompts the second reply ---
Check("scheduled action, phase 1/2: immediate confirmation",
re.compile(rf"^{nick_re}: Tea timer set to"), command="teatimer 3"),
Check(
re.compile(rf"^{nick_re}: Your tea is ready!$"),
timeout=8.0,
),
+
+ # --- network-dependent: loose patterns, generous timeouts ---
+ Check("wikipedia lookup", re.compile(rf"^{nick_re}: "), command="wp Debian", timeout=20.0),
+ Check("duckduckgo instant answer (either a real abstract or the documented no-results reply)",
+ re.compile(r".+"), command="ducksearch python programming language", timeout=20.0),
+ Check("giphy-backed reaction (Consumables, 'please' branch)",
+ re.compile(r"^cake for \S+: "), command="cake please", timeout=20.0),
+ Check("mymemory translation (single word sidesteps a words[2:]-as-list bug)",
+ re.compile(r"^translation: "), command="translate en|de hello", timeout=20.0),
+ Check("isdown check (regression coverage for a fixed off-by-one: parse_body used to read "
+ "words[0], the command word itself, instead of words[1], the actual target)",
+ re.compile(rf"^{nick_re}: "), command="isdown debian.org", timeout=20.0),
+ Check("youtube oEmbed title (stable, long-lived video id also used in this repo's own unit tests)",
+ ANY, raw_message="https://www.youtube.com/watch?v=H27VcmHVRaw", timeout=20.0),
+ Check("URLResolver <title> scrape (stable, non-blacklisted URL)",
+ ANY, raw_message="http://debianforum.de", timeout=20.0),
]
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
+ await self._join(self.nick)
for check in self.checks:
await self.run_check(check)
+
+ for workflow in (self.test_voting, self.test_votepoll, self.test_recorder):
+ try:
+ await workflow()
+ except Exception as e:
+ print(f"[error] {workflow.__name__} crashed: {e!r}", file=sys.stderr)
+ import traceback
+ traceback.print_exc()
+ self.results.append((f"{workflow.__name__} (crashed, see stderr)", False, repr(e)))
except Exception as e:
print(f"[error] session_start crashed: {e!r}", file=sys.stderr)
import traceback
await asyncio.sleep(0.5) # let slixmpp's internal stream tasks wind down quietly
self.loop.stop()
- async def run_check(self, check: Check):
- # drop any stale messages left over from a previous check (e.g. a
- # slow catch-all plugin reply that arrived late) so it can't be
- # mistaken for this check's response.
+ async def _join(self, nick):
+ # 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, nick, maxstanzas=0)
+ self.nick = nick
+ self._debug(f"joined {self.room} as {nick}: {result[0]}")
+ await asyncio.sleep(1) # let the join settle before addressing the bot
+
+ def _drain_queue(self):
while not self._queue.empty():
self._queue.get_nowait()
- if check.command is not None:
- self.send_message(mto=self.room, mbody=f"{self.bot_nick}: {check.command}", mtype='groupchat')
- elif check.raw_message is not None:
- self.send_message(mto=self.room, mbody=check.raw_message, mtype='groupchat')
- # else: nothing to send - just wait for a spontaneous match (e.g. a scheduled event firing)
-
+ async def wait_for_match(self, expect: Pattern, timeout: float) -> tuple[bool, str]:
loop = asyncio.get_event_loop()
- deadline = loop.time() + check.timeout
+ deadline = loop.time() + timeout
while True:
remaining = deadline - loop.time()
if remaining <= 0:
- self.results.append((check, False, "<no matching response within timeout>"))
- return
+ return False, "<no matching response within timeout>"
try:
body = await asyncio.wait_for(self._queue.get(), timeout=remaining)
except asyncio.TimeoutError:
- self.results.append((check, False, "<no matching response within timeout>"))
- return
- if check.expect.search(body):
- self.results.append((check, True, body))
- return
+ return False, "<no matching response within timeout>"
+ if expect.search(body):
+ return True, body
self._debug(f"ignoring non-matching reply while waiting: {body!r}")
+ def send_to_room(self, body: str):
+ self.send_message(mto=self.room, mbody=body, mtype='groupchat')
+
+ async def run_check(self, check: Check):
+ # drop any stale messages left over from a previous check (e.g. a
+ # slow catch-all plugin reply that arrived late) so it can't be
+ # mistaken for this check's response.
+ self._drain_queue()
+
+ if check.command is not None:
+ self.send_to_room(f"{self.bot_nick}: {check.command}")
+ elif check.raw_message is not None:
+ self.send_to_room(check.raw_message)
+ # else: nothing to send - just wait for a spontaneous match (e.g. a scheduled event firing)
+
+ if check.no_response:
+ self.results.append((check.description, True, "<fire-and-forget, not checked>"))
+ return
+
+ ok, body = await self.wait_for_match(check.expect, check.timeout)
+ self.results.append((check.description, ok, body))
+
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 other room members
self._queue.put_nowait(str(msg['body']))
+ # --- stateful multi-step workflows, not expressible as a flat Check list ---
+
+ async def test_voting(self):
+ """Voting (vote.py) ships with a pre-seeded, permanently-running joke
+ election. Ends it (confirmed acceptable), starts a fresh one, casts a
+ vote (which is a silent success - no reply to wait for), and confirms
+ the vote actually landed via the closing tally."""
+ self._drain_queue()
+ self.send_to_room(f"{self.bot_nick}: close-voting")
+ ok, body = await self.wait_for_match(ANY, 10.0)
+ self.results.append(("Voting: close the pre-seeded joke election", ok, body))
+ if not ok:
+ return
+
+ subject = f"smoketest-{int(time.time())}"
+ self._drain_queue()
+ self.send_to_room(f"{self.bot_nick}: new-voting {subject} optionA optionB")
+ ok, body = await self.wait_for_match(re.compile(r"^Voting started\. Please respond with any of "), 10.0)
+ self.results.append(("Voting: start a fresh poll", ok, body))
+ if not ok:
+ return
+
+ self._drain_queue()
+ self.send_to_room("vote 1") # bare, no bot-nick prefix, matches "vote.*"
+ await asyncio.sleep(2) # silent success - just give it a moment to land, nothing to wait for
+
+ self.send_to_room(f"{self.bot_nick}: close-voting")
+ ok, body = await self.wait_for_match(
+ re.compile(rf"^Voting '{re.escape(subject)}'.*optionA: 1", re.DOTALL), 10.0,
+ )
+ self.results.append(("Voting: cast vote reflected in closing tally", ok, body))
+
+ async def test_votepoll(self):
+ """VotePoll (votepoll.py) only allows one active poll globally. If a
+ real user already has one running, don't touch it - skip the rest of
+ this workflow rather than end someone else's poll early."""
+ self._drain_queue()
+ self.send_to_room(f"{self.bot_nick}: pollstatus")
+ ok, body = await self.wait_for_match(ANY, 10.0)
+ if not ok:
+ self.results.append(("VotePoll: check for a pre-existing active poll", False, body))
+ return
+ if "No active poll." not in body:
+ self.results.append((
+ "VotePoll: workflow skipped - a real poll is already active, not touching it",
+ True, body,
+ ))
+ return
+ self.results.append(("VotePoll: confirmed no poll already active", True, body))
+
+ option_a, option_b = f"smoketest-a-{int(time.time())}", f"smoketest-b-{int(time.time())}"
+ self._drain_queue()
+ self.send_to_room(f"{self.bot_nick}: poll {option_a} vs {option_b} 30")
+ ok, body = await self.wait_for_match(re.compile(r"^\*\*New Vote:\*\* "), 10.0)
+ self.results.append(("VotePoll: start a poll", ok, body))
+ if not ok:
+ return
+
+ self._drain_queue()
+ self.send_to_room(f"{self.bot_nick}: vote a")
+ ok, body = await self.wait_for_match(re.compile(r"^Vote added\.$"), 10.0)
+ self.results.append(("VotePoll: cast a vote", ok, body))
+
+ self._drain_queue()
+ self.send_to_room(f"{self.bot_nick}: endpoll")
+ ok, body = await self.wait_for_match(
+ re.compile(rf"^\*\*Poll status:\*\*(?=.*{re.escape(option_a)}: 1)", re.DOTALL), 10.0,
+ )
+ self.results.append(("VotePoll: end poll, vote reflected in tally", ok, body))
+
+ async def test_recorder(self):
+ """Recorder (muc.py) queues a message for an offline user and is
+ supposed to deliver it when that user's nick joins the room - tests
+ this by actually leaving and rejoining under the target nick.
+
+ NOTE: Recorder's "userjoin.*" binding-key entry is itself dead code
+ - the real routing key is "userjoin.<full room JID>/<nick>", and
+ since a real XMPP domain always contains dots, '*' (exactly one
+ dot-free word) can never match it. But this doesn't actually break
+ anything: Recorder's binding_keys also include Worker.CATCH_ALL
+ ("#"), so it receives every message regardless, and its own
+ `if cmd[0] == "userjoin"` check only looks at the first token -
+ confirmed working end to end live, despite the dead binding-key
+ entry.
+ """
+ target_nick = f"smoketest-target-{int(time.time())}"
+ self._drain_queue()
+ self.send_to_room(f"{self.bot_nick}: record {target_nick} a message from the smoke test")
+ ok, body = await self.wait_for_match(re.compile(rf"^Message saved for {re.escape(target_nick)}$"), 10.0)
+ self.results.append(("Recorder: queue an offline message", ok, body))
+ if not ok:
+ return
+
+ original_nick = self.nick
+ self._drain_queue()
+ self.plugin['xep_0045'].leave_muc(self.room, original_nick) # not a coroutine, don't await
+ await asyncio.sleep(1)
+ await self._join(target_nick)
+
+ ok, body = await self.wait_for_match(
+ re.compile(rf"^{re.escape(target_nick)}, there is 1 message for you:"), 10.0,
+ )
+ self.results.append((
+ "Recorder: message delivered on join (works despite a dead 'userjoin.*' "
+ "binding-key entry - see docstring)",
+ ok, body,
+ ))
+
+ self.plugin['xep_0045'].leave_muc(self.room, target_nick) # not a coroutine, don't await
+ await asyncio.sleep(1)
+ await self._join(original_nick)
+
def main():
parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
client.loop.run_forever()
failures = 0
- for check, ok, body in client.results:
+ for description, ok, body in client.results:
status = "OK" if ok else "FAIL"
- print(f"[{status}] {check.description}: {body!r}")
+ print(f"[{status}] {description}: {body!r}")
if not ok:
failures += 1
projects / urlbot-v3.git / commitdiff