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))