Compare commits

...
Author SHA1 Message Date
jiang 38a51ed65c fix(isolation): distinguish accident links from same-numbered nodes
Generic Container CI/CD / test-build-publish (push) Successful in 2m29s
Server CI/CD v2 / build-test-publish-and-deploy (push) Successful in 2m29s
Prefer link endpoints for legacy accident IDs and allow explicit node selection. Read authoritative initial statuses and sources, verify source isolation, and separate direct isolation from newly disconnected supply nodes. Return validation errors as HTTP 422.

The node-first lookup misrouted adjacent Fengyang pipes 19025 and 19026 into unrelated regions. Regression cases fail against the previous algorithm and pass after this change.
2026-10-29 11:20:34 +08:00
jiang a3b4e75317 fix(api): return real total for user pagination
Generic Container CI/CD / test-build-publish (push) Successful in 2m13s
Server CI/CD v2 / build-test-publish-and-deploy (push) Successful in 2m13s
2026-09-15 10:55:29 +08:00
12 changed files with 440 additions and 48 deletions
@@ -1,10 +1,11 @@
"""Topology-only valve isolation search.""" """Topology-only valve isolation search."""
from collections import defaultdict, deque from collections import defaultdict, deque
from typing import Any, Iterable from typing import Any, Iterable, Literal
VALVE_LINK_TYPE = "valve" VALVE_LINK_TYPE = "valve"
AccidentElementType = Literal["auto", "link", "node"]
def _parse_link_entry(link_entry: str) -> tuple[str, str, str, str]: 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] 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( def valve_isolation_analysis(
link_entries: Iterable[str], link_entries: Iterable[str],
accident_elements: str | list[str], accident_elements: str | list[str],
disabled_valves: list[str] | None = None, 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]: ) -> 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 []) disabled_valves_set = set(disabled_valves or [])
target_elements = ( target_elements = (
@@ -27,8 +53,13 @@ def valve_isolation_analysis(
if isinstance(accident_elements, str) if isinstance(accident_elements, str)
else accident_elements 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]] = {} all_valves: dict[str, tuple[str, str]] = {}
link_lookup: dict[str, tuple[str, str, str]] = {} link_lookup: dict[str, tuple[str, str, str]] = {}
node_set: set[str] = set() node_set: set[str] = set()
@@ -39,46 +70,31 @@ def valve_isolation_analysis(
node_set.update((node1, node2)) node_set.update((node1, node2))
if link_type_name == VALVE_LINK_TYPE: if link_type_name == VALVE_LINK_TYPE:
all_valves[link_id] = (node1, node2) all_valves[link_id] = (node1, node2)
else: if link_id not in closed_links_set:
pipe_adj[node1].add(node2) adjacency[node1].append((node2, link_id))
pipe_adj[node2].add(node1) 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() start_nodes: set[str] = set()
for element in target_elements: for element in target_elements:
if element in node_set: if accident_element_type != "node" and element in link_lookup:
start_nodes.add(element)
elif element in link_lookup:
node1, node2, _ = link_lookup[element] node1, node2, _ = link_lookup[element]
start_nodes.update((node1, node2)) start_nodes.update((node1, node2))
elif accident_element_type != "link" and element in node_set:
start_nodes.add(element)
else: else:
raise ValueError(f"Accident element {element} was not found in topology") raise ValueError(f"Accident element {element} was not found in topology")
extra_adj: dict[str, list[str]] = defaultdict(list) closable_valves = set(all_valves) - disabled_valves_set - closed_links_set
boundary_valves: dict[str, tuple[str, str]] = {} affected_nodes = _reachable_nodes(adjacency, start_nodes, closable_valves)
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
)
must_close_valves: list[str] = [] must_close_valves: list[str] = []
optional_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 node1_affected = node1 in affected_nodes
node2_affected = node2 in affected_nodes node2_affected = node2 in affected_nodes
if node1_affected and node2_affected: if node1_affected and node2_affected:
@@ -89,14 +105,37 @@ def valve_isolation_analysis(
must_close_valves.sort() must_close_valves.sort()
optional_valves.sort() optional_valves.sort()
isolatable = bool(must_close_valves) 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] = { result: dict[str, Any] = {
"accident_elements": target_elements, "accident_elements": target_elements,
"accident_element_type": accident_element_type,
"disabled_valves": disabled_valves, "disabled_valves": disabled_valves,
"affected_nodes": sorted(affected_nodes) if isolatable else [], "affected_nodes": sorted(affected_nodes) if isolatable else [],
"affected_node_count": len(affected_nodes), "affected_node_count": len(affected_nodes),
"must_close_valves": must_close_valves, "must_close_valves": must_close_valves,
"optional_valves": optional_valves, "optional_valves": optional_valves,
"isolatable": isolatable, "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: if len(target_elements) == 1:
result["accident_element"] = target_elements[0] result["accident_element"] = target_elements[0]
+4 -1
View File
@@ -7,6 +7,7 @@ from sqlalchemy.engine.url import make_url
from sqlalchemy.exc import IntegrityError, SQLAlchemyError from sqlalchemy.exc import IntegrityError, SQLAlchemyError
from sqlalchemy.ext.asyncio import create_async_engine from sqlalchemy.ext.asyncio import create_async_engine
from app.api.pagination import PaginatedList
from app.auth.metadata_dependencies import ( from app.auth.metadata_dependencies import (
get_current_metadata_admin, get_current_metadata_admin,
get_metadata_repository, get_metadata_repository,
@@ -236,7 +237,9 @@ async def list_metadata_users(
metadata_repo: MetadataRepository = Depends(get_metadata_repository), metadata_repo: MetadataRepository = Depends(get_metadata_repository),
) -> List[MetadataUserResponse]: ) -> List[MetadataUserResponse]:
users = await metadata_repo.list_users(skip=skip, limit=limit) users = await metadata_repo.list_users(skip=skip, limit=limit)
return [MetadataUserResponse.model_validate(user) for user in users] total = await metadata_repo.count_users()
items = [MetadataUserResponse.model_validate(user) for user in users]
return PaginatedList(items, total=total)
@router.get("/admin/projects", response_model=List[AdminProjectResponse]) @router.get("/admin/projects", response_model=List[AdminProjectResponse])
+19 -4
View File
@@ -180,6 +180,9 @@ def valve_isolation_endpoint(
network: str = Query(..., description="管网名称(或数据库名称)"), network: str = Query(..., description="管网名称(或数据库名称)"),
accident_element: List[str] = Query(..., description="发生事故的管段/节点ID列表"), accident_element: List[str] = Query(..., description="发生事故的管段/节点ID列表"),
disabled_valves: List[str] = Query(None, 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**: 管网名称(或数据库名称) - **network**: 管网名称(或数据库名称)
- **accident_element**: 发生事故的管段/节点ID列表 - **accident_element**: 发生事故的管段/节点ID列表
- **disabled_valves**: 已故障的阀门ID列表(可选) - **disabled_valves**: 已故障的阀门ID列表(可选)
- **accident_element_type**: auto优先管段;link仅匹配管段;node仅匹配节点
返回隔离方案,包括: 返回隔离方案,包括:
- must_close_valves: 必须关闭的阀门列表 - must_close_valves: 必须关闭的阀门列表
- optional_valves: 可选关闭的阀门列表 - optional_valves: 可选关闭的阀门列表
- affected_nodes: 受影响的节点列表;不可隔离时为空列表 - affected_nodes / affected_node_count: 最近阀门围成的直接隔离区;不可隔离时列表为空,计数保留
- affected_node_count: 受影响的节点总数 - supply_affected_nodes / supply_affected_node_count: 按建议关阀后新增失去水源连接的节点;未验证或不可隔离时为null
- isolatable: 是否可以有效隔离 - source_check_performed: 是否已执行水源连通检查
- isolation_failure_reason: 不可隔离原因;可隔离时为null
- isolatable: 存在关阀边界,且直接隔离区不再连通任何水源
使用模型初始开关状态进行无向拓扑分析,水池按潜在水源处理。
不包含时变控制、水池蓄水量或压力流量计算,不是现场停水范围的水力证明。
""" """
# result = { # result = {
# "accident_element": "P461309", # "accident_element": "P461309",
@@ -212,7 +221,13 @@ def valve_isolation_endpoint(
# "must_close_valves": ["210521658", "V12974", "V12986", "V12993"], # "must_close_valves": ["210521658", "V12974", "V12986", "V12993"],
# "optional_valves": [], # "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 return result
@@ -4,7 +4,7 @@ from typing import Optional, List
from uuid import UUID, uuid4 from uuid import UUID, uuid4
from cryptography.fernet import InvalidToken from cryptography.fernet import InvalidToken
from sqlalchemy import delete, select from sqlalchemy import delete, func, select
from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.ext.asyncio import AsyncSession
from app.core.encryption import ( from app.core.encryption import (
@@ -122,6 +122,12 @@ class MetadataRepository:
) )
return list(result.scalars().all()) return list(result.scalars().all())
async def count_users(self) -> int:
result = await self.session.execute(
select(func.count()).select_from(models.User)
)
return int(result.scalar() or 0)
async def upsert_user_from_keycloak( async def upsert_user_from_keycloak(
self, self,
*, *,
+41 -1
View File
@@ -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 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]: def get_major_pipe_nodes(name: str, diameter: int) -> list[str]:
"""Return large pipes in the established API wire format.""" """Return large pipes in the established API wire format."""
rows = read_all( rows = read_all(
+14 -5
View File
@@ -1,13 +1,22 @@
from typing import Any from typing import Any
from app.algorithms.valve_isolation.topology_search import valve_isolation_analysis from app.algorithms.valve_isolation.topology_search import (
from app.native.wndb.gis.network_views import get_network_link_nodes AccidentElementType,
valve_isolation_analysis,
)
from app.native.wndb.gis.network_views import get_valve_isolation_topology
def analyze_valve_isolation( def analyze_valve_isolation(
network: str, network: str,
accident_element: str | list[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]: ) -> dict[str, Any]:
link_entries = get_network_link_nodes(network) link_entries, source_nodes, closed_links = get_valve_isolation_topology(network)
return valve_isolation_analysis(link_entries, accident_element, disabled_valves) 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,
)
+1 -1
View File
@@ -3,7 +3,7 @@
"contracts": { "contracts": {
"server": { "server": {
"file": "server-v1.openapi.json", "file": "server-v1.openapi.json",
"sha256": "b565d841061c9091f48ff3118bcc0cbb1b918b0cb8c2316d177570e8b2d8ba29" "sha256": "5b8ec0b469f1194e72214003a9cf427489f48de19ae31ee6492821cf64916b6d"
} }
} }
} }
+17
View File
@@ -35969,6 +35969,23 @@
"type": "array" "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", "in": "header",
"name": "X-Project-Id", "name": "X-Project-Id",
+24 -1
View File
@@ -7,8 +7,9 @@ import pytest
from fastapi import HTTPException from fastapi import HTTPException
from fastapi import Response from fastapi import Response
from app.auth.metadata_dependencies import get_current_metadata_admin
from app.api.v1.endpoints import admin_metadata from app.api.v1.endpoints import admin_metadata
from app.api.pagination import PaginatedList
from app.auth.metadata_dependencies import get_current_metadata_admin
from app.domain.schemas.admin_metadata import ( from app.domain.schemas.admin_metadata import (
AdminProjectCreateRequest, AdminProjectCreateRequest,
MetadataUsersBatchSyncRequest, MetadataUsersBatchSyncRequest,
@@ -83,6 +84,28 @@ def test_to_async_sqlalchemy_url_preserves_password():
assert "***" not in url assert "***" not in url
@pytest.mark.anyio
async def test_list_metadata_users_returns_page_with_real_total():
users = [_user(username="alice"), _user(username="bob")]
repo = SimpleNamespace(
list_users=AsyncMock(return_value=users),
count_users=AsyncMock(return_value=7),
)
response = await admin_metadata.list_metadata_users(
skip=2,
limit=2,
current_user=_user(role="admin", is_superuser=True),
metadata_repo=repo,
)
assert isinstance(response, PaginatedList)
assert response.total == 7
assert [item.username for item in response] == ["alice", "bob"]
repo.list_users.assert_awaited_once_with(skip=2, limit=2)
repo.count_users.assert_awaited_once_with()
@pytest.mark.anyio @pytest.mark.anyio
async def test_sync_metadata_user_upserts_without_password(monkeypatch): async def test_sync_metadata_user_upserts_without_password(monkeypatch):
keycloak_id = uuid4() keycloak_id = uuid4()
+33 -2
View File
@@ -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" 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] = {} 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( captured.update(
network=network, network=network,
accident_element=accident_element, accident_element=accident_element,
disabled_valves=disabled_valves, disabled_valves=disabled_valves,
accident_element_type=accident_element_type,
business_dsn=get_project_pgconn_string(network), business_dsn=get_project_pgconn_string(network),
timescale_dsn=get_project_timescale_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-1"),
("accident_element", "P-2"), ("accident_element", "P-2"),
("disabled_valves", "V-9"), ("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", "network": "fengyang",
"accident_element": ["P-1", "P-2"], "accident_element": ["P-1", "P-2"],
"disabled_valves": ["V-9"], "disabled_valves": ["V-9"],
"accident_element_type": element_type,
"business_dsn": "postgresql://user:password@biz/fengyang", "business_dsn": "postgresql://user:password@biz/fengyang",
"timescale_dsn": "postgresql://user:password@timescale/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: def test_scada_cleaning_runs_are_post() -> None:
methods_by_path = { methods_by_path = {
route.path: route.methods route.path: route.methods
+179
View File
@@ -1,3 +1,5 @@
import pytest
from app.algorithms.valve_isolation import topology_search as valve 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"] 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(): def test_disabled_valve_expands_affected_area_before_counting():
links = [ links = [
"P-1:pipe:A:B", "P-1:pipe:A:B",
+30
View File
@@ -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] 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: def test_topology_rows_are_loaded_in_two_batch_queries(monkeypatch) -> None:
calls = [] calls = []
responses = [ responses = [