AegisGitea-MCP/tests/test_executor.py

"""Tests for the Gitea request executor (HTTP behaviour, envelope, redaction)."""

from __future__ import annotations

import base64
import json
from collections.abc import AsyncIterator
from typing import Any

import httpx
import pytest
from pytest_httpx import HTTPXMock

from aegis_gitea_mcp import __version__
from aegis_gitea_mcp.catalog.model import Parameter
from aegis_gitea_mcp.gitea import (
    CREDENTIAL_ISSUING_OPERATIONS,
    RAW_TOOL_NAME,
    UNTRUSTED_NOTICE,
    GiteaExecutor,
    ToolOutcome,
)
from aegis_gitea_mcp.gitea import executor as executor_module
from aegis_gitea_mcp.identity import Identity
from tests.test_executor_helpers import (
    API,
    BEARER_TOKEN,
    CREATE_ISSUE,
    CREATE_TOKEN,
    CREATE_USER,
    DELETE_BRANCH,
    GET_BRANCH,
    GET_RAW_FILE,
    GITEA_TOKEN,
    LIST_BRANCHES,
    LIST_PULLS,
    OAUTH_CLIENT_SECRET,
    PULL_DIFF,
    PUT_SECRET,
    REGISTRATION_TOKEN,
    RENDER_RAW,
    UPLOAD_ATTACHMENT,
    USER_TOKEN,
    admin_identity,
    make_op,
    make_settings,
    oauth_identity,
    param,
)

BASE_ARGS = {"owner": "octo", "repo": "demo"}
GH_TOKEN = "ghp_" + "a1B2c3D4e5F6g7H8i9J0k1L2m3N4o5"


@pytest.fixture
async def executor() -> AsyncIterator[GiteaExecutor]:
    instance = GiteaExecutor(make_settings())
    yield instance
    await instance.aclose()


def only_request(httpx_mock: HTTPXMock) -> httpx.Request:
    requests = httpx_mock.get_requests()
    assert len(requests) == 1
    return requests[0]


def error_of(outcome: ToolOutcome) -> dict[str, Any]:
    assert outcome.is_error
    error: dict[str, Any] = outcome.payload["error"]
    return error


# URL building, headers and query


async def test_get_builds_url_headers_and_success_envelope(
    executor: GiteaExecutor, httpx_mock: HTTPXMock
) -> None:
    httpx_mock.add_response(json=[{"name": "main"}])

    outcome = await executor.execute(
        LIST_BRANCHES, {**BASE_ARGS, "page": 2, "limit": 10}, admin_identity()
    )

    request = only_request(httpx_mock)
    assert request.method == "GET"
    assert str(request.url) == f"{API}/repos/octo/demo/branches?page=2&limit=10"
    assert request.headers["Authorization"] == f"token {GITEA_TOKEN}"
    assert request.headers["Accept"] == "application/json"
    assert request.headers["User-Agent"] == f"aegis-gitea-mcp/{__version__}"
    assert "Sudo" not in request.headers
    assert request.content == b""
    assert outcome.is_error is False
    assert outcome.status == 200
    assert list(outcome.payload) == [
        "identity",
        "request",
        "status",
        "pagination",
        "untrusted_content",
        "data",
    ]
    assert outcome.payload["identity"] == {"login": "admin", "mode": "admin-token", "sudo": None}
    assert outcome.payload["request"] == {
        "method": "GET",
        "path": "/repos/octo/demo/branches",
        "operation": LIST_BRANCHES.tool_name,
    }
    assert outcome.payload["status"] == 200
    assert outcome.payload["untrusted_content"] == UNTRUSTED_NOTICE
    assert outcome.payload["data"] == [{"name": "main"}]
    assert json.loads(outcome.to_text()) == outcome.payload


async def test_oauth_identity_uses_bearer_and_sudo_header(
    executor: GiteaExecutor, httpx_mock: HTTPXMock
) -> None:
    httpx_mock.add_response(json=[])

    outcome = await executor.execute(
        LIST_BRANCHES, {**BASE_ARGS, "sudo": "bob.smith"}, oauth_identity()
    )

    request = only_request(httpx_mock)
    assert request.headers["Authorization"] == f"Bearer {USER_TOKEN}"
    assert request.headers["Sudo"] == "bob.smith"
    assert outcome.payload["identity"] == {"login": "alice", "mode": "oauth", "sudo": "bob.smith"}


async def test_identity_without_token_sends_no_authorization(
    executor: GiteaExecutor, httpx_mock: HTTPXMock
) -> None:
    httpx_mock.add_response(json=[])
    identity = Identity(login=None, mode="oauth", token="")

    await executor.execute(LIST_BRANCHES, BASE_ARGS, identity)

    assert "Authorization" not in only_request(httpx_mock).headers


@pytest.mark.parametrize("sudo", ["", "bob smith", "bob\r\nX-Evil: 1", 5, "a" * 256])
async def test_invalid_sudo_is_rejected_without_request(
    executor: GiteaExecutor, httpx_mock: HTTPXMock, sudo: Any
) -> None:
    outcome = await executor.execute(LIST_BRANCHES, {**BASE_ARGS, "sudo": sudo}, admin_identity())

    assert error_of(outcome)["kind"] == "invalid_argument"
    assert outcome.status is None
    assert outcome.payload["identity"]["sudo"] is None
    assert httpx_mock.get_requests() == []


async def test_accept_header_lists_all_produced_types(
    executor: GiteaExecutor, httpx_mock: HTTPXMock
) -> None:
    operation = make_op(
        "repoCompareDiff",
        "GET",
        "/repos/{owner}/{repo}/labels",
        [param("owner", "path"), param("repo", "path")],
        produces=("application/json", "text/plain"),
    )
    no_produces = make_op("x", "GET", "/ping", [], produces=())
    httpx_mock.add_response(json={})
    httpx_mock.add_response(json={})

    await executor.execute(operation, BASE_ARGS, admin_identity())
    await executor.execute(no_produces, {}, admin_identity())

    first, second = httpx_mock.get_requests()
    assert first.headers["Accept"] == "application/json, text/plain"
    assert second.headers["Accept"] == "application/json"


async def test_slashes_kept_in_filepath_and_unicode_encoded(
    executor: GiteaExecutor, httpx_mock: HTTPXMock
) -> None:
    httpx_mock.add_response(content=b"data", headers={"content-type": "application/octet-stream"})
    httpx_mock.add_response(json={"name": "x"})

    await executor.execute(
        GET_RAW_FILE,
        {**BASE_ARGS, "filepath": "src/app/mein file.py", "ref": "feature/ünï"},
        admin_identity(),
    )
    await executor.execute(GET_BRANCH, {**BASE_ARGS, "branch": "feature/für/✓"}, admin_identity())

    raw, branch = httpx_mock.get_requests()
    assert raw.url.raw_path == (
        b"/api/v1/repos/octo/demo/raw/src/app/mein%20file.py?ref=feature%2F%C3%BCn%C3%AF"
    )
    assert branch.url.raw_path == (b"/api/v1/repos/octo/demo/branches/feature/f%C3%BCr/%E2%9C%93")


async def test_non_slash_parameters_encode_slashes(
    executor: GiteaExecutor, httpx_mock: HTTPXMock
) -> None:
    httpx_mock.add_response(json=[])

    outcome = await executor.execute(
        LIST_BRANCHES, {"owner": "a/b", "repo": "r e:p"}, admin_identity()
    )

    assert only_request(httpx_mock).url.raw_path == b"/api/v1/repos/a%2Fb/r%20e%3Ap/branches"
    assert outcome.payload["request"]["path"] == "/repos/a/b/r e:p/branches"


async def test_integer_and_dotted_placeholders(
    executor: GiteaExecutor, httpx_mock: HTTPXMock
) -> None:
    httpx_mock.add_response(text="diff --git", headers={"content-type": "text/plain"})

    outcome = await executor.execute(
        PULL_DIFF, {**BASE_ARGS, "index": 7, "diffType": "diff"}, admin_identity()
    )

    request = only_request(httpx_mock)
    assert request.url.path == "/api/v1/repos/octo/demo/pulls/7.diff"
    assert request.headers["Accept"] == "text/plain"
    assert outcome.payload["data"] == "diff --git"


@pytest.mark.parametrize(
    "value",
    [
        "",
        ".",
        "..",
        "a/../b",
        "../etc",
        "a//b",
        "a\\b",
        "a?b=1",
        "a#frag",
        "a%2Fb",
        "100%",
        "a\x00b",
        "a\nb",
        "a\u2028b",
        "a\ud800b",
    ],
)
@pytest.mark.parametrize("name", ["owner", "filepath"])
async def test_unsafe_path_values_are_rejected_without_request(
    executor: GiteaExecutor, httpx_mock: HTTPXMock, name: str, value: str
) -> None:
    arguments = {**BASE_ARGS, "filepath": "ok/file.txt"}
    arguments[name] = value

    outcome = await executor.execute(GET_RAW_FILE, arguments, admin_identity())

    error = error_of(outcome)
    assert error["kind"] == "invalid_argument"
    assert error["status"] is None
    assert outcome.status is None
    assert outcome.payload["request"]["path"] == GET_RAW_FILE.path
    assert httpx_mock.get_requests() == []


@pytest.mark.parametrize(
    ("arguments", "needle"),
    [
        ({"repo": "demo"}, "missing path parameter 'owner'"),
        ({"owner": None, "repo": "demo"}, "missing path parameter 'owner'"),
        ({"owner": True, "repo": "demo"}, "must be a string or integer"),
        ({"owner": 1.5, "repo": "demo"}, "must be a string or integer"),
        ({"owner": ["a"], "repo": "demo"}, "must be a string or integer"),
    ],
)
async def test_missing_or_mistyped_path_parameter(
    executor: GiteaExecutor, httpx_mock: HTTPXMock, arguments: dict[str, Any], needle: str
) -> None:
    outcome = await executor.execute(LIST_BRANCHES, arguments, admin_identity())

    assert needle in error_of(outcome)["message"]
    assert httpx_mock.get_requests() == []


async def test_placeholder_without_parameter_fails_closed(
    executor: GiteaExecutor, httpx_mock: HTTPXMock
) -> None:
    operation = make_op("broken", "GET", "/repos/{owner}/{nothere}", [param("owner", "path")])

    outcome = await executor.execute(operation, {"owner": "o"}, admin_identity())

    assert "has no matching parameter" in error_of(outcome)["message"]
    assert httpx_mock.get_requests() == []


async def test_rejected_value_is_never_echoed(
    executor: GiteaExecutor, httpx_mock: HTTPXMock
) -> None:
    outcome = await executor.execute(
        LIST_BRANCHES, {"owner": f"../{USER_TOKEN}", "repo": "demo"}, admin_identity()
    )

    assert USER_TOKEN not in json.dumps(outcome.payload)
    assert httpx_mock.get_requests() == []


async def test_query_arrays_booleans_and_skipped_values(
    executor: GiteaExecutor, httpx_mock: HTTPXMock
) -> None:
    httpx_mock.add_response(json=[])

    await executor.execute(
        LIST_PULLS,
        {
            **BASE_ARGS,
            "labels": [3, 5],
            "state": None,
            "draft": False,
            "csvlist": ["a", "b"],
            "page": 1,
            "limit": 1.0,
            "unknown": "ignored",
        },
        admin_identity(),
    )

    query = only_request(httpx_mock).url.query.decode()
    assert query == "labels=3&labels=5&draft=false&csvlist=a%2Cb&page=1&limit=1.0"


async def test_query_true_empty_array_and_scalar_to_array(
    executor: GiteaExecutor, httpx_mock: HTTPXMock
) -> None:
    httpx_mock.add_response(json=[])

    await executor.execute(
        LIST_PULLS,
        {**BASE_ARGS, "draft": True, "labels": [], "csvlist": "single"},
        admin_identity(),
    )

    assert only_request(httpx_mock).url.query.decode() == "draft=true&csvlist=single"


async def test_query_with_unsupported_type_is_rejected(
    executor: GiteaExecutor, httpx_mock: HTTPXMock
) -> None:
    outcome = await executor.execute(
        LIST_PULLS, {**BASE_ARGS, "state": {"nested": 1}}, admin_identity()
    )

    assert "query parameter 'state'" in error_of(outcome)["message"]
    assert httpx_mock.get_requests() == []


async def test_reserved_query_names_are_rejected(
    executor: GiteaExecutor, httpx_mock: HTTPXMock
) -> None:
    operation = make_op("x", "GET", "/ping", [param("token", "query"), param("Sudo", "query")])

    first = await executor.execute(operation, {"token": "abc"}, admin_identity())
    second = await executor.execute(operation, {"Sudo": "root"}, admin_identity())

    assert "reserved" in error_of(first)["message"]
    assert "reserved" in error_of(second)["message"]
    assert httpx_mock.get_requests() == []


# Bodies


async def test_json_body_is_sent_and_not_returned(
    executor: GiteaExecutor, httpx_mock: HTTPXMock
) -> None:
    httpx_mock.add_response(status_code=201, json={"number": 1})
    body = {"title": "Bug", "body": "details ünï"}

    outcome = await executor.execute(CREATE_ISSUE, {**BASE_ARGS, "body": body}, admin_identity())

    request = only_request(httpx_mock)
    assert request.method == "POST"
    assert json.loads(request.content) == body
    assert request.headers["Content-Type"] == "application/json"
    assert outcome.status == 201
    assert "details" not in json.dumps(outcome.payload)


async def test_missing_body_sends_no_content(
    executor: GiteaExecutor, httpx_mock: HTTPXMock
) -> None:
    httpx_mock.add_response(status_code=201, json={})

    await executor.execute(CREATE_ISSUE, BASE_ARGS, admin_identity())

    assert only_request(httpx_mock).content == b""


async def test_text_plain_body(executor: GiteaExecutor, httpx_mock: HTTPXMock) -> None:
    httpx_mock.add_response(text="<p>x</p>", headers={"content-type": "text/html"})

    outcome = await executor.execute(RENDER_RAW, {"body": "# héllo"}, admin_identity())

    request = only_request(httpx_mock)
    assert request.content == "# héllo".encode()
    assert request.headers["Content-Type"] == "text/plain"
    assert outcome.payload["data"] == "<p>x</p>"


async def test_text_plain_body_must_be_a_string(
    executor: GiteaExecutor, httpx_mock: HTTPXMock
) -> None:
    outcome = await executor.execute(RENDER_RAW, {"body": {"a": 1}}, admin_identity())

    assert "must be a string" in error_of(outcome)["message"]
    assert httpx_mock.get_requests() == []


async def test_multipart_upload_decodes_base64(
    executor: GiteaExecutor, httpx_mock: HTTPXMock
) -> None:
    httpx_mock.add_response(status_code=201, json={"id": 9})
    payload = bytes(range(256))

    outcome = await executor.execute(
        UPLOAD_ATTACHMENT,
        {
            **BASE_ARGS,
            "id": 4,
            "name": "tool.bin",
            "attachment": {
                "filename": "tool.bin",
                "content_base64": base64.b64encode(payload).decode(),
                "content_type": "application/x-tool",
            },
        },
        admin_identity(),
    )

    request = only_request(httpx_mock)
    assert request.url.query == b"name=tool.bin"
    assert request.headers["Content-Type"].startswith("multipart/form-data; boundary=")
    assert b'name="attachment"; filename="tool.bin"' in request.content
    assert b"Content-Type: application/x-tool" in request.content
    assert payload in request.content
    assert outcome.status == 201


async def test_multipart_default_content_type_and_plain_form_field(
    executor: GiteaExecutor, httpx_mock: HTTPXMock
) -> None:
    operation = make_op(
        "x",
        "POST",
        "/upload",
        [param("attachment", "formData", is_file=True), param("note", "formData")],
        consumes=("multipart/form-data",),
    )
    httpx_mock.add_response(status_code=201, json={})

    await executor.execute(
        operation,
        {"attachment": {"filename": "a.txt", "content_base64": "aGk="}, "note": 5},
        admin_identity(),
    )

    content = only_request(httpx_mock).content
    assert b"Content-Type: application/octet-stream" in content
    assert b'name="note"' in content
    assert b"\r\n\r\n5\r\n" in content


@pytest.mark.parametrize(
    ("attachment", "needle"),
    [
        ("not-an-object", "must be an object"),
        ({"filename": "a", "content_base64": "aGk=", "extra": 1}, "unexpected keys"),
        ({"filename": 5, "content_base64": "aGk="}, "needs string filename"),
        ({"filename": "a"}, "needs string filename"),
        ({"filename": "", "content_base64": "aGk="}, "non-empty single-line"),
        ({"filename": "a\nb", "content_base64": "aGk="}, "non-empty single-line"),
        ({"filename": "x" * 256, "content_base64": "aGk="}, "non-empty single-line"),
        ({"filename": 'a"b', "content_base64": "aGk="}, "double quote"),
        ({"filename": "a", "content_base64": "!!!notbase64"}, "not valid base64"),
        ({"filename": "a", "content_base64": "aGk"}, "not valid base64"),
        ({"filename": "a", "content_base64": "aG k="}, "not valid base64"),
        ({"filename": "a", "content_base64": "aGk=", "content_type": 5}, "must be a string"),
        ({"filename": "a", "content_base64": "aGk=", "content_type": "a\r\nb"}, "single-line"),
    ],
)
async def test_invalid_file_arguments(
    executor: GiteaExecutor, httpx_mock: HTTPXMock, attachment: Any, needle: str
) -> None:
    outcome = await executor.execute(
        UPLOAD_ATTACHMENT, {**BASE_ARGS, "id": 1, "attachment": attachment}, admin_identity()
    )

    error = error_of(outcome)
    assert error["kind"] == "invalid_argument"
    assert needle in error["message"]
    assert httpx_mock.get_requests() == []


async def test_upload_without_file_sends_no_body(
    executor: GiteaExecutor, httpx_mock: HTTPXMock
) -> None:
    httpx_mock.add_response(status_code=201, json={})

    await executor.execute(UPLOAD_ATTACHMENT, {**BASE_ARGS, "id": 1}, admin_identity())

    assert only_request(httpx_mock).content == b""


async def test_delete_with_204_returns_null_data(
    executor: GiteaExecutor, httpx_mock: HTTPXMock
) -> None:
    httpx_mock.add_response(status_code=204)

    outcome = await executor.execute(
        DELETE_BRANCH, {**BASE_ARGS, "branch": "old/branch"}, admin_identity()
    )

    assert only_request(httpx_mock).method == "DELETE"
    assert outcome.is_error is False
    assert outcome.status == 204
    assert outcome.payload["data"] is None


# Responses


async def test_redirects_are_never_followed(executor: GiteaExecutor, httpx_mock: HTTPXMock) -> None:
    httpx_mock.add_response(status_code=302, headers={"Location": "https://evil.example.net/steal"})

    outcome = await executor.execute(LIST_BRANCHES, BASE_ARGS, admin_identity())

    assert len(httpx_mock.get_requests()) == 1
    error = error_of(outcome)
    assert error["kind"] == "redirect"
    assert error["status"] == 302
    assert outcome.status == 302
    assert "evil.example.net" not in json.dumps(outcome.payload)


async def test_text_response(executor: GiteaExecutor, httpx_mock: HTTPXMock) -> None:
    httpx_mock.add_response(
        text="line1\nline2", headers={"content-type": "text/plain; charset=utf-8"}
    )

    outcome = await executor.execute(
        PULL_DIFF, {**BASE_ARGS, "index": 1, "diffType": "patch"}, admin_identity()
    )

    assert outcome.payload["data"] == "line1\nline2"
    assert "truncated" not in outcome.payload


@pytest.mark.parametrize(
    "content_type",
    ["application/xml", "application/x-patch", "application/vnd.foo+xml", "text/x-diff"],
)
async def test_text_media_types(
    executor: GiteaExecutor, httpx_mock: HTTPXMock, content_type: str
) -> None:
    httpx_mock.add_response(content=b"<a/>", headers={"content-type": content_type})

    outcome = await executor.execute(LIST_BRANCHES, BASE_ARGS, admin_identity())

    assert outcome.payload["data"] == "<a/>"


async def test_json_variants_and_fallback_to_text(
    executor: GiteaExecutor, httpx_mock: HTTPXMock
) -> None:
    httpx_mock.add_response(
        content=b'{"a": 1}', headers={"content-type": "application/vnd.api+json"}
    )
    httpx_mock.add_response(content=b"not json {", headers={"content-type": "application/json"})
    httpx_mock.add_response(content=b"\xff\xfe bad", headers={"content-type": "application/json"})
    httpx_mock.add_response(content=b"[1, 2]", headers={})

    outcomes = [
        await executor.execute(LIST_BRANCHES, BASE_ARGS, admin_identity()) for _ in range(4)
    ]

    assert outcomes[0].payload["data"] == {"a": 1}
    assert outcomes[1].payload["data"] == "not json {"
    assert outcomes[2].payload["data"] == "�� bad"
    assert outcomes[3].payload["data"] == [1, 2]


async def test_binary_response_is_base64_with_size(
    executor: GiteaExecutor, httpx_mock: HTTPXMock
) -> None:
    blob = bytes(range(200))
    httpx_mock.add_response(content=blob, headers={"content-type": "application/octet-stream"})
    httpx_mock.add_response(content=b"PK", headers={"content-type": "application/zip"})
    httpx_mock.add_response(content=b"zz", headers={"content-type": "image/png; foo=bar"})

    first = await executor.execute(
        GET_RAW_FILE, {**BASE_ARGS, "filepath": "a.bin"}, admin_identity()
    )
    second = await executor.execute(LIST_BRANCHES, BASE_ARGS, admin_identity())
    third = await executor.execute(LIST_BRANCHES, BASE_ARGS, admin_identity())

    assert first.payload["data"] == {
        "content_base64": base64.b64encode(blob).decode(),
        "content_type": "application/octet-stream",
        "size": 200,
    }
    assert second.payload["data"]["content_type"] == "application/zip"
    assert third.payload["data"]["content_type"] == "image/png; foo=bar"
    assert third.payload["data"]["size"] == 2


async def test_binary_over_limit_by_content_length(httpx_mock: HTTPXMock) -> None:
    executor = GiteaExecutor(make_settings(max_binary_bytes=10))
    httpx_mock.add_response(content=b"x" * 11, headers={"content-type": "application/zip"})

    outcome = await executor.execute(LIST_BRANCHES, BASE_ARGS, admin_identity())

    error = error_of(outcome)
    assert error["kind"] == "payload_too_large"
    assert error["size"] == 11
    assert error["limit"] == 10
    assert "11 bytes" in error["message"]
    assert outcome.status == 200
    await executor.aclose()


async def test_binary_exactly_at_limit_is_allowed(httpx_mock: HTTPXMock) -> None:
    executor = GiteaExecutor(make_settings(max_binary_bytes=10))
    httpx_mock.add_response(content=b"x" * 10, headers={"content-type": "application/zip"})

    outcome = await executor.execute(LIST_BRANCHES, BASE_ARGS, admin_identity())

    assert outcome.payload["data"]["size"] == 10
    await executor.aclose()


class _ChunkedStream(httpx.AsyncByteStream):
    async def __aiter__(self) -> AsyncIterator[bytes]:
        for _ in range(5):
            yield b"x" * 4


async def test_binary_over_limit_without_declared_size(httpx_mock: HTTPXMock) -> None:
    executor = GiteaExecutor(make_settings(max_binary_bytes=10))
    httpx_mock.add_response(headers={"content-type": "application/zip"}, stream=_ChunkedStream())
    httpx_mock.add_response(
        content=b"x" * 20,
        headers={"content-type": "application/zip", "content-length": "bogus"},
    )

    first = await executor.execute(LIST_BRANCHES, BASE_ARGS, admin_identity())
    second = await executor.execute(LIST_BRANCHES, BASE_ARGS, admin_identity())

    for outcome in (first, second):
        error = error_of(outcome)
        assert error["kind"] == "payload_too_large"
        assert error["size"] is None
        assert "larger than the limit of 10 bytes" in error["message"]
    await executor.aclose()


def test_declared_size_ignores_encoded_and_negative_lengths() -> None:
    encoded = httpx.Response(200, headers={"content-encoding": "gzip", "content-length": "5"})
    negative = httpx.Response(200, headers={"content-length": "-1"})
    plain = httpx.Response(200, headers={"content-length": "5"})

    assert executor_module._declared_size(encoded) is None
    assert executor_module._declared_size(negative) is None
    assert executor_module._declared_size(plain) == 5


async def test_json_body_over_hard_cap_is_an_error(
    executor: GiteaExecutor, httpx_mock: HTTPXMock, monkeypatch: pytest.MonkeyPatch
) -> None:
    monkeypatch.setattr(executor_module, "_MAX_TEXT_BYTES", 20)
    httpx_mock.add_response(json={"k": "v" * 100})

    outcome = await executor.execute(LIST_BRANCHES, BASE_ARGS, admin_identity())

    assert error_of(outcome)["kind"] == "payload_too_large"


async def test_text_body_over_hard_cap_is_truncated(
    executor: GiteaExecutor, httpx_mock: HTTPXMock, monkeypatch: pytest.MonkeyPatch
) -> None:
    monkeypatch.setattr(executor_module, "_MAX_TEXT_BYTES", 20)
    httpx_mock.add_response(text="é" * 50, headers={"content-type": "text/plain"})

    outcome = await executor.execute(LIST_BRANCHES, BASE_ARGS, admin_identity())

    assert outcome.is_error is False
    assert outcome.payload["truncated"] == {"returned": 10, "available": None, "unit": "chars"}
    assert outcome.payload["data"] == "é" * 10


async def test_exact_chunk_boundary_for_hard_cap(
    executor: GiteaExecutor, httpx_mock: HTTPXMock, monkeypatch: pytest.MonkeyPatch
) -> None:
    monkeypatch.setattr(executor_module, "_MAX_TEXT_BYTES", 10)
    httpx_mock.add_response(text="a" * 10, headers={"content-type": "text/plain"})

    outcome = await executor.execute(LIST_BRANCHES, BASE_ARGS, admin_identity())

    assert "truncated" not in outcome.payload
    assert outcome.payload["data"] == "a" * 10


async def test_list_is_truncated_to_max_items(httpx_mock: HTTPXMock) -> None:
    executor = GiteaExecutor(make_settings(max_tool_response_items=3))
    httpx_mock.add_response(json=[{"i": i} for i in range(5)])
    httpx_mock.add_response(json=[{"i": i} for i in range(3)])

    big = await executor.execute(LIST_BRANCHES, BASE_ARGS, admin_identity())
    exact = await executor.execute(LIST_BRANCHES, BASE_ARGS, admin_identity())

    assert big.payload["data"] == [{"i": 0}, {"i": 1}, {"i": 2}]
    assert big.payload["truncated"] == {"returned": 3, "available": 5, "unit": "items"}
    assert "truncated" not in exact.payload
    await executor.aclose()


async def test_text_over_char_limit_is_truncated_to_fit_envelope(httpx_mock: HTTPXMock) -> None:
    executor = GiteaExecutor(make_settings(max_tool_response_chars=1000))
    httpx_mock.add_response(text='é"' * 1000, headers={"content-type": "text/plain"})

    outcome = await executor.execute(LIST_BRANCHES, BASE_ARGS, admin_identity())

    assert len(outcome.to_text()) <= 1000
    truncated = outcome.payload["truncated"]
    assert truncated["available"] == 2000
    assert truncated["unit"] == "chars"
    assert truncated["returned"] == len(outcome.payload["data"])
    assert 0 < truncated["returned"] < 2000
    assert list(outcome.payload)[-2:] == ["untrusted_content", "data"]
    await executor.aclose()


async def test_huge_json_object_becomes_truncated_json_string(httpx_mock: HTTPXMock) -> None:
    executor = GiteaExecutor(make_settings(max_tool_response_chars=700))
    httpx_mock.add_response(json={"blob": "x" * 5000})

    outcome = await executor.execute(LIST_BRANCHES, BASE_ARGS, admin_identity())

    assert len(outcome.to_text()) <= 700
    assert isinstance(outcome.payload["data"], str)
    assert outcome.payload["data"].startswith('{"blob": "xxx')
    assert outcome.payload["truncated"]["unit"] == "chars"
    assert outcome.payload["truncated"]["available"] == len(json.dumps({"blob": "x" * 5000}))
    await executor.aclose()


async def test_item_truncation_then_size_truncation(httpx_mock: HTTPXMock) -> None:
    executor = GiteaExecutor(make_settings(max_tool_response_items=2, max_tool_response_chars=900))
    httpx_mock.add_response(json=[{"v": "y" * 400}] * 3)

    outcome = await executor.execute(LIST_BRANCHES, BASE_ARGS, admin_identity())

    truncated = outcome.payload["truncated"]
    assert truncated["items_returned"] == 2
    assert truncated["items_available"] == 3
    assert truncated["unit"] == "chars"
    assert len(outcome.to_text()) <= 900
    await executor.aclose()


async def test_envelope_overhead_larger_than_limit_leaves_empty_data(
    httpx_mock: HTTPXMock,
) -> None:
    executor = GiteaExecutor(make_settings(max_tool_response_chars=10))
    httpx_mock.add_response(json={"a": 1})

    outcome = await executor.execute(LIST_BRANCHES, BASE_ARGS, admin_identity())

    assert outcome.payload["data"] == ""
    assert outcome.payload["truncated"]["returned"] == 0
    await executor.aclose()


async def test_small_response_within_limit_is_untouched(
    executor: GiteaExecutor, httpx_mock: HTTPXMock
) -> None:
    httpx_mock.add_response(json={"a": 1})

    outcome = await executor.execute(LIST_BRANCHES, BASE_ARGS, admin_identity())

    assert outcome.payload["data"] == {"a": 1}
    assert "truncated" not in outcome.payload


# Pagination


async def test_pagination_from_link_and_total_count(
    executor: GiteaExecutor, httpx_mock: HTTPXMock
) -> None:
    link = (
        f'<{API}/repos/octo/demo/branches?page=3&limit=30>; rel="next", '
        f'<{API}/repos/octo/demo/branches?page=4&limit=30>; rel="last", '
        f'<{API}/repos/octo/demo/branches?page=1&limit=30>; rel="first", '
        f'<{API}/repos/octo/demo/branches?page=1&limit=30>; rel="prev"'
    )
    httpx_mock.add_response(json=[], headers={"Link": link, "X-Total-Count": "120"})

    outcome = await executor.execute(LIST_BRANCHES, {**BASE_ARGS, "page": 2}, admin_identity())

    assert outcome.payload["pagination"] == {
        "page": 2,
        "limit": 30,
        "total_count": 120,
        "next_page": 3,
        "last_page": 4,
    }


async def test_pagination_infers_page_when_not_requested(
    executor: GiteaExecutor, httpx_mock: HTTPXMock
) -> None:
    httpx_mock.add_response(
        json=[], headers={"Link": '<https://x/?page=2&limit=5>; rel="next"', "X-Total-Count": "9"}
    )
    httpx_mock.add_response(json=[], headers={"Link": '<https://x/?page=1&limit=5>; rel="prev"'})

    first = await executor.execute(LIST_BRANCHES, BASE_ARGS, admin_identity())
    last = await executor.execute(LIST_BRANCHES, BASE_ARGS, admin_identity())

    assert first.payload["pagination"]["page"] == 1
    assert first.payload["pagination"]["next_page"] == 2
    assert last.payload["pagination"]["page"] == 2
    assert last.payload["pagination"]["last_page"] == 2


async def test_paginated_operation_without_headers_reports_defaults(
    executor: GiteaExecutor, httpx_mock: HTTPXMock
) -> None:
    httpx_mock.add_response(json=[])

    outcome = await executor.execute(LIST_BRANCHES, {**BASE_ARGS, "limit": 50}, admin_identity())

    assert outcome.payload["pagination"] == {
        "page": 1,
        "limit": 50,
        "total_count": None,
        "next_page": None,
        "last_page": 1,
    }


async def test_unpaginated_operation_with_total_count_header(
    executor: GiteaExecutor, httpx_mock: HTTPXMock
) -> None:
    httpx_mock.add_response(json=[], headers={"X-Total-Count": "0"})
    httpx_mock.add_response(json={})

    with_header = await executor.execute(CREATE_ISSUE, BASE_ARGS, admin_identity())
    without = await executor.execute(CREATE_ISSUE, BASE_ARGS, admin_identity())

    assert with_header.payload["pagination"]["total_count"] == 0
    assert "pagination" not in without.payload


# Errors


@pytest.mark.parametrize(
    ("status", "kind"),
    [
        (400, "bad_request"),
        (401, "unauthorized"),
        (403, "forbidden"),
        (404, "not_found"),
        (405, "method_not_allowed"),
        (409, "conflict"),
        (410, "gone"),
        (412, "precondition_failed"),
        (413, "payload_too_large"),
        (422, "validation_failed"),
        (423, "locked"),
        (429, "rate_limited"),
        (418, "client_error"),
        (500, "server_error"),
        (503, "server_error"),
    ],
)
async def test_http_error_kinds(
    executor: GiteaExecutor, httpx_mock: HTTPXMock, status: int, kind: str
) -> None:
    httpx_mock.add_response(
        status_code=status, json={"message": "nope", "url": "https://x/swagger"}
    )

    outcome = await executor.execute(LIST_BRANCHES, BASE_ARGS, admin_identity())

    assert outcome.is_error
    assert outcome.status == status
    assert outcome.payload == {
        "error": {"status": status, "kind": kind, "message": "nope"},
        "identity": {"login": "admin", "mode": "admin-token", "sudo": None},
        "request": {
            "method": "GET",
            "path": "/repos/octo/demo/branches",
            "operation": LIST_BRANCHES.tool_name,
        },
    }


async def test_error_message_joins_validation_errors(
    executor: GiteaExecutor, httpx_mock: HTTPXMock
) -> None:
    httpx_mock.add_response(
        status_code=422,
        json={"message": "Validation failed", "errors": ["title is empty", 7]},
    )

    outcome = await executor.execute(CREATE_ISSUE, BASE_ARGS, admin_identity())

    assert error_of(outcome)["message"] == "Validation failed; title is empty; 7"


@pytest.mark.parametrize(
    ("kwargs", "expected"),
    [
        ({"text": "<html>  Bad\n gateway </html>"}, "<html> Bad gateway </html>"),
        ({"content": b""}, "Gitea returned HTTP 502"),
        ({"json": {"url": "only"}}, "Gitea returned HTTP 502"),
        ({"json": {"message": "", "errors": "notalist"}}, "Gitea returned HTTP 502"),
        ({"json": ["a", "b"]}, '["a","b"]'),
        ({"content": b"\xff\xfe"}, "��"),
    ],
)
async def test_error_message_fallbacks(
    executor: GiteaExecutor, httpx_mock: HTTPXMock, kwargs: dict[str, Any], expected: str
) -> None:
    httpx_mock.add_response(status_code=502, **kwargs)

    outcome = await executor.execute(LIST_BRANCHES, BASE_ARGS, admin_identity())

    assert error_of(outcome)["message"] == expected


async def test_error_message_is_sanitized_and_truncated(
    executor: GiteaExecutor, httpx_mock: HTTPXMock
) -> None:
    message = f"Authorization: Bearer abcdefghijklmnop failed {GH_TOKEN} " + "z" * 600
    httpx_mock.add_response(status_code=400, json={"message": message})

    outcome = await executor.execute(LIST_BRANCHES, BASE_ARGS, admin_identity())

    shown = error_of(outcome)["message"]
    assert "abcdefghijklmnop" not in shown
    assert GH_TOKEN not in shown
    assert len(shown) <= 500
    assert shown.endswith("...")


async def test_error_body_read_is_capped(
    executor: GiteaExecutor, httpx_mock: HTTPXMock, monkeypatch: pytest.MonkeyPatch
) -> None:
    monkeypatch.setattr(executor_module, "_MAX_ERROR_BODY_BYTES", 5)
    httpx_mock.add_response(status_code=500, text="abcdefghij")

    outcome = await executor.execute(LIST_BRANCHES, BASE_ARGS, admin_identity())

    assert error_of(outcome)["message"] == "abcde"


@pytest.mark.parametrize(
    ("exception", "kind"),
    [
        (httpx.ReadTimeout("slow https://gitea.example.com/x?access_token=SECRET"), "timeout"),
        (httpx.ConnectTimeout("slow"), "timeout"),
        (
            httpx.ConnectError("refused https://gitea.example.com/x?access_token=SECRET"),
            "upstream_unreachable",
        ),
        (httpx.RemoteProtocolError("broken"), "upstream_unreachable"),
    ],
)
async def test_transport_errors(
    executor: GiteaExecutor, httpx_mock: HTTPXMock, exception: Exception, kind: str
) -> None:
    httpx_mock.add_exception(exception)

    outcome = await executor.execute(LIST_BRANCHES, {**BASE_ARGS, "page": 1}, admin_identity())

    error = error_of(outcome)
    assert error["kind"] == kind
    assert error["status"] is None
    assert outcome.status is None
    assert "exception" not in error
    dumped = json.dumps(outcome.payload)
    assert "SECRET" not in dumped
    assert "page=1" not in dumped
    assert "gitea.example.com" not in dumped


async def test_timeout_message_names_configured_seconds(httpx_mock: HTTPXMock) -> None:
    executor = GiteaExecutor(make_settings(request_timeout_seconds=7))
    httpx_mock.add_exception(httpx.ReadTimeout("slow"))

    outcome = await executor.execute(LIST_BRANCHES, BASE_ARGS, admin_identity())

    assert "7s" in error_of(outcome)["message"]
    await executor.aclose()


async def test_expose_error_details_adds_only_class_name(httpx_mock: HTTPXMock) -> None:
    executor = GiteaExecutor(make_settings(expose_error_details=True))
    httpx_mock.add_exception(httpx.ConnectError("refused"))

    outcome = await executor.execute(LIST_BRANCHES, BASE_ARGS, admin_identity())

    error = error_of(outcome)
    assert error["exception"] == "ConnectError"
    assert "Traceback" not in json.dumps(outcome.payload)
    await executor.aclose()


class _BrokenStream(httpx.AsyncByteStream):
    async def __aiter__(self) -> AsyncIterator[bytes]:
        yield b"partial"
        raise httpx.ReadError("connection reset")


async def test_failure_while_reading_the_body(
    executor: GiteaExecutor, httpx_mock: HTTPXMock
) -> None:
    httpx_mock.add_response(headers={"content-type": "application/json"}, stream=_BrokenStream())

    outcome = await executor.execute(LIST_BRANCHES, BASE_ARGS, admin_identity())

    assert error_of(outcome)["kind"] == "upstream_unreachable"


# Redaction


@pytest.mark.parametrize(
    ("secret", "identity"),
    [
        (GITEA_TOKEN, admin_identity()),
        (USER_TOKEN, oauth_identity()),
        (BEARER_TOKEN, admin_identity()),
        (OAUTH_CLIENT_SECRET, admin_identity()),
        ("extra-plain-secret-value", admin_identity()),
    ],
)
async def test_every_secret_source_is_redacted_everywhere(
    httpx_mock: HTTPXMock, secret: str, identity: Identity
) -> None:
    executor = GiteaExecutor(make_settings(), extra_secrets=["extra-plain-secret-value"])
    echoed = {
        "plain": f"x {secret} y",
        "nested": [{"k": secret}],
        f"key-{secret}": "v",
        "n": 1,
    }
    httpx_mock.add_response(json=echoed)
    httpx_mock.add_response(status_code=403, json={"message": f"bad token {secret}"})

    ok = await executor.execute(LIST_BRANCHES, BASE_ARGS, identity)
    failed = await executor.execute(LIST_BRANCHES, BASE_ARGS, identity)

    assert secret not in json.dumps(ok.payload)
    assert secret not in json.dumps(failed.payload)
    assert ok.payload["data"]["plain"] == "x [REDACTED] y"
    assert ok.payload["data"]["nested"] == [{"k": "[REDACTED]"}]
    assert ok.payload["data"]["n"] == 1
    assert "[REDACTED]" in failed.payload["error"]["message"]
    await executor.aclose()


async def test_redaction_applies_even_when_secret_detection_is_off(httpx_mock: HTTPXMock) -> None:
    executor = GiteaExecutor(make_settings(secret_detection_mode="off"))
    httpx_mock.add_response(json={"echo": GITEA_TOKEN, "pattern": GH_TOKEN})

    outcome = await executor.execute(LIST_BRANCHES, BASE_ARGS, admin_identity())

    assert outcome.payload["data"] == {"echo": "[REDACTED]", "pattern": GH_TOKEN}
    await executor.aclose()


async def test_exact_redaction_with_no_configured_secrets(httpx_mock: HTTPXMock) -> None:
    settings = make_settings(
        gitea_token=None, mcp_bearer_token=None, gitea_oauth_client_secret=None
    )
    executor = GiteaExecutor(settings)
    httpx_mock.add_response(json={"echo": USER_TOKEN})

    outcome = await executor.execute(LIST_BRANCHES, BASE_ARGS, oauth_identity())

    assert outcome.payload["data"] == {"echo": "[REDACTED]"}
    await executor.aclose()


async def test_mask_mode_masks_secret_like_strings(
    executor: GiteaExecutor, httpx_mock: HTTPXMock
) -> None:
    httpx_mock.add_response(json={"note": f"see {GH_TOKEN}"})

    outcome = await executor.execute(LIST_BRANCHES, BASE_ARGS, admin_identity())

    assert GH_TOKEN not in json.dumps(outcome.payload)
    assert "ghp_...N4o5" in outcome.payload["data"]["note"]
    assert outcome.is_error is False


async def test_credential_issuing_operations_skip_masking_but_not_exact_redaction(
    executor: GiteaExecutor, httpx_mock: HTTPXMock
) -> None:
    minted = {"sha1": GH_TOKEN, "leak": GITEA_TOKEN}
    httpx_mock.add_response(status_code=201, json=minted)

    outcome = await executor.execute(
        CREATE_TOKEN, {"username": "alice", "body": {"name": "ci"}}, admin_identity()
    )

    assert CREATE_TOKEN.operation_id in CREDENTIAL_ISSUING_OPERATIONS
    assert outcome.is_error is False
    assert outcome.payload["data"] == {"sha1": GH_TOKEN, "leak": "[REDACTED]"}


async def test_credential_issuing_errors_are_still_masked(
    executor: GiteaExecutor, httpx_mock: HTTPXMock
) -> None:
    httpx_mock.add_response(status_code=400, json={"message": f"x {GH_TOKEN}"})

    outcome = await executor.execute(
        CREATE_TOKEN, {"username": "alice", "body": {"name": "ci"}}, admin_identity()
    )

    assert GH_TOKEN not in json.dumps(outcome.payload)


async def test_registration_token_is_returned_from_the_response_header(
    executor: GiteaExecutor, httpx_mock: HTTPXMock
) -> None:
    httpx_mock.add_response(headers={"token": GH_TOKEN})
    httpx_mock.add_response()

    issued = await executor.execute(REGISTRATION_TOKEN, BASE_ARGS, admin_identity())
    absent = await executor.execute(REGISTRATION_TOKEN, BASE_ARGS, admin_identity())

    assert issued.payload["data"] == {"token": GH_TOKEN}
    assert absent.payload["data"] is None


async def test_header_data_is_ignored_for_other_operations(
    executor: GiteaExecutor, httpx_mock: HTTPXMock
) -> None:
    httpx_mock.add_response(headers={"token": "not-for-you"})

    outcome = await executor.execute(DELETE_BRANCH, {**BASE_ARGS, "branch": "b"}, admin_identity())

    assert outcome.payload["data"] is None


async def test_block_mode_blocks_secret_like_results(httpx_mock: HTTPXMock) -> None:
    executor = GiteaExecutor(make_settings(secret_detection_mode="block"))
    httpx_mock.add_response(json={"note": f"see {GH_TOKEN}"})
    httpx_mock.add_response(json={"note": "clean"})
    httpx_mock.add_response(status_code=400, json={"message": f"leak {GH_TOKEN}"})

    blocked = await executor.execute(LIST_BRANCHES, BASE_ARGS, admin_identity())
    clean = await executor.execute(LIST_BRANCHES, BASE_ARGS, admin_identity())
    blocked_error = await executor.execute(LIST_BRANCHES, BASE_ARGS, admin_identity())

    assert blocked.is_error
    assert blocked.status == 200
    assert error_of(blocked)["kind"] == "secret_blocked"
    assert error_of(blocked)["status"] == 200
    assert set(blocked.payload) == {"error", "identity", "request"}
    assert GH_TOKEN not in json.dumps(blocked.payload)
    assert clean.is_error is False
    assert clean.payload["data"] == {"note": "clean"}
    assert error_of(blocked_error)["kind"] == "bad_request"
    assert GH_TOKEN not in json.dumps(blocked_error.payload)
    assert blocked_error.status == 400
    await executor.aclose()


async def test_block_mode_exempts_credential_issuing_results(httpx_mock: HTTPXMock) -> None:
    executor = GiteaExecutor(make_settings(secret_detection_mode="block"))
    httpx_mock.add_response(status_code=201, json={"sha1": GH_TOKEN})

    outcome = await executor.execute(
        CREATE_TOKEN, {"username": "alice", "body": {"name": "ci"}}, admin_identity()
    )

    assert outcome.is_error is False
    assert outcome.payload["data"] == {"sha1": GH_TOKEN}
    await executor.aclose()


async def test_block_mode_does_not_echo_secret_like_request_path(httpx_mock: HTTPXMock) -> None:
    executor = GiteaExecutor(make_settings(secret_detection_mode="block"))
    httpx_mock.add_response(json={"note": f"see {GH_TOKEN}"})

    outcome = await executor.execute(
        LIST_BRANCHES, {"owner": GH_TOKEN, "repo": "demo"}, admin_identity()
    )

    assert error_of(outcome)["kind"] == "secret_blocked"
    assert GH_TOKEN not in json.dumps(outcome.payload)
    await executor.aclose()


async def test_action_secret_value_is_never_echoed(
    executor: GiteaExecutor, httpx_mock: HTTPXMock
) -> None:
    value = "s3cr3t-deploy-value-42"
    httpx_mock.add_response(status_code=422, json={"message": f"invalid secret {value}"})
    httpx_mock.add_response(status_code=201, json={"echo": value, "nested": [value]})
    arguments = {**BASE_ARGS, "secretname": "DEPLOY", "body": {"data": value, "description": "d"}}

    failed = await executor.execute(PUT_SECRET, arguments, admin_identity())
    echoed = await executor.execute(PUT_SECRET, arguments, admin_identity())

    assert value not in json.dumps(failed.payload)
    assert value not in json.dumps(echoed.payload)
    assert value in httpx_mock.get_requests()[0].content.decode()
    assert echoed.payload["data"]["echo"] == "[REDACTED]"


async def test_short_action_secret_value_is_redacted_too(
    executor: GiteaExecutor, httpx_mock: HTTPXMock
) -> None:
    httpx_mock.add_response(status_code=422, json={"message": "bad ZQ value"})

    outcome = await executor.execute(
        PUT_SECRET,
        {**BASE_ARGS, "secretname": "S", "body": {"data": "ZQ"}},
        admin_identity(),
    )

    assert "ZQ" not in json.dumps(outcome.payload)


async def test_sensitive_body_fields_are_redacted_from_results(
    executor: GiteaExecutor, httpx_mock: HTTPXMock
) -> None:
    httpx_mock.add_response(
        status_code=422, json={"message": "password Hunter2-Hunter2 rejected, ab, hook-secret-1"}
    )
    body = {
        "username": "bob",
        "password": "Hunter2-Hunter2",
        "short": {"password": "ab"},
        "config": {"secret": "hook-secret-1"},
        "list": [{"token": "list-token-1"}],
    }

    outcome = await executor.execute(CREATE_USER, {"body": body}, admin_identity())

    dumped = json.dumps(outcome.payload)
    assert "Hunter2" not in dumped
    assert "hook-secret-1" not in dumped
    assert "rejected, ab" in dumped


async def test_secret_data_value_not_special_for_other_operations(
    executor: GiteaExecutor, httpx_mock: HTTPXMock
) -> None:
    httpx_mock.add_response(status_code=201, json={"title": "keep-me"})

    outcome = await executor.execute(
        CREATE_ISSUE, {**BASE_ARGS, "body": {"data": "keep-me"}}, admin_identity()
    )

    assert outcome.payload["data"] == {"title": "keep-me"}


async def test_body_deeper_than_limit_is_not_scanned(
    executor: GiteaExecutor, httpx_mock: HTTPXMock
) -> None:
    deep: Any = {"password": "deepdeepdeep"}
    for _ in range(12):
        deep = {"x": deep}
    httpx_mock.add_response(status_code=201, json={})

    outcome = await executor.execute(CREATE_USER, {"body": deep}, admin_identity())

    assert outcome.is_error is False


# Raw request


async def test_raw_request_get(executor: GiteaExecutor, httpx_mock: HTTPXMock) -> None:
    httpx_mock.add_response(json={"version": "1.2"})

    outcome = await executor.request(
        "get",
        "/repos/octo/dem ö/issues",
        oauth_identity(),
        query={"state": "open", "labels": ["a", "b"], "draft": True, "skip": None},
        sudo="bob",
        read_only_hint=True,
    )

    request = only_request(httpx_mock)
    assert request.method == "GET"
    assert request.url.raw_path == (
        b"/api/v1/repos/octo/dem%20%C3%B6/issues?state=open&labels=a&labels=b&draft=true"
    )
    assert request.headers["Authorization"] == f"Bearer {USER_TOKEN}"
    assert request.headers["Sudo"] == "bob"
    assert outcome.payload["request"] == {
        "method": "GET",
        "path": "/repos/octo/dem ö/issues",
        "operation": RAW_TOOL_NAME,
    }
    assert outcome.payload["identity"]["sudo"] == "bob"
    assert outcome.payload["data"] == {"version": "1.2"}
    assert "pagination" not in outcome.payload


async def test_raw_request_with_body_and_pagination(
    executor: GiteaExecutor, httpx_mock: HTTPXMock
) -> None:
    httpx_mock.add_response(json=[])

    outcome = await executor.request(
        "POST", "/repos/o/r/issues", admin_identity(), query={"page": 2}, body={"title": "t"}
    )

    request = only_request(httpx_mock)
    assert json.loads(request.content) == {"title": "t"}
    assert request.headers["Content-Type"] == "application/json"
    assert outcome.payload["pagination"]["page"] == 2


async def test_raw_request_redacts_secret_writes(
    executor: GiteaExecutor, httpx_mock: HTTPXMock
) -> None:
    value = "raw-secret-value-77"
    httpx_mock.add_response(status_code=422, json={"message": f"bad {value}"})

    outcome = await executor.request(
        "PUT", "/repos/o/r/actions/secrets/NAME", admin_identity(), body={"data": value}
    )

    assert value not in json.dumps(outcome.payload)


async def test_raw_request_data_field_is_not_secret_outside_secret_paths(
    executor: GiteaExecutor, httpx_mock: HTTPXMock
) -> None:
    httpx_mock.add_response(json={"echo": "plainvalue"})

    outcome = await executor.request(
        "POST", "/repos/o/r/hooks", admin_identity(), body={"data": "plainvalue"}
    )

    assert outcome.payload["data"] == {"echo": "plainvalue"}


@pytest.mark.parametrize(
    "path",
    [
        "repos/o/r",
        "/",
        "//x",
        "/a/../b",
        "/a/./b",
        "/a//b",
        "/a/",
        "/a?b=1",
        "/a#x",
        "/a%2Fb",
        "/a\\b",
        "/a\x00",
    ],
)
async def test_raw_request_rejects_unsafe_paths(
    executor: GiteaExecutor, httpx_mock: HTTPXMock, path: str
) -> None:
    outcome = await executor.request("GET", path, admin_identity())

    assert error_of(outcome)["kind"] == "invalid_argument"
    assert outcome.payload["request"]["operation"] == RAW_TOOL_NAME
    assert httpx_mock.get_requests() == []


async def test_raw_request_rejects_bad_method_sudo_and_query(
    executor: GiteaExecutor, httpx_mock: HTTPXMock
) -> None:
    results = [
        await executor.request("TRACE", "/user", admin_identity()),
        await executor.request(5, "/user", admin_identity()),  # type: ignore[arg-type]
        await executor.request("GET", "/user", admin_identity(), sudo="bad user"),
        await executor.request("GET", "/user", admin_identity(), query={"access_token": "x"}),
        await executor.request("GET", "/user", admin_identity(), query={"sudo": "root"}),
        await executor.request("GET", "/user", admin_identity(), query={"a": object()}),
        await executor.request("GET", 5, admin_identity()),  # type: ignore[arg-type]
    ]

    assert all(error_of(result)["kind"] == "invalid_argument" for result in results)
    assert results[0].payload["request"]["method"] == "(invalid)"
    assert results[-1].payload["request"]["path"] == "(invalid)"
    assert httpx_mock.get_requests() == []


async def test_raw_request_echoes_only_a_bounded_invalid_path(
    executor: GiteaExecutor,
) -> None:
    outcome = await executor.request("GET", "x" * 500, admin_identity())

    assert len(outcome.payload["request"]["path"]) == 200


# Client handling


async def test_cookies_are_not_persisted_between_calls(httpx_mock: HTTPXMock) -> None:
    executor = GiteaExecutor(make_settings())
    httpx_mock.add_response(json=[], headers={"Set-Cookie": "session=abc; Path=/"})
    httpx_mock.add_response(json=[])

    await executor.execute(LIST_BRANCHES, BASE_ARGS, admin_identity())
    await executor.execute(LIST_BRANCHES, BASE_ARGS, oauth_identity())

    assert "Cookie" not in httpx_mock.get_requests()[1].headers
    await executor.aclose()


async def test_aclose_closes_only_owned_clients() -> None:
    shared = httpx.AsyncClient()
    borrowed = GiteaExecutor(make_settings(), client=shared)
    owned = GiteaExecutor(make_settings())

    await borrowed.aclose()
    assert shared.is_closed is False
    inner = owned._client
    await owned.aclose()
    assert inner.is_closed is True
    await shared.aclose()


async def test_injected_client_is_used_without_following_redirects(
    httpx_mock: HTTPXMock,
) -> None:
    client = httpx.AsyncClient(follow_redirects=True)
    executor = GiteaExecutor(make_settings(), client=client)
    httpx_mock.add_response(status_code=301, headers={"Location": "https://other.example.org/"})

    outcome = await executor.execute(LIST_BRANCHES, BASE_ARGS, admin_identity())

    assert error_of(outcome)["kind"] == "redirect"
    assert len(httpx_mock.get_requests()) == 1
    await client.aclose()


async def test_timeout_comes_from_settings(httpx_mock: HTTPXMock) -> None:
    executor = GiteaExecutor(make_settings(request_timeout_seconds=12))
    httpx_mock.add_response(json=[])

    await executor.execute(LIST_BRANCHES, BASE_ARGS, admin_identity())

    assert only_request(httpx_mock).extensions["timeout"] == {
        "connect": 12.0,
        "read": 12.0,
        "write": 12.0,
        "pool": 12.0,
    }
    await executor.aclose()


async def test_api_url_comes_from_settings(httpx_mock: HTTPXMock) -> None:
    executor = GiteaExecutor(make_settings(gitea_url="http://localhost:3000/gitea/api/v1/"))
    httpx_mock.add_response(json=[])

    await executor.execute(LIST_BRANCHES, BASE_ARGS, admin_identity())

    assert str(only_request(httpx_mock).url).startswith("http://localhost:3000/gitea/api/v1/repos/")
    await executor.aclose()


async def test_query_parameter_named_body_is_mapped_by_arg_name(
    executor: GiteaExecutor, httpx_mock: HTTPXMock
) -> None:
    operation = make_op(
        "repoGetFileContents",
        "GET",
        "/repos/{owner}/{repo}/file-contents",
        [
            param("owner", "path"),
            param("repo", "path"),
            Parameter(
                name="body",
                arg_name="query_body",
                location="query",
                required=False,
                schema={"type": "string"},
            ),
        ],
    )
    httpx_mock.add_response(json=[])

    await executor.execute(
        operation, {**BASE_ARGS, "query_body": "files", "body": {"ignored": 1}}, admin_identity()
    )

    request = only_request(httpx_mock)
    assert request.url.query == b"body=files"
    assert request.content == b""