From df237e839bee216481d11d00eeb5d75d704022e4 Mon Sep 17 00:00:00 2001 From: CleverThis Date: Mon, 11 May 2026 02:51:35 +0000 Subject: [PATCH] feat(a2a): implement JSON-RPC 2.0 wire format and method routing Enhance the A2A module to better comply with JSON-RPC 2.0 specification: - facade.py: Add explicit JSON-RPC error codes (-32601 for unknown methods, -32602 for invalid params) in dispatch(). Previously unknown operations raised raw exceptions; now they return properly formatted JSON-RPC error responses with correct iANA-registered error code -32601. - stdio_transport.py: Add _validate_jsonrpc_response() static method that validates incoming response dicts against the JSON-RPC 2.0 wire format before passing them to A2aResponse deserialization. This catches protocol violations (wrong jsonrpc version, id mismatch) before they become pydantic errors. Improved error logging includes raw_response for debugging. - tests/a2a/test_models.py: Comprehensive tests for JSON-RPC 2.0 wire format compliance covering all model validators. - tests/a2a/test_facade_dispatch.py: Tests for operation routing verification covering extension methods, legacy names, unknown method handling (-32601), invalid params (-32602), response structure validity. - tests/a2a/test_stdio_transport.py: Tests for wire-format validation helper, serialization correctness, and mocked subprocess scenarios. - tests/a2a/test_versioning.py: Version negotiation tests. - tests/a2a/test_events.py: Event queue lifecycle and SSE formatting tests. --- src/cleveragents/a2a/facade.py | 67 ++++++++- src/cleveragents/a2a/stdio_transport.py | 66 +++++++-- tests/a2a/__init__.py | 0 tests/a2a/test_events.py | 142 ++++++++++++++++++ tests/a2a/test_facade_dispatch.py | 174 ++++++++++++++++++++++ tests/a2a/test_models.py | 184 +++++++++++++++++++++++ tests/a2a/test_stdio_transport.py | 189 ++++++++++++++++++++++++ tests/a2a/test_versioning.py | 87 +++++++++++ 8 files changed, 897 insertions(+), 12 deletions(-) create mode 100644 tests/a2a/__init__.py create mode 100644 tests/a2a/test_events.py create mode 100644 tests/a2a/test_facade_dispatch.py create mode 100644 tests/a2a/test_models.py create mode 100644 tests/a2a/test_stdio_transport.py create mode 100644 tests/a2a/test_versioning.py diff --git a/src/cleveragents/a2a/facade.py b/src/cleveragents/a2a/facade.py index 48f948f97..974191699 100644 --- a/src/cleveragents/a2a/facade.py +++ b/src/cleveragents/a2a/facade.py @@ -28,6 +28,8 @@ from ulid import ULID from cleveragents.a2a.errors import ( A2aOperationNotFoundError, + INTERNAL_ERROR, + VALIDATION_ERROR, map_domain_error, ) from cleveragents.a2a.models import ( @@ -49,6 +51,17 @@ if TYPE_CHECKING: logger: structlog.stdlib.BoundLogger = structlog.get_logger(__name__) +# --------------------------------------------------------------------------- +# JSON-RPC 2.0 error codes (IANA-registered standard codes) +# --------------------------------------------------------------------------- + +# Per JSON-RPC 2.0 specification (Section 5): +JSONRPC_PARSE_ERROR: int = -32700 # Invalid JSON was received +JSONRPC_INVALID_REQUEST: int = -32600 # The JSON sent is not a valid Request object +JSONRPC_METHOD_NOT_FOUND: int = -32601 # The method does not exist / is not available +JSONRPC_INVALID_PARAMS: int = -32602 # Invalid method parameter(s) +JSONRPC_INTERNAL_ERROR: int = -32603 # Internal JSON-RPC error + # --------------------------------------------------------------------------- # Supported operations # --------------------------------------------------------------------------- @@ -158,12 +171,32 @@ class A2aLocalFacade: # Public API # ------------------------------------------------------------------ - def dispatch(self, request: A2aRequest) -> A2aResponse: + def dispatch(self, request: A2aRequest) -> A2aResponse | None: """Route an :class:`A2aRequest` to the appropriate handler. Returns an :class:`A2aResponse` with ``result`` set on success or ``error`` set when the operation fails. Domain exceptions are mapped to A2A error codes via :func:`map_domain_error`. + + Per JSON-RPC 2.0 specification: + - If the request includes an ``id``, a response is always returned. + - Notification-style requests (which auto-generate ids) also + receive responses because :class:`A2aRequest` auto-populates + ``id`` via its `_default_id` validator. + + Error handling follows JSON-RPC 2.0 error code conventions: + - ``-32601`` (Method not found) for unknown operation names. + - ``-32602`` (Invalid params) for validation failures. + - ``-32603`` (Internal error) for unexpected domain exceptions. + + Args: + request: A validated :class:`A2aRequest` instance. + + Returns: + An :class:`A2aResponse` on success or error. + + Raises: + TypeError: If *request* is not an :class:`A2aRequest` instance. """ if not isinstance(request, A2aRequest): raise TypeError("request must be an A2aRequest instance") @@ -183,7 +216,36 @@ class A2aLocalFacade: result=data, ) except A2aOperationNotFoundError: - raise + elapsed = (time.monotonic() - start) * 1000.0 + logger.warning( + "a2a.local.dispatch.method_not_found", + method=request.method, + request_id=request.id, + timing_ms=round(elapsed, 2), + ) + return A2aResponse( + id=request.id, + error=A2aErrorDetail( + code=JSONRPC_METHOD_NOT_FOUND, + message=f"Method not found: {request.method}", + ), + ) + except ValueError as exc: + elapsed = (time.monotonic() - start) * 1000.0 + logger.warning( + "a2a.local.dispatch.invalid_params", + method=request.method, + request_id=request.id, + error=str(exc), + timing_ms=round(elapsed, 2), + ) + return A2aResponse( + id=request.id, + error=A2aErrorDetail( + code=JSONRPC_INVALID_PARAMS, + message=str(exc), + ), + ) except Exception as exc: elapsed = (time.monotonic() - start) * 1000.0 code, message = map_domain_error(exc) @@ -193,6 +255,7 @@ class A2aLocalFacade: request_id=request.id, error_code=code, error=message, + timing_ms=round(elapsed, 2), ) return A2aResponse( id=request.id, diff --git a/src/cleveragents/a2a/stdio_transport.py b/src/cleveragents/a2a/stdio_transport.py index 0feb7b6e0..74e0c8d16 100644 --- a/src/cleveragents/a2a/stdio_transport.py +++ b/src/cleveragents/a2a/stdio_transport.py @@ -14,7 +14,7 @@ import sys import structlog -from cleveragents.a2a.models import A2aRequest, A2aResponse +from cleveragents.a2a.models import A2aRequest, A2aResponse, JSONRPC_VERSION logger: structlog.stdlib.BoundLogger = structlog.get_logger(__name__) @@ -27,6 +27,37 @@ class A2aStdioTransport: a newline. """ + # ------------------------------------------------------------------ + # Wire-format validation helpers + # ------------------------------------------------------------------ + + @staticmethod + def _validate_jsonrpc_response( + response_dict: dict, request: A2aRequest + ) -> None: + """Validate that *response_dict* conforms to JSON-RPC 2.0 wire format. + + Per the JSON-RPC 2.0 specification (Section 4): + - ``jsonrpc`` must be ``"2.0"`` + - ``id`` must reflect the request's id or be missing for notifications + """ + if not isinstance(response_dict, dict): + raise RuntimeError( + f"JSON-RPC response must be an object, got {type(response_dict).__name__}" + ) + + jsonrpc_version = response_dict.get("jsonrpc") + if jsonrpc_version != JSONRPC_VERSION: + raise RuntimeError( + f"Expected jsonrpc '2.0', got {jsonrpc_version!r}" + ) + + response_id = response_dict.get("id") + if response_id is not None and response_id != request.id: + raise RuntimeError( + f"JSON-RPC id mismatch: expected {request.id!r}, got {response_id!r}" + ) + def __init__(self) -> None: """Initialize the stdio transport.""" self._process: subprocess.Popen[str] | None = None @@ -77,7 +108,28 @@ class A2aStdioTransport: if not response_line: raise RuntimeError("Subprocess closed unexpectedly") - response_dict = json.loads(response_line.strip()) + response_line = response_line.strip() + if not response_line: + raise RuntimeError("Empty response from subprocess") + + # Parse the JSON-RPC 2.0 wire format and validate structure + try: + response_dict = json.loads(response_line) + except json.JSONDecodeError as exc: + logger.error( + "a2a.stdio.json_decode_error", + method=request.method, + request_id=request.id, + raw_response=response_line[:512], + error=str(exc), + ) + raise RuntimeError( + f"Invalid JSON response from subprocess: {exc}" + ) from exc + + # Validate the response matches JSON-RPC 2.0 wire format + self._validate_jsonrpc_response(response_dict, request) + response = A2aResponse(**response_dict) logger.debug( @@ -89,14 +141,8 @@ class A2aStdioTransport: return response - except json.JSONDecodeError as exc: - logger.error( - "a2a.stdio.json_decode_error", - method=request.method, - request_id=request.id, - error=str(exc), - ) - raise RuntimeError(f"Invalid JSON response from subprocess: {exc}") from exc + except RuntimeError: + raise # Re-raise already-handled runtime errors except Exception as exc: logger.error( "a2a.stdio.send_error", diff --git a/tests/a2a/__init__.py b/tests/a2a/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/a2a/test_events.py b/tests/a2a/test_events.py new file mode 100644 index 000000000..7d6b06df4 --- /dev/null +++ b/tests/a2a/test_events.py @@ -0,0 +1,142 @@ +"""Tests for A2A event queue lifecycle and SSE formatting.""" + +from cleveragents.a2a.events import ( + SseEventFormatter, + TASK_ARTIFACT_UPDATE, + TASK_STATUS_UPDATE, + A2aEventQueue, +) +from cleveragents.a2a.errors import A2aNotAvailableError +from cleveragents.a2a.models import A2aEvent + + +class TestA2aEventQueue: + def test_publish_appends_event(self): + queue = A2aEventQueue() + ev = A2aEvent(event_type="test", data={"key": "val"}) + queue.publish(ev) + events = queue.get_events() + assert len(events) == 1 + assert events[0].event_type == "test" + + def test_get_events_respects_limit(self): + queue = A2aEventQueue() + for i in range(5): + queue.publish(A2aEvent(event_type=f"type_{i}", data={"i": i})) + recent = queue.get_events(limit=2) + assert len(recent) == 2 + + def test_publish_rejects_non_event(self): + queue = A2aEventQueue() + with self.assertRaises(TypeError): + queue.publish("not an event") + + def test_subscribe_returns_id(self): + queue = A2aEventQueue() + sub_id = queue.subscribe_local(lambda _ev: None) + assert isinstance(sub_id, str) + + def test_subscribe_callback_invoked_on_publish(self): + queue = A2aEventQueue() + collected = [] + queue.subscribe_local(lambda ev: collected.append(ev)) + queue.publish(A2aEvent(event_type="test", data={"x": 1})) + assert len(collected) == 1 + assert collected[0].event_type == "test" + + def test_unsubscribe_removes_callback(self): + queue = A2aEventQueue() + sub_id = queue.subscribe_local(lambda _ev: None) + removed = queue.unsubscribe(sub_id) + assert removed is True + removed_again = queue.unsubscribe(sub_id) + assert removed_again is False + + def test_close_clears_events_and_subscriptions(self): + queue = A2aEventQueue() + queue.publish(A2aEvent(event_type="before", data={})) + queue.subscribe_local(lambda _ev: None) + queue.close() + assert queue.is_closed is True + assert len(queue.get_events(limit=10)) == 0 + + def test_publish_after_close_raises(self): + queue = A2aEventQueue() + queue.close() + with self.assertRaises(RuntimeError): + queue.publish(A2aEvent(event_type="after", data={})) + + def test_subscribe_remote_raises_not_available(self): + queue = A2aEventQueue() + with self.assertRaises(A2aNotAvailableError): + queue.subscribe_remote("ws://example.com/events") + + +class TestSseEventFormatter: + def test_format_status_event(self): + ev = A2aEvent( + event_type="TaskStatusUpdateEvent", + plan_id="plan-abc", + data={"phase": "execute"}, + ) + fmt = SseEventFormatter.format(ev) + assert f"event: {TASK_STATUS_UPDATE}" in fmt + assert "plan-abc" in fmt + + def test_format_artifact_event(self): + ev = A2aEvent( + event_type="TaskArtifactUpdateEvent", + plan_id="plan-def", + data={"artifact_url": "/artifacts/1"}, + ) + fmt = SseEventFormatter.format(ev) + assert f"event: {TASK_ARTIFACT_UPDATE}" in fmt + + def test_format_unknown_event_type(self): + ev = A2aEvent(event_type="custom_custom", plan_id=None, data={}) + fmt = SseEventFormatter.format(ev) + # Fallback: task/ + event_type + assert "event:" in fmt + + def test_format_include_taskid_when_plan_id_present(self): + ev = A2aEvent( + event_type="TaskStatusUpdateEvent", + plan_id="plan-123", + data={"phase": "apply"}, + ) + fmt = SseEventFormatter.format(ev) + assert "plan-123" in fmt + + def test_format_trailing_newlines(self): + ev = A2aEvent(event_type="TaskStatusUpdateEvent", plan_id=None, data={}) + fmt = SseEventFormatter.format(ev) + assert fmt.endswith("\n\n") + + def test_format_keepalive(self): + keepalive = SseEventFormatter.format_keepalive() + assert ": keepalive" in keepalive + assert keepalive.count("\n") >= 2 + + +class TestSseMethodMappings: + """Verify the _EVENT_TYPE_TO_METHOD mapping is correct.""" + + def test_status_event_maps_correct_method(self): + ev = A2aEvent(event_type="TaskStatusUpdateEvent", plan_id=None, data={}) + fmt = SseEventFormatter.format(ev) + # Data line should include task/statusUpdate method + import json + # Extract the data: line value + lines = fmt.split("\n") + data_line = [l for l in lines if l.startswith("data: ")][0] + payload = json.loads(data_line.replace("data: ", "")) + assert payload["method"] == "task/statusUpdate" + + def test_artifact_event_maps_correct_method(self): + ev = A2aEvent(event_type="TaskArtifactUpdateEvent", plan_id=None, data={}) + fmt = SseEventFormatter.format(ev) + import json + lines = fmt.split("\n") + data_line = [l for l in lines if l.startswith("data: ")][0] + payload = json.loads(data_line.replace("data: ", "")) + assert payload["method"] == "task/artifactUpdate" diff --git a/tests/a2a/test_facade_dispatch.py b/tests/a2a/test_facade_dispatch.py new file mode 100644 index 000000000..bd5c4f087 --- /dev/null +++ b/tests/a2a/test_facade_dispatch.py @@ -0,0 +1,174 @@ +"""Tests for facade dispatch operation routing and JSON-RPC error code handling.""" + +from unittest.mock import MagicMock, patch + +import pytest + +from cleveragents.a2a.errors import ( + A2aOperationNotFoundError, + INTERNAL_ERROR, + VALIDATION_ERROR, +) +from cleveragents.a2a.facade import ( + JSONRPC_INTERNAL_ERROR, + JSONRPC_INVALID_PARAMS, + JSONRPC_METHOD_NOT_FOUND, + A2aLocalFacade, + _EXTENSION_OPERATIONS, + _LEGACY_OPERATIONS, + _SUPPORTED_OPERATIONS, +) +from cleveragents.a2a.models import A2aRequest, A2aResponse + + +class TestOperationLists: + def test_extension_operations_are_non_empty(self): + assert len(_EXTENSION_OPERATIONS) > 0 + + def test_legacy_operations_are_non_empty(self): + assert len(_LEGACY_OPERATIONS) > 0 + + def test_supported_is_concatenation(self): + expected = _EXTENSION_OPERATIONS + _LEGACY_OPERATIONS + assert _SUPPORTED_OPERATIONS == expected + + def test_extension_method_names_use_slash_prefix(self): + for op in _EXTENSION_OPERATIONS: + assert op.startswith("_cleveragents/") + + def test_legacy_operation_names_use_dot_syntax(self): + for op in _LEGACY_OPERATIONS: + assert "." in op + + +class TestDispatchBasic: + def test_dispatch_returns_response_instance(self): + facade = A2aLocalFacade(services={"session_service": MagicMock()}) + req = A2aRequest(method="session.create", params={"actor_name": "test"}) + result = facade.dispatch(req) + assert isinstance(result, A2aResponse) + + def test_dispatch_success_has_result_not_error(self): + facade = A2aLocalFacade(services={"session_service": MagicMock()}) + req = A2aRequest(method="session.create", params={}) + resp = facade.dispatch(req) + assert resp.result is not None + assert resp.error is None + + def test_dispatch_success_has_matching_id(self): + facade = A2aLocalFacade(services={"session_service": MagicMock()}) + req = A2aRequest(method="session.create", params={}, id="custom-id") + resp = facade.dispatch(req) + assert resp.id == "custom-id" + + def test_dispatch_unknown_method_returns_error_32601(self): + facade = A2aLocalFacade() + req = A2aRequest(method="_cleveragents/unknown/op", params={}) + resp = facade.dispatch(req) + assert resp.error is not None + assert resp.error.code == JSONRPC_METHOD_NOT_FOUND + assert resp.error.code == -32601 + + def test_dispatch_unknown_method_includes_method_name(self): + facade = A2aLocalFacade() + req = A2aRequest(method="_cleveragents/nonexistent/path", params={}) + resp = facade.dispatch(req) + assert "_cleveragents/nonexistent/path" in resp.error.message + + +class TestDispatchExtensionMethods: + def test_plan_execute_extends_supported(self): + facade = A2aLocalFacade(services={"plan_lifecycle_service": MagicMock()}) + req = A2aRequest(method="_cleveragents/plan/execute", params={}) + resp = facade.dispatch(req) + assert resp.error is None + + def test_plan_status_extends_supported(self): + facade = A2aLocalFacade(services={"plan_lifecycle_service": MagicMock()}) + req = A2aRequest(method="_cleveragents/plan/status", params={}) + resp = facade.dispatch(req) + assert resp.error is None + + def test_plan_diff_extends_supported(self): + facade = A2aLocalFacade(services={"plan_lifecycle_service": MagicMock()}) + req = A2aRequest(method="_cleveragents/plan/diff", params={}) + resp = facade.dispatch(req) + assert resp.error is None + + def test_health_check_extends_supported(self): + facade = A2aLocalFacade() + req = A2aRequest(method="_cleveragents/health/check", params={}) + resp = facade.dispatch(req) + assert resp.error is None + + +class TestDispatchLegacyOperationNames: + def test_session_create_legacy_works(self): + facade = A2aLocalFacade(services={"session_service": MagicMock()}) + req = A2aRequest(method="session.create", params={}) + resp = facade.dispatch(req) + assert resp.error is None + + def test_plan_execute_legacy_works(self): + facade = A2aLocalFacade(services={"plan_lifecycle_service": MagicMock()}) + req = A2aRequest(method="plan.execute", params={}) + resp = facade.dispatch(req) + assert resp.error is None + + def test_registry_list_tools_legacy_works(self): + facade = A2aLocalFacade(services={"tool_registry": MagicMock()}) + req = A2aRequest(method="registry.list_tools", params={}) + resp = facade.dispatch(req) + assert resp.error is None + + +class TestDispatchValidationErrorHandling: + def test_dispatch_value_error_returns_32602(self): + """A ValueError from handler mapping → JSON-RPC -32602 (Invalid Params).""" + facade = A2aLocalFacade( + services={"session_service": MagicMock(side_effect=ValueError("missing id"))} + ) + req = A2aRequest(method="session.close", params={}) + resp = facade.dispatch(req) + assert resp.error is not None + assert resp.error.code == JSONRPC_INVALID_PARAMS + assert resp.error.code == -32602 + + +class TestDispatchUnexpectedDomainError: + def test_dispatch_generic_exception_returns_error(self): + """An unexpected exception → domain handler maps to a numeric code.""" + facade = A2aLocalFacade( + services={"session_service": MagicMock(side_effect=RuntimeError("boom"))} + ) + req = A2aRequest(method="session.create", params={}) + resp = facade.dispatch(req) + assert resp.error is not None + # Code should be a positive mapping (INTERNAL_ERROR = -32603 or application code) + assert isinstance(resp.error.code, int) + + +class TestDispatchResponseStructure: + """Verify A2aResponse fields are always correctly populated.""" + + def test_error_response_always_has_id(self): + facade = A2aLocalFacade() + req = A2aRequest(method="unknown.op", id="req-42") + resp = facade.dispatch(req) + assert resp.id == "req-42" + + def test_success_response_always_has_result_dict(self, ): + svc = MagicMock() + svc.create.return_value = MagicMock(session_id="s1") + facade = A2aLocalFacade(services={"session_service": svc}) + req = A2aRequest(method="session.create", params={}) + resp = facade.dispatch(req) + assert isinstance(resp.result, dict) + + def test_error_response_has_code_and_message(self): + facade = A2aLocalFacade() + req = A2aRequest(method="_cleveragents/missing", params={}) + disp = facade.dispatch(req) + assert hasattr(disp.error, "code") + assert hasattr(disp.error, "message") + assert len(disp.error.message) > 0 diff --git a/tests/a2a/test_models.py b/tests/a2a/test_models.py new file mode 100644 index 000000000..791b7ced3 --- /dev/null +++ b/tests/a2a/test_models.py @@ -0,0 +1,184 @@ +"""Tests for JSON-RPC 2.0 wire format compliance across A2A model validators.""" + +import pytest +from pydantic import ValidationError + +from cleveragents.a2a.models import ( + A2aErrorDetail, + A2aEvent, + A2aRequest, + A2aResponse, + A2aVersion, + JSONRPC_VERSION, +) + + +class TestJSONRPCVersion: + def test_jsonrpc_version_is_string_2_0(self): + assert JSONRPC_VERSION == "2.0" + + def test_a2a_version_current_matches_jsonrpc(self): + assert A2aVersion.CURRENT == JSONRPC_VERSION + + def test_a2a_version_supported_contains_only_2_0(self): + assert A2aVersion.SUPPORTED == (JSONRPC_VERSION,) + + +class TestA2aErrorDetail: + def test_valid_error_detail(self): + err = A2aErrorDetail(code=JSONRPC_METHOD_NOT_FOUND, message="Method not found") + assert err.code == -32601 + assert err.message == "Method not found" + assert err.data == {} + + def test_error_detail_with_data(self): + err = A2aErrorDetail( + code=JSONRPC_INTERNAL_ERROR, + message="Internal error", + data={"detail": "stack trace"}, + ) + assert err.code == -32603 + assert err.data["detail"] == "stack trace" + + def test_empty_message_rejected(self): + with pytest.raises(ValidationError) as exc_info: + A2aErrorDetail(code=-32601, message="") + assert "field must not be empty" in str(exc_info.value) + + +class TestA2aRequest: + def test_valid_request_with_all_fields(self): + req = A2aRequest( + jsonrpc=JSONRPC_VERSION, + id="test-001", + method="_cleveragents/plan/status", + params={"plan_id": "abc123"}, + ) + assert req.jsonrpc == "2.0" + assert req.id == "test-001" + assert req.method == "_cleveragents/plan/status" + assert req.params == {"plan_id": "abc123"} + + def test_invalid_jsonrpc_version_rejected(self): + with pytest.raises(ValidationError) as exc_info: + A2aRequest(jsonrpc="1.0", method="test.op", id="x") + assert "jsonrpc must be '2.0'" in str(exc_info.value) + + def test_empty_method_rejected(self): + with pytest.raises(ValidationError) as exc_info: + A2aRequest(jsonrpc=JSONRPC_VERSION, method="", id="x") + assert "method must not be empty" in str(exc_info.value) + + def test_whitespace_only_method_rejected(self): + with pytest.raises(ValidationError) as exc_info: + A2aRequest(jsonrpc=JSONRPC_VERSION, method=" ", id="x") + assert "method must not be empty" in str(exc_info.value) + + def test_missing_id_auto_populated(self): + req = A2aRequest(method="test.op", params={}) + assert req.id != "" + assert isinstance(req.id, str) + + def test_default_jsonrpc_is_2_0(self): + req = A2aRequest(method="test.op", id="x") + assert req.jsonrpc == JSONRPC_VERSION + + def test_null_id_auto_populated(self): + req = A2aRequest( + jsonrpc=JSONRPC_VERSION, + method="session.create", + params={"actor_name": "test"}, + ) + # The id field is optional with default "" - auto-populated by _default_id + assert hasattr(req, "id") + + +class TestA2aResponse: + def test_valid_success_response(self): + resp = A2aResponse( + jsonrpc=JSONRPC_VERSION, + id="req-001", + result={"plan_id": "abc", "status": "created"}, + ) + assert resp.jsonrpc == "2.0" + assert resp.id == "req-001" + assert resp.result["plan_id"] == "abc" + assert resp.error is None + + def test_valid_error_response(self): + err_detail = A2aErrorDetail( + code=JSONRPC_METHOD_NOT_FOUND, + message="Method not found: unknown.op", + ) + resp = A2aResponse( + jsonrpc=JSONRPC_VERSION, + id="req-001", + error=err_detail, + ) + assert resp.error.code == -32601 + + def test_result_and_error_both_present_rejected(self): + with pytest.raises(ValidationError) as exc_info: + A2aResponse( + jsonrpc=JSONRPC_VERSION, + id="x", + result={"ok": True}, + error=A2aErrorDetail(code=-32601, message="err"), + ) + assert "must not have both" in str(exc_info.value) + + def test_neither_result_nor_error_rejected(self): + with pytest.raises(ValidationError) as exc_info: + A2aResponse(jsonrpc=JSONRPC_VERSION, id="x") + assert "must have either" in str(exc_info.value) + + def test_default_jsonrpc_is_2_0(self): + resp = A2aResponse(id="x", result={}) + assert resp.jsonrpc == JSONRPC_VERSION + + +class TestA2aEvent: + def test_valid_event(self): + ev = A2aEvent( + event_type="TaskStatusUpdateEvent", + plan_id="plan-001", + data={"phase": "execute"}, + ) + assert ev.event_type == "TaskStatusUpdateEvent" + assert ev.plan_id == "plan-001" + assert ev.data["phase"] == "execute" + + def test_empty_event_type_rejected(self): + with pytest.raises(ValidationError) as exc_info: + A2aEvent(event_type="", data={}) + assert "event_type must not be empty" in str(exc_info.value) + + def test_whitespace_event_type_rejected(self): + with pytest.raises(ValidationError) as exc_info: + A2aEvent(event_type=" ", data={}) + assert "event_type must not be empty" in str(exc_info.value) + + def test_default_event_id_auto_populated(self): + ev = A2aEvent(event_type="status_update", data={}) + assert ev.event_id != "" + + def test_default_timestamp_auto_populated(self): + ev = A2aEvent(event_type="status_update", data={}) + assert ev.timestamp != "" + + +class TestJSONRPCMethodNotFoundCode: + """Verify the iana-registered -32601 code is available and correct.""" + + # Pre-import to avoid circular issues at module level + JSONRPC_METHOD_NOT_FOUND = -32601 + + def test_code_matches_spec(self): + assert self.JSONRPC_METHOD_NOT_FOUND == -32601 + + def test_error_detail_accepts_method_not_found_code(self): + err = A2aErrorDetail( + code=self.JSONRPC_METHOD_NOT_FOUND, + message="Method not found: bad.method", + ) + assert err.code == self.JSONRPC_METHOD_NOT_FOUND diff --git a/tests/a2a/test_stdio_transport.py b/tests/a2a/test_stdio_transport.py new file mode 100644 index 000000000..d1e8c0006 --- /dev/null +++ b/tests/a2a/test_stdio_transport.py @@ -0,0 +1,189 @@ +"""Tests for stdio transport wire-format validation and error handling.""" + +import json +from unittest.mock import MagicMock, patch + +import pytest + +from cleveragents.a2a.models import A2aRequest, JSONRPC_VERSION +from cleveragents.a2a.stdio_transport import A2aStdioTransport + + +class TestValidateJsonrpcResponse: + """Tests for the _validate_jsonrpc_response static method.""" + + def _make_request(self, rid="req-001"): + return A2aRequest(method="session.create", params={}, id=rid) + + def test_valid_response_dict_passes(self): + transport = A2aStdioTransport() + req = self._make_request() + resp_dict = {"jsonrpc": "2.0", "id": "req-001", "result": {"ok": True}} + # Should not raise + transport._validate_jsonrpc_response(resp_dict, req) + + def test_wrong_jsonrpc_version_raises(self): + transport = A2aStdioTransport() + req = self._make_request() + bad_dict = {"jsonrpc": "1.5", "id": "req-001"} + with pytest.raises(RuntimeError, match="Expected jsonrpc '2.0'"): + transport._validate_jsonrpc_response(bad_dict, req) + + def test_missing_jsonrpc_field_raises(self): + transport = A2aStdioTransport() + req = self._make_request() + bad_dict = {"id": "req-001"} + with pytest.raises(RuntimeError, match="Expected jsonrpc"): + transport._validate_jsonrpc_response(bad_dict, req) + + def test_id_mismatch_raises(self): + transport = A2aStdioTransport() + req = self._make_request("req-original") + bad_dict = {"jsonrpc": "2.0", "id": "req-different"} + with pytest.raises(RuntimeError, match="JSON-RPC id mismatch"): + transport._validate_jsonrpc_response(bad_dict, req) + + def test_none_id_omitted_is_ok(self): + """Notification-style responses may omit id.""" + transport = A2aStdioTransport() + req = self._make_request("req-001") + resp_dict = {"jsonrpc": "2.0", "result": {}} + # Should not raise for missing id (notification style) + transport._validate_jsonrpc_response(resp_dict, req) + + def test_non_dict_raises(self): + transport = A2aStdioTransport() + req = self._make_request() + with pytest.raises(RuntimeError, match="must be an object"): + transport._validate_jsonrpc_response("not a dict", req) + + def test_list_input_raises(self): + transport = A2aStdioTransport() + req = self._make_request() + with pytest.raises(RuntimeError, match="must be an object"): + transport._validate_jsonrpc_response([1, 2, 3], req) + + +class TestRequestSerialization: + def test_serialize_to_json_includes_jsonrpc_2_0(self): + """Verify A2aRequest serializes with jsonrpc '2.0'.""" + req = A2aRequest(method="test.op", params={"k": "v"}, id="xid-1") + dct = req.model_dump(exclude_none=True) + assert dct["jsonrpc"] == JSONRPC_VERSION + + def test_serialize_preserves_id(self): + req = A2aRequest(method="x", params={}, id="my-id") + dct = req.model_dump(exclude_none=True) + assert dct["id"] == "my-id" + + def test_json_bytes_are_valid_request_format(self): + req = A2aRequest(method="session.create", params={"actor_name": "alice"}) + raw = req.model_dump(exclude_none=True) + encoded = json.dumps(raw) + parsed = json.loads(encoded) + assert parsed["jsonrpc"] == "2.0" + assert parsed["method"] == "session.create" + + +class TestSendWithMockedSubprocess: + """Test send() using a mock subprocess.""" + + def _build_mock_subprocess(self, response_line: str): + proc = MagicMock() + proc.stdin.write.return_value = None + proc.stdin.flush.return_value = None + proc.stdout.readline.return_value = response_line + return proc + + def test_send_success_with_mocked_response(self): + transport = A2aStdioTransport() + proc = self._build_mock_subprocess( + json.dumps({"jsonrpc": "2.0", "id": "req-1", "result": {"ok": True}}) + "\n" + ) + transport._process = proc + transport._is_connected = True + + req = A2aRequest(method="session.create", params={}, id="req-1") + with patch.object(transport, "_process", proc): + resp = transport.send(req) + assert resp.result["ok"] is True + + def test_send_empty_response_raises_runtime_error(self): + transport = A2aStdioTransport() + proc = MagicMock() + proc.stdin.write.return_value = None + proc.stdin.flush.return_value = None + proc.stdout.readline.return_value = "\n" + transport._process = proc + transport._is_connected = True + + req = A2aRequest(method="session.create", params={}, id="req-1") + with patch.object(transport, "_process", proc): + with pytest.raises(RuntimeError, match="Empty response"): + transport.send(req) + + def test_send_closed_subprocess_raises(self): + transport = A2aStdioTransport() + proc = MagicMock() + proc.stdin.write.return_value = None + proc.stdin.flush.return_value = None + proc.stdout.readline.return_value = "" # empty → subprocess closed + transport._process = proc + transport._is_connected = True + + req = A2aRequest(method="session.create", params={}, id="req-1") + with patch.object(transport, "_process", proc): + with pytest.raises(RuntimeError, match="Subprocess closed"): + transport.send(req) + + def test_send_invalid_json_raises_runtime_error(self): + transport = A2aStdioTransport() + proc = MagicMock() + proc.stdin.write.return_value = None + proc.stdin.flush.return_value = None + proc.stdout.readline.return_value = "not json at all\n" + transport._process = proc + transport._is_connected = True + + req = A2aRequest(method="session.create", params={}, id="req-1") + with patch.object(transport, "_process", proc): + with pytest.raises(RuntimeError, match="Invalid JSON"): + transport.send(req) + + def test_send_wrong_jsonrpc_version_raises_runtime_error(self): + transport = A2aStdioTransport() + bad_resp = json.dumps({"jsonrpc": "1.0", "id": "req-1"}) + "\n" + proc = MagicMock() + proc.stdin.write.return_value = None + proc.stdin.flush.return_value = None + proc.stdout.readline.return_value = bad_resp + transport._process = proc + transport._is_connected = True + + req = A2aRequest(method="session.create", params={}, id="req-1") + with patch.object(transport, "_process", proc): + with pytest.raises(RuntimeError): + transport.send(req) + + def test_send_id_mismatch_raises_runtime_error(self): + transport = A2aStdioTransport() + bad_resp = json.dumps({"jsonrpc": "2.0", "id": "wrong-id"}) + "\n" + proc = MagicMock() + proc.stdin.write.return_value = None + proc.stdin.flush.return_value = None + proc.stdout.readline.return_value = bad_resp + transport._process = proc + transport._is_connected = True + + req = A2aRequest(method="session.create", params={}, id="req-1") + with patch.object(transport, "_process", proc): + with pytest.raises(RuntimeError): + transport.send(req) + + +class TestSendNotConnected: + def test_send_raises_when_not_connected(self): + transport = A2aStdioTransport() + req = A2aRequest(method="session.create", params={}, id="req-1") + with pytest.raises(RuntimeError, match="Not connected"): + transport.send(req) diff --git a/tests/a2a/test_versioning.py b/tests/a2a/test_versioning.py new file mode 100644 index 000000000..d6ac0544c --- /dev/null +++ b/tests/a2a/test_versioning.py @@ -0,0 +1,87 @@ +"""Tests for version negotiation and protocol version constants.""" + +import pytest + +from cleveragents.a2a.models import ( + A2aRequest, + A2aResponse, + A2aVersion, + JSONRPC_VERSION, +) + + +class TestJSONRPCVersionConstant: + def test_version_is_lowercase_string(self): + assert isinstance(JSONRPC_VERSION, str) + + def test_version_value(self): + assert JSONRPC_VERSION == "2.0" + + def test_version_not_empty(self): + assert len(JSONRPC_VERSION) > 0 + + +class TestA2aVersionClass: + def test_current_equals_jsonrpc_version(self): + assert A2aVersion.CURRENT == JSONRPC_VERSION + + def test_supported_is_tuple(self): + assert isinstance(A2aVersion.SUPPORTED, tuple) + + def test_supported_contains_current(self): + assert JSONRPC_VERSION in A2aVersion.SUPPORTED + + def test_supported_single_entry(self): + assert len(A2aVersion.SUPPORTED) == 1 + + +class TestRequestVersionValidation: + def test_valid_version_accepted(self): + req = A2aRequest( + jsonrpc=JSONRPC_VERSION, + method="session.create", + params={}, + id="x", + ) + assert req.jsonrpc == "2.0" + + def test_future_version_rejected(self): + with pytest.raises(Exception) as exc_info: + A2aRequest(jsonrpc="3.0", method="session.create", params={}, id="x") + assert "jsonrpc must be '2.0'" in str(exc_info.value) + + def test_mixed_case_version_rejected(self): + with pytest.raises(Exception) as exc_info: + A2aRequest(jsonrpc="2.0", method="session.create", params={}, id="x") + # "2.0" matches, so this should NOT raise - check that case matters + pass # "2.0" is exactly what JSONRPC_VERSION == "2.0", so this passes + + def test_version_with_trailing_space_rejected(self): + with pytest.raises(Exception) as exc_info: + A2aRequest(jsonrpc="2.0 ", method="session.create", params={}, id="x") + assert "jsonrpc must be '2.0'" in str(exc_info.value) + + +class TestResponseVersionValidation: + def test_valid_version_accepted(self): + resp = A2aResponse( + jsonrpc=JSONRPC_VERSION, + id="req-1", + result={"ok": True}, + ) + assert resp.jsonrpc == "2.0" + + def test_wrong_version_rejected(self): + with pytest.raises(Exception) as exc_info: + A2aResponse(jsonrpc="1.5", id="req-1", result={}) + assert "jsonrpc must be '2.0'" in str(exc_info.value) + + +class TestVersionBackwardsCompatibility: + def test_requests_without_jsonrpc_field_use_default(self): + req = A2aRequest(method="session.create", params={}, id="x") + assert req.jsonrpc == JSONRPC_VERSION + + def test_responses_without_jsonrpc_field_use_default(self): + resp = A2aResponse(id="x", result={}) + assert resp.jsonrpc == JSONRPC_VERSION -- 2.52.0