Mannu
e6dc39aa33
- Engine: Add ppa_revenue_cr, mcp_revenue_cr, tariff, units to PnLRow - Engine: Split PPA vs MCP revenue in P&L computation - Web: Collapsible rows for PPA/MCP Revenue and Opex - Web: Highlighted rows (Total Revenue, EBITDA, EBIT, PBT, PAT) - Web: Units above Tariff in breakdown, bg-blue-50 highlight - Fix sticky column z-index for horizontal scroll - CLAUDE.md: Add project documentation Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
77 lines
2.2 KiB
Python
77 lines
2.2 KiB
Python
"""S8-T01: Sweep engine tests."""
|
|
|
|
import pytest
|
|
|
|
from remodel_engine.scenarios.sweep import (
|
|
SweepParam,
|
|
TornadoEntry,
|
|
run_sweep,
|
|
run_tornado,
|
|
)
|
|
from remodel_engine.schemas.scenario import ScenarioInput
|
|
|
|
|
|
def _base_inputs() -> ScenarioInput:
|
|
return ScenarioInput()
|
|
|
|
|
|
def test_empty_sweep_returns_base() -> None:
|
|
inp = _base_inputs()
|
|
results = run_sweep(inp, [])
|
|
assert len(results) == 1
|
|
assert results[0].status == "success"
|
|
|
|
|
|
def test_single_axis_sweep() -> None:
|
|
inp = _base_inputs()
|
|
param = SweepParam("commercial.tariff_inr_per_kwh", [3.5, 4.0, 4.5])
|
|
results = run_sweep(inp, [param], max_workers=1)
|
|
assert len(results) == 3
|
|
tariffs = [r.param_values["commercial.tariff_inr_per_kwh"] for r in results]
|
|
assert sorted(tariffs) == [3.5, 4.0, 4.5]
|
|
|
|
|
|
def test_cartesian_sweep_two_axes() -> None:
|
|
inp = _base_inputs()
|
|
params = [
|
|
SweepParam("commercial.tariff_inr_per_kwh", [3.5, 4.5]),
|
|
SweepParam("debt.interest_rate_annual", [0.09, 0.11]),
|
|
]
|
|
results = run_sweep(inp, params, max_workers=2)
|
|
assert len(results) == 4 # 2 * 2
|
|
assert all(r.status == "success" for r in results)
|
|
|
|
|
|
def test_sweep_result_has_kpis() -> None:
|
|
inp = _base_inputs()
|
|
param = SweepParam("commercial.tariff_inr_per_kwh", [4.0])
|
|
results = run_sweep(inp, [param], max_workers=1)
|
|
assert results[0].kpis is not None
|
|
|
|
|
|
def test_sweep_invalid_path_marks_failed() -> None:
|
|
inp = _base_inputs()
|
|
param = SweepParam("nonexistent.field", [1.0])
|
|
results = run_sweep(inp, [param], max_workers=1)
|
|
assert results[0].status == "failed"
|
|
assert results[0].error is not None
|
|
|
|
|
|
def test_tornado_returns_sorted_entries() -> None:
|
|
inp = _base_inputs()
|
|
entries = run_tornado(inp, kpi_key="lcoe_inr_per_kwh", max_workers=2)
|
|
assert len(entries) > 0
|
|
swings = [e.swing for e in entries]
|
|
assert swings == sorted(swings, reverse=True)
|
|
|
|
|
|
def test_tornado_entry_swing_correct() -> None:
|
|
entry = TornadoEntry(
|
|
param_name="Test",
|
|
low_value=0.09,
|
|
high_value=0.12,
|
|
base_kpi=0.18,
|
|
low_kpi=0.20,
|
|
high_kpi=0.15,
|
|
)
|
|
assert entry.swing == pytest.approx(0.05, abs=1e-6)
|