From 8b36eaee4adb8fa7ca4fb32d339fdc08f6923260 Mon Sep 17 00:00:00 2001 From: muaviz Date: Thu, 12 Nov 2026 22:38:53 +0530 Subject: [PATCH 1/2] test(omchem): add comprehensive unit tests for all OMChem unit operations --- tests/test_omchem_models.py | 225 +++++++++++++++++++++++++++++++++--- 1 file changed, 207 insertions(+), 18 deletions(-) diff --git a/tests/test_omchem_models.py b/tests/test_omchem_models.py index b2a5ebc..2ab546b 100644 --- a/tests/test_omchem_models.py +++ b/tests/test_omchem_models.py @@ -30,6 +30,9 @@ def test_mixer(self): m = Mixer() self.assertEqual(m.type, 'Mixer') self.assertTrue(m.name.startswith('Mixer')) + self.assertEqual(m.getname(), m.name) + self.assertEqual(m.modesList(), []) + self.assertEqual(m.paramgetter(), {}) self.assertEqual(m.no_of_input, 2) self.assertEqual(m.no_of_output, 1) @@ -37,7 +40,12 @@ def test_mixer(self): m.paramsetter({"NOI": 3}) self.assertEqual(m.NOI, 3) - # Connect + # Connect with None + m.connect(None, None) + self.assertEqual(m.InputStms, []) + self.assertEqual(m.OutputStms, []) + + # Connect with streams in1 = MagicMock(name="Inlet1") in1.name = "Inlet1" in2 = MagicMock(name="Inlet2") @@ -64,6 +72,9 @@ def test_splitter(self): s = Splitter() self.assertEqual(s.type, 'Splitter') self.assertTrue(s.name.startswith('Splitter')) + self.assertEqual(s.getname(), s.name) + self.assertEqual(s.modesList(), []) + self.assertEqual(s.paramgetter(), {"NOO": None}) self.assertEqual(s.no_of_input, 1) self.assertEqual(s.no_of_output, 4) @@ -90,6 +101,9 @@ def test_flash(self): f = Flash() self.assertEqual(f.type, 'flash') self.assertTrue(f.name.startswith('Flash')) + self.assertEqual(f.getname(), f.name) + self.assertEqual(f.modesList(), []) + self.assertEqual(f.paramgetter(), {"thermoPackage": None}) self.assertEqual(f.no_of_input, 1) self.assertEqual(f.no_of_output, 2) @@ -117,7 +131,9 @@ def test_pump(self): p = Pump() self.assertEqual(p.type, 'Pump') self.assertTrue(p.name.startswith('Pump')) + self.assertEqual(p.getname(), p.name) self.assertIn("pressInc", p.modesList()) + self.assertIn("enFlo", p.modesList()) params = p.paramgetter("pressInc") params["eff"] = 0.85 @@ -137,11 +153,19 @@ def test_pump(self): eqn_code = p.OM_Flowsheet_Eqn(["Water"]) self.assertIn("connect(PumpIn.Out", eqn_code) self.assertIn("PumpOut.In", eqn_code) + self.assertIn(f"{p.name}.pressInc=200000", eqn_code) + + # Test enFlo mode + p.mode = "enFlo" + p.modeVal = "1500" + eqn_code_en = p.OM_Flowsheet_Eqn(["Water"]) + self.assertIn(f"{p.EngStms.name}.enFlo=1500;", eqn_code_en) def test_valve(self): v = Valve() self.assertEqual(v.type, 'Valve') self.assertTrue(v.name.startswith('Valve')) + self.assertEqual(v.getname(), v.name) self.assertIn("pressDrop", v.modesList()) params = v.paramgetter("pressDrop") @@ -166,24 +190,30 @@ def test_heater_and_cooler(self): h = Heater() self.assertEqual(h.type, 'Heater') self.assertTrue(h.name.startswith('Heater')) + self.assertEqual(h.getname(), h.name) + self.assertIn("heatAdd", h.modesList()) + self.assertIn("enFlo", h.modesList()) c = Cooler() self.assertEqual(c.type, 'Cooler') self.assertTrue(c.name.startswith('Cooler')) + self.assertEqual(c.getname(), c.name) + self.assertIn("heatRem", c.modesList()) + self.assertIn("enFlo", c.modesList()) h_params = h.paramgetter("outT") h_params["PressureDrop"] = 0 h_params["eff"] = 1.0 - h_params["outT"] = 350 + h_params["outT"] = "350" h.paramsetter(h_params) - self.assertEqual(h.modeVal, 350) + self.assertEqual(h.modeVal, "350") c_params = c.paramgetter("outT") c_params["PressureDrop"] = 0 c_params["eff"] = 1.0 - c_params["outT"] = 280 + c_params["outT"] = "280" c.paramsetter(c_params) - self.assertEqual(c.modeVal, 280) + self.assertEqual(c.modeVal, "280") in1 = MagicMock() in1.name = "HeatIn" @@ -197,10 +227,28 @@ def test_heater_and_cooler(self): h.connect([in1], [out1]) init_code = h.OM_Flowsheet_Init(["Water"]) self.assertIn("Simulator.Unit_Operations.Heater", init_code) + eqn_code_h = h.OM_Flowsheet_Eqn(["Water"]) + self.assertIn("connect(HeatIn.Out", eqn_code_h) + self.assertIn(f"{h.name}.outT=350;", eqn_code_h) + + # Heater enFlo mode + h.mode = "enFlo" + h.modeVal = "5000" + eqn_code_h_en = h.OM_Flowsheet_Eqn(["Water"]) + self.assertIn(f"{eng.name}.enFlo=5000;", eqn_code_h_en) c.connect([in1], [out1]) init_code_c = c.OM_Flowsheet_Init(["Water"]) self.assertIn("Simulator.Unit_Operations.Cooler", init_code_c) + eqn_code_c = c.OM_Flowsheet_Eqn(["Water"]) + self.assertIn("connect(HeatIn.Out", eqn_code_c) + self.assertIn(f"{c.name}.outT=280;", eqn_code_c) + + # Cooler enFlo mode + c.mode = "enFlo" + c.modeVal = "4000" + eqn_code_c_en = c.OM_Flowsheet_Eqn(["Water"]) + self.assertIn(f"{eng.name}.enFlo=4000;", eqn_code_c_en) def test_energy_stream(self): e = EngStm() @@ -210,36 +258,85 @@ def test_energy_stream(self): init_code = e.OM_Flowsheet_Init(["Water"]) self.assertIn("Simulator.Streams.Energy_Stream", init_code) + eqn_code = e.OM_Flowsheet_Eqn(["Water"]) + self.assertEqual(eqn_code, '') + def test_compressor_and_expander(self): comp = AdiabaticCompressor() self.assertEqual(comp.type, 'AdiabaticCompressor') self.assertTrue(comp.name.startswith('AdiabaticCompressor')) - self.assertIn("pressInc", comp.modesList()) + self.assertEqual(comp.getname(), comp.name) + self.assertEqual(comp.modesList(), ["pressInc", "outP", "reqPow"]) comp_params = comp.paramgetter("pressInc") - comp_params["eff"] = 0.8 - comp_params["pressInc"] = 300000 + comp_params["eff"] = "0.8" + comp_params["pressInc"] = "300000" comp.paramsetter(comp_params) - self.assertEqual(comp.modeVal, 300000) + self.assertEqual(comp.eff, "0.8") + self.assertEqual(comp.modeVal, "300000") + in_c = MagicMock() + in_c.name = "CompFeed" + out_c = MagicMock() + out_c.name = "CompProd" + comp.connect([in_c], [out_c]) + + init_c = comp.OM_Flowsheet_Init(["Air"]) + self.assertIn("Simulator.Unit_Operations.Adiabatic_Compressor", init_c) + self.assertIn("eff=0.8", init_c) + + eqn_c = comp.OM_Flowsheet_Eqn(["Air"]) + self.assertIn(f"connect(CompFeed.Out,{comp.name}.In);", eqn_c) + self.assertIn(f"connect({comp.name}.Out,CompProd.In);", eqn_c) + self.assertIn(f"{comp.name}.pressInc=300000;", eqn_c) + + # Expander exp = AdiabaticExpander() self.assertEqual(exp.type, 'AdiabaticExpander') self.assertTrue(exp.name.startswith('AdiabaticExpander')) - self.assertIn("outP", exp.modesList()) + self.assertEqual(exp.getname(), exp.name) + self.assertEqual(exp.modesList(), ["pressDrop", "outP", "genPow"]) exp_params = exp.paramgetter("outP") - exp_params["eff"] = 0.85 - exp_params["outP"] = 100000 + exp_params["eff"] = "0.85" + exp_params["outP"] = "100000" exp.paramsetter(exp_params) - self.assertEqual(exp.modeVal, 100000) + self.assertEqual(exp.eff, "0.85") + self.assertEqual(exp.modeVal, "100000") + + in_e = MagicMock() + in_e.name = "ExpFeed" + out_e = MagicMock() + out_e.name = "ExpProd" + exp.connect([in_e], [out_e]) + + init_e = exp.OM_Flowsheet_Init(["Steam"]) + self.assertIn("Simulator.Unit_Operations.Adiabatic_Expander", init_e) + self.assertIn("eff=0.85", init_e) + + eqn_e = exp.OM_Flowsheet_Eqn(["Steam"]) + self.assertIn(f"connect(ExpFeed.Out,{exp.name}.In);", eqn_e) + self.assertIn(f"connect({exp.name}.Out,ExpProd.In);", eqn_e) + self.assertIn(f"{exp.name}.outP=100000;", eqn_e) def test_compound_separator(self): cs = CompSep() self.assertEqual(cs.type, 'CompSep') self.assertTrue(cs.name.startswith('CompSep')) + self.assertEqual(cs.getname(), cs.name) + self.assertEqual(cs.modesList(), []) + self.assertEqual(cs.paramgetter(), {"SepStrm": None, "SepFactValue": None, "SepFact": None}) self.assertEqual(cs.no_of_input, 1) self.assertEqual(cs.no_of_output, 2) + cs.paramsetter({ + "SepStrm": "1", + "SepFactValue": "{0.9, 0.1}", + "SepFact": "{\"Molar_Flow\", \"Mass_Flow\"}" + }) + self.assertEqual(cs.SepStrm, "1") + self.assertEqual(cs.SepFactValue, "{0.9, 0.1}") + in1 = MagicMock() in1.name = "SepFeed" out1 = MagicMock() @@ -248,20 +345,26 @@ def test_compound_separator(self): out2.name = "SepOut2" cs.connect([in1], [out1, out2]) + init_code = cs.OM_Flowsheet_Init(["Water", "Ethanol"]) + self.assertIn("Simulator.Unit_Operations.Compound_Separator", init_code) + self.assertIn("sepStrm = 1", init_code) + eqn_code = cs.OM_Flowsheet_Eqn(["Water", "Ethanol"]) self.assertIn("connect(SepFeed.Out", eqn_code) + self.assertIn(f"connect(SepOut1.In,{cs.name}.Out1);", eqn_code) + self.assertIn(f"connect(SepOut2.In,{cs.name}.Out2);", eqn_code) + self.assertIn(f"{cs.name}.sepFactVal= {{{{0.9, 0.1}}}};", eqn_code) def test_distillation_columns(self): dc = DistCol() self.assertEqual(dc.type, 'DistCol') self.assertTrue(dc.name.startswith('DistCol')) + self.assertEqual(dc.getname(), dc.name) + self.assertEqual(dc.modesList(), ["refluxRatio", "sideDrawMolFlo", "T"]) self.assertEqual(dc.no_of_input, 2) self.assertEqual(dc.no_of_output, 2) - sc = ShortcutColumn() - self.assertEqual(sc.type, 'ShortCol') - self.assertTrue(sc.name.startswith('ShortCol')) - + # DistCol refluxRatio mode dc_params = dc.paramgetter("refluxRatio") dc_params["numStage"] = "10" dc_params["numFeeds"] = "1" @@ -270,9 +373,95 @@ def test_distillation_columns(self): dc_params["condensor.P"] = "101325" dc_params["reboiler.P"] = "101325" dc_params["condType"] = "Total" - dc.thermoPackage = "NRTL" + dc.thermoPackage = "Raoults_Law" dc.paramsetter(dc_params) self.assertEqual(dc.numStage, "10") + self.assertEqual(dc.modeVal, "1.5") + + feed1 = MagicMock() + feed1.name = "ColFeed1" + distillate = MagicMock() + distillate.name = "Distillate" + bottoms = MagicMock() + bottoms.name = "Bottoms" + bottoms.Prop = {"totMolFlo[1]": 25.0} + + dc.connect([feed1], [distillate, bottoms]) + + init_code = dc.OM_Flowsheet_Init(["Benzene", "Toluene"]) + self.assertIn("model Condensor", init_code) + self.assertIn("model Tray", init_code) + self.assertIn("model Reboiler", init_code) + self.assertIn(f"noOfStages=10,noOfFeeds=1,feedStages={{5}},condensor.condType=\"Total\"", init_code) + self.assertIn(f"Simulator.Streams.Energy_Stream {dc.EngStm1.name};", init_code) + self.assertIn(f"Simulator.Streams.Energy_Stream {dc.EngStm2.name};", init_code) + + eqn_code = dc.OM_Flowsheet_Eqn(["Benzene", "Toluene"]) + self.assertIn(f"connect({dc.name}.condensor_duty,{dc.EngStm1.name}.In);", eqn_code) + self.assertIn(f"connect({dc.name}.En2, {dc.EngStm2.name}.In);", eqn_code) + self.assertIn(f"connect({dc.name}.Out1, Distillate.In);", eqn_code) + self.assertIn(f"connect({dc.name}.Out2, Bottoms.In);", eqn_code) + self.assertIn(f"connect(ColFeed1.Out, {dc.name}.In[1]);", eqn_code) + self.assertIn(f"{dc.name}.refluxRatio=1.5;", eqn_code) + self.assertIn(f"{dc.name}.reboiler.P=101325;", eqn_code) + self.assertIn(f"{dc.name}.condensor.P=101325;", eqn_code) + + # DistCol alternative mode (e.g. temperature T) + dc2 = DistCol() + dc2_params = dc2.paramgetter("T") + dc2_params["numStage"] = "8" + dc2_params["numFeeds"] = "1" + dc2_params["feedStages"] = "4" + dc2_params["T"] = "360.0" + dc2_params["condensor.P"] = "101325" + dc2_params["reboiler.P"] = "101325" + dc2_params["condType"] = "Partial" + dc2.thermoPackage = "Raoults_Law" + dc2.paramsetter(dc2_params) + dc2.connect([feed1], [distillate, bottoms]) + dc2.OM_Flowsheet_Init(["Benzene", "Toluene"]) + eqn_code2 = dc2.OM_Flowsheet_Eqn(["Benzene", "Toluene"]) + self.assertIn(f"{dc2.name}.condensor.T=360.0;", eqn_code2) + + # Shortcut Column + sc = ShortcutColumn() + self.assertEqual(sc.type, 'ShortCol') + self.assertTrue(sc.name.startswith('ShortCol')) + self.assertEqual(sc.getname(), sc.name) + self.assertEqual(sc.modesList(), []) + self.assertIn("HKey", sc.paramgetter()) + + sc_params = sc.paramgetter() + sc_params["HKey"] = "2" + sc_params["LKey"] = "1" + sc_params["HKeyMolFrac"] = "0.01" + sc_params["LKeyMolFrac"] = "0.99" + sc_params["condP"] = "101325" + sc_params["rebP"] = "105000" + sc_params["actR"] = "1.5" + sc_params["condType"] = "Total" + sc_params["thermoPackage"] = "Raoults_Law" + sc.paramsetter(sc_params) + + sc.connect([feed1], [distillate, bottoms]) + + sc_init = sc.OM_Flowsheet_Init(["Benzene", "Toluene"]) + self.assertIn("extends Simulator.Unit_Operations.Shortcut_Column;", sc_init) + self.assertIn("condType = \"Total\",HKey = 2,LKey = 1);", sc_init) + self.assertIn(f"Simulator.Streams.Energy_Stream {sc.EngStm1.name};", sc_init) + self.assertIn(f"Simulator.Streams.Energy_Stream {sc.EngStm2.name};", sc_init) + + sc_eqn = sc.OM_Flowsheet_Eqn(["Benzene", "Toluene"]) + self.assertIn(f"connect({sc.name}.En2,{sc.EngStm1.name}.Out);", sc_eqn) + self.assertIn(f"connect({sc.EngStm2.name}.In,{sc.name}.En1);", sc_eqn) + self.assertIn(f"connect({sc.name}.Out1,Distillate.In);", sc_eqn) + self.assertIn(f"connect({sc.name}.Out2,Bottoms.In);", sc_eqn) + self.assertIn(f"connect(ColFeed1.Out,{sc.name}.In);", sc_eqn) + self.assertIn(f"{sc.name}.rebP = 105000;", sc_eqn) + self.assertIn(f"{sc.name}.condP = 101325;", sc_eqn) + self.assertIn(f"{sc.name}.mixMolFrac[2,{sc.name}.LKey]=0.99;", sc_eqn) + self.assertIn(f"{sc.name}.mixMolFrac[3,{sc.name}.HKey]=0.01;", sc_eqn) + self.assertIn(f"{sc.name}.actR=1.5;", sc_eqn) def test_conversion_reactor(self): cr = ConvReactor( From f83bdc6ea862cac25334b03b78bdb215a4b4dd1b Mon Sep 17 00:00:00 2001 From: muaviz Date: Thu, 12 Nov 2026 22:42:42 +0530 Subject: [PATCH 2/2] test(omchem): achieve 100% test coverage for Flowsheet simulation lifecycle --- tests/test_flowsheet.py | 351 +++++++++++++++++++++++++++++++++++++++- 1 file changed, 349 insertions(+), 2 deletions(-) diff --git a/tests/test_flowsheet.py b/tests/test_flowsheet.py index 2d47f65..72051db 100644 --- a/tests/test_flowsheet.py +++ b/tests/test_flowsheet.py @@ -96,11 +96,48 @@ def test_get_omc_path_found(self): path = self.flowsheet.get_omc_path() self.assertTrue(path.endswith("omc") or path.endswith("omc.exe")) - def test_get_omc_path_missing(self): + def test_extract_omc_error_detail_line_with_empty_geterrorstring(self): + combined = "Error: Some compilation error\ngetErrorString() \"\"\nNext line error" + error = self.flowsheet._extract_omc_error("", combined) + self.assertIn("Error: Some compilation error", error) + self.assertNotIn('getErrorString() ""', error) + + def test_decode_process_output_exception(self): + # When .decode() raises an exception, _decode_process_output should fallback to str(output) + bad_obj = MagicMock() + bad_obj.decode.side_effect = RuntimeError("decode exploded") + bad_obj.__str__.return_value = "fallback_string" + decoded = self.flowsheet._decode_process_output(bad_obj) + self.assertEqual(decoded, "fallback_string") + + def test_init_creates_sim_dir_path_if_missing(self): with patch("os.path.exists", return_value=False): + with patch("os.makedirs") as mock_makedirs: + fs = Flowsheet() + mock_makedirs.assert_called_once() + + def test_get_omc_path_windows_candidates(self): + with patch("platform.system", return_value="Windows"): + with patch("glob.glob", return_value=[r"C:\Program Files\OpenModelica1.26.3-64bit\bin\omc.exe"]): + with patch("os.path.exists", return_value=True): + path = self.flowsheet.get_omc_path() + self.assertIn("OpenModelica", path) + + def test_get_omc_path_windows_fallback(self): + with patch("platform.system", return_value="Windows"): with patch("glob.glob", return_value=[]): + with patch("os.path.exists", return_value=True): + path = self.flowsheet.get_omc_path() + self.assertIn("OpenModelica1.26.3-64bit", path) + + def test_get_omc_path_missing_with_msg(self): + with patch("os.path.exists", return_value=False): + with patch("glob.glob", return_value=[]): + msg = ["Simulation started"] with self.assertRaises(FileNotFoundError): - self.flowsheet.get_omc_path() + self.flowsheet.get_omc_path(msg=msg) + self.assertTrue(len(msg) > 1) + self.assertIn("OMC not found", msg[1]) class TestFlowsheetSimulationLifecycle(unittest.TestCase): @@ -228,6 +265,316 @@ def test_ext_data_empty_or_single_row(self): self.flowsheet.result_data = [["time", "T"]] self.flowsheet.ext_data() # Must not raise + def test_ext_data_index_or_key_error(self): + """ext_data should catch IndexError or KeyError without raising.""" + stream = MagicMock() + stream.type = "MaterialStream" + stream.name = "Stream1" + stream.variables = {"T": {"value": "0"}} + self.flowsheet.add_unit_operations(stream) + + # Header has Stream1.T at index 1, but data row has length 1 (IndexError) + self.flowsheet.result_data = [ + ["time", "Stream1.T"], + ["0.0"] + ] + self.flowsheet.ext_data() + self.assertEqual(stream.variables["T"]["value"], "0") + + def test_simulate_eqn_package_order_and_stray_cleanup(self): + """Verify simulate_EQN updates package.order and removes stray .mo files.""" + # Create package.order missing Flowsheet + order_path = os.path.join(self.temp_dir, 'package.order') + with open(order_path, 'w') as f: + f.write("SomeExistingModel\n") + + # Create stray .mo file + stray_path = os.path.join(self.temp_dir, 'stray_model.mo') + with open(stray_path, 'w') as f: + f.write("model stray_model end stray_model;") + + self.assertTrue(os.path.exists(stray_path)) + + with patch.object(self.flowsheet, "send_for_simulation_Eqn"): + self.flowsheet.simulate_EQN(msg=[]) + + # Verify Flowsheet was appended to package.order + with open(order_path, 'r') as f: + order_content = f.read() + self.assertIn("Flowsheet", order_content) + self.assertIn("SomeExistingModel", order_content) + + # Verify stray .mo file was removed + self.assertFalse(os.path.exists(stray_path)) + + def test_send_for_simulation_eqn_success(self): + """Verify send_for_simulation_Eqn runs process and loads CSV results.""" + self.flowsheet.sim_method = 'Eqn' + csv_path = os.path.join(self.temp_dir, 'Simulator.Flowsheet.FlowsheetSimulation_res.csv') + # Pre-create csv_path to exercise cleanup at start of send_for_simulation_Eqn + with open(csv_path, 'w') as f: + f.write("old data\n") + + mock_proc = MagicMock() + mock_proc.communicate.return_value = (b"record SimulationResult\n timeSimulation = 1.0\nend SimulationResult;", b"") + + # Mock Popen writing the CSV + def fake_communicate(): + with open(csv_path, 'w') as f: + f.write("time,MaterialStream1.T\n0.0,300\n1.0,350\n") + return (b"record SimulationResult\n timeSimulation = 1.0\nend SimulationResult;", b"") + + mock_proc.communicate.side_effect = fake_communicate + + with patch.object(self.flowsheet, "get_omc_path", return_value="/mock/omc"): + with patch("python.OMChem.Flowsheet.Popen", return_value=mock_proc): + stream = MagicMock() + stream.type = "MaterialStream" + stream.name = "MaterialStream1" + stream.variables = {"T": {"value": "0"}} + self.flowsheet.add_unit_operations(stream) + + self.flowsheet.send_for_simulation_Eqn([]) + + self.assertEqual(len(self.flowsheet.result_data), 3) + self.assertEqual(stream.variables["T"]["value"], "350") + self.assertEqual(self.flowsheet.last_error, "") + + def test_send_for_simulation_eqn_windows_startupinfo(self): + """Verify Windows branch in send_for_simulation_Eqn configures STARTUPINFO.""" + self.flowsheet.sim_method = 'Eqn' + mock_proc = MagicMock() + mock_proc.communicate.return_value = (b"Simulation done", b"") + + mock_startupinfo_cls = MagicMock() + with patch("platform.system", return_value="Windows"): + with patch("subprocess.STARTUPINFO", mock_startupinfo_cls, create=True): + with patch("subprocess.STARTF_USESHOWWINDOW", 1, create=True): + with patch.object(self.flowsheet, "get_omc_path", return_value="C:\\omc.exe"): + with patch("python.OMChem.Flowsheet.Popen", return_value=mock_proc) as mock_popen: + self.flowsheet.send_for_simulation_Eqn([]) + mock_popen.assert_called_once() + _, kwargs = mock_popen.call_args + self.assertIn("startupinfo", kwargs) + + def test_send_for_simulation_eqn_failure(self): + """Verify send_for_simulation_Eqn sets last_error on simulation error or missing CSV.""" + self.flowsheet.sim_method = 'Eqn' + mock_proc = MagicMock() + mock_proc.communicate.return_value = (b"timeSimulation = 0.0,\nError: Model is overdetermined", b"") + + with patch.object(self.flowsheet, "get_omc_path", return_value="/mock/omc"): + with patch("python.OMChem.Flowsheet.Popen", return_value=mock_proc): + self.flowsheet.send_for_simulation_Eqn([]) + + self.assertEqual(self.flowsheet.result_data, []) + self.assertIn("Error: Model is overdetermined", self.flowsheet.last_error) + + def test_send_for_simulation_sm_success(self): + """Verify send_for_simulation_SM executes unit and reads its CSV.""" + mock_unit = MagicMock() + mock_unit.name = "Mixer1" + + csv_path = os.path.join(self.temp_dir, 'Mixer1_res.csv') + mock_proc = MagicMock() + def fake_communicate(): + with open(csv_path, 'w') as f: + f.write("time,Stream1.T\n0.0,300\n1.0,320\n") + return (b"Mixer simulation success", b"") + + mock_proc.communicate.side_effect = fake_communicate + + with patch.object(self.flowsheet, "get_omc_path", return_value="/mock/omc"): + with patch("python.OMChem.Flowsheet.Popen", return_value=mock_proc): + self.flowsheet.send_for_simulation_SM(mock_unit) + + self.assertEqual(len(self.flowsheet.result_data), 3) + + def test_send_for_simulation_sm_missing_csv(self): + """Verify send_for_simulation_SM extracts error when CSV is missing.""" + mock_unit = MagicMock() + mock_unit.name = "Mixer1" + mock_proc = MagicMock() + mock_proc.communicate.return_value = (b"Error: Failed to simulate Mixer1", b"") + + with patch.object(self.flowsheet, "get_omc_path", return_value="/mock/omc"): + with patch("python.OMChem.Flowsheet.Popen", return_value=mock_proc): + self.flowsheet.send_for_simulation_SM(mock_unit) + + self.assertIn("Error: Failed to simulate Mixer1", self.flowsheet.last_error) + + def test_simulate_sm_complete_lifecycle(self): + """Verify simulate_SM correctly sorts units, writes files, transfers stream properties, and combines CSVs.""" + self.flowsheet.add_compound_list(["Water", "Ethanol"]) + + # Streams + s1 = MagicMock() + s1.type = "MaterialStream" + s1.name = "Stream1" + s1.Prop = {"T": "300"} + s1.OM_Flowsheet_Initialize.return_value = "ms Stream1(Nc = 2);" + s1.OM_Flowsheet_Equation.return_value = "Stream1.T = 300;" + + s2 = MagicMock() + s2.type = "MaterialStream" + s2.name = "Stream2" + s2.Prop = {"T": "310"} + s2.OM_Flowsheet_Initialize.return_value = "ms Stream2(Nc = 2);" + s2.OM_Flowsheet_Equation.return_value = "Stream2.T = 310;" + + s3 = MagicMock() + s3.type = "MaterialStream" + s3.name = "Stream3" + s3.Prop = {"T": "0"} + s3.OM_Flowsheet_Initialize.return_value = "ms Stream3(Nc = 2);" + s3.OM_Flowsheet_Equation.return_value = "Stream3.T = 0;" + + s4 = MagicMock() + s4.type = "MaterialStream" + s4.name = "Stream4" + s4.Prop = {"T": "0"} + s4.OM_Flowsheet_Initialize.return_value = "ms Stream4(Nc = 2);" + s4.OM_Flowsheet_Equation.return_value = "Stream4.T = 0;" + + e1 = MagicMock() + e1.type = "EngStm" + e1.name = "EngStm1" + e1.OM_Flowsheet_Initialize.return_value = "Simulator.Streams.Energy_Stream EngStm1;" + + # Unit 1: Mixer (input dict, output list, eng None) + mixer = MagicMock() + mixer.type = "Mixer" + mixer.name = "Mixer1" + mixer.input_stms = {"in1": s1, "in2": s2} + mixer.output_stms = [s3] + mixer.EngStms = None + mixer.OM_Flowsheet_Initialize.return_value = "Simulator.UnitOperations.Mixer Mixer1(Nc = 2);" + mixer.OM_Flowsheet_Equation.return_value = "connect(Stream1.Out, Mixer1.In[1]);" + + # Unit 2: Heater (single input not list/dict, output list, eng stm) + heater = MagicMock() + heater.type = "Heater" + heater.name = "Heater1" + heater.input_stms = s3 + heater.output_stms = [s4] + heater.EngStms = e1 + heater.OM_Flowsheet_Initialize.return_value = "Simulator.Unit_Operations.Heater Heater1(Nc = 2);" + heater.OM_Flowsheet_Equation.return_value = "connect(Stream3.Out, Heater1.In);" + + # Add all to flowsheet, including streams to test unit filtering + self.flowsheet.add_unit_operations(mixer) + self.flowsheet.add_unit_operations(heater) + self.flowsheet.add_unit_operations(s1) + self.flowsheet.add_unit_operations(e1) + + # ip / op structured to exercise all 4 topological sorting branches: + # Branch 1: neither in self.unit -> appends k then i (Mixer1, Heater1) + # Branch 2: i in self.unit -> inserts k before i (Mixer0 inserted before Mixer1) + # Branch 3: k in self.unit -> appends i (Cooler1 appended after Heater1) + # Branch 4: both i and k in self.unit -> pass + ip = { + "Heater1": ["Stream3"], + "Mixer1": ["Stream0"], + "Cooler1": ["Stream4"], + "PassUnit": ["Stream3"], + } + op = { + "Mixer1": ["Stream3"], + "Mixer0": ["Stream0"], + "Heater1": ["Stream4"], + "Heater0": ["Stream3"], + } + + # Extra stream without Prop to exercise line 514 + s_no_prop = MagicMock() + s_no_prop.type = "MaterialStream" + s_no_prop.name = "StreamNoProp" + s_no_prop.Prop = None + s_no_prop.OM_Flowsheet_Initialize.return_value = "ms StreamNoProp(Nc = 2);" + s_no_prop.OM_Flowsheet_Equation.return_value = "StreamNoProp.T = 300;" + + # Extra unit without type / MaterialStream to exercise line 421 + unit_stm = MagicMock() + unit_stm.type = "MaterialStream" + unit_stm.name = "Mixer0" + + # Unit 0 (Mixer0) + mixer0 = MagicMock() + mixer0.type = "MaterialStream" # Intentionally MaterialStream to trigger line 421 continue + mixer0.name = "Mixer0" + + # PassUnit + pass_unit = MagicMock() + pass_unit.type = "EngStm" # Intentionally EngStm to trigger line 421 continue + pass_unit.name = "PassUnit" + + cooler = MagicMock() + cooler.type = "Cooler" + cooler.name = "Cooler1" + cooler.input_stms = [s4, s_no_prop] + cooler.output_stms = [] + cooler.EngStms = None + cooler.OM_Flowsheet_Initialize.return_value = "Simulator.Unit_Operations.Cooler Cooler1(Nc = 2);" + cooler.OM_Flowsheet_Equation.return_value = "connect(Stream4.Out, Cooler1.In);" + + self.flowsheet.add_unit_operations(mixer0) + self.flowsheet.add_unit_operations(pass_unit) + self.flowsheet.add_unit_operations(cooler) + + # Mock Popen to write CSV files for each simulated unit + def fake_communicate(): + # Write results for whatever unit is being simulated + for name in ["mixer1", "heater1", "cooler1"]: + p = os.path.join(self.temp_dir, f"Simulator.{name}_res.csv") + if not os.path.exists(p): + with open(p, 'w') as f: + f.write(f"time,Stream3.T,Stream4.T\n0.0,305,350\n1.0,305,350\n") + break + return (b"Sequential simulation success", b"") + + mock_proc = MagicMock() + mock_proc.communicate.side_effect = fake_communicate + + with patch.object(self.flowsheet, "get_omc_path", return_value="/mock/omc"): + with patch("python.OMChem.Flowsheet.Popen", return_value=mock_proc): + self.flowsheet.simulate_SM(ip, op) + + # Check that FlowsheetSEQ.csv was produced and loaded + seq_csv = os.path.join(self.temp_dir, "FlowsheetSEQ.csv") + self.assertTrue(os.path.exists(seq_csv)) + self.assertTrue(len(self.flowsheet.result_data) >= 2) + # Check property transfer + self.assertEqual(s3.Prop["T"], "305") + self.assertEqual(s4.Prop["T"], "350") + + def test_simulate_sm_failure_when_no_csv(self): + """Verify simulate_SM terminates early and records last_error when a unit simulation produces no CSV.""" + self.flowsheet.add_compound_list(["Water"]) + + mixer = MagicMock() + mixer.type = "Mixer" + mixer.name = "Mixer1" + mixer.input_stms = [] + mixer.output_stms = [] + mixer.EngStms = None + mixer.OM_Flowsheet_Initialize.return_value = "Mixer1_init;" + mixer.OM_Flowsheet_Equation.return_value = "Mixer1_eqn;" + + self.flowsheet.add_unit_operations(mixer) + + # Ensure unit order resolves Mixer1 into self.unit + ip = {"Mixer1": ["s1"]} + op = {"Mixer1": ["s1"]} + + mock_proc = MagicMock() + mock_proc.communicate.return_value = (b"Error: compilation failed for Mixer1", b"") + + with patch.object(self.flowsheet, "get_omc_path", return_value="/mock/omc"): + with patch("python.OMChem.Flowsheet.Popen", return_value=mock_proc): + self.flowsheet.simulate_SM(ip, op) + + self.assertIn("Error: compilation failed for Mixer1", self.flowsheet.last_error) + if __name__ == "__main__": unittest.main()