UmbrelApps: community app store whatsnext
Electrum Gate, Evolu Relay en Atreides Assets, met de image-recepten in tools/ en de tests in tests/. Begonnen zonder voorgeschiedenis op 09-10-2026. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
commit
85d85a1295
63 files changed
+13987
No files matched your search
@@ -0,0 +1,263 @@
|
||||
"""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())
|
||||
Reference in new issue
Block a user