NuklearRabbitandClaude Sonnet 5
4049c0c6b1
Fleet Ops integration status no longer depends only on a config boolean or the most recent outbox event: N8nIntegrationStatus now reports per-canonical-workflow evidence (last successful outbox delivery for the return workflow, latest service-triggered data_quality_scan_run for the scan workflow, latest n8n_workflow_failure_registered for the error handler, and "not built" for the still-blocked RAGcore sync), plus an error-handler summary (total failures registered, latest failure + which workflow). Automation page renders this as a localized workflow table (EN/NL/FR) with technical workflow names tucked under a "Technical details" disclosure, matching the existing progressive-disclosure pattern. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
121 lines
4.5 KiB
Python
121 lines
4.5 KiB
Python
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):
|
|
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
|
|
assert n8n["failed"] >= 1
|
|
# The seed deliberately carries both failed and succeeded events, so a single most-
|
|
# recent-event read would misreport health -- the aggregate must call this "degraded",
|
|
# not "operational" or "unavailable".
|
|
assert n8n["state"] == "degraded"
|
|
assert n8n["latest_success_at"] is not None
|
|
assert n8n["latest_failure_at"] is not None
|
|
|
|
mcp_hub = body["mcp_hub"]
|
|
assert mcp_hub["registration_enabled"] is False
|
|
assert mcp_hub["state"] == "not_configured"
|
|
|
|
|
|
def test_integration_status_is_operational_once_all_failed_events_resolved(ops_client):
|
|
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"])
|
|
assert ragcore_sync["built"] is False
|
|
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
|