allucanget/calminer
2
0
Fork 0
Code Pull Requests Actions Packages Releases Activity

Enhance testing framework and UI feedback

Browse Source 
- Updated architecture documentation to include details on UI rendering checks and Playwright end-to-end tests.
- Revised testing documentation to specify Playwright for frontend E2E tests and added details on running tests.
- Implemented feedback mechanism in scenario form for successful creation notifications.
- Added feedback div in ScenarioForm.html for user notifications.
- Created new fixtures for Playwright tests to manage server and browser instances.
- Developed comprehensive E2E tests for consumption, costs, equipment, maintenance, production, and scenarios.
- Added smoke tests to verify UI page loading and form submissions.
- Enhanced unit tests for simulation and validation, including new tests for report generation and validation errors.
- Created new test files for router validation to ensure consistent error handling.
- Established a new test suite for UI routes to validate dashboard and reporting functionalities.
- Implemented validation tests to ensure proper handling of JSON payloads.
This commit is contained in:
zwitschi
2025-10-21 08:29:11 +02:00
parent ae4b9c136f
commit f020d276bc
20 changed files with 939 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

174
tests/unit/conftest.py
View File

@@ -1,4 +1,6 @@
from typing import Generator
from datetime import date
from typing import Any, Dict, Generator
from uuid import uuid4
import pytest
from fastapi.testclient import TestClient
@@ -8,6 +10,15 @@ from sqlalchemy.pool import StaticPool
from config.database import Base
from main import app
from models.capex import Capex
from models.consumption import Consumption
from models.equipment import Equipment
from models.maintenance import Maintenance
from models.opex import Opex
from models.parameters import Parameter
from models.production_output import ProductionOutput
from models.scenario import Scenario
from models.simulation_result import SimulationResult
SQLALCHEMY_TEST_URL = "sqlite:///:memory:"
engine = create_engine(
@@ -35,6 +46,19 @@ def setup_database() -> Generator[None, None, None]:
simulation_result,
) # noqa: F401 - imported for side effects
_ = (
capex,
consumption,
distribution,
equipment,
maintenance,
opex,
parameters,
production_output,
scenario,
simulation_result,
)
Base.metadata.create_all(bind=engine)
yield
Base.metadata.drop_all(bind=engine)
@@ -65,3 +89,151 @@ def api_client(db_session: Session) -> Generator[TestClient, None, None]:
yield client
app.dependency_overrides.pop(route_dependencies.get_db, None)
@pytest.fixture()
def seeded_ui_data(db_session: Session) -> Generator[Dict[str, Any], None, None]:
"""Populate a scenario with representative related records for UI tests."""
scenario_name = f"Scenario Alpha {uuid4()}"
scenario = Scenario(name=scenario_name,
description="Seeded UI scenario")
db_session.add(scenario)
db_session.flush()
parameter = Parameter(
scenario_id=scenario.id,
name="Ore Grade",
value=1.5,
distribution_type="normal",
distribution_parameters={"mean": 1.5, "std_dev": 0.1},
)
capex = Capex(
scenario_id=scenario.id,
amount=1_000_000.0,
description="Drill purchase",
)
opex = Opex(
scenario_id=scenario.id,
amount=250_000.0,
description="Fuel spend",
)
consumption = Consumption(
scenario_id=scenario.id,
amount=1_200.0,
description="Diesel (L)",
)
production = ProductionOutput(
scenario_id=scenario.id,
amount=800.0,
description="Ore (tonnes)",
)
equipment = Equipment(
scenario_id=scenario.id,
name="Excavator 42",
description="Primary loader",
)
db_session.add_all(
[parameter, capex, opex, consumption, production, equipment]
)
db_session.flush()
maintenance = Maintenance(
scenario_id=scenario.id,
equipment_id=equipment.id,
maintenance_date=date(2025, 1, 15),
description="Hydraulic service",
cost=15_000.0,
)
simulation_results = [
SimulationResult(
scenario_id=scenario.id,
iteration=index,
result=value,
)
for index, value in enumerate((950_000.0, 975_000.0, 990_000.0), start=1)
]
db_session.add(maintenance)
db_session.add_all(simulation_results)
db_session.commit()
try:
yield {
"scenario": scenario,
"equipment": equipment,
"simulation_results": simulation_results,
}
finally:
db_session.query(SimulationResult).filter_by(
scenario_id=scenario.id
).delete()
db_session.query(Maintenance).filter_by(
scenario_id=scenario.id
).delete()
db_session.query(Equipment).filter_by(id=equipment.id).delete()
db_session.query(ProductionOutput).filter_by(
scenario_id=scenario.id
).delete()
db_session.query(Consumption).filter_by(
scenario_id=scenario.id
).delete()
db_session.query(Opex).filter_by(scenario_id=scenario.id).delete()
db_session.query(Capex).filter_by(scenario_id=scenario.id).delete()
db_session.query(Parameter).filter_by(scenario_id=scenario.id).delete()
db_session.query(Scenario).filter_by(id=scenario.id).delete()
db_session.commit()
@pytest.fixture()
def invalid_request_payloads(db_session: Session) -> Generator[Dict[str, Any], None, None]:
"""Provide reusable invalid request bodies for exercising validation branches."""
duplicate_name = f"Scenario Duplicate {uuid4()}"
existing = Scenario(name=duplicate_name,
description="Existing scenario for duplicate checks")
db_session.add(existing)
db_session.commit()
payloads: Dict[str, Any] = {
"existing_scenario": existing,
"scenario_duplicate": {
"name": duplicate_name,
"description": "Second scenario should fail with duplicate name",
},
"parameter_missing_scenario": {
"scenario_id": existing.id + 99,
"name": "Invalid Parameter",
"value": 1.0,
},
"parameter_invalid_distribution": {
"scenario_id": existing.id,
"name": "Weird Dist",
"value": 2.5,
"distribution_type": "invalid",
},
"simulation_unknown_scenario": {
"scenario_id": existing.id + 99,
"iterations": 10,
"parameters": [
{"name": "grade", "value": 1.2, "distribution": "normal"}
],
},
"simulation_missing_parameters": {
"scenario_id": existing.id,
"iterations": 5,
"parameters": [],
},
"reporting_non_list_payload": {"result": 10.0},
"reporting_missing_result": [{"value": 12.0}],
"maintenance_negative_cost": {
"equipment_id": 1,
"scenario_id": existing.id,
"maintenance_date": "2025-01-15",
"cost": -500.0,
},
}
try:
yield payloads
finally:
db_session.query(Scenario).filter_by(id=existing.id).delete()
db_session.commit()

26
tests/unit/test_reporting.py
View File

@@ -49,6 +49,32 @@ def test_generate_report_with_values():
assert math.isclose(float(report["expected_shortfall_95"]), 10.0)
def test_generate_report_single_value():
report = generate_report([
{"iteration": 1, "result": 42.0},
])
assert report["count"] == 1
assert report["std_dev"] == 0.0
assert report["variance"] == 0.0
assert report["percentile_10"] == 42.0
assert report["expected_shortfall_95"] == 42.0
def test_generate_report_ignores_invalid_entries():
raw_values: List[Any] = [
{"iteration": 1, "result": 10.0},
"not-a-mapping",
{"iteration": 2},
{"iteration": 3, "result": None},
{"iteration": 4, "result": 20},
]
report = generate_report(raw_values)
assert report["count"] == 2
assert math.isclose(float(report["mean"]), 15.0)
assert math.isclose(float(report["min"]), 10.0)
assert math.isclose(float(report["max"]), 20.0)
@pytest.fixture
def client(api_client: TestClient) -> TestClient:
return api_client

95
tests/unit/test_router_validation.py Normal file
View File

@@ -0,0 +1,95 @@
from typing import Any, Dict
import pytest
from fastapi.testclient import TestClient
@pytest.mark.usefixtures("invalid_request_payloads")
def test_duplicate_scenario_returns_400(
api_client: TestClient, invalid_request_payloads: Dict[str, Any]
) -> None:
payload = invalid_request_payloads["scenario_duplicate"]
response = api_client.post("/api/scenarios/", json=payload)
assert response.status_code == 400
body = response.json()
assert body["detail"] == "Scenario already exists"
@pytest.mark.usefixtures("invalid_request_payloads")
def test_parameter_create_missing_scenario_returns_404(
api_client: TestClient, invalid_request_payloads: Dict[str, Any]
) -> None:
payload = invalid_request_payloads["parameter_missing_scenario"]
response = api_client.post("/api/parameters/", json=payload)
assert response.status_code == 404
assert response.json()["detail"] == "Scenario not found"
@pytest.mark.usefixtures("invalid_request_payloads")
def test_parameter_create_invalid_distribution_is_422(
api_client: TestClient
) -> None:
response = api_client.post(
"/api/parameters/",
json={
"scenario_id": 1,
"name": "Bad Dist",
"value": 2.0,
"distribution_type": "invalid",
},
)
assert response.status_code == 422
errors = response.json()["detail"]
assert any("distribution_type" in err["loc"] for err in errors)
@pytest.mark.usefixtures("invalid_request_payloads")
def test_simulation_unknown_scenario_returns_404(
api_client: TestClient, invalid_request_payloads: Dict[str, Any]
) -> None:
payload = invalid_request_payloads["simulation_unknown_scenario"]
response = api_client.post("/api/simulations/run", json=payload)
assert response.status_code == 404
assert response.json()["detail"] == "Scenario not found"
@pytest.mark.usefixtures("invalid_request_payloads")
def test_simulation_missing_parameters_returns_400(
api_client: TestClient, invalid_request_payloads: Dict[str, Any]
) -> None:
payload = invalid_request_payloads["simulation_missing_parameters"]
response = api_client.post("/api/simulations/run", json=payload)
assert response.status_code == 400
assert response.json()["detail"] == "No parameters provided"
@pytest.mark.usefixtures("invalid_request_payloads")
def test_reporting_summary_rejects_non_list_payload(
api_client: TestClient, invalid_request_payloads: Dict[str, Any]
) -> None:
payload = invalid_request_payloads["reporting_non_list_payload"]
response = api_client.post("/api/reporting/summary", json=payload)
assert response.status_code == 400
assert response.json()["detail"] == "Invalid input format"
@pytest.mark.usefixtures("invalid_request_payloads")
def test_reporting_summary_requires_result_field(
api_client: TestClient, invalid_request_payloads: Dict[str, Any]
) -> None:
payload = invalid_request_payloads["reporting_missing_result"]
response = api_client.post("/api/reporting/summary", json=payload)
assert response.status_code == 400
assert "must include numeric 'result'" in response.json()["detail"]
@pytest.mark.usefixtures("invalid_request_payloads")
def test_maintenance_negative_cost_rejected_by_schema(
api_client: TestClient, invalid_request_payloads: Dict[str, Any]
) -> None:
payload = invalid_request_payloads["maintenance_negative_cost"]
response = api_client.post("/api/maintenance/", json=payload)
assert response.status_code == 422
error_locations = [tuple(item["loc"])
for item in response.json()["detail"]]
assert ("body", "cost") in error_locations

81
tests/unit/test_simulation.py
View File

@@ -4,6 +4,8 @@ import pytest
from fastapi.testclient import TestClient
from sqlalchemy.orm import Session
from typing import Any, Dict, List
from models.simulation_result import SimulationResult
from services.simulation import run_simulation
@@ -14,7 +16,7 @@ def client(api_client: TestClient) -> TestClient:
def test_run_simulation_function_generates_samples():
params = [
params: List[Dict[str, Any]] = [
{"name": "grade", "value": 1.8, "distribution": "normal", "std_dev": 0.2},
{
"name": "recovery",
@@ -30,8 +32,73 @@ def test_run_simulation_function_generates_samples():
assert results[0]["iteration"] == 1
def test_run_simulation_with_zero_iterations_returns_empty():
params: List[Dict[str, Any]] = [
{"name": "grade", "value": 1.2, "distribution": "normal"}
]
results = run_simulation(params, iterations=0)
assert results == []
@pytest.mark.parametrize(
"parameter_payload,error_message",
[
({"name": "missing-value"}, "Parameter at index 0 must include 'value'"),
(
{
"name": "bad-dist",
"value": 1.0,
"distribution": "unsupported",
},
"Parameter 'bad-dist' has unsupported distribution 'unsupported'",
),
(
{
"name": "uniform-range",
"value": 1.0,
"distribution": "uniform",
"min": 5,
"max": 5,
},
"Parameter 'uniform-range' requires 'min' < 'max' for uniform distribution",
),
(
{
"name": "triangular-mode",
"value": 5.0,
"distribution": "triangular",
"min": 1,
"max": 3,
"mode": 5,
},
"Parameter 'triangular-mode' mode must be within min/max bounds for triangular distribution",
),
],
)
def test_run_simulation_parameter_validation_errors(
parameter_payload: Dict[str, Any], error_message: str
) -> None:
with pytest.raises(ValueError) as exc:
run_simulation([parameter_payload])
assert str(exc.value) == error_message
def test_run_simulation_normal_std_dev_fallback():
params: List[Dict[str, Any]] = [
{
"name": "std-dev-fallback",
"value": 10.0,
"distribution": "normal",
"std_dev": 0,
}
]
results = run_simulation(params, iterations=3, seed=99)
assert len(results) == 3
assert all("result" in entry for entry in results)
def test_simulation_endpoint_no_params(client: TestClient):
scenario_payload = {
scenario_payload: Dict[str, Any] = {
"name": f"NoParamScenario-{uuid4()}",
"description": "No parameters run",
}
@@ -50,7 +117,7 @@ def test_simulation_endpoint_no_params(client: TestClient):
def test_simulation_endpoint_success(
client: TestClient, db_session: Session
):
scenario_payload = {
scenario_payload: Dict[str, Any] = {
"name": f"SimScenario-{uuid4()}",
"description": "Simulation test",
}
@@ -58,10 +125,10 @@ def test_simulation_endpoint_success(
assert scenario_resp.status_code == 200
scenario_id = scenario_resp.json()["id"]
params = [
params: List[Dict[str, Any]] = [
{"name": "param1", "value": 2.5, "distribution": "normal", "std_dev": 0.5}
]
payload = {
payload: Dict[str, Any] = {
"scenario_id": scenario_id,
"parameters": params,
"iterations": 10,
@@ -85,7 +152,7 @@ def test_simulation_endpoint_success(
def test_simulation_endpoint_uses_stored_parameters(client: TestClient):
scenario_payload = {
scenario_payload: Dict[str, Any] = {
"name": f"StoredParams-{uuid4()}",
"description": "Stored parameter simulation",
}
@@ -93,7 +160,7 @@ def test_simulation_endpoint_uses_stored_parameters(client: TestClient):
assert scenario_resp.status_code == 200
scenario_id = scenario_resp.json()["id"]
parameter_payload = {
parameter_payload: Dict[str, Any] = {
"scenario_id": scenario_id,
"name": "grade",
"value": 1.5,

100
tests/unit/test_ui_routes.py Normal file
View File

@@ -0,0 +1,100 @@
from typing import Any, Dict, cast
from fastapi.testclient import TestClient
from models.scenario import Scenario
def test_dashboard_route_provides_summary(
api_client: TestClient, seeded_ui_data: Dict[str, Any]
) -> None:
response = api_client.get("/ui/dashboard")
assert response.status_code == 200
template = getattr(response, "template", None)
assert template is not None
assert template.name == "Dashboard.html"
context = cast(Dict[str, Any], getattr(response, "context", {}))
assert context.get("report_available") is True
metric_labels = {item["label"] for item in context["summary_metrics"]}
assert {"CAPEX Total", "OPEX Total", "Production", "Simulation Iterations"}.issubset(metric_labels)
scenario = cast(Scenario, seeded_ui_data["scenario"])
scenario_row = next(
row for row in context["scenario_rows"] if row["scenario_name"] == scenario.name
)
assert scenario_row["iterations"] == 3
assert scenario_row["simulation_mean_display"] == "971,666.67"
assert scenario_row["capex_display"] == "$1,000,000.00"
assert scenario_row["opex_display"] == "$250,000.00"
assert scenario_row["production_display"] == "800.00"
assert scenario_row["consumption_display"] == "1,200.00"
def test_scenarios_route_lists_seeded_scenario(
api_client: TestClient, seeded_ui_data: Dict[str, Any]
) -> None:
response = api_client.get("/ui/scenarios")
assert response.status_code == 200
template = getattr(response, "template", None)
assert template is not None
assert template.name == "ScenarioForm.html"
context = cast(Dict[str, Any], getattr(response, "context", {}))
names = [item["name"] for item in context["scenarios"]]
scenario = cast(Scenario, seeded_ui_data["scenario"])
assert scenario.name in names
def test_reporting_route_includes_summary(
api_client: TestClient, seeded_ui_data: Dict[str, Any]
) -> None:
response = api_client.get("/ui/reporting")
assert response.status_code == 200
template = getattr(response, "template", None)
assert template is not None
assert template.name == "reporting.html"
context = cast(Dict[str, Any], getattr(response, "context", {}))
summaries = context["report_summaries"]
scenario = cast(Scenario, seeded_ui_data["scenario"])
scenario_summary = next(
item for item in summaries if item["scenario_id"] == scenario.id
)
assert scenario_summary["iterations"] == 3
mean_value = float(scenario_summary["summary"]["mean"])
assert abs(mean_value - 971_666.6666666666) < 1e-6
def test_dashboard_data_endpoint_returns_aggregates(
api_client: TestClient, seeded_ui_data: Dict[str, Any]
) -> None:
response = api_client.get("/ui/dashboard/data")
assert response.status_code == 200
payload = response.json()
assert payload["report_available"] is True
metric_map = {item["label"]: item["value"] for item in payload["summary_metrics"]}
assert metric_map["CAPEX Total"].startswith("$")
assert metric_map["Maintenance Cost"].startswith("$")
scenario = cast(Scenario, seeded_ui_data["scenario"])
scenario_rows = payload["scenario_rows"]
scenario_entry = next(
row for row in scenario_rows if row["scenario_name"] == scenario.name
)
assert scenario_entry["capex_display"] == "$1,000,000.00"
assert scenario_entry["production_display"] == "800.00"
labels = payload["scenario_cost_chart"]["labels"]
idx = labels.index(scenario.name)
assert payload["scenario_cost_chart"]["capex"][idx] == 1_000_000.0
activity_labels = payload["scenario_activity_chart"]["labels"]
activity_idx = activity_labels.index(scenario.name)
assert payload["scenario_activity_chart"]["production"][activity_idx] == 800.0

28
tests/unit/test_validation.py Normal file
View File

@@ -0,0 +1,28 @@
from uuid import uuid4
import pytest
from fastapi import HTTPException
from fastapi.testclient import TestClient
def test_validate_json_allows_valid_payload(api_client: TestClient) -> None:
payload = {
"name": f"ValidJSON-{uuid4()}",
"description": "Middleware should allow valid JSON.",
}
response = api_client.post("/api/scenarios/", json=payload)
assert response.status_code == 200
data = response.json()
assert data["name"] == payload["name"]
def test_validate_json_rejects_invalid_payload(api_client: TestClient) -> None:
with pytest.raises(HTTPException) as exc_info:
api_client.post(
"/api/scenarios/",
content=b"{not valid json",
headers={"Content-Type": "application/json"},
)
assert exc_info.value.status_code == 400
assert exc_info.value.detail == "Invalid JSON payload"