NuklearRabbit
6f77a30dce
The demo seed plants exactly one failed delivery (BK-H-0020) to demonstrate retry and audit. Because derive_n8n_status() counted any failure, every fresh reset pinned the n8n integration to "degraded" -- the demo showed a warning about a prop, which tells a viewer something untrue about the automation. The seeded failure now carries its own error code, demoScenarioTimeout, rather than the generic connectionError a real timeout produces. No column and no migration: last_error_code already existed, is already surfaced to the UI and is already localizable. - integration status splits failed into unexpected_failed and demo_scenario_failed; only unexpected failures may move the state. A staged failure alone leaves n8n operational. - latest_failure_at is a health signal and now ignores the staged failure; latest_demo_scenario_at reports it separately. - /api/v1/workflows exposes is_demo_scenario. The Automation page labels the run as a prepared demo scenario, explains that it is a simulated temporary failure that does not affect automation health, and offers a distinct "retry demo scenario" action. Translated in nl-BE, en-GB and fr-BE. - the carve-out stays narrow: a real failure still degrades n8n, and a genuine later failure of the same event overwrites the demo code with the real one. - the retry itself is unchanged and real: the event goes back on the outbox and the dispatcher delivers it to n8n like any other, so 19+1 becomes 20+0 only on an actual round trip. The audit records which kind of failure was retried. Tests that assert on the seeded scenario now reseed first, since earlier test files legitimately mutate the outbox and the suite shares one database. Verified locally against a real PostgreSQL 16: 181 passed, ruff clean, mypy clean (50 files), tsc clean, frontend build clean. Not deployed and not browser-verified.
190 lines
7.1 KiB
Python
190 lines
7.1 KiB
Python
from sqlalchemy import select
|
|
|
|
from app.core.db import SessionLocal
|
|
from app.models.outbox import OutboxEvent
|
|
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_integration_status_requires_operations_manager(employee_client):
|
|
response = employee_client.get("/api/v1/integrations/status")
|
|
assert response.status_code == 403
|
|
|
|
|
|
def test_integration_status_requires_authentication(client):
|
|
response = client.get("/api/v1/integrations/status")
|
|
assert response.status_code == 401
|
|
|
|
|
|
def test_integration_status_reflects_seeded_mixed_outcomes(ops_client):
|
|
_reseed()
|
|
response = ops_client.get("/api/v1/integrations/status")
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
|
|
n8n = body["n8n"]
|
|
assert n8n["dispatch_enabled"] is True
|
|
assert n8n["succeeded"] >= 1
|
|
# The seeded failure stays visible and counted...
|
|
assert n8n["failed"] >= 1
|
|
assert n8n["demo_scenario_failed"] == 1
|
|
# ...but it is a prepared prop, so it is not an unexpected failure and must not
|
|
# move the integration off "operational". A staged failure that degrades the
|
|
# health badge tells a viewer something untrue about the automation.
|
|
assert n8n["unexpected_failed"] == 0
|
|
assert n8n["state"] == "operational"
|
|
assert n8n["latest_success_at"] is not None
|
|
# "latest failure" is a health signal, so the staged one never sets it; it is
|
|
# reported separately instead.
|
|
assert n8n["latest_failure_at"] is None
|
|
assert n8n["latest_demo_scenario_at"] is not None
|
|
|
|
mcp_hub = body["mcp_hub"]
|
|
assert mcp_hub["registration_enabled"] is False
|
|
assert mcp_hub["state"] == "not_configured"
|
|
|
|
|
|
def test_a_real_failure_still_degrades_the_integration(ops_client):
|
|
"""The demo carve-out must be narrow: a failure that is not the prepared scenario
|
|
still degrades n8n, otherwise this change would hide real breakage."""
|
|
_reseed()
|
|
db = SessionLocal()
|
|
try:
|
|
real_failure = db.scalar(
|
|
select(OutboxEvent).where(OutboxEvent.delivery_status == "succeeded").limit(1)
|
|
)
|
|
assert real_failure is not None
|
|
restore = (real_failure.delivery_status, real_failure.last_error_code)
|
|
real_failure.delivery_status = "failed"
|
|
real_failure.last_error_code = "connectionError"
|
|
db.commit()
|
|
|
|
n8n = ops_client.get("/api/v1/integrations/status").json()["n8n"]
|
|
assert n8n["unexpected_failed"] == 1
|
|
assert n8n["state"] == "degraded"
|
|
assert n8n["latest_failure_at"] is not None
|
|
finally:
|
|
real_failure.delivery_status, real_failure.last_error_code = restore
|
|
db.commit()
|
|
db.close()
|
|
|
|
|
|
def test_prepared_demo_failure_is_reset_back_by_a_demo_reset(ops_client):
|
|
"""A demo reset must recreate the intended 19 succeeded + 1 prepared failure, so the
|
|
scenario can be shown again after it has been retried away."""
|
|
_reseed()
|
|
|
|
n8n = ops_client.get("/api/v1/integrations/status").json()["n8n"]
|
|
assert n8n["succeeded"] == 19
|
|
assert n8n["failed"] == 1
|
|
assert n8n["demo_scenario_failed"] == 1
|
|
assert n8n["unexpected_failed"] == 0
|
|
assert n8n["state"] == "operational"
|
|
|
|
|
|
def test_integration_status_is_operational_once_all_failed_events_resolved(ops_client):
|
|
_reseed()
|
|
failed = ops_client.get("/api/v1/workflows", params={"status": "failed"}).json()
|
|
for run in failed:
|
|
retried = ops_client.post(f"/api/v1/workflows/{run['event_id']}/retry")
|
|
assert retried.status_code == 200
|
|
|
|
response = ops_client.get("/api/v1/integrations/status")
|
|
body = response.json()["n8n"]
|
|
assert body["failed"] == 0
|
|
assert body["state"] == "operational"
|
|
|
|
|
|
def test_integration_status_lists_all_four_canonical_workflows(ops_client):
|
|
body = ops_client.get("/api/v1/integrations/status").json()["n8n"]
|
|
assert body["expected_workflow_count"] == 4
|
|
names = {w["name"] for w in body["workflows"]}
|
|
assert names == {
|
|
"Fleet Ops — Vehicle Return Orchestration",
|
|
"Fleet Ops — Scheduled Data Quality Scan",
|
|
"Fleet Ops — RAGcore Procedure Sync",
|
|
"Fleet Ops — Workflow Error Handler",
|
|
}
|
|
ragcore_sync = next(w for w in body["workflows"] if "RAGcore" in w["name"])
|
|
# All 4 canonical workflows are built (RAGcore Procedure Sync has all 6 nodes saved
|
|
# live); it has no *run* evidence yet since it is deliberately left unpublished.
|
|
assert ragcore_sync["built"] is True
|
|
assert ragcore_sync["last_seen_at"] is None
|
|
|
|
|
|
def test_integration_status_scheduled_scan_evidence_only_counts_service_runs(client, ops_client):
|
|
from app.core.config import get_settings
|
|
|
|
settings = get_settings()
|
|
|
|
before = ops_client.get("/api/v1/integrations/status").json()["n8n"]
|
|
scan_workflow = next(
|
|
w for w in before["workflows"] if w["name"].endswith("Scheduled Data Quality Scan")
|
|
)
|
|
assert scan_workflow["last_seen_at"] is None
|
|
|
|
scan = client.post(
|
|
"/api/v1/integrations/n8n/scheduled-scan",
|
|
headers={"X-Service-Token": settings.n8n_callback_token},
|
|
)
|
|
assert scan.status_code == 200
|
|
|
|
after = ops_client.get("/api/v1/integrations/status").json()["n8n"]
|
|
scan_workflow = next(
|
|
w for w in after["workflows"] if w["name"].endswith("Scheduled Data Quality Scan")
|
|
)
|
|
assert scan_workflow["last_seen_at"] is not None
|
|
assert after["known_workflow_count"] > before["known_workflow_count"]
|
|
|
|
|
|
def test_integration_status_reflects_error_handler_registrations(client, ops_client):
|
|
import uuid
|
|
|
|
from app.core.config import get_settings
|
|
|
|
settings = get_settings()
|
|
before = ops_client.get("/api/v1/integrations/status").json()["n8n"]
|
|
|
|
execution_id = str(uuid.uuid4())
|
|
report = client.post(
|
|
"/api/v1/integrations/n8n/workflow-error",
|
|
json={
|
|
"workflow_id": "mobilityops-return-processing",
|
|
"workflow_name": "Fleet Ops — Vehicle Return Orchestration",
|
|
"execution_id": execution_id,
|
|
"failed_at": "2026-08-04T10:15:00Z",
|
|
"error_category": "httpError",
|
|
"error_summary": "Simulated failure for status test",
|
|
"trigger_context": "webhook",
|
|
"attempt": 1,
|
|
},
|
|
headers={"X-Service-Token": settings.n8n_callback_token},
|
|
)
|
|
assert report.status_code == 200
|
|
|
|
after = ops_client.get("/api/v1/integrations/status").json()["n8n"]
|
|
assert (
|
|
after["error_handler"]["total_failures_registered"]
|
|
== before["error_handler"]["total_failures_registered"] + 1
|
|
)
|
|
assert after["error_handler"]["latest_failure_workflow"] == (
|
|
"Fleet Ops — Vehicle Return Orchestration"
|
|
)
|
|
handler_workflow = next(
|
|
w for w in after["workflows"] if w["name"].endswith("Workflow Error Handler")
|
|
)
|
|
assert handler_workflow["last_seen_at"] is not None
|