221 lines
7.6 KiB
Python
221 lines
7.6 KiB
Python
import pytest
|
|
|
|
from app.algorithms.valve_isolation import topology_search as valve
|
|
|
|
|
|
def test_non_isolatable_omits_affected_node_ids_but_keeps_count():
|
|
links = ["P-1:pipe:A:B", "P-2:pipe:B:C", "V-optional:valve:A:C"]
|
|
|
|
result = valve.valve_isolation_analysis(links, "P-1")
|
|
|
|
assert result["isolatable"] is False
|
|
assert result["affected_node_count"] == 3
|
|
assert result["affected_nodes"] == []
|
|
assert result["optional_valves"] == ["V-optional"]
|
|
|
|
|
|
def test_isolatable_keeps_affected_node_ids_and_count():
|
|
links = ["P-1:pipe:A:B", "V-close:valve:B:C"]
|
|
|
|
result = valve.valve_isolation_analysis(links, "P-1")
|
|
|
|
assert result["isolatable"] is True
|
|
assert result["affected_node_count"] == 2
|
|
assert result["affected_nodes"] == ["A", "B"]
|
|
assert result["must_close_valves"] == ["V-close"]
|
|
|
|
|
|
@pytest.fixture
|
|
def colliding_topology():
|
|
# The reported adjacent pipes and their same-numbered nodes are in
|
|
# different valve-bounded regions. Keep this fixture independent of a DB.
|
|
return [
|
|
"19025:pipe:17038:17037",
|
|
"19026:pipe:17038:17039",
|
|
"AUTO_TCV_4:valve:17037:D",
|
|
"AUTO_TCV_5:valve:17039:E",
|
|
"AUTO_TCV_546:valve:17038:R",
|
|
"32252:pipe:19025:X",
|
|
"15383:pipe:19026:Y",
|
|
"V-wrong-25:valve:X:R",
|
|
"V-wrong-26:valve:Y:R",
|
|
]
|
|
|
|
|
|
@pytest.mark.parametrize("target", ["19025", "19026", ["19025", "19026"]])
|
|
def test_adjacent_pipes_take_priority_over_same_numbered_nodes(colliding_topology, target):
|
|
result = valve.valve_isolation_analysis(colliding_topology, target)
|
|
|
|
assert result["affected_nodes"] == ["17037", "17038", "17039"]
|
|
assert result["must_close_valves"] == ["AUTO_TCV_4", "AUTO_TCV_5", "AUTO_TCV_546"]
|
|
assert result["optional_valves"] == []
|
|
|
|
|
|
def test_explicit_node_type_resolves_the_other_side_of_an_id_collision(colliding_topology):
|
|
result = valve.valve_isolation_analysis(
|
|
colliding_topology, "19025", accident_element_type="node",
|
|
)
|
|
|
|
assert result["affected_nodes"] == ["19025", "X"]
|
|
assert result["must_close_valves"] == ["V-wrong-25"]
|
|
|
|
|
|
def test_unambiguous_legacy_node_input_still_works():
|
|
result = valve.valve_isolation_analysis(["P:pipe:A:B", "V:valve:B:R"], "A")
|
|
assert result["affected_nodes"] == ["A", "B"]
|
|
|
|
|
|
def test_explicit_link_type_does_not_fall_back_to_a_node():
|
|
with pytest.raises(ValueError, match="not found"):
|
|
valve.valve_isolation_analysis(
|
|
["P:pipe:A:B"], "A", accident_element_type="link",
|
|
)
|
|
|
|
|
|
def test_a_boundary_valve_does_not_prove_source_isolation():
|
|
result = valve.valve_isolation_analysis(
|
|
["P:pipe:R:A", "V:valve:A:B"], "P", water_source_nodes=["R"],
|
|
)
|
|
|
|
assert result["isolatable"] is False
|
|
assert result["source_check_performed"] is True
|
|
assert result["isolation_failure_reason"] == "connected_to_water_source"
|
|
assert result["affected_nodes"] == []
|
|
assert result["affected_node_count"] == 2
|
|
assert result["supply_affected_nodes"] is None
|
|
|
|
|
|
def test_supply_impact_includes_downstream_regions_but_not_preexisting_outages():
|
|
links = [
|
|
"P:pipe:A:B", "V-source:valve:A:R", "V-downstream:valve:B:C",
|
|
"P-downstream:pipe:C:D", "P-already-off:pipe:X:Y",
|
|
]
|
|
result = valve.valve_isolation_analysis(links, "P", water_source_nodes=["R"])
|
|
|
|
assert result["isolatable"] is True
|
|
assert result["source_check_performed"] is True
|
|
assert result["affected_nodes"] == ["A", "B"]
|
|
assert result["affected_node_count"] == 2
|
|
assert result["supply_affected_nodes"] == ["A", "B", "C", "D"]
|
|
assert result["supply_affected_node_count"] == 4
|
|
|
|
|
|
def test_closed_links_are_not_traversed_or_recommended_for_closure():
|
|
result = valve.valve_isolation_analysis(
|
|
["P:pipe:A:B", "P-closed:pipe:B:R", "V:valve:A:R", "V-closed:valve:B:C"],
|
|
"P", water_source_nodes=["R"], closed_links=["P-closed", "V-closed"],
|
|
)
|
|
|
|
assert result["isolatable"] is True
|
|
assert result["affected_nodes"] == ["A", "B"]
|
|
assert result["must_close_valves"] == ["V"]
|
|
assert result["supply_affected_nodes"] == ["A", "B"]
|
|
|
|
|
|
def test_disabled_valve_cannot_cut_the_source_path():
|
|
result = valve.valve_isolation_analysis(
|
|
["P:pipe:A:B", "V-disabled:valve:A:R", "V-other:valve:B:C"],
|
|
"P", disabled_valves=["V-disabled"], water_source_nodes=["R"],
|
|
)
|
|
|
|
assert result["isolatable"] is False
|
|
assert result["isolation_failure_reason"] == "connected_to_water_source"
|
|
|
|
|
|
def test_joint_accidents_fail_if_any_region_remains_connected_to_a_source():
|
|
result = valve.valve_isolation_analysis(
|
|
["P1:pipe:A:B", "V1:valve:B:R", "P2:pipe:R:C", "V2:valve:C:D"],
|
|
["P1", "P2"], water_source_nodes=["R"],
|
|
)
|
|
assert result["isolatable"] is False
|
|
|
|
|
|
def test_missing_sources_do_not_claim_verified_isolation():
|
|
result = valve.valve_isolation_analysis(
|
|
["P:pipe:A:B", "V:valve:B:C"], "P", water_source_nodes=[],
|
|
)
|
|
assert result["isolatable"] is False
|
|
assert result["source_check_performed"] is False
|
|
assert result["isolation_failure_reason"] == "water_sources_unavailable"
|
|
assert result["supply_affected_node_count"] is None
|
|
|
|
|
|
@pytest.mark.parametrize("sources", [["R1"], ["R1", "R2"]])
|
|
def test_all_water_sources_are_considered(sources):
|
|
result = valve.valve_isolation_analysis(
|
|
["P:pipe:A:B", "V:valve:A:R1", "P2:pipe:B:R2"],
|
|
"P", water_source_nodes=sources,
|
|
)
|
|
assert result["isolatable"] is (len(sources) == 1)
|
|
|
|
|
|
def test_empty_targets_are_rejected():
|
|
with pytest.raises(ValueError, match="At least one"):
|
|
valve.valve_isolation_analysis(["P:pipe:A:B"], [])
|
|
|
|
|
|
def test_unknown_disabled_valve_is_rejected():
|
|
with pytest.raises(ValueError, match="Disabled valve"):
|
|
valve.valve_isolation_analysis(["P:pipe:A:B"], "P", disabled_valves=["typo"])
|
|
|
|
|
|
def test_service_passes_authoritative_sources_statuses_and_element_type(monkeypatch):
|
|
from app.services import valve_isolation as service
|
|
|
|
projects = []
|
|
|
|
def snapshot(network):
|
|
projects.append(network)
|
|
return (
|
|
["P:pipe:A:B", "P-closed:pipe:B:R", "V:valve:A:R"],
|
|
["R"], ["P-closed"],
|
|
)
|
|
|
|
monkeypatch.setattr(service, "get_valve_isolation_topology", snapshot)
|
|
result = service.analyze_valve_isolation(
|
|
"trusted-project", ["A"], accident_element_type="node",
|
|
)
|
|
|
|
assert projects == ["trusted-project"]
|
|
assert result["isolatable"] is True
|
|
assert result["source_check_performed"] is True
|
|
assert result["must_close_valves"] == ["V"]
|
|
assert result["supply_affected_nodes"] == ["A", "B"]
|
|
|
|
|
|
def test_parallel_links_preserve_an_open_bypass():
|
|
result = valve.valve_isolation_analysis(
|
|
["P:pipe:A:B", "V:valve:B:R", "P-bypass:pipe:B:R"],
|
|
"P", water_source_nodes=["R"],
|
|
)
|
|
assert result["isolatable"] is False
|
|
assert result["isolation_failure_reason"] == "connected_to_water_source"
|
|
|
|
|
|
def test_closed_disabled_valve_does_not_reopen_a_known_closed_link():
|
|
result = valve.valve_isolation_analysis(
|
|
["P:pipe:A:B", "V:valve:B:R", "V-disabled:valve:A:R"],
|
|
"P", disabled_valves=["V-disabled"], closed_links=["V-disabled"],
|
|
water_source_nodes=["R"],
|
|
)
|
|
assert result["isolatable"] is True
|
|
assert result["must_close_valves"] == ["V"]
|
|
|
|
|
|
def test_disabled_valve_expands_affected_area_before_counting():
|
|
links = [
|
|
"P-1:pipe:A:B",
|
|
"V-disabled:valve:B:C",
|
|
"V-close:valve:C:D",
|
|
]
|
|
result = valve.valve_isolation_analysis(
|
|
links,
|
|
"P-1",
|
|
disabled_valves=["V-disabled"],
|
|
)
|
|
|
|
assert result["isolatable"] is True
|
|
assert result["affected_node_count"] == 3
|
|
assert result["affected_nodes"] == ["A", "B", "C"]
|
|
assert result["must_close_valves"] == ["V-close"]
|