cleveragents-core/robot/helper_server_http_client.py

"""Helper script for server_http_client.robot integration tests.

Each subcommand is a self-contained check that prints a sentinel on success.
"""

from __future__ import annotations

import sys
from pathlib import Path
from unittest.mock import patch

import httpx

# Ensure local source tree is importable
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
    sys.path.insert(0, _SRC)

from cleveragents.client.exceptions import (  # noqa: E402
    ServerConnectionError,
    ServerTimeoutError,
    ServerVersionMismatchError,
)
from cleveragents.client.http_client import (  # noqa: E402
    ServerHttpClient,
    _redact_headers,
)


def _mock_response(
    status: int = 200,
    body: dict | list | None = None,
    headers: dict | None = None,
) -> httpx.Response:
    return httpx.Response(
        status_code=status,
        json=body,
        headers=headers or {},
        request=httpx.Request("GET", "http://mock"),
    )


# ---------------------------------------------------------------------------
# Subcommands
# ---------------------------------------------------------------------------


def client_create() -> None:
    """Verify client construction and properties."""
    client = ServerHttpClient(
        base_url="https://test.example.com",
        api_token="tok_testkey1234567890abc",
        tls_verify=True,
        timeout=15.0,
    )
    assert client.base_url == "https://test.example.com"
    assert client.tls_verify is True
    assert client.timeout == 15.0
    client.close()
    print("server-http-client-create-ok")


def health_check() -> None:
    """Verify health check with mock."""
    client = ServerHttpClient()
    with patch("cleveragents.client.http_client.httpx.request") as mock_req:
        mock_req.return_value = _mock_response(200, {"status": "healthy"})
        result = client.health_check()
    assert result is True
    print("server-http-client-health-ok")


def health_check_fail() -> None:
    """Verify health check failure."""
    client = ServerHttpClient(max_retries=1)
    with patch("cleveragents.client.http_client.httpx.request") as mock_req:
        mock_req.side_effect = httpx.ConnectError("refused")
        result = client.health_check()
    assert result is False
    print("server-http-client-health-fail-ok")


def version() -> None:
    """Verify version retrieval."""
    client = ServerHttpClient()
    with patch("cleveragents.client.http_client.httpx.request") as mock_req:
        mock_req.return_value = _mock_response(200, {"version": "1.2.3"})
        result = client.get_version()
    assert result == "1.2.3"
    print("server-http-client-version-ok")


def negotiate() -> None:
    """Verify version negotiation."""
    client = ServerHttpClient()
    with patch("cleveragents.client.http_client.httpx.request") as mock_req:
        mock_req.return_value = _mock_response(200, {"negotiated_version": "1.0"})
        result = client.negotiate_version()
    assert result == "1.0"
    print("server-http-client-negotiate-ok")


def pagination() -> None:
    """Verify paginated list endpoint."""
    client = ServerHttpClient()
    items = [{"id": str(i)} for i in range(3)]
    with patch("cleveragents.client.http_client.httpx.request") as mock_req:
        mock_req.return_value = _mock_response(
            200, {"items": items, "total": 3, "has_next": False}
        )
        result = client.list_endpoint("/items", page=1, per_page=10)
    assert len(result.items) == 3
    assert result.total == 3
    print("server-http-client-pagination-ok")


def error_503() -> None:
    """Verify 503 maps to A2aNotAvailableError."""
    from cleveragents.a2a.errors import A2aNotAvailableError

    client = ServerHttpClient(max_retries=1)
    with patch("cleveragents.client.http_client.httpx.request") as mock_req:
        mock_req.return_value = _mock_response(503, {"error": "unavailable"})
        try:
            client.get_version()
            print("FAIL: should have raised", file=sys.stderr)
            sys.exit(1)
        except A2aNotAvailableError:
            pass
    print("server-http-client-error-503-ok")


def redaction() -> None:
    """Verify auth header redaction."""
    headers = {
        "Authorization": "Bearer tok_secret123abcdefghij",
        "Accept": "application/json",
    }
    redacted = _redact_headers(headers)
    assert redacted["Authorization"] == "***REDACTED***"
    assert redacted["Accept"] == "application/json"
    print("server-http-client-redaction-ok")


def exceptions() -> None:
    """Verify exception attributes."""
    err1 = ServerConnectionError("test", url="https://x.com", cause=RuntimeError("x"))
    assert err1.url == "https://x.com"
    assert err1.cause is not None

    err2 = ServerTimeoutError("timeout", timeout_seconds=5.0)
    assert err2.timeout_seconds == 5.0

    err3 = ServerVersionMismatchError(
        "mm", client_version="2.0", server_versions=["1.0"]
    )
    assert err3.client_version == "2.0"
    assert "1.0" in err3.server_versions

    print("server-http-client-exceptions-ok")


_COMMANDS = {
    "client-create": client_create,
    "health-check": health_check,
    "health-check-fail": health_check_fail,
    "version": version,
    "negotiate": negotiate,
    "pagination": pagination,
    "error-503": error_503,
    "redaction": redaction,
    "exceptions": exceptions,
}


def main() -> None:
    if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS:
        print(f"Usage: {sys.argv[0]} <{'|'.join(_COMMANDS)}>", file=sys.stderr)
        sys.exit(1)
    _COMMANDS[sys.argv[1]]()


if __name__ == "__main__":
    main()