#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ test_omniflo.py - Validiert Omniflo Bogen/Weiche Export. Prueft: 1. CSV-Inhalt gegen Testdefinitionen (Unit-Test, ohne BricsCAD) 2. Merkmale-Werte gegen JSON-Katalog 3. Korrekte Anzahl und Reihenfolge der Elemente inkl. Omniflo-Sum-Zeile 4. Integrationstest: liest omniflo_results.json aus BricsCAD-Lauf (falls vorhanden) 5. Referenz-Vergleich: CSV-Export aus export_csv.py gegen abgenommene Referenz-CSV """ import json import os import sys import pytest # export_csv aus lib/ importieren _LIB_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "lib") sys.path.insert(0, _LIB_DIR) from export_csv import ( build_lookup, process_blocks, build_bogen_merkmale, build_weiche_merkmale, load_json, format_csv_line, format_sivasnr ) # --------------------------------------------------------------------------- # Hilfsfunktionen # --------------------------------------------------------------------------- def _data_dir(): base = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) return os.path.join(base, "data") def _testdata_dir(): return os.getenv("DXFM_TESTDATA", os.path.join(os.path.dirname(os.path.abspath(__file__)), "testdata")) def _output_dir(): return os.getenv("DXFM_TESTOUT", os.path.join(os.path.dirname(os.path.abspath(__file__)), "output")) def _make_block(block_name, x=0.0, y=0.0, z=0.0, rotation=0.0, hoehe=None, drehung=None): """Erzeugt einen Mock-Block wie ihn export.lsp (csv:block-to-json) schreibt.""" attribs = {} if hoehe is not None: attribs["HOEHE"] = str(int(hoehe)) if drehung is not None: attribs["DREHUNG"] = str(drehung) return { "block_name": block_name, "x": x, "y": y, "z": z, "rotation": rotation, "attribs": attribs, } def _load_omniflo_tests(): path = os.path.join(_testdata_dir(), "omniflo_tests.json") with open(path, "r", encoding="utf-8") as f: return json.load(f) def _load_lookup(): data_dir = _data_dir() boegen_path = os.path.join(data_dir, "json", "omniflo_boegen.json") weichen_path = os.path.join(data_dir, "json", "omniflo_weichen.json") boegen = load_json(boegen_path) if os.path.exists(boegen_path) else [] weichen = load_json(weichen_path) if os.path.exists(weichen_path) else [] return build_lookup(boegen, weichen) # --------------------------------------------------------------------------- # Unit-Tests: process_blocks mit Mock-Daten # --------------------------------------------------------------------------- class TestOmnifloExportUnit: """Unit-Tests fuer export_csv.py mit synthetischen Mock-Bloecken.""" @pytest.fixture(autouse=True) def setup(self): self.tests = _load_omniflo_tests() self.lookup = _load_lookup() # Mock-Bloecke aus Testfaellen erzeugen self.blocks = [] for i, tc in enumerate(self.tests): self.blocks.append(_make_block( block_name=tc["sivasnr"], x=float(i) * 200.0, y=0.0, z=float(tc["hoehe"]), rotation=float(tc["drehung"]), hoehe=float(tc["hoehe"]), drehung=float(tc["drehung"]), )) self.items = process_blocks(self.blocks, self.lookup) # Item-Zeilen ohne die abschliessende "Omniflo Sum"-Zeile self.element_items = [it for it in self.items if it["teileart"] != "Omniflo Sum"] self.sum_item = next((it for it in self.items if it["teileart"] == "Omniflo Sum"), None) # CSV in output speichern out_dir = _output_dir() os.makedirs(out_dir, exist_ok=True) csv_path = os.path.join(out_dir, "omniflo_export.csv") header = "Elementnummer;TeileArt;TeileId;Bezeichnung;Anzahl;Merkmale" with open(csv_path, "w", encoding="utf-8") as f: f.write(header + "\n") for item in self.items: f.write(format_csv_line(item) + "\n") def test_item_count_matches_testcases(self): """Anzahl exportierter Elemente muss Anzahl Testfaelle entsprechen.""" assert len(self.element_items) == len(self.tests), ( f"Erwartet {len(self.tests)} Elemente, erhalten {len(self.element_items)}" ) def test_numbering_starts_at_one(self): """Element-Nummerierung muss bei 1 beginnen.""" assert self.element_items[0]["nr"] == 1, ( f"Erstes Element: nr={self.element_items[0]['nr']}, erwartet 1" ) def test_numbering_is_consecutive(self): """Element-Nummern muessen lueckenlos aufsteigen (inkl. abschliessender Sum-Zeile).""" for i, item in enumerate(self.items): assert item["nr"] == i + 1, ( f"Element {i}: nr={item['nr']}, erwartet {i + 1}" ) def test_sum_zeile_angehaengt(self): """Nach allen Omniflo-Elementen muss eine 'Omniflo Sum'-Zeile folgen.""" assert self.sum_item is not None, "Keine 'Omniflo Sum'-Zeile erzeugt" assert self.sum_item["nr"] == len(self.items) assert self.sum_item["teileid"] == "autogenerated_of_json" m = self.sum_item["merkmale"] bogen_tests = [tc for tc in self.tests if tc["type"] == "bogen"] weiche_tests = [tc for tc in self.tests if tc["type"] == "weiche"] assert m["Anzahl Bögen"] == len(bogen_tests) assert (m["Anzahl Weichengrundkörper"] + m["Anzahl Einwegweichen"] + m["Anzahl Deltaweichen"] + m["Anzahl Doppelweichen und Dreiwegeweichen"] + m["Anzahl Sternweichen"]) == len(weiche_tests) def test_bogen_teileart(self): """Boegen muessen TeileArt 'Omniflo Kurve' haben.""" bogen_tests = [tc for tc in self.tests if tc["type"] == "bogen"] bogen_items = [it for it in self.element_items if it["teileart"] == "Omniflo Kurve"] assert len(bogen_items) == len(bogen_tests), ( f"Erwartet {len(bogen_tests)} Boegen, erhalten {len(bogen_items)}" ) def test_weiche_teileart(self): """Weichen muessen TeileArt 'Omniflo Weiche' haben.""" weiche_tests = [tc for tc in self.tests if tc["type"] == "weiche"] weiche_items = [it for it in self.element_items if it["teileart"] == "Omniflo Weiche"] assert len(weiche_items) == len(weiche_tests), ( f"Erwartet {len(weiche_tests)} Weichen, erhalten {len(weiche_items)}" ) def test_bogen_merkmale_felder(self): """Boegen-Merkmale muessen alle Pflichtfelder enthalten.""" required = {"Kurvenwinkel", "Radius", "Höhe", "Drehung", "SivasNummer"} for item in self.element_items: if item["teileart"] != "Omniflo Kurve": continue merkmale = item["merkmale"] missing = required - set(merkmale.keys()) assert not missing, ( f"{item['bezeichnung']}: Fehlende Felder: {missing}" ) def test_weiche_merkmale_felder(self): """Weichen-Merkmale muessen alle Pflichtfelder enthalten.""" required = {"Weichentyp", "Richtung", "Weichenwinkel", "Höhe", "Drehung", "Antrieb Kurve", "SivasNummer"} for item in self.element_items: if item["teileart"] != "Omniflo Weiche": continue merkmale = item["merkmale"] missing = required - set(merkmale.keys()) assert not missing, ( f"{item['bezeichnung']}: Fehlende Felder: {missing}" ) def test_bogen_merkmale_werte(self): """Bogen-Merkmale muessen mit JSON-Katalog uebereinstimmen.""" for tc, item in zip( [t for t in self.tests if t["type"] == "bogen"], [it for it in self.element_items if it["teileart"] == "Omniflo Kurve"] ): sivasnr = tc["sivasnr"] assert sivasnr in self.lookup, f"SivasNr {sivasnr} nicht im Katalog" _, eintrag = self.lookup[sivasnr] m = item["merkmale"] assert m["Kurvenwinkel"] == float(eintrag["KurvenWinkel"]), ( f"{sivasnr}: Kurvenwinkel erwartet={eintrag['KurvenWinkel']}, " f"erhalten={m['Kurvenwinkel']}" ) assert m["Radius"] == float(eintrag["Radius"]), ( f"{sivasnr}: Radius erwartet={eintrag['Radius']}, " f"erhalten={m['Radius']}" ) assert m["Höhe"] == str(int(tc["hoehe"])), ( f"{sivasnr}: Hoehe erwartet={tc['hoehe']}, erhalten={m['Höhe']}" ) assert m["SivasNummer"] == format_sivasnr(eintrag["Sivasnr"]), ( f"{sivasnr}: SivasNummer erwartet={format_sivasnr(eintrag['Sivasnr'])}, " f"erhalten={m['SivasNummer']}" ) def test_weiche_merkmale_werte(self): """Weichen-Merkmale muessen mit JSON-Katalog uebereinstimmen.""" for tc, item in zip( [t for t in self.tests if t["type"] == "weiche"], [it for it in self.element_items if it["teileart"] == "Omniflo Weiche"] ): sivasnr = tc["sivasnr"] assert sivasnr in self.lookup, f"SivasNr {sivasnr} nicht im Katalog" _, eintrag = self.lookup[sivasnr] m = item["merkmale"] assert m["Weichentyp"] == eintrag["WeichenTyp"], ( f"{sivasnr}: Weichentyp erwartet={eintrag['WeichenTyp']}, " f"erhalten={m['Weichentyp']}" ) assert m["Weichenwinkel"] == float(eintrag["KurvenWinkel"]), ( f"{sivasnr}: Weichenwinkel erwartet={eintrag['KurvenWinkel']}, " f"erhalten={m['Weichenwinkel']}" ) assert m["Höhe"] == str(int(tc["hoehe"])), ( f"{sivasnr}: Hoehe erwartet={tc['hoehe']}, erhalten={m['Höhe']}" ) assert m["SivasNummer"] == format_sivasnr(eintrag["Sivasnr"]), ( f"{sivasnr}: SivasNummer erwartet={format_sivasnr(eintrag['Sivasnr'])}, " f"erhalten={m['SivasNummer']}" ) expected_antrieb = eintrag.get("SivasnrTEF") is not None assert m["Antrieb Kurve"] == expected_antrieb, ( f"{sivasnr}: Antrieb Kurve erwartet={expected_antrieb}, " f"erhalten={m['Antrieb Kurve']}" ) def test_bogen_hoehe_als_string(self): """Bogen-Hoehe muss als String vorliegen (z.B. '2000', nicht 2000).""" for item in self.element_items: if item["teileart"] != "Omniflo Kurve": continue assert isinstance(item["merkmale"]["Höhe"], str), ( f"{item['bezeichnung']}: Hoehe muss String sein, " f"ist {type(item['merkmale']['Höhe'])}" ) def test_weiche_hoehe_als_string(self): """Weichen-Hoehe muss als String vorliegen.""" for item in self.element_items: if item["teileart"] != "Omniflo Weiche": continue assert isinstance(item["merkmale"]["Höhe"], str), ( f"{item['bezeichnung']}: Hoehe muss String sein" ) def test_kurvenwinkel_als_float(self): """Kurvenwinkel muss als Float vorliegen (90.0, nicht 90).""" for item in self.element_items: if item["teileart"] == "Omniflo Kurve": assert isinstance(item["merkmale"]["Kurvenwinkel"], float), ( f"{item['bezeichnung']}: Kurvenwinkel muss float sein" ) elif item["teileart"] == "Omniflo Weiche": assert isinstance(item["merkmale"]["Weichenwinkel"], float), ( f"{item['bezeichnung']}: Weichenwinkel muss float sein" ) def test_weiche_alle_typen_vorhanden(self): """Test muss alle Weichen-Typen aus dem Katalog abdecken.""" typen_in_tests = { self.lookup[tc["sivasnr"]][1].get("WeichenTyp") for tc in self.tests if tc["type"] == "weiche" and tc["sivasnr"] in self.lookup } # Mindestens Einzelweiche, Doppelweiche, Dreiwegeweiche muessen vorhanden sein for pflicht in ("Einzelweiche", "Doppelweiche", "Dreiwegeweiche"): assert pflicht in typen_in_tests, ( f"Weichen-Typ '{pflicht}' fehlt in Testfaellen" ) # --------------------------------------------------------------------------- # Integrations-Tests: Omniflo-Ergebnisse aus BricsCAD-Lauf validieren # (tests/output/omniflo_results.json, geschrieben von omniflo:export-results) # --------------------------------------------------------------------------- class TestOmnifloResults: """Validiert omniflo_results.json aus dem BricsCAD-Lauf (c:TEST_OMNIFLO).""" def test_all_tests_executed(self, omniflo_testdata, omniflo_results): """Alle definierten Testfaelle muessen ausgefuehrt worden sein.""" expected_sivas = {t["sivasnr"] for t in omniflo_testdata} result_sivas = {r["sivasnr"] for r in omniflo_results} missing = expected_sivas - result_sivas assert not missing, f"Nicht ausgefuehrte Tests: {missing}" def test_all_executed_successfully(self, omniflo_results): """Alle Ergebnisse muessen Status 'OK' haben.""" for result in omniflo_results: assert result["status"] == "OK", ( f'{result["test_id"]} ({result["sivasnr"]}): Status={result["status"]}' ) def test_hoehe_matches(self, omniflo_testdata, omniflo_results): """Eingefuegte Hoehe muss dem Testfall entsprechen.""" results_by_sivas = {r["sivasnr"]: r for r in omniflo_results} for test in omniflo_testdata: result = results_by_sivas.get(test["sivasnr"]) if not result: continue assert result["hoehe"] == str(int(test["hoehe"])), ( f'{test["sivasnr"]}: Hoehe erwartet={test["hoehe"]}, ' f'ist={result["hoehe"]}' ) # --------------------------------------------------------------------------- # Referenz-Tests: CSV gegen Referenzdatei vergleichen # --------------------------------------------------------------------------- def _reference_dir(): return os.path.join(os.path.dirname(os.path.abspath(__file__)), "reference") def _normalize_row(line): """Splittet eine CSV-Zeile und ersetzt die TeileId (UUID) durch Platzhalter.""" cols = line.strip().split(";", 5) if len(cols) >= 3: cols[2] = "" return cols class TestOmnifloReferenceCSV: """Vergleicht den von export_csv.py erzeugten CSV-Export mit einer Referenzdatei.""" @pytest.fixture(autouse=True) def setup(self): self.out_csv = os.path.join(_output_dir(), "omniflo_export.csv") self.ref_csv = os.path.join(_reference_dir(), "omniflo_export_reference.csv") if not os.path.exists(self.out_csv): pytest.skip( "omniflo_export.csv nicht vorhanden - " "zuerst TestOmnifloExportUnit ausfuehren" ) if not os.path.exists(self.ref_csv): pytest.skip("Referenzdatei nicht vorhanden: " + self.ref_csv) with open(self.out_csv, "r", encoding="utf-8") as f: self.out_lines = f.readlines() with open(self.ref_csv, "r", encoding="utf-8") as f: self.ref_lines = f.readlines() def test_header_matches_reference(self): """Header muss mit Referenz uebereinstimmen.""" assert self.out_lines[0].strip() == self.ref_lines[0].strip(), ( f"Header abweichend:\n Ist: {self.out_lines[0].strip()}\n" f" Soll: {self.ref_lines[0].strip()}" ) def test_zeilenanzahl_matches_reference(self): """Zeilenanzahl muss mit Referenz uebereinstimmen.""" assert len(self.out_lines) == len(self.ref_lines), ( f"Zeilenanzahl abweichend: Ist={len(self.out_lines)}, " f"Soll={len(self.ref_lines)}" ) def test_zeilen_matches_reference(self): """Alle Zeilen muessen mit Referenz uebereinstimmen (TeileId wird ignoriert).""" min_lines = min(len(self.out_lines), len(self.ref_lines)) diffs = [] for i in range(1, min_lines): out_cols = _normalize_row(self.out_lines[i]) ref_cols = _normalize_row(self.ref_lines[i]) if out_cols != ref_cols: diffs.append( f"Zeile {i + 1}:\n" f" Ist: {self.out_lines[i].strip()[:120]}\n" f" Soll: {self.ref_lines[i].strip()[:120]}" ) assert not diffs, ( f"{len(diffs)} abweichende Zeile(n) (erste 5):\n" + "\n".join(diffs[:5]) )