90 lines
3.2 KiB
Python
90 lines
3.2 KiB
Python
from sqlalchemy import select
|
|
|
|
from app.core.db import SessionLocal
|
|
from app.models.vehicle import Vehicle
|
|
|
|
|
|
def test_dashboard_metrics_are_persisted_counts(ops_client):
|
|
response = ops_client.get("/api/v1/dashboard")
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
metrics = body["metrics"]
|
|
total = (
|
|
metrics["available"]
|
|
+ metrics["rented"]
|
|
+ metrics["cleaning"]
|
|
+ metrics["maintenance"]
|
|
+ metrics["blocked"]
|
|
)
|
|
assert total == 50
|
|
assert metrics["open_quality_issues"] >= 1
|
|
assert metrics["pending_or_failed_workflows"] >= 1
|
|
|
|
|
|
def test_dashboard_metrics_change_when_persisted_fleet_data_changes(ops_client):
|
|
before = ops_client.get("/api/v1/dashboard").json()["metrics"]
|
|
db = SessionLocal()
|
|
probe = Vehicle(
|
|
public_ref="MO-METRIC-PROBE",
|
|
make="Synthetic",
|
|
model="Metric probe",
|
|
model_year=2026,
|
|
registration_number="TEST-METRIC-PROBE",
|
|
location="Test fixture",
|
|
operational_status="blocked",
|
|
odometer_km=0,
|
|
next_service_km=1,
|
|
active=True,
|
|
version=1,
|
|
)
|
|
try:
|
|
db.add(probe)
|
|
db.commit()
|
|
|
|
after_response = ops_client.get("/api/v1/dashboard")
|
|
assert after_response.status_code == 200
|
|
after = after_response.json()["metrics"]
|
|
assert after["blocked"] == before["blocked"] + 1
|
|
fleet_statuses = ("available", "rented", "cleaning", "maintenance", "blocked")
|
|
assert (
|
|
sum(after[status] for status in fleet_statuses)
|
|
== sum(before[status] for status in fleet_statuses) + 1
|
|
)
|
|
finally:
|
|
persisted_probe = db.scalar(select(Vehicle).where(Vehicle.public_ref == "MO-METRIC-PROBE"))
|
|
if persisted_probe is not None:
|
|
db.delete(persisted_probe)
|
|
db.commit()
|
|
db.close()
|
|
|
|
|
|
def test_dashboard_attention_items_link_to_records(ops_client):
|
|
response = ops_client.get("/api/v1/dashboard")
|
|
body = response.json()
|
|
assert len(body["attention_items"]) > 0
|
|
for item in body["attention_items"]:
|
|
assert item["link_ref"]
|
|
assert item["severity"] in ("low", "medium", "high")
|
|
|
|
|
|
def test_dashboard_attention_items_expose_localizable_signals_not_raw_text(ops_client):
|
|
"""The dashboard subtext used to be raw, untranslated evidence text (and for most
|
|
seeded issues, the meaningless placeholder 'Synthetic deterministic seed issue').
|
|
The API must never emit prose here -- only stable signal codes + params, exactly
|
|
like the data-quality issue detail page, for the frontend to localize."""
|
|
response = ops_client.get("/api/v1/dashboard")
|
|
body = response.json()
|
|
assert len(body["attention_items"]) > 0
|
|
for item in body["attention_items"]:
|
|
assert "detail" not in item
|
|
assert len(item["evidence_signals"]) > 0
|
|
for signal in item["evidence_signals"]:
|
|
assert signal["code"]
|
|
assert signal["code"] != "Synthetic deterministic seed issue"
|
|
|
|
|
|
def test_dashboard_recent_automation_capped_at_five(ops_client):
|
|
response = ops_client.get("/api/v1/dashboard")
|
|
body = response.json()
|
|
assert len(body["recent_automation"]) == 5
|