"""Supply errors are rejected at UI-independent model and native entry points.""" import unittest from xml.etree import ElementTree as ET from app.main import ReactFlowProjectPayload, build_reactflow_system_xml, compile_reactflow_network from app.simulation.core.port_computation import ( PortSupplyError, port_supply_issue, reference_supply_issues, ) from app.simulation.native_codegen.compiler import compile_native_program from app.simulation.native_codegen.extended import compile_extended_program from app.simulation.registry import COMPONENT_MODEL_REGISTRY, build_component_catalog from app.simulation.core.medium import IdealGasMedium from app.simulation.systems.network import Connection, Endpoint, SimulationNetwork from app.system_xml import validate_system_xml_document from tests.test_amesim_pneumatic_node_xml import amesim_pn3node_project class Circuit: def __init__(self): self.net = SimulationNetwork('port_supply') def add(self, model, name): spec = COMPONENT_MODEL_REGISTRY[model] component = spec.create(name, IdealGasMedium(), {p.name: p.default for p in spec.parameters}) self.net.add_component(component) return component def chamber(self, name): return self.add('amesim_pnch023', name) def connect(self, first, first_port, second, second_port): self.net.connect(first.name, first_port, second.name, second_port) def port(model, name): return next(p for p in COMPONENT_MODEL_REGISTRY[model].ports if p.name == name) class PortComputationTests(unittest.TestCase): def test_catalog_declares_both_sides_without_changing_flow_or_state_contract(self): for spec in COMPONENT_MODEL_REGISTRY.values(): for p in spec.ports: if p.domain == 'pneumatic': self.assertIsNotNone(p.computation, spec.model_type) self.assertEqual([v.name for v in p.variables], ['p', 'm_flow', 'h_outflow', 'volume', 'volume_flow']) reference = port('amesim_p4node2', 'port_2') self.assertEqual(reference.computation.inputs, ('p', 'T')) self.assertEqual(reference.computation.outputs, ('m_flow', 'H_flow')) self.assertEqual(reference.nominal_role, 'bidirectional') self.assertEqual(port('amesim_p4node2', 'port_1').computation.reference_port, 'port_2') catalog = build_component_catalog() declared = next(c for lib in catalog['libraries'] for c in lib['components'] if c['modelType'] == 'amesim_p4node2')['ports'] self.assertEqual(declared[1]['computation'], reference.computation.as_dict()) def test_reference_accepts_storage_and_branch_accepts_flow_in_both_edge_orders(self): for node in ('amesim_pn3node2', 'amesim_p4node2'): for model, name in [('amesim_pnl0001', 'port_2'), ('amesim_pnl0003', 'port_1'), ('amesim_pnch023', 'port_1'), ('tank', 'port_a')]: a, b = port(node, 'port_2'), port(model, name) self.assertIsNone(port_supply_issue(a, b)) self.assertIsNone(port_supply_issue(b, a)) for model, name in [('amesim_pnvo001', 'port_2'), ('amesim_pnl0001', 'port_1'), ('amesim_pnl0002', 'port_2'), ('amesim_pnpl01', 'port_1')]: a, b = port(node, 'port_1'), port(model, name) self.assertIsNone(port_supply_issue(a, b)) self.assertIsNone(port_supply_issue(b, a)) def test_wrong_reference_and_wrong_branch_report_specific_missing_variables(self): for node in ('amesim_pn3node2', 'amesim_p4node2'): a, b = port(node, 'port_2'), port('amesim_pnvo001', 'port_2') for first, second in [(a, b), (b, a)]: issue = port_supply_issue(first, second, 'left.port_2', 'right.port_2') self.assertEqual(issue.code, 'CONNECTION_VARIABLE_SUPPLY_MISSING') self.assertIn('温度', issue.message) self.assertIn('压力', issue.message) self.assertIn('left.port_2', issue.message) issue = port_supply_issue(port(node, 'port_1'), port('amesim_pnl0003', 'port_1')) self.assertIn('质量流率', issue.message) self.assertIn('能量流率', issue.message) self.assertIsNotNone(port_supply_issue(a, port('amesim_pnpl01', 'port_1'))) def test_equation_connections_are_not_mistaken_for_fixed_reference_errors(self): for a, b in [(port('amesim_pnl00r', 'port_1'), port('amesim_pnor001', 'port_1')), (port('amesim_pnl0003', 'port_2'), port('amesim_pnl0001', 'port_2')), (port('tee', 'port_in'), port('orifice', 'port_b'))]: self.assertIsNone(port_supply_issue(a, b)) def test_saved_port_metadata_cannot_override_registered_supply(self): payload = amesim_pn3node_project().model_dump() for edge in payload['edges']: if edge['target'] == 'node_1': edge['targetHandle'] = 'port_2' if edge['source'] == 'node_1' and edge['sourceHandle'] == 'port_2': edge['sourceHandle'] = 'port_1' for node in payload['nodes']: for p in node['data']['ports']: p['computation'] = {'mode': 'equation', 'inputs': [], 'outputs': ['p', 'T']} project = ReactFlowProjectPayload.model_validate(payload) for entry in (build_reactflow_system_xml, compile_reactflow_network): with self.assertRaisesRegex(ValueError, 'CONNECTION_VARIABLE_SUPPLY_MISSING'): entry(project) def test_xml_validation_reports_the_edge_before_native_execution(self): root = ET.fromstring(build_reactflow_system_xml(amesim_pn3node_project())) for endpoint in root.findall('./Connections/Connection/Endpoint'): if endpoint.get('component') == 'node_1': if endpoint.get('port') in ('port_1', 'port_2'): endpoint.set('port', 'port_2' if endpoint.get('port') == 'port_1' else 'port_1') report = validate_system_xml_document(ET.tostring(root)) self.assertFalse(report.valid) errors = [i for i in report.issues if i.code == 'CONNECTION_VARIABLE_SUPPLY_MISSING'] self.assertEqual(len(errors), 2) self.assertTrue(all('Connection[' in i.path for i in errors)) def test_native_compilers_recheck_manually_inserted_connections(self): b = Circuit() b.add('amesim_p4node2', 'node') b.add('amesim_pnvo001', 'valve') b.net.connections.append(Connection('bad', 'physical', 'pneumatic', Endpoint('node', 'port_2'), Endpoint('valve', 'port_2'))) for compiler in (compile_native_program, compile_extended_program): with self.assertRaisesRegex(PortSupplyError, 'node.port_2.*温度'): compiler(b.net) def test_reference_chains_need_a_real_origin(self): b = Circuit() a = b.add('amesim_pn3node2', 'a') c = b.add('amesim_p4node2', 'b') tank = b.chamber('tank') b.connect(a, 'port_2', c, 'port_1') b.connect(c, 'port_2', tank, 'port_1') b.net.validate_port_supplies() b.net.connections.pop() with self.assertRaisesRegex(PortSupplyError, 'REFERENCE_SUPPLY_UNCONNECTED'): b.net.validate_port_supplies() b.connect(c, 'port_2', a, 'port_1') with self.assertRaisesRegex(PortSupplyError, 'REFERENCE_SUPPLY_CYCLE'): b.net.validate_port_supplies() def test_deep_reference_chain_does_not_require_python_recursion(self): reference = port('amesim_pn3node2', 'port_2') branch = port('amesim_pn3node2', 'port_1') ports, adjacency = {}, {} # Longer than Python's default recursion limit; resolved paths are reused. for i in range(1100): ports[str(i), 'port_1'] = branch ports[str(i), 'port_2'] = reference adjacency[str(i), 'port_2'] = (str(i+1), 'port_1') ports['1100', 'port_1'] = port('tank', 'port_a') self.assertEqual(reference_supply_issues(ports, adjacency), []) if __name__ == '__main__': unittest.main()