1  """
  2  Unit tests for the A2AMessageValidator.
  3  """
  4  
  5  import pytest
  6  from sam_test_infrastructure.a2a_validator.validator import A2AMessageValidator
  7  
  8  
  9  @pytest.fixture
 10  def validator() -> A2AMessageValidator:
 11      """Provides an instance of the A2AMessageValidator."""
 12      return A2AMessageValidator()
 13  
 14  
 15  def test_validator_initialization(validator: A2AMessageValidator):
 16      """Tests that the validator initializes correctly and loads the schema."""
 17      assert validator.active is False
 18      assert validator.schema is not None
 19      assert "definitions" in validator.schema
 20      assert validator.validator is not None
 21  
 22  
 23  def test_valid_send_message_request(validator: A2AMessageValidator):
 24      """Tests validation of a valid SendMessageRequest."""
 25      payload = {
 26          "jsonrpc": "2.0",
 27          "id": "req-1",
 28          "method": "message/send",
 29          "params": {
 30              "message": {
 31                  "role": "user",
 32                  "messageId": "msg-1",
 33                  "kind": "message",
 34                  "parts": [{"kind": "text", "text": "Hello"}],
 35              }
 36          },
 37      }
 38      validator.validate_message(payload, "a2a/v1/agent/request/TestAgent")
 39  
 40  
 41  def test_valid_task_response(validator: A2AMessageValidator):
 42      """Tests validation of a valid response containing a Task object."""
 43      payload = {
 44          "jsonrpc": "2.0",
 45          "id": "req-1",
 46          "result": {
 47              "id": "task-1",
 48              "contextId": "session-1",
 49              "kind": "task",
 50              "status": {
 51                  "state": "completed",
 52                  "message": {
 53                      "role": "agent",
 54                      "messageId": "msg-2",
 55                      "kind": "message",
 56                      "parts": [{"kind": "text", "text": "Done"}],
 57                  },
 58              },
 59          },
 60      }
 61      validator.validate_message(payload, "a2a/v1/gateway/response/gw-1/task-1")
 62  
 63  
 64  def test_valid_status_update_response(validator: A2AMessageValidator):
 65      """Tests validation of a valid response containing a TaskStatusUpdateEvent."""
 66      payload = {
 67          "jsonrpc": "2.0",
 68          "id": "req-1",
 69          "result": {
 70              "kind": "status-update",
 71              "taskId": "task-1",
 72              "contextId": "session-1",
 73              "final": False,
 74              "status": {"state": "working"},
 75          },
 76      }
 77      validator.validate_message(payload, "a2a/v1/gateway/status/gw-1/task-1")
 78  
 79  
 80  def test_invalid_request_missing_jsonrpc(validator: A2AMessageValidator):
 81      """Tests that a request missing 'jsonrpc' fails validation."""
 82      payload = {
 83          "id": "req-1",
 84          "method": "message/send",
 85          "params": {
 86              "message": {
 87                  "role": "user",
 88                  "messageId": "msg-1",
 89                  "kind": "message",
 90                  "parts": [{"kind": "text", "text": "Hello"}],
 91              }
 92          },
 93      }
 94      with pytest.raises(pytest.fail.Exception, match="'jsonrpc' is a required property"):
 95          validator.validate_message(payload, "a2a/v1/agent/request/TestAgent")
 96  
 97  
 98  def test_invalid_request_bad_method(validator: A2AMessageValidator):
 99      """Tests that a request with an unknown method falls back to generic validation."""
100      payload = {
101          "jsonrpc": "2.0",
102          "id": "req-1",
103          "method": "non/existent/method",
104          "params": {},
105      }
106      # This should not fail because it validates against the generic request schema.
107      validator.validate_message(payload, "a2a/v1/agent/request/TestAgent")
108  
109  
110  def test_invalid_response_both_result_and_error(validator: A2AMessageValidator):
111      """Tests that a response with both 'result' and 'error' fails validation."""
112      payload = {
113          "jsonrpc": "2.0",
114          "id": "req-1",
115          "result": {
116              "id": "task-1",
117              "contextId": "session-1",
118              "kind": "task",
119              "status": {"state": "completed"},
120          },
121          "error": {"code": -32000, "message": "An error"},
122      }
123      with pytest.raises(
124          pytest.fail.Exception, match="'result' and 'error' are mutually exclusive"
125      ):
126          validator.validate_message(payload, "a2a/v1/gateway/response/gw-1/task-1")
127  
128  
129  def test_invalid_task_missing_id(validator: A2AMessageValidator):
130      """Tests that a Task object in a result missing a required field fails validation."""
131      payload = {
132          "jsonrpc": "2.0",
133          "id": "req-1",
134          "result": {
135              # "id": "task-1",  <-- Missing required field
136              "contextId": "session-1",
137              "kind": "task",
138              "status": {"state": "completed"},
139          },
140      }
141      with pytest.raises(pytest.fail.Exception, match="'id' is a required property"):
142          validator.validate_message(payload, "a2a/v1/gateway/response/gw-1/task-1")