fix(a2a): add input validation for optional parameters in facade handlers
CI / lint (pull_request) Failing after 28s
CI / typecheck (pull_request) Successful in 1m23s
CI / security (pull_request) Successful in 1m17s
CI / quality (pull_request) Successful in 37s
CI / coverage (pull_request) Has been skipped
CI / build (pull_request) Successful in 30s
CI / push-validation (pull_request) Successful in 26s
CI / helm (pull_request) Successful in 39s
CI / integration_tests (pull_request) Successful in 4m13s
CI / e2e_tests (pull_request) Successful in 4m7s
CI / unit_tests (pull_request) Successful in 7m26s
CI / docker (pull_request) Has been skipped
CI / status-check (pull_request) Failing after 1s

- Added input validation for namespace parameter in _handle_registry_list_tools in src/cleveragents/a2a/facade.py: If provided, must be a non-empty string; empty strings raise ValueError; non-string types raise TypeError
- Added input validation for type_name parameter in _handle_registry_list_resources in src/cleveragents/a2a/facade.py: If provided, must be a non-empty string; empty strings raise ValueError; non-string types raise TypeError
- Added input validation for arguments parameter in _handle_plan_create in src/cleveragents/a2a/facade.py: If provided, must be a dict; non-dict types raise TypeError
- Added input validation for created_by parameter in _handle_plan_create in src/cleveragents/a2a/facade.py: If provided, must be a non-empty string; empty strings raise ValueError; non-string types raise TypeError
- Added BDD feature file features/a2a_facade_optional_param_validation.feature with 17 scenarios covering all validation paths
- Added step definitions features/steps/a2a_facade_optional_param_validation_steps.py

ISSUES CLOSED: #9059
This commit is contained in:
2026-04-14 12:53:16 +00:00
parent 64b1f4c0b6
commit c1cf4160a5
3 changed files with 320 additions and 2 deletions
@@ -0,0 +1,122 @@
Feature: A2A facade optional parameter validation
As a developer using the A2A local facade
I want optional parameters to be validated before being passed to services
So that invalid inputs are rejected early with clear error messages
# -------------------------------------------------------------------
# _handle_registry_list_tools namespace validation
# -------------------------------------------------------------------
Scenario: list_tools with valid namespace succeeds
Given an opt-val facade with a mock ToolRegistry
When I dispatch opt-val operation "registry.list_tools" with params {"namespace": "local"}
Then the opt-val response status should be "ok"
And the opt-val response data should contain key "tools"
Scenario: list_tools with None namespace succeeds
Given an opt-val facade with a mock ToolRegistry
When I dispatch opt-val operation "registry.list_tools" with params {}
Then the opt-val response status should be "ok"
And the opt-val response data should contain key "tools"
Scenario: list_tools with empty string namespace returns error
Given an opt-val facade with a mock ToolRegistry
When I dispatch opt-val operation "registry.list_tools" with params {"namespace": ""}
Then the opt-val response status should be "error"
And the opt-val response error message should contain "namespace must be a non-empty string"
Scenario: list_tools with non-string namespace returns error
Given an opt-val facade with a mock ToolRegistry
When I dispatch opt-val operation "registry.list_tools" with params {"namespace": 42}
Then the opt-val response status should be "error"
And the opt-val response error message should contain "namespace must be a string or None"
Scenario: list_tools with no registry returns empty tools list
Given an opt-val facade with no services
When I dispatch opt-val operation "registry.list_tools" with params {"namespace": "local"}
Then the opt-val response status should be "ok"
And the opt-val response data should contain key "tools"
# -------------------------------------------------------------------
# _handle_registry_list_resources — type_name validation
# -------------------------------------------------------------------
Scenario: list_resources with valid type_name succeeds
Given an opt-val facade with a mock ResourceRegistryService
When I dispatch opt-val operation "registry.list_resources" with params {"type_name": "git-checkout"}
Then the opt-val response status should be "ok"
And the opt-val response data should contain key "resources"
Scenario: list_resources with None type_name succeeds
Given an opt-val facade with a mock ResourceRegistryService
When I dispatch opt-val operation "registry.list_resources" with params {}
Then the opt-val response status should be "ok"
And the opt-val response data should contain key "resources"
Scenario: list_resources with empty string type_name returns error
Given an opt-val facade with a mock ResourceRegistryService
When I dispatch opt-val operation "registry.list_resources" with params {"type_name": ""}
Then the opt-val response status should be "error"
And the opt-val response error message should contain "type_name must be a non-empty string"
Scenario: list_resources with non-string type_name returns error
Given an opt-val facade with a mock ResourceRegistryService
When I dispatch opt-val operation "registry.list_resources" with params {"type_name": 99}
Then the opt-val response status should be "error"
And the opt-val response error message should contain "type_name must be a string or None"
Scenario: list_resources with no service returns empty resources list
Given an opt-val facade with no services
When I dispatch opt-val operation "registry.list_resources" with params {"type_name": "git-checkout"}
Then the opt-val response status should be "ok"
And the opt-val response data should contain key "resources"
# -------------------------------------------------------------------
# _handle_plan_create — arguments validation
# -------------------------------------------------------------------
Scenario: plan.create with valid dict arguments succeeds
Given an opt-val facade with a mock PlanLifecycleService
When I dispatch opt-val operation "plan.create" with params {"action_name": "build", "arguments": {"target": "all"}}
Then the opt-val response status should be "ok"
And the opt-val response data key "status" should equal "created"
Scenario: plan.create with None arguments succeeds
Given an opt-val facade with a mock PlanLifecycleService
When I dispatch opt-val operation "plan.create" with params {"action_name": "build"}
Then the opt-val response status should be "ok"
And the opt-val response data key "status" should equal "created"
Scenario: plan.create with non-dict arguments returns error
Given an opt-val facade with a mock PlanLifecycleService
When I dispatch opt-val operation "plan.create" with params {"action_name": "build", "arguments": "not-a-dict"}
Then the opt-val response status should be "error"
And the opt-val response error message should contain "arguments must be a dict or None"
# -------------------------------------------------------------------
# _handle_plan_create — created_by validation
# -------------------------------------------------------------------
Scenario: plan.create with valid created_by succeeds
Given an opt-val facade with a mock PlanLifecycleService
When I dispatch opt-val operation "plan.create" with params {"action_name": "build", "created_by": "alice"}
Then the opt-val response status should be "ok"
And the opt-val response data key "status" should equal "created"
Scenario: plan.create with None created_by succeeds
Given an opt-val facade with a mock PlanLifecycleService
When I dispatch opt-val operation "plan.create" with params {"action_name": "build"}
Then the opt-val response status should be "ok"
And the opt-val response data key "status" should equal "created"
Scenario: plan.create with empty string created_by returns error
Given an opt-val facade with a mock PlanLifecycleService
When I dispatch opt-val operation "plan.create" with params {"action_name": "build", "created_by": ""}
Then the opt-val response status should be "error"
And the opt-val response error message should contain "created_by must be a non-empty string"
Scenario: plan.create with non-string created_by returns error
Given an opt-val facade with a mock PlanLifecycleService
When I dispatch opt-val operation "plan.create" with params {"action_name": "build", "created_by": 123}
Then the opt-val response status should be "error"
And the opt-val response error message should contain "created_by must be a string or None"
@@ -0,0 +1,177 @@
"""Step definitions for a2a_facade_optional_param_validation.feature.
Covers input validation for optional parameters in A2A facade handlers:
| Handler | Parameter | Validation |
|--------------------------------|-------------|-------------------------------------|
| _handle_registry_list_tools | namespace | non-empty string or None |
| _handle_registry_list_resources| type_name | non-empty string or None |
| _handle_plan_create | arguments | dict or None |
| _handle_plan_create | created_by | non-empty string or None |
"""
from __future__ import annotations
import json
from typing import Any
from unittest.mock import MagicMock
from behave import given, then, use_step_matcher, when
from behave.runner import Context
try:
from cleveragents.a2a.facade import A2aLocalFacade
from cleveragents.a2a.models import A2aRequest
except ImportError:
pass # a2a module not available
use_step_matcher("re")
# ---------------------------------------------------------------------------
# Mock helpers
# ---------------------------------------------------------------------------
class _MockPlanIdentity:
"""Minimal PlanIdentity stub."""
def __init__(self, plan_id: str = "MOCK-PLAN-001") -> None:
self.plan_id = plan_id
class _MockPlan:
"""Minimal plan stub."""
def __init__(self, plan_id: str = "MOCK-PLAN-001") -> None:
self.identity = _MockPlanIdentity(plan_id)
self.phase = MagicMock()
self.phase.value = "strategize"
self.state = MagicMock()
self.state.value = "queued"
class _MockToolSpec:
"""Minimal ToolSpec stub."""
def __init__(self, name: str, description: str) -> None:
self.name = name
self.description = description
class _MockResource:
"""Minimal resource stub."""
def __init__(self, resource_id: str, name: str, resource_type_name: str) -> None:
self.resource_id = resource_id
self.name = name
self.resource_type_name = resource_type_name
def _build_mock_plan_lifecycle_service() -> MagicMock:
svc = MagicMock()
svc.use_action.return_value = _MockPlan()
return svc
def _build_mock_tool_registry() -> MagicMock:
registry = MagicMock()
registry.list_tools.return_value = [
_MockToolSpec("local/tool-a", "Tool A"),
]
return registry
def _build_mock_resource_registry_service() -> MagicMock:
svc = MagicMock()
svc.list_resources.return_value = [
_MockResource("RES-001", "my-repo", "git-checkout"),
]
return svc
# ---------------------------------------------------------------------------
# Given steps
# ---------------------------------------------------------------------------
@given(r"an opt-val facade with no services")
def step_ov_facade_no_services(context: Context) -> None:
context.ov_facade = A2aLocalFacade()
@given(r"an opt-val facade with a mock ToolRegistry")
def step_ov_facade_tool_registry(context: Context) -> None:
context.ov_facade = A2aLocalFacade(
services={"tool_registry": _build_mock_tool_registry()}
)
@given(r"an opt-val facade with a mock ResourceRegistryService")
def step_ov_facade_resource_registry(context: Context) -> None:
context.ov_facade = A2aLocalFacade(
services={"resource_registry_service": _build_mock_resource_registry_service()}
)
@given(r"an opt-val facade with a mock PlanLifecycleService")
def step_ov_facade_plan_lifecycle(context: Context) -> None:
context.ov_facade = A2aLocalFacade(
services={"plan_lifecycle_service": _build_mock_plan_lifecycle_service()}
)
# ---------------------------------------------------------------------------
# When steps
# ---------------------------------------------------------------------------
@when(
r'I dispatch opt-val operation "(?P<operation>[^"]+)" '
r"with params (?P<params_json>.+)"
)
def step_ov_dispatch(context: Context, operation: str, params_json: str) -> None:
params: dict[str, Any] = json.loads(params_json)
request = A2aRequest(method=operation, params=params)
context.ov_response = context.ov_facade.dispatch(request)
# ---------------------------------------------------------------------------
# Then steps
# ---------------------------------------------------------------------------
@then(r'the opt-val response status should be "(?P<status>[^"]+)"')
def step_ov_response_status(context: Context, status: str) -> None:
assert (context.ov_response.error is None) == (status == "ok"), (
f"Expected status '{status}', got error={context.ov_response.error}"
)
@then(r'the opt-val response data should contain key "(?P<key>[^"]+)"')
def step_ov_data_has_key(context: Context, key: str) -> None:
assert key in context.ov_response.result, (
f"Expected key '{key}' in response data, "
f"got: {list(context.ov_response.result.keys())}"
)
@then(
r'the opt-val response data key "(?P<key>[^"]+)" should equal "(?P<value>[^"]+)"'
)
def step_ov_data_key_equals(context: Context, key: str, value: str) -> None:
actual = context.ov_response.result.get(key)
assert str(actual) == value, f"Expected data['{key}'] = '{value}', got '{actual}'"
@then(r'the opt-val response error message should contain "(?P<fragment>[^"]+)"')
def step_ov_error_message_contains(context: Context, fragment: str) -> None:
assert context.ov_response.error is not None, "No error in response"
assert fragment in context.ov_response.error.message, (
f"Expected message containing '{fragment}', "
f"got '{context.ov_response.error.message}'"
)
# Reset step matcher to parse (default) so subsequent step files are not affected
use_step_matcher("parse")
+21 -2
View File
@@ -376,10 +376,19 @@ class A2aLocalFacade:
action_name = params.get("action_name", "")
if not action_name:
raise ValueError("action_name is required")
arguments = params.get("arguments")
if arguments is not None and not isinstance(arguments, dict):
raise TypeError("arguments must be a dict or None")
created_by = params.get("created_by")
if created_by is not None:
if not isinstance(created_by, str):
raise TypeError("created_by must be a string or None")
if not created_by:
raise ValueError("created_by must be a non-empty string if provided")
plan = svc.use_action(
action_name=action_name,
arguments=params.get("arguments"),
created_by=params.get("created_by"),
arguments=arguments,
created_by=created_by,
)
return {"plan_id": plan.identity.plan_id, "status": "created"}
@@ -447,6 +456,11 @@ class A2aLocalFacade:
if registry is None:
return {"tools": []}
namespace: str | None = params.get("namespace")
if namespace is not None:
if not isinstance(namespace, str):
raise TypeError("namespace must be a string or None")
if not namespace:
raise ValueError("namespace must be a non-empty string if provided")
specs = registry.list_tools(namespace=namespace)
return {
"tools": [{"name": s.name, "description": s.description} for s in specs],
@@ -457,6 +471,11 @@ class A2aLocalFacade:
if svc is None:
return {"resources": []}
type_name: str | None = params.get("type_name")
if type_name is not None:
if not isinstance(type_name, str):
raise TypeError("type_name must be a string or None")
if not type_name:
raise ValueError("type_name must be a non-empty string if provided")
resources = svc.list_resources(type_name=type_name)
return {
"resources": [