392 lines
16 KiB
Python
392 lines
16 KiB
Python
#!/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
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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"] == 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.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"] == 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.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] = "<ID>"
|
|
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])
|
|
)
|