42 lines
1.8 KiB
Python
42 lines
1.8 KiB
Python
import ctypes
|
|
import os
|
|
from pathlib import Path
|
|
import platform
|
|
import subprocess
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from app.infra.epanet.epanet import Output
|
|
|
|
|
|
@pytest.mark.parametrize("pressure_code,expected", [(0, "PSI"), (1, "KPA"), (2, "MTR")])
|
|
def test_output_pressure_units_follow_native_enum(pressure_code, expected):
|
|
def units(handle, kind, value):
|
|
value._obj.value = {1: 5, 2: pressure_code, 3: 0}[kind.value]
|
|
return 0
|
|
output = Output.__new__(Output)
|
|
output._lib = SimpleNamespace(ENR_getUnits=units, ENR_close=lambda *args: 0)
|
|
output._handle = ctypes.c_void_p()
|
|
output._check = lambda code: None
|
|
assert output.units() == {"flow": "LPS", "pressure": expected, "quality": "NONE"}
|
|
|
|
|
|
def test_bundled_epanet_metric_pressure_matches_head_minus_elevation(tmp_path):
|
|
binary_dir = Path(__file__).resolve().parents[2] / "app/infra/epanet/linux"
|
|
exe = binary_dir / "runepanet"
|
|
if platform.system() != "Linux" or not exe.exists():
|
|
pytest.skip("Bundled Linux EPANET runner unavailable")
|
|
inp, rpt, opt = (tmp_path / name for name in ("model.inp", "model.rpt", "model.opt"))
|
|
inp.write_text("[JUNCTIONS]\nJ1 10 1\n[RESERVOIRS]\nR1 100\n"
|
|
"[PIPES]\nP1 R1 J1 100 200 100 0 OPEN\n"
|
|
"[OPTIONS]\nUNITS LPS\n[END]\n")
|
|
env = {**os.environ, "LD_LIBRARY_PATH": str(binary_dir)}
|
|
result = subprocess.run([str(exe), str(inp), str(rpt), str(opt)], env=env,
|
|
capture_output=True, text=True, timeout=10)
|
|
assert result.returncode == 0, result.stdout + result.stderr
|
|
output = Output(str(opt))
|
|
assert output.units()["pressure"] == "MTR"
|
|
row = next(r for r in output.node_results() if r["node"] == "J1")["result"][0]
|
|
assert row["pressure"] == pytest.approx(row["head"] - 10, abs=1e-4)
|