455 lines
16 KiB
Python
455 lines
16 KiB
Python
import uuid
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
|
|
from app.core.config import get_settings
|
|
|
|
|
|
def _callback_headers(event_id: str, token: str | None = None):
|
|
settings = get_settings()
|
|
return {
|
|
"Idempotency-Key": event_id,
|
|
"X-Service-Token": token if token is not None else settings.n8n_callback_token,
|
|
}
|
|
|
|
|
|
def test_callback_rejects_wrong_service_token(client):
|
|
event_id = str(uuid.uuid4())
|
|
response = client.post(
|
|
"/api/v1/integrations/n8n/return-callback",
|
|
json={
|
|
"event_id": event_id,
|
|
"correlation_id": str(uuid.uuid4()),
|
|
"follow_up": "cleaning",
|
|
},
|
|
headers=_callback_headers(event_id, token="wrong-token"),
|
|
)
|
|
assert response.status_code == 401
|
|
|
|
|
|
def test_callback_unknown_event_returns_404(client):
|
|
event_id = str(uuid.uuid4())
|
|
response = client.post(
|
|
"/api/v1/integrations/n8n/return-callback",
|
|
json={
|
|
"event_id": event_id,
|
|
"correlation_id": str(uuid.uuid4()),
|
|
"follow_up": "cleaning",
|
|
},
|
|
headers=_callback_headers(event_id),
|
|
)
|
|
assert response.status_code == 404
|
|
|
|
|
|
def test_callback_is_idempotent_by_event_id(client, ops_client):
|
|
from sqlalchemy import select
|
|
|
|
from app.core.db import SessionLocal
|
|
from app.models.booking import Booking
|
|
from app.models.outbox import OutboxEvent
|
|
|
|
db = SessionLocal()
|
|
try:
|
|
booking = db.scalar(select(Booking).limit(1))
|
|
correlation_id = uuid.uuid4()
|
|
event = OutboxEvent(
|
|
event_id=uuid.uuid4(),
|
|
event_type="vehicle.returned.v1",
|
|
aggregate_type="booking",
|
|
aggregate_id=booking.id,
|
|
payload_json={
|
|
"correlation_id": str(correlation_id),
|
|
"aggregate": {
|
|
"type": "booking",
|
|
"id": str(booking.id),
|
|
"public_ref": booking.public_ref,
|
|
},
|
|
"data": {},
|
|
"aggregate_ref": booking.public_ref,
|
|
},
|
|
occurred_at=db.execute(select(Booking.starts_at).limit(1)).scalar(),
|
|
delivery_status="delivering",
|
|
attempts=1,
|
|
)
|
|
db.add(event)
|
|
db.commit()
|
|
event_id = str(event.event_id)
|
|
finally:
|
|
db.close()
|
|
|
|
first = client.post(
|
|
"/api/v1/integrations/n8n/return-callback",
|
|
json={
|
|
"event_id": event_id,
|
|
"correlation_id": str(correlation_id),
|
|
"follow_up": "cleaning",
|
|
"summary": "test",
|
|
},
|
|
headers=_callback_headers(event_id),
|
|
)
|
|
second = client.post(
|
|
"/api/v1/integrations/n8n/return-callback",
|
|
json={
|
|
"event_id": event_id,
|
|
"correlation_id": str(correlation_id),
|
|
"follow_up": "cleaning",
|
|
"summary": "test",
|
|
},
|
|
headers=_callback_headers(event_id),
|
|
)
|
|
assert first.status_code == 200
|
|
assert second.status_code == 200
|
|
|
|
audit_events = ops_client.get(
|
|
"/api/v1/audit", params={"action": "n8n_return_followup_recorded"}
|
|
).json()
|
|
matching = [e for e in audit_events if e["metadata"]["event_id"] == event_id]
|
|
assert len(matching) == 1
|
|
assert matching[0]["correlation_id"] == str(correlation_id)
|
|
|
|
|
|
def test_callback_rejects_crossed_event_or_correlation(client):
|
|
from sqlalchemy import select
|
|
|
|
from app.core.db import SessionLocal
|
|
from app.models.outbox import OutboxEvent
|
|
|
|
with SessionLocal() as db:
|
|
event = db.scalar(select(OutboxEvent).limit(1))
|
|
event_id = str(event.event_id)
|
|
correlation_id = event.payload_json["correlation_id"]
|
|
|
|
crossed_event = client.post(
|
|
"/api/v1/integrations/n8n/return-callback",
|
|
json={
|
|
"event_id": str(uuid.uuid4()),
|
|
"correlation_id": correlation_id,
|
|
"follow_up": "cleaning",
|
|
},
|
|
headers=_callback_headers(event_id),
|
|
)
|
|
assert crossed_event.status_code == 409
|
|
assert crossed_event.json()["error"]["code"] == "CALLBACK_EVENT_MISMATCH"
|
|
|
|
crossed_correlation = client.post(
|
|
"/api/v1/integrations/n8n/return-callback",
|
|
json={
|
|
"event_id": event_id,
|
|
"correlation_id": str(uuid.uuid4()),
|
|
"follow_up": "cleaning",
|
|
},
|
|
headers=_callback_headers(event_id),
|
|
)
|
|
assert crossed_correlation.status_code == 409
|
|
assert crossed_correlation.json()["error"]["code"] == "CALLBACK_CORRELATION_MISMATCH"
|
|
|
|
|
|
def test_callback_concurrent_retries_record_one_audit(client, ops_client):
|
|
from sqlalchemy import select
|
|
|
|
from app.core.db import SessionLocal
|
|
from app.models.outbox import OutboxEvent
|
|
|
|
with SessionLocal() as db:
|
|
event = db.scalar(select(OutboxEvent).limit(1))
|
|
event_id = str(event.event_id)
|
|
correlation_id = event.payload_json["correlation_id"]
|
|
body = {
|
|
"event_id": event_id,
|
|
"correlation_id": correlation_id,
|
|
"follow_up": "cleaning",
|
|
}
|
|
|
|
def post_callback(_index: int):
|
|
return client.post(
|
|
"/api/v1/integrations/n8n/return-callback",
|
|
json=body,
|
|
headers=_callback_headers(event_id),
|
|
)
|
|
|
|
with ThreadPoolExecutor(max_workers=2) as pool:
|
|
responses = list(pool.map(post_callback, range(2)))
|
|
assert [response.status_code for response in responses] == [200, 200]
|
|
matching = [
|
|
event
|
|
for event in ops_client.get(
|
|
"/api/v1/audit", params={"action": "n8n_return_followup_recorded"}
|
|
).json()
|
|
if event["metadata"]["event_id"] == event_id
|
|
]
|
|
assert len(matching) == 1
|
|
|
|
|
|
def test_scheduled_scan_rejects_wrong_service_token(client):
|
|
response = client.post(
|
|
"/api/v1/integrations/n8n/scheduled-scan",
|
|
headers={"X-Service-Token": "wrong-token"},
|
|
)
|
|
assert response.status_code == 401
|
|
|
|
|
|
def test_scheduled_scan_requires_service_token_header(client):
|
|
response = client.post("/api/v1/integrations/n8n/scheduled-scan")
|
|
assert response.status_code == 422
|
|
|
|
|
|
def test_scheduled_scan_runs_and_returns_counts_by_rule(client, ops_client):
|
|
settings = get_settings()
|
|
response = client.post(
|
|
"/api/v1/integrations/n8n/scheduled-scan",
|
|
headers={"X-Service-Token": settings.n8n_callback_token},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json() == {"created": {}} # already-seeded conditions, nothing new
|
|
|
|
audit_events = ops_client.get(
|
|
"/api/v1/audit", params={"action": "data_quality_scan_run"}
|
|
).json()
|
|
service_triggered = [e for e in audit_events if e["actor_type"] == "service"]
|
|
assert len(service_triggered) >= 1
|
|
assert service_triggered[0]["actor_label"] == "n8n scheduled scan"
|
|
|
|
|
|
def test_scheduled_scan_is_idempotent_across_repeated_triggers(client):
|
|
settings = get_settings()
|
|
headers = {"X-Service-Token": settings.n8n_callback_token}
|
|
first = client.post("/api/v1/integrations/n8n/scheduled-scan", headers=headers)
|
|
second = client.post("/api/v1/integrations/n8n/scheduled-scan", headers=headers)
|
|
assert first.status_code == 200
|
|
assert second.status_code == 200
|
|
assert second.json()["created"] == {}
|
|
|
|
|
|
def _workflow_error_body(execution_id: str, **overrides):
|
|
body = {
|
|
"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": "Callback request failed with status 500",
|
|
"trigger_context": "webhook",
|
|
"correlation_id": None,
|
|
"attempt": 1,
|
|
"retry_action": "n8n will retry automatically",
|
|
}
|
|
body.update(overrides)
|
|
return body
|
|
|
|
|
|
def test_workflow_error_rejects_wrong_service_token(client):
|
|
response = client.post(
|
|
"/api/v1/integrations/n8n/workflow-error",
|
|
json=_workflow_error_body(str(uuid.uuid4())),
|
|
headers={"X-Service-Token": "wrong-token"},
|
|
)
|
|
assert response.status_code == 401
|
|
|
|
|
|
def test_workflow_error_rejects_unknown_category(client):
|
|
settings = get_settings()
|
|
response = client.post(
|
|
"/api/v1/integrations/n8n/workflow-error",
|
|
json=_workflow_error_body(str(uuid.uuid4()), error_category="somethingElse"),
|
|
headers={"X-Service-Token": settings.n8n_callback_token},
|
|
)
|
|
assert response.status_code == 422
|
|
|
|
|
|
def test_workflow_error_registers_and_is_idempotent_by_execution_id(client, ops_client):
|
|
settings = get_settings()
|
|
headers = {"X-Service-Token": settings.n8n_callback_token}
|
|
execution_id = str(uuid.uuid4())
|
|
body = _workflow_error_body(execution_id)
|
|
|
|
first = client.post("/api/v1/integrations/n8n/workflow-error", json=body, headers=headers)
|
|
second = client.post("/api/v1/integrations/n8n/workflow-error", json=body, headers=headers)
|
|
|
|
assert first.status_code == 200
|
|
assert first.json()["status"] == "registered"
|
|
assert second.status_code == 200
|
|
assert second.json()["status"] == "already_registered"
|
|
|
|
audit_events = ops_client.get(
|
|
"/api/v1/audit", params={"action": "n8n_workflow_failure_registered"}
|
|
).json()
|
|
matching = [e for e in audit_events if e["metadata"]["execution_id"] == execution_id]
|
|
assert len(matching) == 1
|
|
assert matching[0]["after"]["error_category"] == "httpError"
|
|
assert matching[0]["after"]["retry_action"] == "n8n will retry automatically"
|
|
|
|
|
|
def test_workflow_error_preserves_correlation_and_rejects_malformed(client, ops_client):
|
|
settings = get_settings()
|
|
headers = {"X-Service-Token": settings.n8n_callback_token}
|
|
execution_id = str(uuid.uuid4())
|
|
correlation_id = str(uuid.uuid4())
|
|
accepted = client.post(
|
|
"/api/v1/integrations/n8n/workflow-error",
|
|
json=_workflow_error_body(execution_id, correlation_id=correlation_id),
|
|
headers=headers,
|
|
)
|
|
assert accepted.status_code == 200
|
|
matching = [
|
|
event
|
|
for event in ops_client.get(
|
|
"/api/v1/audit", params={"action": "n8n_workflow_failure_registered"}
|
|
).json()
|
|
if event["metadata"]["execution_id"] == execution_id
|
|
]
|
|
assert len(matching) == 1
|
|
assert matching[0]["correlation_id"] == correlation_id
|
|
|
|
malformed_execution = str(uuid.uuid4())
|
|
malformed = client.post(
|
|
"/api/v1/integrations/n8n/workflow-error",
|
|
json=_workflow_error_body(malformed_execution, correlation_id="not-a-uuid"),
|
|
headers=headers,
|
|
)
|
|
assert malformed.status_code == 422
|
|
assert all(
|
|
event["metadata"]["execution_id"] != malformed_execution
|
|
for event in ops_client.get(
|
|
"/api/v1/audit", params={"action": "n8n_workflow_failure_registered"}
|
|
).json()
|
|
)
|
|
|
|
|
|
def test_workflow_error_bounds_summary_length(client):
|
|
settings = get_settings()
|
|
response = client.post(
|
|
"/api/v1/integrations/n8n/workflow-error",
|
|
json=_workflow_error_body(str(uuid.uuid4()), error_summary="x" * 501),
|
|
headers={"X-Service-Token": settings.n8n_callback_token},
|
|
)
|
|
assert response.status_code == 422
|
|
|
|
|
|
def test_procedures_rejects_wrong_service_token(client):
|
|
response = client.get(
|
|
"/api/v1/integrations/n8n/procedures", headers={"X-Service-Token": "wrong-token"}
|
|
)
|
|
assert response.status_code == 401
|
|
|
|
|
|
def test_procedures_lists_every_language_with_stable_ids(client):
|
|
settings = get_settings()
|
|
response = client.get(
|
|
"/api/v1/integrations/n8n/procedures",
|
|
headers={"X-Service-Token": settings.n8n_callback_token},
|
|
)
|
|
assert response.status_code == 200
|
|
documents = response.json()["documents"]
|
|
assert len(documents) > 0
|
|
assert {d["language"] for d in documents} == {"en-GB", "nl-BE", "fr-BE"}
|
|
checkout_docs = [d for d in documents if d["document_id"] == "vehicle-checkout-procedure"]
|
|
assert len(checkout_docs) == 3 # one per language
|
|
assert all(d["content"] and d["content_hash"] for d in checkout_docs)
|
|
# Same document_id, different language, must not collide on id.
|
|
assert len({d["id"] for d in checkout_docs}) == 3
|
|
|
|
second_response = client.get(
|
|
"/api/v1/integrations/n8n/procedures",
|
|
headers={"X-Service-Token": settings.n8n_callback_token},
|
|
)
|
|
second_ids = {d["id"] for d in second_response.json()["documents"]}
|
|
assert second_ids == {d["id"] for d in documents} # ids are stable across requests
|
|
|
|
|
|
def test_procedures_sync_result_rejects_wrong_service_token(client):
|
|
response = client.post(
|
|
"/api/v1/integrations/n8n/procedures-sync-result",
|
|
json={"execution_id": str(uuid.uuid4()), "synced": 5, "failed": 0},
|
|
headers={"X-Service-Token": "wrong-token"},
|
|
)
|
|
assert response.status_code == 401
|
|
|
|
|
|
def test_procedures_sync_result_registers_and_is_idempotent(client, ops_client):
|
|
settings = get_settings()
|
|
headers = {"X-Service-Token": settings.n8n_callback_token}
|
|
execution_id = str(uuid.uuid4())
|
|
body = {"execution_id": execution_id, "synced": 33, "failed": 1}
|
|
|
|
first = client.post(
|
|
"/api/v1/integrations/n8n/procedures-sync-result", json=body, headers=headers
|
|
)
|
|
second = client.post(
|
|
"/api/v1/integrations/n8n/procedures-sync-result", json=body, headers=headers
|
|
)
|
|
|
|
assert first.status_code == 200
|
|
assert first.json()["status"] == "registered"
|
|
assert second.status_code == 200
|
|
assert second.json()["status"] == "already_registered"
|
|
|
|
audit_events = ops_client.get(
|
|
"/api/v1/audit", params={"action": "n8n_procedures_synced"}
|
|
).json()
|
|
matching = [e for e in audit_events if e["metadata"]["execution_id"] == execution_id]
|
|
assert len(matching) == 1
|
|
assert matching[0]["after"] == {"synced": 33, "failed": 1}
|
|
|
|
# The workflow's own callback is its evidence -- same pattern the scheduled scan and
|
|
# error handler already use -- so this real report must now show up as run evidence
|
|
# in the integration status, not stay hardcoded to "no evidence yet".
|
|
status = ops_client.get("/api/v1/integrations/status").json()["n8n"]
|
|
ragcore_sync = next(w for w in status["workflows"] if "RAGcore" in w["name"])
|
|
assert ragcore_sync["last_seen_at"] is not None
|
|
|
|
|
|
def test_failed_or_partial_procedure_sync_is_never_reported_healthy(client, ops_client):
|
|
settings = get_settings()
|
|
headers = {"X-Service-Token": settings.n8n_callback_token}
|
|
for synced, failed in ((0, 33), (32, 1)):
|
|
execution_id = str(uuid.uuid4())
|
|
response = client.post(
|
|
"/api/v1/integrations/n8n/procedures-sync-result",
|
|
json={"execution_id": execution_id, "synced": synced, "failed": failed},
|
|
headers=headers,
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
status = ops_client.get("/api/v1/integrations/status").json()["n8n"]
|
|
workflow = next(w for w in status["workflows"] if "RAGcore" in w["name"])
|
|
assert workflow["state"] == "failed"
|
|
assert workflow["last_status"] == "failed"
|
|
assert workflow["last_execution_id"] == execution_id
|
|
assert status["state"] == "degraded"
|
|
|
|
|
|
def test_workflow_heartbeat_is_idempotent_and_drives_live_status(client, ops_client):
|
|
settings = get_settings()
|
|
execution_id = str(uuid.uuid4())
|
|
body = {
|
|
"workflow_id": "mobilityops-scheduled-quality-scan",
|
|
"workflow_name": "Fleet Ops — Scheduled Data Quality Scan",
|
|
"execution_id": execution_id,
|
|
"status": "succeeded",
|
|
}
|
|
headers = {"X-Service-Token": settings.n8n_callback_token}
|
|
first = client.post("/api/v1/integrations/n8n/heartbeat", json=body, headers=headers)
|
|
second = client.post("/api/v1/integrations/n8n/heartbeat", json=body, headers=headers)
|
|
assert first.status_code == 200
|
|
assert first.json()["status"] == "registered"
|
|
assert second.json()["status"] == "already_registered"
|
|
|
|
status = ops_client.get("/api/v1/integrations/status").json()["n8n"]
|
|
workflow = next(w for w in status["workflows"] if w["name"] == body["workflow_name"])
|
|
assert workflow["state"] == "healthy"
|
|
assert workflow["last_status"] == "succeeded"
|
|
assert workflow["last_execution_id"] == execution_id
|
|
|
|
|
|
def test_workflow_heartbeat_rejects_unknown_workflow(client):
|
|
response = client.post(
|
|
"/api/v1/integrations/n8n/heartbeat",
|
|
json={
|
|
"workflow_id": "unknown",
|
|
"workflow_name": "Unknown workflow",
|
|
"execution_id": "test-unknown-001",
|
|
"status": "succeeded",
|
|
},
|
|
headers={"X-Service-Token": get_settings().n8n_callback_token},
|
|
)
|
|
assert response.status_code == 422
|