test: add A2UI schema conformance tests

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}"
+            )