# Copyright (c) 2025-2026, RTE (https://www.rte-france.com)
# This Source Code Form is subject to the terms of the Mozilla Public License, version 2.0.
# If a copy of the Mozilla Public License, version 2.0 was not distributed with this file,
# you can obtain one at http://mozilla.org/MPL/2.0/.
# SPDX-License-Identifier: MPL-2.0
# This file is part of Co-Study4Grid a Power Grid Study tool Assistant Interface to help solve contigencies for a grid state under study.
"""Tests for FastAPI API endpoints using TestClient with mocked services."""
import json
import pytest
from unittest.mock import patch, MagicMock, PropertyMock
from fastapi.testclient import TestClient
@pytest.fixture
def mock_services():
"""Patch both singleton services before importing the app."""
with patch(
"expert_backend.main.network_service"
) as mock_ns, patch(
"expert_backend.main.recommender_service"
) as mock_rs:
yield mock_ns, mock_rs
@pytest.fixture
def client(mock_services):
"""Create a TestClient with mocked services."""
from expert_backend.main import app
return TestClient(app)
class TestGetBranches:
def test_success(self, client, mock_services):
mock_ns, _ = mock_services
mock_ns.get_disconnectable_elements.return_value = [
"LINE_A",
"LINE_B",
"TRAFO_1",
]
response = client.get("/api/branches")
assert response.status_code == 200
data = response.json()
assert data["branches"] == ["LINE_A", "LINE_B", "TRAFO_1"]
def test_error_returns_400(self, client, mock_services):
mock_ns, _ = mock_services
mock_ns.get_disconnectable_elements.side_effect = ValueError("Network not loaded")
response = client.get("/api/branches")
assert response.status_code == 400
assert "Network not loaded" in response.json()["detail"]
class TestGetVoltageLevels:
def test_success(self, client, mock_services):
mock_ns, _ = mock_services
mock_ns.get_voltage_levels.return_value = ["VL1", "VL2", "VL3"]
response = client.get("/api/voltage-levels")
assert response.status_code == 200
assert response.json()["voltage_levels"] == ["VL1", "VL2", "VL3"]
class TestGetNominalVoltages:
def test_success(self, client, mock_services):
mock_ns, _ = mock_services
mock_ns.get_nominal_voltages.return_value = {"VL1": 400.0, "VL2": 225.0}
response = client.get("/api/nominal-voltages")
assert response.status_code == 200
data = response.json()
assert data["mapping"] == {"VL1": 400.0, "VL2": 225.0}
assert data["unique_kv"] == [225.0, 400.0]
class TestSetRecommenderModel:
"""POST /api/recommender-model — lightweight model swap on the
running RecommenderService (no network reload, no action-dict
rebuild). The frontend fires this whenever the operator changes the
model dropdown (Settings modal OR the selector above Analyze &
Suggest), so the next /api/run-analysis-step2 uses the chosen
model.
"""
def test_success_applies_model_and_echoes_active(self, client, mock_services):
_, mock_rs = mock_services
mock_rs.get_active_model_name.return_value = "random_overflow"
mock_rs.get_compute_overflow_graph.return_value = True
response = client.post(
"/api/recommender-model",
json={"model": "random_overflow", "compute_overflow_graph": True},
)
assert response.status_code == 200
data = response.json()
assert data == {
"status": "success",
"active_model": "random_overflow",
"compute_overflow_graph": True,
}
# The request object is threaded straight into _apply_model_settings.
mock_rs._apply_model_settings.assert_called_once()
applied = mock_rs._apply_model_settings.call_args[0][0]
assert applied.model == "random_overflow"
assert applied.compute_overflow_graph is True
def test_compute_overflow_graph_is_optional(self, client, mock_services):
# The pydantic model allows omitting compute_overflow_graph;
# _apply_model_settings then falls back to its own default.
_, mock_rs = mock_services
mock_rs.get_active_model_name.return_value = "random"
mock_rs.get_compute_overflow_graph.return_value = False
response = client.post("/api/recommender-model", json={"model": "random"})
assert response.status_code == 200
applied = mock_rs._apply_model_settings.call_args[0][0]
assert applied.model == "random"
assert applied.compute_overflow_graph is None
def test_missing_model_returns_422(self, client, mock_services):
# `model` is a required field on RecommenderModelRequest.
response = client.post("/api/recommender-model", json={})
assert response.status_code == 422
def test_error_returns_400(self, client, mock_services):
_, mock_rs = mock_services
mock_rs._apply_model_settings.side_effect = ValueError("bad model")
response = client.post("/api/recommender-model", json={"model": "expert"})
assert response.status_code == 400
assert "bad model" in response.json()["detail"]
class TestGetElementVoltageLevels:
def test_success(self, client, mock_services):
mock_ns, _ = mock_services
mock_ns.get_element_voltage_levels.return_value = ["VL1", "VL2"]
response = client.get("/api/element-voltage-levels?element_id=LINE_A")
assert response.status_code == 200
assert response.json()["voltage_level_ids"] == ["VL1", "VL2"]
def test_missing_param(self, client, mock_services):
response = client.get("/api/element-voltage-levels")
assert response.status_code == 422 # Validation error
class TestGetNetworkDiagram:
def test_success(self, client, mock_services):
_, mock_rs = mock_services
# No NAD prefetch was queued by update_config (or the test doesn't
# exercise the prefetch path) — fall through to fresh compute.
mock_rs.get_prefetched_base_nad.return_value = None
mock_rs.get_network_diagram.return_value = {
"svg": "",
"metadata": '{"nodes":[],"edges":[]}',
}
response = client.get("/api/network-diagram")
assert response.status_code == 200
data = response.json()
assert "svg" in data
assert "metadata" in data
def test_error_returns_400(self, client, mock_services):
_, mock_rs = mock_services
mock_rs.get_prefetched_base_nad.return_value = None
mock_rs.get_network_diagram.side_effect = Exception("No network loaded")
response = client.get("/api/network-diagram")
assert response.status_code == 400
def test_uses_prefetched_nad_when_available(self, client, mock_services):
"""When `prefetch_base_nad_async` has populated the cache during
`/api/config`, `/api/network-diagram` serves the cached NAD and
does NOT re-run the expensive `get_network_diagram()` pypowsybl
code path. This is the core of the perf #2 optimisation — see
docs/performance/history/nad-prefetch.md.
"""
_, mock_rs = mock_services
mock_rs.get_prefetched_base_nad.return_value = {
"svg": "",
"metadata": '{"from":"prefetch"}',
}
response = client.get("/api/network-diagram")
assert response.status_code == 200
data = response.json()
assert data["svg"] == ""
# The fresh compute path MUST NOT have been taken.
mock_rs.get_network_diagram.assert_not_called()
def test_falls_through_to_fresh_compute_on_prefetch_timeout(self, client, mock_services):
"""If `get_prefetched_base_nad` returns None (timeout or never
started), the endpoint falls back to the synchronous path. Keeps
the endpoint usable when `update_config` was never called (e.g.
external callers, process restart)."""
_, mock_rs = mock_services
mock_rs.get_prefetched_base_nad.return_value = None # timed out / not queued
mock_rs.get_network_diagram.return_value = {
"svg": "",
"metadata": None,
}
response = client.get("/api/network-diagram")
assert response.status_code == 200
assert response.json()["svg"] == ""
mock_rs.get_network_diagram.assert_called_once()
def test_prefetched_path_supports_text_format(self, client, mock_services):
"""format=text must work with the prefetched payload too."""
_, mock_rs = mock_services
mock_rs.get_prefetched_base_nad.return_value = {
"svg": "",
"metadata": None,
"lines_overloaded": ["L1"],
"lines_overloaded_rho": [1.05],
}
response = client.get("/api/network-diagram?format=text")
assert response.status_code == 200
assert response.headers["content-type"].startswith("text/plain")
body = response.text
nl = body.index("\n")
assert body[nl + 1:] == ""
mock_rs.get_network_diagram.assert_not_called()
def test_text_format_returns_header_plus_svg(self, client, mock_services):
"""format=text returns a small JSON header on the first line,
then the raw SVG as the rest of the body. The SVG must NOT be
JSON-escaped (savings = no 25 MB JSON.parse on the client)."""
import json as json_module
_, mock_rs = mock_services
mock_rs.get_prefetched_base_nad.return_value = None
mock_rs.get_network_diagram.return_value = {
"svg": "",
"metadata": '{"nodes":[]}',
"lines_overloaded": ["L1"],
"lines_overloaded_rho": [1.05],
}
response = client.get("/api/network-diagram?format=text")
assert response.status_code == 200
assert response.headers["content-type"].startswith("text/plain")
body = response.text
nl = body.index("\n")
header = json_module.loads(body[:nl])
svg = body[nl + 1:]
assert svg == ""
assert "svg" not in header # must be stripped from the JSON header
assert header["lines_overloaded"] == ["L1"]
assert header["lines_overloaded_rho"] == [1.05]
assert header["metadata"] == '{"nodes":[]}'
def test_text_format_gzip(self, client, mock_services):
"""Large text-format responses are gzip-compressed on the wire
when the client signals Accept-Encoding: gzip."""
import gzip as gzip_module
_, mock_rs = mock_services
mock_rs.get_prefetched_base_nad.return_value = None
big_svg = ""
mock_rs.get_network_diagram.return_value = {
"svg": big_svg,
"metadata": None,
}
response = client.get(
"/api/network-diagram?format=text",
headers={"Accept-Encoding": "gzip"},
)
assert response.status_code == 200
# TestClient auto-decodes gzip, so check the header explicitly.
assert response.headers.get("content-encoding") == "gzip"
body = response.text # auto-decoded
nl = body.index("\n")
assert body[nl + 1:] == big_svg
# Sanity: compressed body is much smaller than raw
raw_len = len(big_svg) + 100
# Re-encode to confirm compression ratio is credible
assert len(gzip_module.compress(body.encode("utf-8"))) < raw_len
class TestGetN1Diagram:
def test_success(self, client, mock_services):
_, mock_rs = mock_services
mock_rs.get_contingency_diagram.return_value = {
"svg": "",
"metadata": "{}",
"lf_converged": True,
"lf_status": "CONVERGED",
}
response = client.post(
"/api/contingency-diagram",
json={"disconnected_elements": ["LINE_A"]},
)
assert response.status_code == 200
data = response.json()
assert data["lf_converged"] is True
def test_missing_element(self, client, mock_services):
response = client.post("/api/contingency-diagram", json={})
assert response.status_code == 422
# ------------------------------------------------------------------
# SVG DOM recycling — patch endpoints
# ------------------------------------------------------------------
# /api/contingency-diagram-patch and /api/action-variant-diagram-patch return the
# same delta / overload payload as their full-NAD siblings but SKIP the
# ~2-4 s pypowsybl NAD generation and the ~20-28 MB SVG transfer. The
# frontend patches a clone of the already-loaded N-state SVG DOM.
# See docs/performance/history/svg-dom-recycling.md.
class TestGetN1DiagramPatch:
def test_returns_patchable_payload_without_svg(self, client, mock_services):
_, mock_rs = mock_services
mock_rs.get_contingency_diagram_patch.return_value = {
"patchable": True,
"contingency_id": "LINE_A",
"lf_converged": True,
"lf_status": "CONVERGED",
"disconnected_edges": ["LINE_A"],
"absolute_flows": {
"p1": {"LINE_B": 12.3}, "p2": {"LINE_B": -12.3},
"q1": {"LINE_B": 4.1}, "q2": {"LINE_B": -4.1},
"vl1": {"LINE_B": "VL_1"}, "vl2": {"LINE_B": "VL_2"},
},
"lines_overloaded": ["LINE_C"],
"lines_overloaded_rho": [1.05],
"flow_deltas": {"LINE_B": {"delta": -1.0, "category": "negative", "flip_arrow": False}},
"reactive_flow_deltas": {},
"asset_deltas": {},
"meta": {"base_state": "N", "elapsed_ms": 120},
}
response = client.post(
"/api/contingency-diagram-patch",
json={"disconnected_elements": ["LINE_A"]},
)
assert response.status_code == 200
data = response.json()
assert data["patchable"] is True
assert data["contingency_id"] == "LINE_A"
assert data["disconnected_edges"] == ["LINE_A"]
# SVG-less payload is the whole point of this endpoint.
assert "svg" not in data
assert "metadata" not in data
def test_missing_element_returns_422(self, client, mock_services):
response = client.post("/api/contingency-diagram-patch", json={})
assert response.status_code == 422
def test_service_error_returns_400(self, client, mock_services):
_, mock_rs = mock_services
mock_rs.get_contingency_diagram_patch.side_effect = ValueError("N state unavailable")
response = client.post(
"/api/contingency-diagram-patch",
json={"disconnected_elements": ["LINE_A"]},
)
assert response.status_code == 400
assert "N state unavailable" in response.json()["detail"]
def test_payload_carries_same_non_svg_fields_as_full_endpoint(self, client, mock_services):
"""Frontend contract: every non-SVG field returned by the full
endpoint is also present in the patch payload. Enforces parity
at the HTTP boundary so the frontend branch-logic can swap
transparently."""
_, mock_rs = mock_services
shared_fields = {
"lf_converged": True,
"lf_status": "CONVERGED",
"lines_overloaded": ["X"],
"lines_overloaded_rho": [1.05],
"flow_deltas": {"X": {"delta": 1.0, "category": "positive", "flip_arrow": False}},
"reactive_flow_deltas": {},
"asset_deltas": {},
}
mock_rs.get_contingency_diagram_patch.return_value = {
"patchable": True,
"contingency_id": "LINE_A",
"disconnected_edges": ["LINE_A"],
"absolute_flows": {
"p1": {}, "p2": {}, "q1": {}, "q2": {}, "vl1": {}, "vl2": {}
},
"meta": {"base_state": "N", "elapsed_ms": 1},
**shared_fields,
}
response = client.post(
"/api/contingency-diagram-patch",
json={"disconnected_elements": ["LINE_A"]},
)
data = response.json()
for k, v in shared_fields.items():
assert data[k] == v, f"mismatch on field {k}"
class TestGetActionVariantDiagramPatch:
def test_returns_patchable_payload_for_pst_action(self, client, mock_services):
_, mock_rs = mock_services
mock_rs.get_action_variant_diagram_patch.return_value = {
"patchable": True,
"action_id": "PST_42",
"lf_converged": True,
"lf_status": "CONVERGED",
"non_convergence": None,
"disconnected_edges": [],
"absolute_flows": {
"p1": {"LINE_B": 20.0}, "p2": {"LINE_B": -20.0},
"q1": {"LINE_B": 5.0}, "q2": {"LINE_B": -5.0},
"vl1": {"LINE_B": "VL_1"}, "vl2": {"LINE_B": "VL_2"},
},
"lines_overloaded": [],
"lines_overloaded_rho": [],
"flow_deltas": {},
"reactive_flow_deltas": {},
"asset_deltas": {},
"meta": {"base_state": "N-1", "elapsed_ms": 80},
}
response = client.post(
"/api/action-variant-diagram-patch",
json={"action_id": "PST_42"},
)
assert response.status_code == 200
data = response.json()
assert data["patchable"] is True
assert data["action_id"] == "PST_42"
assert "svg" not in data
assert "metadata" not in data
def test_patchable_for_disco_action(self, client, mock_services):
"""Line-disconnection actions flip switch breakers but don't
change VL bus counts; rendering is a pure dashed-class toggle
on the disconnected line. The backend keeps them patchable
and includes the line in `disconnected_edges`."""
_, mock_rs = mock_services
mock_rs.get_action_variant_diagram_patch.return_value = {
"patchable": True,
"action_id": "disco_LINE_Y",
"lf_converged": True,
"lf_status": "CONVERGED",
"non_convergence": None,
"disconnected_edges": ["CONTINGENCY", "LINE_Y"],
"absolute_flows": {
"p1": {}, "p2": {}, "q1": {}, "q2": {}, "vl1": {}, "vl2": {},
},
"lines_overloaded": [],
"lines_overloaded_rho": [],
"flow_deltas": {},
"reactive_flow_deltas": {},
"asset_deltas": {},
"meta": {"base_state": "N-1", "elapsed_ms": 30},
}
response = client.post(
"/api/action-variant-diagram-patch",
json={"action_id": "disco_LINE_Y"},
)
data = response.json()
assert data["patchable"] is True
assert "LINE_Y" in data["disconnected_edges"]
def test_patchable_for_line_reconnect_action(self, client, mock_services):
"""Single-line reconnections / extra disconnections render as a
pure dashed/solid toggle on one edge element, so the backend
keeps them patchable and lets svgPatch swap the class list."""
_, mock_rs = mock_services
mock_rs.get_action_variant_diagram_patch.return_value = {
"patchable": True,
"action_id": "RECO_LINE_Y",
"lf_converged": True,
"lf_status": "CONVERGED",
"non_convergence": None,
"disconnected_edges": [], # reco_* reconnects the contingency
"absolute_flows": {
"p1": {}, "p2": {}, "q1": {}, "q2": {}, "vl1": {}, "vl2": {},
},
"lines_overloaded": [],
"lines_overloaded_rho": [],
"flow_deltas": {},
"reactive_flow_deltas": {},
"asset_deltas": {},
"meta": {"base_state": "N-1", "elapsed_ms": 42},
}
response = client.post(
"/api/action-variant-diagram-patch",
json={"action_id": "RECO_LINE_Y"},
)
data = response.json()
assert data["patchable"] is True
assert data["disconnected_edges"] == []
def test_patchable_carries_contingency_on_disconnected_edges(self, client, mock_services):
"""Post-action patch must mark the N-1 contingency as dashed
on the action tab — it is still disconnected after the action
unless the action explicitly reconnects it."""
_, mock_rs = mock_services
mock_rs.get_action_variant_diagram_patch.return_value = {
"patchable": True,
"action_id": "LOAD_SHED_X",
"lf_converged": True,
"lf_status": "CONVERGED",
"non_convergence": None,
"disconnected_edges": ["CONTINGENCY_LINE"],
"absolute_flows": {
"p1": {}, "p2": {}, "q1": {}, "q2": {}, "vl1": {}, "vl2": {},
},
"lines_overloaded": [],
"lines_overloaded_rho": [],
"flow_deltas": {},
"reactive_flow_deltas": {},
"asset_deltas": {},
"meta": {"base_state": "N-1", "elapsed_ms": 42},
}
response = client.post(
"/api/action-variant-diagram-patch",
json={"action_id": "LOAD_SHED_X"},
)
data = response.json()
assert data["patchable"] is True
assert data["disconnected_edges"] == ["CONTINGENCY_LINE"]
def test_patchable_with_vl_subtrees_for_node_merging(self, client, mock_services):
"""Node-merging / node-splitting / coupling actions change bus
counts per VL. The backend now patches those rendering
changes via pypowsybl-native focused sub-diagrams instead of
falling back to the full NAD. The frontend splices each
`` subtree into the cloned base."""
_, mock_rs = mock_services
mock_rs.get_action_variant_diagram_patch.return_value = {
"patchable": True,
"action_id": "node_merging_PYMONP3",
"lf_converged": True,
"lf_status": "CONVERGED",
"non_convergence": None,
"disconnected_edges": ["CONTINGENCY_LINE"],
"absolute_flows": {
"p1": {}, "p2": {}, "q1": {}, "q2": {}, "vl1": {}, "vl2": {},
},
"lines_overloaded": [],
"lines_overloaded_rho": [],
"flow_deltas": {},
"reactive_flow_deltas": {},
"asset_deltas": {},
"vl_subtrees": {
"PYMONP3": {"node_svg": ""},
},
"meta": {"base_state": "N-1", "elapsed_ms": 120},
}
response = client.post(
"/api/action-variant-diagram-patch",
json={"action_id": "node_merging_PYMONP3"},
)
assert response.status_code == 200
data = response.json()
assert data["patchable"] is True
assert "PYMONP3" in data["vl_subtrees"]
assert data["vl_subtrees"]["PYMONP3"]["node_svg"].startswith(" service plumbing (e.g. someone forgets to
add the field to the Pydantic model)."""
_, mock_rs = mock_services
def fake_step2(selected_overloads, all_overloads=None, monitor_deselected=False, additional_lines_to_cut=None):
yield {"type": "result", "actions": {}, "action_scores": {}, "lines_overloaded": []}
mock_rs.run_analysis_step2.side_effect = fake_step2
response = client.post(
"/api/run-analysis-step2",
json={
"selected_overloads": ["LINE_1"],
"all_overloads": ["LINE_1"],
"monitor_deselected": False,
"additional_lines_to_cut": ["EXTRA_A", "EXTRA_B"],
},
)
assert response.status_code == 200
mock_rs.run_analysis_step2.assert_called_once_with(
["LINE_1"],
all_overloads=["LINE_1"],
monitor_deselected=False,
additional_lines_to_cut=["EXTRA_A", "EXTRA_B"],
)
def test_error_in_streaming(self, client, mock_services):
_, mock_rs = mock_services
def failing_analysis_step2(*args, **kwargs):
raise RuntimeError("Step 2 exploded")
mock_rs.run_analysis_step2.side_effect = failing_analysis_step2
response = client.post(
"/api/run-analysis-step2",
json={"selected_overloads": ["LINE_1"]},
)
assert response.status_code == 200 # Streaming response
lines = [
line for line in response.text.strip().split("\n") if line.strip()
]
error_event = json.loads(lines[0])
assert error_event["type"] == "error"
assert "Step 2 exploded" in error_event["message"]
class TestActionVariantDiagram:
def test_success(self, client, mock_services):
_, mock_rs = mock_services
mock_rs.get_action_variant_diagram.return_value = {
"svg": "",
"metadata": "{}",
"action_id": "action_1",
}
response = client.post(
"/api/action-variant-diagram",
json={"action_id": "action_1"},
)
assert response.status_code == 200
assert response.json()["action_id"] == "action_1"
def test_with_mode(self, client, mock_services):
_, mock_rs = mock_services
mock_rs.get_action_variant_diagram.return_value = {
"svg": "",
"metadata": "{}",
}
response = client.post(
"/api/action-variant-diagram",
json={"action_id": "action_1", "mode": "delta"},
)
assert response.status_code == 200
mock_rs.get_action_variant_diagram.assert_called_once_with(
"action_1", mode="delta"
)
class TestFocusedDiagram:
def test_with_disconnected_element(self, client, mock_services):
mock_ns, mock_rs = mock_services
mock_ns.get_element_voltage_levels.return_value = ["VL1"]
mock_rs.get_contingency_diagram.return_value = {
"svg": "",
"metadata": "{}",
}
response = client.post(
"/api/focused-diagram",
json={
"element_id": "LINE_A",
"depth": 2,
"disconnected_elements": ["LINE_B"],
},
)
assert response.status_code == 200
data = response.json()
assert data["voltage_level_ids"] == ["VL1"]
assert data["depth"] == 2
def test_without_disconnected_element(self, client, mock_services):
mock_ns, mock_rs = mock_services
mock_ns.get_element_voltage_levels.return_value = ["VL1", "VL2"]
mock_rs.get_network_diagram.return_value = {
"svg": "",
"metadata": "{}",
}
response = client.post(
"/api/focused-diagram",
json={"element_id": "LINE_A"},
)
assert response.status_code == 200
def test_no_voltage_levels_returns_404(self, client, mock_services):
mock_ns, _ = mock_services
mock_ns.get_element_voltage_levels.return_value = []
response = client.post(
"/api/focused-diagram",
json={"element_id": "NONEXISTENT"},
)
assert response.status_code == 404
class TestGetActions:
def test_success(self, client, mock_services):
_, mock_rs = mock_services
mock_rs.get_all_action_ids.return_value = [
{"id": "action_1", "description": "Test action", "type": "line_disconnection"},
]
response = client.get("/api/actions")
assert response.status_code == 200
data = response.json()
assert len(data["actions"]) == 1
assert data["actions"][0]["id"] == "action_1"
class TestSimulateManualAction:
def test_success(self, client, mock_services):
_, mock_rs = mock_services
mock_rs.simulate_manual_action.return_value = {
"action_id": "action_1",
"description_unitaire": "Open line X",
"rho_before": [0.95],
"rho_after": [0.80],
"max_rho": 0.80,
"max_rho_line": "LINE_A",
"is_rho_reduction": True,
"lines_overloaded": ["LINE_A"],
}
response = client.post(
"/api/simulate-manual-action",
json={
"action_id": "action_1",
"disconnected_elements": ["LINE_B"],
},
)
assert response.status_code == 200
data = response.json()
assert data["is_rho_reduction"] is True
assert data["max_rho"] == 0.80
def test_missing_fields(self, client, mock_services):
response = client.post(
"/api/simulate-manual-action",
json={"action_id": "action_1"},
)
assert response.status_code == 422
class TestPydanticModels:
"""Test request model validation."""
def test_config_request_defaults(self, client, mock_services):
"""ConfigRequest should use default values for optional fields."""
mock_ns, mock_rs = mock_services
mock_ns.get_disconnectable_elements.return_value = []
# Mock the import of expert_op4grid_recommender.config
with patch("expert_backend.main.network_service", mock_ns):
# Just test the model itself
from expert_backend.main import ConfigRequest
config = ConfigRequest(
network_path="/test/path",
action_file_path="/test/actions.json",
)
assert config.min_line_reconnections == 2.0
assert config.min_close_coupling == 3.0
assert config.min_open_coupling == 2.0
assert config.min_line_disconnections == 3.0
assert config.n_prioritized_actions == 10
assert config.monitoring_factor == 0.95
assert config.pre_existing_overload_threshold == 0.02
assert config.ignore_reconnections is False
assert config.pypowsybl_fast_mode is True
assert config.lines_monitoring_path is None
# Recommender restriction defaults to "no restriction" (None → []).
assert config.allowed_action_types is None
assert config.min_redispatch == 0
def test_analysis_request_validation(self):
from expert_backend.main import AnalysisRequest
req = AnalysisRequest(disconnected_elements=["LINE_A"])
assert req.disconnected_elements == ["LINE_A"]
def test_focused_diagram_request_defaults(self):
from expert_backend.main import FocusedDiagramRequest
req = FocusedDiagramRequest(element_id="LINE_A")
assert req.depth == 1
assert req.disconnected_elements is None
def test_action_variant_request_defaults(self):
from expert_backend.main import ActionVariantRequest
req = ActionVariantRequest(action_id="act_1")
assert req.mode == "network"
class TestRestoreAnalysisContext:
"""Tests for POST /api/restore-analysis-context endpoint."""
def test_restore_with_all_fields(self, client, mock_services):
_, mock_rs = mock_services
mock_rs.restore_analysis_context.return_value = None
response = client.post(
"/api/restore-analysis-context",
json={
"lines_we_care_about": ["LINE_A", "LINE_B", "LINE_C"],
"disconnected_elements": ["LINE_X"],
"lines_overloaded": ["LINE_A"],
"computed_pairs": {"LINE_A+LINE_B": {"max_rho": 0.5}},
},
)
assert response.status_code == 200
data = response.json()
assert data["status"] == "success"
assert data["lines_we_care_about_count"] == 3
assert data["computed_pairs_count"] == 1
mock_rs.restore_analysis_context.assert_called_once_with(
lines_we_care_about=["LINE_A", "LINE_B", "LINE_C"],
disconnected_elements=["LINE_X"],
lines_overloaded=["LINE_A"],
computed_pairs={"LINE_A+LINE_B": {"max_rho": 0.5}},
)
def test_restore_with_minimal_fields(self, client, mock_services):
_, mock_rs = mock_services
mock_rs.restore_analysis_context.return_value = None
response = client.post(
"/api/restore-analysis-context",
json={
"lines_we_care_about": ["LINE_A"],
},
)
assert response.status_code == 200
data = response.json()
assert data["status"] == "success"
assert data["lines_we_care_about_count"] == 1
assert data["computed_pairs_count"] == 0
def test_restore_with_null_lines(self, client, mock_services):
_, mock_rs = mock_services
mock_rs.restore_analysis_context.return_value = None
response = client.post(
"/api/restore-analysis-context",
json={
"lines_we_care_about": None,
},
)
assert response.status_code == 200
data = response.json()
assert data["lines_we_care_about_count"] == 0
def test_restore_error_returns_400(self, client, mock_services):
_, mock_rs = mock_services
mock_rs.restore_analysis_context.side_effect = Exception("Restore failed")
response = client.post(
"/api/restore-analysis-context",
json={
"lines_we_care_about": ["LINE_A"],
},
)
assert response.status_code == 400
assert "Restore failed" in response.json()["detail"]
class TestRestoreAnalysisContextModel:
"""Tests for the RestoreAnalysisContextRequest Pydantic model."""
def test_defaults(self):
from expert_backend.main import RestoreAnalysisContextRequest
req = RestoreAnalysisContextRequest()
assert req.lines_we_care_about is None
assert req.disconnected_elements is None
assert req.lines_overloaded is None
assert req.computed_pairs is None
def test_full_model(self):
from expert_backend.main import RestoreAnalysisContextRequest
req = RestoreAnalysisContextRequest(
lines_we_care_about=["L1", "L2"],
disconnected_elements=["LINE_X"],
lines_overloaded=["L1"],
computed_pairs={"pair1": {"betas": [0.1, 0.2]}},
)
assert req.lines_we_care_about == ["L1", "L2"]
assert req.disconnected_elements == ["LINE_X"]
assert req.lines_overloaded == ["L1"]
assert req.computed_pairs == {"pair1": {"betas": [0.1, 0.2]}}
class TestRunAnalysisLinesWeCareAbout:
"""Tests that analysis result events include lines_we_care_about."""
def test_run_analysis_includes_lines_we_care_about(self, client, mock_services):
_, mock_rs = mock_services
def fake_analysis(element):
yield {"type": "pdf", "pdf_path": "/tmp/graph.pdf"}
yield {
"type": "result",
"actions": {},
"action_scores": {},
"lines_overloaded": ["LINE_A"],
"lines_we_care_about": ["LINE_A", "LINE_B", "LINE_C"],
"message": "Analysis completed",
"dc_fallback": False,
}
mock_rs.run_analysis.return_value = fake_analysis("LINE_X")
response = client.post(
"/api/run-analysis",
json={"disconnected_elements": ["LINE_X"]},
)
assert response.status_code == 200
lines = [line for line in response.text.strip().split("\n") if line.strip()]
result_event = json.loads(lines[1])
assert result_event["type"] == "result"
assert result_event["lines_we_care_about"] == ["LINE_A", "LINE_B", "LINE_C"]
def test_run_analysis_null_lines_we_care_about(self, client, mock_services):
_, mock_rs = mock_services
def fake_analysis(element):
yield {
"type": "result",
"actions": {},
"action_scores": {},
"lines_overloaded": [],
"lines_we_care_about": None,
"message": "Done",
"dc_fallback": False,
}
mock_rs.run_analysis.return_value = fake_analysis("LINE_X")
response = client.post(
"/api/run-analysis",
json={"disconnected_elements": ["LINE_X"]},
)
lines = [line for line in response.text.strip().split("\n") if line.strip()]
result_event = json.loads(lines[0])
assert result_event["lines_we_care_about"] is None
def test_step2_includes_lines_we_care_about(self, client, mock_services):
_, mock_rs = mock_services
def fake_step2(selected_overloads, all_overloads=None, monitor_deselected=False, additional_lines_to_cut=None):
yield {
"type": "result",
"actions": {},
"action_scores": {},
"lines_overloaded": ["LINE_1"],
"lines_we_care_about": ["LINE_1", "LINE_2"],
"combined_actions": {},
"pre_existing_overloads": [],
"message": "Done",
"dc_fallback": False,
}
mock_rs.run_analysis_step2.side_effect = fake_step2
response = client.post(
"/api/run-analysis-step2",
json={
"selected_overloads": ["LINE_1"],
"all_overloads": ["LINE_1", "LINE_2"],
},
)
lines = [line for line in response.text.strip().split("\n") if line.strip()]
result_event = json.loads(lines[0])
assert result_event["lines_we_care_about"] == ["LINE_1", "LINE_2"]
class TestSimulateManualActionWithContext:
"""Tests that simulate_manual_action passes lines_overloaded to the service."""
def test_passes_lines_overloaded(self, client, mock_services):
_, mock_rs = mock_services
mock_rs.simulate_manual_action.return_value = {
"action_id": "action_1",
"description_unitaire": "Open line X",
"rho_before": [0.95],
"rho_after": [0.70],
"max_rho": 0.70,
"max_rho_line": "LINE_A",
"is_rho_reduction": True,
"lines_overloaded": ["LINE_A"],
}
response = client.post(
"/api/simulate-manual-action",
json={
"action_id": "action_1",
"disconnected_elements": ["LINE_B"],
"lines_overloaded": ["LINE_A", "LINE_C"],
},
)
assert response.status_code == 200
# Verify lines_overloaded was passed through to the service
call_kwargs = mock_rs.simulate_manual_action.call_args
assert call_kwargs[1].get("lines_overloaded") == ["LINE_A", "LINE_C"] or \
(len(call_kwargs[0]) >= 4 and call_kwargs[0][3] == ["LINE_A", "LINE_C"])
class TestSimulateAndVariantDiagramStream:
"""Combined `POST /api/simulate-and-variant-diagram` NDJSON endpoint.
Replaces the legacy two-shot
`simulateManualAction -> getActionVariantDiagram` fallback path with a
single streamed response so the action card's rho numbers can paint
as soon as the grid2op simulation completes without waiting for the
(expensive) post-action NAD regeneration.
These tests lock in:
- the two-event ordering (`metrics` first, then `diagram`);
- the streaming content-type (`application/x-ndjson`);
- the no-gzip invariant on streamed responses — wrapping this endpoint
in `_maybe_gzip_json` or the global `GZipMiddleware` would break
early-event delivery, which was the root cause of the step-2 rollback
(see docs/performance/history/per-endpoint-gzip.md).
"""
_SIM_RESULT = {
"action_id": "act_1",
"description_unitaire": "Open LINE_A",
"rho_before": [0.95, 1.02],
"rho_after": [0.80, 0.70],
"max_rho": 0.80,
"max_rho_line": "LINE_A",
"is_rho_reduction": True,
"non_convergence": None,
"lines_overloaded": ["LINE_A"],
}
_DIAG_RESULT = {
"svg": "",
"metadata": "{}",
"action_id": "act_1",
"lf_converged": True,
"lf_status": "CONVERGED",
"non_convergence": None,
}
def test_streams_metrics_then_diagram(self, client, mock_services):
_, mock_rs = mock_services
mock_rs.simulate_manual_action.return_value = self._SIM_RESULT
mock_rs.get_action_variant_diagram.return_value = self._DIAG_RESULT
response = client.post(
"/api/simulate-and-variant-diagram",
json={
"action_id": "act_1",
"disconnected_elements": ["LINE_B"],
},
)
assert response.status_code == 200
assert response.headers.get("content-type", "").startswith(
"application/x-ndjson"
)
# Must NOT be gzipped — see class docstring.
assert response.headers.get("content-encoding") != "gzip"
lines = [ln for ln in response.text.splitlines() if ln.strip()]
assert len(lines) == 2
metrics = json.loads(lines[0])
diagram = json.loads(lines[1])
assert metrics["type"] == "metrics"
assert metrics["description_unitaire"] == "Open LINE_A"
assert metrics["rho_after"] == [0.80, 0.70]
assert diagram["type"] == "diagram"
assert diagram["svg"] == ""
assert diagram["lf_converged"] is True
def test_ignores_accept_encoding_gzip(self, client, mock_services):
"""Client-sent `Accept-Encoding: gzip` must not cause the stream to be
wrapped — the frontend reads NDJSON with TextDecoder and needs each
event to arrive as soon as the server yields it.
"""
_, mock_rs = mock_services
mock_rs.simulate_manual_action.return_value = self._SIM_RESULT
mock_rs.get_action_variant_diagram.return_value = self._DIAG_RESULT
response = client.post(
"/api/simulate-and-variant-diagram",
json={"action_id": "act_1", "disconnected_elements": ["LINE_B"]},
headers={"Accept-Encoding": "gzip"},
)
assert response.status_code == 200
assert response.headers.get("content-encoding") != "gzip"
def test_error_on_simulate_is_reported_and_closes_stream(self, client, mock_services):
_, mock_rs = mock_services
mock_rs.simulate_manual_action.side_effect = ValueError("no dict loaded")
response = client.post(
"/api/simulate-and-variant-diagram",
json={"action_id": "act_1", "disconnected_elements": ["LINE_B"]},
)
assert response.status_code == 200
lines = [ln for ln in response.text.splitlines() if ln.strip()]
assert len(lines) == 1
event = json.loads(lines[0])
assert event["type"] == "error"
assert "no dict loaded" in event["message"]
mock_rs.get_action_variant_diagram.assert_not_called()
def test_error_on_diagram_after_successful_metrics(self, client, mock_services):
_, mock_rs = mock_services
mock_rs.simulate_manual_action.return_value = self._SIM_RESULT
mock_rs.get_action_variant_diagram.side_effect = ValueError("no observation")
response = client.post(
"/api/simulate-and-variant-diagram",
json={"action_id": "act_1", "disconnected_elements": ["LINE_B"]},
)
assert response.status_code == 200
lines = [ln for ln in response.text.splitlines() if ln.strip()]
# Metrics event still arrives — the UI can at least update the sidebar.
assert len(lines) == 2
assert json.loads(lines[0])["type"] == "metrics"
assert json.loads(lines[1])["type"] == "error"
def test_passes_optional_params_through(self, client, mock_services):
_, mock_rs = mock_services
mock_rs.simulate_manual_action.return_value = self._SIM_RESULT
mock_rs.get_action_variant_diagram.return_value = self._DIAG_RESULT
client.post(
"/api/simulate-and-variant-diagram",
json={
"action_id": "act_1",
"disconnected_elements": ["LINE_B"],
"action_content": {"switches": {"sw1": True}},
"lines_overloaded": ["LINE_A", "LINE_C"],
"target_mw": 42.5,
"target_tap": 3,
"mode": "delta",
},
)
call = mock_rs.simulate_manual_action.call_args
assert call.kwargs["action_content"] == {"switches": {"sw1": True}}
assert call.kwargs["lines_overloaded"] == ["LINE_A", "LINE_C"]
assert call.kwargs["target_mw"] == 42.5
assert call.kwargs["target_tap"] == 3
diag_call = mock_rs.get_action_variant_diagram.call_args
assert diag_call.kwargs["mode"] == "delta"
class TestDiagramGzipCompression:
"""Per-endpoint gzip on the 3 large SVG + 1 actions endpoints.
We deliberately do NOT use Starlette's global `GZipMiddleware` because it
buffers `StreamingResponse` bodies (NDJSON on /api/run-analysis(-step2)),
which was the rollback cause in commits 8c15de7 -> 26bc49d. These tests
lock in the per-endpoint behaviour so that future refactors can't
silently regress back to the global middleware.
"""
# A large-enough SVG body to cross the _GZIP_MIN_BYTES threshold.
_BIG_SVG = ""
def _assert_gzip(self, response):
assert response.status_code == 200
# httpx decodes Content-Encoding transparently, but exposes the header
# value as seen on the wire — so we can still verify it was gzipped.
assert response.headers.get("content-encoding") == "gzip"
assert "accept-encoding" in response.headers.get("vary", "").lower()
# Response body is still valid JSON after transparent decompression.
data = response.json()
assert "svg" in data or "actions" in data
def _assert_no_gzip(self, response):
assert response.status_code == 200
assert response.headers.get("content-encoding") != "gzip"
assert "accept-encoding" in response.headers.get("vary", "").lower()
assert response.json()
def test_network_diagram_gzip_when_accepted(self, client, mock_services):
_, mock_rs = mock_services
mock_rs.get_prefetched_base_nad.return_value = None
mock_rs.get_network_diagram.return_value = {
"svg": self._BIG_SVG,
"metadata": "{}",
}
response = client.get(
"/api/network-diagram",
headers={"Accept-Encoding": "gzip"},
)
self._assert_gzip(response)
def test_network_diagram_no_gzip_when_not_accepted(self, client, mock_services):
_, mock_rs = mock_services
mock_rs.get_prefetched_base_nad.return_value = None
mock_rs.get_network_diagram.return_value = {
"svg": self._BIG_SVG,
"metadata": "{}",
}
response = client.get(
"/api/network-diagram",
headers={"Accept-Encoding": "identity"},
)
self._assert_no_gzip(response)
def test_n1_diagram_gzip_when_accepted(self, client, mock_services):
_, mock_rs = mock_services
mock_rs.get_contingency_diagram.return_value = {
"svg": self._BIG_SVG,
"metadata": "{}",
"lf_converged": True,
}
response = client.post(
"/api/contingency-diagram",
json={"disconnected_elements": ["LINE_A"]},
headers={"Accept-Encoding": "gzip"},
)
self._assert_gzip(response)
assert response.json()["lf_converged"] is True
def test_action_variant_diagram_gzip_when_accepted(self, client, mock_services):
_, mock_rs = mock_services
mock_rs.get_action_variant_diagram.return_value = {
"svg": self._BIG_SVG,
"metadata": "{}",
}
response = client.post(
"/api/action-variant-diagram",
json={"action_id": "action_1"},
headers={"Accept-Encoding": "gzip"},
)
self._assert_gzip(response)
def test_small_payload_below_threshold_is_not_compressed(self, client, mock_services):
_, mock_rs = mock_services
mock_rs.get_prefetched_base_nad.return_value = None
# Tiny SVG — well below _GZIP_MIN_BYTES (10 kB).
mock_rs.get_network_diagram.return_value = {
"svg": "",
"metadata": "{}",
}
response = client.get(
"/api/network-diagram",
headers={"Accept-Encoding": "gzip"},
)
assert response.status_code == 200
assert response.headers.get("content-encoding") != "gzip"
assert response.json()["svg"] == ""
def test_streaming_analysis_is_not_affected(self, client, mock_services):
"""/api/run-analysis-step2 MUST keep streaming NDJSON one event per line
(the overflow PDF event must reach the browser before the result
event). The per-endpoint gzip helper must not be wrapped around it.
"""
_, mock_rs = mock_services
def fake_step2(*args, **kwargs):
yield {"type": "pdf", "pdf_path": "/tmp/graph.pdf"}
yield {"type": "result", "analysis_result": {"actions": []}}
mock_rs.run_analysis_step2.side_effect = fake_step2
response = client.post(
"/api/run-analysis-step2",
json={"selected_overloads": ["LINE_A"]},
headers={"Accept-Encoding": "gzip"},
)
assert response.status_code == 200
# NDJSON streaming response — never gzip-wrapped by the helper.
assert response.headers.get("content-encoding") != "gzip"
assert response.headers.get("content-type", "").startswith("application/x-ndjson")
lines = [ln for ln in response.text.splitlines() if ln.strip()]
assert len(lines) == 2
assert json.loads(lines[0])["type"] == "pdf"
assert json.loads(lines[1])["type"] == "result"
class TestRegenerateOverflowGraph:
"""`POST /api/regenerate-overflow-graph` — cache-backed Hierarchical
/ Geo toggle for the Overflow Analysis tab."""
def test_success_adds_pdf_url(self, client, mock_services):
_, mock_rs = mock_services
mock_rs.regenerate_overflow_graph.return_value = {
"pdf_path": "/tmp/Overflow_Graph/overflow_geo.html",
"mode": "geo",
"cached": False,
}
response = client.post(
"/api/regenerate-overflow-graph",
json={"mode": "geo"},
)
assert response.status_code == 200
data = response.json()
assert data["mode"] == "geo"
assert data["cached"] is False
assert data["pdf_path"] == "/tmp/Overflow_Graph/overflow_geo.html"
# `pdf_url` must be derived from the basename so the static mount
# at /results/pdf/ can serve the file regardless of its on-disk
# parent directory.
assert data["pdf_url"] == "/results/pdf/overflow_geo.html"
mock_rs.regenerate_overflow_graph.assert_called_once_with("geo")
def test_cached_response_passes_through(self, client, mock_services):
_, mock_rs = mock_services
mock_rs.regenerate_overflow_graph.return_value = {
"pdf_path": "/tmp/Overflow_Graph/overflow_hierarchi.html",
"mode": "hierarchical",
"cached": True,
}
response = client.post(
"/api/regenerate-overflow-graph",
json={"mode": "hierarchical"},
)
assert response.status_code == 200
assert response.json()["cached"] is True
def test_no_prior_step2_returns_400(self, client, mock_services):
_, mock_rs = mock_services
mock_rs.regenerate_overflow_graph.side_effect = ValueError(
"No Step-2 context available. Run the analysis first."
)
response = client.post(
"/api/regenerate-overflow-graph",
json={"mode": "geo"},
)
assert response.status_code == 400
assert "Step-2" in response.json()["detail"]
def test_rejects_bad_mode_from_service(self, client, mock_services):
_, mock_rs = mock_services
mock_rs.regenerate_overflow_graph.side_effect = ValueError(
"Unknown overflow layout mode: 'foo'; expected 'hierarchical' or 'geo'."
)
response = client.post(
"/api/regenerate-overflow-graph",
json={"mode": "foo"},
)
assert response.status_code == 400
assert "foo" in response.json()["detail"]
def test_missing_pdf_path_does_not_set_url(self, client, mock_services):
"""When graphviz produced nothing (older recommender install,
for example), pdf_path is None — the route must not synthesize
a bogus pdf_url."""
_, mock_rs = mock_services
mock_rs.regenerate_overflow_graph.return_value = {
"pdf_path": None,
"mode": "geo",
"cached": False,
}
response = client.post(
"/api/regenerate-overflow-graph",
json={"mode": "geo"},
)
assert response.status_code == 200
assert "pdf_url" not in response.json()
class TestResponseModels:
"""The D2 response models on the small control endpoints must serialize
the EXACT field set — a `response_model` silently drops any field the
handler returns that isn't declared, so these lock the shape (and the
OpenAPI snapshot documents it)."""
def test_recommender_model_response_exact_fields(self, client, mock_services):
_, mock_rs = mock_services
mock_rs.get_active_model_name.return_value = "random"
mock_rs.get_compute_overflow_graph.return_value = False
resp = client.post("/api/recommender-model", json={"model": "random"})
assert resp.status_code == 200
assert resp.json() == {
"status": "success",
"active_model": "random",
"compute_overflow_graph": False,
}
def test_restore_analysis_context_response_exact_fields(self, client, mock_services):
resp = client.post(
"/api/restore-analysis-context",
json={
"lines_we_care_about": ["L1", "L2"],
"computed_pairs": {"a+b": {}},
},
)
assert resp.status_code == 200
assert resp.json() == {
"status": "success",
"lines_we_care_about_count": 2,
"computed_pairs_count": 1,
}
def test_save_session_response_exact_fields(self, client, mock_services, tmp_path):
resp = client.post(
"/api/save-session",
json={
"session_name": "costudy4grid_session_x",
"json_content": "{}",
"output_folder_path": str(tmp_path),
},
)
assert resp.status_code == 200
assert set(resp.json().keys()) == {"session_folder", "pdf_copied"}
assert resp.json()["pdf_copied"] is False
class TestSessionPathTraversal:
"""`session_name` is joined onto a caller-supplied folder to build the
session directory, so it must be a single path component — a `..`, a
path separator, or an absolute path (which `os.path.join` honours,
silently dropping the base) would let save/load escape the output
folder (QW7, 2026-07). `_safe_session_dir` rejects those with a 400
before any filesystem write."""
def test_save_session_rejects_parent_traversal(self, client, mock_services, tmp_path):
base = tmp_path / "sessions"
base.mkdir()
resp = client.post(
"/api/save-session",
json={
"session_name": "../escaped",
"json_content": "{}",
"output_folder_path": str(base),
},
)
assert resp.status_code == 400
assert resp.json()["detail"] == "Invalid session name"
# Nothing was written outside the base folder.
assert not (tmp_path / "escaped").exists()
def test_save_session_rejects_path_separator(self, client, mock_services, tmp_path):
resp = client.post(
"/api/save-session",
json={
"session_name": "nested/child",
"json_content": "{}",
"output_folder_path": str(tmp_path),
},
)
assert resp.status_code == 400
assert not (tmp_path / "nested").exists()
def test_save_session_rejects_absolute_name(self, client, mock_services, tmp_path):
evil = tmp_path / "abs_escape"
resp = client.post(
"/api/save-session",
json={
"session_name": str(evil),
"json_content": "{}",
"output_folder_path": str(tmp_path / "base"),
},
)
assert resp.status_code == 400
assert not evil.exists()
def test_load_session_rejects_parent_traversal(self, client, mock_services, tmp_path):
resp = client.post(
"/api/load-session",
json={"folder_path": str(tmp_path), "session_name": "../../etc"},
)
assert resp.status_code == 400
assert resp.json()["detail"] == "Invalid session name"
def test_save_session_allows_plain_name(self, client, mock_services, tmp_path):
resp = client.post(
"/api/save-session",
json={
"session_name": "costudy4grid_session_ok",
"json_content": "{}",
"output_folder_path": str(tmp_path),
},
)
assert resp.status_code == 200
assert (tmp_path / "costudy4grid_session_ok" / "session.json").is_file()
def test_load_session_allows_plain_name(self, client, mock_services, tmp_path):
# The guard must not break a legitimate round-trip.
import json as _json
sess = tmp_path / "costudy4grid_session_ok"
sess.mkdir()
(sess / "session.json").write_text(_json.dumps({"hello": "world"}))
resp = client.post(
"/api/load-session",
json={"folder_path": str(tmp_path), "session_name": "costudy4grid_session_ok"},
)
assert resp.status_code == 200
assert resp.json()["hello"] == "world"
class TestCorsDefault:
"""CORS defaults to the loopback dev-server origins, NOT a wildcard
(QW3, 2026-07): a `*` default lets any web page the operator visits
read `/api/*` (incl. the filesystem RPCs) off the localhost backend,
so the wildcard is now explicit opt-in via `CORS_ALLOWED_ORIGINS`."""
def test_unset_env_resolves_to_loopback_not_wildcard(self):
from expert_backend import main
origins = main._resolve_cors_origins(None)
assert "*" not in origins
assert origins == main._CORS_DEFAULT_ORIGINS
assert "http://localhost:5173" in origins
assert "http://127.0.0.1:5173" in origins
def test_empty_env_resolves_to_loopback(self):
from expert_backend import main
assert main._resolve_cors_origins(" ") == main._CORS_DEFAULT_ORIGINS
def test_explicit_wildcard_is_opt_in(self):
from expert_backend import main
assert main._resolve_cors_origins("*") == ["*"]
def test_comma_separated_list_overrides(self):
from expert_backend import main
assert main._resolve_cors_origins("https://a.example, https://b.example") == [
"https://a.example",
"https://b.example",
]
def test_live_app_default_is_not_wildcard(self):
# The app is configured at import from the (unset in tests) env.
from expert_backend import main
assert main._CORS_ORIGINS != ["*"]
def test_credentials_invariant(self):
# The middleware sets allow_credentials = (origins != ["*"]):
# ON for a concrete allow-list, OFF under the wildcard (browsers
# reject `*` + credentials anyway).
from expert_backend import main
assert (main._resolve_cors_origins("") != ["*"]) is True
assert (main._resolve_cors_origins("*") != ["*"]) is False
def test_disallowed_origin_gets_no_cors_header(self, client):
resp = client.get("/api/user-config", headers={"Origin": "http://evil.example"})
acao = resp.headers.get("access-control-allow-origin")
assert acao != "*"
assert acao != "http://evil.example"
def test_allowed_loopback_origin_is_reflected(self, client):
resp = client.get("/api/user-config", headers={"Origin": "http://localhost:5173"})
assert resp.headers.get("access-control-allow-origin") == "http://localhost:5173"
class TestLockdownProfile:
"""D7: on a locked-down (hosted, anonymous-visitor) deployment the
desktop-era filesystem RPCs are disabled with a 403 `{code:
LOCKED_DOWN}`, while the read-only app config stays available so the
SPA still boots. Toggled by `COSTUDY4GRID_LOCKDOWN` (read into
`main._LOCKDOWN` at import; patched here)."""
def test_filesystem_rpcs_are_403_when_locked_down(self, client, mock_services, monkeypatch, tmp_path):
from expert_backend import main
monkeypatch.setattr(main, "_LOCKDOWN", True)
cases = [
("post", "/api/config-file-path", {"json": {"path": "/etc/evil.json"}}),
("get", "/api/pick-path", {"params": {"type": "file"}}),
("post", "/api/save-session", {"json": {
"session_name": "s", "json_content": "{}", "output_folder_path": str(tmp_path),
}}),
("get", "/api/list-sessions", {"params": {"folder_path": str(tmp_path)}}),
("post", "/api/load-session", {"json": {"folder_path": str(tmp_path), "session_name": "s"}}),
]
for method, path, kwargs in cases:
resp = getattr(client, method)(path, **kwargs)
assert resp.status_code == 403, f"{path} should be 403 when locked down"
assert resp.json()["code"] == "LOCKED_DOWN", f"{path} should carry the LOCKED_DOWN code"
# Nothing was written under the locked-down deployment.
assert not (tmp_path / "s").exists()
def test_read_only_config_still_available_when_locked_down(self, client, mock_services, monkeypatch):
from expert_backend import main
monkeypatch.setattr(main, "_LOCKDOWN", True)
# The SPA boot reads still work.
assert client.get("/api/user-config").status_code == 200
assert client.get("/api/config-file-path").status_code == 200
def test_filesystem_rpcs_work_when_not_locked_down(self, client, mock_services, monkeypatch, tmp_path):
from expert_backend import main
monkeypatch.setattr(main, "_LOCKDOWN", False)
resp = client.post("/api/save-session", json={
"session_name": "costudy4grid_session_ok", "json_content": "{}",
"output_folder_path": str(tmp_path),
})
assert resp.status_code == 200
assert (tmp_path / "costudy4grid_session_ok" / "session.json").is_file()
class TestEventLoopSafety:
"""`/api/run-analysis-step1` runs seconds of synchronous pypowsybl /
grid2op work, so it MUST be a sync `def` route (dispatched to
Starlette's threadpool) — an `async def` would run that blocking work
on the event loop and freeze every other request for its duration
(QW2, 2026-07). The streaming analysis routes stay `async def`: they
return a StreamingResponse immediately and their sync generators are
iterated in the threadpool."""
def test_run_analysis_step1_is_sync_def(self):
import inspect
from expert_backend import main
assert not inspect.iscoroutinefunction(main.run_analysis_step1), (
"run_analysis_step1 must be a sync `def` route so its blocking "
"pypowsybl work runs in the threadpool, not on the event loop."
)
class TestStudyMutationBusyGate:
"""The study-mutation gate (D3, 2026-07): config load + the analysis
pipeline mutate shared singleton state, so at most one runs at a
time; a second concurrent one is rejected with HTTP 409 rather than
queued behind seconds of work."""
def test_config_returns_409_when_busy(self, client, mock_services):
_, mock_rs = mock_services
mock_rs.try_begin_study_mutation.return_value = False
response = client.post(
"/api/config",
json={"network_path": "/n", "action_file_path": "/a"},
)
assert response.status_code == 409
assert "in progress" in response.json()["detail"]
# A rejected request must NOT release a gate it never acquired.
mock_rs.end_study_mutation.assert_not_called()
def test_step1_returns_409_when_busy(self, client, mock_services):
_, mock_rs = mock_services
mock_rs.try_begin_study_mutation.return_value = False
response = client.post(
"/api/run-analysis-step1",
json={"disconnected_elements": ["LINE_A"]},
)
assert response.status_code == 409
mock_rs.run_analysis_step1.assert_not_called()
mock_rs.end_study_mutation.assert_not_called()
def test_step1_releases_gate_on_success(self, client, mock_services):
_, mock_rs = mock_services
mock_rs.try_begin_study_mutation.return_value = True
mock_rs.run_analysis_step1.return_value = {
"lines_overloaded": [], "can_proceed": False, "step1_time": 0.1,
}
response = client.post(
"/api/run-analysis-step1",
json={"disconnected_elements": ["LINE_A"]},
)
assert response.status_code == 200
mock_rs.end_study_mutation.assert_called_once()
def test_step1_releases_gate_on_error(self, client, mock_services):
_, mock_rs = mock_services
mock_rs.try_begin_study_mutation.return_value = True
mock_rs.run_analysis_step1.side_effect = ValueError("boom")
response = client.post(
"/api/run-analysis-step1",
json={"disconnected_elements": ["LINE_A"]},
)
assert response.status_code == 400
# Gate released even when the analysis raised.
mock_rs.end_study_mutation.assert_called_once()
def test_step2_returns_409_when_busy(self, client, mock_services):
_, mock_rs = mock_services
mock_rs.try_begin_study_mutation.return_value = False
response = client.post(
"/api/run-analysis-step2",
json={"selected_overloads": ["LINE_A"]},
)
assert response.status_code == 409
mock_rs.run_analysis_step2.assert_not_called()
def test_step2_releases_gate_after_stream(self, client, mock_services):
_, mock_rs = mock_services
mock_rs.try_begin_study_mutation.return_value = True
mock_rs.run_analysis_step2.return_value = iter([
{"type": "pdf", "pdf_path": None},
{"type": "result", "actions": {}},
])
with client.stream(
"POST", "/api/run-analysis-step2",
json={"selected_overloads": ["LINE_A"]},
) as response:
assert response.status_code == 200
_ = list(response.iter_lines())
# Gate released in the generator's finally once the stream drains.
mock_rs.end_study_mutation.assert_called_once()
class TestPickPath:
"""Native file/folder picker — must succeed on macOS via osascript
(no tkinter dependency) and on other platforms via the existing
tkinter subprocess. The endpoint always returns 200 with a body
of {"path": ...} or {"path": "", "error": ...}; it never relies
on transport-level errors to surface a failure to the frontend."""
def test_macos_file_picker_uses_osascript(self, client):
completed = MagicMock(returncode=0, stdout="/Users/me/grid.xiidm\n", stderr="")
with patch("expert_backend.main.platform.system", return_value="Darwin"), \
patch("expert_backend.main.subprocess.run", return_value=completed) as run:
response = client.get("/api/pick-path?type=file")
assert response.status_code == 200
assert response.json() == {"path": "/Users/me/grid.xiidm"}
# The dispatcher must invoke osascript with the file-picker prompt.
argv = run.call_args.args[0]
assert argv[0] == "osascript"
assert any("choose file" in a for a in argv)
def test_macos_dir_picker_uses_osascript(self, client):
completed = MagicMock(returncode=0, stdout="/Users/me/output\n", stderr="")
with patch("expert_backend.main.platform.system", return_value="Darwin"), \
patch("expert_backend.main.subprocess.run", return_value=completed) as run:
response = client.get("/api/pick-path?type=dir")
assert response.status_code == 200
assert response.json() == {"path": "/Users/me/output"}
argv = run.call_args.args[0]
assert argv[0] == "osascript"
assert any("choose folder" in a for a in argv)
def test_macos_user_cancel_is_silent(self, client):
# AppleScript exits with code 1 + stderr "User canceled. (-128)"
# when the user dismisses the dialog. The frontend must NOT see
# an `error` field — it would pop a noisy alert for a benign
# cancel gesture.
cancelled = MagicMock(
returncode=1,
stdout="",
stderr="0:55: execution error: User canceled. (-128)",
)
with patch("expert_backend.main.platform.system", return_value="Darwin"), \
patch("expert_backend.main.subprocess.run", return_value=cancelled):
response = client.get("/api/pick-path?type=file")
assert response.status_code == 200
assert response.json() == {"path": ""}
def test_macos_real_failure_surfaces_error(self, client):
failed = MagicMock(
returncode=1, stdout="", stderr="syntax error: unexpected end of file",
)
with patch("expert_backend.main.platform.system", return_value="Darwin"), \
patch("expert_backend.main.subprocess.run", return_value=failed):
response = client.get("/api/pick-path?type=file")
assert response.status_code == 200
body = response.json()
assert body["path"] == ""
assert "syntax error" in body["error"]
def test_macos_osascript_missing_falls_back_to_error(self, client):
# If osascript is absent (extremely rare on macOS), surface a
# helpful error instead of crashing the route.
with patch("expert_backend.main.platform.system", return_value="Darwin"), \
patch("expert_backend.main.subprocess.run", side_effect=FileNotFoundError("osascript")):
response = client.get("/api/pick-path?type=file")
assert response.status_code == 200
body = response.json()
assert body["path"] == ""
assert "osascript" in body["error"]
def test_non_macos_uses_tkinter_subprocess(self, client):
completed = MagicMock(returncode=0, stdout="/home/user/grid.xiidm\n", stderr="")
with patch("expert_backend.main.platform.system", return_value="Linux"), \
patch("expert_backend.main.subprocess.run", return_value=completed) as run:
response = client.get("/api/pick-path?type=file")
assert response.status_code == 200
assert response.json() == {"path": "/home/user/grid.xiidm"}
# tkinter path runs the python interpreter, NOT osascript.
argv = run.call_args.args[0]
assert argv[0] != "osascript"
# The inlined script must contain the tkinter import.
assert any("import tkinter" in a for a in argv)
class TestErrorLeakGuard:
"""QW6 — the FS/config endpoints must not leak filesystem paths (or raw
exception text) into client-visible error `detail`. D2 genericises only
UNCAUGHT exceptions; these endpoints raise explicit HTTPException(400)s, so
they need their own generic messages + server-side logger.exception."""
def test_save_user_config_failure_is_generic(self, client, monkeypatch):
def boom(_cfg):
raise OSError("/secret/abs/path/config.json: permission denied")
monkeypatch.setattr("expert_backend.main._save_user_config", boom)
r = client.post("/api/user-config", json={"foo": "bar"})
assert r.status_code == 400
detail = r.json()["detail"]
assert detail == "Failed to save configuration."
assert "/secret" not in detail
def test_save_session_failure_is_generic(self, client, monkeypatch):
# os.makedirs failing must not echo the absolute session path.
def boom(*_a, **_k):
raise OSError("/private/sessions/leak: no space left")
monkeypatch.setattr("expert_backend.main.os.makedirs", boom)
r = client.post("/api/save-session", json={
"output_folder_path": "out", "session_name": "s1", "json_content": "{}",
})
assert r.status_code == 400
detail = r.json()["detail"]
assert detail == "Cannot create the session directory."
assert "/private" not in detail