MobilityOps/backend/tests/test_mcp_integrations.py

from app.core.config import get_settings


def _headers(token: str | None = None, client_id: str = "test-mcp-client"):
    settings = get_settings()
    return {
        "X-Service-Token": token if token is not None else settings.mcp_hub_service_token,
        "X-Client-Id": f"itworx-mcp-hub:mobilityops:{client_id}",
    }


def test_operations_summary_requires_service_token(client):
    response = client.get(
        "/api/v1/integrations/mcp/operations-summary", headers=_headers(token="wrong")
    )
    assert response.status_code == 401


def test_operations_summary_rejects_spoofed_client_identity(client):
    settings = get_settings()
    response = client.get(
        "/api/v1/integrations/mcp/operations-summary",
        headers={"X-Service-Token": settings.mcp_hub_service_token, "X-Client-Id": "spoofed"},
    )
    assert response.status_code == 403


def test_operations_summary_rejects_cross_tenant_request(client):
    response = client.get(
        "/api/v1/integrations/mcp/operations-summary",
        headers={**_headers(), "X-Tenant-Id": "another-tenant"},
    )
    assert response.status_code == 403


def test_operations_summary_returns_metrics(client):
    response = client.get("/api/v1/integrations/mcp/operations-summary", headers=_headers())
    assert response.status_code == 200
    body = response.json()
    assert "metrics" in body
    assert body["metrics"]["available"] >= 0
    assert response.headers["x-correlation-id"]
    assert response.headers["x-tenant-id"] == get_settings().ragcore_tenant


def test_attention_vehicles_filters_by_severity(client):
    response = client.get(
        "/api/v1/integrations/mcp/attention-vehicles",
        params={"minimum_severity": "high", "limit": 50},
        headers=_headers(),
    )
    assert response.status_code == 200
    rows = response.json()
    assert all(r["severity"] == "high" for r in rows)


def test_attention_vehicles_respects_limit(client):
    response = client.get(
        "/api/v1/integrations/mcp/attention-vehicles",
        params={"minimum_severity": "low", "limit": 2},
        headers=_headers(),
    )
    assert response.status_code == 200
    assert len(response.json()) <= 2


def test_vehicle_details_known_ref(client):
    response = client.get(
        "/api/v1/integrations/mcp/vehicles/MO-016", headers=_headers()
    )
    assert response.status_code == 200
    body = response.json()
    assert body["public_ref"] == "MO-016"
    assert "registration_number" not in body  # narrow read-only view, not the full record


def test_vehicle_details_unknown_ref_is_404(client):
    response = client.get(
        "/api/v1/integrations/mcp/vehicles/MO-999", headers=_headers()
    )
    assert response.status_code == 404


def test_search_knowledge_grounded_and_respects_max_sources(client):
    response = client.post(
        "/api/v1/integrations/mcp/search-knowledge",
        json={"question": "What must I do when a vehicle returns with damage?", "max_sources": 1},
        headers=_headers(),
    )
    assert response.status_code == 200
    body = response.json()
    assert body["evidence_state"] == "grounded"
    assert len(body["sources"]) == 1
    assert int(response.headers["x-sources-available"]) >= 1
    assert response.headers["x-sources-returned"] == "1"


def test_mcp_tool_requests_are_audited(client, ops_client):
    client.get("/api/v1/integrations/mcp/operations-summary", headers=_headers(client_id="probe-1"))
    events = ops_client.get("/api/v1/audit", params={"action": "mcp_tool_request"}).json()
    assert len(events) >= 1
    assert events[0]["actor_type"] == "service"


def test_search_knowledge_respects_requested_locale(client):
    response = client.post(
        "/api/v1/integrations/mcp/search-knowledge",
        json={
            "question": "Wat moet ik doen wanneer een voertuig beschadigd terugkomt?",
            "max_sources": 1,
            "locale": "nl-BE",
        },
        headers=_headers(),
    )
    assert response.status_code == 200
    body = response.json()
    assert body["evidence_state"] == "grounded"


def test_search_knowledge_preserves_inbound_correlation_id(client, ops_client):
    inbound = "11111111-1111-1111-1111-111111111111"
    response = client.post(
        "/api/v1/integrations/mcp/search-knowledge",
        json={"question": "What must I do when a vehicle returns with damage?", "max_sources": 1},
        headers={**_headers(client_id="correlation-probe"), "X-Correlation-Id": inbound},
    )
    assert response.status_code == 200
    assert response.json()["correlation_id"] == inbound

    events = ops_client.get("/api/v1/audit", params={"action": "mcp_tool_request"}).json()
    matching = [e for e in events if e["correlation_id"] == inbound]
    assert len(matching) == 1


def test_operations_summary_mints_correlation_id_when_none_supplied(client, ops_client):
    response = client.get(
        "/api/v1/integrations/mcp/operations-summary",
        headers=_headers(client_id="no-correlation-probe"),
    )
    assert response.status_code == 200
    events = ops_client.get("/api/v1/audit", params={"action": "mcp_tool_request"}).json()
    matching = [e for e in events if e["actor_label"].endswith(":no-correlation-probe")]
    assert len(matching) >= 1
    assert matching[0]["correlation_id"]  # a fresh UUID was minted, not left empty


def test_no_write_endpoints_exist_under_mcp_namespace(client):
    for method, path in [
        ("post", "/api/v1/integrations/mcp/vehicles/MO-016"),
        ("put", "/api/v1/integrations/mcp/vehicles/MO-016"),
        ("delete", "/api/v1/integrations/mcp/vehicles/MO-016"),
    ]:
        response = getattr(client, method)(path, headers=_headers())
        assert response.status_code in (404, 405)