feat(auth): sync Keycloak users for authorization

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.

.env.example

@@ -1,90 +1,83 @@
-# TJWater Server 环境变量配置模板
-# 复制此文件为 .env 并填写实际值
-# CI/CD: 生产环境变量由 Dev 主机的受控 backend.env 注入，不要将完整 .env 保存为 Gitea 仓库密钥。
-ENVIRONMENT="production"
-NETWORK_NAME="tjwater"
-# ============================================
-# 敏感配置加密 (必填)
-# ============================================
+# TJWaterServerBinary 环境变量模板
+# 复制为 .env 后按启用的功能填写；不要提交 .env 或任何真实凭据。
+# 生产配置由部署主机的受控 backend.env 注入，不随镜像或 CI/CD 同步本地 .env。
+# 进程环境变量优先；修改 Compose env_file 后需重新创建容器。
 
-# Fernet 格式，生产环境必须替换为独立密钥
-# 生成方式: python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())"
-# 用于项目数据库 DSN、GeoServer 管理密码等敏感配置
+# 基础运行
+PROJECT_NAME="TJWater Server"
+ENVIRONMENT="production"
+API_V1_STR="/api/v1"
+NETWORK_NAME="tjwater"
+
+# Fernet 格式；用于加密项目数据库 DSN、GeoServer 管理密码等敏感配置。
+# 生成方式：python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())"
 DATABASE_ENCRYPTION_KEY="replace-with-generated-fernet-key"
 
-# ============================================
-# 数据库配置 (PostgreSQL)
-# ============================================
+# 业务数据库（PostgreSQL）
 DB_NAME="tjwater"
 DB_HOST="localhost"
 DB_PORT="5432"
 DB_USER="tjwater"
 DB_PASSWORD="password"
 
-# ============================================
-# 数据库配置 (TimescaleDB)
-# ============================================
+# 时序数据库（TimescaleDB）
 TIMESCALEDB_DB_NAME="tjwater"
 TIMESCALEDB_DB_HOST="localhost"
 TIMESCALEDB_DB_PORT="5433"
 TIMESCALEDB_DB_USER="tjwater"
 TIMESCALEDB_DB_PASSWORD="password"
 
-# ============================================
-# 元数据数据库配置 (Metadata DB)
-# ============================================
+# 元数据数据库
 METADATA_DB_NAME="system_hub"
 METADATA_DB_HOST="localhost"
 METADATA_DB_PORT="5432"
 METADATA_DB_USER="tjwater"
 METADATA_DB_PASSWORD="password"
+METADATA_DB_POOL_SIZE=5
+METADATA_DB_MAX_OVERFLOW=10
 
-# Per-project synchronous connection pools
-PROJECT_PG_CACHE_SIZE="16"
-PROJECT_TS_CACHE_SIZE="16"
-PROJECT_PG_POOL_MIN_SIZE="0"
-PROJECT_PG_POOL_SIZE="4"
-PROJECT_PG_MAX_OVERFLOW="2"
-PROJECT_TS_POOL_MIN_SIZE="0"
-PROJECT_TS_POOL_MAX_SIZE="4"
+# 项目数据库连接池、缓存与模板
+PROJECT_PG_CACHE_SIZE=16
+PROJECT_TS_CACHE_SIZE=16
+PROJECT_PG_POOL_MIN_SIZE=0
+PROJECT_PG_POOL_SIZE=4
+PROJECT_PG_MAX_OVERFLOW=2
+PROJECT_TS_POOL_MIN_SIZE=0
+PROJECT_TS_POOL_MAX_SIZE=4
 WNDB_SCHEMA_TEMPLATE_DB_NAME="tjwater_v2_schema_template"
 TIMESCALEDB_SCHEMA_TEMPLATE_DB_NAME="tjwater_v2_timescale_template"
-WNDB_TEMP_DB_MAX_COUNT="8"
+WNDB_TEMP_DB_MAX_COUNT=8
 
-# ============================================
 # GeoServer 项目供应
-# ============================================
 GEOSERVER_URL="http://localhost:8080/geoserver"
 GEOSERVER_USERNAME="admin"
 GEOSERVER_PASSWORD="password"
-# 留空时分别复用 DB_HOST / DB_PORT / DB_USER / DB_PASSWORD。
+# 留空时分别复用 DB_HOST、DB_PORT、DB_USER、DB_PASSWORD。
 # GeoServer 在容器内运行时，GEOSERVER_DB_HOST 通常应填写数据库服务名。
 GEOSERVER_DB_HOST=""
 GEOSERVER_DB_PORT=""
 GEOSERVER_DB_USER=""
 GEOSERVER_DB_PASSWORD=""
-# 30 天客户端缓存，单位为秒。
-GEOSERVER_CLIENT_CACHE_SECONDS="2592000"
+# 客户端缓存时间，单位为秒；默认 30 天。
+GEOSERVER_CLIENT_CACHE_SECONDS=2592000
 
-# ============================================
-# Keycloak JWT (可选)
-# ============================================
-KEYCLOAK_PUBLIC_KEY="-----BEGIN PUBLIC KEY-----\n...\n-----END PUBLIC KEY-----"
-KEYCLOAK_ALGORITHM=RS256
+# Keycloak JWT 校验
+# 未启用认证时可留空；生产环境应注入完整公钥和预期 audience。
+KEYCLOAK_PUBLIC_KEY=""
+KEYCLOAK_ALGORITHM="RS256"
 KEYCLOAK_AUDIENCE="account"
 KEYCLOAK_ACCESS_TOKEN_MAX_AGE_SECONDS=900
 
+# 浏览器跨域访问（主 API）
+# 必须填写明确来源（协议、域名或 IP、端口），英文逗号分隔；不允许 *。
+CORS_ALLOWED_ORIGINS=http://127.0.0.1:5173,http://localhost:5173,http://127.0.0.1:15173,http://localhost:15173
 
-# ============================================
-# Bocha Web Search API
-# ============================================
-BOCHA_API_KEY="sk-your-bocha-api-key"
+# Bocha 网页搜索（旧版仍保留；未启用时令牌留空）
+BOCHA_API_KEY=""
 BOCHA_WEB_SEARCH_URL="https://api.bochaai.com/v1/web-search"
 BOCHA_WEB_SEARCH_TIMEOUT_SECONDS=30
 
-# ============================================
-# Tianditu Geocoding API
-# ============================================
-TIANDITU_GEOCODER_TOKEN="your-tianditu-geocoder-token"
+# 天地图地理编码（旧版使用 TIANDITU_GEOCODER_TOKEN）
+TIANDITU_GEOCODER_TOKEN=""
 TIANDITU_GEOCODER_URL="https://api.tianditu.gov.cn/geocoder"
 TIANDITU_GEOCODER_TIMEOUT_SECONDS=30

AUTHENTICATION_AND_USER_MANAGEMENT.md

@@ -37,11 +37,17 @@ but still need a project membership for project business APIs.
 Every authenticated metadata-user resolution validates the Keycloak access token
 and reads `sub`, `preferred_username`, and `email` claims. The
 backend finds `users` by `keycloak_id = sub`, rejects inactive or missing users,
-then refreshes `username`, `email`, and `last_login_at`.
+then refreshes `username`, `email`, and `last_login_at` for active users.
 
-This keeps local display data current without changing the identity binding.
-There is no Keycloak webhook requirement; second-level user or permission sync is
-out of scope unless explicitly requested later.
+System administrators explicitly synchronize the Keycloak user directory from
+the system-management page. New identities are stored with role `user` and
+`is_active = false`; existing identities keep their TJWater role, active state,
+superuser flag, and project memberships. This makes pending accounts visible
+without requiring the user to log in and without granting access automatically.
+The administrator's Keycloak account must have the `realm-management`
+`query-users` and `view-users` client roles. The backend forwards that access
+token only to the same token issuer's Keycloak Admin API; no management
+credential is exposed to the browser.
 
 ## Admin APIs
 

README.md

@@ -43,6 +43,22 @@ conda run -n server uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload
 conda activate server
 ```
 
+## CORS 配置
+
+后端通过 `.env` 或进程环境变量中的 `CORS_ALLOWED_ORIGINS` 设置允许的浏览器页面来源，
+进程环境变量优先。必须使用英文逗号分隔的明确来源，不能填写 `*`，例如：
+
+```dotenv
+CORS_ALLOWED_ORIGINS=https://demo.waternetwork.cn,http://192.168.1.133,https://192.168.1.133,http://192.168.1.110,https://192.168.1.110
+```
+
+未设置时默认只允许本机 `5173`、`15173` 开发来源。来源需包含协议、域名或 IP、端口，
+不带路径；非默认端口须逐项列出。CORS 匹配浏览器页面来源，不是 Nginx 连接 IP 或域名解析结果。
+保留跨域凭据支持及 `Authorization`、`X-Project-Id` 请求头，接口认证和权限校验继续生效。
+
+修改后需重启后端进程。容器部署需同步更新宿主机的受控 `backend.env`；
+使用 Compose `env_file` 注入时，需要重新创建容器才能读取新值，CI/CD 不会自动同步本地 `.env`。
+
 ## 常用命令
 
 ```bash

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]

app/api/v1/endpoints/admin_metadata.py

@@ -12,11 +12,17 @@ from app.auth.metadata_dependencies import (
     get_current_metadata_admin,
     get_metadata_repository,
 )
+from app.auth.keycloak_dependencies import (
+    get_current_keycloak_payload,
+    oauth2_optional,
+)
 from app.core.audit import AuditAction, log_audit_event
 from app.domain.schemas.admin_metadata import (
     AdminProjectCreateRequest,
     AdminProjectResponse,
     AdminProjectUpdateRequest,
+    KeycloakUserSyncIssue,
+    KeycloakUserSyncResponse,
     MetadataUsersBatchSyncRequest,
     MetadataUserResponse,
     MetadataUserSyncRequest,
@@ -36,6 +42,11 @@ from app.infra.db.metadb.repositories.metadata_repository import (
     MetadataRepository,
     ProjectDbRouting,
 )
+from app.services.keycloak_user_sync import (
+    KeycloakUserSyncConfigurationError,
+    KeycloakUserSyncRequestError,
+    fetch_keycloak_users,
+)
 
 router = APIRouter()
 
@@ -229,6 +240,94 @@ async def sync_metadata_users_batch(
     return results
 
 
+@router.post(
+    "/admin/keycloak-user-syncs",
+    response_model=KeycloakUserSyncResponse,
+)
+async def sync_keycloak_user_directory(
+    current_user=Depends(get_current_metadata_admin),
+    keycloak_payload: dict = Depends(get_current_keycloak_payload),
+    keycloak_access_token: str | None = Depends(oauth2_optional),
+    metadata_repo: MetadataRepository = Depends(get_metadata_repository),
+) -> KeycloakUserSyncResponse:
+    try:
+        keycloak_users = await fetch_keycloak_users(
+            issuer=str(keycloak_payload.get("iss") or ""),
+            bearer_token=keycloak_access_token or "",
+        )
+    except KeycloakUserSyncConfigurationError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail=str(exc),
+        ) from exc
+    except KeycloakUserSyncRequestError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_502_BAD_GATEWAY,
+            detail=str(exc),
+        ) from exc
+
+    created = 0
+    updated = 0
+    issues: list[KeycloakUserSyncIssue] = []
+    for keycloak_user in keycloak_users:
+        if keycloak_user.service_account_client_id:
+            issues.append(
+                KeycloakUserSyncIssue(
+                    username=keycloak_user.username,
+                    reason="service_account",
+                )
+            )
+            continue
+        if not keycloak_user.email:
+            issues.append(
+                KeycloakUserSyncIssue(
+                    username=keycloak_user.username,
+                    reason="missing_email",
+                )
+            )
+            continue
+        try:
+            _, was_created = await metadata_repo.sync_user_snapshot_from_keycloak(
+                keycloak_id=keycloak_user.keycloak_id,
+                username=keycloak_user.username,
+                email=keycloak_user.email,
+            )
+        except IntegrityError:
+            await metadata_repo.session.rollback()
+            issues.append(
+                KeycloakUserSyncIssue(
+                    username=keycloak_user.username,
+                    reason="identity_conflict",
+                )
+            )
+        except SQLAlchemyError as exc:
+            await metadata_repo.session.rollback()
+            raise HTTPException(
+                status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+                detail="Metadata database error during Keycloak user synchronization",
+            ) from exc
+        else:
+            created += int(was_created)
+            updated += int(not was_created)
+
+    result = KeycloakUserSyncResponse(
+        scanned=len(keycloak_users),
+        created=created,
+        updated=updated,
+        skipped=len(issues),
+        issues=issues,
+    )
+    await log_audit_event(
+        action=AuditAction.UPDATE,
+        user_id=current_user.id,
+        resource_type="keycloak_user_directory",
+        request_data=result.model_dump(mode="json", exclude={"issues"}),
+        response_status=status.HTTP_200_OK,
+        session=metadata_repo.session,
+    )
+    return result
+
+
 @router.get("/admin/users", response_model=List[MetadataUserResponse])
 async def list_metadata_users(
     skip: int = Query(0, ge=0),

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
 
 

app/auth/metadata_dependencies.py

@@ -52,6 +52,7 @@ async def get_current_metadata_user(
 ):
     keycloak_sub = _keycloak_sub_from_payload(keycloak_payload)
     username = get_keycloak_preferred_username(keycloak_payload)
+    email = _email_from_payload(keycloak_payload)
     try:
         user = await metadata_repo.get_user_by_keycloak_id(keycloak_sub)
     except SQLAlchemyError as exc:
@@ -65,13 +66,14 @@ async def get_current_metadata_user(
         ) from exc
     if not user or not user.is_active:
         raise HTTPException(
-            status_code=status.HTTP_403_FORBIDDEN, detail="Inactive user"
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="User authorization pending",
         )
     try:
         user = await metadata_repo.refresh_user_keycloak_snapshot(
             user,
             username=username,
-            email=_email_from_payload(keycloak_payload),
+            email=email,
         )
     except SQLAlchemyError as exc:
         logger.error(

app/core/config.py

@@ -63,6 +63,19 @@ class Settings(BaseSettings):
     KEYCLOAK_AUDIENCE: str = ""
     KEYCLOAK_ACCESS_TOKEN_MAX_AGE_SECONDS: int = 900
 
+    # Browser cross-origin access; process environment overrides .env.
+    CORS_ALLOWED_ORIGINS: str = (
+        "http://127.0.0.1:5173,http://localhost:5173,"
+        "http://127.0.0.1:15173,http://localhost:15173"
+    )
+
+    @property
+    def cors_allowed_origins(self) -> list[str]:
+        origins = [value.strip() for value in self.CORS_ALLOWED_ORIGINS.split(",") if value.strip()]
+        if "*" in origins:
+            raise ValueError("CORS_ALLOWED_ORIGINS must list explicit origins")
+        return origins
+
     # Bocha Web Search API
     BOCHA_API_KEY: str = ""
     BOCHA_WEB_SEARCH_URL: str = "https://api.bochaai.com/v1/web-search"

app/domain/schemas/admin_metadata.py

@@ -50,6 +50,19 @@ class MetadataUserSyncResult(BaseModel):
     error: str | None = None
 
 
+class KeycloakUserSyncIssue(BaseModel):
+    username: str
+    reason: str
+
+
+class KeycloakUserSyncResponse(BaseModel):
+    scanned: int
+    created: int
+    updated: int
+    skipped: int
+    issues: list[KeycloakUserSyncIssue] = Field(default_factory=list)
+
+
 class ProjectMemberCreateRequest(BaseModel):
     user_id: UUID
     project_role: ProjectRole = "viewer"

app/infra/db/metadb/repositories/metadata_repository.py

@@ -5,6 +5,7 @@ from uuid import UUID, uuid4
 
 from cryptography.fernet import InvalidToken
 from sqlalchemy import delete, func, select
+from sqlalchemy.dialects.postgresql import insert
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.core.encryption import (
@@ -158,6 +159,44 @@ class MetadataRepository:
         await self.session.refresh(user)
         return user
 
+    async def sync_user_snapshot_from_keycloak(
+        self,
+        *,
+        keycloak_id: UUID,
+        username: str,
+        email: str,
+    ) -> tuple[models.User, bool]:
+        user = await self.get_user_by_keycloak_id(keycloak_id)
+        if user is not None:
+            user.username = username
+            user.email = email
+            user.updated_at = _utcnow()
+            await self.session.commit()
+            await self.session.refresh(user)
+            return user, False
+
+        statement = (
+            insert(models.User)
+            .values(
+                id=uuid4(),
+                keycloak_id=keycloak_id,
+                username=username,
+                email=email,
+                role="user",
+                is_active=False,
+                is_superuser=False,
+            )
+            .on_conflict_do_nothing(index_elements=[models.User.keycloak_id])
+            .returning(models.User.id)
+        )
+        result = await self.session.execute(statement)
+        inserted_id = result.scalar_one_or_none()
+        await self.session.commit()
+        user = await self.get_user_by_keycloak_id(keycloak_id)
+        if user is None:
+            raise RuntimeError("Keycloak user synchronization did not persist a user")
+        return user, inserted_id is not None
+
     async def refresh_user_keycloak_snapshot(
         self,
         user: models.User,

app/main.py

@@ -66,7 +66,7 @@ app.add_middleware(AuditMiddleware)
 # 确保这是你最后一个添加的 app.add_middleware
 app.add_middleware(
     CORSMiddleware,
-    allow_origins=["*"],  # 允许所有来源
+    allow_origins=settings.cors_allowed_origins,
     allow_credentials=True,
     allow_methods=["*"],
     allow_headers=["*"],

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(

app/native/wndb/inp/exporter.py

@@ -120,7 +120,7 @@ def dump_inp(project: str, inp: str, version: str = '3'):
             file.write('\n'.join(inp_out_demand(project)))
 
         elif name == STATUS:
-            file.write('\n'.join(inp_out_status(project)))
+            file.write('\n'.join(inp_out_status(project, version=version)))
 
         elif name == PATTERNS:
             if version == '3':
@@ -232,7 +232,7 @@ def export_inp(project: str, version: str = '3') -> ChangeSet:
             inp += '\n'.join(inp_out_demand(project))
 
         elif name == STATUS:
-            inp += '\n'.join(inp_out_status(project))
+            inp += '\n'.join(inp_out_status(project, version=version))
 
         elif name == PATTERNS:
             if version == '3':

app/native/wndb/model/status.py

@@ -93,14 +93,16 @@ def inp_in_status(line: str) -> str:
         return str(f"insert into network.link_initial_settings (link_id, status, setting) values ({sql_literal(link)}, null, {sql_literal(float(value))});")
 
 
-def inp_out_status(name: str) -> list[str]:
+def inp_out_status(name: str, version: str = '2') -> list[str]:
     lines = []
     objs = read_all(name, 'select link_id as link, status, setting from network.link_initial_settings order by link_id')
     for obj in objs:
         link = obj['link']
         status = obj['status'] if obj['status'] != None else ''
         setting = obj['setting'] if obj['setting'] != None else ''
-        if status != '':
+        # EPANET 2 valves use their configured setting by default; ACTIVE is
+        # an internal state, not a valid keyword in its [STATUS] section.
+        if status != '' and not (version == '2' and status.upper() == LINK_STATUS_ACTIVE):
             lines.append(f'{link} {status}')
         if setting != '':
             lines.append(f'{link} {setting}')

app/services/keycloak_user_sync.py

@@ -0,0 +1,117 @@
+from dataclasses import dataclass
+from urllib.parse import quote
+from uuid import UUID
+
+import httpx
+
+
+class KeycloakUserSyncConfigurationError(ValueError):
+    pass
+
+
+class KeycloakUserSyncRequestError(RuntimeError):
+    pass
+
+
+@dataclass(frozen=True)
+class KeycloakUserSnapshot:
+    keycloak_id: UUID
+    username: str
+    email: str | None
+    service_account_client_id: str | None = None
+
+
+def _keycloak_admin_users_url(issuer: str) -> str:
+    issuer = issuer.rstrip("/")
+    marker = "/realms/"
+    if not issuer or marker not in issuer:
+        raise KeycloakUserSyncConfigurationError(
+            "Keycloak token issuer is not a complete realm URL"
+        )
+    server_url, realm = issuer.rsplit(marker, 1)
+    if not server_url or not realm:
+        raise KeycloakUserSyncConfigurationError(
+            "Keycloak token issuer is not a complete realm URL"
+        )
+    return f"{server_url}/admin/realms/{quote(realm, safe='')}/users"
+
+
+async def fetch_keycloak_users(
+    *,
+    issuer: str,
+    bearer_token: str,
+    client: httpx.AsyncClient | None = None,
+) -> list[KeycloakUserSnapshot]:
+    admin_users_url = _keycloak_admin_users_url(issuer)
+    if not bearer_token:
+        raise KeycloakUserSyncConfigurationError(
+            "Keycloak administrator access token is missing"
+        )
+    owns_client = client is None
+    if client is None:
+        client = httpx.AsyncClient(timeout=15.0)
+
+    try:
+        try:
+            snapshots: list[KeycloakUserSnapshot] = []
+            first = 0
+            page_size = 100
+            while True:
+                response = await client.get(
+                    admin_users_url,
+                    params={"first": first, "max": page_size},
+                    headers={
+                        "Accept": "application/json",
+                        "Authorization": f"Bearer {bearer_token}",
+                    },
+                )
+                response.raise_for_status()
+                payload = response.json()
+                if not isinstance(payload, list):
+                    raise KeycloakUserSyncRequestError(
+                        "Keycloak users response is not a list"
+                    )
+                for item in payload:
+                    if not isinstance(item, dict):
+                        continue
+                    try:
+                        keycloak_id = UUID(str(item.get("id", "")))
+                    except ValueError:
+                        continue
+                    username = str(item.get("username") or "").strip()
+                    if not username:
+                        continue
+                    email_value = item.get("email")
+                    email = str(email_value).strip() if email_value else None
+                    service_account_value = item.get("serviceAccountClientId")
+                    snapshots.append(
+                        KeycloakUserSnapshot(
+                            keycloak_id=keycloak_id,
+                            username=username,
+                            email=email,
+                            service_account_client_id=(
+                                str(service_account_value)
+                                if service_account_value
+                                else None
+                            ),
+                        )
+                    )
+                if len(payload) < page_size:
+                    break
+                first += page_size
+            return snapshots
+        except httpx.HTTPStatusError as exc:
+            if exc.response.status_code == 403:
+                raise KeycloakUserSyncRequestError(
+                    "Keycloak administrator requires query-users and view-users roles"
+                ) from exc
+            raise KeycloakUserSyncRequestError(
+                "Keycloak user directory request failed"
+            ) from exc
+        except httpx.HTTPError as exc:
+            raise KeycloakUserSyncRequestError(
+                "Keycloak user directory request failed"
+            ) from exc
+    finally:
+        if owns_client:
+            await client.aclose()

app/services/simulation.py

@@ -490,6 +490,23 @@ def run_simulation(
                     set_status(name_c, cs)
         # 运行并返回结果
         result_data = json.loads(run_project(name_c))
+        if result_data.get("simulation_result") == "failed":
+            report = result_data.get("report") or ""
+            logging.error("EPANET failed for project %s:\n%s", name_c, report)
+            lines = report.splitlines()
+            error_index = next(
+                (i for i, line in enumerate(lines) if line.strip().startswith("Error ")),
+                None,
+            )
+            detail = (
+                " ".join(line.strip() for line in lines[error_index:error_index + 2])[:600]
+                if error_index is not None
+                else "引擎未返回错误详情，请查看服务日志"
+            )
+            raise RuntimeError(f"EPANET 计算失败：{detail}")
+        output_data = result_data.get("output")
+        if not isinstance(output_data, dict):
+            raise RuntimeError("run_project did not return JSON output content")
     refresh_materialized_views_after_commit(name_c)
     time_cost_end = time.perf_counter()
     print(
@@ -498,9 +515,6 @@ def run_simulation(
             time_cost_end - time_cost_start,
         )
     )
-    output_data = result_data.get("output")
-    if not isinstance(output_data, dict):
-        raise RuntimeError("run_project did not return JSON output content")
     node_result = output_data.get("node_results")
     link_result = output_data.get("link_results")
     if node_result is None or link_result is None:

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,
+    )

contracts/manifest.json

@@ -3,7 +3,7 @@
   "contracts": {
     "server": {
       "file": "server-v1.openapi.json",
-      "sha256": "b565d841061c9091f48ff3118bcc0cbb1b918b0cb8c2316d177570e8b2d8ba29"
+      "sha256": "19961572fad04f8545a4f607d4bf7276de64a84b266daecbb21d74e0975f97ef"
     }
   }
 }

tests/api/test_admin_metadata_endpoints.py

@@ -20,6 +20,7 @@ from app.domain.schemas.admin_metadata import (
     ProjectMemberUpdateRequest,
 )
 from app.infra.db.metadb.repositories.metadata_repository import ProjectDbRouting
+from app.services.keycloak_user_sync import KeycloakUserSnapshot
 
 
 @pytest.fixture
@@ -176,6 +177,66 @@ async def test_batch_sync_metadata_users_returns_per_user_results(monkeypatch):
     assert admin_metadata.log_audit_event.await_count == 2
 
 
+@pytest.mark.anyio
+async def test_sync_keycloak_user_directory_preserves_authorization(monkeypatch):
+    new_id = uuid4()
+    existing_id = uuid4()
+    repo = SimpleNamespace(
+        session=SimpleNamespace(rollback=AsyncMock()),
+        sync_user_snapshot_from_keycloak=AsyncMock(
+            side_effect=[
+                (_user(keycloak_id=new_id, is_active=False), True),
+                (_user(keycloak_id=existing_id, role="admin"), False),
+            ]
+        ),
+    )
+    monkeypatch.setattr(
+        admin_metadata,
+        "fetch_keycloak_users",
+        AsyncMock(
+            return_value=[
+                KeycloakUserSnapshot(new_id, "new-user", "new@example.com"),
+                KeycloakUserSnapshot(existing_id, "admin-user", "admin@example.com"),
+                KeycloakUserSnapshot(uuid4(), "no-email", None),
+                KeycloakUserSnapshot(
+                    uuid4(),
+                    "service-account-client",
+                    "service@example.com",
+                    service_account_client_id="client",
+                ),
+            ]
+        ),
+    )
+    monkeypatch.setattr(admin_metadata, "log_audit_event", AsyncMock())
+
+    response = await admin_metadata.sync_keycloak_user_directory(
+        current_user=_user(role="admin", is_superuser=True),
+        keycloak_payload={"iss": "https://keycloak.example.com/realms/tjwater"},
+        keycloak_access_token="admin-token",
+        metadata_repo=repo,
+    )
+
+    assert response.scanned == 4
+    assert response.created == 1
+    assert response.updated == 1
+    assert response.skipped == 2
+    assert [issue.reason for issue in response.issues] == [
+        "missing_email",
+        "service_account",
+    ]
+    assert repo.sync_user_snapshot_from_keycloak.await_count == 2
+    admin_metadata.fetch_keycloak_users.assert_awaited_once_with(
+        issuer="https://keycloak.example.com/realms/tjwater",
+        bearer_token="admin-token",
+    )
+    assert repo.sync_user_snapshot_from_keycloak.await_args_list[1].kwargs == {
+        "keycloak_id": existing_id,
+        "username": "admin-user",
+        "email": "admin@example.com",
+    }
+    admin_metadata.log_audit_event.assert_awaited_once()
+
+
 @pytest.mark.anyio
 async def test_update_metadata_user_updates_role_and_active_status(monkeypatch):
     user_id = uuid4()

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

tests/unit/test_analysis_simulation.py

@@ -277,3 +277,51 @@ def test_extended_simulation_marks_run_failed_when_result_storage_fails(monkeypa
         )
 
     assert [call[1]["status"] for call in lifecycle_calls] == ["failed"]
+
+
+@pytest.mark.parametrize("has_output", [False, True])
+def test_engine_failure_preserves_report_and_rolls_back_model_changes(monkeypatch, has_output):
+    from app.services import simulation
+
+    transaction_outcome = []
+
+    @contextmanager
+    def project_transaction(_name):
+        try:
+            yield
+        except Exception:
+            transaction_outcome.append("rollback")
+            raise
+        else:
+            transaction_outcome.append("commit")
+
+    monkeypatch.setattr(simulation, "project_transaction", project_transaction)
+    monkeypatch.setattr(simulation, "get_time", lambda _: {
+        "HYDRAULIC TIMESTEP": "0:15:00",
+        "DURATION": "0:00",
+        "PATTERN START": "0:00",
+    })
+    monkeypatch.setattr(simulation, "set_time", lambda *_: None)
+    result = {
+        "simulation_result": "failed",
+        "report": "EPANET report\n  Error 202: illegal numeric value ACTIVE in [STATUS] section:\n  V1 ACTIVE\n",
+    }
+    if has_output:
+        result["output"] = {"node_results": [], "link_results": [], "times": {"num_periods": 1}}
+    monkeypatch.setattr(simulation, "run_project", lambda _: json.dumps(result))
+    refresh = Mock()
+    store = Mock()
+    monkeypatch.setattr(simulation, "refresh_materialized_views_after_commit", refresh)
+    monkeypatch.setattr(simulation.TimescaleInternalStorage, "store_realtime_simulation", store)
+
+    with pytest.raises(RuntimeError, match="EPANET.*Error 202"):
+        simulation.run_simulation(
+            name="demo",
+            simulation_type="realtime",
+            modify_pattern_start_time="2026-07-30T04:30:00Z",
+            scada_mappings=_empty_scada_mappings(simulation),
+        )
+
+    assert transaction_outcome == ["rollback"]
+    refresh.assert_not_called()
+    store.assert_not_called()

tests/unit/test_cors.py

@@ -0,0 +1,73 @@
+import pytest
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.testclient import TestClient
+
+from app.core.config import Settings
+
+
+ALLOWED_ORIGINS = [
+    "https://demo.waternetwork.cn",
+    "http://192.168.1.133",
+    "https://192.168.1.133",
+    "http://192.168.1.110",
+    "https://192.168.1.110",
+]
+
+
+@pytest.mark.parametrize("origins", ["*", " *, http://localhost:5173", "http://localhost:5173,*"])
+def test_wildcard_origins_are_rejected(origins: str):
+    with pytest.raises(ValueError, match="must list explicit origins"):
+        Settings(_env_file=None, CORS_ALLOWED_ORIGINS=origins).cors_allowed_origins
+
+
+@pytest.mark.parametrize(
+    "origin",
+    ALLOWED_ORIGINS + [
+        "http://192.168.1.114",
+        "http://192.168.1.133:5173",
+        "https://next-demo.waternetwork.cn",
+        "http://192.168.1.133.evil.example",
+        "null",
+    ],
+)
+def test_explicit_origin_preflight(origin: str):
+    settings = Settings(_env_file=None, CORS_ALLOWED_ORIGINS=", ".join(ALLOWED_ORIGINS))
+    app = FastAPI()
+    app.add_middleware(
+        CORSMiddleware,
+        allow_origins=settings.cors_allowed_origins,
+        allow_credentials=True,
+        allow_methods=["*"],
+        allow_headers=["*"],
+    )
+    with TestClient(app) as client:
+        response = client.options(
+            "/api/v1/projects",
+            headers={
+                "Origin": origin,
+                "Access-Control-Request-Method": "GET",
+                "Access-Control-Request-Headers": "authorization,x-project-id",
+            },
+        )
+    allowed = origin in ALLOWED_ORIGINS
+    assert response.status_code == (200 if allowed else 400)
+    assert response.headers.get("access-control-allow-origin") == (origin if allowed else None)
+    if allowed:
+        assert response.headers["access-control-allow-credentials"] == "true"
+        assert "authorization" in response.headers["access-control-allow-headers"]
+
+
+def test_default_origins_are_explicit_local_sources(monkeypatch):
+    monkeypatch.delenv("CORS_ALLOWED_ORIGINS", raising=False)
+    settings = Settings(_env_file=None)
+    assert settings.cors_allowed_origins == [
+        "http://127.0.0.1:5173", "http://localhost:5173",
+        "http://127.0.0.1:15173", "http://localhost:15173",
+    ]
+
+
+def test_environment_controls_origins(monkeypatch):
+    monkeypatch.setenv("CORS_ALLOWED_ORIGINS", "http://192.168.1.133,https://192.168.1.110")
+    settings = Settings(_env_file=None)
+    assert settings.cors_allowed_origins == ["http://192.168.1.133", "https://192.168.1.110"]

tests/unit/test_epanet_status_export.py

@@ -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)

tests/unit/test_keycloak_user_sync.py

@@ -0,0 +1,54 @@
+from uuid import uuid4
+
+import httpx
+import pytest
+
+from app.services import keycloak_user_sync
+
+
+@pytest.fixture
+def anyio_backend():
+    return "asyncio"
+
+
+@pytest.mark.anyio
+async def test_fetch_keycloak_users_uses_admin_token_and_paginates():
+    user_ids = [uuid4() for _ in range(101)]
+    requested_first: list[int] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        assert request.headers["authorization"] == "Bearer admin-token"
+        first = int(request.url.params["first"])
+        requested_first.append(first)
+        page_ids = user_ids[first : first + 100]
+        return httpx.Response(
+            200,
+            json=[
+                {
+                    "id": str(user_id),
+                    "username": f"user-{first + index}",
+                    "email": f"user-{first + index}@example.com",
+                }
+                for index, user_id in enumerate(page_ids)
+            ],
+        )
+
+    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
+        users = await keycloak_user_sync.fetch_keycloak_users(
+            issuer="https://keycloak.example.com/realms/tjwater",
+            bearer_token="admin-token",
+            client=client,
+        )
+
+    assert len(users) == 101
+    assert requested_first == [0, 100]
+    assert users[-1].username == "user-100"
+
+
+@pytest.mark.anyio
+async def test_fetch_keycloak_users_requires_realm_issuer():
+    with pytest.raises(keycloak_user_sync.KeycloakUserSyncConfigurationError):
+        await keycloak_user_sync.fetch_keycloak_users(
+            issuer="",
+            bearer_token="admin-token",
+        )

tests/unit/test_metadata_repository_users.py

@@ -0,0 +1,44 @@
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+from uuid import uuid4
+
+import pytest
+
+from app.infra.db.metadb.repositories.metadata_repository import MetadataRepository
+
+
+@pytest.fixture
+def anyio_backend():
+    return "asyncio"
+
+
+@pytest.mark.anyio
+async def test_keycloak_snapshot_sync_preserves_existing_authorization():
+    keycloak_id = uuid4()
+    user = SimpleNamespace(
+        keycloak_id=keycloak_id,
+        username="old-name",
+        email="old@example.com",
+        role="admin",
+        is_active=True,
+        is_superuser=True,
+        updated_at=None,
+    )
+    session = SimpleNamespace(commit=AsyncMock(), refresh=AsyncMock())
+    repository = MetadataRepository(session)
+    repository.get_user_by_keycloak_id = AsyncMock(return_value=user)
+
+    result, created = await repository.sync_user_snapshot_from_keycloak(
+        keycloak_id=keycloak_id,
+        username="new-name",
+        email="new@example.com",
+    )
+
+    assert created is False
+    assert result.username == "new-name"
+    assert result.email == "new@example.com"
+    assert result.role == "admin"
+    assert result.is_active is True
+    assert result.is_superuser is True
+    session.commit.assert_awaited_once()
+    session.refresh.assert_awaited_once_with(user)

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",

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 = [