Michael StanglandClaude Opus 5.5
29077caa81
- OpenPoint -> model.TroCandidate (type, position, belongsTo, note); TroDocument.candidates ersetzt openPoints, Schema sps_skel-tro-v3 - Positionsbehandlung von TroBase und TroCandidate im Mixin _Positioned - tro_overrides.py: eigene ConnectionSpec/OpenPoint entfallen, stattdessen model.Connection / model.TroCandidate; einheitliche Feldnamen (tro_id/separator, hosts als Liste) - Fixtures, Beispiel-JSON und Doku auf v3 nachgezogen Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
139 lines
4.9 KiB
Python
139 lines
4.9 KiB
Python
# -*- 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
|