# Copyright (c) 2026, AG2ai, Inc., AG2ai open-source projects maintainers and core contributors
#
# SPDX-License-Identifier: Apache-2.0

"""Regression tests for issue #2904 — AG2 A2A must reject AgentCards whose
selected interface advertises an A2A protocol version < 1.0, while still
accepting interfaces that omit the optional ``protocol_version`` field."""

import httpx
import pytest
from a2a.client.client_factory import TransportProtocol
from a2a.types import AgentCard, AgentInterface
from a2a.utils.constants import PROTOCOL_VERSION_CURRENT

from ag2 import Agent
from ag2.a2a import A2AConfig, A2AServer, build_card
from ag2.a2a.errors import A2AError, A2AIncompatibleProtocolVersionError
from ag2.a2a.testing import make_test_client_factory
from ag2.a2a.transports import TransportName
from ag2.testing import TestConfig

URL = "http://test"
LEGACY_URL = "http://legacy"
REPLY = "pong"


def _iface(*, url: str, version: str, binding: str = TransportProtocol.JSONRPC.value) -> AgentInterface:
    return AgentInterface(url=url, protocol_binding=binding, protocol_version=version)


def _card(*interfaces: AgentInterface) -> AgentCard:
    """A serviceable AG2 card whose ``supported_interfaces`` are overridden.

    ``AgentCard`` is a protobuf message, so a repeated field is rebuilt by
    clearing and extending — it cannot be assigned to.
    """
    card = build_card(Agent("server", config=TestConfig(REPLY)), url=URL)
    card.ClearField("supported_interfaces")
    card.supported_interfaces.extend(interfaces)
    return card


def _client(card: AgentCard, *, card_url: str = URL, prefer: TransportName | None = None) -> Agent:
    """Client that connects using ``card``, backed by an in-process server.

    ``preset_card`` is what puts ``card`` in front of the version gate; the
    factory is only there to carry the request once the gate lets it through.
    """
    server = A2AServer(Agent("server", config=TestConfig(REPLY)))
    return Agent(
        "client",
        config=A2AConfig(
            card_url=card_url,
            preset_card=card,
            prefer=prefer,
            httpx_client_factory=make_test_client_factory(server, url=URL),
        ),
    )


class _RecordingTransport(httpx.AsyncBaseTransport):
    """In-process transport to ``server`` that records the host of every request."""

    def __init__(self, server: A2AServer) -> None:
        self._inner = httpx.ASGITransport(app=server.build_jsonrpc(url=URL))
        self.hosts: list[str] = []

    async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
        self.hosts.append(request.url.host)
        return await self._inner.handle_async_request(request)


@pytest.mark.asyncio
class TestVersionGate:
    @pytest.mark.parametrize("version", ["1.0", "1.0.0", "2.5", PROTOCOL_VERSION_CURRENT])
    async def test_connects_on_a_compatible_version(self, version: str) -> None:
        reply = await _client(_card(_iface(url=URL, version=version))).ask("ping")

        assert reply.body == REPLY

    async def test_connects_when_the_version_is_absent(self) -> None:
        # ``protocol_version`` is optional and the A2A SDK defaults a missing
        # one to the current version, so an empty field is not legacy.
        reply = await _client(_card(_iface(url=URL, version=""))).ask("ping")

        assert reply.body == REPLY

    @pytest.mark.parametrize("version", ["garbage", "not-a-version"])
    async def test_an_unparsable_version_is_not_treated_as_legacy(self, version: str) -> None:
        # The #2904 guard is specifically that *AG2's* gate stays out of the
        # way here. Connecting still fails — the SDK refuses to build a
        # transport for a version it cannot parse — but that is the SDK's
        # own ValueError, not anything from AG2's A2A error hierarchy.
        client = _client(_card(_iface(url=URL, version=version)))

        with pytest.raises(ValueError) as exc_info:
            await client.ask("ping")

        assert not isinstance(exc_info.value, A2AError)

    @pytest.mark.parametrize("version", ["0.3", "0.9", "0.3.0"])
    async def test_refuses_a_legacy_version(self, version: str) -> None:
        client = _client(_card(_iface(url=URL, version=version)))

        with pytest.raises(A2AIncompatibleProtocolVersionError) as exc_info:
            await client.ask("ping")

        assert exc_info.value.protocol_version == version
        assert exc_info.value.transport == "jsonrpc"
        assert exc_info.value.url == URL


@pytest.mark.asyncio
class TestInterfaceSelection:
    async def test_connect_url_picks_its_own_interface_not_the_first_listed(self) -> None:
        # A legacy interface is listed first, but the connect URL names the
        # current one — resolution must follow the URL, or the version gate
        # would inspect the wrong interface and reject a valid server.
        card = _card(
            _iface(url=LEGACY_URL, version="0.3"),
            _iface(url=URL, version=PROTOCOL_VERSION_CURRENT),
        )

        reply = await _client(card).ask("ping")

        assert reply.body == REPLY

    async def test_requests_go_to_the_interface_that_was_checked(self) -> None:
        # Two current interfaces with the same binding: the one the URL selected (and the gates
        # checked) is the one the requests go to, not the SDK's own first pick for that binding.
        card = _card(
            _iface(url="http://other", version=PROTOCOL_VERSION_CURRENT),
            _iface(url=URL, version=PROTOCOL_VERSION_CURRENT),
        )
        transport = _RecordingTransport(A2AServer(Agent("server", config=TestConfig(REPLY))))
        client = Agent(
            "client",
            config=A2AConfig(
                card_url=URL,
                preset_card=card,
                httpx_client_factory=lambda: httpx.AsyncClient(transport=transport, base_url=URL),
            ),
        )

        reply = await client.ask("ping")

        assert reply.body == REPLY
        assert set(transport.hosts) == {"test"}

    async def test_connecting_to_the_legacy_interface_is_refused(self) -> None:
        card = _card(
            _iface(url=LEGACY_URL, version="0.3"),
            _iface(url=URL, version=PROTOCOL_VERSION_CURRENT),
        )

        client = _client(card, card_url=LEGACY_URL)

        with pytest.raises(A2AIncompatibleProtocolVersionError) as exc_info:
            await client.ask("ping")

        assert exc_info.value.url == LEGACY_URL

    async def test_prefer_overrides_the_url_match(self) -> None:
        # ``prefer`` must win over the URL match. The gRPC interface is the
        # legacy one, so a gRPC-tagged version error is proof that selection
        # went to gRPC rather than quietly staying on JSON-RPC.
        card = _card(
            _iface(url=URL, version=PROTOCOL_VERSION_CURRENT),
            _iface(url="grpc.example:50051", version="0.3", binding=TransportProtocol.GRPC.value),
        )

        client = _client(card, prefer="grpc")

        with pytest.raises(A2AIncompatibleProtocolVersionError) as exc_info:
            await client.ask("ping")

        assert exc_info.value.transport == "grpc"
        assert exc_info.value.protocol_version == "0.3"