cleveragents-core/robot/helper_a2a_facade.py

"""Helper script for a2a_facade.robot smoke tests.

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

from __future__ import annotations

import sys
from pathlib import Path

# 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.a2a.errors import (  # noqa: E402
    A2aNotAvailableError,
    A2aVersionMismatchError,
)
from cleveragents.a2a.events import A2aEventQueue  # noqa: E402
from cleveragents.a2a.facade import A2aLocalFacade  # noqa: E402
from cleveragents.a2a.models import A2aEvent, A2aRequest  # noqa: E402
from cleveragents.a2a.transport import A2aHttpTransport  # noqa: E402
from cleveragents.a2a.versioning import A2aVersionNegotiator  # noqa: E402

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


def facade_dispatch() -> None:
    """Dispatch session.create via local facade."""
    facade = A2aLocalFacade()
    request = A2aRequest(method="session.create")
    response = facade.dispatch(request)
    if (
        response.error is None
        and response.result is not None
        and "session_id" in response.result
    ):
        print("a2a-facade-dispatch-ok")
    else:
        print(f"FAIL: unexpected response {response}", file=sys.stderr)
        sys.exit(1)


def transport_stub() -> None:
    """Verify transport stub raises A2aNotAvailableError."""
    transport = A2aHttpTransport()
    try:
        transport.connect("http://localhost:8080")
        print("FAIL: should have raised", file=sys.stderr)
        sys.exit(1)
    except A2aNotAvailableError:
        pass

    if transport.is_connected() is False:
        print("a2a-transport-stub-ok")
    else:
        print("FAIL: is_connected should be False", file=sys.stderr)
        sys.exit(1)


def event_queue() -> None:
    """Verify local event queue publish/subscribe."""
    queue = A2aEventQueue()
    received: list[A2aEvent] = []
    queue.subscribe_local(lambda e: received.append(e))
    queue.publish(A2aEvent(event_type="test.event"))
    if len(received) == 1 and received[0].event_type == "test.event":
        print("a2a-event-queue-ok")
    else:
        print(f"FAIL: received={received}", file=sys.stderr)
        sys.exit(1)


def version_negotiate() -> None:
    """Verify version negotiation."""
    negotiator = A2aVersionNegotiator()
    result = negotiator.negotiate("1.0")
    if result != "1.0":
        print("FAIL: expected 1.0", file=sys.stderr)
        sys.exit(1)

    try:
        negotiator.negotiate("2.0")
        print("FAIL: should have raised", file=sys.stderr)
        sys.exit(1)
    except A2aVersionMismatchError:
        pass

    print("a2a-version-negotiate-ok")


def list_operations() -> None:
    """Verify list_operations returns expected operations."""
    facade = A2aLocalFacade()
    ops = facade.list_operations()
    expected = {"session.create", "plan.create", "plan.execute", "context.get"}
    if expected.issubset(set(ops)) and len(ops) == 44:
        print("a2a-list-operations-ok")
    else:
        print(f"FAIL: ops={ops}", file=sys.stderr)
        sys.exit(1)


# ---------------------------------------------------------------------------
# Dispatch
# ---------------------------------------------------------------------------

_COMMANDS = {
    "facade-dispatch": facade_dispatch,
    "transport-stub": transport_stub,
    "event-queue": event_queue,
    "version-negotiate": version_negotiate,
    "list-operations": list_operations,
}


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(2)
    _COMMANDS[sys.argv[1]]()


if __name__ == "__main__":
    main()