Classroom_Desktop/tests/test-classroom-parent-call.sh

#!/bin/bash
set -uo pipefail

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

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

from classroom_parent_call import (
    ERROR, EXPIRED, MODE_APPROVAL, MODE_AUTO, MODE_OFF, REJECTED,
    SENT_TO_SANCTUARY, SUBMITTED, UNAVAILABLE, WAITING_APPROVAL, WORDING,
    Topics, availability, clean_child, new_request_id, read_coordinator,
    read_mode, read_status, request_payload,
)

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 coordinator's topics, exactly")
t = Topics()
check("root is the church's, with no product name",
      t.root == "classroom/parent-call")
check("mode", t.mode == "classroom/parent-call/mode")
check("coordinator", t.coordinator == "classroom/parent-call/coordinator")
check("request", t.request == "classroom/parent-call/request")
check("status is per request",
      t.status("nursery-8f21c3") == "classroom/parent-call/status/nursery-8f21c3")
check("a different root carries through",
      Topics("test/pc").request == "test/pc/request")
check("a trailing slash does not double up",
      Topics("test/pc/").mode == "test/pc/mode")

print()
print("2. a request is four fields and nothing else")
body = request_payload("nursery-8f21c3", "nursery", "Emma", now=1780000000)
check("exactly the four the coordinator reads",
      set(body) == {"id", "room", "child", "created_at"}, sorted(body))
check("created_at is a number", isinstance(body["created_at"], int))
check("it round-trips as json", json.loads(json.dumps(body)) == body)

print()
print("3. ids the coordinator will accept")
check("room-scoped", new_request_id("nursery", "8f21c3") == "nursery-8f21c3")
check("random by default, and different each time",
      new_request_id("k2") != new_request_id("k2"))
check("no topic syntax can get into one",
      "/" not in new_request_id("older") and "#" not in new_request_id("older"))
check("two rooms cannot collide on a fixed part",
      new_request_id("k2", "aaa") != new_request_id("older", "aaa"))

print()
print("4. the name the teacher typed is the name that is sent")
check("ordinary name", clean_child("Emma") == "Emma")
check("outer space is trimmed", clean_child("  Emma  ") == "Emma")
check("inner runs collapse", clean_child("Emma   Grace") == "Emma Grace")
check("a newline becomes a space, not a weld",
      clean_child("Emma\nGrace") == "Emma Grace")
check("a zero width joiner cannot hide in it",
      clean_child("Em‍ma") == "Em ma")
check("empty is not a name", clean_child("   ") is None)
check("nothing is not a name", clean_child(None) is None)
check("longer than the coordinator accepts is refused here first",
      clean_child("E" * 41) is None)
check("exactly the limit is fine", clean_child("E" * 40) == "E" * 40)
check("accents normalise to one form",
      clean_child("José") == "José")

print()
print("5. reading the mode")
for raw, expected in [(b"auto", MODE_AUTO), ("approval", MODE_APPROVAL),
                      (" off\n", MODE_OFF), ("AUTO", MODE_AUTO)]:
    check(f"{raw!r} -> {expected}", read_mode(raw) == expected)
check("anything else is not a mode", read_mode("banana") is None)
check("an empty retained mode is not a mode", read_mode("") is None)

print()
print("6. is the sanctuary end up")
check("online", read_coordinator('{"online": true}') is True)
check("explicitly offline", read_coordinator('{"online": false}') is False)
check("a cleared retained topic is not up", read_coordinator("") is False)
check("junk is not up", read_coordinator("yes please") is False)
check("a bare true is not the contract", read_coordinator("true") is False)

print()
print("7. statuses, as the coordinator sends them")
def status(**kw):
    body = {"id": "nursery-8f21c3", "room": "nursery",
            "status": SUBMITTED, "message": "Sending", "updated_at": 1}
    body.update(kw)
    return json.dumps(body)

seen = read_status(status(status=WAITING_APPROVAL, message="Waiting for approval"))
check("waiting for approval is not terminal",
      seen["status"] == WAITING_APPROVAL and not seen["terminal"])
check("its wording is the coordinator's, verbatim",
      seen["message"] == "Waiting for approval")

seen = read_status(status(status=SENT_TO_SANCTUARY, message="Sent to sanctuary"))
check("sent to sanctuary is terminal and good",
      seen["terminal"] and seen["good"])
check("and it never says displayed",
      "displayed" not in seen["message"].lower())

for bad in (REJECTED, EXPIRED, ERROR, UNAVAILABLE):
    seen = read_status(status(status=bad))
    check(f"{bad} is terminal but not good",
          seen["terminal"] and not seen["good"])

check("an unknown status is not shown",
      read_status(status(status="displayed")) is None)
check("json that is not an object is not a status",
      read_status("[1,2,3]") is None)
check("junk is not a status", read_status("not json") is None)
check("a status with no usable message falls back to the coordinator's word",
      read_status(status(message=""))["message"] == WORDING[SUBMITTED])
check("another room's request is never shown here",
      read_status(status(id="k2-ffffff"), request_id="nursery-8f21c3") is None)
check("our own request is", 
      read_status(status(), request_id="nursery-8f21c3")["status"] == SUBMITTED)
check("a detail is carried but bounded",
      len(read_status(status(detail="x" * 500))["detail"]) == 200)

print()
print("8. when a teacher may send at all")
can, why = availability(MODE_AUTO, True)
check("auto with the coordinator up", can and why is None)
can, why = availability(MODE_APPROVAL, True)
check("approval with the coordinator up", can and why is None)

can, why = availability(MODE_OFF, True)
check("off cannot send", not can)
check("and says so in the coordinator's own words",
      why == WORDING[UNAVAILABLE], why)

can, why = availability(MODE_AUTO, False)
check("the coordinator being down stops it", not can)
check("and does not pretend a call would arrive",
      "not running" in why, why)

can, why = availability(MODE_AUTO, True, broker_reachable=False)
check("no broker stops it", not can)
check("and says the network, not the sanctuary",
      "network" in why, why)

# No retained mode on the broker. The coordinator publishes one only when
# an operator writes to …/mode/set, and on the church's broker nobody
# ever has — so every room was told it had never been set up.
can, why = availability(None, True)
check("an undecided mode is not yet a yes", not can)
check("and is not a refusal either", why is None, why)

can, why = availability(None, True, assume_default=True)
check("once the retained topics have had their chance, the call goes",
      can)
check("with nothing to say about it", why is None, why)

can, why = availability(MODE_OFF, True, assume_default=True)
check("a mode that really is off still stops it", not can)
check("in the sanctuary's own words", why == WORDING[UNAVAILABLE], why)

can, why = availability(None, False, assume_default=True)
check("and a coordinator that is down still stops it", not can)
check("naming the sanctuary, not this room", "sanctuary" in why, why)

can, why = availability(None, True, broker_reachable=False,
                        why="not configured")
check("a room with no broker configured is still its own answer", not can)
check("and says this room, not the network",
      "no church network set up" in why, why)

print()
print(f"{passed} passed, {failed} failed")
raise SystemExit(1 if failed else 0)
PY