diff --git a/app/algorithms/valve_isolation/topology_search.py b/app/algorithms/valve_isolation/topology_search.py index 9baee6f..cc35bd9 100644 --- a/app/algorithms/valve_isolation/topology_search.py +++ b/app/algorithms/valve_isolation/topology_search.py @@ -1,10 +1,11 @@ """Topology-only valve isolation search.""" from collections import defaultdict, deque -from typing import Any, Iterable +from typing import Any, Iterable, Literal VALVE_LINK_TYPE = "valve" +AccidentElementType = Literal["auto", "link", "node"] def _parse_link_entry(link_entry: str) -> tuple[str, str, str, str]: @@ -14,12 +15,37 @@ def _parse_link_entry(link_entry: str) -> tuple[str, str, str, str]: return parts[0], parts[1], parts[2], parts[3] +def _reachable_nodes( + adjacency: dict[str, list[tuple[str, str]]], + start_nodes: Iterable[str], + blocked_links: set[str], +) -> set[str]: + visited = set(start_nodes) + queue = deque(visited) + while queue: + for neighbor, link_id in adjacency.get(queue.popleft(), ()): + if link_id not in blocked_links and neighbor not in visited: + visited.add(neighbor) + queue.append(neighbor) + return visited + + def valve_isolation_analysis( link_entries: Iterable[str], accident_elements: str | list[str], disabled_valves: list[str] | None = None, + *, + accident_element_type: AccidentElementType = "auto", + water_source_nodes: Iterable[str] | None = None, + closed_links: Iterable[str] = (), ) -> dict[str, Any]: - """Determine boundary valves and affected nodes from a topology snapshot.""" + """Find nearest valve boundaries, then check source connectivity. + + Legacy untyped IDs prefer links; explicit node mode handles cross-type ID + collisions. Supply impact is a before/after connectivity difference, not + a hydraulic pressure or demand calculation. Without source metadata only + the provisional boundary search is available. + """ disabled_valves_set = set(disabled_valves or []) target_elements = ( @@ -27,8 +53,13 @@ def valve_isolation_analysis( if isinstance(accident_elements, str) else accident_elements ) + if not target_elements: + raise ValueError("At least one accident element is required") + if accident_element_type not in {"auto", "link", "node"}: + raise ValueError(f"Invalid accident element type: {accident_element_type}") + closed_links_set = set(closed_links) - pipe_adj: dict[str, set[str]] = defaultdict(set) + adjacency: dict[str, list[tuple[str, str]]] = defaultdict(list) all_valves: dict[str, tuple[str, str]] = {} link_lookup: dict[str, tuple[str, str, str]] = {} node_set: set[str] = set() @@ -39,46 +70,31 @@ def valve_isolation_analysis( node_set.update((node1, node2)) if link_type_name == VALVE_LINK_TYPE: all_valves[link_id] = (node1, node2) - else: - pipe_adj[node1].add(node2) - pipe_adj[node2].add(node1) + if link_id not in closed_links_set: + adjacency[node1].append((node2, link_id)) + adjacency[node2].append((node1, link_id)) + + unknown_disabled = disabled_valves_set - all_valves.keys() + if unknown_disabled: + raise ValueError(f"Disabled valve was not found in topology: {sorted(unknown_disabled)}") start_nodes: set[str] = set() for element in target_elements: - if element in node_set: - start_nodes.add(element) - elif element in link_lookup: + if accident_element_type != "node" and element in link_lookup: node1, node2, _ = link_lookup[element] start_nodes.update((node1, node2)) + elif accident_element_type != "link" and element in node_set: + start_nodes.add(element) else: raise ValueError(f"Accident element {element} was not found in topology") - extra_adj: dict[str, list[str]] = defaultdict(list) - boundary_valves: dict[str, tuple[str, str]] = {} - for valve_id, (node1, node2) in all_valves.items(): - if valve_id in disabled_valves_set: - extra_adj[node1].append(node2) - extra_adj[node2].append(node1) - else: - boundary_valves[valve_id] = (node1, node2) - - affected_nodes: set[str] = set() - queue = deque(start_nodes) - while queue: - node = queue.popleft() - if node in affected_nodes: - continue - affected_nodes.add(node) - queue.extend(pipe_adj.get(node, set()) - affected_nodes) - queue.extend( - neighbor - for neighbor in extra_adj.get(node, ()) - if neighbor not in affected_nodes - ) + closable_valves = set(all_valves) - disabled_valves_set - closed_links_set + affected_nodes = _reachable_nodes(adjacency, start_nodes, closable_valves) must_close_valves: list[str] = [] optional_valves: list[str] = [] - for valve_id, (node1, node2) in boundary_valves.items(): + for valve_id in closable_valves: + node1, node2 = all_valves[valve_id] node1_affected = node1 in affected_nodes node2_affected = node2 in affected_nodes if node1_affected and node2_affected: @@ -89,14 +105,37 @@ def valve_isolation_analysis( must_close_valves.sort() optional_valves.sort() isolatable = bool(must_close_valves) + failure_reason = None if isolatable else "no_boundary_valves" + sources = set(water_source_nodes or ()) + source_check_performed = bool(sources) + supply_affected_nodes = None + if water_source_nodes is not None: + if not sources: + isolatable = False + failure_reason = "water_sources_unavailable" + else: + after_closure = _reachable_nodes(adjacency, sources, set(must_close_valves)) + if affected_nodes & after_closure: + isolatable = False + failure_reason = "connected_to_water_source" + if isolatable: + before_closure = _reachable_nodes(adjacency, sources, set()) + supply_affected_nodes = sorted(before_closure - after_closure) result: dict[str, Any] = { "accident_elements": target_elements, + "accident_element_type": accident_element_type, "disabled_valves": disabled_valves, "affected_nodes": sorted(affected_nodes) if isolatable else [], "affected_node_count": len(affected_nodes), "must_close_valves": must_close_valves, "optional_valves": optional_valves, "isolatable": isolatable, + "source_check_performed": source_check_performed, + "isolation_failure_reason": failure_reason, + "supply_affected_nodes": supply_affected_nodes, + "supply_affected_node_count": ( + len(supply_affected_nodes) if supply_affected_nodes is not None else None + ), } if len(target_elements) == 1: result["accident_element"] = target_elements[0] diff --git a/app/api/v1/endpoints/simulation.py b/app/api/v1/endpoints/simulation.py index 59ce57e..35beb5e 100644 --- a/app/api/v1/endpoints/simulation.py +++ b/app/api/v1/endpoints/simulation.py @@ -180,6 +180,9 @@ def valve_isolation_endpoint( network: str = Query(..., description="管网名称（或数据库名称）"), accident_element: List[str] = Query(..., description="发生事故的管段/节点ID列表"), disabled_valves: List[str] = Query(None, description="已故障的阀门ID列表（可选）"), + accident_element_type: Literal["auto", "link", "node"] = Query( + "auto", description="事故要素类型；auto优先管段再节点；分析同名节点须指定node，纯管段分析可指定link", + ), ): """ 阀门隔离分析 @@ -187,13 +190,19 @@ def valve_isolation_endpoint( - **network**: 管网名称（或数据库名称） - **accident_element**: 发生事故的管段/节点ID列表 - **disabled_valves**: 已故障的阀门ID列表（可选） + - **accident_element_type**: auto优先管段；link仅匹配管段；node仅匹配节点 返回隔离方案，包括： - must_close_valves: 必须关闭的阀门列表 - optional_valves: 可选关闭的阀门列表 - - affected_nodes: 受影响的节点列表；不可隔离时为空列表 - - affected_node_count: 受影响的节点总数 - - isolatable: 是否可以有效隔离 + - affected_nodes / affected_node_count: 最近阀门围成的直接隔离区；不可隔离时列表为空，计数保留 + - supply_affected_nodes / supply_affected_node_count: 按建议关阀后新增失去水源连接的节点；未验证或不可隔离时为null + - source_check_performed: 是否已执行水源连通检查 + - isolation_failure_reason: 不可隔离原因；可隔离时为null + - isolatable: 存在关阀边界，且直接隔离区不再连通任何水源 + + 使用模型初始开关状态进行无向拓扑分析，水池按潜在水源处理。 + 不包含时变控制、水池蓄水量或压力流量计算，不是现场停水范围的水力证明。 """ # result = { # "accident_element": "P461309", @@ -212,7 +221,13 @@ def valve_isolation_endpoint( # "must_close_valves": ["210521658", "V12974", "V12986", "V12993"], # "optional_valves": [], # } - result = analyze_valve_isolation(network, accident_element, disabled_valves) + try: + result = analyze_valve_isolation( + network, accident_element, disabled_valves, + accident_element_type=accident_element_type, + ) + except ValueError as exc: + raise HTTPException(status_code=422, detail=str(exc)) from exc return result diff --git a/app/native/wndb/gis/network_views.py b/app/native/wndb/gis/network_views.py index 25b6e99..e00d348 100644 --- a/app/native/wndb/gis/network_views.py +++ b/app/native/wndb/gis/network_views.py @@ -1,4 +1,4 @@ -"""Read-only projections backed by the GIS materialized-view query layer.""" +"""Read-only GIS projections and authoritative network topology snapshots.""" from typing import Any @@ -69,6 +69,46 @@ def get_network_link_nodes(name: str) -> list[str]: ] +def get_valve_isolation_topology(name: str) -> tuple[list[str], list[str], list[str]]: + """Read links, closed statuses and sources from one consistent SQL snapshot. + + Isolation must include model links without published geometries, and use + initial statuses rather than treating every link as open. UNION ALL keeps + source nodes in the same statement snapshot without repeating them for + every link or relying on materialized view refresh timing. + """ + rows = read_all( + name, + """ + SELECT l.id, l.link_type::text AS element_type, + l.start_node_id, l.end_node_id, + COALESCE(s.status::text, p.status::text) AS status + FROM network.links AS l + LEFT JOIN network.pipes AS p ON p.link_id = l.id + LEFT JOIN network.link_initial_settings AS s ON s.link_id = l.id + UNION ALL + SELECT n.id, n.node_type::text AS element_type, + NULL, NULL, NULL + FROM network.nodes AS n + WHERE n.node_type::text IN ('reservoir', 'tank') + """, + ) + link_entries: list[str] = [] + source_nodes: list[str] = [] + closed_links: list[str] = [] + for row in rows: + element_id = str(row["id"]) + if row["element_type"] in {"reservoir", "tank"}: + source_nodes.append(element_id) + else: + link_entries.append( + f"{element_id}:{row['element_type']}:{row['start_node_id']}:{row['end_node_id']}" + ) + if row["status"] == "CLOSED": + closed_links.append(element_id) + return link_entries, source_nodes, closed_links + + def get_major_pipe_nodes(name: str, diameter: int) -> list[str]: """Return large pipes in the established API wire format.""" rows = read_all( diff --git a/app/services/valve_isolation.py b/app/services/valve_isolation.py index 6f88db8..ba2c733 100644 --- a/app/services/valve_isolation.py +++ b/app/services/valve_isolation.py @@ -1,13 +1,22 @@ from typing import Any -from app.algorithms.valve_isolation.topology_search import valve_isolation_analysis -from app.native.wndb.gis.network_views import get_network_link_nodes +from app.algorithms.valve_isolation.topology_search import ( + AccidentElementType, + valve_isolation_analysis, +) +from app.native.wndb.gis.network_views import get_valve_isolation_topology def analyze_valve_isolation( network: str, accident_element: str | list[str], - disabled_valves: list[str] = None, + disabled_valves: list[str] | None = None, + accident_element_type: AccidentElementType = "auto", ) -> dict[str, Any]: - link_entries = get_network_link_nodes(network) - return valve_isolation_analysis(link_entries, accident_element, disabled_valves) + link_entries, source_nodes, closed_links = get_valve_isolation_topology(network) + return valve_isolation_analysis( + link_entries, accident_element, disabled_valves, + accident_element_type=accident_element_type, + water_source_nodes=source_nodes, + closed_links=closed_links, + ) diff --git a/contracts/manifest.json b/contracts/manifest.json index 49cf8d2..66a3720 100644 --- a/contracts/manifest.json +++ b/contracts/manifest.json @@ -3,7 +3,7 @@ "contracts": { "server": { "file": "server-v1.openapi.json", - "sha256": "b565d841061c9091f48ff3118bcc0cbb1b918b0cb8c2316d177570e8b2d8ba29" + "sha256": "5b8ec0b469f1194e72214003a9cf427489f48de19ae31ee6492821cf64916b6d" } } } diff --git a/contracts/server-v1.openapi.json b/contracts/server-v1.openapi.json index 5b68dce..b986ce0 100644 --- a/contracts/server-v1.openapi.json +++ b/contracts/server-v1.openapi.json @@ -35969,6 +35969,23 @@ "type": "array" } }, + { + "description": "事故要素类型；auto优先管段再节点；分析同名节点须指定node，纯管段分析可指定link", + "in": "query", + "name": "accident_element_type", + "required": false, + "schema": { + "default": "auto", + "description": "事故要素类型；auto优先管段再节点；分析同名节点须指定node，纯管段分析可指定link", + "enum": [ + "auto", + "link", + "node" + ], + "title": "Accident Element Type", + "type": "string" + } + }, { "in": "header", "name": "X-Project-Id", diff --git a/tests/api/test_openapi_contract.py b/tests/api/test_openapi_contract.py index eb8519e..03f0236 100644 --- a/tests/api/test_openapi_contract.py +++ b/tests/api/test_openapi_contract.py @@ -254,14 +254,16 @@ def test_committed_write_refresh_failure_has_explicit_http_contract() -> None: assert response.json()["code"] == "materialized_view_refresh_failed_after_commit" -def test_valve_isolation_runtime_accepts_frontend_query(monkeypatch) -> None: +@pytest.mark.parametrize("element_type", ["auto", "link", "node"]) +def test_valve_isolation_runtime_accepts_frontend_query(monkeypatch, element_type) -> None: captured: dict[str, object] = {} - def fake_analyze_valve_isolation(network, accident_element, disabled_valves): + def fake_analyze_valve_isolation(network, accident_element, disabled_valves, accident_element_type): captured.update( network=network, accident_element=accident_element, disabled_valves=disabled_valves, + accident_element_type=accident_element_type, business_dsn=get_project_pgconn_string(network), timescale_dsn=get_project_timescale_pgconn_string(network), ) @@ -288,6 +290,7 @@ def test_valve_isolation_runtime_accepts_frontend_query(monkeypatch) -> None: ("accident_element", "P-1"), ("accident_element", "P-2"), ("disabled_valves", "V-9"), + *([] if element_type == "auto" else [("accident_element_type", element_type)]), ], ) @@ -297,11 +300,39 @@ def test_valve_isolation_runtime_accepts_frontend_query(monkeypatch) -> None: "network": "fengyang", "accident_element": ["P-1", "P-2"], "disabled_valves": ["V-9"], + "accident_element_type": element_type, "business_dsn": "postgresql://user:password@biz/fengyang", "timescale_dsn": "postgresql://user:password@timescale/fengyang", } +@pytest.mark.parametrize( + "target,disabled,detail", + [("P", "typo", "Disabled valve"), ("missing", "V", "Accident element")], +) +def test_valve_isolation_invalid_ids_return_validation_error(monkeypatch, target, disabled, detail): + from app.services import valve_isolation as service + + monkeypatch.setattr( + service, "get_valve_isolation_topology", + lambda _network: (["P:pipe:A:B", "V:valve:B:R"], ["R"], []), + ) + monkeypatch.setattr(simulation_endpoint, "analyze_valve_isolation", service.analyze_valve_isolation) + app = FastAPI() + app.include_router(api_router, prefix="/api/v1") + _override_project_routing(app, ProjectContext( + project_id=uuid4(), project_code="project", user_id=uuid4(), project_role="member", + )) + + response = TestClient(app, raise_server_exceptions=False).post( + "/api/v1/valve-isolation-analyses", + params={"accident_element": target, "disabled_valves": disabled}, + ) + + assert response.status_code == 422 + assert detail in response.json()["detail"] + + def test_scada_cleaning_runs_are_post() -> None: methods_by_path = { route.path: route.methods diff --git a/tests/unit/test_valve_isolation.py b/tests/unit/test_valve_isolation.py index 34af18e..4a3b2be 100644 --- a/tests/unit/test_valve_isolation.py +++ b/tests/unit/test_valve_isolation.py @@ -1,3 +1,5 @@ +import pytest + from app.algorithms.valve_isolation import topology_search as valve @@ -23,6 +25,183 @@ def test_isolatable_keeps_affected_node_ids_and_count(): 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", diff --git a/tests/unit/test_wndb_network_views.py b/tests/unit/test_wndb_network_views.py index 86c62af..a266145 100644 --- a/tests/unit/test_wndb_network_views.py +++ b/tests/unit/test_wndb_network_views.py @@ -52,6 +52,36 @@ def test_network_link_nodes_use_one_unified_view_query(monkeypatch) -> None: assert "FROM gis.network_links" in calls[0][1] +def test_isolation_snapshot_reads_authoritative_links_statuses_and_sources_once(monkeypatch): + calls = [] + + def fake_read_all(name, statement, params=None): + calls.append((name, statement, params)) + return [ + {"id": "P", "element_type": "pipe", "start_node_id": "A", "end_node_id": "B", "status": None}, + {"id": "P-closed", "element_type": "pipe", "start_node_id": "A", "end_node_id": "R", "status": "CLOSED"}, + {"id": "V", "element_type": "valve", "start_node_id": "B", "end_node_id": "R", "status": "CLOSED"}, + {"id": "PU", "element_type": "pump", "start_node_id": "T", "end_node_id": "A", "status": "CLOSED"}, + {"id": "R", "element_type": "reservoir", "start_node_id": None, "end_node_id": None, "status": None}, + {"id": "T", "element_type": "tank", "start_node_id": None, "end_node_id": None, "status": None}, + ] + + monkeypatch.setattr(network_views, "read_all", fake_read_all) + + assert network_views.get_valve_isolation_topology("project_a") == ( + ["P:pipe:A:B", "P-closed:pipe:A:R", "V:valve:B:R", "PU:pump:T:A"], + ["R", "T"], ["P-closed", "V", "PU"], + ) + assert len(calls) == 1 + name, statement, _params = calls[0] + assert name == "project_a" + assert "FROM network.links" in statement + assert "network.link_initial_settings" in statement + assert "COALESCE(s.status::text, p.status::text)" in statement + assert "FROM network.nodes" in statement + assert "UNION ALL" in statement + + def test_topology_rows_are_loaded_in_two_batch_queries(monkeypatch) -> None: calls = [] responses = [