293 lines
11 KiB
Python
293 lines
11 KiB
Python
import uuid
|
|
from datetime import UTC, datetime
|
|
|
|
import pytest
|
|
from sqlalchemy import delete, select
|
|
|
|
from app.core.db import SessionLocal
|
|
from app.models.audit import AuditEvent
|
|
from app.models.outbox import OutboxEvent
|
|
from app.seed_loader import reset_and_seed
|
|
from app.services import integration_status
|
|
|
|
|
|
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"
|
|
|
|
|
|
@pytest.mark.parametrize("delivery_status", ["pending", "delivering"])
|
|
def test_queued_work_without_any_success_is_not_reported_operational(delivery_status):
|
|
_reseed()
|
|
db = SessionLocal()
|
|
try:
|
|
for event in db.scalars(select(OutboxEvent)).all():
|
|
event.delivery_status = delivery_status
|
|
event.last_error = None
|
|
event.last_error_code = None
|
|
db.execute(
|
|
delete(AuditEvent).where(
|
|
AuditEvent.action.in_(
|
|
{
|
|
"data_quality_scan_run",
|
|
"n8n_procedures_synced",
|
|
"n8n_workflow_failure_registered",
|
|
"n8n_workflow_heartbeat",
|
|
}
|
|
)
|
|
)
|
|
)
|
|
db.flush()
|
|
|
|
status = integration_status.derive_n8n_status(db)
|
|
|
|
assert status.succeeded == 0
|
|
assert getattr(status, delivery_status) > 0
|
|
assert status.state == "no_evidence"
|
|
finally:
|
|
db.rollback()
|
|
db.close()
|
|
|
|
|
|
def test_historical_success_is_not_green_when_webhook_is_unconfigured(monkeypatch):
|
|
_reseed()
|
|
with SessionLocal() as db:
|
|
monkeypatch.setattr(integration_status.settings, "n8n_dispatch_enabled", True)
|
|
monkeypatch.setattr(integration_status.settings, "n8n_webhook_url", "")
|
|
status = integration_status.derive_n8n_status(db)
|
|
assert status.configured is False
|
|
assert status.succeeded > 0
|
|
assert status.state == "unavailable"
|
|
|
|
|
|
def test_null_coded_real_failure_sets_latest_failure_timestamp():
|
|
_reseed()
|
|
with SessionLocal() as db:
|
|
event = db.scalar(
|
|
select(OutboxEvent).where(OutboxEvent.delivery_status == "succeeded").limit(1)
|
|
)
|
|
event.delivery_status = "failed"
|
|
event.last_error_code = None
|
|
db.flush()
|
|
status = integration_status.derive_n8n_status(db)
|
|
assert status.unexpected_failed == 1
|
|
assert status.latest_failure_at is not None
|
|
db.rollback()
|
|
|
|
|
|
def test_unreachable_hub_invalidates_historical_operational_claim(monkeypatch):
|
|
db = SessionLocal()
|
|
try:
|
|
db.add(
|
|
AuditEvent(
|
|
actor_type="service",
|
|
actor_id=None,
|
|
actor_label="itworx-mcp-hub",
|
|
action="mcp_tool_request",
|
|
entity_type="integration",
|
|
entity_id=None,
|
|
correlation_id=uuid.uuid4(),
|
|
before_json=None,
|
|
after_json=None,
|
|
metadata_json={"tool": "operations_summary"},
|
|
occurred_at=datetime.now(UTC),
|
|
)
|
|
)
|
|
db.flush()
|
|
monkeypatch.setattr(integration_status.settings, "mcp_hub_registration_enabled", True)
|
|
monkeypatch.setattr(integration_status, "_check_hub_reachable", lambda: False)
|
|
|
|
unreachable = integration_status.derive_mcp_hub_status(db)
|
|
|
|
assert unreachable.total_calls > 0
|
|
assert unreachable.hub_reachable is False
|
|
assert unreachable.state == "no_evidence"
|
|
|
|
monkeypatch.setattr(integration_status, "_check_hub_reachable", lambda: True)
|
|
reachable = integration_status.derive_mcp_hub_status(db)
|
|
assert reachable.state == "operational"
|
|
finally:
|
|
db.rollback()
|
|
db.close()
|
|
|
|
|
|
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 (all 6 nodes saved live). This fresh test run
|
|
# has reported no real sync result yet, so -- like the other three workflows before
|
|
# their own first real signal -- there is no run evidence yet either.
|
|
assert ragcore_sync["built"] is True
|
|
assert ragcore_sync["last_seen_at"] is None
|
|
assert ragcore_sync["state"] == "no_evidence"
|
|
|
|
|
|
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
|