#!/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 4. Integrationstest: liest CSV aus BricsCAD-Lauf (falls vorhanden) """ import 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 ) # --------------------------------------------------------------------------- # 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) def test_item_count_matches_testcases(self): """Anzahl exportierter Elemente muss Anzahl Testfaelle entsprechen.""" assert len(self.items) == len(self.tests), ( f"Erwartet {len(self.tests)} Elemente, erhalten {len(self.items)}" ) def test_numbering_starts_at_one(self): """Element-Nummerierung muss bei 1 beginnen.""" assert self.items[0]["nr"] == 1, ( f"Erstes Element: nr={self.items[0]['nr']}, erwartet 1" ) def test_numbering_is_consecutive(self): """Element-Nummern muessen lueckenlos aufsteigen.""" for i, item in enumerate(self.items): assert item["nr"] == i + 1, ( f"Element {i}: nr={item['nr']}, erwartet {i + 1}" ) 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.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.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.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.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.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"] == str(eintrag["Sivasnr"]), ( f"{sivasnr}: SivasNummer erwartet={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.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"] == str(eintrag["Sivasnr"]), ( f"{sivasnr}: SivasNummer erwartet={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.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.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.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: CSV aus BricsCAD-Lauf validieren # --------------------------------------------------------------------------- class TestOmnifloExportCSV: """Validiert CSV-Datei aus dem BricsCAD-Lauf (c:TEST_OMNIFLO_EXPORT).""" @pytest.fixture(autouse=True) def setup(self): csv_path = os.path.join(_output_dir(), "omniflo_export.csv") if not os.path.exists(csv_path): pytest.skip( "omniflo_export.csv nicht vorhanden - " "TEST_OMNIFLO_EXPORT in BricsCAD ausfuehren" ) self.csv_path = csv_path self.rows = [] with open(csv_path, "r", encoding="utf-8") as f: reader = csv.reader(f, delimiter=";") self.header = next(reader) for row in reader: self.rows.append(row) def test_header_felder(self): """CSV-Header muss korrekte Spaltenbezeichnungen haben.""" assert self.header[0] == "Elementnummer" assert self.header[1] == "TeileArt" assert self.header[2] == "TeileId" assert self.header[3] == "Bezeichnung" assert self.header[4] == "Anzahl" assert self.header[5] == "Merkmale" def test_mindest_anzahl_zeilen(self): """CSV muss mindestens so viele Zeilen haben wie Testfaelle.""" tests = _load_omniflo_tests() assert len(self.rows) >= len(tests), ( f"CSV hat {len(self.rows)} Zeilen, " f"erwartet mind. {len(tests)}" ) def test_anzahl_spalte_ist_eins(self): """Anzahl-Spalte muss fuer alle Zeilen '1' sein.""" for i, row in enumerate(self.rows): if len(row) < 5: continue assert row[4] == "1", ( f"Zeile {i + 2}: Anzahl={row[4]}, erwartet 1" ) def test_merkmale_sind_valides_json(self): """Merkmale-Spalte muss valides JSON enthalten.""" for i, row in enumerate(self.rows): if len(row) < 6: continue try: json.loads(row[5]) except json.JSONDecodeError as e: pytest.fail( f"Zeile {i + 2}: Merkmale ist kein valides JSON: {e}\n" f" Inhalt: {row[5][:100]}" ) def test_bogen_merkmale_felder_in_csv(self): """Bogen-Zeilen muessen Pflichtfelder in Merkmale haben.""" required = {"Kurvenwinkel", "Radius", "Höhe", "Drehung", "SivasNummer"} for i, row in enumerate(self.rows): if len(row) < 6 or row[1].strip('"') != "Omniflo Kurve": continue merkmale = json.loads(row[5]) missing = required - set(merkmale.keys()) assert not missing, ( f"Zeile {i + 2}: Fehlende Bogen-Felder: {missing}" ) def test_weiche_merkmale_felder_in_csv(self): """Weichen-Zeilen muessen Pflichtfelder in Merkmale haben.""" required = {"Weichentyp", "Richtung", "Weichenwinkel", "Höhe", "Drehung", "Antrieb Kurve", "SivasNummer"} for i, row in enumerate(self.rows): if len(row) < 6 or row[1].strip('"') != "Omniflo Weiche": continue merkmale = json.loads(row[5]) missing = required - set(merkmale.keys()) assert not missing, ( f"Zeile {i + 2}: Fehlende Weichen-Felder: {missing}" )