prowler/mcp_server/tests/lib/test_errors.py

"""Tests for the shared failure classifier.

Two properties are pinned here: a failed tool call answers with ``isError: true``
rather than a result object the client reads as a success, and the only text that
reaches a model is text this server produced.
"""

import json

import httpx
import pytest
from fastmcp import Client
from pydantic import BaseModel, ValidationError

from prowler_mcp_server.lib.errors import (
    CredentialError,
    InvalidArgument,
    UpstreamInvalidResponse,
    _describe_failure,
    parse_json_response,
)
from prowler_mcp_server.prowler_app.utils.api_client import (
    ProwlerAPIError,
    ProwlerAPIInvalidResponse,
    ProwlerAPIUnreachable,
)
from tests.helpers.jsonapi import jsonapi_error

LATEST = "/api/v1/findings/latest"


# --------------------------------------------------------------- json bodies


def _answer(
    body: str, *, url: str = "https://hub.prowler.com/api/check"
) -> httpx.Response:
    """An answer as a client would hand it back, request attached."""
    return httpx.Response(200, text=body, request=httpx.Request("GET", url))


def test_a_json_body_is_returned_as_it_is():
    """The helper only classifies the failure; the success path is untouched."""
    assert parse_json_response(_answer('{"id": "s3_bucket_public_access"}')) == {
        "id": "s3_bucket_public_access"
    }


def test_a_body_that_is_not_json_names_the_host_that_answered():
    """Which upstream is misbehaving is the one useful fact here, and the shared
    helper is reached from every sub-server that reads an upstream directly."""
    with pytest.raises(UpstreamInvalidResponse) as raised:
        parse_json_response(_answer("<html><body>502 Bad Gateway</body></html>"))

    assert raised.value.host == "hub.prowler.com"
    assert "Bad Gateway" not in str(raised.value)


def test_a_body_that_is_not_json_is_not_a_valueerror():
    """`JSONDecodeError` is a ValueError, and callers tell an upstream fault from
    a bad argument by type alone."""
    with pytest.raises(UpstreamInvalidResponse) as raised:
        parse_json_response(_answer("not json"))

    assert not isinstance(raised.value, ValueError)


# ------------------------------------------------------------ classification


@pytest.mark.parametrize(
    ("status", "expected"),
    [
        (401, "missing, malformed or expired"),
        (403, "prowler_get_current_user"),
        (429, "rate limiting"),
        (503, "failed on Prowler's side"),
    ],
    ids=["unauthorized", "forbidden", "rate-limited", "unavailable"],
)
def test_the_failures_every_authenticated_tool_shares_get_one_message(status, expected):
    """These four mean the same thing whichever tool hit them."""
    message = _describe_failure(ProwlerAPIError("failed", status))

    assert expected in message


def test_a_rejection_relays_the_apis_own_reason():
    """`errors[].detail` is written by Prowler for a caller, so it is ours to relay."""
    message = _describe_failure(
        ProwlerAPIError("failed", 400, detail="scan_id is not a valid UUID.")
    )

    assert "scan_id is not a valid UUID." in message


def test_a_rejection_with_no_trustworthy_reason_says_so_instead_of_guessing():
    """A body that was not JSON:API leaves `detail` unset, and it stays unrelayed."""
    message = _describe_failure(ProwlerAPIError("failed", 400))

    assert "gave no reason" in message


def test_a_request_that_got_no_answer_says_the_outcome_is_unknown():
    """An unanswered write may well have landed, so repeating it can duplicate it."""
    message = _describe_failure(ProwlerAPIUnreachable("POST /providers got no answer"))

    assert "could not be reached" in message
    assert "unknown" in message


def test_an_unreadable_api_answer_is_not_blamed_on_the_arguments():
    """The same `JSONDecodeError` means opposite things on the two sides."""
    message = _describe_failure(
        ProwlerAPIInvalidResponse(
            "GET /findings answered 200 with a body that is not JSON"
        )
    )

    assert "could not read" in message
    assert "argument" not in message


def test_an_unreadable_api_answer_is_never_called_safe_to_repeat():
    """`post`, `patch` and `delete` reach this too, and a write may have landed."""
    message = _describe_failure(
        ProwlerAPIInvalidResponse(
            "POST /providers answered 201 with a body that is not JSON"
        )
    )

    assert "unknown" in message
    assert "check the current state" in message


def test_an_unreadable_upstream_answer_is_not_blamed_on_the_arguments():
    """A `JSONDecodeError` from an upstream and one from an argument are the same
    exception and opposite instructions."""
    message = _describe_failure(
        UpstreamInvalidResponse("200 body is not JSON", host="hub.prowler.com")
    )

    assert "hub.prowler.com" in message
    assert "could not read as JSON" in message
    assert "changing them will not help" in message


def test_an_unreadable_upstream_answer_never_quotes_the_body():
    """The body is someone else's text, so only the host and the status leave here."""
    message = _describe_failure(
        UpstreamInvalidResponse(
            "502 body is not JSON", host="raw.githubusercontent.com"
        )
    )

    assert "body is not JSON" not in message


def test_an_argument_this_server_rejected_is_repeated_verbatim():
    """`InvalidArgument` exists to mark a message as one we wrote."""
    message = _describe_failure(
        InvalidArgument("page_size must be between 1 and 1000.")
    )

    assert message == "page_size must be between 1 and 1000."


def test_a_credential_caught_here_is_answered_like_the_401_it_would_have_got():
    """It is not an argument problem, and saying so stops a pointless retry."""
    message = _describe_failure(CredentialError("the token has expired"))

    assert "the token has expired" in message
    assert "changing the arguments will not help" in message


def test_a_transport_this_server_cannot_serve_is_left_masked():
    """No call caused a bad PROWLER_MCP_TRANSPORT_MODE and no call can fix it."""
    assert _describe_failure(RuntimeError("Invalid mode: websocket")) is None


def test_a_pydantic_rejection_names_the_field_without_echoing_the_value():
    """Pydantic quotes `input_value` back, and these tools take credentials."""

    class Credentials(BaseModel):
        api_token: int

    with pytest.raises(ValidationError) as raised:
        Credentials(api_token="hunter2-the-real-secret")

    message = _describe_failure(raised.value)

    assert "api_token" in message
    assert "hunter2-the-real-secret" not in message


def test_unparseable_json_is_reported_without_quoting_the_input():
    """`JSONDecodeError` is a ValueError whose message quotes what it was given."""
    with pytest.raises(json.JSONDecodeError) as raised:
        json.loads('{"api_token": "hunter2-the-real-secret"')

    message = _describe_failure(raised.value)

    assert "could not be parsed" in message
    assert "hunter2" not in message


def test_an_unrecognised_failure_is_left_masked():
    """Saying nothing is the safe default; the alternative is relaying anything."""
    assert (
        _describe_failure(RuntimeError("connection pool exhausted at 10.0.0.4:5432"))
        is None
    )


# --------------------------------------------------------- through the server


async def test_a_failing_tool_answers_with_is_error_not_a_result(
    mcp_root_server, mock_api_client, mock_router
):
    """An error dict would arrive as `isError: false` and read as a success."""
    mock_router.add("GET", LATEST, status=403, json=jsonapi_error(403, "Denied."))

    async with Client(mcp_root_server) as client:
        result = await client.call_tool_mcp("prowler_search_security_findings", {})

    assert result.isError is True
    assert result.structuredContent is None
    assert "prowler_get_current_user" in result.content[0].text


async def test_an_upstream_body_never_reaches_the_agent(
    mcp_root_server, mock_api_client, mock_router
):
    """A body this server did not write is logged and replaced, never relayed."""
    mock_router.add(
        "GET",
        LATEST,
        status=500,
        text="Traceback: psycopg2 could not connect to internal-db:5432",
    )

    async with Client(mcp_root_server) as client:
        result = await client.call_tool_mcp("prowler_search_security_findings", {})

    assert result.isError is True
    assert "internal-db" not in result.content[0].text
    assert "failed on Prowler's side" in result.content[0].text


async def test_a_bad_argument_is_rejected_before_any_request_goes_out(
    mcp_root_server, mock_api_client, mock_router
):
    """Local validation saves a round trip, and its message is safe to repeat."""
    async with Client(mcp_root_server) as client:
        result = await client.call_tool_mcp(
            "prowler_search_security_findings", {"page_size": 5000}
        )

    assert result.isError is True
    assert "Must be between 1 and 1000" in result.content[0].text
    assert mock_router.requests == []


async def test_an_unreadable_api_answer_does_not_reach_the_agent_as_a_bad_argument(
    mcp_root_server, mock_api_client, mock_router
):
    """Told apart by type, so the agent is not sent to fix an argument that is fine."""
    mock_router.add("GET", LATEST, text="<html>gateway timeout</html>")

    async with Client(mcp_root_server) as client:
        result = await client.call_tool_mcp("prowler_search_security_findings", {})

    assert result.isError is True
    assert "gateway timeout" not in result.content[0].text
    assert "argument" not in result.content[0].text


async def test_a_tool_specific_message_survives_masking(
    mcp_root_server, mock_api_client, mock_router
):
    """A `ToolError` raised without a `from` clause is the final word."""
    mock_router.add("GET", "/api/v1/integrations/i1", json={"data": None})

    async with Client(mcp_root_server) as client:
        result = await client.call_tool_mcp(
            "prowler_get_integration", {"integration_id": "i1"}
        )

    assert result.isError is True
    assert "prowler_list_integrations" in result.content[0].text


async def test_a_hub_tool_failure_says_which_host_refused_it(
    mcp_root_server, hub_router
):
    """Hub failures arrive as raw httpx errors: host and status relayed, body not."""
    hub_router.add(
        "GET", "/api/check", status=503, text="<html>upstream nginx 10.1.2.3</html>"
    )

    async with Client(mcp_root_server) as client:
        result = await client.call_tool_mcp("prowler_hub_list_checks", {})

    assert result.isError is True
    assert "hub.prowler.com" in result.content[0].text
    assert "10.1.2.3" not in result.content[0].text