401 lines
14 KiB
Python
401 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
import uuid
|
|
from datetime import UTC, datetime, timedelta
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from sqlalchemy import select
|
|
|
|
from app.core.config import get_settings
|
|
from app.core.db import SessionLocal
|
|
from app.models.booking import Booking
|
|
from app.models.outbox import OutboxEvent
|
|
from app.services import dispatcher
|
|
|
|
|
|
def _make_pending_event(vehicle_ref: str) -> uuid.UUID:
|
|
db = SessionLocal()
|
|
try:
|
|
booking = db.scalar(select(Booking).where(Booking.status == "returned").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": {
|
|
"vehicle_ref": vehicle_ref,
|
|
"inspection_ref": "INSP-TEST",
|
|
"resulting_vehicle_status": "cleaning",
|
|
"attention_reasons": [],
|
|
},
|
|
"aggregate_ref": booking.public_ref,
|
|
},
|
|
occurred_at=datetime.now(UTC),
|
|
delivery_status="pending",
|
|
attempts=0,
|
|
)
|
|
db.add(event)
|
|
db.commit()
|
|
return event.event_id
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def _get_event(event_id: uuid.UUID) -> OutboxEvent:
|
|
db = SessionLocal()
|
|
try:
|
|
return db.scalar(select(OutboxEvent).where(OutboxEvent.event_id == event_id))
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_claim_marks_events_delivering():
|
|
event_id = _make_pending_event("MO-001")
|
|
claimed = dispatcher._claim_due_events()
|
|
assert event_id in claimed
|
|
assert _get_event(event_id).delivery_status == "delivering"
|
|
|
|
|
|
def test_deliver_one_success(monkeypatch):
|
|
event_id = _make_pending_event("MO-002")
|
|
execution_id = "n8n-execution-123"
|
|
dispatcher._claim_due_events()
|
|
|
|
def fake_post(url, json, headers, timeout):
|
|
return SimpleNamespace(
|
|
raise_for_status=lambda: None,
|
|
json=lambda: {
|
|
"ok": True,
|
|
"event_id": str(event_id),
|
|
"result": {"execution_id": execution_id},
|
|
},
|
|
)
|
|
|
|
monkeypatch.setattr(dispatcher.httpx, "post", fake_post)
|
|
dispatcher._deliver_one(event_id)
|
|
|
|
event = _get_event(event_id)
|
|
assert event.delivery_status == "succeeded"
|
|
assert event.attempts == 1
|
|
assert event.external_run_id == execution_id
|
|
assert event.last_error is None
|
|
assert event.last_error_code is None
|
|
|
|
|
|
def test_deliver_one_failure_schedules_retry(monkeypatch):
|
|
event_id = _make_pending_event("MO-003")
|
|
dispatcher._claim_due_events()
|
|
|
|
def fake_post(url, json, headers, timeout):
|
|
raise dispatcher.httpx.ConnectError("simulated connection failure")
|
|
|
|
monkeypatch.setattr(dispatcher.httpx, "post", fake_post)
|
|
dispatcher._deliver_one(event_id)
|
|
|
|
event = _get_event(event_id)
|
|
assert event.delivery_status == "pending"
|
|
assert event.attempts == 1
|
|
assert event.next_attempt_at is not None
|
|
assert "simulated connection failure" in event.last_error
|
|
assert event.last_error_code == "connectionError"
|
|
|
|
|
|
def test_deliver_one_treats_empty_2xx_body_as_failure(monkeypatch):
|
|
# Reproduces a real failure mode found while live-validating the n8n webhook auth
|
|
# fix: a workflow that errors internally before its "Respond to Webhook" node runs
|
|
# can still answer with a 2xx status and an empty body. response.json() on that body
|
|
# raises json.JSONDecodeError -- this must be treated as a retryable failure, not an
|
|
# unhandled exception that leaves the event stuck in "delivering" forever.
|
|
event_id = _make_pending_event("MO-005")
|
|
dispatcher._claim_due_events()
|
|
|
|
def fake_post(url, json, headers, timeout):
|
|
def raise_json_error():
|
|
raise ValueError("Expecting value: line 1 column 1 (char 0)")
|
|
|
|
return SimpleNamespace(
|
|
raise_for_status=lambda: None, json=raise_json_error, status_code=200
|
|
)
|
|
|
|
monkeypatch.setattr(dispatcher.httpx, "post", fake_post)
|
|
dispatcher._deliver_one(event_id)
|
|
|
|
event = _get_event(event_id)
|
|
assert event.delivery_status == "pending"
|
|
assert event.attempts == 1
|
|
assert event.next_attempt_at is not None
|
|
assert event.last_error_code == "malformedResponse"
|
|
|
|
|
|
def test_deliver_one_rejects_json_object_without_explicit_success_ack(monkeypatch):
|
|
event_id = _make_pending_event("MO-010")
|
|
dispatcher._claim_due_events()
|
|
|
|
def fake_post(url, json, headers, timeout):
|
|
return SimpleNamespace(
|
|
raise_for_status=lambda: None,
|
|
json=lambda: {},
|
|
status_code=200,
|
|
)
|
|
|
|
monkeypatch.setattr(dispatcher.httpx, "post", fake_post)
|
|
dispatcher._deliver_one(event_id)
|
|
|
|
event = _get_event(event_id)
|
|
assert event.delivery_status == "pending"
|
|
assert event.attempts == 1
|
|
assert event.last_error_code == "malformedResponse"
|
|
assert event.external_run_id is None
|
|
|
|
|
|
@pytest.mark.parametrize("result", [{}, {"execution_id": 123}, {"execution_id": " "}])
|
|
def test_deliver_one_rejects_ack_without_valid_execution_id(monkeypatch, result):
|
|
event_id = _make_pending_event("MO-EXEC-ID")
|
|
dispatcher._claim_due_events()
|
|
|
|
def fake_post(url, json, headers, timeout):
|
|
return SimpleNamespace(
|
|
raise_for_status=lambda: None,
|
|
json=lambda: {"ok": True, "event_id": str(event_id), "result": result},
|
|
status_code=200,
|
|
)
|
|
|
|
monkeypatch.setattr(dispatcher.httpx, "post", fake_post)
|
|
dispatcher._deliver_one(event_id)
|
|
event = _get_event(event_id)
|
|
assert event.delivery_status == "pending"
|
|
assert event.last_error_code == "malformedResponse"
|
|
assert event.external_run_id is None
|
|
|
|
|
|
def test_deliver_one_rejects_success_ack_for_a_different_event(monkeypatch):
|
|
event_id = _make_pending_event("MO-011")
|
|
dispatcher._claim_due_events()
|
|
other_event_id = uuid.uuid4()
|
|
|
|
def fake_post(url, json, headers, timeout):
|
|
return SimpleNamespace(
|
|
raise_for_status=lambda: None,
|
|
json=lambda: {"ok": True, "event_id": str(other_event_id)},
|
|
status_code=200,
|
|
)
|
|
|
|
monkeypatch.setattr(dispatcher.httpx, "post", fake_post)
|
|
dispatcher._deliver_one(event_id)
|
|
|
|
event = _get_event(event_id)
|
|
assert event.delivery_status == "pending"
|
|
assert event.attempts == 1
|
|
assert event.last_error_code == "mismatchedEventId"
|
|
assert event.external_run_id is None
|
|
|
|
|
|
def test_late_delivery_outcome_cannot_overwrite_a_newer_lease(monkeypatch):
|
|
event_id = _make_pending_event("MO-LEASE-RACE")
|
|
dispatcher._claim_due_events()
|
|
|
|
def fake_post(url, json, headers, timeout):
|
|
# Simulate lease expiry/reclaim while the original worker is still in network
|
|
# I/O. The later claimant owns a different token and its state must win.
|
|
with SessionLocal() as db:
|
|
event = db.scalar(
|
|
select(OutboxEvent).where(OutboxEvent.event_id == event_id).with_for_update()
|
|
)
|
|
event.payload_json = {
|
|
**event.payload_json,
|
|
"_delivery_claim_token": "newer-worker-token",
|
|
}
|
|
event.next_attempt_at = datetime.now(UTC) + timedelta(minutes=1)
|
|
db.commit()
|
|
return SimpleNamespace(
|
|
raise_for_status=lambda: None,
|
|
json=lambda: {
|
|
"ok": True,
|
|
"event_id": str(event_id),
|
|
"result": {"execution_id": "stale-worker-execution"},
|
|
},
|
|
status_code=200,
|
|
)
|
|
|
|
monkeypatch.setattr(dispatcher.httpx, "post", fake_post)
|
|
dispatcher._deliver_one(event_id)
|
|
|
|
event = _get_event(event_id)
|
|
assert event.delivery_status == "delivering"
|
|
assert event.attempts == 0
|
|
assert event.external_run_id is None
|
|
assert event.payload_json["_delivery_claim_token"] == "newer-worker-token"
|
|
|
|
|
|
def test_deliver_one_exhausts_attempts_to_failed(monkeypatch):
|
|
event_id = _make_pending_event("MO-004")
|
|
settings = get_settings()
|
|
|
|
def fake_post(url, json, headers, timeout):
|
|
raise dispatcher.httpx.ConnectError("still down")
|
|
|
|
monkeypatch.setattr(dispatcher.httpx, "post", fake_post)
|
|
|
|
for _ in range(settings.n8n_max_attempts):
|
|
db = SessionLocal()
|
|
try:
|
|
event = db.scalar(select(OutboxEvent).where(OutboxEvent.event_id == event_id))
|
|
event.next_attempt_at = None
|
|
db.commit()
|
|
finally:
|
|
db.close()
|
|
claimed = dispatcher._claim_due_events(batch_size=1000)
|
|
assert event_id in claimed
|
|
dispatcher._deliver_one(event_id)
|
|
|
|
event = _get_event(event_id)
|
|
assert event.delivery_status == "failed"
|
|
assert event.attempts == settings.n8n_max_attempts
|
|
|
|
|
|
def test_deliver_one_handles_malformed_payload_without_getting_stuck(monkeypatch):
|
|
# Regression test: seeded/legacy outbox rows may lack the full event envelope. Delivery
|
|
# must resolve the claimed "delivering" row to pending/failed, never leave it stuck.
|
|
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={"aggregate_ref": booking.public_ref}, # missing correlation_id/etc.
|
|
occurred_at=datetime.now(UTC),
|
|
delivery_status="pending",
|
|
attempts=0,
|
|
)
|
|
db.add(event)
|
|
db.commit()
|
|
event_id = event.event_id
|
|
finally:
|
|
db.close()
|
|
|
|
def fake_post(url, json, headers, timeout):
|
|
raise AssertionError("must not attempt delivery with a malformed payload")
|
|
|
|
monkeypatch.setattr(dispatcher.httpx, "post", fake_post)
|
|
|
|
dispatcher._claim_due_events()
|
|
dispatcher._deliver_one(event_id)
|
|
|
|
event = _get_event(event_id)
|
|
assert event.delivery_status in ("pending", "failed")
|
|
assert event.attempts == 1
|
|
assert "Malformed outbox payload" in event.last_error
|
|
assert event.last_error_code == "malformedPayload"
|
|
|
|
|
|
def test_claim_sets_a_lease_deadline():
|
|
event_id = _make_pending_event("MO-006")
|
|
settings = get_settings()
|
|
before = datetime.now(UTC)
|
|
dispatcher._claim_due_events()
|
|
|
|
event = _get_event(event_id)
|
|
assert event.delivery_status == "delivering"
|
|
assert event.next_attempt_at is not None
|
|
lease = settings.n8n_delivery_lease_seconds
|
|
assert event.next_attempt_at > before + timedelta(seconds=lease - 5)
|
|
|
|
|
|
def test_reclaim_ignores_an_active_unexpired_lease():
|
|
# A worker that is still within its lease window must not be disturbed -- this is
|
|
# what prevents double delivery of an event another (still-alive) worker is handling.
|
|
event_id = _make_pending_event("MO-007")
|
|
dispatcher._claim_due_events()
|
|
|
|
reclaimed = dispatcher._reclaim_stale_deliveries()
|
|
assert reclaimed == 0
|
|
assert _get_event(event_id).delivery_status == "delivering"
|
|
|
|
|
|
def test_reclaim_recovers_an_expired_lease_and_preserves_attempts(monkeypatch):
|
|
# Simulates a process crash: the row was claimed (delivering) but no outcome was ever
|
|
# recorded, and its lease has since expired.
|
|
event_id = _make_pending_event("MO-008")
|
|
dispatcher._claim_due_events()
|
|
|
|
db = SessionLocal()
|
|
try:
|
|
event = db.scalar(select(OutboxEvent).where(OutboxEvent.event_id == event_id))
|
|
event.attempts = 2
|
|
event.next_attempt_at = datetime.now(UTC) - timedelta(seconds=1)
|
|
db.commit()
|
|
finally:
|
|
db.close()
|
|
|
|
reclaimed = dispatcher._reclaim_stale_deliveries()
|
|
assert reclaimed == 1
|
|
|
|
event = _get_event(event_id)
|
|
assert event.delivery_status == "pending"
|
|
assert event.next_attempt_at is None
|
|
assert event.attempts == 2
|
|
assert "stale" in event.last_error.lower()
|
|
assert event.last_error_code == "staleLeaseRecovered"
|
|
|
|
# The reclaimed event is now a normal pending event, immediately claimable again.
|
|
claimed = dispatcher._claim_due_events()
|
|
assert event_id in claimed
|
|
|
|
|
|
def test_run_dispatch_cycle_recovers_a_stale_lease_before_claiming(monkeypatch):
|
|
event_id = _make_pending_event("MO-009")
|
|
dispatcher._claim_due_events()
|
|
db = SessionLocal()
|
|
try:
|
|
event = db.scalar(select(OutboxEvent).where(OutboxEvent.event_id == event_id))
|
|
event.next_attempt_at = datetime.now(UTC) - timedelta(seconds=1)
|
|
db.commit()
|
|
finally:
|
|
db.close()
|
|
|
|
def fake_post(url, json, headers, timeout):
|
|
return SimpleNamespace(
|
|
raise_for_status=lambda: None,
|
|
json=lambda: {
|
|
"ok": True,
|
|
"event_id": str(event_id),
|
|
"result": {"execution_id": "n8n-recovered-execution"},
|
|
},
|
|
)
|
|
|
|
monkeypatch.setattr(dispatcher.httpx, "post", fake_post)
|
|
processed = dispatcher.run_dispatch_cycle()
|
|
|
|
assert processed >= 1
|
|
assert _get_event(event_id).delivery_status == "succeeded"
|
|
|
|
|
|
def test_run_dispatch_cycle_end_to_end(monkeypatch):
|
|
event_id = _make_pending_event("MO-005")
|
|
|
|
def fake_post(url, json, headers, timeout):
|
|
return SimpleNamespace(
|
|
raise_for_status=lambda: None,
|
|
json=lambda: {
|
|
"ok": True,
|
|
"event_id": str(event_id),
|
|
"result": {"execution_id": "n8n-cycle-execution"},
|
|
},
|
|
)
|
|
|
|
monkeypatch.setattr(dispatcher.httpx, "post", fake_post)
|
|
processed = dispatcher.run_dispatch_cycle()
|
|
|
|
assert processed >= 1
|
|
assert _get_event(event_id).delivery_status == "succeeded"
|