From 64d10675f97a84b32789d21b6a1a29f370f5bc0c Mon Sep 17 00:00:00 2001 From: wrenj Date: Thu, 12 Feb 2026 13:00:27 -0500 Subject: [PATCH 1/6] initial validation draft --- .../extension/send_a2ui_to_client_toolset.py | 3 +- .../src/a2ui/extension/validation.py | 295 +++++++++++ .../a2ui_agent/tests/test_validation.py | 461 ++++++++++++++++++ 3 files changed, 758 insertions(+), 1 deletion(-) create mode 100644 a2a_agents/python/a2ui_agent/src/a2ui/extension/validation.py create mode 100644 a2a_agents/python/a2ui_agent/tests/test_validation.py diff --git a/a2a_agents/python/a2ui_agent/src/a2ui/extension/send_a2ui_to_client_toolset.py b/a2a_agents/python/a2ui_agent/src/a2ui/extension/send_a2ui_to_client_toolset.py index 1101869cf..baba339f4 100644 --- a/a2a_agents/python/a2ui_agent/src/a2ui/extension/send_a2ui_to_client_toolset.py +++ b/a2a_agents/python/a2ui_agent/src/a2ui/extension/send_a2ui_to_client_toolset.py @@ -87,6 +87,7 @@ async def get_schema(ctx: ReadonlyContext) -> dict[str, Any]: from a2a import types as a2a_types from a2ui.extension.a2ui_extension import create_a2ui_part from a2ui.extension.a2ui_schema_utils import wrap_as_json_array +from a2ui.extension.validation import validate_a2ui_json from google.adk.a2a.converters import part_converter from google.adk.agents.readonly_context import ReadonlyContext from google.adk.models import LlmRequest @@ -262,7 +263,7 @@ async def run_async( a2ui_json_payload = [a2ui_json_payload] a2ui_schema = await self.get_a2ui_schema(tool_context) - jsonschema.validate(instance=a2ui_json_payload, schema=a2ui_schema) + validate_a2ui_json(a2ui_json_payload, a2ui_schema) logger.info( f"Validated call to tool {self.TOOL_NAME} with {self.A2UI_JSON_ARG_NAME}" diff --git a/a2a_agents/python/a2ui_agent/src/a2ui/extension/validation.py b/a2a_agents/python/a2ui_agent/src/a2ui/extension/validation.py new file mode 100644 index 000000000..8e088fa71 --- /dev/null +++ b/a2a_agents/python/a2ui_agent/src/a2ui/extension/validation.py @@ -0,0 +1,295 @@ +from typing import Any, Dict, List, Set, Union +import jsonschema + +def validate_a2ui_json(a2ui_json: Union[Dict[str, Any], List[Any]], a2ui_schema: Dict[str, Any]) -> None: + """ + Validates the A2UI JSON payload against the provided schema and checks for integrity. + + Args: + a2ui_json: The JSON payload to validate. + a2ui_schema: The schema to validate against. + + Raises: + jsonschema.ValidationError: If the payload does not match the schema. + ValueError: If integrity or topology checks fail. + """ + jsonschema.validate(instance=a2ui_json, schema=a2ui_schema) + + # Normalize to list for iteration + messages = a2ui_json if isinstance(a2ui_json, list) else [a2ui_json] + + for message in messages: + if not isinstance(message, dict): + continue + + # Check for SurfaceUpdate which has 'components' + if "components" in message: + _validate_component_integrity(message["components"], a2ui_schema) + _validate_topology(message["components"], a2ui_schema) + + _validate_recursion_and_paths(message) + + +def _validate_component_integrity(components: List[Dict[str, Any]], a2ui_schema: Dict[str, Any] = None) -> None: + """ + Validates that: + 1. All component IDs are unique. + 2. A 'root' component exists. + 3. All references (children, child, etc.) point to existing IDs. + """ + ids: Set[str] = set() + + # 1. Collect IDs and check for duplicates + for comp in components: + comp_id = comp.get("id") + if comp_id is None: + continue + + if comp_id in ids: + raise ValueError(f"Duplicate component ID found: '{comp_id}'") + ids.add(comp_id) + + # 2. Check for root component + if "root" not in ids: + raise ValueError("Missing 'root' component: One component must have 'id' set to 'root'.") + + # 3. Check for dangling references using schema-driven extraction + ref_fields_map = _extract_component_ref_fields(a2ui_schema) if a2ui_schema else {} + + for comp in components: + comp_props_container = comp.get("componentProperties") + if not isinstance(comp_props_container, dict): + continue + + for comp_type, props in comp_props_container.items(): + if not isinstance(props, dict): + continue + + # Determine fields to check for this component type + # Strictly use schema; if not found, assume no references (generic schema support) + single_refs, list_refs = ref_fields_map.get(comp_type, (set(), set())) + + for key, value in props.items(): + if key in single_refs: + if isinstance(value, str) and value not in ids: + raise ValueError(f"Component '{comp.get('id')}' references missing ID '{value}' in field '{key}'") + elif key in list_refs: + if isinstance(value, list): + for item in value: + if isinstance(item, str) and item not in ids: + raise ValueError(f"Component '{comp.get('id')}' references missing ID '{item}' in field '{key}'") + + +def _validate_topology(components: List[Dict[str, Any]], a2ui_schema: Dict[str, Any] = None) -> None: + """ + Validates the topology of the component tree: + 1. No circular references (including self-references). + 2. No orphaned components (all components must be reachable from 'root'). + """ + adj_list: Dict[str, List[str]] = {} + all_ids: Set[str] = set() + + ref_fields_map = _extract_component_ref_fields(a2ui_schema) if a2ui_schema else {} + + # Build Adjacency List + for comp in components: + comp_id = comp.get("id") + if comp_id is None: + continue + + all_ids.add(comp_id) + if comp_id not in adj_list: + adj_list[comp_id] = [] + + comp_props_container = comp.get("componentProperties") + if not isinstance(comp_props_container, dict): + continue + + for comp_type, props in comp_props_container.items(): + if not isinstance(props, dict): + continue + + # Determine fields to check + # Strictly use schema + single_refs, list_refs = ref_fields_map.get(comp_type, (set(), set())) + + for key, value in props.items(): + if key in single_refs: + if isinstance(value, str): + if value == comp_id: + raise ValueError(f"Self-reference detected: Component '{comp_id}' references itself in field '{key}'") + adj_list[comp_id].append(value) + elif key in list_refs: + if isinstance(value, list): + for item in value: + if isinstance(item, str): + if item == comp_id: + raise ValueError(f"Self-reference detected: Component '{comp_id}' references itself in field '{key}'") + adj_list[comp_id].append(item) + + # Detect Cycles using DFS + visited: Set[str] = set() + recursion_stack: Set[str] = set() + + def dfs(node_id: str): + visited.add(node_id) + recursion_stack.add(node_id) + + for neighbor in adj_list.get(node_id, []): + if neighbor not in visited: + dfs(neighbor) + elif neighbor in recursion_stack: + raise ValueError(f"Circular reference detected involving component '{neighbor}'") + + recursion_stack.remove(node_id) + + if "root" in all_ids: + dfs("root") + + # Check for Orphans + orphans = all_ids - visited + if orphans: + sorted_orphans = sorted(list(orphans)) + raise ValueError(f"Orphaned components detected (not reachable from 'root'): {sorted_orphans}") + + +def _extract_component_ref_fields(schema: Dict[str, Any]) -> Dict[str, tuple[Set[str], Set[str]]]: + """ + Parses the JSON schema to identify which component properties reference other components. + Returns a map: { component_name: (set_of_single_ref_fields, set_of_list_ref_fields) } + """ + # print(f"DEBUG: _extract_component_ref_fields called with schema keys: {schema.keys()}") + ref_map = {} + + # We expect schema structure to have 'properties' -> 'components' -> 'items' -> ... + # OR we might be passed the root schema. + # A typical A2UI schema has definitions in $defs or definitions. + + # 1. Locate component definitions + # In the testing schema (and likely real one), component properties are inside: + # properties -> components -> items -> properties -> componentProperties -> properties -> [ComponentType] + + try: + # Navigate to componentProperties definitions in the schema + # This path depends on the exact schema structure provided to validate_a2ui_json + # We'll try to walk down standard paths. + + root_defs = schema.get("$defs") or schema.get("definitions", {}) + + # Helper to check if a property schema looks like a ComponentId reference + def is_component_id_ref(prop_schema: Dict[str, Any]) -> bool: + ref = prop_schema.get("$ref", "") + if ref.endswith("ComponentId"): + return True + # Check if it's an expanded schema that refs ComponentId (unlikely for direct prop but possible) + return False + + def is_child_list_ref(prop_schema: Dict[str, Any]) -> bool: + ref = prop_schema.get("$ref", "") + if ref.endswith("ChildList"): + return True + # Or array of ComponentIds + if prop_schema.get("type") == "array": + items = prop_schema.get("items", {}) + if is_component_id_ref(items): + return True + return False + + # Find where components are defined. + # In the provided common_types.json / standard_catalog.json, components are usually in a specific location + # but the schema passed to validate_a2ui_json *is* the message schema which *includes* catalog definitions + # often via $defs or inline? + # Actually message schema has 'components' array items matching catalog entries. + + # Let's search for 'componentProperties' in the schema + # We can implement a search or targeted lookup + + # Target path: properties.components.items.properties.componentProperties.properties + # (This matches the structure in test_validation.py) + + comps_schema = schema.get("properties", {}).get("components", {}) + items_schema = comps_schema.get("items", {}) + comp_props_schema = items_schema.get("properties", {}).get("componentProperties", {}) + all_components = comp_props_schema.get("properties", {}) + + for comp_name, comp_schema in all_components.items(): + single_refs = set() + list_refs = set() + + props = comp_schema.get("properties", {}) + for prop_name, prop_schema in props.items(): + if is_component_id_ref(prop_schema): + single_refs.add(prop_name) + elif is_child_list_ref(prop_schema): + list_refs.add(prop_name) + + if single_refs or list_refs: + ref_map[comp_name] = (single_refs, list_refs) + except Exception: + # If schema traversal fails (structure mismatch), return empty to trigger fallback + pass + + return ref_map + + +def _validate_recursion_and_paths(data: Any) -> None: + """ + Validates: + 1. Global recursion depth limit (50). + 2. FunctionCall recursion depth limit (5). + 3. Path syntax for DataBindings/DataModelUpdates. + """ + def traverse(item: Any, global_depth: int, func_depth: int): + if global_depth > 50: + raise ValueError("Global recursion limit exceeded: Depth > 50") + + if isinstance(item, list): + for x in item: + traverse(x, global_depth + 1, func_depth) + elif isinstance(item, dict): + # Check for path + if "path" in item and isinstance(item["path"], str): + _validate_path_syntax(item["path"]) + + # Check for FunctionCall (heuristic: has 'call' and 'args') + is_func = "call" in item and "args" in item + + if is_func: + if func_depth >= 5: + raise ValueError("Recursion limit exceeded: FunctionCall depth > 5") + + # Increment func_depth only for this branch, but global_depth matches traversal + for k, v in item.items(): + if k == "args": + traverse(v, global_depth + 1, func_depth + 1) + else: + traverse(v, global_depth + 1, func_depth) + else: + for v in item.values(): + traverse(v, global_depth + 1, func_depth) + + traverse(data, 0, 0) + + +def _validate_path_syntax(path: str) -> None: + """ + Validates that the path is either a valid JSON Pointer (starts with /) + or a valid relative path (no empty segments). + Also checks for spaces. + """ + if not path: + return + + parts = path.split('/') + if path.startswith('/'): + # JSON Pointer - starts with / + # We allow empty keys (//) as per RFC 6901 but typical A2UI usage might not. + # For now, we only enforce that it starts with / if intended to be absolute. + pass + else: + # Relative path - should not have empty segments + if any(not p for p in parts): + raise ValueError(f"Invalid data model path: '{path}' contains empty segments (//)") + + if " " in path: + raise ValueError(f"Invalid data model path: '{path}' contains spaces") diff --git a/a2a_agents/python/a2ui_agent/tests/test_validation.py b/a2a_agents/python/a2ui_agent/tests/test_validation.py new file mode 100644 index 000000000..5f7e3ab16 --- /dev/null +++ b/a2a_agents/python/a2ui_agent/tests/test_validation.py @@ -0,0 +1,461 @@ + +import pytest +import jsonschema +from a2ui.extension.validation import validate_a2ui_json + +# Simple schema for testing +SCHEMA = { + "type": "object", + "$defs": { + "ComponentId": {"type": "string"}, + "ChildList": { + "type": "array", + "items": {"$ref": "#/$defs/ComponentId"} + } + }, + "properties": { + "components": { + "type": "array", + "items": { + "type": "object", + "properties": { + "id": {"$ref": "#/$defs/ComponentId"}, + "componentProperties": { + "type": "object", + "properties": { + "Column": { + "type": "object", + "properties": { + "children": {"$ref": "#/$defs/ChildList"} + } + }, + "Row": { + "type": "object", + "properties": { + "children": {"$ref": "#/$defs/ChildList"} + } + }, + "Container": { + "type": "object", + "properties": { + "children": {"$ref": "#/$defs/ChildList"} + } + }, + "Card": { + "type": "object", + "properties": { + "child": {"$ref": "#/$defs/ComponentId"} + } + }, + "Button": { + "type": "object", + "properties": { + "child": {"$ref": "#/$defs/ComponentId"}, + "action": { + # Minimal action schema for recursion tests + "properties": { + "functionCall": { + "properties": { + "call": {"type": "string"}, + "args": {"type": "object"} + } + } + } + } + } + }, + "Text": { + "type": "object", + "properties": { + "text": { + "oneOf": [ + {"type": "string"}, + {"type": "object"} + ] + } + } + } + } + } + }, + "required": ["id"] + } + } + } +} + +def test_validate_a2ui_json_valid_integrity(): + payload = { + "components": [ + { + "id": "root", + "componentProperties": { + "Column": { + "children": ["child1"] + } + } + }, + { + "id": "child1", + "componentProperties": { + "Text": { + "text": "Hello" + } + } + } + ] + } + # Should not raise + validate_a2ui_json(payload, SCHEMA) + +def test_validate_a2ui_json_duplicate_ids(): + payload = { + "components": [ + {"id": "root", "componentProperties": {}}, + {"id": "root", "componentProperties": {}} + ] + } + with pytest.raises(ValueError, match="Duplicate component ID found: 'root'"): + validate_a2ui_json(payload, SCHEMA) + +def test_validate_a2ui_json_missing_root(): + payload = { + "components": [ + {"id": "not-root", "componentProperties": {}} + ] + } + with pytest.raises(ValueError, match="Missing 'root' component"): + validate_a2ui_json(payload, SCHEMA) + +def test_validate_a2ui_json_dangling_reference_child(): + payload = { + "components": [ + { + "id": "root", + "componentProperties": { + "Card": { + "child": "missing_child" + } + } + } + ] + } + with pytest.raises(ValueError, match="Component 'root' references missing ID 'missing_child' in field 'child'"): + validate_a2ui_json(payload, SCHEMA) + +def test_validate_a2ui_json_dangling_reference_children(): + payload = { + "components": [ + { + "id": "root", + "componentProperties": { + "Column": { + "children": ["child1", "missing_child"] + } + } + }, + {"id": "child1", "componentProperties": {}} + ] + } + with pytest.raises(ValueError, match="Component 'root' references missing ID 'missing_child' in field 'children'"): + validate_a2ui_json(payload, SCHEMA) + + +def test_validate_a2ui_json_self_reference(): + payload = { + "components": [ + { + "id": "root", + "componentProperties": { + "Container": { + "children": ["root"] + } + } + } + ] + } + with pytest.raises(ValueError, match="Self-reference detected: Component 'root' references itself in field 'children'"): + validate_a2ui_json(payload, SCHEMA) + +def test_validate_a2ui_json_circular_reference(): + payload = { + "components": [ + { + "id": "root", + "componentProperties": { + "Container": { + "children": ["child1"] + } + } + }, + { + "id": "child1", + "componentProperties": { + "Container": { + "children": ["root"] + } + } + } + ] + } + with pytest.raises(ValueError, match="Circular reference detected involving component"): + # The exact message depends on DFS order, but it should contain "Circular reference detected involving component" + validate_a2ui_json(payload, SCHEMA) + +def test_validate_a2ui_json_orphaned_component(): + payload = { + "components": [ + { + "id": "root", + "componentProperties": { + "Container": { + "children": [] + } + } + }, + { + "id": "orphan", + "componentProperties": {} + } + ] + } + # We use regex match because the list order in set conversion might vary, though I sorted it in code. + # But to be safe on Python versions or whatever... + # I sorted it in code: sorted(list(orphans)) + with pytest.raises(ValueError, match=r"Orphaned components detected \(not reachable from 'root'\): \['orphan'\]"): + validate_a2ui_json(payload, SCHEMA) + +def test_validate_a2ui_json_valid_topology_complex(): + """Test a valid topology with multiple levels.""" + payload = { + "components": [ + { + "id": "root", + "componentProperties": { + "Container": { + "children": ["child1", "child2"] + } + } + }, + { + "id": "child1", + "componentProperties": { + "Text": {"text": "Hello"} + } + }, + { + "id": "child2", + "componentProperties": { + "Container": { + "children": ["child3"] + } + } + }, + { + "id": "child3", + "componentProperties": { + "Text": {"text": "World"} + } + } + ] + } + # Should not raise + validate_a2ui_json(payload, SCHEMA) + +def test_validate_recursion_limit_exceeded(): + """Test that recursion depth > 5 raises ValueError.""" + # Nesting level 6 + payload = { + "components": [ + { + "id": "root", + "componentProperties": { + "Button": { + "label": "Click me", + "action": { + "functionCall": { + "call": "fn1", + "args": { + "arg1": { + # Depth 2 + "call": "fn2", + "args": { + "arg2": { + # Depth 3 + "call": "fn3", + "args": { + "arg3": { + # Depth 4 + "call": "fn4", + "args": { + "arg4": { + # Depth 5 + "call": "fn5", + "args": { + "arg5": { + # Depth 6 - Should fail + "call": "fn6", + "args": {} + } + } + } + } + } + } + } + } + } + } + } + } + } + } + } + ] + } + with pytest.raises(ValueError, match="Recursion limit exceeded"): + validate_a2ui_json(payload, SCHEMA) + +def test_validate_recursion_limit_valid(): + """Test that recursion depth <= 5 is allowed.""" + # Nesting level 5 + payload = { + "components": [ + { + "id": "root", + "componentProperties": { + "Button": { + "label": "Click me", + "action": { + "functionCall": { + "call": "fn1", + "args": { + "arg1": { + "call": "fn2", + "args": { + "arg2": { + "call": "fn3", + "args": { + "arg3": { + "call": "fn4", + "args": { + "arg4": { + "call": "fn5", + "args": {} + } + } + } + } + } + } + } + } + } + } + } + } + } + ] + } + # Should not raise + validate_a2ui_json(payload, SCHEMA) + +def test_validate_invalid_datamodel_path_update(): + """Test invalid path in UpdateDataModelMessage.""" + payload = { + "updateDataModel": { + "surfaceId": "surface1", + "path": "invalid//path", + "value": "data" + } + } + with pytest.raises(ValueError, match="Invalid data model path"): + validate_a2ui_json(payload, SCHEMA) + +def test_validate_invalid_databinding_path(): + """Test invalid path in DataBinding.""" + payload = { + "components": [ + { + "id": "root", + "componentProperties": { + "Text": { + "text": { + "path": "invalid path with spaces" + } + } + } + } + ] + } + with pytest.raises(ValueError, match="Invalid data model path"): + validate_a2ui_json(payload, SCHEMA) + +def test_validate_global_recursion_limit_exceeded(): + """Test that global recursion depth > 50 raises ValueError.""" + # Create a deeply nested dictionary + deep_payload = {"level": 0} + current = deep_payload + for i in range(55): + current["next"] = {"level": i + 1} + current = current["next"] + + with pytest.raises(ValueError, match="Global recursion limit exceeded"): + validate_a2ui_json(deep_payload, SCHEMA) + + +def test_validate_custom_schema_reference(): + """Test validation with a custom schema where a component has a non-standard reference field.""" + # Custom schema extending the base one + custom_schema = { + "type": "object", + "$defs": { + "ComponentId": {"type": "string"}, + "ChildList": { + "type": "array", + "items": {"$ref": "#/$defs/ComponentId"} + } + }, + "properties": { + "components": { + "type": "array", + "items": { + "type": "object", + "properties": { + "id": {"$ref": "#/$defs/ComponentId"}, + "componentProperties": { + "type": "object", + "properties": { + "CustomLink": { + "type": "object", + "properties": { + # "linkedComponentId" should be picked up because it refs ComponentId + "linkedComponentId": {"$ref": "#/$defs/ComponentId"} + } + } + } + } + }, + "required": ["id"] + } + } + } + } + + payload = { + "components": [ + { + "id": "root", + "componentProperties": { + "CustomLink": { + "linkedComponentId": "missing_target" + } + } + } + ] + } + + # Validation should fail because "linkedComponentId" references "missing_target" + # and the logic should have extracted "linkedComponentId" as a reference field from the schema. + with pytest.raises(ValueError, match="Component 'root' references missing ID 'missing_target' in field 'linkedComponentId'"): + validate_a2ui_json(payload, custom_schema) From afca7f6fad1cac3611090446b57f63e966661465 Mon Sep 17 00:00:00 2001 From: wrenj Date: Thu, 12 Feb 2026 13:32:46 -0500 Subject: [PATCH 2/6] fix path validation --- .../src/a2ui/extension/validation.py | 165 +++++++----------- .../a2ui_agent/tests/test_validation.py | 29 ++- 2 files changed, 79 insertions(+), 115 deletions(-) diff --git a/a2a_agents/python/a2ui_agent/src/a2ui/extension/validation.py b/a2a_agents/python/a2ui_agent/src/a2ui/extension/validation.py index 8e088fa71..ef71d6d46 100644 --- a/a2a_agents/python/a2ui_agent/src/a2ui/extension/validation.py +++ b/a2a_agents/python/a2ui_agent/src/a2ui/extension/validation.py @@ -1,17 +1,36 @@ -from typing import Any, Dict, List, Set, Union +from typing import Any, Dict, Iterator, List, Set, Tuple, Union import jsonschema +import re + +# RFC 6901 compliant regex for JSON Pointer +JSON_POINTER_PATTERN = re.compile(r"^(?:\/(?:[^~\/]|~[01])*)*$") def validate_a2ui_json(a2ui_json: Union[Dict[str, Any], List[Any]], a2ui_schema: Dict[str, Any]) -> None: """ Validates the A2UI JSON payload against the provided schema and checks for integrity. + Checks performed: + 1. **JSON Schema Validation**: Ensures payload adheres to the A2UI schema. + 2. **Component Integrity**: + - All component IDs are unique. + - A 'root' component exists. + - All unique component references point to valid IDs. + 3. **Topology**: + - No circular references (including self-references). + - No orphaned components (all components must be reachable from 'root'). + 4. **Recursion Limits**: + - Global recursion depth limit (50). + - FunctionCall recursion depth limit (5). + 5. **Path Syntax**: + - Validates JSON Pointer syntax for data paths. + Args: a2ui_json: The JSON payload to validate. a2ui_schema: The schema to validate against. Raises: jsonschema.ValidationError: If the payload does not match the schema. - ValueError: If integrity or topology checks fail. + ValueError: If integrity, topology, or recursion checks fail. """ jsonschema.validate(instance=a2ui_json, schema=a2ui_schema) @@ -24,13 +43,14 @@ def validate_a2ui_json(a2ui_json: Union[Dict[str, Any], List[Any]], a2ui_schema: # Check for SurfaceUpdate which has 'components' if "components" in message: - _validate_component_integrity(message["components"], a2ui_schema) - _validate_topology(message["components"], a2ui_schema) + ref_map = _extract_component_ref_fields(a2ui_schema) + _validate_component_integrity(message["components"], ref_map) + _validate_topology(message["components"], ref_map) _validate_recursion_and_paths(message) -def _validate_component_integrity(components: List[Dict[str, Any]], a2ui_schema: Dict[str, Any] = None) -> None: +def _validate_component_integrity(components: List[Dict[str, Any]], ref_fields_map: Dict[str, tuple[Set[str], Set[str]]]) -> None: """ Validates that: 1. All component IDs are unique. @@ -53,34 +73,14 @@ def _validate_component_integrity(components: List[Dict[str, Any]], a2ui_schema: if "root" not in ids: raise ValueError("Missing 'root' component: One component must have 'id' set to 'root'.") - # 3. Check for dangling references using schema-driven extraction - ref_fields_map = _extract_component_ref_fields(a2ui_schema) if a2ui_schema else {} - + # 3. Check for dangling references using helper for comp in components: - comp_props_container = comp.get("componentProperties") - if not isinstance(comp_props_container, dict): - continue - - for comp_type, props in comp_props_container.items(): - if not isinstance(props, dict): - continue + for ref_id, field_name in _get_component_references(comp, ref_fields_map): + if ref_id not in ids: + raise ValueError(f"Component '{comp.get('id')}' references missing ID '{ref_id}' in field '{field_name}'") - # Determine fields to check for this component type - # Strictly use schema; if not found, assume no references (generic schema support) - single_refs, list_refs = ref_fields_map.get(comp_type, (set(), set())) - - for key, value in props.items(): - if key in single_refs: - if isinstance(value, str) and value not in ids: - raise ValueError(f"Component '{comp.get('id')}' references missing ID '{value}' in field '{key}'") - elif key in list_refs: - if isinstance(value, list): - for item in value: - if isinstance(item, str) and item not in ids: - raise ValueError(f"Component '{comp.get('id')}' references missing ID '{item}' in field '{key}'") - - -def _validate_topology(components: List[Dict[str, Any]], a2ui_schema: Dict[str, Any] = None) -> None: + +def _validate_topology(components: List[Dict[str, Any]], ref_fields_map: Dict[str, tuple[Set[str], Set[str]]]) -> None: """ Validates the topology of the component tree: 1. No circular references (including self-references). @@ -89,8 +89,6 @@ def _validate_topology(components: List[Dict[str, Any]], a2ui_schema: Dict[str, adj_list: Dict[str, List[str]] = {} all_ids: Set[str] = set() - ref_fields_map = _extract_component_ref_fields(a2ui_schema) if a2ui_schema else {} - # Build Adjacency List for comp in components: comp_id = comp.get("id") @@ -101,31 +99,10 @@ def _validate_topology(components: List[Dict[str, Any]], a2ui_schema: Dict[str, if comp_id not in adj_list: adj_list[comp_id] = [] - comp_props_container = comp.get("componentProperties") - if not isinstance(comp_props_container, dict): - continue - - for comp_type, props in comp_props_container.items(): - if not isinstance(props, dict): - continue - - # Determine fields to check - # Strictly use schema - single_refs, list_refs = ref_fields_map.get(comp_type, (set(), set())) - - for key, value in props.items(): - if key in single_refs: - if isinstance(value, str): - if value == comp_id: - raise ValueError(f"Self-reference detected: Component '{comp_id}' references itself in field '{key}'") - adj_list[comp_id].append(value) - elif key in list_refs: - if isinstance(value, list): - for item in value: - if isinstance(item, str): - if item == comp_id: - raise ValueError(f"Self-reference detected: Component '{comp_id}' references itself in field '{key}'") - adj_list[comp_id].append(item) + for ref_id, field_name in _get_component_references(comp, ref_fields_map): + if ref_id == comp_id: + raise ValueError(f"Self-reference detected: Component '{comp_id}' references itself in field '{field_name}'") + adj_list[comp_id].append(ref_id) # Detect Cycles using DFS visited: Set[str] = set() @@ -158,22 +135,9 @@ def _extract_component_ref_fields(schema: Dict[str, Any]) -> Dict[str, tuple[Set Parses the JSON schema to identify which component properties reference other components. Returns a map: { component_name: (set_of_single_ref_fields, set_of_list_ref_fields) } """ - # print(f"DEBUG: _extract_component_ref_fields called with schema keys: {schema.keys()}") ref_map = {} - # We expect schema structure to have 'properties' -> 'components' -> 'items' -> ... - # OR we might be passed the root schema. - # A typical A2UI schema has definitions in $defs or definitions. - - # 1. Locate component definitions - # In the testing schema (and likely real one), component properties are inside: - # properties -> components -> items -> properties -> componentProperties -> properties -> [ComponentType] - try: - # Navigate to componentProperties definitions in the schema - # This path depends on the exact schema structure provided to validate_a2ui_json - # We'll try to walk down standard paths. - root_defs = schema.get("$defs") or schema.get("definitions", {}) # Helper to check if a property schema looks like a ComponentId reference @@ -181,7 +145,6 @@ def is_component_id_ref(prop_schema: Dict[str, Any]) -> bool: ref = prop_schema.get("$ref", "") if ref.endswith("ComponentId"): return True - # Check if it's an expanded schema that refs ComponentId (unlikely for direct prop but possible) return False def is_child_list_ref(prop_schema: Dict[str, Any]) -> bool: @@ -195,18 +158,6 @@ def is_child_list_ref(prop_schema: Dict[str, Any]) -> bool: return True return False - # Find where components are defined. - # In the provided common_types.json / standard_catalog.json, components are usually in a specific location - # but the schema passed to validate_a2ui_json *is* the message schema which *includes* catalog definitions - # often via $defs or inline? - # Actually message schema has 'components' array items matching catalog entries. - - # Let's search for 'componentProperties' in the schema - # We can implement a search or targeted lookup - - # Target path: properties.components.items.properties.componentProperties.properties - # (This matches the structure in test_validation.py) - comps_schema = schema.get("properties", {}).get("components", {}) items_schema = comps_schema.get("items", {}) comp_props_schema = items_schema.get("properties", {}).get("componentProperties", {}) @@ -226,12 +177,38 @@ def is_child_list_ref(prop_schema: Dict[str, Any]) -> bool: if single_refs or list_refs: ref_map[comp_name] = (single_refs, list_refs) except Exception: - # If schema traversal fails (structure mismatch), return empty to trigger fallback + # If schema traversal fails, return empty map pass return ref_map +def _get_component_references(component: Dict[str, Any], ref_fields_map: Dict[str, tuple[Set[str], Set[str]]]) -> Iterator[Tuple[str, str]]: + """ + Helper to extract all referenced component IDs from a component. + Yields (referenced_id, field_name). + """ + comp_props_container = component.get("componentProperties") + if not isinstance(comp_props_container, dict): + return + + for comp_type, props in comp_props_container.items(): + if not isinstance(props, dict): + continue + + single_refs, list_refs = ref_fields_map.get(comp_type, (set(), set())) + + for key, value in props.items(): + if key in single_refs: + if isinstance(value, str): + yield value, key + elif key in list_refs: + if isinstance(value, list): + for item in value: + if isinstance(item, str): + yield item, key + + def _validate_recursion_and_paths(data: Any) -> None: """ Validates: @@ -277,19 +254,7 @@ def _validate_path_syntax(path: str) -> None: or a valid relative path (no empty segments). Also checks for spaces. """ - if not path: - return - - parts = path.split('/') - if path.startswith('/'): - # JSON Pointer - starts with / - # We allow empty keys (//) as per RFC 6901 but typical A2UI usage might not. - # For now, we only enforce that it starts with / if intended to be absolute. - pass - else: - # Relative path - should not have empty segments - if any(not p for p in parts): - raise ValueError(f"Invalid data model path: '{path}' contains empty segments (//)") + if not re.fullmatch(JSON_POINTER_PATTERN, path): + raise ValueError(f"Invalid JSON Pointer syntax: '{path}'") - if " " in path: - raise ValueError(f"Invalid data model path: '{path}' contains spaces") + \ No newline at end of file diff --git a/a2a_agents/python/a2ui_agent/tests/test_validation.py b/a2a_agents/python/a2ui_agent/tests/test_validation.py index 5f7e3ab16..8adb6c34c 100644 --- a/a2a_agents/python/a2ui_agent/tests/test_validation.py +++ b/a2a_agents/python/a2ui_agent/tests/test_validation.py @@ -52,7 +52,6 @@ "properties": { "child": {"$ref": "#/$defs/ComponentId"}, "action": { - # Minimal action schema for recursion tests "properties": { "functionCall": { "properties": { @@ -105,7 +104,6 @@ def test_validate_a2ui_json_valid_integrity(): } ] } - # Should not raise validate_a2ui_json(payload, SCHEMA) def test_validate_a2ui_json_duplicate_ids(): @@ -199,7 +197,6 @@ def test_validate_a2ui_json_circular_reference(): ] } with pytest.raises(ValueError, match="Circular reference detected involving component"): - # The exact message depends on DFS order, but it should contain "Circular reference detected involving component" validate_a2ui_json(payload, SCHEMA) def test_validate_a2ui_json_orphaned_component(): @@ -219,9 +216,6 @@ def test_validate_a2ui_json_orphaned_component(): } ] } - # We use regex match because the list order in set conversion might vary, though I sorted it in code. - # But to be safe on Python versions or whatever... - # I sorted it in code: sorted(list(orphans)) with pytest.raises(ValueError, match=r"Orphaned components detected \(not reachable from 'root'\): \['orphan'\]"): validate_a2ui_json(payload, SCHEMA) @@ -259,12 +253,10 @@ def test_validate_a2ui_json_valid_topology_complex(): } ] } - # Should not raise validate_a2ui_json(payload, SCHEMA) def test_validate_recursion_limit_exceeded(): """Test that recursion depth > 5 raises ValueError.""" - # Nesting level 6 payload = { "components": [ { @@ -293,7 +285,6 @@ def test_validate_recursion_limit_exceeded(): "call": "fn5", "args": { "arg5": { - # Depth 6 - Should fail "call": "fn6", "args": {} } @@ -357,7 +348,6 @@ def test_validate_recursion_limit_valid(): } ] } - # Should not raise validate_a2ui_json(payload, SCHEMA) def test_validate_invalid_datamodel_path_update(): @@ -369,7 +359,7 @@ def test_validate_invalid_datamodel_path_update(): "value": "data" } } - with pytest.raises(ValueError, match="Invalid data model path"): + with pytest.raises(ValueError, match="Invalid JSON Pointer syntax"): validate_a2ui_json(payload, SCHEMA) def test_validate_invalid_databinding_path(): @@ -388,7 +378,7 @@ def test_validate_invalid_databinding_path(): } ] } - with pytest.raises(ValueError, match="Invalid data model path"): + with pytest.raises(ValueError, match="Invalid JSON Pointer syntax"): validate_a2ui_json(payload, SCHEMA) def test_validate_global_recursion_limit_exceeded(): @@ -429,7 +419,6 @@ def test_validate_custom_schema_reference(): "CustomLink": { "type": "object", "properties": { - # "linkedComponentId" should be picked up because it refs ComponentId "linkedComponentId": {"$ref": "#/$defs/ComponentId"} } } @@ -455,7 +444,17 @@ def test_validate_custom_schema_reference(): ] } - # Validation should fail because "linkedComponentId" references "missing_target" - # and the logic should have extracted "linkedComponentId" as a reference field from the schema. with pytest.raises(ValueError, match="Component 'root' references missing ID 'missing_target' in field 'linkedComponentId'"): validate_a2ui_json(payload, custom_schema) + +def test_validate_invalid_json_pointer_escape(): + """Test invalid escape sequence in JSON Pointer.""" + payload = { + "updateDataModel": { + "surfaceId": "surface1", + "path": "/invalid/escape/~2", + "value": "data" + } + } + with pytest.raises(ValueError, match="Invalid JSON Pointer syntax"): + validate_a2ui_json(payload, SCHEMA) From 45dcf620bf5f42eedda9a1c00a7b1dda312f1d40 Mon Sep 17 00:00:00 2001 From: wrenj Date: Thu, 12 Feb 2026 13:35:25 -0500 Subject: [PATCH 3/6] refactor --- .../src/a2ui/extension/validation.py | 18 ++++-------------- 1 file changed, 4 insertions(+), 14 deletions(-) diff --git a/a2a_agents/python/a2ui_agent/src/a2ui/extension/validation.py b/a2a_agents/python/a2ui_agent/src/a2ui/extension/validation.py index ef71d6d46..8f2e0b4d0 100644 --- a/a2a_agents/python/a2ui_agent/src/a2ui/extension/validation.py +++ b/a2a_agents/python/a2ui_agent/src/a2ui/extension/validation.py @@ -226,7 +226,9 @@ def traverse(item: Any, global_depth: int, func_depth: int): elif isinstance(item, dict): # Check for path if "path" in item and isinstance(item["path"], str): - _validate_path_syntax(item["path"]) + path = item["path"] + if not re.fullmatch(JSON_POINTER_PATTERN, path): + raise ValueError(f"Invalid JSON Pointer syntax: '{path}'") # Check for FunctionCall (heuristic: has 'call' and 'args') is_func = "call" in item and "args" in item @@ -245,16 +247,4 @@ def traverse(item: Any, global_depth: int, func_depth: int): for v in item.values(): traverse(v, global_depth + 1, func_depth) - traverse(data, 0, 0) - - -def _validate_path_syntax(path: str) -> None: - """ - Validates that the path is either a valid JSON Pointer (starts with /) - or a valid relative path (no empty segments). - Also checks for spaces. - """ - if not re.fullmatch(JSON_POINTER_PATTERN, path): - raise ValueError(f"Invalid JSON Pointer syntax: '{path}'") - - \ No newline at end of file + traverse(data, 0, 0) \ No newline at end of file From 78efe010b18db257a90f050bc48891708d4df64a Mon Sep 17 00:00:00 2001 From: wrenj Date: Thu, 12 Feb 2026 13:42:27 -0500 Subject: [PATCH 4/6] refactor --- .../src/a2ui/extension/validation.py | 56 +-- .../a2ui_agent/tests/test_validation.py | 328 ++++++++---------- 2 files changed, 174 insertions(+), 210 deletions(-) diff --git a/a2a_agents/python/a2ui_agent/src/a2ui/extension/validation.py b/a2a_agents/python/a2ui_agent/src/a2ui/extension/validation.py index 8f2e0b4d0..0291eaedc 100644 --- a/a2a_agents/python/a2ui_agent/src/a2ui/extension/validation.py +++ b/a2a_agents/python/a2ui_agent/src/a2ui/extension/validation.py @@ -5,6 +5,16 @@ # RFC 6901 compliant regex for JSON Pointer JSON_POINTER_PATTERN = re.compile(r"^(?:\/(?:[^~\/]|~[01])*)*$") +# Constants +COMPONENTS = "components" +ID = "id" +COMPONENT_PROPERTIES = "componentProperties" +ROOT = "root" +PATH = "path" +FUNCTION_CALL = "functionCall" +CALL = "call" +ARGS = "args" + def validate_a2ui_json(a2ui_json: Union[Dict[str, Any], List[Any]], a2ui_schema: Dict[str, Any]) -> None: """ Validates the A2UI JSON payload against the provided schema and checks for integrity. @@ -42,10 +52,10 @@ def validate_a2ui_json(a2ui_json: Union[Dict[str, Any], List[Any]], a2ui_schema: continue # Check for SurfaceUpdate which has 'components' - if "components" in message: + if COMPONENTS in message: ref_map = _extract_component_ref_fields(a2ui_schema) - _validate_component_integrity(message["components"], ref_map) - _validate_topology(message["components"], ref_map) + _validate_component_integrity(message[COMPONENTS], ref_map) + _validate_topology(message[COMPONENTS], ref_map) _validate_recursion_and_paths(message) @@ -61,7 +71,7 @@ def _validate_component_integrity(components: List[Dict[str, Any]], ref_fields_m # 1. Collect IDs and check for duplicates for comp in components: - comp_id = comp.get("id") + comp_id = comp.get(ID) if comp_id is None: continue @@ -70,14 +80,14 @@ def _validate_component_integrity(components: List[Dict[str, Any]], ref_fields_m ids.add(comp_id) # 2. Check for root component - if "root" not in ids: - raise ValueError("Missing 'root' component: One component must have 'id' set to 'root'.") + if ROOT not in ids: + raise ValueError(f"Missing '{ROOT}' component: One component must have '{ID}' set to '{ROOT}'.") # 3. Check for dangling references using helper for comp in components: for ref_id, field_name in _get_component_references(comp, ref_fields_map): if ref_id not in ids: - raise ValueError(f"Component '{comp.get('id')}' references missing ID '{ref_id}' in field '{field_name}'") + raise ValueError(f"Component '{comp.get(ID)}' references missing ID '{ref_id}' in field '{field_name}'") def _validate_topology(components: List[Dict[str, Any]], ref_fields_map: Dict[str, tuple[Set[str], Set[str]]]) -> None: @@ -91,7 +101,7 @@ def _validate_topology(components: List[Dict[str, Any]], ref_fields_map: Dict[st # Build Adjacency List for comp in components: - comp_id = comp.get("id") + comp_id = comp.get(ID) if comp_id is None: continue @@ -120,14 +130,14 @@ def dfs(node_id: str): recursion_stack.remove(node_id) - if "root" in all_ids: - dfs("root") + if ROOT in all_ids: + dfs(ROOT) # Check for Orphans orphans = all_ids - visited if orphans: sorted_orphans = sorted(list(orphans)) - raise ValueError(f"Orphaned components detected (not reachable from 'root'): {sorted_orphans}") + raise ValueError(f"Orphaned components detected (not reachable from '{ROOT}'): {sorted_orphans}") def _extract_component_ref_fields(schema: Dict[str, Any]) -> Dict[str, tuple[Set[str], Set[str]]]: @@ -158,9 +168,9 @@ def is_child_list_ref(prop_schema: Dict[str, Any]) -> bool: return True return False - comps_schema = schema.get("properties", {}).get("components", {}) + comps_schema = schema.get("properties", {}).get(COMPONENTS, {}) items_schema = comps_schema.get("items", {}) - comp_props_schema = items_schema.get("properties", {}).get("componentProperties", {}) + comp_props_schema = items_schema.get("properties", {}).get(COMPONENT_PROPERTIES, {}) all_components = comp_props_schema.get("properties", {}) for comp_name, comp_schema in all_components.items(): @@ -188,7 +198,7 @@ def _get_component_references(component: Dict[str, Any], ref_fields_map: Dict[st Helper to extract all referenced component IDs from a component. Yields (referenced_id, field_name). """ - comp_props_container = component.get("componentProperties") + comp_props_container = component.get(COMPONENT_PROPERTIES) if not isinstance(comp_props_container, dict): return @@ -223,23 +233,25 @@ def traverse(item: Any, global_depth: int, func_depth: int): if isinstance(item, list): for x in item: traverse(x, global_depth + 1, func_depth) - elif isinstance(item, dict): + return + + if isinstance(item, dict): # Check for path - if "path" in item and isinstance(item["path"], str): - path = item["path"] + if PATH in item and isinstance(item[PATH], str): + path = item[PATH] if not re.fullmatch(JSON_POINTER_PATTERN, path): raise ValueError(f"Invalid JSON Pointer syntax: '{path}'") - # Check for FunctionCall (heuristic: has 'call' and 'args') - is_func = "call" in item and "args" in item + # Check for FunctionCall + is_func = CALL in item and ARGS in item if is_func: if func_depth >= 5: - raise ValueError("Recursion limit exceeded: FunctionCall depth > 5") + raise ValueError(f"Recursion limit exceeded: {FUNCTION_CALL} depth > 5") - # Increment func_depth only for this branch, but global_depth matches traversal + # Increment func_depth only for 'args', but global_depth matches traversal for k, v in item.items(): - if k == "args": + if k == ARGS: traverse(v, global_depth + 1, func_depth + 1) else: traverse(v, global_depth + 1, func_depth) diff --git a/a2a_agents/python/a2ui_agent/tests/test_validation.py b/a2a_agents/python/a2ui_agent/tests/test_validation.py index 8adb6c34c..11483528c 100644 --- a/a2a_agents/python/a2ui_agent/tests/test_validation.py +++ b/a2a_agents/python/a2ui_agent/tests/test_validation.py @@ -3,87 +3,89 @@ import jsonschema from a2ui.extension.validation import validate_a2ui_json -# Simple schema for testing -SCHEMA = { - "type": "object", - "$defs": { - "ComponentId": {"type": "string"}, - "ChildList": { - "type": "array", - "items": {"$ref": "#/$defs/ComponentId"} - } - }, - "properties": { - "components": { - "type": "array", - "items": { - "type": "object", - "properties": { - "id": {"$ref": "#/$defs/ComponentId"}, - "componentProperties": { - "type": "object", - "properties": { - "Column": { - "type": "object", - "properties": { - "children": {"$ref": "#/$defs/ChildList"} - } - }, - "Row": { - "type": "object", - "properties": { - "children": {"$ref": "#/$defs/ChildList"} - } - }, - "Container": { - "type": "object", - "properties": { - "children": {"$ref": "#/$defs/ChildList"} - } - }, - "Card": { - "type": "object", - "properties": { - "child": {"$ref": "#/$defs/ComponentId"} - } - }, - "Button": { - "type": "object", - "properties": { - "child": {"$ref": "#/$defs/ComponentId"}, - "action": { - "properties": { - "functionCall": { - "properties": { - "call": {"type": "string"}, - "args": {"type": "object"} +# Fixture for the schema +@pytest.fixture +def schema(): + return { + "type": "object", + "$defs": { + "ComponentId": {"type": "string"}, + "ChildList": { + "type": "array", + "items": {"$ref": "#/$defs/ComponentId"} + } + }, + "properties": { + "components": { + "type": "array", + "items": { + "type": "object", + "properties": { + "id": {"$ref": "#/$defs/ComponentId"}, + "componentProperties": { + "type": "object", + "properties": { + "Column": { + "type": "object", + "properties": { + "children": {"$ref": "#/$defs/ChildList"} + } + }, + "Row": { + "type": "object", + "properties": { + "children": {"$ref": "#/$defs/ChildList"} + } + }, + "Container": { + "type": "object", + "properties": { + "children": {"$ref": "#/$defs/ChildList"} + } + }, + "Card": { + "type": "object", + "properties": { + "child": {"$ref": "#/$defs/ComponentId"} + } + }, + "Button": { + "type": "object", + "properties": { + "child": {"$ref": "#/$defs/ComponentId"}, + "action": { + "properties": { + "functionCall": { + "properties": { + "call": {"type": "string"}, + "args": {"type": "object"} + } } } + } + } + }, + "Text": { + "type": "object", + "properties": { + "text": { + "oneOf": [ + {"type": "string"}, + {"type": "object"} + ] } - } - } - }, - "Text": { - "type": "object", - "properties": { - "text": { - "oneOf": [ - {"type": "string"}, - {"type": "object"} - ] } } } } - } - }, - "required": ["id"] + }, + "required": ["id"] + } } } } -} -def test_validate_a2ui_json_valid_integrity(): +def test_validate_a2ui_json_valid_integrity(schema): payload = { "components": [ { @@ -104,9 +106,9 @@ def test_validate_a2ui_json_valid_integrity(): } ] } - validate_a2ui_json(payload, SCHEMA) + validate_a2ui_json(payload, schema) -def test_validate_a2ui_json_duplicate_ids(): +def test_validate_a2ui_json_duplicate_ids(schema): payload = { "components": [ {"id": "root", "componentProperties": {}}, @@ -114,52 +116,46 @@ def test_validate_a2ui_json_duplicate_ids(): ] } with pytest.raises(ValueError, match="Duplicate component ID found: 'root'"): - validate_a2ui_json(payload, SCHEMA) + validate_a2ui_json(payload, schema) -def test_validate_a2ui_json_missing_root(): +def test_validate_a2ui_json_missing_root(schema): payload = { "components": [ {"id": "not-root", "componentProperties": {}} ] } with pytest.raises(ValueError, match="Missing 'root' component"): - validate_a2ui_json(payload, SCHEMA) + validate_a2ui_json(payload, schema) -def test_validate_a2ui_json_dangling_reference_child(): +@pytest.mark.parametrize("component_type, field_name, ids_to_ref", [ + ("Card", "child", "missing_child"), + ("Column", "children", ["child1", "missing_child"]), +]) +def test_validate_a2ui_json_dangling_references(schema, component_type, field_name, ids_to_ref): + """Test dangling references for both single and list fields.""" + # Construct payload dynamically + props = {field_name: ids_to_ref} payload = { "components": [ { "id": "root", "componentProperties": { - "Card": { - "child": "missing_child" - } + component_type: props } } ] } - with pytest.raises(ValueError, match="Component 'root' references missing ID 'missing_child' in field 'child'"): - validate_a2ui_json(payload, SCHEMA) + if isinstance(ids_to_ref, list): + # Add valid children if any + for child_id in ids_to_ref: + if child_id != "missing_child": + payload["components"].append({"id": child_id, "componentProperties": {}}) -def test_validate_a2ui_json_dangling_reference_children(): - payload = { - "components": [ - { - "id": "root", - "componentProperties": { - "Column": { - "children": ["child1", "missing_child"] - } - } - }, - {"id": "child1", "componentProperties": {}} - ] - } - with pytest.raises(ValueError, match="Component 'root' references missing ID 'missing_child' in field 'children'"): - validate_a2ui_json(payload, SCHEMA) + with pytest.raises(ValueError, match=f"Component 'root' references missing ID 'missing_child' in field '{field_name}'"): + validate_a2ui_json(payload, schema) -def test_validate_a2ui_json_self_reference(): +def test_validate_a2ui_json_self_reference(schema): payload = { "components": [ { @@ -173,9 +169,9 @@ def test_validate_a2ui_json_self_reference(): ] } with pytest.raises(ValueError, match="Self-reference detected: Component 'root' references itself in field 'children'"): - validate_a2ui_json(payload, SCHEMA) + validate_a2ui_json(payload, schema) -def test_validate_a2ui_json_circular_reference(): +def test_validate_a2ui_json_circular_reference(schema): payload = { "components": [ { @@ -197,9 +193,9 @@ def test_validate_a2ui_json_circular_reference(): ] } with pytest.raises(ValueError, match="Circular reference detected involving component"): - validate_a2ui_json(payload, SCHEMA) + validate_a2ui_json(payload, schema) -def test_validate_a2ui_json_orphaned_component(): +def test_validate_a2ui_json_orphaned_component(schema): payload = { "components": [ { @@ -217,9 +213,9 @@ def test_validate_a2ui_json_orphaned_component(): ] } with pytest.raises(ValueError, match=r"Orphaned components detected \(not reachable from 'root'\): \['orphan'\]"): - validate_a2ui_json(payload, SCHEMA) + validate_a2ui_json(payload, schema) -def test_validate_a2ui_json_valid_topology_complex(): +def test_validate_a2ui_json_valid_topology_complex(schema): """Test a valid topology with multiple levels.""" payload = { "components": [ @@ -253,10 +249,17 @@ def test_validate_a2ui_json_valid_topology_complex(): } ] } - validate_a2ui_json(payload, SCHEMA) + validate_a2ui_json(payload, schema) -def test_validate_recursion_limit_exceeded(): +def test_validate_recursion_limit_exceeded(schema): """Test that recursion depth > 5 raises ValueError.""" + # Construct deep function call + args = {} + current = args + for i in range(5): # Depth 0 to 5 (6 levels) + current["arg"] = {"call": f"fn{i}", "args": {}} + current = current["arg"]["args"] + payload = { "components": [ { @@ -266,37 +269,8 @@ def test_validate_recursion_limit_exceeded(): "label": "Click me", "action": { "functionCall": { - "call": "fn1", - "args": { - "arg1": { - # Depth 2 - "call": "fn2", - "args": { - "arg2": { - # Depth 3 - "call": "fn3", - "args": { - "arg3": { - # Depth 4 - "call": "fn4", - "args": { - "arg4": { - # Depth 5 - "call": "fn5", - "args": { - "arg5": { - "call": "fn6", - "args": {} - } - } - } - } - } - } - } - } - } - } + "call": "fn_top", + "args": args } } } @@ -305,11 +279,17 @@ def test_validate_recursion_limit_exceeded(): ] } with pytest.raises(ValueError, match="Recursion limit exceeded"): - validate_a2ui_json(payload, SCHEMA) + validate_a2ui_json(payload, schema) -def test_validate_recursion_limit_valid(): +def test_validate_recursion_limit_valid(schema): """Test that recursion depth <= 5 is allowed.""" - # Nesting level 5 + # Construct max depth function call (Depth 5) + args = {} + current = args + for i in range(4): # Depth 0 to 4 (5 levels) + current["arg"] = {"call": f"fn{i}", "args": {}} + current = current["arg"]["args"] + payload = { "components": [ { @@ -319,28 +299,8 @@ def test_validate_recursion_limit_valid(): "label": "Click me", "action": { "functionCall": { - "call": "fn1", - "args": { - "arg1": { - "call": "fn2", - "args": { - "arg2": { - "call": "fn3", - "args": { - "arg3": { - "call": "fn4", - "args": { - "arg4": { - "call": "fn5", - "args": {} - } - } - } - } - } - } - } - } + "call": "fn_top", + "args": args } } } @@ -348,23 +308,17 @@ def test_validate_recursion_limit_valid(): } ] } - validate_a2ui_json(payload, SCHEMA) + validate_a2ui_json(payload, schema) -def test_validate_invalid_datamodel_path_update(): - """Test invalid path in UpdateDataModelMessage.""" - payload = { +@pytest.mark.parametrize("payload", [ + { "updateDataModel": { "surfaceId": "surface1", "path": "invalid//path", "value": "data" } - } - with pytest.raises(ValueError, match="Invalid JSON Pointer syntax"): - validate_a2ui_json(payload, SCHEMA) - -def test_validate_invalid_databinding_path(): - """Test invalid path in DataBinding.""" - payload = { + }, + { "components": [ { "id": "root", @@ -377,11 +331,21 @@ def test_validate_invalid_databinding_path(): } } ] + }, + { + "updateDataModel": { + "surfaceId": "surface1", + "path": "/invalid/escape/~2", + "value": "data" + } } +]) +def test_validate_invalid_paths(schema, payload): + """Test various invalid paths (JSON Pointer syntax).""" with pytest.raises(ValueError, match="Invalid JSON Pointer syntax"): - validate_a2ui_json(payload, SCHEMA) + validate_a2ui_json(payload, schema) -def test_validate_global_recursion_limit_exceeded(): +def test_validate_global_recursion_limit_exceeded(schema): """Test that global recursion depth > 50 raises ValueError.""" # Create a deeply nested dictionary deep_payload = {"level": 0} @@ -391,7 +355,7 @@ def test_validate_global_recursion_limit_exceeded(): current = current["next"] with pytest.raises(ValueError, match="Global recursion limit exceeded"): - validate_a2ui_json(deep_payload, SCHEMA) + validate_a2ui_json(deep_payload, schema) def test_validate_custom_schema_reference(): @@ -446,15 +410,3 @@ def test_validate_custom_schema_reference(): with pytest.raises(ValueError, match="Component 'root' references missing ID 'missing_target' in field 'linkedComponentId'"): validate_a2ui_json(payload, custom_schema) - -def test_validate_invalid_json_pointer_escape(): - """Test invalid escape sequence in JSON Pointer.""" - payload = { - "updateDataModel": { - "surfaceId": "surface1", - "path": "/invalid/escape/~2", - "value": "data" - } - } - with pytest.raises(ValueError, match="Invalid JSON Pointer syntax"): - validate_a2ui_json(payload, SCHEMA) From 2e4efe3a2810f619f1b295098ed1c1e20dc6046f Mon Sep 17 00:00:00 2001 From: wrenj Date: Fri, 13 Feb 2026 16:33:00 -0500 Subject: [PATCH 5/6] format, fix warnings --- .../src/a2ui/extension/validation.py | 512 +++++++++--------- 1 file changed, 270 insertions(+), 242 deletions(-) diff --git a/a2a_agents/python/a2ui_agent/src/a2ui/extension/validation.py b/a2a_agents/python/a2ui_agent/src/a2ui/extension/validation.py index 0291eaedc..db98e16f9 100644 --- a/a2a_agents/python/a2ui_agent/src/a2ui/extension/validation.py +++ b/a2a_agents/python/a2ui_agent/src/a2ui/extension/validation.py @@ -5,6 +5,10 @@ # RFC 6901 compliant regex for JSON Pointer JSON_POINTER_PATTERN = re.compile(r"^(?:\/(?:[^~\/]|~[01])*)*$") +# Recursion Limits +MAX_GLOBAL_DEPTH = 50 +MAX_FUNC_CALL_DEPTH = 5 + # Constants COMPONENTS = "components" ID = "id" @@ -15,248 +19,272 @@ CALL = "call" ARGS = "args" -def validate_a2ui_json(a2ui_json: Union[Dict[str, Any], List[Any]], a2ui_schema: Dict[str, Any]) -> None: - """ - Validates the A2UI JSON payload against the provided schema and checks for integrity. - - Checks performed: - 1. **JSON Schema Validation**: Ensures payload adheres to the A2UI schema. - 2. **Component Integrity**: - - All component IDs are unique. - - A 'root' component exists. - - All unique component references point to valid IDs. - 3. **Topology**: - - No circular references (including self-references). - - No orphaned components (all components must be reachable from 'root'). - 4. **Recursion Limits**: - - Global recursion depth limit (50). - - FunctionCall recursion depth limit (5). - 5. **Path Syntax**: - - Validates JSON Pointer syntax for data paths. - - Args: - a2ui_json: The JSON payload to validate. - a2ui_schema: The schema to validate against. - - Raises: - jsonschema.ValidationError: If the payload does not match the schema. - ValueError: If integrity, topology, or recursion checks fail. - """ - jsonschema.validate(instance=a2ui_json, schema=a2ui_schema) - - # Normalize to list for iteration - messages = a2ui_json if isinstance(a2ui_json, list) else [a2ui_json] - - for message in messages: - if not isinstance(message, dict): - continue - - # Check for SurfaceUpdate which has 'components' - if COMPONENTS in message: - ref_map = _extract_component_ref_fields(a2ui_schema) - _validate_component_integrity(message[COMPONENTS], ref_map) - _validate_topology(message[COMPONENTS], ref_map) - - _validate_recursion_and_paths(message) - - -def _validate_component_integrity(components: List[Dict[str, Any]], ref_fields_map: Dict[str, tuple[Set[str], Set[str]]]) -> None: - """ - Validates that: - 1. All component IDs are unique. - 2. A 'root' component exists. - 3. All references (children, child, etc.) point to existing IDs. - """ - ids: Set[str] = set() - - # 1. Collect IDs and check for duplicates - for comp in components: - comp_id = comp.get(ID) - if comp_id is None: - continue - - if comp_id in ids: - raise ValueError(f"Duplicate component ID found: '{comp_id}'") - ids.add(comp_id) - - # 2. Check for root component - if ROOT not in ids: - raise ValueError(f"Missing '{ROOT}' component: One component must have '{ID}' set to '{ROOT}'.") - - # 3. Check for dangling references using helper - for comp in components: - for ref_id, field_name in _get_component_references(comp, ref_fields_map): - if ref_id not in ids: - raise ValueError(f"Component '{comp.get(ID)}' references missing ID '{ref_id}' in field '{field_name}'") - - -def _validate_topology(components: List[Dict[str, Any]], ref_fields_map: Dict[str, tuple[Set[str], Set[str]]]) -> None: - """ - Validates the topology of the component tree: - 1. No circular references (including self-references). - 2. No orphaned components (all components must be reachable from 'root'). - """ - adj_list: Dict[str, List[str]] = {} - all_ids: Set[str] = set() - - # Build Adjacency List - for comp in components: - comp_id = comp.get(ID) - if comp_id is None: - continue - - all_ids.add(comp_id) - if comp_id not in adj_list: - adj_list[comp_id] = [] - - for ref_id, field_name in _get_component_references(comp, ref_fields_map): - if ref_id == comp_id: - raise ValueError(f"Self-reference detected: Component '{comp_id}' references itself in field '{field_name}'") - adj_list[comp_id].append(ref_id) - - # Detect Cycles using DFS - visited: Set[str] = set() - recursion_stack: Set[str] = set() - - def dfs(node_id: str): - visited.add(node_id) - recursion_stack.add(node_id) - - for neighbor in adj_list.get(node_id, []): - if neighbor not in visited: - dfs(neighbor) - elif neighbor in recursion_stack: - raise ValueError(f"Circular reference detected involving component '{neighbor}'") - - recursion_stack.remove(node_id) - - if ROOT in all_ids: - dfs(ROOT) - - # Check for Orphans - orphans = all_ids - visited - if orphans: - sorted_orphans = sorted(list(orphans)) - raise ValueError(f"Orphaned components detected (not reachable from '{ROOT}'): {sorted_orphans}") - - -def _extract_component_ref_fields(schema: Dict[str, Any]) -> Dict[str, tuple[Set[str], Set[str]]]: - """ - Parses the JSON schema to identify which component properties reference other components. - Returns a map: { component_name: (set_of_single_ref_fields, set_of_list_ref_fields) } - """ - ref_map = {} - - try: - root_defs = schema.get("$defs") or schema.get("definitions", {}) - - # Helper to check if a property schema looks like a ComponentId reference - def is_component_id_ref(prop_schema: Dict[str, Any]) -> bool: - ref = prop_schema.get("$ref", "") - if ref.endswith("ComponentId"): - return True - return False - - def is_child_list_ref(prop_schema: Dict[str, Any]) -> bool: - ref = prop_schema.get("$ref", "") - if ref.endswith("ChildList"): - return True - # Or array of ComponentIds - if prop_schema.get("type") == "array": - items = prop_schema.get("items", {}) - if is_component_id_ref(items): - return True - return False - - comps_schema = schema.get("properties", {}).get(COMPONENTS, {}) - items_schema = comps_schema.get("items", {}) - comp_props_schema = items_schema.get("properties", {}).get(COMPONENT_PROPERTIES, {}) - all_components = comp_props_schema.get("properties", {}) - - for comp_name, comp_schema in all_components.items(): - single_refs = set() - list_refs = set() - - props = comp_schema.get("properties", {}) - for prop_name, prop_schema in props.items(): - if is_component_id_ref(prop_schema): - single_refs.add(prop_name) - elif is_child_list_ref(prop_schema): - list_refs.add(prop_name) - - if single_refs or list_refs: - ref_map[comp_name] = (single_refs, list_refs) - except Exception: - # If schema traversal fails, return empty map - pass - - return ref_map - - -def _get_component_references(component: Dict[str, Any], ref_fields_map: Dict[str, tuple[Set[str], Set[str]]]) -> Iterator[Tuple[str, str]]: - """ - Helper to extract all referenced component IDs from a component. - Yields (referenced_id, field_name). - """ - comp_props_container = component.get(COMPONENT_PROPERTIES) - if not isinstance(comp_props_container, dict): - return - - for comp_type, props in comp_props_container.items(): - if not isinstance(props, dict): - continue - - single_refs, list_refs = ref_fields_map.get(comp_type, (set(), set())) - - for key, value in props.items(): - if key in single_refs: - if isinstance(value, str): - yield value, key - elif key in list_refs: - if isinstance(value, list): - for item in value: - if isinstance(item, str): - yield item, key + +def validate_a2ui_json( + a2ui_json: Union[Dict[str, Any], List[Any]], a2ui_schema: Dict[str, Any] +) -> None: + """ + Validates the A2UI JSON payload against the provided schema and checks for integrity. + + Checks performed: + 1. **JSON Schema Validation**: Ensures payload adheres to the A2UI schema. + 2. **Component Integrity**: + - All component IDs are unique. + - A 'root' component exists. + - All unique component references point to valid IDs. + 3. **Topology**: + - No circular references (including self-references). + - No orphaned components (all components must be reachable from 'root'). + 4. **Recursion Limits**: + - Global recursion depth limit (50). + - FunctionCall recursion depth limit (5). + 5. **Path Syntax**: + - Validates JSON Pointer syntax for data paths. + + Args: + a2ui_json: The JSON payload to validate. + a2ui_schema: The schema to validate against. + + Raises: + jsonschema.ValidationError: If the payload does not match the schema. + ValueError: If integrity, topology, or recursion checks fail. + """ + jsonschema.validate(instance=a2ui_json, schema=a2ui_schema) + + # Normalize to list for iteration + messages = a2ui_json if isinstance(a2ui_json, list) else [a2ui_json] + + for message in messages: + if not isinstance(message, dict): + continue + + # Check for SurfaceUpdate which has 'components' + if COMPONENTS in message: + ref_map = _extract_component_ref_fields(a2ui_schema) + _validate_component_integrity(message[COMPONENTS], ref_map) + _validate_topology(message[COMPONENTS], ref_map) + + _validate_recursion_and_paths(message) + + +def _validate_component_integrity( + components: List[Dict[str, Any]], + ref_fields_map: Dict[str, tuple[Set[str], Set[str]]], +) -> None: + """ + Validates that: + 1. All component IDs are unique. + 2. A 'root' component exists. + 3. All references (children, child, etc.) point to existing IDs. + """ + ids: Set[str] = set() + + # 1. Collect IDs and check for duplicates + for comp in components: + comp_id = comp.get(ID) + if comp_id is None: + continue + + if comp_id in ids: + raise ValueError(f"Duplicate component ID found: '{comp_id}'") + ids.add(comp_id) + + # 2. Check for root component + if ROOT not in ids: + raise ValueError( + f"Missing '{ROOT}' component: One component must have '{ID}' set to '{ROOT}'." + ) + + # 3. Check for dangling references using helper + for comp in components: + for ref_id, field_name in _get_component_references(comp, ref_fields_map): + if ref_id not in ids: + raise ValueError( + f"Component '{comp.get(ID)}' references missing ID '{ref_id}' in field" + f" '{field_name}'" + ) + + +def _validate_topology( + components: List[Dict[str, Any]], + ref_fields_map: Dict[str, tuple[Set[str], Set[str]]], +) -> None: + """ + Validates the topology of the component tree: + 1. No circular references (including self-references). + 2. No orphaned components (all components must be reachable from 'root'). + """ + adj_list: Dict[str, List[str]] = {} + all_ids: Set[str] = set() + + # Build Adjacency List + for comp in components: + comp_id = comp.get(ID) + if comp_id is None: + continue + + all_ids.add(comp_id) + if comp_id not in adj_list: + adj_list[comp_id] = [] + + for ref_id, field_name in _get_component_references(comp, ref_fields_map): + if ref_id == comp_id: + raise ValueError( + f"Self-reference detected: Component '{comp_id}' references itself in field" + f" '{field_name}'" + ) + adj_list[comp_id].append(ref_id) + + # Detect Cycles using DFS + visited: Set[str] = set() + recursion_stack: Set[str] = set() + + def dfs(node_id: str): + visited.add(node_id) + recursion_stack.add(node_id) + + for neighbor in adj_list.get(node_id, []): + if neighbor not in visited: + dfs(neighbor) + elif neighbor in recursion_stack: + raise ValueError( + f"Circular reference detected involving component '{neighbor}'" + ) + + recursion_stack.remove(node_id) + + if ROOT in all_ids: + dfs(ROOT) + + # Check for Orphans + orphans = all_ids - visited + if orphans: + sorted_orphans = sorted(list(orphans)) + raise ValueError( + f"Orphaned components detected (not reachable from '{ROOT}'): {sorted_orphans}" + ) + + +def _extract_component_ref_fields( + schema: Dict[str, Any], +) -> Dict[str, tuple[Set[str], Set[str]]]: + """ + Parses the JSON schema to identify which component properties reference other components. + Returns a map: { component_name: (set_of_single_ref_fields, set_of_list_ref_fields) } + """ + ref_map = {} + + root_defs = schema.get("$defs") or schema.get("definitions", {}) + + # Helper to check if a property schema looks like a ComponentId reference + def is_component_id_ref(prop_schema: Dict[str, Any]) -> bool: + ref = prop_schema.get("$ref", "") + if ref.endswith("ComponentId"): + return True + return False + + def is_child_list_ref(prop_schema: Dict[str, Any]) -> bool: + ref = prop_schema.get("$ref", "") + if ref.endswith("ChildList"): + return True + # Or array of ComponentIds + if prop_schema.get("type") == "array": + items = prop_schema.get("items", {}) + if is_component_id_ref(items): + return True + return False + + comps_schema = schema.get("properties", {}).get(COMPONENTS, {}) + items_schema = comps_schema.get("items", {}) + comp_props_schema = items_schema.get("properties", {}).get(COMPONENT_PROPERTIES, {}) + all_components = comp_props_schema.get("properties", {}) + + for comp_name, comp_schema in all_components.items(): + single_refs = set() + list_refs = set() + + props = comp_schema.get("properties", {}) + for prop_name, prop_schema in props.items(): + if is_component_id_ref(prop_schema): + single_refs.add(prop_name) + elif is_child_list_ref(prop_schema): + list_refs.add(prop_name) + + if single_refs or list_refs: + ref_map[comp_name] = (single_refs, list_refs) + + return ref_map + + +def _get_component_references( + component: Dict[str, Any], ref_fields_map: Dict[str, tuple[Set[str], Set[str]]] +) -> Iterator[Tuple[str, str]]: + """ + Helper to extract all referenced component IDs from a component. + Yields (referenced_id, field_name). + """ + comp_props_container = component.get(COMPONENT_PROPERTIES) + if not isinstance(comp_props_container, dict): + return + + for comp_type, props in comp_props_container.items(): + if not isinstance(props, dict): + continue + + single_refs, list_refs = ref_fields_map.get(comp_type, (set(), set())) + + for key, value in props.items(): + if key in single_refs: + if isinstance(value, str): + yield value, key + elif key in list_refs: + if isinstance(value, list): + for item in value: + if isinstance(item, str): + yield item, key def _validate_recursion_and_paths(data: Any) -> None: - """ - Validates: - 1. Global recursion depth limit (50). - 2. FunctionCall recursion depth limit (5). - 3. Path syntax for DataBindings/DataModelUpdates. - """ - def traverse(item: Any, global_depth: int, func_depth: int): - if global_depth > 50: - raise ValueError("Global recursion limit exceeded: Depth > 50") - - if isinstance(item, list): - for x in item: - traverse(x, global_depth + 1, func_depth) - return - - if isinstance(item, dict): - # Check for path - if PATH in item and isinstance(item[PATH], str): - path = item[PATH] - if not re.fullmatch(JSON_POINTER_PATTERN, path): - raise ValueError(f"Invalid JSON Pointer syntax: '{path}'") - - # Check for FunctionCall - is_func = CALL in item and ARGS in item - - if is_func: - if func_depth >= 5: - raise ValueError(f"Recursion limit exceeded: {FUNCTION_CALL} depth > 5") - - # Increment func_depth only for 'args', but global_depth matches traversal - for k, v in item.items(): - if k == ARGS: - traverse(v, global_depth + 1, func_depth + 1) - else: - traverse(v, global_depth + 1, func_depth) - else: - for v in item.values(): - traverse(v, global_depth + 1, func_depth) - - traverse(data, 0, 0) \ No newline at end of file + """ + Validates: + 1. Global recursion depth limit (50). + 2. FunctionCall recursion depth limit (5). + 3. Path syntax for DataBindings/DataModelUpdates. + """ + + def traverse(item: Any, global_depth: int, func_depth: int): + if global_depth > MAX_GLOBAL_DEPTH: + raise ValueError(f"Global recursion limit exceeded: Depth > {MAX_GLOBAL_DEPTH}") + + if isinstance(item, list): + for x in item: + traverse(x, global_depth + 1, func_depth) + return + + if isinstance(item, dict): + # Check for path + if PATH in item and isinstance(item[PATH], str): + path = item[PATH] + if not re.fullmatch(JSON_POINTER_PATTERN, path): + raise ValueError(f"Invalid JSON Pointer syntax: '{path}'") + + # Check for FunctionCall + is_func = CALL in item and ARGS in item + + if is_func: + if func_depth >= MAX_FUNC_CALL_DEPTH: + raise ValueError( + f"Recursion limit exceeded: {FUNCTION_CALL} depth > {MAX_FUNC_CALL_DEPTH}" + ) + + # Increment func_depth only for 'args', but global_depth matches traversal + for k, v in item.items(): + if k == ARGS: + traverse(v, global_depth + 1, func_depth + 1) + else: + traverse(v, global_depth + 1, func_depth) + else: + for v in item.values(): + traverse(v, global_depth + 1, func_depth) + + traverse(data, 0, 0) From eb04d27d286d0908d2c610c4cf5ace53f98263a7 Mon Sep 17 00:00:00 2001 From: wrenj Date: Fri, 13 Feb 2026 16:34:04 -0500 Subject: [PATCH 6/6] fix test format --- .../a2ui_agent/tests/test_validation.py | 694 ++++++++---------- 1 file changed, 320 insertions(+), 374 deletions(-) diff --git a/a2a_agents/python/a2ui_agent/tests/test_validation.py b/a2a_agents/python/a2ui_agent/tests/test_validation.py index 11483528c..afd642e58 100644 --- a/a2a_agents/python/a2ui_agent/tests/test_validation.py +++ b/a2a_agents/python/a2ui_agent/tests/test_validation.py @@ -1,412 +1,358 @@ - import pytest import jsonschema from a2ui.extension.validation import validate_a2ui_json + # Fixture for the schema @pytest.fixture def schema(): - return { - "type": "object", - "$defs": { - "ComponentId": {"type": "string"}, - "ChildList": { - "type": "array", - "items": {"$ref": "#/$defs/ComponentId"} - } - }, - "properties": { - "components": { - "type": "array", - "items": { - "type": "object", - "properties": { - "id": {"$ref": "#/$defs/ComponentId"}, - "componentProperties": { - "type": "object", - "properties": { - "Column": { - "type": "object", - "properties": { - "children": {"$ref": "#/$defs/ChildList"} - } - }, - "Row": { - "type": "object", - "properties": { - "children": {"$ref": "#/$defs/ChildList"} - } - }, - "Container": { - "type": "object", - "properties": { - "children": {"$ref": "#/$defs/ChildList"} - } - }, - "Card": { - "type": "object", - "properties": { - "child": {"$ref": "#/$defs/ComponentId"} - } - }, - "Button": { - "type": "object", - "properties": { - "child": {"$ref": "#/$defs/ComponentId"}, - "action": { - "properties": { - "functionCall": { - "properties": { - "call": {"type": "string"}, - "args": {"type": "object"} - } - } - } - } - } - }, - "Text": { - "type": "object", - "properties": { - "text": { - "oneOf": [ - {"type": "string"}, - {"type": "object"} - ] - } - } - } - } - } - }, - "required": ["id"] - } - } - } - } + return { + "type": "object", + "$defs": { + "ComponentId": {"type": "string"}, + "ChildList": {"type": "array", "items": {"$ref": "#/$defs/ComponentId"}}, + }, + "properties": { + "components": { + "type": "array", + "items": { + "type": "object", + "properties": { + "id": {"$ref": "#/$defs/ComponentId"}, + "componentProperties": { + "type": "object", + "properties": { + "Column": { + "type": "object", + "properties": { + "children": {"$ref": "#/$defs/ChildList"} + }, + }, + "Row": { + "type": "object", + "properties": { + "children": {"$ref": "#/$defs/ChildList"} + }, + }, + "Container": { + "type": "object", + "properties": { + "children": {"$ref": "#/$defs/ChildList"} + }, + }, + "Card": { + "type": "object", + "properties": { + "child": {"$ref": "#/$defs/ComponentId"} + }, + }, + "Button": { + "type": "object", + "properties": { + "child": {"$ref": "#/$defs/ComponentId"}, + "action": { + "properties": { + "functionCall": { + "properties": { + "call": {"type": "string"}, + "args": {"type": "object"}, + } + } + } + }, + }, + }, + "Text": { + "type": "object", + "properties": { + "text": { + "oneOf": [ + {"type": "string"}, + {"type": "object"}, + ] + } + }, + }, + }, + }, + }, + "required": ["id"], + }, + } + }, + } + def test_validate_a2ui_json_valid_integrity(schema): - payload = { - "components": [ - { - "id": "root", - "componentProperties": { - "Column": { - "children": ["child1"] - } - } - }, - { - "id": "child1", - "componentProperties": { - "Text": { - "text": "Hello" - } - } - } - ] - } - validate_a2ui_json(payload, schema) + payload = { + "components": [ + {"id": "root", "componentProperties": {"Column": {"children": ["child1"]}}}, + {"id": "child1", "componentProperties": {"Text": {"text": "Hello"}}}, + ] + } + validate_a2ui_json(payload, schema) + def test_validate_a2ui_json_duplicate_ids(schema): - payload = { - "components": [ - {"id": "root", "componentProperties": {}}, - {"id": "root", "componentProperties": {}} - ] - } - with pytest.raises(ValueError, match="Duplicate component ID found: 'root'"): - validate_a2ui_json(payload, schema) + payload = { + "components": [ + {"id": "root", "componentProperties": {}}, + {"id": "root", "componentProperties": {}}, + ] + } + with pytest.raises(ValueError, match="Duplicate component ID found: 'root'"): + validate_a2ui_json(payload, schema) + def test_validate_a2ui_json_missing_root(schema): - payload = { - "components": [ - {"id": "not-root", "componentProperties": {}} - ] - } - with pytest.raises(ValueError, match="Missing 'root' component"): - validate_a2ui_json(payload, schema) - -@pytest.mark.parametrize("component_type, field_name, ids_to_ref", [ - ("Card", "child", "missing_child"), - ("Column", "children", ["child1", "missing_child"]), -]) -def test_validate_a2ui_json_dangling_references(schema, component_type, field_name, ids_to_ref): - """Test dangling references for both single and list fields.""" - # Construct payload dynamically - props = {field_name: ids_to_ref} - payload = { - "components": [ - { - "id": "root", - "componentProperties": { - component_type: props - } - } - ] - } - if isinstance(ids_to_ref, list): - # Add valid children if any - for child_id in ids_to_ref: - if child_id != "missing_child": - payload["components"].append({"id": child_id, "componentProperties": {}}) + payload = {"components": [{"id": "not-root", "componentProperties": {}}]} + with pytest.raises(ValueError, match="Missing 'root' component"): + validate_a2ui_json(payload, schema) - with pytest.raises(ValueError, match=f"Component 'root' references missing ID 'missing_child' in field '{field_name}'"): - validate_a2ui_json(payload, schema) + +@pytest.mark.parametrize( + "component_type, field_name, ids_to_ref", + [ + ("Card", "child", "missing_child"), + ("Column", "children", ["child1", "missing_child"]), + ], +) +def test_validate_a2ui_json_dangling_references( + schema, component_type, field_name, ids_to_ref +): + """Test dangling references for both single and list fields.""" + # Construct payload dynamically + props = {field_name: ids_to_ref} + payload = { + "components": [{"id": "root", "componentProperties": {component_type: props}}] + } + if isinstance(ids_to_ref, list): + # Add valid children if any + for child_id in ids_to_ref: + if child_id != "missing_child": + payload["components"].append({"id": child_id, "componentProperties": {}}) + + with pytest.raises( + ValueError, + match=( + "Component 'root' references missing ID 'missing_child' in field" + f" '{field_name}'" + ), + ): + validate_a2ui_json(payload, schema) def test_validate_a2ui_json_self_reference(schema): - payload = { - "components": [ - { - "id": "root", - "componentProperties": { - "Container": { - "children": ["root"] - } - } - } - ] - } - with pytest.raises(ValueError, match="Self-reference detected: Component 'root' references itself in field 'children'"): - validate_a2ui_json(payload, schema) + payload = { + "components": [ + {"id": "root", "componentProperties": {"Container": {"children": ["root"]}}} + ] + } + with pytest.raises( + ValueError, + match=( + "Self-reference detected: Component 'root' references itself in field" + " 'children'" + ), + ): + validate_a2ui_json(payload, schema) + def test_validate_a2ui_json_circular_reference(schema): - payload = { - "components": [ - { - "id": "root", - "componentProperties": { - "Container": { - "children": ["child1"] - } - } - }, - { - "id": "child1", - "componentProperties": { - "Container": { - "children": ["root"] - } - } - } - ] - } - with pytest.raises(ValueError, match="Circular reference detected involving component"): - validate_a2ui_json(payload, schema) + payload = { + "components": [ + { + "id": "root", + "componentProperties": {"Container": {"children": ["child1"]}}, + }, + { + "id": "child1", + "componentProperties": {"Container": {"children": ["root"]}}, + }, + ] + } + with pytest.raises( + ValueError, match="Circular reference detected involving component" + ): + validate_a2ui_json(payload, schema) + def test_validate_a2ui_json_orphaned_component(schema): - payload = { - "components": [ - { - "id": "root", - "componentProperties": { - "Container": { - "children": [] - } - } - }, - { - "id": "orphan", - "componentProperties": {} - } - ] - } - with pytest.raises(ValueError, match=r"Orphaned components detected \(not reachable from 'root'\): \['orphan'\]"): - validate_a2ui_json(payload, schema) + payload = { + "components": [ + {"id": "root", "componentProperties": {"Container": {"children": []}}}, + {"id": "orphan", "componentProperties": {}}, + ] + } + with pytest.raises( + ValueError, + match=r"Orphaned components detected \(not reachable from 'root'\): \['orphan'\]", + ): + validate_a2ui_json(payload, schema) + def test_validate_a2ui_json_valid_topology_complex(schema): - """Test a valid topology with multiple levels.""" - payload = { - "components": [ - { - "id": "root", - "componentProperties": { - "Container": { - "children": ["child1", "child2"] - } - } - }, - { - "id": "child1", - "componentProperties": { - "Text": {"text": "Hello"} - } - }, - { - "id": "child2", - "componentProperties": { - "Container": { - "children": ["child3"] - } - } - }, - { - "id": "child3", - "componentProperties": { - "Text": {"text": "World"} - } - } - ] - } - validate_a2ui_json(payload, schema) + """Test a valid topology with multiple levels.""" + payload = { + "components": [ + { + "id": "root", + "componentProperties": {"Container": {"children": ["child1", "child2"]}}, + }, + {"id": "child1", "componentProperties": {"Text": {"text": "Hello"}}}, + { + "id": "child2", + "componentProperties": {"Container": {"children": ["child3"]}}, + }, + {"id": "child3", "componentProperties": {"Text": {"text": "World"}}}, + ] + } + validate_a2ui_json(payload, schema) + def test_validate_recursion_limit_exceeded(schema): - """Test that recursion depth > 5 raises ValueError.""" - # Construct deep function call - args = {} - current = args - for i in range(5): # Depth 0 to 5 (6 levels) - current["arg"] = {"call": f"fn{i}", "args": {}} - current = current["arg"]["args"] - - payload = { - "components": [ - { - "id": "root", - "componentProperties": { - "Button": { - "label": "Click me", - "action": { - "functionCall": { - "call": "fn_top", - "args": args - } - } - } - } - } - ] - } - with pytest.raises(ValueError, match="Recursion limit exceeded"): - validate_a2ui_json(payload, schema) + """Test that recursion depth > 5 raises ValueError.""" + # Construct deep function call + args = {} + current = args + for i in range(5): # Depth 0 to 5 (6 levels) + current["arg"] = {"call": f"fn{i}", "args": {}} + current = current["arg"]["args"] -def test_validate_recursion_limit_valid(schema): - """Test that recursion depth <= 5 is allowed.""" - # Construct max depth function call (Depth 5) - args = {} - current = args - for i in range(4): # Depth 0 to 4 (5 levels) - current["arg"] = {"call": f"fn{i}", "args": {}} - current = current["arg"]["args"] - - payload = { - "components": [ - { - "id": "root", - "componentProperties": { - "Button": { - "label": "Click me", - "action": { - "functionCall": { - "call": "fn_top", - "args": args - } - } - } - } - } - ] - } + payload = { + "components": [{ + "id": "root", + "componentProperties": { + "Button": { + "label": "Click me", + "action": {"functionCall": {"call": "fn_top", "args": args}}, + } + }, + }] + } + with pytest.raises(ValueError, match="Recursion limit exceeded"): validate_a2ui_json(payload, schema) -@pytest.mark.parametrize("payload", [ - { - "updateDataModel": { - "surfaceId": "surface1", - "path": "invalid//path", - "value": "data" - } - }, - { - "components": [ - { + +def test_validate_recursion_limit_valid(schema): + """Test that recursion depth <= 5 is allowed.""" + # Construct max depth function call (Depth 5) + args = {} + current = args + for i in range(4): # Depth 0 to 4 (5 levels) + current["arg"] = {"call": f"fn{i}", "args": {}} + current = current["arg"]["args"] + + payload = { + "components": [{ + "id": "root", + "componentProperties": { + "Button": { + "label": "Click me", + "action": {"functionCall": {"call": "fn_top", "args": args}}, + } + }, + }] + } + validate_a2ui_json(payload, schema) + + +@pytest.mark.parametrize( + "payload", + [ + { + "updateDataModel": { + "surfaceId": "surface1", + "path": "invalid//path", + "value": "data", + } + }, + { + "components": [{ "id": "root", "componentProperties": { - "Text": { - "text": { - "path": "invalid path with spaces" - } - } - } + "Text": {"text": {"path": "invalid path with spaces"}} + }, + }] + }, + { + "updateDataModel": { + "surfaceId": "surface1", + "path": "/invalid/escape/~2", + "value": "data", } - ] - }, - { - "updateDataModel": { - "surfaceId": "surface1", - "path": "/invalid/escape/~2", - "value": "data" - } - } -]) + }, + ], +) def test_validate_invalid_paths(schema, payload): - """Test various invalid paths (JSON Pointer syntax).""" - with pytest.raises(ValueError, match="Invalid JSON Pointer syntax"): - validate_a2ui_json(payload, schema) + """Test various invalid paths (JSON Pointer syntax).""" + with pytest.raises(ValueError, match="Invalid JSON Pointer syntax"): + validate_a2ui_json(payload, schema) + def test_validate_global_recursion_limit_exceeded(schema): - """Test that global recursion depth > 50 raises ValueError.""" - # Create a deeply nested dictionary - deep_payload = {"level": 0} - current = deep_payload - for i in range(55): - current["next"] = {"level": i + 1} - current = current["next"] - - with pytest.raises(ValueError, match="Global recursion limit exceeded"): - validate_a2ui_json(deep_payload, schema) + """Test that global recursion depth > 50 raises ValueError.""" + # Create a deeply nested dictionary + deep_payload = {"level": 0} + current = deep_payload + for i in range(55): + current["next"] = {"level": i + 1} + current = current["next"] + + with pytest.raises(ValueError, match="Global recursion limit exceeded"): + validate_a2ui_json(deep_payload, schema) def test_validate_custom_schema_reference(): - """Test validation with a custom schema where a component has a non-standard reference field.""" - # Custom schema extending the base one - custom_schema = { - "type": "object", - "$defs": { - "ComponentId": {"type": "string"}, - "ChildList": { - "type": "array", - "items": {"$ref": "#/$defs/ComponentId"} - } - }, - "properties": { - "components": { - "type": "array", - "items": { - "type": "object", - "properties": { - "id": {"$ref": "#/$defs/ComponentId"}, - "componentProperties": { - "type": "object", - "properties": { - "CustomLink": { - "type": "object", - "properties": { - "linkedComponentId": {"$ref": "#/$defs/ComponentId"} - } - } - } - } - }, - "required": ["id"] - } - } - } - } + """Test validation with a custom schema where a component has a non-standard reference field.""" + # Custom schema extending the base one + custom_schema = { + "type": "object", + "$defs": { + "ComponentId": {"type": "string"}, + "ChildList": {"type": "array", "items": {"$ref": "#/$defs/ComponentId"}}, + }, + "properties": { + "components": { + "type": "array", + "items": { + "type": "object", + "properties": { + "id": {"$ref": "#/$defs/ComponentId"}, + "componentProperties": { + "type": "object", + "properties": { + "CustomLink": { + "type": "object", + "properties": { + "linkedComponentId": { + "$ref": "#/$defs/ComponentId" + } + }, + } + }, + }, + }, + "required": ["id"], + }, + } + }, + } - payload = { - "components": [ - { - "id": "root", - "componentProperties": { - "CustomLink": { - "linkedComponentId": "missing_target" - } - } - } - ] - } - - with pytest.raises(ValueError, match="Component 'root' references missing ID 'missing_target' in field 'linkedComponentId'"): - validate_a2ui_json(payload, custom_schema) + payload = { + "components": [{ + "id": "root", + "componentProperties": { + "CustomLink": {"linkedComponentId": "missing_target"} + }, + }] + } + + with pytest.raises( + ValueError, + match=( + "Component 'root' references missing ID 'missing_target' in field" + " 'linkedComponentId'" + ), + ): + validate_a2ui_json(payload, custom_schema)