Refactor and enhance CalMiner application

- Updated README.md to reflect new features and usage instructions.
- Removed deprecated Dashboard.html component and integrated dashboard functionality directly into the main application.
- Revised architecture documentation for clarity and added module map and request flow diagrams.
- Enhanced maintenance model to include equipment association and cost tracking.
- Updated requirements.txt to include new dependencies (httpx, pandas, numpy).
- Improved consumption, maintenance, production, and reporting routes with better validation and response handling.
- Added unit tests for maintenance and production routes, ensuring proper CRUD operations and validation.
- Enhanced reporting service to calculate and return detailed summary statistics.
- Redesigned Dashboard.html for improved user experience and integrated Chart.js for visualizing simulation results.
This commit is contained in:
2025-10-20 20:53:55 +02:00
parent fee857637f
commit e73a987d25
19 changed files with 794 additions and 184 deletions

View File

@@ -1,8 +1,30 @@
from fastapi.testclient import TestClient
from main import app
from config.database import Base, engine
from uuid import uuid4
# Setup and teardown
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from config.database import Base
from main import app
from routes import (
consumption,
costs,
distributions,
equipment,
maintenance,
parameters,
production,
reporting,
scenarios,
simulations,
ui,
)
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={
"check_same_thread": False})
TestingSessionLocal = sessionmaker(
autocommit=False, autoflush=False, bind=engine)
def setup_module(module):
@@ -13,29 +35,145 @@ def teardown_module(module):
Base.metadata.drop_all(bind=engine)
def override_get_db():
db = TestingSessionLocal()
try:
yield db
finally:
db.close()
app.dependency_overrides[maintenance.get_db] = override_get_db
app.dependency_overrides[equipment.get_db] = override_get_db
app.dependency_overrides[scenarios.get_db] = override_get_db
app.dependency_overrides[distributions.get_db] = override_get_db
app.dependency_overrides[parameters.get_db] = override_get_db
app.dependency_overrides[costs.get_db] = override_get_db
app.dependency_overrides[consumption.get_db] = override_get_db
app.dependency_overrides[production.get_db] = override_get_db
app.dependency_overrides[reporting.get_db] = override_get_db
app.dependency_overrides[simulations.get_db] = override_get_db
app.include_router(maintenance.router)
app.include_router(equipment.router)
app.include_router(scenarios.router)
app.include_router(distributions.router)
app.include_router(ui.router)
app.include_router(parameters.router)
app.include_router(costs.router)
app.include_router(consumption.router)
app.include_router(production.router)
app.include_router(reporting.router)
app.include_router(simulations.router)
client = TestClient(app)
def _create_scenario_and_equipment():
scenario_payload = {
"name": f"Test Scenario {uuid4()}",
"description": "Scenario for maintenance tests",
}
scenario_response = client.post("/api/scenarios/", json=scenario_payload)
assert scenario_response.status_code == 200
scenario_id = scenario_response.json()["id"]
equipment_payload = {
"scenario_id": scenario_id,
"name": f"Test Equipment {uuid4()}",
"description": "Equipment linked to maintenance",
}
equipment_response = client.post("/api/equipment/", json=equipment_payload)
assert equipment_response.status_code == 200
equipment_id = equipment_response.json()["id"]
return scenario_id, equipment_id
def _create_maintenance_payload(equipment_id: int, scenario_id: int, description: str):
return {
"equipment_id": equipment_id,
"scenario_id": scenario_id,
"maintenance_date": "2025-10-20",
"description": description,
"cost": 100.0,
}
def test_create_and_list_maintenance():
# Create a scenario to attach maintenance
resp = client.post(
"/api/scenarios/", json={"name": "MaintScenario", "description": "maintenance scenario"}
scenario_id, equipment_id = _create_scenario_and_equipment()
payload = _create_maintenance_payload(
equipment_id, scenario_id, "Create maintenance")
response = client.post("/api/maintenance/", json=payload)
assert response.status_code == 201
created = response.json()
assert created["equipment_id"] == equipment_id
assert created["scenario_id"] == scenario_id
assert created["description"] == "Create maintenance"
list_response = client.get("/api/maintenance/")
assert list_response.status_code == 200
items = list_response.json()
assert any(item["id"] == created["id"] for item in items)
def test_get_maintenance():
scenario_id, equipment_id = _create_scenario_and_equipment()
payload = _create_maintenance_payload(
equipment_id, scenario_id, "Retrieve maintenance"
)
assert resp.status_code == 200
scenario = resp.json()
sid = scenario["id"]
create_response = client.post("/api/maintenance/", json=payload)
assert create_response.status_code == 201
maintenance_id = create_response.json()["id"]
# Create Maintenance record
maint_payload = {"scenario_id": sid, "details": "Routine check"}
resp2 = client.post("/api/maintenance/", json=maint_payload)
assert resp2.status_code == 200
maint = resp2.json()
assert maint["scenario_id"] == sid
assert maint["details"] == "Routine check"
response = client.get(f"/api/maintenance/{maintenance_id}")
assert response.status_code == 200
data = response.json()
assert data["id"] == maintenance_id
assert data["equipment_id"] == equipment_id
assert data["description"] == "Retrieve maintenance"
# List Maintenance records
resp3 = client.get("/api/maintenance/")
assert resp3.status_code == 200
data = resp3.json()
assert any(item["details"] ==
"Routine check" and item["scenario_id"] == sid for item in data)
def test_update_maintenance():
scenario_id, equipment_id = _create_scenario_and_equipment()
create_response = client.post(
"/api/maintenance/",
json=_create_maintenance_payload(
equipment_id, scenario_id, "Maintenance before update"
),
)
assert create_response.status_code == 201
maintenance_id = create_response.json()["id"]
update_payload = {
"equipment_id": equipment_id,
"scenario_id": scenario_id,
"maintenance_date": "2025-11-01",
"description": "Maintenance after update",
"cost": 250.0,
}
response = client.put(
f"/api/maintenance/{maintenance_id}", json=update_payload)
assert response.status_code == 200
updated = response.json()
assert updated["maintenance_date"] == "2025-11-01"
assert updated["description"] == "Maintenance after update"
assert updated["cost"] == 250.0
def test_delete_maintenance():
scenario_id, equipment_id = _create_scenario_and_equipment()
create_response = client.post(
"/api/maintenance/",
json=_create_maintenance_payload(
equipment_id, scenario_id, "Delete maintenance"),
)
assert create_response.status_code == 201
maintenance_id = create_response.json()["id"]
delete_response = client.delete(f"/api/maintenance/{maintenance_id}")
assert delete_response.status_code == 204
get_response = client.get(f"/api/maintenance/{maintenance_id}")
assert get_response.status_code == 404