215 lines
6.9 KiB
Python
215 lines
6.9 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from xml.etree import ElementTree as ET
|
|
|
|
from app.main import ReactFlowProjectPayload, build_reactflow_system_xml
|
|
|
|
|
|
def physical_port(
|
|
name: str,
|
|
role: str,
|
|
side: str,
|
|
*,
|
|
domain: str = "pneumatic",
|
|
) -> dict[str, str]:
|
|
return {
|
|
"name": name,
|
|
"kind": "physical",
|
|
"domain": domain,
|
|
"nominalRole": role,
|
|
"positiveFlowDirection": "intoComponent",
|
|
"side": side,
|
|
}
|
|
|
|
|
|
def physical_connection_project(*, reverse_edge: bool = False) -> ReactFlowProjectPayload:
|
|
source = {
|
|
"id": "cylinder_1",
|
|
"type": "simulationComponent",
|
|
"position": {"x": 12.5, "y": 24.0},
|
|
"data": {
|
|
"label": "source-cylinder",
|
|
"componentType": "cylinder",
|
|
"modelType": "cylinder",
|
|
"modelVersion": "1.0.0",
|
|
"ports": [physical_port("port_b", "outlet", "right")],
|
|
"parameters": {"volume": 0.01, "p0": 35000000, "T0": 300},
|
|
},
|
|
}
|
|
target = {
|
|
"id": "tank_1",
|
|
"type": "simulationComponent",
|
|
"position": {"x": 420.0, "y": 24.0},
|
|
"data": {
|
|
"label": "receiver-tank",
|
|
"componentType": "tank",
|
|
"modelType": "tank",
|
|
"modelVersion": "1.0.0",
|
|
"ports": [physical_port("port_a", "inlet", "left")],
|
|
"parameters": {"volume": 0.1, "p0": 100000, "T0": 300},
|
|
},
|
|
}
|
|
edge = (
|
|
{
|
|
"id": "edge-1",
|
|
"source": "tank_1",
|
|
"sourceHandle": "port_a",
|
|
"target": "cylinder_1",
|
|
"targetHandle": "port_b",
|
|
}
|
|
if reverse_edge
|
|
else {
|
|
"id": "edge-1",
|
|
"source": "cylinder_1",
|
|
"sourceHandle": "port_b",
|
|
"target": "tank_1",
|
|
"targetHandle": "port_a",
|
|
}
|
|
)
|
|
return ReactFlowProjectPayload(
|
|
projectSchemaVersion=1,
|
|
name="protocol-test",
|
|
nodes=[source, target],
|
|
edges=[edge],
|
|
simulation={
|
|
"t_start": 1.0,
|
|
"t_stop": 5.0,
|
|
"step": 0.2,
|
|
"max_step": 0.01,
|
|
"method": "BDF",
|
|
},
|
|
)
|
|
|
|
|
|
class SystemXmlProtocolTests(unittest.TestCase):
|
|
def test_v3_xml_contains_only_execution_model_fields(self) -> None:
|
|
root = ET.fromstring(build_reactflow_system_xml(physical_connection_project()))
|
|
|
|
self.assertEqual(root.tag, "System")
|
|
self.assertEqual(
|
|
root.attrib,
|
|
{
|
|
"name": "protocol-test",
|
|
"schemaVersion": "3",
|
|
"unitSystem": "SI",
|
|
},
|
|
)
|
|
self.assertEqual(
|
|
[child.tag for child in root],
|
|
["Simulation", "Components", "Connections"],
|
|
)
|
|
|
|
simulation = root.find("Simulation")
|
|
assert simulation is not None
|
|
self.assertEqual(
|
|
simulation.attrib,
|
|
{
|
|
"tStart": "1.0",
|
|
"tStop": "5.0",
|
|
"sampleStep": "0.2",
|
|
"maxStep": "0.01",
|
|
"method": "BDF",
|
|
},
|
|
)
|
|
|
|
component = root.find("./Components/Component[@id='cylinder_1']")
|
|
assert component is not None
|
|
self.assertEqual(
|
|
component.attrib,
|
|
{
|
|
"id": "cylinder_1",
|
|
"type": "cylinder",
|
|
"modelVersion": "1.0.0",
|
|
},
|
|
)
|
|
self.assertEqual(component.findall("Port"), [])
|
|
self.assertEqual(
|
|
{
|
|
parameter.attrib["name"]: float(parameter.attrib["value"])
|
|
for parameter in component.findall("Parameter")
|
|
},
|
|
{"volume": 0.01, "p0": 35_000_000.0, "T0": 300.0},
|
|
)
|
|
|
|
connection = root.find("./Connections/Connection")
|
|
assert connection is not None
|
|
self.assertEqual(connection.attrib, {"id": "edge-1"})
|
|
self.assertEqual(
|
|
{
|
|
(endpoint.attrib["component"], endpoint.attrib["port"])
|
|
for endpoint in connection.findall("Endpoint")
|
|
},
|
|
{("cylinder_1", "port_b"), ("tank_1", "port_a")},
|
|
)
|
|
self.assertTrue(all(len(endpoint.attrib) == 2 for endpoint in connection))
|
|
|
|
def test_physical_connection_meaning_does_not_depend_on_drag_direction(self) -> None:
|
|
forward = ET.fromstring(
|
|
build_reactflow_system_xml(physical_connection_project())
|
|
)
|
|
reverse = ET.fromstring(
|
|
build_reactflow_system_xml(physical_connection_project(reverse_edge=True))
|
|
)
|
|
|
|
def endpoint_set(root: ET.Element) -> set[tuple[str, str]]:
|
|
return {
|
|
(endpoint.attrib["component"], endpoint.attrib["port"])
|
|
for endpoint in root.findall("./Connections/Connection/Endpoint")
|
|
}
|
|
|
|
self.assertEqual(endpoint_set(forward), endpoint_set(reverse))
|
|
|
|
def test_editor_layout_and_display_fields_are_not_exported(self) -> None:
|
|
project = physical_connection_project()
|
|
project.nodes[0].position.x = 987
|
|
project.nodes[0].position.y = 654
|
|
project.nodes[0].data.label = "display-only-name"
|
|
project.nodes[0].data.rotation = 90
|
|
project.nodes[0].data.mirrored = True
|
|
|
|
root = ET.fromstring(build_reactflow_system_xml(project))
|
|
component = root.find("./Components/Component[@id='cylinder_1']")
|
|
assert component is not None
|
|
self.assertEqual(
|
|
component.attrib,
|
|
{
|
|
"id": "cylinder_1",
|
|
"type": "cylinder",
|
|
"modelVersion": "1.0.0",
|
|
},
|
|
)
|
|
|
|
def test_incompatible_physical_domains_are_rejected(self) -> None:
|
|
project = physical_connection_project()
|
|
tank_port = project.nodes[1].data.ports[0]
|
|
assert not isinstance(tank_port, str)
|
|
tank_port.domain = "hydraulic"
|
|
|
|
with self.assertRaisesRegex(ValueError, "incompatible port type"):
|
|
build_reactflow_system_xml(project)
|
|
|
|
def test_exporter_never_returns_xml_that_fails_the_v3_contract(self) -> None:
|
|
invalid_time = physical_connection_project()
|
|
invalid_time.simulation.t_stop = invalid_time.simulation.t_start
|
|
with self.assertRaisesRegex(ValueError, "SIMULATION_TIME_RANGE_INVALID"):
|
|
build_reactflow_system_xml(invalid_time)
|
|
|
|
duplicate_component = physical_connection_project()
|
|
duplicate_component.nodes[1].id = duplicate_component.nodes[0].id
|
|
duplicate_component.edges = []
|
|
with self.assertRaisesRegex(ValueError, "COMPONENT_ID_DUPLICATE"):
|
|
build_reactflow_system_xml(duplicate_component)
|
|
|
|
def test_blank_project_name_is_omitted(self) -> None:
|
|
project = physical_connection_project()
|
|
project.name = " "
|
|
|
|
root = ET.fromstring(build_reactflow_system_xml(project))
|
|
|
|
self.assertNotIn("name", root.attrib)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|