#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ test_hm_recformat.py - Validiert die 5 VarioFoerderer-Linienzuege der Anlage HundM. Zwei Ebenen: 1. Die Testdaten selbst (tests/testdata/hm_recformat.json): Struktur der Eingabe-Journale und Lesbarkeit durch den LISP-Parser. 2. Die von TEST_HM_RECFORMAT (test_hm_recformat.lsp) erzeugten Ergebnisse (output/hmrec_results.json + output/hm_recformat_tests.dxf): jede Kette gebaut, kein Journal-Wert offen, Attribute passen zu den Eingaben. Die Testdaten stammen aus lib/vf_journal_export.py (Journale der VF_n-Bloecke aus data/polylines.dxf, XDATA-App SSG_VF_EDIT) und nicht aus Handarbeit - Schema siehe tests/testdata/object_data.md Abschnitt 8. """ import json import os import pytest EINGABE_TYPEN = ("point_abs", "point_rel", "real", "int", "string", "step") # ============================================================ # Hilfsfunktionen: flaches JSON in Ketten gruppieren # ============================================================ # Gleiche Regel wie hmrec:gruppiere in test_hm_recformat.lsp: ein Objekt mit # "test_id" ist ein Kopf und beginnt eine Kette, die folgenden Objekte mit # "typ" sind ihre Eingaben. def _ketten(testdata): ketten = [] for obj in testdata: if "test_id" in obj: ketten.append((obj, [])) elif "typ" in obj and ketten: ketten[-1][1].append(obj) return ketten def _journal_laenge(eingaben): """Anzahl der Journal-Eintraege, die der Treiber daraus baut: ein point_rel mit "hz" wird zu ZWEI Eintraegen (DL + REAL).""" return sum(2 if (e["typ"] == "point_rel" and "hz" in e) else 1 for e in eingaben) def _pfad(*teile): return os.path.join(os.path.dirname(os.path.abspath(__file__)), *teile) # ============================================================ # Testdaten-Pruefung (hm_recformat.json selbst) # ============================================================ class TestAufzeichnungTestdaten: def test_fuenf_ketten(self, hmrec_testdata): """Die Anlage hat fuenf am Kreisel haengende Linienzuege.""" ketten = _ketten(hmrec_testdata) assert len(ketten) == 5, \ f"{len(ketten)} Ketten in hm_recformat.json, erwartet 5" def test_jedes_objekt_zuordenbar(self, hmrec_testdata): """Kein Objekt darf zwischen Kopf und Eingabe fallen - sonst verschluckt hmrec:gruppiere es stillschweigend.""" for i, obj in enumerate(hmrec_testdata): assert ("test_id" in obj) or ("typ" in obj), \ f"Objekt {i} ist weder Kopf noch Eingabe: {sorted(obj)}" def test_kopf_felder(self, hmrec_testdata): for kopf, _ in _ketten(hmrec_testdata): assert kopf["test_id"].startswith("VF_"), kopf["test_id"] # "typ" wuerde den Kopf zu einer Eingabe machen (siehe # hmrec:gruppiere) - der Anlagetyp heisst darum "anlagetyp". assert "typ" not in kopf, \ f'{kopf["test_id"]}: Kopf darf kein Feld "typ" tragen' assert kopf["anlagetyp"] == "linienzug", kopf["test_id"] assert kopf["modus"] == 1, kopf["test_id"] assert kopf["block"].startswith("VF_"), kopf["block"] assert len(kopf["startpunkt_mm"]) == 3, kopf["test_id"] def test_anzahl_eingaben_stimmt(self, hmrec_testdata): """anzahl_eingaben ist die Selbstauskunft des Kopfes - weicht sie ab, fehlen Eingaben (z.B. durch eine Handaenderung).""" for kopf, eingaben in _ketten(hmrec_testdata): assert kopf["anzahl_eingaben"] == len(eingaben), \ f'{kopf["test_id"]}: anzahl_eingaben=' \ f'{kopf["anzahl_eingaben"]}, gezaehlt {len(eingaben)}' def test_eingabe_typen_bekannt(self, hmrec_testdata): for kopf, eingaben in _ketten(hmrec_testdata): for i, e in enumerate(eingaben): assert e["typ"] in EINGABE_TYPEN, \ f'{kopf["test_id"]} Eingabe {i}: unbekannter Typ ' \ f'{e["typ"]}' assert e.get("kommentar"), \ f'{kopf["test_id"]} Eingabe {i}: kein Kommentar' def test_kette_beginnt_mit_startpunkt(self, hmrec_testdata): """Erste Eingabe = absoluter Startpunkt, zweite = Starthoehe. Fehlt der Punkt, wuerde das Replay ihn interaktiv erfragen und die Kette an der falschen Stelle aufbauen (siehe hmrec:build-linienzug).""" for kopf, eingaben in _ketten(hmrec_testdata): assert eingaben[0]["typ"] == "point_abs", kopf["test_id"] assert eingaben[1]["typ"] == "real", kopf["test_id"] punkt = eingaben[0]["wert"] assert len(punkt) == 3, kopf["test_id"] assert punkt == kopf["startpunkt_mm"], \ f'{kopf["test_id"]}: point_abs {punkt} != startpunkt_mm ' \ f'{kopf["startpunkt_mm"]}' assert eingaben[1]["wert"] == punkt[2], \ f'{kopf["test_id"]}: Starthoehe {eingaben[1]["wert"]} != ' \ f'Z des Startpunkts {punkt[2]}' def test_hz_nur_beim_ersten_segment(self, hmrec_testdata): """vfl-in-abstand journalisiert die Fahrtrichtung NUR beim ersten Segment der Kette; jedes weitere erbt sie vom Vorgaenger. Ein "hz" an spaeterer Stelle waere ein Wert zu viel in der Replay-Queue und verschiebt alle folgenden Antworten (Desync).""" for kopf, eingaben in _ketten(hmrec_testdata): rel = [e for e in eingaben if e["typ"] == "point_rel"] assert rel, f'{kopf["test_id"]}: kein Segment' assert "hz" in rel[0], \ f'{kopf["test_id"]}: erstes Segment ohne hz' for i, e in enumerate(rel[1:], start=2): assert "hz" not in e, \ f'{kopf["test_id"]}: Segment {i} traegt hz - nur das ' \ f'erste Segment darf eines haben' def test_segmentlaengen_positiv(self, hmrec_testdata): for kopf, eingaben in _ketten(hmrec_testdata): for e in eingaben: if e["typ"] == "point_rel": assert e["dL"] > 0, f'{kopf["test_id"]}: dL={e["dL"]}' def test_menue_antworten_sind_strings(self, hmrec_testdata): """Menue-Antworten muessen Strings bleiben ("1"/"2"/"90"/"30"): der Produktionscode vergleicht mit (= antwort "1"), eine Zahl wuerde stumm in den Default-Zweig laufen.""" for kopf, eingaben in _ketten(hmrec_testdata): for i, e in enumerate(eingaben): if e["typ"] in ("string", "step"): assert isinstance(e["wert"], str), \ f'{kopf["test_id"]} Eingabe {i}: {e["wert"]!r} ist ' \ f'kein String' def test_glieder_vorhanden(self, hmrec_testdata): """Jede Kette hat Glied-Marker (STEP) - sie sind der Anker fuer den Sektions-/Glied-Edit und die Kurzbeschreibung.""" for kopf, eingaben in _ketten(hmrec_testdata): glieder = [e["wert"] for e in eingaben if e["typ"] == "step"] assert glieder, f'{kopf["test_id"]}: kein STEP-Marker' for g in glieder: assert g in ("Linie-GF", "Linie-VF", "Linie", "GF-Bogen", "Horizontal-VF", "Vario-Kurve", "ES"), \ f'{kopf["test_id"]}: unbekanntes Glied {g}' def test_arrays_einzeilig(self): """Jedes Array muss komplett in EINER Zeile stehen. ssg-parse-json-array (Lisp/ssg_core.lsp) liest die Datei zeilenweise, und ssg-cfg-parse-array braucht das ganze Array in einem String. Ein von json.dump(indent=2) umgebrochenes Array wird zu nil - der Treiber baut die Kette dann nicht an ihrer Position. Das ist genau der Fehler, der die Anlage einmal komplett am Ursprung aufgebaut hat, und er ist in der Zeichnung nicht als Datenfehler zu erkennen. """ with open(_pfad("testdata", "hm_recformat.json"), encoding="utf-8") as fh: for nr, zeile in enumerate(fh, start=1): offen = zeile.count("[") - zeile.count("]") # Die aeusserste Klammer des Dokuments darf allein stehen. if zeile.strip() in ("[", "]"): continue assert offen == 0, \ f"Zeile {nr}: Array ueber mehrere Zeilen umgebrochen - " \ f"ssg-cfg-parse-array kann das nicht lesen: {zeile.strip()!r}" def test_flach_fuer_lisp_parser(self, hmrec_testdata): """Die Datei muss FLACH bleiben. ssg-load-json / ssg-parse-json-array behandelt jedes "{" als Objektanfang und jedes "}"/"}," als Objektende - es kennt keine Verschachtelung. Ein Objekt in einem Objekt (oder ein Array von Objekten wie "eingaben": [...]) wuerde das umgebende Objekt verschlucken. Dieser Test zaehlt die Objekte so, wie der LISP-Parser sie zaehlen wuerde, und vergleicht mit dem echten JSON. """ objekte = 0 with open(_pfad("testdata", "hm_recformat.json"), encoding="utf-8") as fh: for zeile in fh: gestrippt = zeile.strip() if gestrippt == "{": objekte += 1 elif gestrippt in ("}", "},"): continue elif gestrippt.startswith('"'): # Ein Wert, der ein Objekt oder eine Objektliste eroeffnet, # macht die Datei fuer den LISP-Parser unlesbar. assert not gestrippt.endswith(("{", "[")), \ f"verschachtelter Wert: {gestrippt!r}" assert objekte == len(hmrec_testdata), \ f"LISP-Parser saehe {objekte} Objekte, json.load sieht " \ f"{len(hmrec_testdata)}" # ============================================================ # Ergebnis-Pruefung (results.json vs. testdata) # ============================================================ class TestAufzeichnungResults: def test_je_kette_ein_ergebnis(self, hmrec_testdata, hmrec_results): soll = [k["test_id"] for k, _ in _ketten(hmrec_testdata)] ist = [r["test_id"] for r in hmrec_results] assert ist == soll, f"Ergebnisse {ist}, erwartet {soll}" for r in hmrec_results: assert r["kind"] == "linienzug", f'{r["test_id"]}: {r["kind"]}' def test_status_executed(self, hmrec_results): """"desync" (Journal-Werte nicht verbraucht) und "failed" (Abbruch) sind beide echte Fehler - siehe hmrec:build-linienzug.""" for r in hmrec_results: assert r["status"] == "executed", \ f'{r["test_id"]}: Status={r["status"]}' def test_journal_vollstaendig_verbraucht(self, hmrec_testdata, hmrec_results): """Alle aufgezeichneten Werte muessen im Bau abgerufen worden sein. Bleibt etwas offen, ist der Ablauf von der Aufzeichnung abgewichen und die Geometrie kann der Vorlage nicht entsprechen.""" soll = {k["test_id"]: _journal_laenge(e) for k, e in _ketten(hmrec_testdata)} for r in hmrec_results: assert "eingaben_offen" in r, ( f'{r["test_id"]}: kein eingaben_offen im Ergebnis - die Kette ' f'ist vor der Auswertung abgebrochen (status=' f'{r.get("status")!r}, {r.get("fehler_text")})') assert r["eingaben_offen"] == 0, \ f'{r["test_id"]}: {r["eingaben_offen"]} Journal-Werte offen' assert r["eingaben_gesamt"] == soll[r["test_id"]], \ f'{r["test_id"]}: {r["eingaben_gesamt"]} Journal-Eintraege ' \ f'abgespielt, erwartet {soll[r["test_id"]]}' def test_keine_live_eingabe(self, hmrec_results): """Kein Prompt, kein Dialog: der Treiber schaltet *vfl-headless* ein (harter Abbruch statt stillem Rueckfall auf Live-Eingabe, siehe vfl-headless-abbruch in Lisp/vf_linienzug.lsp) UND ersetzt zusaetzlich getpoint/getstring/getint/getreal durch zaehlende Stubs. Beide Wege muessen leer bleiben - "prompts" ist der positive Beweis, dass keine Eingabe ausserhalb der vfl-in-*-Wrapper erreicht wurde.""" for r in hmrec_results: if "prompts" in r: assert r["prompts"] == 0, \ f'{r["test_id"]}: {r["prompts"]} Live-Eingabe(n) erreicht' assert not r.get("headless_fehler"), \ f'{r["test_id"]}: Headless-Abbruch bei {r["headless_fehler"]}' def test_keine_geometrie_meldungen(self, hmrec_results): """*vfl-meldungen* sammelt die abgewiesenen Sektionen (frueher blockierende alert-Fenster, jetzt vfl-meldung). Eine Meldung heisst: ein Glied liess sich geometrisch nicht bauen - der Nachbau weicht dann von der Vorlage ab, auch wenn das Journal aufgeht.""" for r in hmrec_results: assert not r.get("meldungen"), \ f'{r["test_id"]}: Meldungen aus dem Bau: {r["meldungen"]}' def test_vf_block_erzeugt(self, hmrec_results): for r in hmrec_results: assert r["block_name"].startswith("VF_"), \ f'{r["test_id"]}: Block "{r["block_name"]}" ist kein VF_-Block' assert r["block_handle"], f'{r["test_id"]}: kein Handle' def test_dimension_einheitlich(self, hmrec_results): """Alle Ergebnisse eines Laufs stammen aus EINER Bau-Dimension. Die Eingabedaten sind dimensionsfrei; ob _2D- oder _3D-Blockvarianten verwendet werden, loest ssg-ils-dim-aktuell auf. Ein Lauf, in dem beide vorkommen, waere ein halb umgeschalteter Zustand (z.B. weil ein vorheriger Test *ssg-ils-dim* nicht zurueckgesetzt hat). """ dims = {r.get("dimension") for r in hmrec_results} assert len(dims) == 1, f"gemischte Bau-Dimensionen im Lauf: {dims}" dim = dims.pop() assert dim in ("2D", "3D"), \ f'Bau-Dimension "{dim}" ist weder 2D noch 3D' def test_startpunkt_getroffen(self, hmrec_testdata, hmrec_results): """Der VF_n-Block wird mit dem Kettenstartpunkt als Basispunkt gewickelt (ssg-block-wrap-welt in vfl-block-erstellen) - sein Einfuegepunkt muss also der Startpunkt aus den Eingaben sein.""" soll = {k["test_id"]: k["startpunkt_mm"] for k, _ in _ketten(hmrec_testdata)} for r in hmrec_results: p = soll[r["test_id"]] ist = r["insert_point"] for achse, i in (("X", 0), ("Y", 1), ("Z", 2)): assert abs(ist[i] - p[i]) < 1.0, \ f'{r["test_id"]}: Einfuegepunkt {achse} {ist[i]} != ' \ f'{p[i]}' def test_as_es_attribute_aus_eingaben(self, hmrec_testdata, hmrec_results): """WINKEL_AS/SEITE_AS und WINKEL_ES/SEITE_ES muessen die Antworten aus dem Journal wiedergeben (AS steht in der Praeambel, ES im ES-Glied).""" seiten = {"1": "links", "2": "rechts"} nach_id = {r["test_id"]: r for r in hmrec_results} for kopf, eingaben in _ketten(hmrec_testdata): r = nach_id[kopf["test_id"]] attrs = r["actual_attributes"] # Praeambel: [0] Punkt, [1] Hoehe, [2] AS setzen, [3] Winkel, # [4] Seite if eingaben[2]["wert"] == "1": assert attrs.get("WINKEL_AS") == eingaben[3]["wert"], \ f'{kopf["test_id"]}: WINKEL_AS=' \ f'{attrs.get("WINKEL_AS")}, erwartet {eingaben[3]["wert"]}' assert attrs.get("SEITE_AS") == seiten[eingaben[4]["wert"]], \ f'{kopf["test_id"]}: SEITE_AS={attrs.get("SEITE_AS")}' # ES: die beiden Strings direkt nach dem STEP-Marker "ES" for i, e in enumerate(eingaben): if e["typ"] == "step" and e["wert"] == "ES": assert attrs.get("WINKEL_ES") == eingaben[i + 1]["wert"], \ f'{kopf["test_id"]}: WINKEL_ES=' \ f'{attrs.get("WINKEL_ES")}' assert attrs.get("SEITE_ES") == \ seiten[eingaben[i + 2]["wert"]], \ f'{kopf["test_id"]}: SEITE_ES={attrs.get("SEITE_ES")}' break def test_vario_kurven_gezaehlt(self, hmrec_testdata, hmrec_results): """Jede Vario-Kurve im Journal muss in den VF_Bogen_*-Attributen auftauchen (vfl-insert-vario-kurve-block zaehlt sie mit).""" nach_id = {r["test_id"]: r for r in hmrec_results} for kopf, eingaben in _ketten(hmrec_testdata): soll = sum(1 for e in eingaben if e["typ"] == "step" and e["wert"] == "Vario-Kurve") attrs = nach_id[kopf["test_id"]]["actual_attributes"] ist = sum(int(attrs.get(f"VF_Bogen_{lage}_{w}", 0) or 0) for lage in ("A", "I") for w in (90, 60, 30)) assert ist == soll, \ f'{kopf["test_id"]}: {ist} Vario-Kurven in den Attributen, ' \ f'{soll} im Journal' def test_hoehe_von_ist_starthoehe(self, hmrec_testdata, hmrec_results): nach_id = {r["test_id"]: r for r in hmrec_results} for kopf, eingaben in _ketten(hmrec_testdata): attrs = nach_id[kopf["test_id"]]["actual_attributes"] if "HOEHE_VON_mm" not in attrs: continue assert abs(int(attrs["HOEHE_VON_mm"]) - eingaben[1]["wert"]) <= 1, \ f'{kopf["test_id"]}: HOEHE_VON_mm={attrs["HOEHE_VON_mm"]}, ' \ f'Starthoehe {eingaben[1]["wert"]}' # ============================================================ # DXF Geometrie-Pruefung (ezdxf) # ============================================================ class TestAufzeichnungGeometry: def test_block_anzahl(self, hmrec_dxf, hmrec_testdata): """Je Kette muss ein VF_-Block im Modelspace stehen.""" n_vf = sum(1 for e in hmrec_dxf.modelspace() if e.dxftype() == "INSERT" and e.dxf.name.startswith("VF_")) soll = len(_ketten(hmrec_testdata)) assert n_vf >= soll, f"VF_: {n_vf} < {soll}" def test_handles_im_modelspace(self, hmrec_dxf, hmrec_results): """Jedes gebaute Block-Handle (executed) liegt im Modelspace.""" handles = {e.dxf.handle for e in hmrec_dxf.modelspace() if e.dxftype() == "INSERT"} for r in hmrec_results: if r["status"] == "executed" and r["block_handle"]: assert r["block_handle"] in handles, \ f'{r["test_id"]}: Handle {r["block_handle"]} nicht im ' \ f'Modelspace'