152 lines
8.5 KiB
Python
152 lines
8.5 KiB
Python
from copy import deepcopy
|
|
import json
|
|
from pathlib import Path
|
|
import unittest
|
|
from urllib.parse import unquote
|
|
from xml.etree import ElementTree as ET
|
|
|
|
from tests.test_native_result_transport import AsgiClient
|
|
|
|
from app.main import app, ReactFlowProjectPayload, build_reactflow_system_xml
|
|
from app.project_parameters import expression_value, prepare_project
|
|
from app.simulation.native_codegen.input import project_xml
|
|
from tests.test_pressure_units import pressure_project
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def param(xml, name="p0"):
|
|
return float(ET.fromstring(xml).find(f"./Components/Component[@id='chamber_1']/Parameter[@name='{name}']").get("value"))
|
|
|
|
|
|
class ProjectInputContractTests(unittest.TestCase):
|
|
def test_shared_expression_grammar(self):
|
|
cases = json.loads((ROOT / "tests/fixtures/parameter-expressions.json").read_text())
|
|
for source, expected in cases["valid"]:
|
|
with self.subTest(source=source):
|
|
self.assertAlmostEqual(expression_value(source), expected, places=12)
|
|
for source in cases["invalid"] + ["(" * 34 + "1" + ")" * 34, "1+" * 256 + "1", "1" * 513, "min(" + ",".join(["1"] * 17) + ")"]:
|
|
with self.subTest(source=source):
|
|
with self.assertRaises(ValueError):
|
|
expression_value(source)
|
|
|
|
def test_v2_numeric_string_and_expression_have_same_si_value_on_http_and_cli(self):
|
|
client = AsgiClient(app)
|
|
for unit, magnitude in (("bar", 2.5), ("kPa", 250), ("MPa", .25), ("Pa", 250000)):
|
|
for value in (magnitude, str(magnitude), f"={magnitude}", f"sqrt({magnitude}^2)"):
|
|
with self.subTest(unit=unit, value=value):
|
|
project = pressure_project(value, unit)
|
|
project["projectSchemaVersion"] = 2
|
|
original = deepcopy(project)
|
|
xml = project_xml(project)
|
|
response = client.post("/api/reactflow/system-xml", content=json.dumps(project).encode(), headers={"content-type": "application/json"})
|
|
self.assertEqual(response.status_code, 200, response.content)
|
|
self.assertEqual(response.content, xml)
|
|
self.assertAlmostEqual(param(xml), 250000)
|
|
self.assertEqual(project, original)
|
|
|
|
def test_affine_temperature_and_si_defaults_are_not_double_converted(self):
|
|
for value in (20, "20", "=10+10"):
|
|
project = pressure_project(2.5)
|
|
project["projectSchemaVersion"] = 2
|
|
data = project["nodes"][0]["data"]
|
|
data["parameters"]["T0"] = value
|
|
data["parameterUnits"]["T0"] = "degC"
|
|
self.assertEqual(param(project_xml(project), "T0"), 293.15)
|
|
del data["parameters"]["T0"]
|
|
# Omitted defaults remain defined in SI, independent of display metadata.
|
|
self.assertEqual(param(project_xml(project), "T0"), 293.15)
|
|
|
|
def test_old_versions_warn_at_adapter_but_core_remains_strict(self):
|
|
client = AsgiClient(app)
|
|
for version in (None, "0.0.1", "99.0.0"):
|
|
project = pressure_project(250000)
|
|
project["nodes"][0]["data"]["modelVersion"] = version
|
|
original = deepcopy(project)
|
|
response = client.post("/api/reactflow/system-xml", content=json.dumps(project).encode(), headers={"content-type": "application/json"})
|
|
self.assertEqual(response.status_code, 200, response.content)
|
|
warning = json.loads(unquote(response.headers["x-component-version-warnings"]))
|
|
self.assertEqual(warning["components"][0]["storedVersion"], version)
|
|
self.assertEqual(param(response.content), 250000)
|
|
self.assertEqual(project, original)
|
|
with self.assertRaisesRegex(ValueError, "MODEL_VERSION"):
|
|
build_reactflow_system_xml(ReactFlowProjectPayload.model_validate(project))
|
|
|
|
def test_compile_http_adapter_uses_current_version_and_si(self):
|
|
client = AsgiClient(app)
|
|
project = pressure_project("sqrt(6.25)")
|
|
project["projectSchemaVersion"] = 2
|
|
project["nodes"][0]["data"]["modelVersion"] = "0.0.1"
|
|
response = client.post("/api/reactflow/compile-model", content=json.dumps(project).encode(),
|
|
headers={"content-type": "application/json"})
|
|
self.assertEqual(response.status_code, 200, response.content)
|
|
self.assertEqual(response.json()["warnings"][0]["components"][0]["storedVersion"], "0.0.1")
|
|
|
|
def test_header_warning_stays_bounded_for_large_old_projects(self):
|
|
data = json.loads((ROOT / "tests/data/test-mql-8-corrected.json").read_text())
|
|
for node in data["nodes"]:
|
|
node["data"]["modelVersion"] = "0.0.1"
|
|
response = AsgiClient(app).post("/api/reactflow/system-xml", content=json.dumps(data).encode(),
|
|
headers={"content-type": "application/json"})
|
|
self.assertEqual(response.status_code, 200, response.content)
|
|
header = response.headers["x-component-version-warnings"]
|
|
self.assertLessEqual(len(header), 3800)
|
|
self.assertEqual(json.loads(unquote(header))["totalCount"], len(data["nodes"]))
|
|
|
|
def test_old_version_does_not_bypass_structural_or_parameter_errors(self):
|
|
client = AsgiClient(app)
|
|
for change in ("type", "port", "parameter", "range", "unit", "discrete"):
|
|
project = pressure_project(250000)
|
|
data = project["nodes"][0]["data"]
|
|
data["modelVersion"] = "0.0.1"
|
|
if change == "type": data["componentType"] = "tank"
|
|
if change == "port": data["ports"][0]["name"] = "unknown"
|
|
if change == "parameter": data["parameters"]["unknown"] = 1
|
|
if change == "range": data["parameters"]["p0"] = -1
|
|
if change == "unit": data["parameterUnits"]["p0"] = "psi"
|
|
if change == "discrete": data["parameters"]["gi"] = "=0"
|
|
with self.subTest(change=change):
|
|
self.assertEqual(client.post("/api/reactflow/system-xml", content=json.dumps(project).encode(), headers={"content-type": "application/json"}).status_code, 400)
|
|
|
|
def test_strict_execution_boundary_rejects_expressions_and_nonfinite(self):
|
|
for value in ("=2.5", "250000", float("inf"), float("nan"), True):
|
|
with self.subTest(value=value):
|
|
with self.assertRaisesRegex(ValueError, "finite SI number"):
|
|
build_reactflow_system_xml(ReactFlowProjectPayload.model_validate(pressure_project(value)))
|
|
project = ReactFlowProjectPayload.model_validate(pressure_project(2.5))
|
|
project.projectSchemaVersion = 2
|
|
with self.assertRaisesRegex(ValueError, "normalized SI"):
|
|
build_reactflow_system_xml(project)
|
|
|
|
def test_simulation_expressions_are_also_normalized_before_execution(self):
|
|
project = ReactFlowProjectPayload.model_validate(pressure_project(250000))
|
|
project.simulation.t_stop = "=1/500"
|
|
normalized, _ = prepare_project(project)
|
|
self.assertEqual(normalized.simulation.t_stop, .002)
|
|
self.assertEqual(project.simulation.t_stop, "=1/500")
|
|
with self.assertRaisesRegex(ValueError, "finite SI"):
|
|
build_reactflow_system_xml(project)
|
|
|
|
def test_legacy_lmechn1_does_not_guess_a_missing_dynamic_port_count(self):
|
|
data = json.loads((ROOT / "tests/data/test-mql-8-corrected.json").read_text())
|
|
legacy = next(n for n in data["nodes"] if n["data"]["modelType"] == "amesim_lmechn1")
|
|
legacy["data"]["modelVersion"] = "0.1.0"
|
|
legacy["data"]["ports"] = legacy["data"]["ports"][:9]
|
|
del legacy["data"]["parameters"]["v1"]
|
|
normalized, notices = prepare_project(ReactFlowProjectPayload.model_validate(data))
|
|
self.assertTrue(notices)
|
|
with self.assertRaisesRegex(ValueError, "port names"):
|
|
build_reactflow_system_xml(normalized)
|
|
|
|
def test_eight_branch_legacy_project_stays_unchanged(self):
|
|
data = json.loads((ROOT / "tests/data/test-mql-8-corrected.json").read_text())
|
|
original = deepcopy(data)
|
|
normalized, warnings = prepare_project(ReactFlowProjectPayload.model_validate(data))
|
|
self.assertFalse(warnings)
|
|
self.assertEqual(data, original)
|
|
for source, target in zip(data["nodes"], normalized.nodes):
|
|
for name, value in source["data"]["parameters"].items():
|
|
if isinstance(value, (int, float)):
|
|
self.assertEqual(value, target.data.parameters[name])
|
|
self.assertIn(b'unitSystem="SI"', build_reactflow_system_xml(normalized))
|