NuklearRabbit
e115031a57
The original docs described two n8n workflows but the repository only ever shipped one (return-processing); the sketched second workflow (knowledge sync) depends on RAGcore, which isn't connected here, so it stays deferred. Add POST /api/v1/integrations/n8n/scheduled-scan (X-Service-Token protected, same pattern as the return callback), calling the same run_scan() the manual "Run quality scan" UI action uses and recording a service-actor data_quality_scan_run audit event. run_scan() already only creates an issue for a condition without one open, so overlapping triggers do no duplicate domain work. n8n/mobilityops-scheduled-quality-scan.json (hourly schedule + manual test trigger, both feeding the same HTTP call) ships "active": false so it can't fire anywhere until deliberately published. Verified live against the local n8n instance via the Manual test trigger: full green execution, and the resulting data_quality_scan_run audit event (actor_type=service, actor_label="n8n scheduled scan") confirms the real round trip, not just a contract test. deploy/unraid/setup-scheduled-scan.sh mirrors the existing return-workflow publish script for the shared Unraid n8n.
125 lines
4.1 KiB
Python
125 lines
4.1 KiB
Python
import uuid
|
|
|
|
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):
|
|
response = client.post(
|
|
"/api/v1/integrations/n8n/return-callback",
|
|
json={"follow_up": "cleaning"},
|
|
headers=_callback_headers(str(uuid.uuid4()), token="wrong-token"),
|
|
)
|
|
assert response.status_code == 401
|
|
|
|
|
|
def test_callback_unknown_event_returns_404(client):
|
|
response = client.post(
|
|
"/api/v1/integrations/n8n/return-callback",
|
|
json={"follow_up": "cleaning"},
|
|
headers=_callback_headers(str(uuid.uuid4())),
|
|
)
|
|
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))
|
|
event = OutboxEvent(
|
|
event_id=uuid.uuid4(),
|
|
event_type="vehicle.returned.v1",
|
|
aggregate_type="booking",
|
|
aggregate_id=booking.id,
|
|
payload_json={
|
|
"correlation_id": str(uuid.uuid4()),
|
|
"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={"follow_up": "cleaning", "summary": "test"},
|
|
headers=_callback_headers(event_id),
|
|
)
|
|
second = client.post(
|
|
"/api/v1/integrations/n8n/return-callback",
|
|
json={"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
|
|
|
|
|
|
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"] == {}
|