from __future__ import annotations import copy import hashlib import json from pathlib import Path import tempfile import unittest from unittest.mock import patch from app.simulation.benchmark_regression import summarize_simulation_result from app.simulation.physical_state_v21 import ( PHYSICAL_STATE_V21_ID, PhysicalStateV21Error, approve_candidate_golden, assess_report_against_amesim, build_amesim_reference, build_candidate_golden, compare_contract_to_amesim, evaluate_physical_state_v21, load_approved_golden, project_physical_state_v21, write_candidate_golden, ) REPO_ROOT = Path(__file__).resolve().parents[1] AMESIM_ARCHIVE = REPO_ROOT / "AmesimModels" / "test_mql.ame" CANDIDATE_REPORT = ( REPO_ROOT / "tests/baselines/simulation/test_mql_8/runs" / "2026-08-18-production-amesim-aligned-v1-candidate-0.2.json" ) def _variable(key: str, quantity: str, unit: str, category: str) -> dict[str, str]: return {"key": key, "quantity": quantity, "unit": unit, "category": category} def _synthetic_result() -> dict[str, object]: times = [0.0, 0.04, 0.2] variables: list[dict[str, str]] = [] series: dict[str, list[float]] = {"time": times} def add( key: str, values: list[float], quantity: str, unit: str, category: str, ) -> None: variables.append(_variable(key, quantity, unit, category)) series[key] = values add("amesim_pnch012_8.p", [100_000.0, 200_000.0, 300_000.0], "pressure", "Pa", "thermodynamic") add("amesim_pnl0001_20.p", [100_000.0, 210_000.0, 310_000.0], "pressure", "Pa", "thermodynamic") add("amesim_pnl0001_20.port_1.m_flow", [-0.1, -0.2, -0.3], "mass_flow", "kg/s", "port") add("amesim_pnch012_8.port_1.m_flow", [0.1, 0.2, 0.3], "mass_flow", "kg/s", "port") add("amesim_pnvo001_5.port_2.m_flow", [-0.2, -0.3, -0.4], "mass_flow", "kg/s", "port") add("amesim_pnvo001_5.port_3.m_flow", [0.2, 0.3, 0.4], "mass_flow", "kg/s", "port") for port, values in enumerate( ([0.4, 0.5, 0.6], [-0.1, -0.2, -0.3], [-0.2, -0.2, -0.2], [-0.1, -0.1, -0.1]), start=1, ): add(f"amesim_p4node2_8.port_{port}.m_flow", list(values), "mass_flow", "kg/s", "port") add("amesim_pnvo001_5.xv", [0.0, 1.0, 1.0], "dimensionless", "", "derived") add("amesim_mecmas21_9.x", [0.0, 0.0, 0.0], "length", "m", "state") add("amesim_mecmas21_9.v", [0.0, 0.0, 0.0], "velocity", "m/s", "state") add("amesim_mecmas21_9.port_1.f", [1.0, 1.0, 1.0], "force", "N", "port") add("amesim_mecmas21_9.port_2.f", [2.0, 2.0, 2.0], "force", "N", "port") add("amesim_mecmas21_10.x", [0.0, 0.0, 0.0], "length", "m", "state") add("amesim_mecmas21_10.v", [0.0, 0.0, 0.0], "velocity", "m/s", "state") add("amesim_mecmas21_10.port_1.f", [-2.0, -2.0, -2.0], "force", "N", "port") add("amesim_mecmas21_10.port_2.f", [-1.0, -1.0, -1.0], "force", "N", "port") add("storage_a.m", [1.0, 2.0, 3.0], "mass", "kg", "state") add("storage_b.m", [4.0, 3.0, 2.0], "mass", "kg", "state") return { "success": True, "status": "completed", "partial": False, "simulatedUntil": 0.2, "requestedStopTime": 0.2, "variables": variables, "series": series, "final": {key: values[-1] for key, values in series.items() if key != "time"}, "diagnostics": {}, } class PhysicalStateProjectionTests(unittest.TestCase): def test_projects_all_categories_with_units_and_strict_balances(self) -> None: contract = project_physical_state_v21( _synthetic_result(), checkpoint_times=(0.0, 0.04, 0.2), sample_step=0.04 ) self.assertEqual(contract["id"], PHYSICAL_STATE_V21_ID) self.assertEqual( contract["projectionCategories"], ["pressure", "massFlow", "conservation", "discreteMode"], ) self.assertEqual(len(contract["layout"]["projectionKeys"]), 11) endpoint = contract["checkpoints"][-1]["values"] self.assertEqual(endpoint["conservation.totalStoredGasMass"], 5.0) self.assertAlmostEqual(endpoint["conservation.p4node2_8.massBalance"], 0.0) self.assertAlmostEqual(endpoint["conservation.pnvo001_5.massBalance"], 0.0) self.assertAlmostEqual( endpoint["conservation.pnl0001_20_pnch012_8.connectionMassBalance"], 0.0 ) self.assertEqual(endpoint["discrete.pnvo001_5.openMode"], 1.0) self.assertEqual(endpoint["discrete.mecmas21_9.endstopMode"], 1.0) self.assertEqual(endpoint["discrete.mecmas21_10.endstopMode"], -1.0) pressure = contract["layout"]["projections"][0]["amesim"] self.assertEqual(pressure["offset"], 101_300.0) flow = contract["layout"]["projections"][2]["amesim"] self.assertEqual(flow["scale"], -1.0e-3) def test_fails_closed_on_wrong_source_unit(self) -> None: result = _synthetic_result() next( variable for variable in result["variables"] if variable["key"] == "amesim_pnch012_8.p" )["unit"] = "bar" with self.assertRaisesRegex(PhysicalStateV21Error, "Unexpected metadata"): project_physical_state_v21(result, checkpoint_times=(0.2,), sample_step=0.2) def test_benchmark_summary_attaches_v21_only_to_matching_result(self) -> None: summary = summarize_simulation_result( _synthetic_result(), checkpoint_times=(0.0, 0.2), sample_step=0.2 ) self.assertEqual(summary["physicalStateV21"]["id"], PHYSICAL_STATE_V21_ID) class AmesimReferenceTests(unittest.TestCase): def test_reference_records_archive_hash_units_signs_and_known_endpoint(self) -> None: reference = build_amesim_reference(AMESIM_ARCHIVE, checkpoint_times=(0.2,)) values = reference["checkpoints"][0]["values"] self.assertAlmostEqual(values["pressure.pnch012_8.absolute"], 3_625_053.2009413484) self.assertAlmostEqual( values["massFlow.pnl0001_20.port_1.intoComponent"], -0.3728806429214546, ) self.assertEqual(values["discrete.mecmas21_9.endstopMode"], 1.0) self.assertEqual(values["discrete.mecmas21_10.endstopMode"], -1.0) provenance = reference["provenance"] self.assertEqual(provenance["archivePath"], "AmesimModels/test_mql.ame") self.assertEqual(len(provenance["archiveSha256"]), 64) self.assertEqual( [member["name"] for member in provenance["members"]], ["test_mql_.var", "test_mql_.results"], ) def test_current_candidate_endpoint_passes_explicit_amesim_alignment_profile(self) -> None: assessment = assess_report_against_amesim( CANDIDATE_REPORT, archive_path=AMESIM_ARCHIVE ) comparison = assessment["comparison"] self.assertEqual(assessment["coverage"], "legacyEndpointOnly") self.assertTrue(comparison["passed"]) self.assertEqual(comparison["comparedValueCount"], 9) self.assertEqual(comparison["metricCount"], 11) self.assertEqual(comparison["availableBaselineValueCount"], 9) self.assertEqual(comparison["unavailableBaselineValueCount"], 2) self.assertEqual(len(comparison["metrics"]), 11) self.assertTrue( all( { "requestedTime", "key", "category", "actual", "amesimBaseline", "absoluteError", "relativeError", "evaluated", "passed", } <= set(metric) for metric in comparison["metrics"] ) ) self.assertEqual( comparison["worstValue"]["key"], "massFlow.pnl0001_20.port_1.intoComponent", ) self.assertAlmostEqual( comparison["worstValue"]["relativeError"], 0.0013934852135053246, ) self.assertEqual( comparison["toleranceProfile"]["byCategory"]["discreteMode"]["relative"], 0.0, ) def test_signal_jump_mass_flows_record_errors_but_are_not_evaluated(self) -> None: contract = project_physical_state_v21( _synthetic_result(), checkpoint_times=(0.0, 0.04, 0.2), sample_step=0.04 ) reference_checkpoints = [] for checkpoint in contract["checkpoints"]: values = dict(checkpoint["values"]) values["conservation.p4node2_8.massBalance"] = None values[ "conservation.pnl0001_20_pnch012_8.connectionMassBalance" ] = None if checkpoint["requestedTime"] == 0.04: values["massFlow.pnl0001_20.port_1.intoComponent"] += 100.0 values["massFlow.pnvo001_5.port_2.intoComponent"] -= 100.0 reference_checkpoints.append( {"requestedTime": checkpoint["requestedTime"], "values": values} ) comparison = compare_contract_to_amesim( contract, {"checkpoints": reference_checkpoints} ) excluded = [ metric for metric in comparison["metrics"] if metric["requestedTime"] == 0.04 and metric["category"] == "massFlow" ] self.assertTrue(comparison["passed"]) self.assertEqual(comparison["metricCount"], 33) self.assertEqual(comparison["availableBaselineValueCount"], 27) self.assertEqual(comparison["unavailableBaselineValueCount"], 6) self.assertEqual(comparison["comparedValueCount"], 25) self.assertEqual(comparison["excludedValueCount"], 2) self.assertEqual(len(comparison["metrics"]), 33) self.assertEqual(len(excluded), 2) self.assertTrue(all(metric["evaluated"] is False for metric in excluded)) self.assertTrue(all(metric["passed"] is None for metric in excluded)) self.assertTrue(all(metric["absoluteError"] > 99.0 for metric in excluded)) self.assertTrue( all("LeftRightLimit" in metric["exclusionReason"] for metric in excluded) ) self.assertNotIn(comparison["worstValue"], excluded) unavailable = [ metric for metric in comparison["metrics"] if metric["amesimBaseline"] is None ] self.assertEqual(len(unavailable), 6) self.assertTrue(all(metric["evaluated"] is False for metric in unavailable)) self.assertTrue(all(metric["absoluteError"] is None for metric in unavailable)) self.assertTrue(all(metric["relativeError"] is None for metric in unavailable)) def test_relative_error_uses_the_amesim_baseline_denominator(self) -> None: contract = project_physical_state_v21( _synthetic_result(), checkpoint_times=(0.2,), sample_step=0.2 ) values = dict(contract["checkpoints"][0]["values"]) key = "pressure.pnch012_8.absolute" values[key] = 0.5 * values[key] comparison = compare_contract_to_amesim( contract, {"checkpoints": [{"requestedTime": 0.2, "values": values}]}, ) metric = next(item for item in comparison["metrics"] if item["key"] == key) expected = abs(metric["actual"] - metric["amesimBaseline"]) / abs( metric["amesimBaseline"] ) self.assertAlmostEqual(metric["relativeError"], expected) self.assertAlmostEqual(metric["relativeErrorPercent"], 100.0 * expected) self.assertTrue(metric["relativeErrorDefined"]) def test_zero_amesim_baseline_has_undefined_relative_error(self) -> None: contract = project_physical_state_v21( _synthetic_result(), checkpoint_times=(0.2,), sample_step=0.2 ) values = dict(contract["checkpoints"][0]["values"]) key = "massFlow.pnl0001_20.port_1.intoComponent" values[key] = -0.0 comparison = compare_contract_to_amesim( contract, {"checkpoints": [{"requestedTime": 0.2, "values": values}]}, ) metric = next(item for item in comparison["metrics"] if item["key"] == key) self.assertIsNone(metric["relativeError"]) self.assertIsNone(metric["relativeErrorPercent"]) self.assertFalse(metric["relativeErrorDefined"]) self.assertEqual(metric["relativeErrorReason"], "zeroAmesimBaseline") self.assertEqual(metric["comparisonBasis"], "absoluteNearZero") self.assertEqual( json.loads(json.dumps(comparison, allow_nan=False))["baselineAuthority"], "amesim", ) class ReviewedGoldenProtocolTests(unittest.TestCase): def _report_and_reference(self): contract = project_physical_state_v21( _synthetic_result(), checkpoint_times=(0.0, 0.04, 0.2), sample_step=0.04 ) report = { "generatedAt": "2026-08-18T00:00:00+00:00", "source": {"sha256": "a" * 64}, "cases": [{ "caseId": "0.2s", "lane": "production", "worker": {"summary": {"physicalStateV21": contract}}, }], } keys = contract["layout"]["projectionKeys"] reference_checkpoints = [] for checkpoint in contract["checkpoints"]: reference_checkpoints.append({ "requestedTime": checkpoint["requestedTime"], "values": {key: checkpoint["values"][key] for key in keys}, }) reference = { "schemaVersion": 1, "provenance": { "archivePath": "AmesimModels/synthetic.ame", "archiveBytes": 2, "archiveSha256": "b" * 64, "members": [ {"name": "synthetic.var", "bytes": 1, "sha256": "c" * 64}, { "name": "synthetic.results", "bytes": 1, "sha256": "d" * 64, }, ], }, "storedMassDataPaths": ["mgas@synthetic"], "checkpoints": reference_checkpoints, } return contract, report, reference def test_candidate_requires_exact_hash_and_is_approved_to_a_new_file(self) -> None: contract, report, reference = self._report_and_reference() with tempfile.TemporaryDirectory( dir=REPO_ROOT, prefix=".physical-state-v21-test-" ) as temporary_directory: root = Path(temporary_directory) report_path = root / "report.json" report_path.write_text(json.dumps(report), encoding="utf-8") candidate_path = root / "candidate.json" approved_path = root / "approved.json" with patch( "app.simulation.physical_state_v21.build_amesim_reference", return_value=reference, ): write_candidate_golden(report_path, candidate_path) candidate = json.loads(candidate_path.read_text(encoding="utf-8")) local_report = candidate["provenance"]["localSourceReport"] self.assertFalse(Path(local_report["path"]).is_absolute()) self.assertEqual(local_report["bytes"], report_path.stat().st_size) candidate_sha = hashlib.sha256(candidate_path.read_bytes()).hexdigest() with self.assertRaisesRegex(PhysicalStateV21Error, "SHA mismatch"): approve_candidate_golden( candidate_path, approved_path, expected_candidate_sha256="0" * 64, reviewed_by="reviewer", note="reviewed", ) approve_candidate_golden( candidate_path, approved_path, expected_candidate_sha256=candidate_sha, reviewed_by="reviewer", note="units, signs, and errors reviewed", ) golden = load_approved_golden(approved_path) self.assertFalse(Path(golden["approval"]["candidatePath"]).is_absolute()) golden["checkpoints"][-1]["values"][0] += 1.0e9 self.assertTrue(evaluate_physical_state_v21(contract, golden)["passed"]) changed = copy.deepcopy(contract) changed["checkpoints"][-1]["values"]["pressure.pnch012_8.absolute"] += 1000.0 failure = evaluate_physical_state_v21(changed, golden) self.assertFalse(failure["passed"]) self.assertIn("amesimReferenceToleranceMismatch", failure["issues"]) def test_approval_rejects_failed_or_tampered_alignment(self) -> None: _contract, report, reference = self._report_and_reference() with tempfile.TemporaryDirectory() as temporary_directory: root = Path(temporary_directory) report_path = root / "report.json" report_path.write_text(json.dumps(report), encoding="utf-8") original_path = root / "candidate.json" with patch( "app.simulation.physical_state_v21.build_amesim_reference", return_value=reference, ): write_candidate_golden(report_path, original_path) original = json.loads(original_path.read_text(encoding="utf-8")) failed_alignment = copy.deepcopy(original) failed_alignment["amesimAlignmentAtGeneration"]["passed"] = False failed_path = root / "failed-alignment.json" failed_path.write_text(json.dumps(failed_alignment), encoding="utf-8") with self.assertRaisesRegex( PhysicalStateV21Error, "AMESim alignment at generation" ): approve_candidate_golden( failed_path, root / "failed-approved.json", expected_candidate_sha256=hashlib.sha256( failed_path.read_bytes() ).hexdigest(), reviewed_by="reviewer", note="must not override a failed alignment", ) failed_metric = copy.deepcopy(original) evaluated_metric = next( metric for metric in failed_metric["amesimAlignmentAtGeneration"]["metrics"] if metric["evaluated"] ) evaluated_metric["passed"] = False failed_metric_path = root / "failed-metric.json" failed_metric_path.write_text(json.dumps(failed_metric), encoding="utf-8") with self.assertRaisesRegex(PhysicalStateV21Error, "evaluated metric"): approve_candidate_golden( failed_metric_path, root / "failed-metric-approved.json", expected_candidate_sha256=hashlib.sha256( failed_metric_path.read_bytes() ).hexdigest(), reviewed_by="reviewer", note="must not override a failed metric", ) def test_approval_rejects_local_conservation_residual(self) -> None: _contract, report, reference = self._report_and_reference() with tempfile.TemporaryDirectory() as temporary_directory: root = Path(temporary_directory) report_path = root / "report.json" report_path.write_text(json.dumps(report), encoding="utf-8") original_path = root / "candidate.json" with patch( "app.simulation.physical_state_v21.build_amesim_reference", return_value=reference, ): write_candidate_golden(report_path, original_path) candidate = json.loads(original_path.read_text(encoding="utf-8")) key = "conservation.p4node2_8.massBalance" value_index = candidate["layout"]["projectionKeys"].index(key) candidate["checkpoints"][1]["values"][value_index] = 2.0e-9 tampered_path = root / "conservation-residual.json" tampered_path.write_text(json.dumps(candidate), encoding="utf-8") with self.assertRaisesRegex( PhysicalStateV21Error, "local conservation residual" ): approve_candidate_golden( tampered_path, root / "residual-approved.json", expected_candidate_sha256=hashlib.sha256( tampered_path.read_bytes() ).hexdigest(), reviewed_by="reviewer", note="must not override a local conservation failure", ) def test_candidate_writer_refuses_overwrite_and_unapproved_file_is_rejected(self) -> None: _contract, report, reference = self._report_and_reference() with tempfile.TemporaryDirectory() as temporary_directory: root = Path(temporary_directory) report_path = root / "report.json" report_path.write_text(json.dumps(report), encoding="utf-8") output = root / "candidate.json" with patch( "app.simulation.physical_state_v21.build_amesim_reference", return_value=reference, ): write_candidate_golden(report_path, output) with self.assertRaisesRegex(PhysicalStateV21Error, "overwrite"): write_candidate_golden(report_path, output) with self.assertRaisesRegex(PhysicalStateV21Error, "approval.status"): load_approved_golden(output) if __name__ == "__main__": unittest.main()