96 lines
3.7 KiB
Python
96 lines
3.7 KiB
Python
from app.core.db import SessionLocal
|
|
from app.seed_loader import reset_and_seed
|
|
|
|
|
|
def _reseed() -> None:
|
|
"""Restore the canonical demo dataset (19 succeeded + 1 prepared failure).
|
|
|
|
The suite shares one session-scoped database and earlier files legitimately mutate
|
|
the outbox, so any test that asserts on the *seeded* scenario has to re-establish it
|
|
rather than depend on file ordering.
|
|
"""
|
|
db = SessionLocal()
|
|
try:
|
|
reset_and_seed(db)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_list_workflows_requires_operations_manager(employee_client):
|
|
response = employee_client.get("/api/v1/workflows")
|
|
assert response.status_code == 403
|
|
|
|
|
|
def test_list_workflows_includes_seeded_failed_run(ops_client):
|
|
_reseed()
|
|
response = ops_client.get("/api/v1/workflows", params={"status": "failed"})
|
|
assert response.status_code == 200
|
|
runs = response.json()
|
|
assert len(runs) >= 1
|
|
assert all(r["status"] == "failed" for r in runs)
|
|
|
|
|
|
def test_retry_requires_failed_status(ops_client):
|
|
succeeded = ops_client.get("/api/v1/workflows", params={"status": "succeeded"}).json()
|
|
target = succeeded[0]["event_id"]
|
|
response = ops_client.post(f"/api/v1/workflows/{target}/retry")
|
|
assert response.status_code == 409
|
|
assert response.json()["error"]["code"] == "NOT_RETRYABLE"
|
|
|
|
|
|
def test_retry_failed_run_moves_to_pending_and_audits(ops_client):
|
|
_reseed()
|
|
failed = ops_client.get("/api/v1/workflows", params={"status": "failed"}).json()
|
|
target = failed[0]["event_id"]
|
|
|
|
response = ops_client.post(f"/api/v1/workflows/{target}/retry")
|
|
assert response.status_code == 200
|
|
assert response.json()["status"] == "pending"
|
|
|
|
audit_events = ops_client.get("/api/v1/audit", params={"action": "workflow_retry"}).json()
|
|
assert len(audit_events) >= 1
|
|
|
|
|
|
def test_retry_requires_operations_manager(employee_client):
|
|
response = employee_client.post("/api/v1/workflows/00000000-0000-4000-8000-000000000020/retry")
|
|
assert response.status_code == 403
|
|
|
|
|
|
def test_seeded_failure_is_labelled_as_a_prepared_demo_scenario(ops_client):
|
|
"""The one seeded failure must announce itself as staged. An unexplained red row in
|
|
a demo reads as a broken product; a labelled one reads as the retry story it is."""
|
|
_reseed()
|
|
runs = ops_client.get("/api/v1/workflows", params={"status": "failed"}).json()
|
|
demo_runs = [run for run in runs if run["is_demo_scenario"]]
|
|
assert len(demo_runs) == 1
|
|
assert demo_runs[0]["last_error_code"] == "demoScenarioTimeout"
|
|
assert demo_runs[0]["aggregate_ref"] == "BK-H-0020"
|
|
|
|
|
|
def test_succeeded_runs_are_never_marked_as_a_demo_scenario(ops_client):
|
|
runs = ops_client.get("/api/v1/workflows", params={"status": "succeeded"}).json()
|
|
assert runs
|
|
assert all(run["is_demo_scenario"] is False for run in runs)
|
|
|
|
|
|
def test_retry_of_the_demo_scenario_is_audited_as_a_demo_scenario(ops_client):
|
|
"""The retry is a real redelivery either way; the audit records which kind of
|
|
failure it resolved so a staged retry is never mistaken for a production fix."""
|
|
_reseed()
|
|
failed = ops_client.get("/api/v1/workflows", params={"status": "failed"}).json()
|
|
demo_run = next(run for run in failed if run["is_demo_scenario"])
|
|
|
|
response = ops_client.post(f"/api/v1/workflows/{demo_run['event_id']}/retry")
|
|
assert response.status_code == 200
|
|
assert response.json()["status"] == "pending"
|
|
|
|
audit = ops_client.get("/api/v1/audit", params={"action": "workflow_retry"}).json()
|
|
entry = next(
|
|
event
|
|
for event in audit
|
|
if (event.get("metadata") or event.get("metadata_json") or {}).get("event_id")
|
|
== demo_run["event_id"]
|
|
)
|
|
metadata = entry.get("metadata") or entry.get("metadata_json") or {}
|
|
assert metadata["demo_scenario"] is True
|