zwitschi
1feae7ff85
- 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.
311 lines
11 KiB
Python
311 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Callable
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from services.unit_of_work import UnitOfWork
|
|
|
|
|
|
def _create_project(client: TestClient, name: str) -> int:
|
|
response = client.post(
|
|
"/projects",
|
|
json={
|
|
"name": name,
|
|
"location": "Nevada",
|
|
"operation_type": "open_pit",
|
|
"description": "Project for processing opex testing",
|
|
},
|
|
)
|
|
assert response.status_code == 201
|
|
return response.json()["id"]
|
|
|
|
|
|
def _create_scenario(client: TestClient, project_id: int, name: str) -> int:
|
|
response = client.post(
|
|
f"/projects/{project_id}/scenarios",
|
|
json={
|
|
"name": name,
|
|
"description": "Processing opex scenario",
|
|
"status": "draft",
|
|
"currency": "usd",
|
|
"primary_resource": "diesel",
|
|
},
|
|
)
|
|
assert response.status_code == 201
|
|
return response.json()["id"]
|
|
|
|
|
|
def test_processing_opex_calculation_html_flow(
|
|
client: TestClient,
|
|
unit_of_work_factory: Callable[[], UnitOfWork],
|
|
) -> None:
|
|
project_id = _create_project(client, "Opex HTML Project")
|
|
scenario_id = _create_scenario(client, project_id, "Opex HTML Scenario")
|
|
|
|
form_page = client.get(
|
|
f"/calculations/processing-opex?project_id={project_id}&scenario_id={scenario_id}"
|
|
)
|
|
assert form_page.status_code == 200
|
|
assert "Processing Opex Planner" in form_page.text
|
|
|
|
response = client.post(
|
|
f"/calculations/processing-opex?project_id={project_id}&scenario_id={scenario_id}",
|
|
data={
|
|
"components[0][name]": "Power",
|
|
"components[0][category]": "energy",
|
|
"components[0][unit_cost]": "1000",
|
|
"components[0][quantity]": "1",
|
|
"components[0][frequency]": "monthly",
|
|
"components[0][currency]": "USD",
|
|
"components[0][period_start]": "1",
|
|
"components[0][period_end]": "3",
|
|
"components[1][name]": "Maintenance",
|
|
"components[1][category]": "maintenance",
|
|
"components[1][unit_cost]": "2500",
|
|
"components[1][quantity]": "1",
|
|
"components[1][frequency]": "quarterly",
|
|
"components[1][currency]": "USD",
|
|
"components[1][period_start]": "1",
|
|
"components[1][period_end]": "2",
|
|
"parameters[currency_code]": "USD",
|
|
"parameters[escalation_pct]": "5",
|
|
"parameters[discount_rate_pct]": "3",
|
|
"parameters[evaluation_horizon_years]": "3",
|
|
"parameters[apply_escalation]": "1",
|
|
"options[persist]": "1",
|
|
"options[snapshot_notes]": "Processing opex HTML flow",
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
assert "Processing opex calculation completed successfully." in response.text
|
|
assert "Opex Summary" in response.text
|
|
assert "$22,000.00" in response.text or "22,000" in response.text
|
|
|
|
with unit_of_work_factory() as uow:
|
|
assert uow.project_processing_opex is not None
|
|
assert uow.scenario_processing_opex is not None
|
|
|
|
project_snapshots = uow.project_processing_opex.list_for_project(
|
|
project_id)
|
|
scenario_snapshots = uow.scenario_processing_opex.list_for_scenario(
|
|
scenario_id)
|
|
|
|
assert len(project_snapshots) == 1
|
|
assert len(scenario_snapshots) == 1
|
|
|
|
project_snapshot = project_snapshots[0]
|
|
scenario_snapshot = scenario_snapshots[0]
|
|
|
|
assert project_snapshot.overall_annual is not None
|
|
assert float(
|
|
project_snapshot.overall_annual) == pytest.approx(22_000.0)
|
|
assert project_snapshot.escalated_total is not None
|
|
assert float(
|
|
project_snapshot.escalated_total) == pytest.approx(58_330.0)
|
|
assert project_snapshot.apply_escalation is True
|
|
assert project_snapshot.component_count == 2
|
|
assert project_snapshot.currency_code == "USD"
|
|
|
|
assert scenario_snapshot.overall_annual is not None
|
|
assert float(
|
|
scenario_snapshot.overall_annual) == pytest.approx(22_000.0)
|
|
assert scenario_snapshot.escalated_total is not None
|
|
assert float(
|
|
scenario_snapshot.escalated_total) == pytest.approx(58_330.0)
|
|
assert scenario_snapshot.apply_escalation is True
|
|
assert scenario_snapshot.component_count == 2
|
|
assert scenario_snapshot.currency_code == "USD"
|
|
|
|
|
|
def test_processing_opex_calculation_json_flow(
|
|
client: TestClient,
|
|
unit_of_work_factory: Callable[[], UnitOfWork],
|
|
) -> None:
|
|
project_id = _create_project(client, "Opex JSON Project")
|
|
scenario_id = _create_scenario(client, project_id, "Opex JSON Scenario")
|
|
|
|
payload = {
|
|
"components": [
|
|
{
|
|
"name": "Reagents",
|
|
"category": "materials",
|
|
"unit_cost": 400,
|
|
"quantity": 10,
|
|
"frequency": "monthly",
|
|
"currency": "USD",
|
|
"period_start": 1,
|
|
"period_end": 3,
|
|
},
|
|
{
|
|
"name": "Labor",
|
|
"category": "labor",
|
|
"unit_cost": 1500,
|
|
"quantity": 4,
|
|
"frequency": "weekly",
|
|
"currency": "USD",
|
|
"period_start": 1,
|
|
"period_end": 3,
|
|
},
|
|
{
|
|
"name": "Maintenance",
|
|
"category": "maintenance",
|
|
"unit_cost": 12000,
|
|
"quantity": 1,
|
|
"frequency": "annually",
|
|
"currency": "USD",
|
|
"period_start": 1,
|
|
"period_end": 3,
|
|
},
|
|
],
|
|
"parameters": {
|
|
"currency_code": "USD",
|
|
"escalation_pct": 4,
|
|
"discount_rate_pct": 2,
|
|
"evaluation_horizon_years": 3,
|
|
"apply_escalation": True,
|
|
},
|
|
"options": {"persist": True, "snapshot_notes": "Processing opex JSON flow"},
|
|
}
|
|
|
|
response = client.post(
|
|
f"/calculations/processing-opex?project_id={project_id}&scenario_id={scenario_id}",
|
|
json=payload,
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
data = response.json()
|
|
assert data["currency"] == "USD"
|
|
|
|
expected_overall = 372_000.0
|
|
escalation_factor = 1 + (payload["parameters"]["escalation_pct"] / 100.0)
|
|
expected_timeline = [
|
|
expected_overall * (escalation_factor ** i) for i in range(payload["parameters"]["evaluation_horizon_years"])
|
|
]
|
|
expected_escalated_total = sum(expected_timeline)
|
|
expected_average = expected_escalated_total / len(expected_timeline)
|
|
|
|
assert data["totals"]["overall_annual"] == pytest.approx(expected_overall)
|
|
assert data["totals"]["escalated_total"] == pytest.approx(
|
|
expected_escalated_total)
|
|
assert data["totals"]["escalation_pct"] == pytest.approx(4.0)
|
|
|
|
by_category = {entry["category"] : entry for entry in data["totals"]["by_category"]}
|
|
assert by_category["materials"]["annual_cost"] == pytest.approx(48_000.0)
|
|
assert by_category["labor"]["annual_cost"] == pytest.approx(312_000.0)
|
|
assert by_category["maintenance"]["annual_cost"] == pytest.approx(12_000.0)
|
|
|
|
assert len(data["timeline"]) == 3
|
|
for index, entry in enumerate(data["timeline"], start=0):
|
|
assert entry["period"] == index + 1
|
|
assert entry["base_cost"] == pytest.approx(expected_overall)
|
|
assert entry["escalated_cost"] == pytest.approx(
|
|
expected_timeline[index])
|
|
|
|
assert data["metrics"]["annual_average"] == pytest.approx(expected_average)
|
|
|
|
with unit_of_work_factory() as uow:
|
|
assert uow.project_processing_opex is not None
|
|
assert uow.scenario_processing_opex is not None
|
|
|
|
project_snapshot = uow.project_processing_opex.latest_for_project(
|
|
project_id)
|
|
scenario_snapshot = uow.scenario_processing_opex.latest_for_scenario(
|
|
scenario_id)
|
|
|
|
assert project_snapshot is not None
|
|
assert scenario_snapshot is not None
|
|
|
|
assert project_snapshot.overall_annual is not None
|
|
assert float(project_snapshot.overall_annual) == pytest.approx(
|
|
expected_overall)
|
|
assert project_snapshot.escalated_total is not None
|
|
assert float(project_snapshot.escalated_total) == pytest.approx(
|
|
expected_escalated_total)
|
|
assert project_snapshot.apply_escalation is True
|
|
|
|
assert scenario_snapshot.annual_average is not None
|
|
assert float(scenario_snapshot.annual_average) == pytest.approx(
|
|
expected_average)
|
|
assert scenario_snapshot.apply_escalation is True
|
|
|
|
|
|
@pytest.mark.parametrize("content_type", ["form", "json"])
|
|
def test_processing_opex_calculation_currency_mismatch(
|
|
client: TestClient,
|
|
unit_of_work_factory: Callable[[], UnitOfWork],
|
|
content_type: str,
|
|
) -> None:
|
|
project_id = _create_project(
|
|
client, f"Opex {content_type.title()} Error Project")
|
|
scenario_id = _create_scenario(
|
|
client, project_id, f"Opex {content_type.title()} Error Scenario")
|
|
|
|
if content_type == "json":
|
|
payload = {
|
|
"components": [
|
|
{
|
|
"name": "Power",
|
|
"category": "energy",
|
|
"unit_cost": 500,
|
|
"quantity": 1,
|
|
"frequency": "monthly",
|
|
"currency": "USD",
|
|
"period_start": 1,
|
|
"period_end": 2,
|
|
}
|
|
],
|
|
"parameters": {"currency_code": "CAD"},
|
|
"options": {"persist": True},
|
|
}
|
|
response = client.post(
|
|
f"/calculations/processing-opex?project_id={project_id}&scenario_id={scenario_id}",
|
|
json=payload,
|
|
)
|
|
assert response.status_code == 422
|
|
body = response.json()
|
|
assert "Component currency does not match" in body.get("message", "")
|
|
assert any(
|
|
"components[0].currency" in entry for entry in body.get("errors", []))
|
|
else:
|
|
response = client.post(
|
|
f"/calculations/processing-opex?project_id={project_id}&scenario_id={scenario_id}",
|
|
data={
|
|
"components[0][name]": "Power",
|
|
"components[0][category]": "energy",
|
|
"components[0][unit_cost]": "500",
|
|
"components[0][quantity]": "1",
|
|
"components[0][frequency]": "monthly",
|
|
"components[0][currency]": "USD",
|
|
"components[0][period_start]": "1",
|
|
"components[0][period_end]": "2",
|
|
"parameters[currency_code]": "CAD",
|
|
"options[persist]": "1",
|
|
},
|
|
)
|
|
assert response.status_code == 422
|
|
assert hasattr(response, "context")
|
|
context = getattr(response, "context", {}) or {}
|
|
combined_errors = [
|
|
str(entry)
|
|
for entry in (
|
|
(context.get("errors") or [])
|
|
+ (context.get("component_errors") or [])
|
|
)
|
|
]
|
|
assert any(
|
|
"components[0].currency" in entry for entry in combined_errors)
|
|
|
|
with unit_of_work_factory() as uow:
|
|
assert uow.project_processing_opex is not None
|
|
assert uow.scenario_processing_opex is not None
|
|
|
|
project_snapshots = uow.project_processing_opex.list_for_project(
|
|
project_id)
|
|
scenario_snapshots = uow.scenario_processing_opex.list_for_scenario(
|
|
scenario_id)
|
|
|
|
assert project_snapshots == []
|
|
assert scenario_snapshots == []
|