sps_skel/tests/test_model.py

# -*- coding: utf-8 -*-
"""Tests fuer lib/model.py - die TRO-JSON zwischen tro_flow und scl_skeleton."""

import json

import pytest
from pydantic import ValidationError

from model import SCHEMA, Analysis, Connection, DerivedTro, TroCandidate, TroDocument
from scl_skeleton import render


def _tro(tro_id: str, type_name: str, **extra) -> DerivedTro:
    data = dict(
        tro_id=tro_id,
        type_name=type_name,
        label=f"Label {tro_id}",
        hosts=["0001"],
        separators=["0040"],
        scanners=[],
        reason="Test",
        confidence="mittel",
    )
    data.update(extra)
    return DerivedTro(**data)


@pytest.fixture
def document() -> TroDocument:
    """Kleine Anlage: SYS1 -> TRO01 (Weiche) -> TRO02 / TRO03 -> SYS2."""
    analysis = Analysis(
        tros=[
            _tro("TRO01", "1Sep1Swi", predecessors=["SYS1"], successors=["TRO02", "TRO03"],
                 position=(1030.0, 4700.0, 1500.0)),
            _tro("TRO02", "1Sep", separators=["0041"], predecessors=["TRO01"],
                 successors=["SYS2"]),
            _tro("TRO03", "Vario", separators=["0042"], predecessors=["TRO01"]),
        ],
        boundaries=[
            _tro("SYS1", "Materialeingang", separators=[], successors=["TRO01"]),
            _tro("SYS2", "Materialausgang", separators=[], predecessors=["TRO02"]),
        ],
        edges=[("SYS1", "TRO01"), ("TRO01", "TRO02"), ("TRO01", "TRO03"), ("TRO02", "SYS2")],
        edge_kinds={("TRO01", "TRO02"): "weiche"},
        findings=["ein Befund"],
    )
    return TroDocument.from_analysis(analysis, source="test.csv", note="Test")


def test_round_trip_is_lossless(document):
    text = document.to_json()
    again = TroDocument.from_json(text)
    assert again.to_json() == text
    assert again.by_id("TRO01").successors == ["TRO02", "TRO03"]


def test_json_uses_tro_json_keys(document):
    data = json.loads(document.to_json())
    assert data["schema"] == SCHEMA
    first = data["tros"][0]
    assert first["id"] == "TRO01"
    assert first["type"] == "1Sep1Swi"
    assert first["fbType"] == "FB_ILS_MTRO_1Sep1Swi"
    assert first["items"] == {"Separator": 1, "Switch": 1}
    assert first["position"] == {"x": 1030.0, "y": 4700.0, "z": 1500.0}
    assert data["connections"][1] == {"from": "TRO01", "to": "TRO02", "kind": "weiche"}
    assert data["candidates"] == []


def test_candidate_round_trip():
    candidate = TroCandidate(name="Qualitaetskontrolle", type_name="1Sep1Swi",
                             position=(-1026.0, 5250.05, 1478.5), belongs_to="TRO09")
    data = json.loads(candidate.model_dump_json(by_alias=True))
    assert data["type"] == "1Sep1Swi"
    assert data["position"] == {"x": -1026.0, "y": 5250.05, "z": 1478.5}
    assert data["belongsTo"] == "TRO09"
    assert TroCandidate.model_validate(data) == candidate
    assert candidate.describe_position() == "-1026, 5250"


def test_items_and_fb_block_follow_the_type():
    tro = _tro("TRO01", "1Sep")
    assert tro.items == {"Separator": 1}
    tro.type_name = "1Sep2Swi"
    assert tro.items == {"Separator": 1, "Switch": 2, "Scanner": 1}
    assert tro.fb_block == "FB_ILS_MTRO_1Sep2Swi"


def test_unknown_type_has_no_catalog_data():
    station = _tro("SYS1", "Materialeingang")
    assert station.definition() is None
    assert station.items == {}
    assert station.fb_block == ""


def test_catalog_fields_cannot_be_assigned():
    tro = _tro("TRO01", "1Sep")
    with pytest.raises((AttributeError, ValidationError)):
        tro.fb_block = "FB_Anders"


def test_unknown_key_is_rejected(document):
    data = json.loads(document.to_json())
    data["tros"][0]["separatorz"] = data["tros"][0].pop("separators")
    with pytest.raises(ValidationError, match="separatorz"):
        TroDocument.model_validate(data)


def test_items_not_matching_catalog_are_rejected(document):
    data = json.loads(document.to_json())
    data["tros"][1]["items"] = {"Separator": 2}
    with pytest.raises(ValidationError, match="passt nicht zum Typ"):
        TroDocument.model_validate(data)


def test_fb_type_not_matching_catalog_is_rejected(document):
    data = json.loads(document.to_json())
    data["tros"][1]["fbType"] = "FB_ILS_MTRO_Vario"
    with pytest.raises(ValidationError, match="fbType"):
        TroDocument.model_validate(data)


def test_connection_accepts_alias_and_field_name():
    assert Connection(**{"from": "A", "to": "B"}) == Connection(src="A", dst="B")


def test_scl_skeleton_renders_from_document(document):
    scl = render(document, "test_tro.json")
    assert 'TRO01 : "FB_ILS_MTRO_1Sep1Swi";' in scl
    assert 'TRO03 : "FB_ILS_MTRO_Vario";' in scl
    # Weiche: FC_Direction und eine Zielliste je Abgang
    assert '"FC_Direction"' in scl
    assert "//   ueber TRO02: SYS2, TRO02" in scl
    # Vario-TRO bekommt den Vario-Parameterblock
    assert "#TRO03.stInVario.Settings" in scl
    # Systemgrenzen sind keine Instanzen
    assert "SYS1 :" not in scl
    assert "Anfang bei TRO01" in scl