281 lines
9.8 KiB
Python
281 lines
9.8 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import unittest
|
|
from xml.etree import ElementTree as ET
|
|
|
|
from fastapi import HTTPException, Request
|
|
|
|
from app.main import (
|
|
ReactFlowEdgePayload,
|
|
ReactFlowNodePayload,
|
|
build_reactflow_system_xml,
|
|
compile_system_xml_model,
|
|
compile_system_xml_network,
|
|
parse_system_xml,
|
|
)
|
|
from app.system_xml import validate_system_xml_document
|
|
from tests.test_system_xml_protocol import (
|
|
physical_connection_project,
|
|
physical_port,
|
|
)
|
|
|
|
|
|
def valid_xml() -> bytes:
|
|
return build_reactflow_system_xml(physical_connection_project())
|
|
|
|
|
|
def issue_codes(xml: bytes | str) -> set[str]:
|
|
return {
|
|
issue.code
|
|
for issue in validate_system_xml_document(xml).issues
|
|
}
|
|
|
|
|
|
def with_parameter_value(
|
|
xml: bytes,
|
|
component_id: str,
|
|
parameter_name: str,
|
|
value: float,
|
|
) -> bytes:
|
|
root = ET.fromstring(xml)
|
|
parameter = root.find(
|
|
"./Components/Component[@id='{}']/Parameter[@name='{}']".format(
|
|
component_id,
|
|
parameter_name,
|
|
)
|
|
)
|
|
assert parameter is not None
|
|
parameter.set("value", str(value))
|
|
return ET.tostring(root, encoding="utf-8", xml_declaration=True)
|
|
|
|
|
|
def without_parameter(
|
|
xml: bytes,
|
|
component_id: str,
|
|
parameter_name: str,
|
|
) -> bytes:
|
|
root = ET.fromstring(xml)
|
|
component = root.find(f"./Components/Component[@id='{component_id}']")
|
|
assert component is not None
|
|
parameter = component.find(f"./Parameter[@name='{parameter_name}']")
|
|
assert parameter is not None
|
|
component.remove(parameter)
|
|
return ET.tostring(root, encoding="utf-8", xml_declaration=True)
|
|
|
|
|
|
def xml_request(body: bytes) -> Request:
|
|
delivered = False
|
|
|
|
async def receive():
|
|
nonlocal delivered
|
|
if delivered:
|
|
return {"type": "http.disconnect"}
|
|
delivered = True
|
|
return {"type": "http.request", "body": body, "more_body": False}
|
|
|
|
return Request(
|
|
{
|
|
"type": "http",
|
|
"method": "POST",
|
|
"path": "/api/system-xml/compile-model",
|
|
"headers": [(b"content-type", b"application/xml")],
|
|
},
|
|
receive,
|
|
)
|
|
|
|
|
|
class SystemXmlParserTests(unittest.TestCase):
|
|
def test_valid_v3_xml_parses_and_compiles_directly(self) -> None:
|
|
report = validate_system_xml_document(valid_xml())
|
|
|
|
self.assertTrue(report.valid)
|
|
self.assertEqual(report.issues, ())
|
|
assert report.document is not None
|
|
network = compile_system_xml_network(report.document)
|
|
|
|
self.assertEqual(report.document.name, "protocol-test")
|
|
self.assertEqual(report.document.simulation.method, "BDF")
|
|
self.assertEqual(report.document.simulation.sample_step, 0.2)
|
|
self.assertEqual(set(network.components), {"cylinder_1", "tank_1"})
|
|
self.assertEqual(len(network.connections), 1)
|
|
self.assertEqual(network.connections[0].id, "edge-1")
|
|
|
|
def test_compile_api_consumes_raw_xml_body(self) -> None:
|
|
response = asyncio.run(compile_system_xml_model(xml_request(valid_xml())))
|
|
|
|
self.assertTrue(response["success"])
|
|
self.assertEqual(response["name"], "protocol-test")
|
|
self.assertEqual(len(response["components"]), 2)
|
|
self.assertEqual(len(response["connections"]), 1)
|
|
self.assertEqual(response["simulation"]["sample_step"], 0.2)
|
|
self.assertTrue(response["validation"]["valid"])
|
|
|
|
def test_parse_api_returns_execution_model_not_editor_project(self) -> None:
|
|
response = asyncio.run(parse_system_xml(xml_request(valid_xml())))
|
|
|
|
self.assertTrue(response["success"])
|
|
self.assertNotIn("project", response)
|
|
model = response["model"]
|
|
self.assertEqual(model["schemaVersion"], "3")
|
|
self.assertEqual(model["unitSystem"], "SI")
|
|
self.assertEqual(model["name"], "protocol-test")
|
|
self.assertEqual(model["simulation"]["sample_step"], 0.2)
|
|
self.assertEqual(model["components"][0]["model_type"], "cylinder")
|
|
self.assertEqual(model["components"][0]["model_version"], "1.0.0")
|
|
self.assertNotIn("nodes", model)
|
|
self.assertNotIn("edges", model)
|
|
|
|
def test_parse_api_returns_structured_422_diagnostics(self) -> None:
|
|
with self.assertRaises(HTTPException) as caught:
|
|
asyncio.run(parse_system_xml(xml_request(b"<broken>")))
|
|
|
|
self.assertEqual(caught.exception.status_code, 422)
|
|
detail = caught.exception.detail
|
|
self.assertEqual(detail["message"], "System XML validation failed.")
|
|
self.assertEqual(detail["issues"][0]["layer"], "xml")
|
|
self.assertEqual(detail["issues"][0]["code"], "XML_SYNTAX_ERROR")
|
|
|
|
def test_malformed_xml_reports_xml_layer_error(self) -> None:
|
|
report = validate_system_xml_document("<System><broken></System>")
|
|
|
|
self.assertFalse(report.valid)
|
|
self.assertIsNone(report.document)
|
|
self.assertEqual(report.issues[0].layer, "xml")
|
|
self.assertEqual(report.issues[0].code, "XML_SYNTAX_ERROR")
|
|
self.assertIsNotNone(report.issues[0].line)
|
|
|
|
def test_dtd_is_rejected_before_schema_validation(self) -> None:
|
|
xml = valid_xml().replace(
|
|
b"<System ",
|
|
b"<!DOCTYPE System [<!ENTITY demo 'unsafe'>]>\n<System ",
|
|
1,
|
|
)
|
|
|
|
report = validate_system_xml_document(xml)
|
|
|
|
self.assertFalse(report.valid)
|
|
self.assertEqual(report.issues[0].layer, "xml")
|
|
self.assertEqual(report.issues[0].code, "XML_DTD_NOT_ALLOWED")
|
|
|
|
def test_wrong_schema_version_reports_xsd_error(self) -> None:
|
|
xml = valid_xml().replace(b'schemaVersion="3"', b'schemaVersion="2"')
|
|
report = validate_system_xml_document(xml)
|
|
|
|
self.assertFalse(report.valid)
|
|
self.assertIsNone(report.document)
|
|
self.assertTrue(report.issues)
|
|
self.assertTrue(all(issue.layer == "schema" for issue in report.issues))
|
|
self.assertIn("XSD_VALIDATION_ERROR", issue_codes(xml))
|
|
|
|
def test_parameter_range_is_checked_semantically(self) -> None:
|
|
xml = with_parameter_value(valid_xml(), "cylinder_1", "volume", -1)
|
|
|
|
report = validate_system_xml_document(xml)
|
|
|
|
self.assertFalse(report.valid)
|
|
self.assertIn("PARAMETER_VALUE_INVALID", issue_codes(xml))
|
|
issue = next(
|
|
issue for issue in report.issues if issue.code == "PARAMETER_VALUE_INVALID"
|
|
)
|
|
self.assertEqual(issue.layer, "semantic")
|
|
self.assertIn("cylinder_1.volume", issue.message)
|
|
|
|
def test_zero_initial_pressure_is_rejected_before_model_creation(self) -> None:
|
|
xml = with_parameter_value(valid_xml(), "cylinder_1", "p0", 0)
|
|
|
|
report = validate_system_xml_document(xml)
|
|
|
|
self.assertFalse(report.valid)
|
|
self.assertIn("PARAMETER_VALUE_INVALID", issue_codes(xml))
|
|
|
|
def test_endpoint_port_is_checked_against_registered_contract(self) -> None:
|
|
xml = valid_xml().replace(
|
|
b'component="cylinder_1" port="port_b"',
|
|
b'component="cylinder_1" port="missing_port"',
|
|
1,
|
|
)
|
|
|
|
self.assertIn("ENDPOINT_PORT_UNKNOWN", issue_codes(xml))
|
|
|
|
def test_unknown_connection_endpoint_is_reported(self) -> None:
|
|
xml = valid_xml().replace(
|
|
b'component="tank_1" port="port_a"',
|
|
b'component="missing_tank" port="port_a"',
|
|
1,
|
|
)
|
|
|
|
self.assertIn("ENDPOINT_COMPONENT_UNKNOWN", issue_codes(xml))
|
|
|
|
def test_physical_port_cannot_be_used_by_two_connections(self) -> None:
|
|
project = physical_connection_project()
|
|
project.nodes.append(
|
|
ReactFlowNodePayload(**{
|
|
"id": "tank_2",
|
|
"type": "simulationComponent",
|
|
"position": {"x": 420, "y": 180},
|
|
"data": {
|
|
"label": "receiver-tank-2",
|
|
"componentType": "tank",
|
|
"modelType": "tank",
|
|
"modelVersion": "1.0.0",
|
|
"ports": [physical_port("port_a", "inlet", "left")],
|
|
"parameters": {"volume": 0.1, "p0": 100000, "T0": 300},
|
|
},
|
|
})
|
|
)
|
|
project.edges.append(
|
|
ReactFlowEdgePayload(**{
|
|
"id": "edge-2",
|
|
"source": "cylinder_1",
|
|
"sourceHandle": "port_b",
|
|
"target": "tank_2",
|
|
"targetHandle": "port_a",
|
|
})
|
|
)
|
|
|
|
with self.assertRaisesRegex(
|
|
ValueError,
|
|
"PHYSICAL_PORT_ALREADY_CONNECTED",
|
|
):
|
|
build_reactflow_system_xml(project)
|
|
|
|
def test_unconnected_registered_port_is_a_warning(self) -> None:
|
|
project = physical_connection_project()
|
|
project.edges.clear()
|
|
|
|
report = validate_system_xml_document(build_reactflow_system_xml(project))
|
|
|
|
self.assertTrue(report.valid)
|
|
self.assertEqual(report.as_dict()["warningCount"], 2)
|
|
self.assertEqual(
|
|
{issue.code for issue in report.issues},
|
|
{"PORT_UNCONNECTED"},
|
|
)
|
|
|
|
def test_missing_required_parameter_is_reported(self) -> None:
|
|
xml = without_parameter(valid_xml(), "cylinder_1", "T0")
|
|
|
|
self.assertIn("PARAMETER_REQUIRED_MISSING", issue_codes(xml))
|
|
|
|
def test_unsupported_solver_method_is_reported(self) -> None:
|
|
xml = valid_xml().replace(b'method="BDF"', b'method="unknown"')
|
|
|
|
self.assertIn("SIMULATION_METHOD_UNSUPPORTED", issue_codes(xml))
|
|
|
|
def test_non_positive_simulation_steps_are_rejected_by_schema(self) -> None:
|
|
xml = valid_xml().replace(b'sampleStep="0.2"', b'sampleStep="0"')
|
|
xml = xml.replace(b'maxStep="0.01"', b'maxStep="-1"')
|
|
|
|
report = validate_system_xml_document(xml)
|
|
|
|
self.assertFalse(report.valid)
|
|
self.assertEqual(
|
|
[issue.code for issue in report.issues].count("XSD_VALIDATION_ERROR"),
|
|
2,
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|