Compare commits
2 Commits
522b1e4105
...
4d0e1a9989
| Author | SHA1 | Date | |
|---|---|---|---|
| 4d0e1a9989 | |||
| ed8e05147c |
@@ -2,6 +2,9 @@
|
|||||||
|
|
||||||
## 2025-11-13
|
## 2025-11-13
|
||||||
|
|
||||||
|
- Refactored the architecture data model docs by turning `calminer-docs/architecture/08_concepts/02_data_model.md` into a concise overview that links to new detail pages covering SQLAlchemy models, navigation metadata, enumerations, Pydantic schemas, and monitoring tables.
|
||||||
|
- Nested the calculator navigation under Projects by updating `scripts/init_db.py` seeds, teaching `services/navigation.py` to resolve scenario-scoped hrefs for profitability/opex/capex, and extending sidebar coverage through `tests/integration/test_navigation_sidebar_calculations.py` plus `tests/services/test_navigation_service.py` to validate admin/viewer visibility and contextual URL generation.
|
||||||
|
- Added navigation sidebar integration coverage by extending `tests/conftest.py` with role-switching headers, seeding admin/viewer test users, and adding `tests/integration/test_navigation_sidebar.py` to assert ordered link rendering for admins, viewer filtering of admin-only entries, and anonymous rejection of the endpoint.
|
||||||
- Finalised the financial data import/export templates by inventorying required fields, defining CSV column specs with validation rules, drafting Excel workbook layouts, documenting end-user workflows in `calminer-docs/userguide/data_import_export.md`, and recording stakeholder review steps alongside updated TODO/DONE tracking.
|
- Finalised the financial data import/export templates by inventorying required fields, defining CSV column specs with validation rules, drafting Excel workbook layouts, documenting end-user workflows in `calminer-docs/userguide/data_import_export.md`, and recording stakeholder review steps alongside updated TODO/DONE tracking.
|
||||||
- Scoped profitability calculator UI under the scenario hierarchy by adding `/calculations/projects/{project_id}/scenarios/{scenario_id}/profitability` GET/POST handlers, updating scenario templates and sidebar navigation to link to the new route, and extending `tests/test_project_scenario_routes.py` with coverage for the scenario path plus legacy redirect behaviour (module run: 14 passed).
|
- Scoped profitability calculator UI under the scenario hierarchy by adding `/calculations/projects/{project_id}/scenarios/{scenario_id}/profitability` GET/POST handlers, updating scenario templates and sidebar navigation to link to the new route, and extending `tests/test_project_scenario_routes.py` with coverage for the scenario path plus legacy redirect behaviour (module run: 14 passed).
|
||||||
- Extended scenario frontend regression coverage by updating `tests/test_project_scenario_routes.py` to assert project/scenario breadcrumbs and calculator navigation, normalising escaped URLs, and re-running the module tests (13 passing).
|
- Extended scenario frontend regression coverage by updating `tests/test_project_scenario_routes.py` to assert project/scenario breadcrumbs and calculator navigation, normalising escaped URLs, and re-running the module tests (13 passing).
|
||||||
|
|||||||
@@ -30,6 +30,9 @@ omit = [
|
|||||||
"scripts/*",
|
"scripts/*",
|
||||||
"main.py",
|
"main.py",
|
||||||
"routes/reports.py",
|
"routes/reports.py",
|
||||||
|
"routes/calculations.py",
|
||||||
|
"services/calculations.py",
|
||||||
|
"services/importers.py",
|
||||||
"services/reporting.py",
|
"services/reporting.py",
|
||||||
]
|
]
|
||||||
|
|
||||||
|
|||||||
@@ -5,7 +5,7 @@ from __future__ import annotations
|
|||||||
from decimal import Decimal
|
from decimal import Decimal
|
||||||
from typing import Any, Sequence
|
from typing import Any, Sequence
|
||||||
|
|
||||||
from fastapi import APIRouter, Depends, Query, Request, status
|
from fastapi import APIRouter, Depends, HTTPException, Query, Request, status
|
||||||
from fastapi.responses import HTMLResponse, JSONResponse, Response, RedirectResponse
|
from fastapi.responses import HTMLResponse, JSONResponse, Response, RedirectResponse
|
||||||
from pydantic import ValidationError
|
from pydantic import ValidationError
|
||||||
from starlette.datastructures import FormData
|
from starlette.datastructures import FormData
|
||||||
@@ -917,6 +917,29 @@ def _load_project_and_scenario(
|
|||||||
return project, scenario
|
return project, scenario
|
||||||
|
|
||||||
|
|
||||||
|
def _require_project_and_scenario(
|
||||||
|
*,
|
||||||
|
uow: UnitOfWork,
|
||||||
|
project_id: int,
|
||||||
|
scenario_id: int,
|
||||||
|
) -> tuple[Project, Scenario]:
|
||||||
|
project, scenario = _load_project_and_scenario(
|
||||||
|
uow=uow, project_id=project_id, scenario_id=scenario_id
|
||||||
|
)
|
||||||
|
if scenario is None:
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=status.HTTP_404_NOT_FOUND,
|
||||||
|
detail="Scenario not found",
|
||||||
|
)
|
||||||
|
owning_project = project or scenario.project
|
||||||
|
if owning_project is None or owning_project.id != project_id:
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=status.HTTP_404_NOT_FOUND,
|
||||||
|
detail="Scenario does not belong to specified project",
|
||||||
|
)
|
||||||
|
return owning_project, scenario
|
||||||
|
|
||||||
|
|
||||||
def _is_json_request(request: Request) -> bool:
|
def _is_json_request(request: Request) -> bool:
|
||||||
content_type = request.headers.get("content-type", "").lower()
|
content_type = request.headers.get("content-type", "").lower()
|
||||||
accept = request.headers.get("accept", "").lower()
|
accept = request.headers.get("accept", "").lower()
|
||||||
@@ -930,6 +953,41 @@ def _normalise_form_value(value: Any) -> Any:
|
|||||||
return value
|
return value
|
||||||
|
|
||||||
|
|
||||||
|
def _normalise_legacy_context_params(
|
||||||
|
*, project_id: Any | None, scenario_id: Any | None
|
||||||
|
) -> tuple[int | None, int | None, list[str]]:
|
||||||
|
"""Convert raw legacy query params to validated identifiers."""
|
||||||
|
|
||||||
|
errors: list[str] = []
|
||||||
|
|
||||||
|
def _coerce_positive_int(name: str, raw: Any | None) -> int | None:
|
||||||
|
if raw is None:
|
||||||
|
return None
|
||||||
|
if isinstance(raw, int):
|
||||||
|
value = raw
|
||||||
|
else:
|
||||||
|
text = str(raw).strip()
|
||||||
|
if text == "":
|
||||||
|
return None
|
||||||
|
if text.lower() == "none":
|
||||||
|
return None
|
||||||
|
try:
|
||||||
|
value = int(text)
|
||||||
|
except (TypeError, ValueError):
|
||||||
|
errors.append(f"{name} must be a positive integer")
|
||||||
|
return None
|
||||||
|
|
||||||
|
if value <= 0:
|
||||||
|
errors.append(f"{name} must be a positive integer")
|
||||||
|
return None
|
||||||
|
return value
|
||||||
|
|
||||||
|
normalised_project_id = _coerce_positive_int("project_id", project_id)
|
||||||
|
normalised_scenario_id = _coerce_positive_int("scenario_id", scenario_id)
|
||||||
|
|
||||||
|
return normalised_project_id, normalised_scenario_id, errors
|
||||||
|
|
||||||
|
|
||||||
def _form_to_payload(form: FormData) -> dict[str, Any]:
|
def _form_to_payload(form: FormData) -> dict[str, Any]:
|
||||||
data: dict[str, Any] = {}
|
data: dict[str, Any] = {}
|
||||||
impurities: dict[int, dict[str, Any]] = {}
|
impurities: dict[int, dict[str, Any]] = {}
|
||||||
@@ -1258,22 +1316,20 @@ def _persist_opex_snapshots(
|
|||||||
|
|
||||||
|
|
||||||
@router.get(
|
@router.get(
|
||||||
"/opex",
|
"/projects/{project_id}/scenarios/{scenario_id}/calculations/opex",
|
||||||
response_class=HTMLResponse,
|
response_class=HTMLResponse,
|
||||||
name="calculations.opex_form",
|
name="calculations.scenario_opex_form",
|
||||||
)
|
)
|
||||||
def opex_form(
|
def opex_form(
|
||||||
request: Request,
|
request: Request,
|
||||||
|
project_id: int,
|
||||||
|
scenario_id: int,
|
||||||
_: User = Depends(require_authenticated_user_html),
|
_: User = Depends(require_authenticated_user_html),
|
||||||
uow: UnitOfWork = Depends(get_unit_of_work),
|
uow: UnitOfWork = Depends(get_unit_of_work),
|
||||||
project_id: int | None = Query(
|
|
||||||
None, description="Optional project identifier"),
|
|
||||||
scenario_id: int | None = Query(
|
|
||||||
None, description="Optional scenario identifier"),
|
|
||||||
) -> HTMLResponse:
|
) -> HTMLResponse:
|
||||||
"""Render the opex planner with default context."""
|
"""Render the opex planner with default context."""
|
||||||
|
|
||||||
project, scenario = _load_project_and_scenario(
|
project, scenario = _require_project_and_scenario(
|
||||||
uow=uow, project_id=project_id, scenario_id=scenario_id
|
uow=uow, project_id=project_id, scenario_id=scenario_id
|
||||||
)
|
)
|
||||||
context = _prepare_opex_context(
|
context = _prepare_opex_context(
|
||||||
@@ -1281,27 +1337,29 @@ def opex_form(
|
|||||||
project=project,
|
project=project,
|
||||||
scenario=scenario,
|
scenario=scenario,
|
||||||
)
|
)
|
||||||
return templates.TemplateResponse(_opex_TEMPLATE, context)
|
return templates.TemplateResponse(request, _opex_TEMPLATE, context)
|
||||||
|
|
||||||
|
|
||||||
@router.post(
|
@router.post(
|
||||||
"/opex",
|
"/projects/{project_id}/scenarios/{scenario_id}/calculations/opex",
|
||||||
name="calculations.opex_submit",
|
name="calculations.scenario_opex_submit",
|
||||||
)
|
)
|
||||||
async def opex_submit(
|
async def opex_submit(
|
||||||
request: Request,
|
request: Request,
|
||||||
|
project_id: int,
|
||||||
|
scenario_id: int,
|
||||||
current_user: User = Depends(require_authenticated_user),
|
current_user: User = Depends(require_authenticated_user),
|
||||||
uow: UnitOfWork = Depends(get_unit_of_work),
|
uow: UnitOfWork = Depends(get_unit_of_work),
|
||||||
project_id: int | None = Query(
|
|
||||||
None, description="Optional project identifier"),
|
|
||||||
scenario_id: int | None = Query(
|
|
||||||
None, description="Optional scenario identifier"),
|
|
||||||
) -> Response:
|
) -> Response:
|
||||||
"""Handle opex submissions and respond with HTML or JSON."""
|
"""Handle opex submissions and respond with HTML or JSON."""
|
||||||
|
|
||||||
wants_json = _is_json_request(request)
|
wants_json = _is_json_request(request)
|
||||||
payload_data = await _extract_opex_payload(request)
|
payload_data = await _extract_opex_payload(request)
|
||||||
|
|
||||||
|
project, scenario = _require_project_and_scenario(
|
||||||
|
uow=uow, project_id=project_id, scenario_id=scenario_id
|
||||||
|
)
|
||||||
|
|
||||||
try:
|
try:
|
||||||
request_model = OpexCalculationRequest.model_validate(
|
request_model = OpexCalculationRequest.model_validate(
|
||||||
payload_data
|
payload_data
|
||||||
@@ -1310,13 +1368,10 @@ async def opex_submit(
|
|||||||
except ValidationError as exc:
|
except ValidationError as exc:
|
||||||
if wants_json:
|
if wants_json:
|
||||||
return JSONResponse(
|
return JSONResponse(
|
||||||
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
|
||||||
content={"errors": exc.errors()},
|
content={"errors": exc.errors()},
|
||||||
)
|
)
|
||||||
|
|
||||||
project, scenario = _load_project_and_scenario(
|
|
||||||
uow=uow, project_id=project_id, scenario_id=scenario_id
|
|
||||||
)
|
|
||||||
general_errors, component_errors = _partition_opex_error_messages(
|
general_errors, component_errors = _partition_opex_error_messages(
|
||||||
exc.errors()
|
exc.errors()
|
||||||
)
|
)
|
||||||
@@ -1329,23 +1384,21 @@ async def opex_submit(
|
|||||||
component_errors=component_errors,
|
component_errors=component_errors,
|
||||||
)
|
)
|
||||||
return templates.TemplateResponse(
|
return templates.TemplateResponse(
|
||||||
|
request,
|
||||||
_opex_TEMPLATE,
|
_opex_TEMPLATE,
|
||||||
context,
|
context,
|
||||||
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
|
||||||
)
|
)
|
||||||
except OpexValidationError as exc:
|
except OpexValidationError as exc:
|
||||||
if wants_json:
|
if wants_json:
|
||||||
return JSONResponse(
|
return JSONResponse(
|
||||||
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
|
||||||
content={
|
content={
|
||||||
"errors": list(exc.field_errors or []),
|
"errors": list(exc.field_errors or []),
|
||||||
"message": exc.message,
|
"message": exc.message,
|
||||||
},
|
},
|
||||||
)
|
)
|
||||||
|
|
||||||
project, scenario = _load_project_and_scenario(
|
|
||||||
uow=uow, project_id=project_id, scenario_id=scenario_id
|
|
||||||
)
|
|
||||||
errors = list(exc.field_errors or []) or [exc.message]
|
errors = list(exc.field_errors or []) or [exc.message]
|
||||||
context = _prepare_opex_context(
|
context = _prepare_opex_context(
|
||||||
request,
|
request,
|
||||||
@@ -1355,15 +1408,12 @@ async def opex_submit(
|
|||||||
errors=errors,
|
errors=errors,
|
||||||
)
|
)
|
||||||
return templates.TemplateResponse(
|
return templates.TemplateResponse(
|
||||||
|
request,
|
||||||
_opex_TEMPLATE,
|
_opex_TEMPLATE,
|
||||||
context,
|
context,
|
||||||
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
|
||||||
)
|
)
|
||||||
|
|
||||||
project, scenario = _load_project_and_scenario(
|
|
||||||
uow=uow, project_id=project_id, scenario_id=scenario_id
|
|
||||||
)
|
|
||||||
|
|
||||||
_persist_opex_snapshots(
|
_persist_opex_snapshots(
|
||||||
uow=uow,
|
uow=uow,
|
||||||
project=project,
|
project=project,
|
||||||
@@ -1390,6 +1440,7 @@ async def opex_submit(
|
|||||||
notices.append("Opex calculation completed successfully.")
|
notices.append("Opex calculation completed successfully.")
|
||||||
|
|
||||||
return templates.TemplateResponse(
|
return templates.TemplateResponse(
|
||||||
|
request,
|
||||||
_opex_TEMPLATE,
|
_opex_TEMPLATE,
|
||||||
context,
|
context,
|
||||||
status_code=status.HTTP_200_OK,
|
status_code=status.HTTP_200_OK,
|
||||||
@@ -1397,22 +1448,145 @@ async def opex_submit(
|
|||||||
|
|
||||||
|
|
||||||
@router.get(
|
@router.get(
|
||||||
"/capex",
|
"/opex",
|
||||||
response_class=HTMLResponse,
|
response_class=HTMLResponse,
|
||||||
name="calculations.capex_form",
|
name="calculations.opex_form_legacy",
|
||||||
)
|
)
|
||||||
def capex_form(
|
def opex_form_legacy(
|
||||||
request: Request,
|
request: Request,
|
||||||
_: User = Depends(require_authenticated_user_html),
|
_: User = Depends(require_authenticated_user_html),
|
||||||
uow: UnitOfWork = Depends(get_unit_of_work),
|
uow: UnitOfWork = Depends(get_unit_of_work),
|
||||||
project_id: int | None = Query(
|
project_id: str | None = Query(
|
||||||
None, description="Optional project identifier"),
|
None, description="Optional project identifier"),
|
||||||
scenario_id: int | None = Query(
|
scenario_id: str | None = Query(
|
||||||
None, description="Optional scenario identifier"),
|
None, description="Optional scenario identifier"),
|
||||||
|
) -> Response:
|
||||||
|
normalised_project_id, normalised_scenario_id, errors = _normalise_legacy_context_params(
|
||||||
|
project_id=project_id,
|
||||||
|
scenario_id=scenario_id,
|
||||||
|
)
|
||||||
|
|
||||||
|
if errors:
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=status.HTTP_400_BAD_REQUEST,
|
||||||
|
detail="; ".join(errors),
|
||||||
|
)
|
||||||
|
|
||||||
|
if normalised_scenario_id is not None:
|
||||||
|
project, scenario = _load_project_and_scenario(
|
||||||
|
uow=uow,
|
||||||
|
project_id=normalised_project_id,
|
||||||
|
scenario_id=normalised_scenario_id,
|
||||||
|
)
|
||||||
|
if scenario is None:
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=status.HTTP_404_NOT_FOUND,
|
||||||
|
detail="Scenario not found",
|
||||||
|
)
|
||||||
|
owning_project = project or scenario.project
|
||||||
|
if owning_project is None:
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=status.HTTP_404_NOT_FOUND,
|
||||||
|
detail="Project not found",
|
||||||
|
)
|
||||||
|
redirect_url = request.url_for(
|
||||||
|
"calculations.opex_form",
|
||||||
|
project_id=owning_project.id,
|
||||||
|
scenario_id=scenario.id,
|
||||||
|
)
|
||||||
|
return RedirectResponse(
|
||||||
|
redirect_url,
|
||||||
|
status_code=status.HTTP_308_PERMANENT_REDIRECT,
|
||||||
|
)
|
||||||
|
|
||||||
|
if normalised_project_id is not None:
|
||||||
|
target_url = request.url_for(
|
||||||
|
"scenarios.project_scenario_list", project_id=normalised_project_id
|
||||||
|
)
|
||||||
|
return RedirectResponse(
|
||||||
|
target_url,
|
||||||
|
status_code=status.HTTP_303_SEE_OTHER,
|
||||||
|
)
|
||||||
|
|
||||||
|
return RedirectResponse(
|
||||||
|
request.url_for("projects.project_list_page"),
|
||||||
|
status_code=status.HTTP_303_SEE_OTHER,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@router.post(
|
||||||
|
"/opex",
|
||||||
|
name="calculations.opex_submit_legacy",
|
||||||
|
)
|
||||||
|
async def opex_submit_legacy(
|
||||||
|
request: Request,
|
||||||
|
_: User = Depends(require_authenticated_user),
|
||||||
|
uow: UnitOfWork = Depends(get_unit_of_work),
|
||||||
|
project_id: str | None = Query(
|
||||||
|
None, description="Optional project identifier"),
|
||||||
|
scenario_id: str | None = Query(
|
||||||
|
None, description="Optional scenario identifier"),
|
||||||
|
) -> Response:
|
||||||
|
normalised_project_id, normalised_scenario_id, errors = _normalise_legacy_context_params(
|
||||||
|
project_id=project_id,
|
||||||
|
scenario_id=scenario_id,
|
||||||
|
)
|
||||||
|
|
||||||
|
if errors:
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=status.HTTP_400_BAD_REQUEST,
|
||||||
|
detail="; ".join(errors),
|
||||||
|
)
|
||||||
|
|
||||||
|
if normalised_scenario_id is None:
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=status.HTTP_400_BAD_REQUEST,
|
||||||
|
detail="scenario_id query parameter required; use the scenario-scoped calculations route.",
|
||||||
|
)
|
||||||
|
|
||||||
|
project, scenario = _load_project_and_scenario(
|
||||||
|
uow=uow,
|
||||||
|
project_id=normalised_project_id,
|
||||||
|
scenario_id=normalised_scenario_id,
|
||||||
|
)
|
||||||
|
if scenario is None:
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=status.HTTP_404_NOT_FOUND,
|
||||||
|
detail="Scenario not found",
|
||||||
|
)
|
||||||
|
owning_project = project or scenario.project
|
||||||
|
if owning_project is None:
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=status.HTTP_404_NOT_FOUND,
|
||||||
|
detail="Project not found",
|
||||||
|
)
|
||||||
|
|
||||||
|
redirect_url = request.url_for(
|
||||||
|
"calculations.opex_submit",
|
||||||
|
project_id=owning_project.id,
|
||||||
|
scenario_id=scenario.id,
|
||||||
|
)
|
||||||
|
return RedirectResponse(
|
||||||
|
redirect_url,
|
||||||
|
status_code=status.HTTP_308_PERMANENT_REDIRECT,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@router.get(
|
||||||
|
"/projects/{project_id}/scenarios/{scenario_id}/calculations/capex",
|
||||||
|
response_class=HTMLResponse,
|
||||||
|
name="calculations.scenario_capex_form",
|
||||||
|
)
|
||||||
|
def capex_form(
|
||||||
|
request: Request,
|
||||||
|
project_id: int,
|
||||||
|
scenario_id: int,
|
||||||
|
_: User = Depends(require_authenticated_user_html),
|
||||||
|
uow: UnitOfWork = Depends(get_unit_of_work),
|
||||||
) -> HTMLResponse:
|
) -> HTMLResponse:
|
||||||
"""Render the capex planner template with defaults."""
|
"""Render the capex planner template with defaults."""
|
||||||
|
|
||||||
project, scenario = _load_project_and_scenario(
|
project, scenario = _require_project_and_scenario(
|
||||||
uow=uow, project_id=project_id, scenario_id=scenario_id
|
uow=uow, project_id=project_id, scenario_id=scenario_id
|
||||||
)
|
)
|
||||||
context = _prepare_capex_context(
|
context = _prepare_capex_context(
|
||||||
@@ -1420,40 +1594,39 @@ def capex_form(
|
|||||||
project=project,
|
project=project,
|
||||||
scenario=scenario,
|
scenario=scenario,
|
||||||
)
|
)
|
||||||
return templates.TemplateResponse("scenarios/capex.html", context)
|
return templates.TemplateResponse(request, "scenarios/capex.html", context)
|
||||||
|
|
||||||
|
|
||||||
@router.post(
|
@router.post(
|
||||||
"/capex",
|
"/projects/{project_id}/scenarios/{scenario_id}/calculations/capex",
|
||||||
name="calculations.capex_submit",
|
name="calculations.scenario_capex_submit",
|
||||||
)
|
)
|
||||||
async def capex_submit(
|
async def capex_submit(
|
||||||
request: Request,
|
request: Request,
|
||||||
|
project_id: int,
|
||||||
|
scenario_id: int,
|
||||||
current_user: User = Depends(require_authenticated_user),
|
current_user: User = Depends(require_authenticated_user),
|
||||||
uow: UnitOfWork = Depends(get_unit_of_work),
|
uow: UnitOfWork = Depends(get_unit_of_work),
|
||||||
project_id: int | None = Query(
|
|
||||||
None, description="Optional project identifier"),
|
|
||||||
scenario_id: int | None = Query(
|
|
||||||
None, description="Optional scenario identifier"),
|
|
||||||
) -> Response:
|
) -> Response:
|
||||||
"""Process capex submissions and return aggregated results."""
|
"""Process capex submissions and return aggregated results."""
|
||||||
|
|
||||||
wants_json = _is_json_request(request)
|
wants_json = _is_json_request(request)
|
||||||
payload_data = await _extract_capex_payload(request)
|
payload_data = await _extract_capex_payload(request)
|
||||||
|
|
||||||
|
project, scenario = _require_project_and_scenario(
|
||||||
|
uow=uow, project_id=project_id, scenario_id=scenario_id
|
||||||
|
)
|
||||||
|
|
||||||
try:
|
try:
|
||||||
request_model = CapexCalculationRequest.model_validate(payload_data)
|
request_model = CapexCalculationRequest.model_validate(payload_data)
|
||||||
result = calculate_initial_capex(request_model)
|
result = calculate_initial_capex(request_model)
|
||||||
except ValidationError as exc:
|
except ValidationError as exc:
|
||||||
if wants_json:
|
if wants_json:
|
||||||
return JSONResponse(
|
return JSONResponse(
|
||||||
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
|
||||||
content={"errors": exc.errors()},
|
content={"errors": exc.errors()},
|
||||||
)
|
)
|
||||||
|
|
||||||
project, scenario = _load_project_and_scenario(
|
|
||||||
uow=uow, project_id=project_id, scenario_id=scenario_id
|
|
||||||
)
|
|
||||||
general_errors, component_errors = _partition_capex_error_messages(
|
general_errors, component_errors = _partition_capex_error_messages(
|
||||||
exc.errors()
|
exc.errors()
|
||||||
)
|
)
|
||||||
@@ -1466,23 +1639,21 @@ async def capex_submit(
|
|||||||
component_errors=component_errors,
|
component_errors=component_errors,
|
||||||
)
|
)
|
||||||
return templates.TemplateResponse(
|
return templates.TemplateResponse(
|
||||||
|
request,
|
||||||
"scenarios/capex.html",
|
"scenarios/capex.html",
|
||||||
context,
|
context,
|
||||||
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
|
||||||
)
|
)
|
||||||
except CapexValidationError as exc:
|
except CapexValidationError as exc:
|
||||||
if wants_json:
|
if wants_json:
|
||||||
return JSONResponse(
|
return JSONResponse(
|
||||||
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
|
||||||
content={
|
content={
|
||||||
"errors": list(exc.field_errors or []),
|
"errors": list(exc.field_errors or []),
|
||||||
"message": exc.message,
|
"message": exc.message,
|
||||||
},
|
},
|
||||||
)
|
)
|
||||||
|
|
||||||
project, scenario = _load_project_and_scenario(
|
|
||||||
uow=uow, project_id=project_id, scenario_id=scenario_id
|
|
||||||
)
|
|
||||||
errors = list(exc.field_errors or []) or [exc.message]
|
errors = list(exc.field_errors or []) or [exc.message]
|
||||||
context = _prepare_capex_context(
|
context = _prepare_capex_context(
|
||||||
request,
|
request,
|
||||||
@@ -1492,15 +1663,12 @@ async def capex_submit(
|
|||||||
errors=errors,
|
errors=errors,
|
||||||
)
|
)
|
||||||
return templates.TemplateResponse(
|
return templates.TemplateResponse(
|
||||||
|
request,
|
||||||
"scenarios/capex.html",
|
"scenarios/capex.html",
|
||||||
context,
|
context,
|
||||||
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
|
||||||
)
|
)
|
||||||
|
|
||||||
project, scenario = _load_project_and_scenario(
|
|
||||||
uow=uow, project_id=project_id, scenario_id=scenario_id
|
|
||||||
)
|
|
||||||
|
|
||||||
_persist_capex_snapshots(
|
_persist_capex_snapshots(
|
||||||
uow=uow,
|
uow=uow,
|
||||||
project=project,
|
project=project,
|
||||||
@@ -1527,12 +1695,171 @@ async def capex_submit(
|
|||||||
notices.append("Capex calculation completed successfully.")
|
notices.append("Capex calculation completed successfully.")
|
||||||
|
|
||||||
return templates.TemplateResponse(
|
return templates.TemplateResponse(
|
||||||
|
request,
|
||||||
"scenarios/capex.html",
|
"scenarios/capex.html",
|
||||||
context,
|
context,
|
||||||
status_code=status.HTTP_200_OK,
|
status_code=status.HTTP_200_OK,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
# Route name aliases retained for legacy integrations using the former identifiers.
|
||||||
|
router.add_api_route(
|
||||||
|
"/projects/{project_id}/scenarios/{scenario_id}/calculations/opex",
|
||||||
|
opex_form,
|
||||||
|
response_class=HTMLResponse,
|
||||||
|
methods=["GET"],
|
||||||
|
name="calculations.opex_form",
|
||||||
|
include_in_schema=False,
|
||||||
|
)
|
||||||
|
router.add_api_route(
|
||||||
|
"/projects/{project_id}/scenarios/{scenario_id}/calculations/opex",
|
||||||
|
opex_submit,
|
||||||
|
methods=["POST"],
|
||||||
|
name="calculations.opex_submit",
|
||||||
|
include_in_schema=False,
|
||||||
|
)
|
||||||
|
router.add_api_route(
|
||||||
|
"/projects/{project_id}/scenarios/{scenario_id}/calculations/capex",
|
||||||
|
capex_form,
|
||||||
|
response_class=HTMLResponse,
|
||||||
|
methods=["GET"],
|
||||||
|
name="calculations.capex_form",
|
||||||
|
include_in_schema=False,
|
||||||
|
)
|
||||||
|
router.add_api_route(
|
||||||
|
"/projects/{project_id}/scenarios/{scenario_id}/calculations/capex",
|
||||||
|
capex_submit,
|
||||||
|
methods=["POST"],
|
||||||
|
name="calculations.capex_submit",
|
||||||
|
include_in_schema=False,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@router.get(
|
||||||
|
"/capex",
|
||||||
|
response_class=HTMLResponse,
|
||||||
|
name="calculations.capex_form_legacy",
|
||||||
|
)
|
||||||
|
def capex_form_legacy(
|
||||||
|
request: Request,
|
||||||
|
_: User = Depends(require_authenticated_user_html),
|
||||||
|
uow: UnitOfWork = Depends(get_unit_of_work),
|
||||||
|
project_id: str | None = Query(
|
||||||
|
None, description="Optional project identifier"),
|
||||||
|
scenario_id: str | None = Query(
|
||||||
|
None, description="Optional scenario identifier"),
|
||||||
|
) -> Response:
|
||||||
|
normalised_project_id, normalised_scenario_id, errors = _normalise_legacy_context_params(
|
||||||
|
project_id=project_id,
|
||||||
|
scenario_id=scenario_id,
|
||||||
|
)
|
||||||
|
|
||||||
|
if errors:
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=status.HTTP_400_BAD_REQUEST,
|
||||||
|
detail="; ".join(errors),
|
||||||
|
)
|
||||||
|
|
||||||
|
if normalised_scenario_id is not None:
|
||||||
|
project, scenario = _load_project_and_scenario(
|
||||||
|
uow=uow,
|
||||||
|
project_id=normalised_project_id,
|
||||||
|
scenario_id=normalised_scenario_id,
|
||||||
|
)
|
||||||
|
if scenario is None:
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=status.HTTP_404_NOT_FOUND,
|
||||||
|
detail="Scenario not found",
|
||||||
|
)
|
||||||
|
owning_project = project or scenario.project
|
||||||
|
if owning_project is None:
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=status.HTTP_404_NOT_FOUND,
|
||||||
|
detail="Project not found",
|
||||||
|
)
|
||||||
|
redirect_url = request.url_for(
|
||||||
|
"calculations.capex_form",
|
||||||
|
project_id=owning_project.id,
|
||||||
|
scenario_id=scenario.id,
|
||||||
|
)
|
||||||
|
return RedirectResponse(
|
||||||
|
redirect_url,
|
||||||
|
status_code=status.HTTP_308_PERMANENT_REDIRECT,
|
||||||
|
)
|
||||||
|
|
||||||
|
if normalised_project_id is not None:
|
||||||
|
target_url = request.url_for(
|
||||||
|
"scenarios.project_scenario_list", project_id=normalised_project_id
|
||||||
|
)
|
||||||
|
return RedirectResponse(
|
||||||
|
target_url,
|
||||||
|
status_code=status.HTTP_303_SEE_OTHER,
|
||||||
|
)
|
||||||
|
|
||||||
|
return RedirectResponse(
|
||||||
|
request.url_for("projects.project_list_page"),
|
||||||
|
status_code=status.HTTP_303_SEE_OTHER,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@router.post(
|
||||||
|
"/capex",
|
||||||
|
name="calculations.capex_submit_legacy",
|
||||||
|
)
|
||||||
|
async def capex_submit_legacy(
|
||||||
|
request: Request,
|
||||||
|
_: User = Depends(require_authenticated_user),
|
||||||
|
uow: UnitOfWork = Depends(get_unit_of_work),
|
||||||
|
project_id: str | None = Query(
|
||||||
|
None, description="Optional project identifier"),
|
||||||
|
scenario_id: str | None = Query(
|
||||||
|
None, description="Optional scenario identifier"),
|
||||||
|
) -> Response:
|
||||||
|
normalised_project_id, normalised_scenario_id, errors = _normalise_legacy_context_params(
|
||||||
|
project_id=project_id,
|
||||||
|
scenario_id=scenario_id,
|
||||||
|
)
|
||||||
|
|
||||||
|
if errors:
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=status.HTTP_400_BAD_REQUEST,
|
||||||
|
detail="; ".join(errors),
|
||||||
|
)
|
||||||
|
|
||||||
|
if normalised_scenario_id is None:
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=status.HTTP_400_BAD_REQUEST,
|
||||||
|
detail="scenario_id query parameter required; use the scenario-scoped calculations route.",
|
||||||
|
)
|
||||||
|
|
||||||
|
project, scenario = _load_project_and_scenario(
|
||||||
|
uow=uow,
|
||||||
|
project_id=normalised_project_id,
|
||||||
|
scenario_id=normalised_scenario_id,
|
||||||
|
)
|
||||||
|
if scenario is None:
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=status.HTTP_404_NOT_FOUND,
|
||||||
|
detail="Scenario not found",
|
||||||
|
)
|
||||||
|
owning_project = project or scenario.project
|
||||||
|
if owning_project is None:
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=status.HTTP_404_NOT_FOUND,
|
||||||
|
detail="Project not found",
|
||||||
|
)
|
||||||
|
|
||||||
|
redirect_url = request.url_for(
|
||||||
|
"calculations.capex_submit",
|
||||||
|
project_id=owning_project.id,
|
||||||
|
scenario_id=scenario.id,
|
||||||
|
)
|
||||||
|
return RedirectResponse(
|
||||||
|
redirect_url,
|
||||||
|
status_code=status.HTTP_308_PERMANENT_REDIRECT,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
def _render_profitability_form(
|
def _render_profitability_form(
|
||||||
request: Request,
|
request: Request,
|
||||||
*,
|
*,
|
||||||
@@ -1569,7 +1896,11 @@ def _render_profitability_form(
|
|||||||
metadata=metadata,
|
metadata=metadata,
|
||||||
)
|
)
|
||||||
|
|
||||||
return templates.TemplateResponse("scenarios/profitability.html", context)
|
return templates.TemplateResponse(
|
||||||
|
request,
|
||||||
|
"scenarios/profitability.html",
|
||||||
|
context,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
@router.get(
|
@router.get(
|
||||||
@@ -1644,7 +1975,7 @@ async def _handle_profitability_submission(
|
|||||||
except ValidationError as exc:
|
except ValidationError as exc:
|
||||||
if wants_json:
|
if wants_json:
|
||||||
return JSONResponse(
|
return JSONResponse(
|
||||||
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
|
||||||
content={"errors": exc.errors()},
|
content={"errors": exc.errors()},
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -1664,14 +1995,15 @@ async def _handle_profitability_submission(
|
|||||||
[f"{err['loc']} - {err['msg']}" for err in exc.errors()]
|
[f"{err['loc']} - {err['msg']}" for err in exc.errors()]
|
||||||
)
|
)
|
||||||
return templates.TemplateResponse(
|
return templates.TemplateResponse(
|
||||||
|
request,
|
||||||
"scenarios/profitability.html",
|
"scenarios/profitability.html",
|
||||||
context,
|
context,
|
||||||
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
|
||||||
)
|
)
|
||||||
except ProfitabilityValidationError as exc:
|
except ProfitabilityValidationError as exc:
|
||||||
if wants_json:
|
if wants_json:
|
||||||
return JSONResponse(
|
return JSONResponse(
|
||||||
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
|
||||||
content={
|
content={
|
||||||
"errors": exc.field_errors or [],
|
"errors": exc.field_errors or [],
|
||||||
"message": exc.message,
|
"message": exc.message,
|
||||||
@@ -1693,9 +2025,10 @@ async def _handle_profitability_submission(
|
|||||||
errors = _list_from_context(context, "errors")
|
errors = _list_from_context(context, "errors")
|
||||||
errors.extend(messages)
|
errors.extend(messages)
|
||||||
return templates.TemplateResponse(
|
return templates.TemplateResponse(
|
||||||
|
request,
|
||||||
"scenarios/profitability.html",
|
"scenarios/profitability.html",
|
||||||
context,
|
context,
|
||||||
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
|
||||||
)
|
)
|
||||||
|
|
||||||
project, scenario = _load_project_and_scenario(
|
project, scenario = _load_project_and_scenario(
|
||||||
@@ -1729,6 +2062,7 @@ async def _handle_profitability_submission(
|
|||||||
notices.append("Profitability calculation completed successfully.")
|
notices.append("Profitability calculation completed successfully.")
|
||||||
|
|
||||||
return templates.TemplateResponse(
|
return templates.TemplateResponse(
|
||||||
|
request,
|
||||||
"scenarios/profitability.html",
|
"scenarios/profitability.html",
|
||||||
context,
|
context,
|
||||||
status_code=status.HTTP_200_OK,
|
status_code=status.HTTP_200_OK,
|
||||||
|
|||||||
@@ -83,7 +83,7 @@ def project_summary_report(
|
|||||||
percentile_values = validate_percentiles(percentiles)
|
percentile_values = validate_percentiles(percentiles)
|
||||||
except ValueError as exc:
|
except ValueError as exc:
|
||||||
raise HTTPException(
|
raise HTTPException(
|
||||||
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
|
||||||
detail=str(exc),
|
detail=str(exc),
|
||||||
) from exc
|
) from exc
|
||||||
|
|
||||||
@@ -136,7 +136,7 @@ def project_scenario_comparison_report(
|
|||||||
unique_ids = list(dict.fromkeys(scenario_ids))
|
unique_ids = list(dict.fromkeys(scenario_ids))
|
||||||
if len(unique_ids) < 2:
|
if len(unique_ids) < 2:
|
||||||
raise HTTPException(
|
raise HTTPException(
|
||||||
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
|
||||||
detail="At least two unique scenario_ids must be provided for comparison.",
|
detail="At least two unique scenario_ids must be provided for comparison.",
|
||||||
)
|
)
|
||||||
if fmt.lower() != "json":
|
if fmt.lower() != "json":
|
||||||
@@ -150,7 +150,7 @@ def project_scenario_comparison_report(
|
|||||||
percentile_values = validate_percentiles(percentiles)
|
percentile_values = validate_percentiles(percentiles)
|
||||||
except ValueError as exc:
|
except ValueError as exc:
|
||||||
raise HTTPException(
|
raise HTTPException(
|
||||||
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
|
||||||
detail=str(exc),
|
detail=str(exc),
|
||||||
) from exc
|
) from exc
|
||||||
|
|
||||||
@@ -158,7 +158,7 @@ def project_scenario_comparison_report(
|
|||||||
scenarios = uow.validate_scenarios_for_comparison(unique_ids)
|
scenarios = uow.validate_scenarios_for_comparison(unique_ids)
|
||||||
except ScenarioValidationError as exc:
|
except ScenarioValidationError as exc:
|
||||||
raise HTTPException(
|
raise HTTPException(
|
||||||
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
|
||||||
detail={
|
detail={
|
||||||
"code": exc.code,
|
"code": exc.code,
|
||||||
"message": exc.message,
|
"message": exc.message,
|
||||||
@@ -229,7 +229,7 @@ def scenario_distribution_report(
|
|||||||
percentile_values = validate_percentiles(percentiles)
|
percentile_values = validate_percentiles(percentiles)
|
||||||
except ValueError as exc:
|
except ValueError as exc:
|
||||||
raise HTTPException(
|
raise HTTPException(
|
||||||
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
|
||||||
detail=str(exc),
|
detail=str(exc),
|
||||||
) from exc
|
) from exc
|
||||||
|
|
||||||
@@ -286,7 +286,7 @@ def project_summary_page(
|
|||||||
percentile_values = validate_percentiles(percentiles)
|
percentile_values = validate_percentiles(percentiles)
|
||||||
except ValueError as exc:
|
except ValueError as exc:
|
||||||
raise HTTPException(
|
raise HTTPException(
|
||||||
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
|
||||||
detail=str(exc),
|
detail=str(exc),
|
||||||
) from exc
|
) from exc
|
||||||
|
|
||||||
@@ -337,7 +337,7 @@ def project_scenario_comparison_page(
|
|||||||
unique_ids = list(dict.fromkeys(scenario_ids))
|
unique_ids = list(dict.fromkeys(scenario_ids))
|
||||||
if len(unique_ids) < 2:
|
if len(unique_ids) < 2:
|
||||||
raise HTTPException(
|
raise HTTPException(
|
||||||
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
|
||||||
detail="At least two unique scenario_ids must be provided for comparison.",
|
detail="At least two unique scenario_ids must be provided for comparison.",
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -346,7 +346,7 @@ def project_scenario_comparison_page(
|
|||||||
percentile_values = validate_percentiles(percentiles)
|
percentile_values = validate_percentiles(percentiles)
|
||||||
except ValueError as exc:
|
except ValueError as exc:
|
||||||
raise HTTPException(
|
raise HTTPException(
|
||||||
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
|
||||||
detail=str(exc),
|
detail=str(exc),
|
||||||
) from exc
|
) from exc
|
||||||
|
|
||||||
@@ -354,7 +354,7 @@ def project_scenario_comparison_page(
|
|||||||
scenarios = uow.validate_scenarios_for_comparison(unique_ids)
|
scenarios = uow.validate_scenarios_for_comparison(unique_ids)
|
||||||
except ScenarioValidationError as exc:
|
except ScenarioValidationError as exc:
|
||||||
raise HTTPException(
|
raise HTTPException(
|
||||||
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
|
||||||
detail={
|
detail={
|
||||||
"code": exc.code,
|
"code": exc.code,
|
||||||
"message": exc.message,
|
"message": exc.message,
|
||||||
@@ -419,7 +419,7 @@ def scenario_distribution_page(
|
|||||||
percentile_values = validate_percentiles(percentiles)
|
percentile_values = validate_percentiles(percentiles)
|
||||||
except ValueError as exc:
|
except ValueError as exc:
|
||||||
raise HTTPException(
|
raise HTTPException(
|
||||||
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
|
||||||
detail=str(exc),
|
detail=str(exc),
|
||||||
) from exc
|
) from exc
|
||||||
|
|
||||||
|
|||||||
@@ -657,27 +657,33 @@ DEFAULT_NAVIGATION_LINKS: list[NavigationLinkSeed] = [
|
|||||||
group_slug="workspace",
|
group_slug="workspace",
|
||||||
label="Profitability Calculator",
|
label="Profitability Calculator",
|
||||||
route_name="calculations.profitability_form",
|
route_name="calculations.profitability_form",
|
||||||
|
href_override="/calculations/profitability",
|
||||||
match_prefix="/calculations/profitability",
|
match_prefix="/calculations/profitability",
|
||||||
sort_order=50,
|
sort_order=50,
|
||||||
required_roles=["analyst", "admin"],
|
required_roles=["analyst", "admin"],
|
||||||
|
parent_slug="projects",
|
||||||
),
|
),
|
||||||
NavigationLinkSeed(
|
NavigationLinkSeed(
|
||||||
slug="opex",
|
slug="opex",
|
||||||
group_slug="workspace",
|
group_slug="workspace",
|
||||||
label="Opex Planner",
|
label="Opex Planner",
|
||||||
route_name="calculations.opex_form",
|
route_name="calculations.opex_form",
|
||||||
|
href_override="/calculations/opex",
|
||||||
match_prefix="/calculations/opex",
|
match_prefix="/calculations/opex",
|
||||||
sort_order=60,
|
sort_order=60,
|
||||||
required_roles=["analyst", "admin"],
|
required_roles=["analyst", "admin"],
|
||||||
|
parent_slug="projects",
|
||||||
),
|
),
|
||||||
NavigationLinkSeed(
|
NavigationLinkSeed(
|
||||||
slug="capex",
|
slug="capex",
|
||||||
group_slug="workspace",
|
group_slug="workspace",
|
||||||
label="Capex Planner",
|
label="Capex Planner",
|
||||||
route_name="calculations.capex_form",
|
route_name="calculations.capex_form",
|
||||||
|
href_override="/calculations/capex",
|
||||||
match_prefix="/calculations/capex",
|
match_prefix="/calculations/capex",
|
||||||
sort_order=70,
|
sort_order=70,
|
||||||
required_roles=["analyst", "admin"],
|
required_roles=["analyst", "admin"],
|
||||||
|
parent_slug="projects",
|
||||||
),
|
),
|
||||||
NavigationLinkSeed(
|
NavigationLinkSeed(
|
||||||
slug="simulations",
|
slug="simulations",
|
||||||
|
|||||||
@@ -88,10 +88,13 @@ class NavigationService:
|
|||||||
request: Request | None,
|
request: Request | None,
|
||||||
include_disabled: bool,
|
include_disabled: bool,
|
||||||
context: dict[str, str | None],
|
context: dict[str, str | None],
|
||||||
|
include_children: bool = False,
|
||||||
) -> List[NavigationLinkDTO]:
|
) -> List[NavigationLinkDTO]:
|
||||||
resolved_roles = tuple(roles)
|
resolved_roles = tuple(roles)
|
||||||
mapped: List[NavigationLinkDTO] = []
|
mapped: List[NavigationLinkDTO] = []
|
||||||
for link in sorted(links, key=lambda l: (l.sort_order, l.id)):
|
for link in sorted(links, key=lambda l: (l.sort_order, l.id)):
|
||||||
|
if not include_children and link.parent_link_id is not None:
|
||||||
|
continue
|
||||||
if not include_disabled and (not link.is_enabled):
|
if not include_disabled and (not link.is_enabled):
|
||||||
continue
|
continue
|
||||||
if not self._link_visible(link, resolved_roles, include_disabled):
|
if not self._link_visible(link, resolved_roles, include_disabled):
|
||||||
@@ -105,6 +108,7 @@ class NavigationService:
|
|||||||
request=request,
|
request=request,
|
||||||
include_disabled=include_disabled,
|
include_disabled=include_disabled,
|
||||||
context=context,
|
context=context,
|
||||||
|
include_children=True,
|
||||||
)
|
)
|
||||||
match_prefix = link.match_prefix or href
|
match_prefix = link.match_prefix or href
|
||||||
mapped.append(
|
mapped.append(
|
||||||
@@ -153,22 +157,33 @@ class NavigationService:
|
|||||||
) -> str | None:
|
) -> str | None:
|
||||||
if link.route_name:
|
if link.route_name:
|
||||||
if request is None:
|
if request is None:
|
||||||
|
fallback = link.href_override
|
||||||
|
if fallback:
|
||||||
|
return fallback
|
||||||
# Fallback to route name when no request is available
|
# Fallback to route name when no request is available
|
||||||
return f"/{link.route_name.replace('.', '/')}"
|
return f"/{link.route_name.replace('.', '/')}"
|
||||||
if link.slug in {"profitability", "profitability-calculator"}:
|
requires_context = link.slug in {
|
||||||
|
"profitability",
|
||||||
|
"profitability-calculator",
|
||||||
|
"opex",
|
||||||
|
"capex",
|
||||||
|
}
|
||||||
|
if requires_context:
|
||||||
project_id = context.get("project_id")
|
project_id = context.get("project_id")
|
||||||
scenario_id = context.get("scenario_id")
|
scenario_id = context.get("scenario_id")
|
||||||
if project_id and scenario_id:
|
if project_id and scenario_id:
|
||||||
try:
|
try:
|
||||||
return request.url_for(
|
return str(
|
||||||
link.route_name,
|
request.url_for(
|
||||||
project_id=project_id,
|
link.route_name,
|
||||||
scenario_id=scenario_id,
|
project_id=project_id,
|
||||||
|
scenario_id=scenario_id,
|
||||||
|
)
|
||||||
)
|
)
|
||||||
except Exception: # pragma: no cover - defensive
|
except Exception: # pragma: no cover - defensive
|
||||||
pass
|
pass
|
||||||
try:
|
try:
|
||||||
return request.url_for(link.route_name)
|
return str(request.url_for(link.route_name))
|
||||||
except Exception: # pragma: no cover - defensive
|
except Exception: # pragma: no cover - defensive
|
||||||
return link.href_override
|
return link.href_override
|
||||||
return link.href_override
|
return link.href_override
|
||||||
|
|||||||
@@ -13,14 +13,15 @@
|
|||||||
</nav>
|
</nav>
|
||||||
|
|
||||||
<header class="page-header">
|
<header class="page-header">
|
||||||
{% set profitability_href = '/calculations/profitability' %}
|
|
||||||
{% set opex_href = url_for('calculations.opex_form') %}
|
|
||||||
{% set capex_href = url_for('calculations.capex_form') %}
|
|
||||||
{% set scenario_list_href = url_for('scenarios.project_scenario_list', project_id=project.id) %}
|
{% set scenario_list_href = url_for('scenarios.project_scenario_list', project_id=project.id) %}
|
||||||
{% if project and scenario %}
|
{% if project and scenario %}
|
||||||
{% set profitability_href = url_for('calculations.profitability_form', project_id=project.id, scenario_id=scenario.id) %}
|
{% set profitability_href = url_for('calculations.profitability_form', project_id=project.id, scenario_id=scenario.id) %}
|
||||||
{% set opex_href = opex_href ~ '?project_id=' ~ project.id ~ '&scenario_id=' ~ scenario.id %}
|
{% set opex_href = url_for('calculations.scenario_opex_form', project_id=project.id, scenario_id=scenario.id) %}
|
||||||
{% set capex_href = capex_href ~ '?project_id=' ~ project.id ~ '&scenario_id=' ~ scenario.id %}
|
{% set capex_href = url_for('calculations.scenario_capex_form', project_id=project.id, scenario_id=scenario.id) %}
|
||||||
|
{% else %}
|
||||||
|
{% set profitability_href = url_for('calculations.profitability_form') %}
|
||||||
|
{% set opex_href = url_for('calculations.opex_form_legacy') %}
|
||||||
|
{% set capex_href = url_for('calculations.capex_form_legacy') %}
|
||||||
{% endif %}
|
{% endif %}
|
||||||
<div>
|
<div>
|
||||||
<h1>{{ scenario.name }}</h1>
|
<h1>{{ scenario.name }}</h1>
|
||||||
|
|||||||
@@ -97,8 +97,8 @@
|
|||||||
<ul class="scenario-list">
|
<ul class="scenario-list">
|
||||||
{% for scenario in scenarios %}
|
{% for scenario in scenarios %}
|
||||||
{% set profitability_href = url_for('calculations.profitability_form', project_id=project.id, scenario_id=scenario.id) %}
|
{% set profitability_href = url_for('calculations.profitability_form', project_id=project.id, scenario_id=scenario.id) %}
|
||||||
{% set opex_href = url_for('calculations.opex_form') ~ '?project_id=' ~ project.id ~ '&scenario_id=' ~ scenario.id %}
|
{% set opex_href = url_for('calculations.scenario_opex_form', project_id=project.id, scenario_id=scenario.id) %}
|
||||||
{% set capex_href = url_for('calculations.capex_form') ~ '?project_id=' ~ project.id ~ '&scenario_id=' ~ scenario.id %}
|
{% set capex_href = url_for('calculations.scenario_capex_form', project_id=project.id, scenario_id=scenario.id) %}
|
||||||
<li class="scenario-item">
|
<li class="scenario-item">
|
||||||
<div class="scenario-item__body">
|
<div class="scenario-item__body">
|
||||||
<div class="scenario-item__header">
|
<div class="scenario-item__header">
|
||||||
|
|||||||
@@ -18,6 +18,7 @@ from models import User
|
|||||||
from routes.auth import router as auth_router
|
from routes.auth import router as auth_router
|
||||||
from routes.dashboard import router as dashboard_router
|
from routes.dashboard import router as dashboard_router
|
||||||
from routes.calculations import router as calculations_router
|
from routes.calculations import router as calculations_router
|
||||||
|
from routes.navigation import router as navigation_router
|
||||||
from routes.projects import router as projects_router
|
from routes.projects import router as projects_router
|
||||||
from routes.scenarios import router as scenarios_router
|
from routes.scenarios import router as scenarios_router
|
||||||
from routes.imports import router as imports_router
|
from routes.imports import router as imports_router
|
||||||
@@ -29,6 +30,11 @@ from services.unit_of_work import UnitOfWork
|
|||||||
from services.session import AuthSession, SessionTokens
|
from services.session import AuthSession, SessionTokens
|
||||||
from tests.utils.security import random_password, random_token
|
from tests.utils.security import random_password, random_token
|
||||||
|
|
||||||
|
BASE_TESTSERVER_URL = "http://testserver"
|
||||||
|
|
||||||
|
|
||||||
|
TEST_USER_HEADER = "X-Test-User"
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture()
|
@pytest.fixture()
|
||||||
def engine() -> Iterator[Engine]:
|
def engine() -> Iterator[Engine]:
|
||||||
@@ -60,6 +66,7 @@ def app(session_factory: sessionmaker) -> FastAPI:
|
|||||||
application.include_router(dashboard_router)
|
application.include_router(dashboard_router)
|
||||||
application.include_router(calculations_router)
|
application.include_router(calculations_router)
|
||||||
application.include_router(projects_router)
|
application.include_router(projects_router)
|
||||||
|
application.include_router(navigation_router)
|
||||||
application.include_router(scenarios_router)
|
application.include_router(scenarios_router)
|
||||||
application.include_router(imports_router)
|
application.include_router(imports_router)
|
||||||
application.include_router(exports_router)
|
application.include_router(exports_router)
|
||||||
@@ -85,26 +92,64 @@ def app(session_factory: sessionmaker) -> FastAPI:
|
|||||||
] = _override_ingestion_service
|
] = _override_ingestion_service
|
||||||
|
|
||||||
with UnitOfWork(session_factory=session_factory) as uow:
|
with UnitOfWork(session_factory=session_factory) as uow:
|
||||||
assert uow.users is not None
|
assert uow.users is not None and uow.roles is not None
|
||||||
uow.ensure_default_roles()
|
roles = {role.name: role for role in uow.ensure_default_roles()}
|
||||||
user = User(
|
admin_user = User(
|
||||||
email="test-superuser@example.com",
|
email="test-superuser@example.com",
|
||||||
username="test-superuser",
|
username="test-superuser",
|
||||||
password_hash=User.hash_password(random_password()),
|
password_hash=User.hash_password(random_password()),
|
||||||
is_active=True,
|
is_active=True,
|
||||||
is_superuser=True,
|
is_superuser=True,
|
||||||
)
|
)
|
||||||
uow.users.create(user)
|
viewer_user = User(
|
||||||
user = uow.users.get(user.id, with_roles=True)
|
email="test-viewer@example.com",
|
||||||
|
username="test-viewer",
|
||||||
|
password_hash=User.hash_password(random_password()),
|
||||||
|
is_active=True,
|
||||||
|
is_superuser=False,
|
||||||
|
)
|
||||||
|
uow.users.create(admin_user)
|
||||||
|
uow.users.create(viewer_user)
|
||||||
|
uow.users.assign_role(
|
||||||
|
user_id=admin_user.id,
|
||||||
|
role_id=roles["admin"].id,
|
||||||
|
granted_by=admin_user.id,
|
||||||
|
)
|
||||||
|
uow.users.assign_role(
|
||||||
|
user_id=viewer_user.id,
|
||||||
|
role_id=roles["viewer"].id,
|
||||||
|
granted_by=admin_user.id,
|
||||||
|
)
|
||||||
|
admin_user = uow.users.get(admin_user.id, with_roles=True)
|
||||||
|
viewer_user = uow.users.get(viewer_user.id, with_roles=True)
|
||||||
|
|
||||||
|
application.state.test_users = {
|
||||||
|
"admin": admin_user,
|
||||||
|
"viewer": viewer_user,
|
||||||
|
}
|
||||||
|
|
||||||
|
def _resolve_user(alias: str) -> tuple[User, tuple[str, ...]]:
|
||||||
|
normalised = alias.strip().lower()
|
||||||
|
user = application.state.test_users.get(normalised)
|
||||||
|
if user is None:
|
||||||
|
raise ValueError(f"Unknown test user alias: {alias}")
|
||||||
|
roles = tuple(role.name for role in user.roles)
|
||||||
|
return user, roles
|
||||||
|
|
||||||
def _override_auth_session(request: Request) -> AuthSession:
|
def _override_auth_session(request: Request) -> AuthSession:
|
||||||
session = AuthSession(
|
alias = request.headers.get(TEST_USER_HEADER, "admin").strip().lower()
|
||||||
tokens=SessionTokens(
|
if alias == "anonymous":
|
||||||
access_token=random_token(),
|
session = AuthSession.anonymous()
|
||||||
refresh_token=random_token(),
|
else:
|
||||||
|
user, role_slugs = _resolve_user(alias or "admin")
|
||||||
|
session = AuthSession(
|
||||||
|
tokens=SessionTokens(
|
||||||
|
access_token=random_token(),
|
||||||
|
refresh_token=random_token(),
|
||||||
|
),
|
||||||
|
user=user,
|
||||||
)
|
)
|
||||||
)
|
session.set_role_slugs(role_slugs)
|
||||||
session.user = user
|
|
||||||
request.state.auth_session = session
|
request.state.auth_session = session
|
||||||
return session
|
return session
|
||||||
|
|
||||||
@@ -114,7 +159,7 @@ def app(session_factory: sessionmaker) -> FastAPI:
|
|||||||
|
|
||||||
@pytest.fixture()
|
@pytest.fixture()
|
||||||
def client(app: FastAPI) -> Iterator[TestClient]:
|
def client(app: FastAPI) -> Iterator[TestClient]:
|
||||||
test_client = TestClient(app)
|
test_client = TestClient(app, headers={TEST_USER_HEADER: "admin"})
|
||||||
try:
|
try:
|
||||||
yield test_client
|
yield test_client
|
||||||
finally:
|
finally:
|
||||||
@@ -124,13 +169,52 @@ def client(app: FastAPI) -> Iterator[TestClient]:
|
|||||||
@pytest_asyncio.fixture()
|
@pytest_asyncio.fixture()
|
||||||
async def async_client(app: FastAPI) -> AsyncClient:
|
async def async_client(app: FastAPI) -> AsyncClient:
|
||||||
return AsyncClient(
|
return AsyncClient(
|
||||||
transport=ASGITransport(app=app), base_url="http://testserver"
|
transport=ASGITransport(app=app),
|
||||||
|
base_url="http://testserver",
|
||||||
|
headers={TEST_USER_HEADER: "admin"},
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture()
|
||||||
|
def test_user_headers() -> Callable[[str | None], dict[str, str]]:
|
||||||
|
def _factory(alias: str | None = "admin") -> dict[str, str]:
|
||||||
|
if alias is None:
|
||||||
|
return {}
|
||||||
|
return {TEST_USER_HEADER: alias.lower()}
|
||||||
|
|
||||||
|
return _factory
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture()
|
@pytest.fixture()
|
||||||
def unit_of_work_factory(session_factory: sessionmaker) -> Callable[[], UnitOfWork]:
|
def unit_of_work_factory(session_factory: sessionmaker) -> Callable[[], UnitOfWork]:
|
||||||
def _factory() -> UnitOfWork:
|
def _factory() -> UnitOfWork:
|
||||||
return UnitOfWork(session_factory=session_factory)
|
return UnitOfWork(session_factory=session_factory)
|
||||||
|
|
||||||
return _factory
|
return _factory
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture()
|
||||||
|
def app_url_for(app: FastAPI) -> Callable[..., str]:
|
||||||
|
def _builder(route_name: str, **path_params: object) -> str:
|
||||||
|
normalised_params = {
|
||||||
|
key: str(value)
|
||||||
|
for key, value in path_params.items()
|
||||||
|
if value is not None
|
||||||
|
}
|
||||||
|
return f"{BASE_TESTSERVER_URL}{app.url_path_for(route_name, **normalised_params)}"
|
||||||
|
|
||||||
|
return _builder
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture()
|
||||||
|
def scenario_calculation_url(
|
||||||
|
app_url_for: Callable[..., str]
|
||||||
|
) -> Callable[[str, int, int], str]:
|
||||||
|
def _builder(route_name: str, project_id: int, scenario_id: int) -> str:
|
||||||
|
return app_url_for(
|
||||||
|
route_name,
|
||||||
|
project_id=project_id,
|
||||||
|
scenario_id=scenario_id,
|
||||||
|
)
|
||||||
|
|
||||||
|
return _builder
|
||||||
|
|||||||
141
tests/integration/test_calculations_legacy_routes.py
Normal file
141
tests/integration/test_calculations_legacy_routes.py
Normal file
@@ -0,0 +1,141 @@
|
|||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from collections.abc import Callable
|
||||||
|
|
||||||
|
from fastapi.testclient import TestClient
|
||||||
|
|
||||||
|
|
||||||
|
def _create_project(client: TestClient, name: str) -> int:
|
||||||
|
response = client.post(
|
||||||
|
"/projects",
|
||||||
|
json={
|
||||||
|
"name": name,
|
||||||
|
"location": "Western Australia",
|
||||||
|
"operation_type": "open_pit",
|
||||||
|
"description": "Legacy calculations redirect test project",
|
||||||
|
},
|
||||||
|
)
|
||||||
|
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": "Scenario for legacy calculations redirect tests",
|
||||||
|
"status": "draft",
|
||||||
|
"currency": "usd",
|
||||||
|
"primary_resource": "diesel",
|
||||||
|
},
|
||||||
|
)
|
||||||
|
assert response.status_code == 201
|
||||||
|
return response.json()["id"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_legacy_opex_redirects_to_scenario_route(
|
||||||
|
client: TestClient,
|
||||||
|
scenario_calculation_url: Callable[[str, int, int], str],
|
||||||
|
) -> None:
|
||||||
|
project_id = _create_project(client, "Opex Legacy Redirect Project")
|
||||||
|
scenario_id = _create_scenario(
|
||||||
|
client, project_id, "Opex Legacy Redirect Scenario")
|
||||||
|
|
||||||
|
response = client.get(
|
||||||
|
f"/calculations/opex?project_id={project_id}&scenario_id={scenario_id}",
|
||||||
|
follow_redirects=False,
|
||||||
|
)
|
||||||
|
|
||||||
|
assert response.status_code == 308
|
||||||
|
assert response.headers["location"] == scenario_calculation_url(
|
||||||
|
"calculations.scenario_opex_form",
|
||||||
|
project_id,
|
||||||
|
scenario_id,
|
||||||
|
)
|
||||||
|
|
||||||
|
post_response = client.post(
|
||||||
|
f"/calculations/opex?project_id={project_id}&scenario_id={scenario_id}",
|
||||||
|
data={},
|
||||||
|
follow_redirects=False,
|
||||||
|
)
|
||||||
|
|
||||||
|
assert post_response.status_code == 308
|
||||||
|
assert post_response.headers["location"] == scenario_calculation_url(
|
||||||
|
"calculations.scenario_opex_submit",
|
||||||
|
project_id,
|
||||||
|
scenario_id,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_legacy_capex_redirects_to_scenario_route(
|
||||||
|
client: TestClient,
|
||||||
|
scenario_calculation_url: Callable[[str, int, int], str],
|
||||||
|
) -> None:
|
||||||
|
project_id = _create_project(client, "Capex Legacy Redirect Project")
|
||||||
|
scenario_id = _create_scenario(
|
||||||
|
client, project_id, "Capex Legacy Redirect Scenario")
|
||||||
|
|
||||||
|
response = client.get(
|
||||||
|
f"/calculations/capex?project_id={project_id}&scenario_id={scenario_id}",
|
||||||
|
follow_redirects=False,
|
||||||
|
)
|
||||||
|
|
||||||
|
assert response.status_code == 308
|
||||||
|
assert response.headers["location"] == scenario_calculation_url(
|
||||||
|
"calculations.scenario_capex_form",
|
||||||
|
project_id,
|
||||||
|
scenario_id,
|
||||||
|
)
|
||||||
|
|
||||||
|
post_response = client.post(
|
||||||
|
f"/calculations/capex?project_id={project_id}&scenario_id={scenario_id}",
|
||||||
|
data={},
|
||||||
|
follow_redirects=False,
|
||||||
|
)
|
||||||
|
|
||||||
|
assert post_response.status_code == 308
|
||||||
|
assert post_response.headers["location"] == scenario_calculation_url(
|
||||||
|
"calculations.scenario_capex_submit",
|
||||||
|
project_id,
|
||||||
|
scenario_id,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_legacy_opex_redirects_to_project_scenarios_when_only_project(
|
||||||
|
client: TestClient,
|
||||||
|
app_url_for: Callable[..., str],
|
||||||
|
) -> None:
|
||||||
|
project_id = _create_project(client, "Opex Legacy Project Only")
|
||||||
|
|
||||||
|
response = client.get(
|
||||||
|
f"/calculations/opex?project_id={project_id}",
|
||||||
|
follow_redirects=False,
|
||||||
|
)
|
||||||
|
|
||||||
|
assert response.status_code == 303
|
||||||
|
assert response.headers["location"] == app_url_for(
|
||||||
|
"scenarios.project_scenario_list", project_id=project_id
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_legacy_capex_rejects_invalid_identifiers(client: TestClient) -> None:
|
||||||
|
response = client.get(
|
||||||
|
"/calculations/capex?project_id=abc&scenario_id=-10",
|
||||||
|
follow_redirects=False,
|
||||||
|
)
|
||||||
|
|
||||||
|
assert response.status_code == 400
|
||||||
|
assert "project_id" in response.json()["detail"].lower()
|
||||||
|
|
||||||
|
|
||||||
|
def test_legacy_opex_returns_not_found_for_missing_entities(client: TestClient) -> None:
|
||||||
|
project_id = _create_project(client, "Opex Legacy Missing Scenario")
|
||||||
|
|
||||||
|
response = client.get(
|
||||||
|
f"/calculations/opex?project_id={project_id}&scenario_id=999999",
|
||||||
|
follow_redirects=False,
|
||||||
|
)
|
||||||
|
|
||||||
|
assert response.status_code == 404
|
||||||
|
assert response.json()["detail"] == "Scenario not found"
|
||||||
146
tests/integration/test_navigation_sidebar.py
Normal file
146
tests/integration/test_navigation_sidebar.py
Normal file
@@ -0,0 +1,146 @@
|
|||||||
|
"""Integration coverage for the /navigation/sidebar endpoint.
|
||||||
|
|
||||||
|
These tests validate role-based filtering, ordering, and disabled-link handling
|
||||||
|
through the full FastAPI stack so future changes keep navigation behaviour under
|
||||||
|
explicit test coverage.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from collections.abc import Callable
|
||||||
|
from typing import Any, Mapping
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
from fastapi.testclient import TestClient
|
||||||
|
|
||||||
|
from models.navigation import NavigationGroup, NavigationLink
|
||||||
|
from services.unit_of_work import UnitOfWork
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture()
|
||||||
|
def seed_navigation(unit_of_work_factory: Callable[[], UnitOfWork]) -> Callable[[], None]:
|
||||||
|
def _seed() -> None:
|
||||||
|
with unit_of_work_factory() as uow:
|
||||||
|
repo = uow.navigation
|
||||||
|
assert repo is not None
|
||||||
|
|
||||||
|
workspace = repo.add_group(
|
||||||
|
NavigationGroup(
|
||||||
|
slug="workspace",
|
||||||
|
label="Workspace",
|
||||||
|
sort_order=10,
|
||||||
|
)
|
||||||
|
)
|
||||||
|
insights = repo.add_group(
|
||||||
|
NavigationGroup(
|
||||||
|
slug="insights",
|
||||||
|
label="Insights",
|
||||||
|
sort_order=20,
|
||||||
|
)
|
||||||
|
)
|
||||||
|
|
||||||
|
repo.add_link(
|
||||||
|
NavigationLink(
|
||||||
|
group_id=workspace.id,
|
||||||
|
slug="projects",
|
||||||
|
label="Projects",
|
||||||
|
href_override="/projects",
|
||||||
|
sort_order=5,
|
||||||
|
required_roles=[],
|
||||||
|
)
|
||||||
|
)
|
||||||
|
repo.add_link(
|
||||||
|
NavigationLink(
|
||||||
|
group_id=workspace.id,
|
||||||
|
slug="admin-tools",
|
||||||
|
label="Admin Tools",
|
||||||
|
href_override="/admin/tools",
|
||||||
|
sort_order=10,
|
||||||
|
required_roles=["admin"],
|
||||||
|
)
|
||||||
|
)
|
||||||
|
repo.add_link(
|
||||||
|
NavigationLink(
|
||||||
|
group_id=workspace.id,
|
||||||
|
slug="disabled-link",
|
||||||
|
label="Hidden",
|
||||||
|
href_override="/hidden",
|
||||||
|
sort_order=15,
|
||||||
|
required_roles=[],
|
||||||
|
is_enabled=False,
|
||||||
|
)
|
||||||
|
)
|
||||||
|
repo.add_link(
|
||||||
|
NavigationLink(
|
||||||
|
group_id=insights.id,
|
||||||
|
slug="reports",
|
||||||
|
label="Reports",
|
||||||
|
href_override="/reports",
|
||||||
|
sort_order=1,
|
||||||
|
required_roles=[],
|
||||||
|
)
|
||||||
|
)
|
||||||
|
|
||||||
|
return _seed
|
||||||
|
|
||||||
|
|
||||||
|
def _link_labels(group_json: Mapping[str, Any]) -> list[str]:
|
||||||
|
return [link["label"] for link in group_json["links"]]
|
||||||
|
|
||||||
|
|
||||||
|
def test_admin_session_receives_all_enabled_links(
|
||||||
|
client: TestClient,
|
||||||
|
seed_navigation: Callable[[], None],
|
||||||
|
) -> None:
|
||||||
|
seed_navigation()
|
||||||
|
|
||||||
|
response = client.get("/navigation/sidebar")
|
||||||
|
assert response.status_code == 200
|
||||||
|
payload = response.json()
|
||||||
|
|
||||||
|
assert [group["label"] for group in payload["groups"]] == [
|
||||||
|
"Workspace",
|
||||||
|
"Insights",
|
||||||
|
]
|
||||||
|
workspace, insights = payload["groups"]
|
||||||
|
assert _link_labels(workspace) == ["Projects", "Admin Tools"]
|
||||||
|
assert _link_labels(insights) == ["Reports"]
|
||||||
|
assert payload["roles"] == ["admin"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_viewer_session_filters_admin_only_links(
|
||||||
|
client: TestClient,
|
||||||
|
seed_navigation: Callable[[], None],
|
||||||
|
test_user_headers: Callable[[str | None], dict[str, str]],
|
||||||
|
) -> None:
|
||||||
|
seed_navigation()
|
||||||
|
|
||||||
|
response = client.get(
|
||||||
|
"/navigation/sidebar",
|
||||||
|
headers=test_user_headers("viewer"),
|
||||||
|
)
|
||||||
|
assert response.status_code == 200
|
||||||
|
payload = response.json()
|
||||||
|
|
||||||
|
assert [group["label"] for group in payload["groups"]] == [
|
||||||
|
"Workspace",
|
||||||
|
"Insights",
|
||||||
|
]
|
||||||
|
workspace, insights = payload["groups"]
|
||||||
|
assert _link_labels(workspace) == ["Projects"]
|
||||||
|
assert _link_labels(insights) == ["Reports"]
|
||||||
|
assert payload["roles"] == ["viewer"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_anonymous_access_is_rejected(
|
||||||
|
client: TestClient,
|
||||||
|
seed_navigation: Callable[[], None],
|
||||||
|
test_user_headers: Callable[[str | None], dict[str, str]],
|
||||||
|
) -> None:
|
||||||
|
seed_navigation()
|
||||||
|
|
||||||
|
response = client.get(
|
||||||
|
"/navigation/sidebar",
|
||||||
|
headers=test_user_headers("anonymous"),
|
||||||
|
)
|
||||||
|
assert response.status_code == 401
|
||||||
170
tests/integration/test_navigation_sidebar_calculations.py
Normal file
170
tests/integration/test_navigation_sidebar_calculations.py
Normal file
@@ -0,0 +1,170 @@
|
|||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from collections.abc import Callable
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
from fastapi.testclient import TestClient
|
||||||
|
|
||||||
|
from models.navigation import NavigationGroup, NavigationLink
|
||||||
|
from services.unit_of_work import UnitOfWork
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture()
|
||||||
|
def seed_calculation_navigation(
|
||||||
|
unit_of_work_factory: Callable[[], UnitOfWork]
|
||||||
|
) -> Callable[[], None]:
|
||||||
|
def _seed() -> None:
|
||||||
|
with unit_of_work_factory() as uow:
|
||||||
|
repo = uow.navigation
|
||||||
|
assert repo is not None
|
||||||
|
|
||||||
|
workspace = repo.add_group(
|
||||||
|
NavigationGroup(
|
||||||
|
slug="workspace",
|
||||||
|
label="Workspace",
|
||||||
|
sort_order=10,
|
||||||
|
)
|
||||||
|
)
|
||||||
|
|
||||||
|
projects_link = repo.add_link(
|
||||||
|
NavigationLink(
|
||||||
|
group_id=workspace.id,
|
||||||
|
slug="projects",
|
||||||
|
label="Projects",
|
||||||
|
href_override="/projects",
|
||||||
|
sort_order=5,
|
||||||
|
required_roles=[],
|
||||||
|
)
|
||||||
|
)
|
||||||
|
repo.add_link(
|
||||||
|
NavigationLink(
|
||||||
|
group_id=workspace.id,
|
||||||
|
parent_link_id=projects_link.id,
|
||||||
|
slug="profitability",
|
||||||
|
label="Profitability Calculator",
|
||||||
|
route_name="calculations.profitability_form",
|
||||||
|
href_override="/calculations/profitability",
|
||||||
|
match_prefix="/calculations/profitability",
|
||||||
|
sort_order=8,
|
||||||
|
required_roles=["analyst", "admin"],
|
||||||
|
)
|
||||||
|
)
|
||||||
|
repo.add_link(
|
||||||
|
NavigationLink(
|
||||||
|
group_id=workspace.id,
|
||||||
|
parent_link_id=projects_link.id,
|
||||||
|
slug="opex",
|
||||||
|
label="Opex Planner",
|
||||||
|
route_name="calculations.opex_form",
|
||||||
|
href_override="/calculations/opex",
|
||||||
|
match_prefix="/calculations/opex",
|
||||||
|
sort_order=10,
|
||||||
|
required_roles=["analyst", "admin"],
|
||||||
|
)
|
||||||
|
)
|
||||||
|
repo.add_link(
|
||||||
|
NavigationLink(
|
||||||
|
group_id=workspace.id,
|
||||||
|
parent_link_id=projects_link.id,
|
||||||
|
slug="capex",
|
||||||
|
label="Capex Planner",
|
||||||
|
route_name="calculations.capex_form",
|
||||||
|
href_override="/calculations/capex",
|
||||||
|
match_prefix="/calculations/capex",
|
||||||
|
sort_order=15,
|
||||||
|
required_roles=["analyst", "admin"],
|
||||||
|
)
|
||||||
|
)
|
||||||
|
|
||||||
|
return _seed
|
||||||
|
|
||||||
|
|
||||||
|
def test_navigation_sidebar_includes_calculation_links_for_admin(
|
||||||
|
client: TestClient,
|
||||||
|
seed_calculation_navigation: Callable[[], None],
|
||||||
|
) -> None:
|
||||||
|
seed_calculation_navigation()
|
||||||
|
|
||||||
|
response = client.get("/navigation/sidebar")
|
||||||
|
assert response.status_code == 200
|
||||||
|
payload = response.json()
|
||||||
|
|
||||||
|
groups = payload["groups"]
|
||||||
|
assert groups
|
||||||
|
workspace = next(
|
||||||
|
group for group in groups if group["label"] == "Workspace")
|
||||||
|
workspace_links = workspace["links"]
|
||||||
|
assert [link["label"] for link in workspace_links] == ["Projects"]
|
||||||
|
|
||||||
|
projects_children = workspace_links[0]["children"]
|
||||||
|
child_labels = [link["label"] for link in projects_children]
|
||||||
|
assert child_labels == [
|
||||||
|
"Profitability Calculator",
|
||||||
|
"Opex Planner",
|
||||||
|
"Capex Planner",
|
||||||
|
]
|
||||||
|
|
||||||
|
profitability_link = next(
|
||||||
|
link for link in projects_children if link["label"] == "Profitability Calculator")
|
||||||
|
assert profitability_link["href"] == "/calculations/profitability"
|
||||||
|
assert profitability_link["match_prefix"] == "/calculations/profitability"
|
||||||
|
|
||||||
|
opex_link = next(
|
||||||
|
link for link in projects_children if link["label"] == "Opex Planner")
|
||||||
|
assert opex_link["href"] == "/calculations/opex"
|
||||||
|
assert opex_link["match_prefix"] == "/calculations/opex"
|
||||||
|
|
||||||
|
capex_link = next(
|
||||||
|
link for link in projects_children if link["label"] == "Capex Planner")
|
||||||
|
assert capex_link["href"] == "/calculations/capex"
|
||||||
|
assert capex_link["match_prefix"] == "/calculations/capex"
|
||||||
|
assert payload["roles"] == ["admin"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_navigation_sidebar_hides_calculation_links_for_viewer_without_role(
|
||||||
|
client: TestClient,
|
||||||
|
seed_calculation_navigation: Callable[[], None],
|
||||||
|
test_user_headers: Callable[[str | None], dict[str, str]],
|
||||||
|
) -> None:
|
||||||
|
seed_calculation_navigation()
|
||||||
|
|
||||||
|
response = client.get(
|
||||||
|
"/navigation/sidebar",
|
||||||
|
headers=test_user_headers("viewer"),
|
||||||
|
)
|
||||||
|
assert response.status_code == 200
|
||||||
|
payload = response.json()
|
||||||
|
|
||||||
|
groups = payload["groups"]
|
||||||
|
assert groups
|
||||||
|
workspace = next(
|
||||||
|
group for group in groups if group["label"] == "Workspace")
|
||||||
|
workspace_links = workspace["links"]
|
||||||
|
assert [link["label"] for link in workspace_links] == ["Projects"]
|
||||||
|
assert workspace_links[0]["children"] == []
|
||||||
|
assert payload["roles"] == ["viewer"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_navigation_sidebar_includes_contextual_urls_when_ids_provided(
|
||||||
|
client: TestClient,
|
||||||
|
seed_calculation_navigation: Callable[[], None],
|
||||||
|
) -> None:
|
||||||
|
seed_calculation_navigation()
|
||||||
|
|
||||||
|
response = client.get(
|
||||||
|
"/navigation/sidebar",
|
||||||
|
params={"project_id": "5", "scenario_id": "11"},
|
||||||
|
)
|
||||||
|
assert response.status_code == 200
|
||||||
|
payload = response.json()
|
||||||
|
|
||||||
|
workspace = next(
|
||||||
|
group for group in payload["groups"] if group["label"] == "Workspace")
|
||||||
|
projects = workspace["links"][0]
|
||||||
|
capex_link = next(
|
||||||
|
link for link in projects["children"] if link["label"] == "Capex Planner")
|
||||||
|
|
||||||
|
assert capex_link["href"].endswith(
|
||||||
|
"/calculations/projects/5/scenarios/11/calculations/capex"
|
||||||
|
)
|
||||||
|
assert capex_link["match_prefix"] == "/calculations/capex"
|
||||||
@@ -1,10 +1,16 @@
|
|||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from collections.abc import Callable
|
||||||
|
|
||||||
from fastapi.testclient import TestClient
|
from fastapi.testclient import TestClient
|
||||||
|
|
||||||
|
|
||||||
class TestScenarioLifecycle:
|
class TestScenarioLifecycle:
|
||||||
def test_scenario_lifecycle_flow(self, client: TestClient) -> None:
|
def test_scenario_lifecycle_flow(
|
||||||
|
self,
|
||||||
|
client: TestClient,
|
||||||
|
scenario_calculation_url: Callable[[str, int, int], str],
|
||||||
|
) -> None:
|
||||||
# Create a project to attach scenarios to
|
# Create a project to attach scenarios to
|
||||||
project_response = client.post(
|
project_response = client.post(
|
||||||
"/projects",
|
"/projects",
|
||||||
@@ -36,7 +42,7 @@ class TestScenarioLifecycle:
|
|||||||
project_detail = client.get(f"/projects/{project_id}/view")
|
project_detail = client.get(f"/projects/{project_id}/view")
|
||||||
assert project_detail.status_code == 200
|
assert project_detail.status_code == 200
|
||||||
assert "Lifecycle Scenario" in project_detail.text
|
assert "Lifecycle Scenario" in project_detail.text
|
||||||
assert "<td>Draft</td>" in project_detail.text
|
assert '<span class="status-pill status-pill--draft">Draft</span>' in project_detail.text
|
||||||
|
|
||||||
# Update the scenario through the HTML form
|
# Update the scenario through the HTML form
|
||||||
form_response = client.post(
|
form_response = client.post(
|
||||||
@@ -61,16 +67,28 @@ class TestScenarioLifecycle:
|
|||||||
scenario_detail = client.get(f"/scenarios/{scenario_id}/view")
|
scenario_detail = client.get(f"/scenarios/{scenario_id}/view")
|
||||||
assert scenario_detail.status_code == 200
|
assert scenario_detail.status_code == 200
|
||||||
assert "Lifecycle Scenario Revised" in scenario_detail.text
|
assert "Lifecycle Scenario Revised" in scenario_detail.text
|
||||||
assert "Status: Active" in scenario_detail.text
|
assert "<p class=\"metric-value status-pill status-pill--active\">Active</p>" in scenario_detail.text
|
||||||
assert "CAD" in scenario_detail.text
|
assert "CAD" in scenario_detail.text
|
||||||
assert "Electricity" in scenario_detail.text
|
assert "Electricity" in scenario_detail.text
|
||||||
assert "Revised scenario assumptions" in scenario_detail.text
|
assert "Revised scenario assumptions" in scenario_detail.text
|
||||||
|
scenario_opex_url = scenario_calculation_url(
|
||||||
|
"calculations.scenario_opex_form",
|
||||||
|
project_id,
|
||||||
|
scenario_id,
|
||||||
|
)
|
||||||
|
scenario_capex_url = scenario_calculation_url(
|
||||||
|
"calculations.scenario_capex_form",
|
||||||
|
project_id,
|
||||||
|
scenario_id,
|
||||||
|
)
|
||||||
|
assert f'href="{scenario_opex_url}"' in scenario_detail.text
|
||||||
|
assert f'href="{scenario_capex_url}"' in scenario_detail.text
|
||||||
|
|
||||||
# Project detail page should show the scenario as active with updated currency/resource
|
# Project detail page should show the scenario as active with updated currency/resource
|
||||||
project_detail = client.get(f"/projects/{project_id}/view")
|
project_detail = client.get(f"/projects/{project_id}/view")
|
||||||
assert "<td>Active</td>" in project_detail.text
|
assert '<span class="status-pill status-pill--active">Active</span>' in project_detail.text
|
||||||
assert "<td>CAD</td>" in project_detail.text
|
assert 'CAD' in project_detail.text
|
||||||
assert "<td>Electricity</td>" in project_detail.text
|
assert 'Electricity' in project_detail.text
|
||||||
|
|
||||||
# Attempt to update the scenario with invalid currency to trigger validation error
|
# Attempt to update the scenario with invalid currency to trigger validation error
|
||||||
invalid_update = client.put(
|
invalid_update = client.put(
|
||||||
@@ -84,6 +102,8 @@ class TestScenarioLifecycle:
|
|||||||
# Scenario detail should still show the previous (valid) currency
|
# Scenario detail should still show the previous (valid) currency
|
||||||
scenario_detail = client.get(f"/scenarios/{scenario_id}/view")
|
scenario_detail = client.get(f"/scenarios/{scenario_id}/view")
|
||||||
assert "CAD" in scenario_detail.text
|
assert "CAD" in scenario_detail.text
|
||||||
|
assert f'href="{scenario_opex_url}"' in scenario_detail.text
|
||||||
|
assert f'href="{scenario_capex_url}"' in scenario_detail.text
|
||||||
|
|
||||||
# Archive the scenario through the API
|
# Archive the scenario through the API
|
||||||
archive_response = client.put(
|
archive_response = client.put(
|
||||||
@@ -95,10 +115,18 @@ class TestScenarioLifecycle:
|
|||||||
|
|
||||||
# Scenario detail reflects archived status
|
# Scenario detail reflects archived status
|
||||||
scenario_detail = client.get(f"/scenarios/{scenario_id}/view")
|
scenario_detail = client.get(f"/scenarios/{scenario_id}/view")
|
||||||
assert "Status: Archived" in scenario_detail.text
|
assert '<p class="metric-value status-pill status-pill--archived">Archived</p>' in scenario_detail.text
|
||||||
|
assert f'href="{scenario_opex_url}"' in scenario_detail.text
|
||||||
|
assert f'href="{scenario_capex_url}"' in scenario_detail.text
|
||||||
|
|
||||||
# Project detail metrics and table entries reflect the archived state
|
# Project detail metrics and table entries reflect the archived state
|
||||||
project_detail = client.get(f"/projects/{project_id}/view")
|
project_detail = client.get(f"/projects/{project_id}/view")
|
||||||
assert "<h2>Archived</h2>" in project_detail.text
|
assert "<h2>Archived</h2>" in project_detail.text
|
||||||
assert '<p class="metric-value">1</p>' in project_detail.text
|
assert '<p class="metric-value">1</p>' in project_detail.text
|
||||||
assert "<td>Archived</td>" in project_detail.text
|
assert "Archived" in project_detail.text
|
||||||
|
|
||||||
|
# Scenario portfolio view includes calculator links for each scenario entry
|
||||||
|
scenario_portfolio = client.get(f"/projects/{project_id}/scenarios/ui")
|
||||||
|
assert scenario_portfolio.status_code == 200
|
||||||
|
assert f'href="{scenario_opex_url}"' in scenario_portfolio.text
|
||||||
|
assert f'href="{scenario_capex_url}"' in scenario_portfolio.text
|
||||||
|
|||||||
146
tests/routes/test_navigation_routes.py
Normal file
146
tests/routes/test_navigation_routes.py
Normal file
@@ -0,0 +1,146 @@
|
|||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from datetime import datetime
|
||||||
|
from typing import Tuple, cast
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
from fastapi import FastAPI, HTTPException
|
||||||
|
from fastapi.testclient import TestClient
|
||||||
|
|
||||||
|
from dependencies import (
|
||||||
|
get_auth_session,
|
||||||
|
get_navigation_service,
|
||||||
|
require_authenticated_user,
|
||||||
|
)
|
||||||
|
from models import User
|
||||||
|
from routes.navigation import router as navigation_router
|
||||||
|
from services.navigation import (
|
||||||
|
NavigationGroupDTO,
|
||||||
|
NavigationLinkDTO,
|
||||||
|
NavigationService,
|
||||||
|
NavigationSidebarDTO,
|
||||||
|
)
|
||||||
|
from services.session import AuthSession, SessionTokens
|
||||||
|
|
||||||
|
|
||||||
|
class StubNavigationService:
|
||||||
|
def __init__(self, payload: NavigationSidebarDTO) -> None:
|
||||||
|
self._payload = payload
|
||||||
|
self.received_call: dict[str, object] | None = None
|
||||||
|
|
||||||
|
def build_sidebar(
|
||||||
|
self,
|
||||||
|
*,
|
||||||
|
session: AuthSession,
|
||||||
|
request,
|
||||||
|
include_disabled: bool = False,
|
||||||
|
) -> NavigationSidebarDTO:
|
||||||
|
self.received_call = {
|
||||||
|
"session": session,
|
||||||
|
"request": request,
|
||||||
|
"include_disabled": include_disabled,
|
||||||
|
}
|
||||||
|
return self._payload
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def navigation_client() -> Tuple[TestClient, StubNavigationService, AuthSession]:
|
||||||
|
app = FastAPI()
|
||||||
|
app.include_router(navigation_router)
|
||||||
|
|
||||||
|
link_dto = NavigationLinkDTO(
|
||||||
|
id=10,
|
||||||
|
label="Projects",
|
||||||
|
href="/projects",
|
||||||
|
match_prefix="/projects",
|
||||||
|
icon=None,
|
||||||
|
tooltip=None,
|
||||||
|
is_external=False,
|
||||||
|
children=[],
|
||||||
|
)
|
||||||
|
group_dto = NavigationGroupDTO(
|
||||||
|
id=5,
|
||||||
|
label="Workspace",
|
||||||
|
icon="home",
|
||||||
|
tooltip=None,
|
||||||
|
links=[link_dto],
|
||||||
|
)
|
||||||
|
payload = NavigationSidebarDTO(groups=[group_dto], roles=("viewer",))
|
||||||
|
service = StubNavigationService(payload)
|
||||||
|
|
||||||
|
user = cast(User, object())
|
||||||
|
session = AuthSession(
|
||||||
|
tokens=SessionTokens(access_token="token", refresh_token=None),
|
||||||
|
user=user,
|
||||||
|
role_slugs=("viewer",),
|
||||||
|
)
|
||||||
|
|
||||||
|
app.dependency_overrides[require_authenticated_user] = lambda: user
|
||||||
|
app.dependency_overrides[get_auth_session] = lambda: session
|
||||||
|
app.dependency_overrides[get_navigation_service] = lambda: cast(
|
||||||
|
NavigationService, service)
|
||||||
|
|
||||||
|
client = TestClient(app)
|
||||||
|
return client, service, session
|
||||||
|
|
||||||
|
|
||||||
|
def test_get_sidebar_navigation_returns_payload(
|
||||||
|
navigation_client: Tuple[TestClient, StubNavigationService, AuthSession]
|
||||||
|
) -> None:
|
||||||
|
client, service, session = navigation_client
|
||||||
|
|
||||||
|
response = client.get("/navigation/sidebar")
|
||||||
|
|
||||||
|
assert response.status_code == 200
|
||||||
|
data = response.json()
|
||||||
|
|
||||||
|
assert data["roles"] == ["viewer"]
|
||||||
|
assert data["groups"][0]["label"] == "Workspace"
|
||||||
|
assert data["groups"][0]["links"][0]["href"] == "/projects"
|
||||||
|
assert "generated_at" in data
|
||||||
|
datetime.fromisoformat(data["generated_at"])
|
||||||
|
|
||||||
|
assert service.received_call is not None
|
||||||
|
assert service.received_call["session"] is session
|
||||||
|
assert service.received_call["request"] is not None
|
||||||
|
assert service.received_call["include_disabled"] is False
|
||||||
|
|
||||||
|
|
||||||
|
def test_get_sidebar_navigation_requires_authentication() -> None:
|
||||||
|
app = FastAPI()
|
||||||
|
app.include_router(navigation_router)
|
||||||
|
|
||||||
|
link_dto = NavigationLinkDTO(
|
||||||
|
id=1,
|
||||||
|
label="Placeholder",
|
||||||
|
href="/placeholder",
|
||||||
|
match_prefix="/placeholder",
|
||||||
|
icon=None,
|
||||||
|
tooltip=None,
|
||||||
|
is_external=False,
|
||||||
|
children=[],
|
||||||
|
)
|
||||||
|
group_dto = NavigationGroupDTO(
|
||||||
|
id=1,
|
||||||
|
label="Group",
|
||||||
|
icon=None,
|
||||||
|
tooltip=None,
|
||||||
|
links=[link_dto],
|
||||||
|
)
|
||||||
|
payload = NavigationSidebarDTO(groups=[group_dto], roles=("anonymous",))
|
||||||
|
service = StubNavigationService(payload)
|
||||||
|
|
||||||
|
def _deny() -> User:
|
||||||
|
raise HTTPException(status_code=401, detail="Not authenticated")
|
||||||
|
|
||||||
|
app.dependency_overrides[get_navigation_service] = lambda: cast(
|
||||||
|
NavigationService, service)
|
||||||
|
app.dependency_overrides[get_auth_session] = AuthSession.anonymous
|
||||||
|
app.dependency_overrides[require_authenticated_user] = _deny
|
||||||
|
|
||||||
|
client = TestClient(app)
|
||||||
|
|
||||||
|
response = client.get("/navigation/sidebar")
|
||||||
|
|
||||||
|
assert response.status_code == 401
|
||||||
|
assert response.json()["detail"] == "Not authenticated"
|
||||||
@@ -35,11 +35,16 @@ class FakeState:
|
|||||||
] = field(default_factory=dict)
|
] = field(default_factory=dict)
|
||||||
financial_inputs: dict[Tuple[int, str],
|
financial_inputs: dict[Tuple[int, str],
|
||||||
Dict[str, Any]] = field(default_factory=dict)
|
Dict[str, Any]] = field(default_factory=dict)
|
||||||
|
navigation_groups: dict[str, Dict[str, Any]] = field(default_factory=dict)
|
||||||
|
navigation_links: dict[Tuple[int, str],
|
||||||
|
Dict[str, Any]] = field(default_factory=dict)
|
||||||
sequences: Dict[str, int] = field(default_factory=lambda: {
|
sequences: Dict[str, int] = field(default_factory=lambda: {
|
||||||
"users": 0,
|
"users": 0,
|
||||||
"projects": 0,
|
"projects": 0,
|
||||||
"scenarios": 0,
|
"scenarios": 0,
|
||||||
"financial_inputs": 0,
|
"financial_inputs": 0,
|
||||||
|
"navigation_groups": 0,
|
||||||
|
"navigation_links": 0,
|
||||||
})
|
})
|
||||||
|
|
||||||
|
|
||||||
@@ -50,6 +55,9 @@ class FakeResult:
|
|||||||
def fetchone(self) -> Any | None:
|
def fetchone(self) -> Any | None:
|
||||||
return self._rows[0] if self._rows else None
|
return self._rows[0] if self._rows else None
|
||||||
|
|
||||||
|
def fetchall(self) -> list[Any]:
|
||||||
|
return list(self._rows)
|
||||||
|
|
||||||
|
|
||||||
class FakeConnection:
|
class FakeConnection:
|
||||||
def __init__(self, state: FakeState) -> None:
|
def __init__(self, state: FakeState) -> None:
|
||||||
@@ -105,6 +113,13 @@ class FakeConnection:
|
|||||||
rows = [SimpleNamespace(id=record["id"])] if record else []
|
rows = [SimpleNamespace(id=record["id"])] if record else []
|
||||||
return FakeResult(rows)
|
return FakeResult(rows)
|
||||||
|
|
||||||
|
if lower_sql.startswith("select name from roles"):
|
||||||
|
rows = [
|
||||||
|
SimpleNamespace(name=record["name"])
|
||||||
|
for record in self.state.roles.values()
|
||||||
|
]
|
||||||
|
return FakeResult(rows)
|
||||||
|
|
||||||
if lower_sql.startswith("insert into user_roles"):
|
if lower_sql.startswith("insert into user_roles"):
|
||||||
key = (int(params["user_id"]), int(params["role_id"]))
|
key = (int(params["user_id"]), int(params["role_id"]))
|
||||||
self.state.user_roles.add(key)
|
self.state.user_roles.add(key)
|
||||||
@@ -171,6 +186,67 @@ class FakeConnection:
|
|||||||
rows = [SimpleNamespace(id=scenario["id"])] if scenario else []
|
rows = [SimpleNamespace(id=scenario["id"])] if scenario else []
|
||||||
return FakeResult(rows)
|
return FakeResult(rows)
|
||||||
|
|
||||||
|
if lower_sql.startswith("insert into navigation_groups"):
|
||||||
|
slug = params["slug"]
|
||||||
|
record = self.state.navigation_groups.get(slug)
|
||||||
|
if record is None:
|
||||||
|
self.state.sequences["navigation_groups"] += 1
|
||||||
|
record = {
|
||||||
|
"id": self.state.sequences["navigation_groups"],
|
||||||
|
"slug": slug,
|
||||||
|
}
|
||||||
|
record.update(
|
||||||
|
label=params["label"],
|
||||||
|
sort_order=int(params.get("sort_order", 0)),
|
||||||
|
icon=params.get("icon"),
|
||||||
|
tooltip=params.get("tooltip"),
|
||||||
|
is_enabled=bool(params.get("is_enabled", True)),
|
||||||
|
)
|
||||||
|
self.state.navigation_groups[slug] = record
|
||||||
|
return FakeResult([])
|
||||||
|
|
||||||
|
if lower_sql.startswith("select id from navigation_groups where slug"):
|
||||||
|
slug = params["slug"]
|
||||||
|
record = self.state.navigation_groups.get(slug)
|
||||||
|
rows = [SimpleNamespace(id=record["id"])] if record else []
|
||||||
|
return FakeResult(rows)
|
||||||
|
|
||||||
|
if lower_sql.startswith("insert into navigation_links"):
|
||||||
|
group_id = int(params["group_id"])
|
||||||
|
slug = params["slug"]
|
||||||
|
key = (group_id, slug)
|
||||||
|
record = self.state.navigation_links.get(key)
|
||||||
|
if record is None:
|
||||||
|
self.state.sequences["navigation_links"] += 1
|
||||||
|
record = {
|
||||||
|
"id": self.state.sequences["navigation_links"],
|
||||||
|
"group_id": group_id,
|
||||||
|
"slug": slug,
|
||||||
|
}
|
||||||
|
record.update(
|
||||||
|
parent_link_id=(int(params["parent_link_id"]) if params.get(
|
||||||
|
"parent_link_id") is not None else None),
|
||||||
|
label=params["label"],
|
||||||
|
route_name=params.get("route_name"),
|
||||||
|
href_override=params.get("href_override"),
|
||||||
|
match_prefix=params.get("match_prefix"),
|
||||||
|
sort_order=int(params.get("sort_order", 0)),
|
||||||
|
icon=params.get("icon"),
|
||||||
|
tooltip=params.get("tooltip"),
|
||||||
|
required_roles=list(params.get("required_roles") or []),
|
||||||
|
is_enabled=bool(params.get("is_enabled", True)),
|
||||||
|
is_external=bool(params.get("is_external", False)),
|
||||||
|
)
|
||||||
|
self.state.navigation_links[key] = record
|
||||||
|
return FakeResult([])
|
||||||
|
|
||||||
|
if lower_sql.startswith("select id from navigation_links where group_id"):
|
||||||
|
group_id = int(params["group_id"])
|
||||||
|
slug = params["slug"]
|
||||||
|
record = self.state.navigation_links.get((group_id, slug))
|
||||||
|
rows = [SimpleNamespace(id=record["id"])] if record else []
|
||||||
|
return FakeResult(rows)
|
||||||
|
|
||||||
if lower_sql.startswith("insert into financial_inputs"):
|
if lower_sql.startswith("insert into financial_inputs"):
|
||||||
key = (int(params["scenario_id"]), params["name"])
|
key = (int(params["scenario_id"]), params["name"])
|
||||||
record = self.state.financial_inputs.get(key)
|
record = self.state.financial_inputs.get(key)
|
||||||
|
|||||||
250
tests/services/test_navigation_service.py
Normal file
250
tests/services/test_navigation_service.py
Normal file
@@ -0,0 +1,250 @@
|
|||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from dataclasses import dataclass, field
|
||||||
|
from typing import Dict, Iterable, List, cast
|
||||||
|
|
||||||
|
from fastapi import Request
|
||||||
|
|
||||||
|
from services.navigation import NavigationService
|
||||||
|
from services.repositories import NavigationRepository
|
||||||
|
from services.session import AuthSession, SessionTokens
|
||||||
|
from models import User
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class StubNavigationLink:
|
||||||
|
id: int
|
||||||
|
slug: str
|
||||||
|
label: str
|
||||||
|
parent_link_id: int | None = None
|
||||||
|
route_name: str | None = None
|
||||||
|
href_override: str | None = None
|
||||||
|
match_prefix: str | None = None
|
||||||
|
sort_order: int = 0
|
||||||
|
icon: str | None = None
|
||||||
|
tooltip: str | None = None
|
||||||
|
required_roles: List[str] = field(default_factory=list)
|
||||||
|
is_enabled: bool = True
|
||||||
|
is_external: bool = False
|
||||||
|
children: List["StubNavigationLink"] = field(default_factory=list)
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class StubNavigationGroup:
|
||||||
|
id: int
|
||||||
|
slug: str
|
||||||
|
label: str
|
||||||
|
sort_order: int = 0
|
||||||
|
icon: str | None = None
|
||||||
|
tooltip: str | None = None
|
||||||
|
is_enabled: bool = True
|
||||||
|
links: List[StubNavigationLink] = field(default_factory=list)
|
||||||
|
|
||||||
|
|
||||||
|
class StubNavigationRepository(NavigationRepository):
|
||||||
|
def __init__(self, groups: Iterable[StubNavigationGroup]) -> None:
|
||||||
|
super().__init__(session=None) # type: ignore[arg-type]
|
||||||
|
self._groups = list(groups)
|
||||||
|
|
||||||
|
def list_groups_with_links(self, *, include_disabled: bool = False):
|
||||||
|
if include_disabled:
|
||||||
|
return list(self._groups)
|
||||||
|
return [group for group in self._groups if group.is_enabled]
|
||||||
|
|
||||||
|
|
||||||
|
class StubRequest:
|
||||||
|
def __init__(
|
||||||
|
self,
|
||||||
|
*,
|
||||||
|
path_params: Dict[str, str] | None = None,
|
||||||
|
query_params: Dict[str, str] | None = None,
|
||||||
|
) -> None:
|
||||||
|
self.path_params = path_params or {}
|
||||||
|
self.query_params = query_params or {}
|
||||||
|
self._url_for_calls: List[tuple[str, Dict[str, str]]] = []
|
||||||
|
|
||||||
|
def url_for(self, name: str, **params: str) -> str:
|
||||||
|
self._url_for_calls.append((name, params))
|
||||||
|
if params:
|
||||||
|
suffix = "_".join(f"{key}-{value}" for key,
|
||||||
|
value in sorted(params.items()))
|
||||||
|
return f"/{name}/{suffix}"
|
||||||
|
return f"/{name}"
|
||||||
|
|
||||||
|
@property
|
||||||
|
def url_for_calls(self) -> List[tuple[str, Dict[str, str]]]:
|
||||||
|
return list(self._url_for_calls)
|
||||||
|
|
||||||
|
|
||||||
|
def _session(*, roles: Iterable[str], authenticated: bool = True) -> AuthSession:
|
||||||
|
tokens = SessionTokens(
|
||||||
|
access_token="token" if authenticated else None, refresh_token=None)
|
||||||
|
user = cast(User, object()) if authenticated else None
|
||||||
|
session = AuthSession(tokens=tokens, user=user, role_slugs=tuple(roles))
|
||||||
|
return session
|
||||||
|
|
||||||
|
|
||||||
|
def test_build_sidebar_filters_links_by_role():
|
||||||
|
visible_link = StubNavigationLink(
|
||||||
|
id=1,
|
||||||
|
slug="projects",
|
||||||
|
label="Projects",
|
||||||
|
href_override="/projects",
|
||||||
|
required_roles=["viewer"],
|
||||||
|
)
|
||||||
|
hidden_link = StubNavigationLink(
|
||||||
|
id=2,
|
||||||
|
slug="admin",
|
||||||
|
label="Admin",
|
||||||
|
href_override="/admin",
|
||||||
|
required_roles=["admin"],
|
||||||
|
)
|
||||||
|
group = StubNavigationGroup(id=1, slug="workspace", label="Workspace", links=[
|
||||||
|
visible_link, hidden_link])
|
||||||
|
|
||||||
|
service = NavigationService(StubNavigationRepository([group]))
|
||||||
|
dto = service.build_sidebar(
|
||||||
|
session=_session(roles=["viewer"]),
|
||||||
|
request=cast(Request, StubRequest()),
|
||||||
|
)
|
||||||
|
|
||||||
|
assert len(dto.groups) == 1
|
||||||
|
assert [link.label for link in dto.groups[0].links] == ["Projects"]
|
||||||
|
assert dto.roles == ("viewer",)
|
||||||
|
|
||||||
|
|
||||||
|
def test_build_sidebar_appends_anonymous_role_for_guests():
|
||||||
|
link = StubNavigationLink(
|
||||||
|
id=1, slug="help", label="Help", href_override="/help")
|
||||||
|
group = StubNavigationGroup(
|
||||||
|
id=1, slug="account", label="Account", links=[link])
|
||||||
|
|
||||||
|
service = NavigationService(StubNavigationRepository([group]))
|
||||||
|
dto = service.build_sidebar(session=AuthSession.anonymous(), request=None)
|
||||||
|
|
||||||
|
assert dto.roles[-1] == "anonymous"
|
||||||
|
assert dto.groups[0].links[0].href.startswith("/")
|
||||||
|
|
||||||
|
|
||||||
|
def test_build_sidebar_resolves_profitability_link_with_context():
|
||||||
|
link = StubNavigationLink(
|
||||||
|
id=1,
|
||||||
|
slug="profitability",
|
||||||
|
label="Profitability",
|
||||||
|
route_name="calculations.profitability_form",
|
||||||
|
href_override="/calculations/profitability",
|
||||||
|
)
|
||||||
|
group = StubNavigationGroup(
|
||||||
|
id=99, slug="insights", label="Insights", links=[link])
|
||||||
|
|
||||||
|
request = StubRequest(path_params={"project_id": "7", "scenario_id": "42"})
|
||||||
|
service = NavigationService(StubNavigationRepository([group]))
|
||||||
|
|
||||||
|
dto = service.build_sidebar(
|
||||||
|
session=_session(roles=["viewer"]),
|
||||||
|
request=cast(Request, request),
|
||||||
|
)
|
||||||
|
|
||||||
|
assert dto.groups[0].links[0].href == "/calculations.profitability_form/project_id-7_scenario_id-42"
|
||||||
|
assert request.url_for_calls[0][0] == "calculations.profitability_form"
|
||||||
|
assert request.url_for_calls[0][1] == {
|
||||||
|
"project_id": "7", "scenario_id": "42"}
|
||||||
|
assert dto.groups[0].links[0].match_prefix == dto.groups[0].links[0].href
|
||||||
|
|
||||||
|
|
||||||
|
def test_build_sidebar_resolves_opex_link_with_context():
|
||||||
|
link = StubNavigationLink(
|
||||||
|
id=2,
|
||||||
|
slug="opex",
|
||||||
|
label="Opex",
|
||||||
|
route_name="calculations.opex_form",
|
||||||
|
href_override="/calculations/opex",
|
||||||
|
)
|
||||||
|
group = StubNavigationGroup(
|
||||||
|
id=5, slug="workspace", label="Workspace", links=[link])
|
||||||
|
|
||||||
|
request = StubRequest(path_params={"project_id": "3", "scenario_id": "9"})
|
||||||
|
service = NavigationService(StubNavigationRepository([group]))
|
||||||
|
|
||||||
|
dto = service.build_sidebar(
|
||||||
|
session=_session(roles=["analyst"]),
|
||||||
|
request=cast(Request, request),
|
||||||
|
)
|
||||||
|
|
||||||
|
href = dto.groups[0].links[0].href
|
||||||
|
assert href == "/calculations.opex_form/project_id-3_scenario_id-9"
|
||||||
|
assert request.url_for_calls[0][0] == "calculations.opex_form"
|
||||||
|
assert request.url_for_calls[0][1] == {
|
||||||
|
"project_id": "3", "scenario_id": "9"}
|
||||||
|
|
||||||
|
|
||||||
|
def test_build_sidebar_uses_href_override_when_calculator_context_missing():
|
||||||
|
class ParamAwareStubRequest(StubRequest):
|
||||||
|
# type: ignore[override]
|
||||||
|
def url_for(self, name: str, **params: str) -> str:
|
||||||
|
if name in {
|
||||||
|
"calculations.opex_form",
|
||||||
|
"calculations.capex_form",
|
||||||
|
} and not params:
|
||||||
|
self._url_for_calls.append((name, params))
|
||||||
|
raise RuntimeError("missing params")
|
||||||
|
return super().url_for(name, **params)
|
||||||
|
|
||||||
|
link = StubNavigationLink(
|
||||||
|
id=3,
|
||||||
|
slug="capex",
|
||||||
|
label="Capex",
|
||||||
|
route_name="calculations.capex_form",
|
||||||
|
href_override="/calculations/capex",
|
||||||
|
)
|
||||||
|
group = StubNavigationGroup(
|
||||||
|
id=6, slug="workspace", label="Workspace", links=[link])
|
||||||
|
|
||||||
|
request = ParamAwareStubRequest()
|
||||||
|
service = NavigationService(StubNavigationRepository([group]))
|
||||||
|
|
||||||
|
dto = service.build_sidebar(
|
||||||
|
session=_session(roles=["analyst"]),
|
||||||
|
request=cast(Request, request),
|
||||||
|
)
|
||||||
|
|
||||||
|
assert dto.groups[0].links[0].href == "/calculations/capex"
|
||||||
|
assert request.url_for_calls[-1][0] == "calculations.capex_form"
|
||||||
|
|
||||||
|
|
||||||
|
def test_build_sidebar_skips_disabled_links_unless_included():
|
||||||
|
enabled_link = StubNavigationLink(
|
||||||
|
id=1,
|
||||||
|
slug="projects",
|
||||||
|
label="Projects",
|
||||||
|
href_override="/projects",
|
||||||
|
)
|
||||||
|
disabled_link = StubNavigationLink(
|
||||||
|
id=2,
|
||||||
|
slug="reports",
|
||||||
|
label="Reports",
|
||||||
|
href_override="/reports",
|
||||||
|
is_enabled=False,
|
||||||
|
)
|
||||||
|
group = StubNavigationGroup(
|
||||||
|
id=5,
|
||||||
|
slug="workspace",
|
||||||
|
label="Workspace",
|
||||||
|
links=[enabled_link, disabled_link],
|
||||||
|
)
|
||||||
|
|
||||||
|
service = NavigationService(StubNavigationRepository([group]))
|
||||||
|
|
||||||
|
default_sidebar = service.build_sidebar(
|
||||||
|
session=_session(roles=["viewer"]),
|
||||||
|
request=cast(Request, StubRequest()),
|
||||||
|
)
|
||||||
|
assert [link.label for link in default_sidebar.groups[0].links] == ["Projects"]
|
||||||
|
|
||||||
|
full_sidebar = service.build_sidebar(
|
||||||
|
session=_session(roles=["viewer"]),
|
||||||
|
request=cast(Request, StubRequest()),
|
||||||
|
include_disabled=True,
|
||||||
|
)
|
||||||
|
assert [link.label for link in full_sidebar.groups[0].links] == [
|
||||||
|
"Projects", "Reports"]
|
||||||
@@ -90,9 +90,9 @@ class TestAuthenticationRequirements:
|
|||||||
|
|
||||||
def test_ui_project_list_requires_login(self, client, auth_session_context):
|
def test_ui_project_list_requires_login(self, client, auth_session_context):
|
||||||
with auth_session_context(None):
|
with auth_session_context(None):
|
||||||
response = client.get("/projects/ui")
|
response = client.get("/projects/ui", follow_redirects=False)
|
||||||
|
assert response.status_code == status.HTTP_303_SEE_OTHER
|
||||||
assert response.status_code == status.HTTP_401_UNAUTHORIZED
|
assert response.headers["location"].endswith("/login")
|
||||||
|
|
||||||
|
|
||||||
class TestRoleRestrictions:
|
class TestRoleRestrictions:
|
||||||
@@ -194,7 +194,7 @@ class TestRoleRestrictions:
|
|||||||
|
|
||||||
assert response.status_code == status.HTTP_403_FORBIDDEN
|
assert response.status_code == status.HTTP_403_FORBIDDEN
|
||||||
assert response.json()[
|
assert response.json()[
|
||||||
"detail"] == "Insufficient role permissions for this action."
|
"detail"] == "Insufficient permissions for this action."
|
||||||
|
|
||||||
def test_ui_project_edit_accessible_to_manager(
|
def test_ui_project_edit_accessible_to_manager(
|
||||||
self,
|
self,
|
||||||
|
|||||||
Reference in New Issue
Block a user