feat(navigation): Enhance navigation links and add legacy route redirects
Some checks failed
CI / test (push) Has been skipped
CI / build (push) Has been skipped
CI / lint (push) Failing after 14s
CI / deploy (push) Has been skipped

- Updated navigation links in `init_db.py` to include href overrides and parent slugs for profitability, opex, and capex planners.
- Modified `NavigationService` to handle child links and href overrides, ensuring proper routing when context is missing.
- Adjusted scenario detail and list templates to use new route names for opex and capex forms, with legacy fallbacks.
- Introduced integration tests for legacy calculation routes to ensure proper redirection and error handling.
- Added tests for navigation sidebar to validate role-based access and link visibility.
- Enhanced navigation sidebar tests to include calculation links and contextual URLs based on project and scenario IDs.
This commit is contained in:
2025-11-13 20:23:53 +01:00
parent ed8e05147c
commit 4d0e1a9989
12 changed files with 1050 additions and 73 deletions

View File

@@ -2,6 +2,9 @@
## 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.
- 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).

View File

@@ -5,7 +5,7 @@ from __future__ import annotations
from decimal import Decimal
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 pydantic import ValidationError
from starlette.datastructures import FormData
@@ -917,6 +917,29 @@ def _load_project_and_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:
content_type = request.headers.get("content-type", "").lower()
accept = request.headers.get("accept", "").lower()
@@ -930,6 +953,41 @@ def _normalise_form_value(value: Any) -> Any:
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]:
data: dict[str, Any] = {}
impurities: dict[int, dict[str, Any]] = {}
@@ -1258,22 +1316,20 @@ def _persist_opex_snapshots(
@router.get(
"/opex",
"/projects/{project_id}/scenarios/{scenario_id}/calculations/opex",
response_class=HTMLResponse,
name="calculations.opex_form",
name="calculations.scenario_opex_form",
)
def opex_form(
request: Request,
project_id: int,
scenario_id: int,
_: User = Depends(require_authenticated_user_html),
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:
"""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
)
context = _prepare_opex_context(
@@ -1285,23 +1341,25 @@ def opex_form(
@router.post(
"/opex",
name="calculations.opex_submit",
"/projects/{project_id}/scenarios/{scenario_id}/calculations/opex",
name="calculations.scenario_opex_submit",
)
async def opex_submit(
request: Request,
project_id: int,
scenario_id: int,
current_user: User = Depends(require_authenticated_user),
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:
"""Handle opex submissions and respond with HTML or JSON."""
wants_json = _is_json_request(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:
request_model = OpexCalculationRequest.model_validate(
payload_data
@@ -1314,9 +1372,6 @@ async def opex_submit(
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(
exc.errors()
)
@@ -1344,9 +1399,6 @@ async def opex_submit(
},
)
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]
context = _prepare_opex_context(
request,
@@ -1362,10 +1414,6 @@ async def opex_submit(
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(
uow=uow,
project=project,
@@ -1400,22 +1448,145 @@ async def opex_submit(
@router.get(
"/capex",
"/opex",
response_class=HTMLResponse,
name="calculations.capex_form",
name="calculations.opex_form_legacy",
)
def capex_form(
def opex_form_legacy(
request: Request,
_: User = Depends(require_authenticated_user_html),
uow: UnitOfWork = Depends(get_unit_of_work),
project_id: int | None = Query(
project_id: str | None = Query(
None, description="Optional project identifier"),
scenario_id: int | None = Query(
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.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:
"""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
)
context = _prepare_capex_context(
@@ -1427,23 +1598,25 @@ def capex_form(
@router.post(
"/capex",
name="calculations.capex_submit",
"/projects/{project_id}/scenarios/{scenario_id}/calculations/capex",
name="calculations.scenario_capex_submit",
)
async def capex_submit(
request: Request,
project_id: int,
scenario_id: int,
current_user: User = Depends(require_authenticated_user),
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:
"""Process capex submissions and return aggregated results."""
wants_json = _is_json_request(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:
request_model = CapexCalculationRequest.model_validate(payload_data)
result = calculate_initial_capex(request_model)
@@ -1454,9 +1627,6 @@ async def capex_submit(
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(
exc.errors()
)
@@ -1484,9 +1654,6 @@ async def capex_submit(
},
)
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]
context = _prepare_capex_context(
request,
@@ -1502,10 +1669,6 @@ async def capex_submit(
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(
uow=uow,
project=project,
@@ -1539,6 +1702,164 @@ async def capex_submit(
)
# 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(
request: Request,
*,

View File

@@ -657,27 +657,33 @@ DEFAULT_NAVIGATION_LINKS: list[NavigationLinkSeed] = [
group_slug="workspace",
label="Profitability Calculator",
route_name="calculations.profitability_form",
href_override="/calculations/profitability",
match_prefix="/calculations/profitability",
sort_order=50,
required_roles=["analyst", "admin"],
parent_slug="projects",
),
NavigationLinkSeed(
slug="opex",
group_slug="workspace",
label="Opex Planner",
route_name="calculations.opex_form",
href_override="/calculations/opex",
match_prefix="/calculations/opex",
sort_order=60,
required_roles=["analyst", "admin"],
parent_slug="projects",
),
NavigationLinkSeed(
slug="capex",
group_slug="workspace",
label="Capex Planner",
route_name="calculations.capex_form",
href_override="/calculations/capex",
match_prefix="/calculations/capex",
sort_order=70,
required_roles=["analyst", "admin"],
parent_slug="projects",
),
NavigationLinkSeed(
slug="simulations",

View File

@@ -88,10 +88,13 @@ class NavigationService:
request: Request | None,
include_disabled: bool,
context: dict[str, str | None],
include_children: bool = False,
) -> List[NavigationLinkDTO]:
resolved_roles = tuple(roles)
mapped: List[NavigationLinkDTO] = []
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):
continue
if not self._link_visible(link, resolved_roles, include_disabled):
@@ -105,6 +108,7 @@ class NavigationService:
request=request,
include_disabled=include_disabled,
context=context,
include_children=True,
)
match_prefix = link.match_prefix or href
mapped.append(
@@ -153,22 +157,33 @@ class NavigationService:
) -> str | None:
if link.route_name:
if request is None:
fallback = link.href_override
if fallback:
return fallback
# Fallback to route name when no request is available
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")
scenario_id = context.get("scenario_id")
if project_id and scenario_id:
try:
return request.url_for(
link.route_name,
project_id=project_id,
scenario_id=scenario_id,
return str(
request.url_for(
link.route_name,
project_id=project_id,
scenario_id=scenario_id,
)
)
except Exception: # pragma: no cover - defensive
pass
try:
return request.url_for(link.route_name)
return str(request.url_for(link.route_name))
except Exception: # pragma: no cover - defensive
return link.href_override
return link.href_override

View File

@@ -13,14 +13,15 @@
</nav>
<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) %}
{% if project and scenario %}
{% 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 capex_href = capex_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 = 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 %}
<div>
<h1>{{ scenario.name }}</h1>

View File

@@ -97,8 +97,8 @@
<ul class="scenario-list">
{% for scenario in scenarios %}
{% 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 capex_href = url_for('calculations.capex_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.scenario_capex_form', project_id=project.id, scenario_id=scenario.id) %}
<li class="scenario-item">
<div class="scenario-item__body">
<div class="scenario-item__header">

View File

@@ -18,6 +18,7 @@ from models import User
from routes.auth import router as auth_router
from routes.dashboard import router as dashboard_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.scenarios import router as scenarios_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 tests.utils.security import random_password, random_token
BASE_TESTSERVER_URL = "http://testserver"
TEST_USER_HEADER = "X-Test-User"
@pytest.fixture()
def engine() -> Iterator[Engine]:
@@ -60,6 +66,7 @@ def app(session_factory: sessionmaker) -> FastAPI:
application.include_router(dashboard_router)
application.include_router(calculations_router)
application.include_router(projects_router)
application.include_router(navigation_router)
application.include_router(scenarios_router)
application.include_router(imports_router)
application.include_router(exports_router)
@@ -85,26 +92,64 @@ def app(session_factory: sessionmaker) -> FastAPI:
] = _override_ingestion_service
with UnitOfWork(session_factory=session_factory) as uow:
assert uow.users is not None
uow.ensure_default_roles()
user = User(
assert uow.users is not None and uow.roles is not None
roles = {role.name: role for role in uow.ensure_default_roles()}
admin_user = User(
email="test-superuser@example.com",
username="test-superuser",
password_hash=User.hash_password(random_password()),
is_active=True,
is_superuser=True,
)
uow.users.create(user)
user = uow.users.get(user.id, with_roles=True)
viewer_user = User(
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:
session = AuthSession(
tokens=SessionTokens(
access_token=random_token(),
refresh_token=random_token(),
alias = request.headers.get(TEST_USER_HEADER, "admin").strip().lower()
if alias == "anonymous":
session = AuthSession.anonymous()
else:
user, role_slugs = _resolve_user(alias or "admin")
session = AuthSession(
tokens=SessionTokens(
access_token=random_token(),
refresh_token=random_token(),
),
user=user,
)
)
session.user = user
session.set_role_slugs(role_slugs)
request.state.auth_session = session
return session
@@ -114,7 +159,7 @@ def app(session_factory: sessionmaker) -> FastAPI:
@pytest.fixture()
def client(app: FastAPI) -> Iterator[TestClient]:
test_client = TestClient(app)
test_client = TestClient(app, headers={TEST_USER_HEADER: "admin"})
try:
yield test_client
finally:
@@ -124,13 +169,52 @@ def client(app: FastAPI) -> Iterator[TestClient]:
@pytest_asyncio.fixture()
async def async_client(app: FastAPI) -> 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()
def unit_of_work_factory(session_factory: sessionmaker) -> Callable[[], UnitOfWork]:
def _factory() -> UnitOfWork:
return UnitOfWork(session_factory=session_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

View 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"

View 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

View 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"

View File

@@ -1,10 +1,16 @@
from __future__ import annotations
from collections.abc import Callable
from fastapi.testclient import TestClient
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
project_response = client.post(
"/projects",
@@ -65,6 +71,18 @@ class TestScenarioLifecycle:
assert "CAD" in scenario_detail.text
assert "Electricity" 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 = client.get(f"/projects/{project_id}/view")
@@ -84,6 +102,8 @@ class TestScenarioLifecycle:
# Scenario detail should still show the previous (valid) currency
scenario_detail = client.get(f"/scenarios/{scenario_id}/view")
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_response = client.put(
@@ -96,9 +116,17 @@ class TestScenarioLifecycle:
# Scenario detail reflects archived status
scenario_detail = client.get(f"/scenarios/{scenario_id}/view")
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 = client.get(f"/projects/{project_id}/view")
assert "<h2>Archived</h2>" in project_detail.text
assert '<p class="metric-value">1</p>' 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

View File

@@ -16,6 +16,7 @@ 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
@@ -131,6 +132,7 @@ def test_build_sidebar_resolves_profitability_link_with_context():
slug="profitability",
label="Profitability",
route_name="calculations.profitability_form",
href_override="/calculations/profitability",
)
group = StubNavigationGroup(
id=99, slug="insights", label="Insights", links=[link])
@@ -150,6 +152,66 @@ def test_build_sidebar_resolves_profitability_link_with_context():
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,