fix(a2a): add input validation for optional parameters in facade handlers
- 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:
@@ -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")
|
||||
@@ -539,10 +539,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"}
|
||||
|
||||
@@ -610,6 +619,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],
|
||||
@@ -620,6 +634,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": [
|
||||
|
||||
Reference in New Issue
Block a user