fix(simulation): export ACTIVE valves correctly for EPANET 2
Generic Container CI/CD / test-build-publish (push) Successful in 2m12s
Server CI/CD v2 / build-test-publish-and-deploy (push) Successful in 2m12s

The legacy online simulation path still exported internal ACTIVE states as STATUS keywords, which EPANET 2 rejects. Preserve numeric settings and EPANET 3 output while passing the format through both exporters. Surface the original engine error inside the model transaction so failed runs roll back before result storage.

Regression guard: six failures reproduced on the deployed baseline, including native input validation. All 34 focused export, service and endpoint checks pass after the fix; the real project also returns native node and link results.
This commit is contained in:
2026-11-03 13:48:50 +08:00
parent 38a51ed65c
commit 9ab7b10e3c
5 changed files with 141 additions and 7 deletions
+70
View File
@@ -0,0 +1,70 @@
import ctypes
import platform
from pathlib import Path
import pytest
from app.native.wndb.model import status
def test_epanet2_preserves_valve_setting_without_emitting_active(monkeypatch):
monkeypatch.setattr(status, "read_all", lambda *_: [
{"link": "V1", "status": "ACTIVE", "setting": None},
{"link": "V2", "status": "ACTIVE", "setting": 2.5},
{"link": "V3", "status": "CLOSED", "setting": None},
{"link": "P1", "status": "OPEN", "setting": 0},
])
assert status.inp_out_status("model") == ["V2 2.5", "V3 CLOSED", "P1 OPEN", "P1 0"]
def test_epanet3_retains_explicit_active_status(monkeypatch):
monkeypatch.setattr(status, "read_all", lambda *_: [
{"link": "V1", "status": "ACTIVE", "setting": 100},
])
assert status.inp_out_status("model", version="3") == ["V1 ACTIVE", "V1 100"]
@pytest.mark.parametrize("version", ["2", "3"])
def test_file_and_text_exporters_forward_the_selected_format(monkeypatch, tmp_path, version):
from app.native.wndb.inp import exporter
monkeypatch.setattr(exporter, "have_project", lambda _: True)
monkeypatch.setattr(exporter, "section_name", [exporter.STATUS])
monkeypatch.setattr(exporter, "section_names_for_epanetv2", [exporter.STATUS])
monkeypatch.setattr(status, "read_all", lambda *_: [
{"link": "V1", "status": "ACTIVE", "setting": 100},
])
path = tmp_path / "model.inp"
exporter.dump_inp("model", str(path), version)
text = exporter.export_inp("model", version).operations[0]["inp"]
assert "V1 100" in path.read_text() and "V1 100" in text
assert ("V1 ACTIVE" in path.read_text()) == (version == "3")
assert ("V1 ACTIVE" in text) == (version == "3")
def test_exported_active_valve_passes_bundled_epanet2_input_validation(monkeypatch, tmp_path):
library = Path(__file__).resolve().parents[2] / "app/infra/epanet/linux/libepanet2.so"
if platform.system() != "Linux" or not library.exists():
pytest.skip("Bundled Linux EPANET library is unavailable")
monkeypatch.setattr(status, "read_all", lambda *_: [
{"link": "V1", "status": "ACTIVE", "setting": None},
])
inp = tmp_path / "valve.inp"
inp.write_text(
"[JUNCTIONS]\nJ1 0 1\nJ2 0 1\n[RESERVOIRS]\nR1 100\n"
"[PIPES]\nP1 R1 J1 100 300 100 0 OPEN\n"
"[VALVES]\nV1 J1 J2 600 TCV 100 0\n[STATUS]\n"
+ "\n".join(status.inp_out_status("model"))
+ "\n[OPTIONS]\nUNITS LPS\n[END]\n"
)
lib = ctypes.CDLL(str(library))
lib.EN_createproject.argtypes = [ctypes.POINTER(ctypes.c_void_p)]
lib.EN_open.argtypes = [ctypes.c_void_p, ctypes.c_char_p, ctypes.c_char_p, ctypes.c_char_p]
lib.EN_deleteproject.argtypes = [ctypes.c_void_p]
handle = ctypes.c_void_p()
assert lib.EN_createproject(ctypes.byref(handle)) == 0
try:
code = lib.EN_open(handle, str(inp).encode(), str(tmp_path / "valve.rpt").encode(), b"")
assert code == 0
finally:
lib.EN_deleteproject(handle)