dxfmakros/tests/test_hm_recformat.py

#!/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'