Classroom_Desktop/tests/test-classroom-mqtt.sh

#!/bin/bash
set -uo pipefail

# The rules for what the classroom accepts from the broker: room
# identity, the topic contract, and which payloads become notifications.
# No broker involved — these are decisions, not plumbing.
#
#   tests/test-classroom-mqtt.sh

ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"

PYTHONPATH="$ROOT/lib" python3 - <<'PY'
import json
import sys

import classroom_mqtt as m

passed = failed = 0


def check(label, condition, detail=""):
    global passed, failed
    if condition:
        print(f"  ok   {label}")
        passed += 1
    else:
        print(f"  FAIL {label}")
        if detail:
            print(f"       {detail}")
        failed += 1


print("1. the room is the machine")
check("hostname becomes the room", m.room_name({}, "nursery") == "nursery")
check("domain is dropped", m.room_name({}, "k2.church.local") == "k2")
check("case is normalised", m.room_name({}, "Older") == "older")
check("ROOM overrides the hostname",
      m.room_name({"ROOM": "older"}, "box-17") == "older")
check("unsafe characters cannot reach a topic",
      m.room_name({}, "Room 3!!") == "room-3")
check("a nameless machine still has a room",
      m.room_name({}, "") == "unknown")

print()
print("2. the topic contract")
topics = m.notify_topics("nursery")
check("listens to every room", "classroom/notify/all" in topics)
check("listens to its own room", "classroom/notify/nursery" in topics)
check("status is per room", m.status_topic("nursery") == "classroom/status/nursery")
check("no product name in any topic",
      not any("ekklesia" in t for t in topics + (m.status_topic("nursery"),)))

print()
print("3. a classroom without a broker stays quiet")
check("disabled means no broker", m.broker({"MQTT_HOST": "h"}) is None)
check("enabled with no host means no broker",
      m.broker({"MQTT_ENABLED": "true"}) is None)
check("no config at all means no broker", m.broker({}) is None)
settings = m.broker({"MQTT_ENABLED": "yes", "MQTT_HOST": "10.40.0.2"})
check("host is taken", settings and settings["host"] == "10.40.0.2")
check("port defaults to 1883", settings and settings["port"] == 1883)
settings = m.broker({"MQTT_ENABLED": "1", "MQTT_HOST": "h", "MQTT_PORT": "not a port"})
check("a nonsense port falls back rather than crashing",
      settings and settings["port"] == 1883)

print()
print("4. which messages become notifications")

now = 1_780_000_000


def payload(**fields):
    body = {"title": "Sermon Starting", "message": "The sermon has started."}
    body.update(fields)
    return json.dumps(body)


n = m.Notifications()
note, reason = n.accept(payload(id="a1"), now=now)
check("a good message is accepted", reason == "ok", reason)
check("title survives", note and note["title"] == "Sermon Starting")
check("message survives", note and note["message"] == "The sermon has started.")
check("urgency defaults to normal", note and note["urgency"] == "normal")

note, reason = n.accept(payload(id="a2", urgency="critical"), now=now)
check("critical is honoured", note and note["urgency"] == "critical")
note, reason = n.accept(payload(id="a3", urgency="screaming"), now=now)
check("a nonsense urgency falls back to normal",
      note and note["urgency"] == "normal")

print()
print("5. what gets dropped, without anything falling over")

n = m.Notifications()
for label, args, expected in [
    ("a retained notification is stale by definition",
     {"payload": payload(id="r"), "retained": True}, "retained"),
    ("not JSON", {"payload": "{{{"}, "not json"),
    ("a JSON array is not a notification", {"payload": "[1,2,3]"}, "not an object"),
    ("no title", {"payload": json.dumps({"message": "x"})}, "no title"),
    ("a blank title", {"payload": json.dumps({"title": "   "})}, "no title"),
    ("a title that is not text",
     {"payload": json.dumps({"title": 7})}, "no title"),
    ("a message that is not text",
     {"payload": json.dumps({"title": "t", "message": 7})}, "message is not text"),
    ("an expiry that is not a time",
     {"payload": payload(expires_at="soon")}, "expires_at is not a time"),
    ("an id that is not text", {"payload": payload(id=7)}, "id is not text"),
    ("invalid utf-8", {"payload": b"\xff\xfe"}, "not utf-8"),
]:
    note, reason = n.accept(now=now, **args)
    check(label, note is None and reason == expected, f"got {reason}")

print()
print("6. expiry and duplicates")

n = m.Notifications()
note, reason = n.accept(payload(id="old", expires_at=now - 1), now=now)
check("an expired message is ignored", note is None and reason == "expired")
note, reason = n.accept(payload(id="future", expires_at=now + 60), now=now)
check("a live message is accepted", reason == "ok")
note, reason = n.accept(payload(id="future"), now=now)
check("the same id is not shown twice", note is None and reason == "duplicate")
note, reason = n.accept(payload(id="different"), now=now)
check("a different id still gets through", reason == "ok")
note, reason = n.accept(payload(), now=now)
check("a message with no id is not deduplicated", reason == "ok")
note, reason = n.accept(payload(), now=now)
check("and still is not", reason == "ok")

n = m.Notifications(remember=4)
for index in range(10):
    n.accept(payload(id=f"m{index}"), now=now)
check("the id history stays bounded", len(n.seen) == 4, f"{len(n.seen)} kept")
note, reason = n.accept(payload(id="m0"), now=now)
check("an id forgotten long ago can come round again", reason == "ok")

print()
print(f"passed: {passed}  failed: {failed}")
sys.exit(1 if failed else 0)
PY