dxfmakros/tests/test_omniflo.py

#!/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}"
            )