UmbrelApps/tests/test_relay_agent.py

"""Toetst de agent van Evolu Relay: de labels.

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;
Tot 0.6.0 rekende hij daarnaast het tijdvenster voor nieuwe eigenaars uit. Dat
venster is er in 0.7.0 uit: er is nog één weg naar binnen, en dat is de wachtlijst
met handmatig toelaten. De toetsen daarover zijn met de functie mee verdwenen.

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 `tools/evolu-relay/agent.py` rechtstreeks: het bestand dat de
Dockerfile in de image zet. Tot 0.6.0 heette dat `agent.py.template` en stond het
in de app-map; de eerste toets hieronder is uit die tijd en blijft staan.
"""

import sys

# Vóór de imports, want anders is het te laat: Python legt bytecode naast agent.py
# zodra die geïmporteerd wordt, en die rommel hoort niet in de repo. 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__))
AGENT = os.path.join(HERE, os.pardir, "tools", "evolu-relay", "agent.py")


def load_agent(state_dir):
    """Laadt agent.py 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, buiten sys.path om.
    """
    os.environ["RELAY_STATE_DIR"] = state_dir
    loader = importlib.machinery.SourceFileLoader("relay_agent", AGENT)
    spec = importlib.util.spec_from_file_location("relay_agent", AGENT, 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):
    """Een restant uit de template-tijd dat blijft staan.

    Het bestand zit sinds 0.6.0 in de image en gaat door geen envsubst meer. Maar
    als iemand het ooit weer als template in een app-map zet, is dit de toets die
    het meteen ziet.
    """
    with open(AGENT, "r", encoding="utf-8") as f:
        inhoud = f.read()
    u.check("agent.py bevat geen dollartekens",
            "$" not in inhoud,
            "een envsubst-stap zou die invullen en de Python-code slopen")


# ── 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 status ───────────────────────────────────────────────────────────────

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)

    # De leerstand is er in 0.7.0 uit. Deze toets houdt vast dat de agent hem ook
    # niet meer dóórgeeft: een pagina die het veld nog zou lezen, zou anders een
    # knop kunnen tonen voor iets dat niet meer bestaat.
    u.check("de leerstand staat niet meer in de status",
            "learning" not in owners and "learningSecondsLeft" not 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_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_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())