diff --git a/lib/crewai/tests/a2a/extensions/test_a2ui_schema_conformance.py b/lib/crewai/tests/a2a/extensions/test_a2ui_schema_conformance.py new file mode 100644 index 000000000..d60a9767a --- /dev/null +++ b/lib/crewai/tests/a2a/extensions/test_a2ui_schema_conformance.py @@ -0,0 +1,324 @@ +"""Cross-validate A2UI Pydantic models against vendored JSON schemas. + +Ensures the two validation sources stay in sync: representative payloads +must be accepted or rejected consistently by both the Pydantic models and +the JSON schemas. +""" + +from __future__ import annotations + +from typing import Any + +import jsonschema +import pytest + +from crewai.a2a.extensions.a2ui.models import A2UIEvent, A2UIMessage +from crewai.a2a.extensions.a2ui.schema import load_schema + + +SERVER_SCHEMA = load_schema("server_to_client") +CLIENT_SCHEMA = load_schema("client_to_server") +CATALOG_SCHEMA = load_schema("standard_catalog_definition") + + +def _json_schema_valid(schema: dict[str, Any], instance: dict[str, Any]) -> bool: + """Return True if *instance* validates against *schema*.""" + try: + jsonschema.validate(instance, schema) + return True + except jsonschema.ValidationError: + return False + + +def _pydantic_valid_message(data: dict[str, Any]) -> bool: + """Return True if *data* validates as an A2UIMessage.""" + try: + A2UIMessage.model_validate(data) + return True + except Exception: + return False + + +def _pydantic_valid_event(data: dict[str, Any]) -> bool: + """Return True if *data* validates as an A2UIEvent.""" + try: + A2UIEvent.model_validate(data) + return True + except Exception: + return False + + +# --------------------------------------------------------------------------- +# Valid server-to-client payloads +# --------------------------------------------------------------------------- + +VALID_SERVER_MESSAGES: list[dict[str, Any]] = [ + { + "beginRendering": { + "surfaceId": "s1", + "root": "root-col", + }, + }, + { + "beginRendering": { + "surfaceId": "s2", + "root": "root-col", + "catalogId": "standard (v0.8)", + "styles": {"primaryColor": "#FF0000", "font": "Roboto"}, + }, + }, + { + "surfaceUpdate": { + "surfaceId": "s1", + "components": [ + { + "id": "title", + "component": { + "Text": {"text": {"literalString": "Hello"}}, + }, + }, + ], + }, + }, + { + "surfaceUpdate": { + "surfaceId": "s1", + "components": [ + { + "id": "weighted", + "weight": 2.0, + "component": { + "Column": { + "children": {"explicitList": ["a", "b"]}, + }, + }, + }, + ], + }, + }, + { + "dataModelUpdate": { + "surfaceId": "s1", + "contents": [ + {"key": "name", "valueString": "Alice"}, + {"key": "score", "valueNumber": 42}, + {"key": "active", "valueBoolean": True}, + ], + }, + }, + { + "dataModelUpdate": { + "surfaceId": "s1", + "path": "/user", + "contents": [ + { + "key": "prefs", + "valueMap": [ + {"key": "theme", "valueString": "dark"}, + ], + }, + ], + }, + }, + { + "deleteSurface": {"surfaceId": "s1"}, + }, +] + +# --------------------------------------------------------------------------- +# Invalid server-to-client payloads +# --------------------------------------------------------------------------- + +INVALID_SERVER_MESSAGES: list[dict[str, Any]] = [ + {}, + {"beginRendering": {"surfaceId": "s1"}}, + {"surfaceUpdate": {"surfaceId": "s1", "components": []}}, + { + "beginRendering": {"surfaceId": "s1", "root": "r"}, + "deleteSurface": {"surfaceId": "s1"}, + }, + {"unknownType": {"surfaceId": "s1"}}, +] + +# --------------------------------------------------------------------------- +# Valid client-to-server payloads +# --------------------------------------------------------------------------- + +VALID_CLIENT_EVENTS: list[dict[str, Any]] = [ + { + "userAction": { + "name": "click", + "surfaceId": "s1", + "sourceComponentId": "btn-1", + "timestamp": "2026-03-12T10:00:00Z", + "context": {}, + }, + }, + { + "userAction": { + "name": "submit", + "surfaceId": "s1", + "sourceComponentId": "btn-2", + "timestamp": "2026-03-12T10:00:00Z", + "context": {"field": "value"}, + }, + }, + { + "error": {"message": "render failed", "code": 500}, + }, +] + +# --------------------------------------------------------------------------- +# Invalid client-to-server payloads +# --------------------------------------------------------------------------- + +INVALID_CLIENT_EVENTS: list[dict[str, Any]] = [ + {}, + {"userAction": {"name": "click"}}, + { + "userAction": { + "name": "click", + "surfaceId": "s1", + "sourceComponentId": "btn-1", + "timestamp": "2026-03-12T10:00:00Z", + "context": {}, + }, + "error": {"message": "oops"}, + }, +] + +# --------------------------------------------------------------------------- +# Catalog component payloads (validated structurally) +# --------------------------------------------------------------------------- + +VALID_COMPONENTS: dict[str, dict[str, Any]] = { + "Text": {"text": {"literalString": "hello"}, "usageHint": "h1"}, + "Image": {"url": {"path": "/img/url"}, "fit": "cover", "usageHint": "avatar"}, + "Icon": {"name": {"literalString": "home"}}, + "Video": {"url": {"literalString": "https://example.com/video.mp4"}}, + "AudioPlayer": {"url": {"literalString": "https://example.com/audio.mp3"}}, + "Row": {"children": {"explicitList": ["a", "b"]}, "distribution": "center"}, + "Column": {"children": {"template": {"componentId": "c1", "dataBinding": "/list"}}}, + "List": {"children": {"explicitList": ["x"]}, "direction": "horizontal"}, + "Card": {"child": "inner"}, + "Tabs": {"tabItems": [{"title": {"literalString": "Tab 1"}, "child": "content"}]}, + "Divider": {"axis": "horizontal"}, + "Modal": {"entryPointChild": "trigger", "contentChild": "body"}, + "Button": {"child": "label", "action": {"name": "go"}}, + "CheckBox": {"label": {"literalString": "Accept"}, "value": {"literalBoolean": False}}, + "TextField": {"label": {"literalString": "Name"}}, + "DateTimeInput": {"value": {"path": "/date"}}, + "MultipleChoice": { + "selections": {"literalArray": ["a"]}, + "options": [{"label": {"literalString": "A"}, "value": "a"}], + }, + "Slider": {"value": {"literalNumber": 50}, "minValue": 0, "maxValue": 100}, +} + + +class TestServerToClientConformance: + """Pydantic models and JSON schema must agree on server-to-client messages.""" + + @pytest.mark.parametrize("payload", VALID_SERVER_MESSAGES) + def test_valid_accepted_by_both(self, payload: dict[str, Any]) -> None: + assert _json_schema_valid(SERVER_SCHEMA, payload), ( + f"JSON schema rejected valid payload: {payload}" + ) + assert _pydantic_valid_message(payload), ( + f"Pydantic rejected valid payload: {payload}" + ) + + @pytest.mark.parametrize("payload", INVALID_SERVER_MESSAGES) + def test_invalid_rejected_by_pydantic(self, payload: dict[str, Any]) -> None: + assert not _pydantic_valid_message(payload), ( + f"Pydantic accepted invalid payload: {payload}" + ) + + +class TestClientToServerConformance: + """Pydantic models and JSON schema must agree on client-to-server events.""" + + @pytest.mark.parametrize("payload", VALID_CLIENT_EVENTS) + def test_valid_accepted_by_both(self, payload: dict[str, Any]) -> None: + assert _json_schema_valid(CLIENT_SCHEMA, payload), ( + f"JSON schema rejected valid payload: {payload}" + ) + assert _pydantic_valid_event(payload), ( + f"Pydantic rejected valid payload: {payload}" + ) + + @pytest.mark.parametrize("payload", INVALID_CLIENT_EVENTS) + def test_invalid_rejected_by_pydantic(self, payload: dict[str, Any]) -> None: + assert not _pydantic_valid_event(payload), ( + f"Pydantic accepted invalid payload: {payload}" + ) + + +class TestCatalogConformance: + """Catalog component schemas and Pydantic models must define the same components.""" + + def test_catalog_component_names_match(self) -> None: + from crewai.a2a.extensions.a2ui.catalog import STANDARD_CATALOG_COMPONENTS + + schema_components = set(CATALOG_SCHEMA["components"].keys()) + assert schema_components == STANDARD_CATALOG_COMPONENTS + + @pytest.mark.parametrize( + "name,props", + list(VALID_COMPONENTS.items()), + ) + def test_valid_component_accepted_by_catalog_schema( + self, name: str, props: dict[str, Any] + ) -> None: + component_schema = CATALOG_SCHEMA["components"][name] + assert _json_schema_valid(component_schema, props), ( + f"Catalog schema rejected valid {name}: {props}" + ) + + @pytest.mark.parametrize( + "name,props", + list(VALID_COMPONENTS.items()), + ) + def test_valid_component_accepted_by_pydantic( + self, name: str, props: dict[str, Any] + ) -> None: + import crewai.a2a.extensions.a2ui.catalog as catalog_mod + + model_cls = getattr(catalog_mod, name) + try: + model_cls.model_validate(props) + except Exception as exc: + pytest.fail(f"Pydantic {name} rejected valid props: {exc}") + + def test_catalog_required_fields_match(self) -> None: + """Required fields in the JSON schema match non-optional Pydantic fields.""" + import crewai.a2a.extensions.a2ui.catalog as catalog_mod + + for comp_name, comp_schema in CATALOG_SCHEMA["components"].items(): + schema_required = set(comp_schema.get("required", [])) + model_cls = getattr(catalog_mod, comp_name) + pydantic_required = { + info.alias or field_name + for field_name, info in model_cls.model_fields.items() + if info.is_required() + } + assert schema_required == pydantic_required, ( + f"{comp_name}: schema requires {schema_required}, " + f"Pydantic requires {pydantic_required}" + ) + + def test_catalog_fields_match(self) -> None: + """Field names in JSON schema match Pydantic model aliases.""" + import crewai.a2a.extensions.a2ui.catalog as catalog_mod + + for comp_name, comp_schema in CATALOG_SCHEMA["components"].items(): + schema_fields = set(comp_schema.get("properties", {}).keys()) + model_cls = getattr(catalog_mod, comp_name) + pydantic_fields = { + info.alias or field_name + for field_name, info in model_cls.model_fields.items() + } + assert schema_fields == pydantic_fields, ( + f"{comp_name}: schema has {schema_fields}, " + f"Pydantic has {pydantic_fields}" + ) \ No newline at end of file