feat: Add Processing Opex functionality

- Introduced OpexValidationError for handling validation errors in processing opex calculations.
- Implemented ProjectProcessingOpexRepository and ScenarioProcessingOpexRepository for managing project and scenario-level processing opex snapshots.
- Enhanced UnitOfWork to include repositories for processing opex.
- Updated sidebar navigation and scenario detail templates to include links to the new Processing Opex Planner.
- Created a new template for the Processing Opex Planner with form handling for input components and parameters.
- Developed integration tests for processing opex calculations, covering HTML and JSON flows, including validation for currency mismatches and unsupported frequencies.
- Added unit tests for the calculation logic, ensuring correct handling of various scenarios and edge cases.
This commit is contained in:
2025-11-13 09:26:57 +01:00
parent 1240b08740
commit 1feae7ff85
16 changed files with 1931 additions and 11 deletions

View File

@@ -0,0 +1,159 @@
import pytest
from schemas.calculations import (
ProcessingOpexCalculationRequest,
ProcessingOpexComponentInput,
ProcessingOpexOptions,
ProcessingOpexParameters,
)
from services.calculations import calculate_processing_opex
from services.exceptions import OpexValidationError
def _component(**overrides) -> ProcessingOpexComponentInput:
defaults = {
"id": None,
"name": "Component",
"category": "energy",
"unit_cost": 1000.0,
"quantity": 1.0,
"frequency": "monthly",
"currency": "USD",
"period_start": 1,
"period_end": 1,
"notes": None,
}
defaults.update(overrides)
return ProcessingOpexComponentInput(**defaults)
def test_calculate_processing_opex_success():
request = ProcessingOpexCalculationRequest(
components=[
_component(
name="Power",
category="energy",
unit_cost=1000.0,
quantity=1,
frequency="monthly",
period_start=1,
period_end=3,
),
_component(
name="Maintenance",
category="maintenance",
unit_cost=2500.0,
quantity=1,
frequency="quarterly",
period_start=1,
period_end=2,
),
],
parameters=ProcessingOpexParameters(
currency_code="USD",
escalation_pct=5,
discount_rate_pct=None,
evaluation_horizon_years=2,
apply_escalation=True,
),
options=ProcessingOpexOptions(persist=True, snapshot_notes=None),
)
result = calculate_processing_opex(request)
assert result.currency == "USD"
assert result.options.persist is True
assert result.totals.overall_annual == pytest.approx(22_000.0)
assert result.totals.escalated_total == pytest.approx(58_330.0, rel=1e-4)
assert result.totals.escalation_pct == pytest.approx(5.0)
categories = {entry.category: entry for entry in result.totals.by_category}
assert categories["energy"].annual_cost == pytest.approx(12_000.0)
assert categories["maintenance"].annual_cost == pytest.approx(10_000.0)
assert len(result.timeline) == 3
timeline = {entry.period: entry for entry in result.timeline}
assert timeline[1].base_cost == pytest.approx(22_000.0)
assert timeline[2].base_cost == pytest.approx(22_000.0)
assert timeline[3].base_cost == pytest.approx(12_000.0)
assert timeline[1].escalated_cost == pytest.approx(22_000.0)
assert timeline[2].escalated_cost == pytest.approx(23_100.0, rel=1e-4)
assert timeline[3].escalated_cost == pytest.approx(13_230.0, rel=1e-4)
assert result.metrics.annual_average == pytest.approx(
19_443.3333, rel=1e-4)
assert len(result.components) == 2
assert result.components[0].frequency == "monthly"
assert result.components[1].frequency == "quarterly"
def test_calculate_processing_opex_currency_mismatch():
request = ProcessingOpexCalculationRequest(
components=[_component(currency="USD")],
parameters=ProcessingOpexParameters(
currency_code="CAD",
escalation_pct=None,
discount_rate_pct=None,
evaluation_horizon_years=10,
),
)
with pytest.raises(OpexValidationError) as exc:
calculate_processing_opex(request)
assert "Component currency does not match" in exc.value.message
assert exc.value.field_errors and "components[0].currency" in exc.value.field_errors[0]
def test_calculate_processing_opex_unsupported_frequency():
request = ProcessingOpexCalculationRequest(
components=[_component(frequency="biweekly")],
parameters=ProcessingOpexParameters(
currency_code="USD",
escalation_pct=None,
discount_rate_pct=None,
evaluation_horizon_years=2,
),
)
with pytest.raises(OpexValidationError) as exc:
calculate_processing_opex(request)
assert "Unsupported frequency" in exc.value.message
assert exc.value.field_errors and "components[0].frequency" in exc.value.field_errors[0]
def test_calculate_processing_opex_requires_components():
request = ProcessingOpexCalculationRequest(components=[])
with pytest.raises(OpexValidationError) as exc:
calculate_processing_opex(request)
assert "At least one processing opex component" in exc.value.message
assert exc.value.field_errors and "components" in exc.value.field_errors[0]
def test_calculate_processing_opex_extends_evaluation_horizon():
request = ProcessingOpexCalculationRequest(
components=[
_component(period_start=1, period_end=4),
],
parameters=ProcessingOpexParameters(
currency_code="USD",
discount_rate_pct=0,
escalation_pct=0,
evaluation_horizon_years=2,
apply_escalation=False,
),
)
result = calculate_processing_opex(request)
assert len(result.timeline) == 4
assert result.timeline[-1].period == 4
assert all(entry.escalated_cost is None for entry in result.timeline)
assert result.timeline[-1].base_cost == pytest.approx(12_000.0)
assert result.metrics.annual_average == pytest.approx(
12_000.0, rel=1e-4)