fix(demo): anchor seeded dates to the real reset moment

Booking/inspection/maintenance/outbox dates were authored as absolute
timestamps around a fixed 2026-08-01 anchor and never re-anchored at
seed/reset time, so demo scenarios (e.g. BK-DEMO-RETURN) silently drifted
into the past. Every reset now shifts seeded dates by (today - authored
anchor); dashboard's "today" filter uses real wall-clock time instead of
the now-removed frozen demo_today setting. Adds seed-validation tests
proving scenarios S1/S2/S4/S5 are present and internally consistent after
every reset.
This commit is contained in:
NuklearRabbit
2026-08-03 13:08:02 +02:00
parent 7c94eb9e87
commit 8989ffb23c
9 changed files with 218 additions and 15 deletions
+37 -7
View File
@@ -3,7 +3,7 @@ from __future__ import annotations
import csv
import uuid
from dataclasses import dataclass
from datetime import UTC, datetime
from datetime import UTC, date, datetime, timedelta
from pathlib import Path
from sqlalchemy import delete, insert, update
@@ -20,6 +20,7 @@ from app.models.maintenance import MaintenanceRecord
from app.models.outbox import OutboxEvent
from app.models.user import User
from app.models.vehicle import Vehicle
from app.services.audit import record_audit_event
settings = get_settings()
@@ -36,6 +37,17 @@ DEMO_USERS = [
},
]
# seed/generate_seed.py authored the committed CSVs relative to this fixed date
# (`--anchor 2026-08-01`, matching Settings.demo_today). Every reset shifts every
# seeded date by (today - SEED_AUTHORED_ANCHOR) so "today" / "near-future" / "overlaps
# right now" scenarios stay true to the actual reset moment instead of decaying as real
# time passes between resets -- a fixed anchor with no shift goes stale within days.
SEED_AUTHORED_ANCHOR = date(2026, 8, 1)
def _seed_anchor_shift(today: date) -> timedelta:
return today - SEED_AUTHORED_ANCHOR
def _parse_dt(value: str) -> datetime:
return datetime.fromisoformat(value.replace("Z", "+00:00"))
@@ -53,6 +65,8 @@ def _parse_optional_int(value: str) -> int | None:
@dataclass
class SeedResult:
counts: dict[str, int]
anchor_date: date
seeded_at: datetime
def _seed_dir() -> Path:
@@ -83,6 +97,8 @@ def clear_all(db: Session) -> None:
def load_seed(db: Session) -> SeedResult:
counts: dict[str, int] = {}
today = datetime.now(UTC).date()
shift = _seed_anchor_shift(today)
user_rows = [
{"id": uuid.uuid4(), **user, "active": True} for user in DEMO_USERS
@@ -154,8 +170,8 @@ def load_seed(db: Session) -> SeedResult:
"public_ref": row["public_ref"],
"customer_id": customer_id_by_ref[row["customer_ref"]],
"vehicle_id": vehicle_id_by_ref[row["vehicle_ref"]],
"starts_at": _parse_dt(row["starts_at"]),
"ends_at": _parse_dt(row["ends_at"]),
"starts_at": _parse_dt(row["starts_at"]) + shift,
"ends_at": _parse_dt(row["ends_at"]) + shift,
"status": row["status"],
"start_odometer_km": _parse_optional_int(row["start_odometer_km"]),
"end_odometer_km": _parse_optional_int(row["end_odometer_km"]),
@@ -179,7 +195,7 @@ def load_seed(db: Session) -> SeedResult:
"damage_reported": _parse_bool(row["damage_reported"]),
"technical_warning": _parse_bool(row["technical_warning"]),
"odometer_km": int(row["odometer_km"]),
"completed_at": _parse_dt(row["completed_at"]),
"completed_at": _parse_dt(row["completed_at"]) + shift,
"completed_by": None,
}
)
@@ -193,7 +209,7 @@ def load_seed(db: Session) -> SeedResult:
"id": uuid.uuid4(),
"public_ref": row["public_ref"],
"vehicle_id": vehicle_id_by_ref[row["vehicle_ref"]],
"occurred_at": _parse_dt(row["occurred_at"]),
"occurred_at": _parse_dt(row["occurred_at"]) + shift,
"odometer_km": int(row["odometer_km"]),
"category": row["category"],
"summary": row["summary"],
@@ -266,7 +282,7 @@ def load_seed(db: Session) -> SeedResult:
},
"aggregate_ref": row["aggregate_ref"],
},
"occurred_at": _parse_dt(row["occurred_at"]),
"occurred_at": _parse_dt(row["occurred_at"]) + shift,
"delivery_status": row["status"],
"attempts": int(row["attempts"]),
"next_attempt_at": None,
@@ -277,7 +293,21 @@ def load_seed(db: Session) -> SeedResult:
db.execute(insert(OutboxEvent), outbox_rows)
counts["workflow_runs"] = len(outbox_rows)
return SeedResult(counts=counts)
seeded_at = datetime.now(UTC)
record_audit_event(
db,
actor_type="system",
actor_label="seed loader",
action="demo_data_seeded",
entity_type="system",
metadata={
"anchor_date": today.isoformat(),
"seed_authored_anchor": SEED_AUTHORED_ANCHOR.isoformat(),
"counts": counts,
},
)
return SeedResult(counts=counts, anchor_date=today, seeded_at=seeded_at)
def reset_and_seed(db: Session) -> SeedResult: