MobilityOps/backend/tests/test_integration_status.py

from sqlalchemy import select

from app.core.db import SessionLocal
from app.models.outbox import OutboxEvent
from app.seed_loader import reset_and_seed


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"


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


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