UmbrelApps/tests/test_relay_agent.py

"""Toetst de agent van Evolu Relay: de labels en het tijdvenster.

Waarom deze test bestaat. De agent van deze app beslist niets over toegang, en dat
is precies waarom hij tot 30-08-2026 ongetoetst kon blijven: hij las een bestand en
legde opdrachten in een postbus, en het beleid eromheen staat in `policy.js` met
`tests/test_limiter.mjs` erop. Met deze release doet hij twee dingen erbij die
eigen logica hebben:

- **hij schrijft zelf een bestand**, `labels.json`, en is daar de enige schrijver
  van. Dat is de eerste plek waar de agent staat bijhoudt in plaats van doorgeeft,
  en labelen is lezen-wijzigen-schrijven, dus er zit een slot om;
- **hij rekent het tijdvenster uit** dat de pagina toont. Een verlopen venster moet
  hij als dicht rapporteren, ook al staat er in owners.json nog dat de leerstand
  aan is: het relay-proces ruimt dat op zijn eigen moment op.

Wat deze test NIET dekt: de HTTP-laag. De handlers zitten in een
BaseHTTPRequestHandler en die is zonder socket niet aan te roepen; wat eronder
hangt, de validatie en het schrijven, is hier wél getoetst. De pagina zelf blijft
handwerk in een browser.

Draaien:

    python tests/test_relay_agent.py

De test laadt `agent.py.template` rechtstreeks. Dat kan omdat dat bestand geen
accolade-variabelen bevat en de invulling door umbreld hem dus onveranderd laat;
de eerste toets hieronder controleert precies dat.
"""

import sys

# Vóór de imports, want anders is het te laat: Python legt bytecode naast
# agent.py.template zodra die geïmporteerd wordt, en die rommel hoort niet in de
# app-map. Een keer is zo'n .pyc meegegaan in een commit.
sys.dont_write_bytecode = True

import importlib.machinery  # noqa: E402
import importlib.util  # noqa: E402
import json  # noqa: E402
import os  # noqa: E402
import tempfile  # noqa: E402
from datetime import datetime, timedelta, timezone  # noqa: E402

HERE = os.path.dirname(os.path.abspath(__file__))
APP = os.path.join(HERE, os.pardir, "whatsnext-evolu-relay")
TEMPLATE = os.path.join(APP, "agent.py.template")


def load_agent(state_dir):
    """Laadt agent.py.template als module, met zijn staat in een tijdelijke map.

    De agent leest `RELAY_STATE_DIR` op moduleniveau, dus die moet vóór het laden
    in de omgeving staan. Met een expliciete loader, want importlib kijkt normaal
    naar de extensie en .template staat daar niet tussen.
    """
    os.environ["RELAY_STATE_DIR"] = state_dir
    loader = importlib.machinery.SourceFileLoader("relay_agent", TEMPLATE)
    spec = importlib.util.spec_from_file_location("relay_agent", TEMPLATE, loader=loader)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


class Uitslag:
    def __init__(self):
        self.goed = 0
        self.fout = []

    def check(self, naam, gelukt, uitleg=""):
        if gelukt:
            self.goed += 1
        else:
            self.fout.append(naam + ((" - " + uitleg) if uitleg else ""))

    def rapport(self):
        print()
        print("%d goed, %d fout" % (self.goed, len(self.fout)))
        for f in self.fout:
            print("  FOUT: " + f)
        return 0 if not self.fout else 1


def iso(seconden_vanaf_nu):
    """Een tijdstip in de vorm die het relay-proces schrijft: ISO, met een Z."""
    when = datetime.now(timezone.utc) + timedelta(seconds=seconden_vanaf_nu)
    return when.isoformat().replace("+00:00", "Z")


def schrijf_owners(state_dir, staat):
    with open(os.path.join(state_dir, "owners.json"), "w", encoding="utf-8") as f:
        json.dump(staat, f)


# ── De aanname waar deze hele test op rust ───────────────────────────────────

def test_template_is_invulbaar_zonder_schade(u):
    with open(TEMPLATE, "r", encoding="utf-8") as f:
        inhoud = f.read()
    u.check("template bevat geen dollartekens",
            "$" not in inhoud,
            "umbreld zou die invullen en de Python-code slopen")


# ── Het tijdvenster ─────────────────────────────────────────────────────────

def test_venster_dat_nog_loopt(agent, u):
    feiten = agent.learning_facts({"learning": True, "learningUntil": iso(90)})
    u.check("een lopend venster leest als open", feiten["learning"] is True)
    u.check("met een resterende tijd", isinstance(feiten["learningSecondsLeft"], int))
    u.check("die ergens rond de negentig seconden ligt",
            85 <= feiten["learningSecondsLeft"] <= 90,
            "kreeg %r" % feiten["learningSecondsLeft"])


def test_venster_dat_verlopen_is(agent, u):
    """De toets die er het meest toe doet aan deze kant.

    In het bestand staat `learning: true` en een tijdstip dat voorbij is. Het
    relay-proces ruimt dat op in zijn lus van twee seconden, dus er is een moment
    waarop het bestand nog "aan" zegt terwijl de deur dicht is. De pagina hoort dan
    "closed" te tonen en geen open deur.
    """
    feiten = agent.learning_facts({"learning": True, "learningUntil": iso(-30)})
    u.check("een verlopen venster leest als dicht", feiten["learning"] is False)
    u.check("met nul seconden over", feiten["learningSecondsLeft"] == 0)
    u.check("en zonder tijdstip", feiten["learningUntil"] is None)


def test_venster_zonder_tijdslot(agent, u):
    feiten = agent.learning_facts({"learning": True, "learningUntil": None})
    u.check("open zonder tijdslot blijft open", feiten["learning"] is True)
    u.check("en heeft geen teller", feiten["learningSecondsLeft"] is None)


def test_onleesbaar_tijdslot(agent, u):
    """Onleesbaar hoort niet als "voor altijd open" te lezen, maar wel als open.

    Anders dan bij `policy.js`, en dat is met opzet: dit is de weergave en niet de
    beslissing. Het relay-proces beslist, en dáár is een onleesbaar tijdstip dicht.
    De pagina die iets anders toont dan het bestand zegt zou verwarrender zijn dan
    een pagina die de leerstand toont die er staat.
    """
    for onzin in ["morgen", "", 42, {}, []]:
        feiten = agent.learning_facts({"learning": True, "learningUntil": onzin})
        u.check("een onleesbaar tijdslot valt terug op de leerstand zelf: %r" % (onzin,),
                feiten["learning"] is True and feiten["learningSecondsLeft"] is None)


def test_leerstand_onbekend(agent, u):
    """Onbekend is geen synonym voor uit.

    Ontbreekt owners.json, dan is er nog nooit iets geschreven en weet niemand wat
    de leerstand is. Uit is een keuze, onbekend is een reden om te kijken, en de
    pagina hoort dat verschil te tonen.
    """
    feiten = agent.learning_facts({})
    u.check("een ontbrekende leerstand blijft onbekend", feiten["learning"] is None)
    u.check("en heeft geen teller", feiten["learningSecondsLeft"] is None)

    uit = agent.learning_facts({"learning": False, "learningUntil": None})
    u.check("en uit blijft uit", uit["learning"] is False)


# ── De labels ───────────────────────────────────────────────────────────────

def test_label_zetten_en_weghalen(agent, u):
    u.check("een label zetten lukt", agent.apply_label("owner-a", "Laptop") is None)
    u.check("en staat er daarna", agent.read_labels().get("owner-a") == "Laptop")

    u.check("overschrijven lukt", agent.apply_label("owner-a", "Werklaptop") is None)
    u.check("en vervangt de vorige", agent.read_labels().get("owner-a") == "Werklaptop")

    u.check("een leeg label haalt hem weg", agent.apply_label("owner-a", "") is None)
    u.check("en dan is hij er niet meer", "owner-a" not in agent.read_labels())

    u.check("nog een keer weghalen is geen fout",
            agent.apply_label("owner-a", "") is None)


def test_label_blijft_op_schijf(agent, u, state_dir):
    """Het bestand is JSON en te lezen zonder de agent.

    Dat is geen formaliteit: dit bestand staat onder de app-datamap en gaat mee in
    de back-up van umbrelOS. Een gebruiker die zijn labels kwijt is, moet ze daar
    kunnen terugvinden.
    """
    agent.apply_label("owner-b", "Telefoon")
    with open(os.path.join(state_dir, "labels.json"), "r", encoding="utf-8") as f:
        op_schijf = json.load(f)
    u.check("het label staat in labels.json", op_schijf.get("owner-b") == "Telefoon")


def test_label_validatie(agent, u):
    goed = agent.valid_label({"ownerId": "owner-c", "label": "  Mijn  laptop  "})
    u.check("witruimte wordt samengetrokken", goed[1] == "Mijn laptop")
    u.check("en er is geen fout", goed[2] is None)

    # Regeleindes eruit: dit is één regel naast een id, en een label met een nieuwe
    # regel erin zou de lijst uit elkaar trekken.
    plat = agent.valid_label({"ownerId": "owner-c", "label": "een\ntwee"})
    u.check("een regeleinde wordt een spatie", plat[1] == "een twee")

    leeg = agent.valid_label({"ownerId": "owner-c"})
    u.check("een ontbrekend label is leeg en geen fout", leeg[1] == "" and leeg[2] is None)

    for rommel, waarom in [
        ({}, "geen ownerId"),
        ({"ownerId": ""}, "leeg ownerId"),
        ({"ownerId": 42}, "ownerId is geen tekst"),
        ({"ownerId": "x" * (agent.MAX_OWNER_ID_LENGTH + 1)}, "ownerId te lang"),
        ({"ownerId": "owner-c", "label": 42}, "label is geen tekst"),
        ({"ownerId": "owner-c", "label": "x" * (agent.MAX_LABEL_LENGTH + 1)}, "label te lang"),
        ("geen object", "geen object"),
    ]:
        u.check("een onbruikbaar label wordt geweigerd: %s" % waarom,
                agent.valid_label(rommel)[2] is not None)


def test_labels_lopen_niet_vol(agent, u):
    """Er zit een plafond op. Niet tegen een aanvaller, want deze pagina zit achter
    de inlog van umbrelOS, maar tegen een lus die per ongeluk blijft schrijven.
    """
    for i in range(agent.MAX_LABELS):
        agent.apply_label("bulk-%d" % i, "label %d" % i)

    u.check("het plafond is bereikt", len(agent.read_labels()) == agent.MAX_LABELS)
    u.check("en een label erboven wordt geweigerd",
            agent.apply_label("een-te-veel", "nog een") is not None)
    u.check("maar een bestaand label mag nog wél gewijzigd worden",
            agent.apply_label("bulk-0", "gewijzigd") is None,
            "anders kun je bij een vol bestand niets meer verbeteren")


def test_onleesbare_labels_zijn_geen_ramp(agent, u, state_dir):
    """Anders dan owners.json: daar hangt aan een half begrepen bestand de vraag wie
    er binnen mag, en dan is weigeren het antwoord. Hier gaat het om een naam naast
    een id, en het ergste gevolg is dat je de rauwe ids ziet.
    """
    with open(os.path.join(state_dir, "labels.json"), "w", encoding="utf-8") as f:
        f.write("dit is geen json {{{")
    u.check("een onleesbaar labelbestand leest als leeg", agent.read_labels() == {})

    with open(os.path.join(state_dir, "labels.json"), "w", encoding="utf-8") as f:
        json.dump({"owner-d": "Goed", "owner-e": 42, "": "geen id", "owner-f": "  "}, f)
    labels = agent.read_labels()
    u.check("de goede regel blijft", labels.get("owner-d") == "Goed")
    u.check("een label dat geen tekst is valt weg", "owner-e" not in labels)
    u.check("een leeg id valt weg", "" not in labels)
    u.check("een label van alleen witruimte valt weg", "owner-f" not in labels)


# ── De opdracht met seconden ────────────────────────────────────────────────

def test_set_learning_met_seconden(agent, u):
    goed, problem = agent.valid_command({"action": "set-learning", "value": True, "seconds": 120})
    u.check("openzetten met seconden mag", problem is None)
    u.check("en de seconden gaan mee naar de relay", goed.get("seconds") == 120)

    zonder, problem = agent.valid_command({"action": "set-learning", "value": True})
    u.check("openzetten zonder seconden mag ook", problem is None)
    u.check("en dan staat er geen seconds in de opdracht", "seconds" not in zonder)

    for seconden, waarom in [
        (0, "nul"),
        (-60, "negatief"),
        (1.5, "geen heel getal"),
        ("120", "tekst"),
        # isinstance(True, int) is in Python waar, dus zonder de uitsluiting in de
        # agent zou een boolean hier als aantal seconden doorglippen.
        (True, "een boolean"),
        (agent.MAX_LEARNING_SECONDS + 1, "boven de bovengrens"),
    ]:
        _, problem = agent.valid_command(
            {"action": "set-learning", "value": True, "seconds": seconden})
        u.check("een onbruikbaar aantal seconden wordt geweigerd: %s" % waarom,
                problem is not None)

    _, problem = agent.valid_command(
        {"action": "set-learning", "value": True, "seconds": agent.MAX_LEARNING_SECONDS})
    u.check("de bovengrens zelf mag wel", problem is None)

    # Seconden bij dichtzetten is onzin en hoort dus geweigerd te worden in plaats
    # van stil genegeerd: stil negeren zou een sluitopdracht met een tikfout laten
    # slagen terwijl er iets anders gebeurt dan er staat.
    _, problem = agent.valid_command(
        {"action": "set-learning", "value": False, "seconds": 120})
    u.check("seconden bij dichtzetten wordt geweigerd", problem is not None)


def test_status_hangt_labels_aan_de_regels(agent, u, state_dir):
    """De pagina krijgt het label bij de eigenaar, en het bestand van de relay
    blijft ongemoeid. Dat tweede is de reden dat `met_label` een kopie maakt.
    """
    schrijf_owners(state_dir, {
        "version": 1,
        "learning": False,
        "learningUntil": None,
        "owners": [
            {"id": "owner-g", "allowed": True, "firstSeen": iso(-600), "lastSeen": iso(-60)},
            {"id": "owner-h", "allowed": False, "firstSeen": iso(-900), "lastSeen": None},
        ],
        "rejected": [{"id": "owner-i", "firstSeen": iso(-300), "lastSeen": iso(-10),
                      "attempts": 3}],
    })
    agent.apply_label("owner-g", "Laptop")
    agent.apply_label("owner-i", "Onbekend apparaat")

    status = agent.build_status()
    owners = status["owners"]

    u.check("de toegelaten eigenaar krijgt zijn label",
            owners["allowed"][0]["label"] == "Laptop")
    u.check("de geblokkeerde staat in blocked", owners["blocked"][0]["id"] == "owner-h")
    u.check("en heeft geen label", owners["blocked"][0]["label"] is None)
    u.check("de geweigerde poging staat in rejected", owners["rejected"][0]["id"] == "owner-i")
    u.check("en krijgt zijn label", owners["rejected"][0]["label"] == "Onbekend apparaat")
    u.check("het aantal pogingen gaat mee", owners["rejected"][0]["attempts"] == 3)

    u.check("de leerstand komt uit learning_facts", owners["learning"] is False)
    u.check("en de statusvelden voor de teller staan erin",
            "learningSecondsLeft" in owners and "learningUntil" in owners)

    with open(os.path.join(state_dir, "owners.json"), "r", encoding="utf-8") as f:
        op_schijf = json.load(f)
    u.check("owners.json is niet aangeraakt",
            all("label" not in regel for regel in op_schijf["owners"]),
            "de agent hoort niet in het bestand van de relay te schrijven")


def main():
    u = Uitslag()
    test_template_is_invulbaar_zonder_schade(u)

    with tempfile.TemporaryDirectory() as state_dir:
        agent = load_agent(state_dir)

        test_venster_dat_nog_loopt(agent, u)
        test_venster_dat_verlopen_is(agent, u)
        test_venster_zonder_tijdslot(agent, u)
        test_onleesbaar_tijdslot(agent, u)
        test_leerstand_onbekend(agent, u)

        test_label_zetten_en_weghalen(agent, u)
        test_label_blijft_op_schijf(agent, u, state_dir)
        test_label_validatie(agent, u)
        test_onleesbare_labels_zijn_geen_ramp(agent, u, state_dir)

        test_set_learning_met_seconden(agent, u)
        test_status_hangt_labels_aan_de_regels(agent, u, state_dir)

        # Als laatste: deze vult het labelbestand tot het plafond en laat dus geen
        # bruikbare staat achter voor een toets erna.
        test_labels_lopen_niet_vol(agent, u)

    return u.rapport()


if __name__ == "__main__":
    sys.exit(main())