M1: implement operational core
Demo auth, seed import/reset, dashboard, vehicle/booking list+detail, audit trail. Backend: 19 tests passing, ruff clean. Frontend: React Router shell, typed API client, responsive pages. Verified end-to-end via curl and browser.
This commit is contained in:
@@ -0,0 +1,38 @@
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from app.core.db import Base, SessionLocal, engine
|
||||
from app.main import app
|
||||
from app.seed_loader import reset_and_seed
|
||||
|
||||
|
||||
@pytest.fixture(scope="session", autouse=True)
|
||||
def _seeded_database():
|
||||
Base.metadata.create_all(engine)
|
||||
db = SessionLocal()
|
||||
try:
|
||||
reset_and_seed(db)
|
||||
finally:
|
||||
db.close()
|
||||
yield
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client() -> TestClient:
|
||||
return TestClient(app)
|
||||
|
||||
|
||||
def login(client: TestClient, role: str) -> TestClient:
|
||||
response = client.post("/api/v1/demo/login", json={"role": role})
|
||||
assert response.status_code == 200
|
||||
return client
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def ops_client(client: TestClient) -> TestClient:
|
||||
return login(client, "operations_manager")
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def employee_client(client: TestClient) -> TestClient:
|
||||
return login(client, "rental_employee")
|
||||
@@ -0,0 +1,11 @@
|
||||
def test_demo_login_is_audited(ops_client):
|
||||
response = ops_client.get("/api/v1/audit", params={"action": "demo_login"})
|
||||
assert response.status_code == 200
|
||||
events = response.json()
|
||||
assert len(events) >= 1
|
||||
assert events[0]["action"] == "demo_login"
|
||||
|
||||
|
||||
def test_audit_requires_authentication(client):
|
||||
response = client.get("/api/v1/audit")
|
||||
assert response.status_code == 401
|
||||
@@ -0,0 +1,20 @@
|
||||
def test_unauthenticated_dashboard_is_rejected(client):
|
||||
response = client.get("/api/v1/dashboard")
|
||||
assert response.status_code == 401
|
||||
assert response.json()["error"]["code"] == "401"
|
||||
|
||||
|
||||
def test_demo_login_grants_access(ops_client):
|
||||
response = ops_client.get("/api/v1/dashboard")
|
||||
assert response.status_code == 200
|
||||
|
||||
|
||||
def test_rental_employee_cannot_reset_demo(employee_client):
|
||||
response = employee_client.post("/api/v1/demo/reset")
|
||||
assert response.status_code == 403
|
||||
|
||||
|
||||
def test_operations_manager_can_reset_demo(ops_client):
|
||||
response = ops_client.post("/api/v1/demo/reset")
|
||||
assert response.status_code == 200
|
||||
assert response.json()["counts"]["vehicles"] == 50
|
||||
@@ -0,0 +1,19 @@
|
||||
def test_list_bookings_filters_by_vehicle(ops_client):
|
||||
response = ops_client.get("/api/v1/bookings", params={"vehicle_ref": "MO-024"})
|
||||
assert response.status_code == 200
|
||||
bookings = response.json()
|
||||
assert len(bookings) >= 1
|
||||
assert all(b["vehicle_ref"] == "MO-024" for b in bookings)
|
||||
|
||||
|
||||
def test_get_booking_detail(ops_client):
|
||||
response = ops_client.get("/api/v1/bookings/BK-DEMO-RETURN")
|
||||
assert response.status_code == 200
|
||||
body = response.json()
|
||||
assert body["status"] == "active"
|
||||
assert body["vehicle_ref"] == "MO-024"
|
||||
|
||||
|
||||
def test_get_booking_404_for_unknown_ref(ops_client):
|
||||
response = ops_client.get("/api/v1/bookings/BK-UNKNOWN")
|
||||
assert response.status_code == 404
|
||||
@@ -0,0 +1,30 @@
|
||||
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_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_recent_automation_capped_at_five(ops_client):
|
||||
response = ops_client.get("/api/v1/dashboard")
|
||||
body = response.json()
|
||||
assert len(body["recent_automation"]) == 5
|
||||
@@ -0,0 +1,46 @@
|
||||
from sqlalchemy import func, select
|
||||
|
||||
from app.core.db import SessionLocal
|
||||
from app.models.booking import Booking
|
||||
from app.models.customer import Customer
|
||||
from app.models.data_quality import DataQualityIssue
|
||||
from app.models.outbox import OutboxEvent
|
||||
from app.models.user import User
|
||||
from app.models.vehicle import Vehicle
|
||||
|
||||
|
||||
def test_seed_counts_match_deterministic_dataset():
|
||||
db = SessionLocal()
|
||||
try:
|
||||
assert db.scalar(select(func.count()).select_from(Vehicle)) == 50
|
||||
assert db.scalar(select(func.count()).select_from(Customer)) == 180
|
||||
assert db.scalar(select(func.count()).select_from(Booking)) == 246
|
||||
assert db.scalar(select(func.count()).select_from(DataQualityIssue)) == 15
|
||||
assert db.scalar(select(func.count()).select_from(OutboxEvent)) == 20
|
||||
assert db.scalar(select(func.count()).select_from(User)) == 2
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
def test_seed_demo_scenarios_present():
|
||||
db = SessionLocal()
|
||||
try:
|
||||
booking = db.scalar(select(Booking).where(Booking.public_ref == "BK-DEMO-RETURN"))
|
||||
assert booking is not None
|
||||
assert booking.status == "active"
|
||||
|
||||
duplicate_customer = db.scalar(select(Customer).where(Customer.public_ref == "CUS-0178"))
|
||||
assert duplicate_customer is not None
|
||||
|
||||
duplicate_issue = db.scalar(
|
||||
select(DataQualityIssue).where(DataQualityIssue.public_ref == "DQ-DEMO-DUPLICATE")
|
||||
)
|
||||
assert duplicate_issue is not None
|
||||
assert duplicate_issue.rule_type == "possible_duplicate_customer"
|
||||
|
||||
failed_run = db.scalar(
|
||||
select(OutboxEvent).where(OutboxEvent.delivery_status == "failed")
|
||||
)
|
||||
assert failed_run is not None
|
||||
finally:
|
||||
db.close()
|
||||
@@ -0,0 +1,27 @@
|
||||
def test_list_vehicles_filters_by_status(ops_client):
|
||||
response = ops_client.get("/api/v1/vehicles", params={"status": "maintenance"})
|
||||
assert response.status_code == 200
|
||||
vehicles = response.json()
|
||||
assert len(vehicles) > 0
|
||||
assert all(v["operational_status"] == "maintenance" for v in vehicles)
|
||||
|
||||
|
||||
def test_attention_only_filters_flagged_vehicles(ops_client):
|
||||
response = ops_client.get("/api/v1/vehicles", params={"attention_only": True})
|
||||
assert response.status_code == 200
|
||||
vehicles = response.json()
|
||||
assert len(vehicles) > 0
|
||||
assert all(v["attention"] for v in vehicles)
|
||||
|
||||
|
||||
def test_vehicle_detail_includes_related_records(ops_client):
|
||||
response = ops_client.get("/api/v1/vehicles/MO-016")
|
||||
assert response.status_code == 200
|
||||
body = response.json()
|
||||
assert body["public_ref"] == "MO-016"
|
||||
assert len(body["quality_issues"]) >= 1
|
||||
|
||||
|
||||
def test_vehicle_detail_404_for_unknown_ref(ops_client):
|
||||
response = ops_client.get("/api/v1/vehicles/MO-999")
|
||||
assert response.status_code == 404
|
||||
Reference in New Issue
Block a user