Skip to content
Open
462 changes: 13 additions & 449 deletions agent_sdks/python/a2ui_agent/tests/express/test_compiler.py

Large diffs are not rendered by default.

237 changes: 0 additions & 237 deletions agent_sdks/python/a2ui_agent/tests/express/test_integration.py
Original file line number Diff line number Diff line change
Expand Up @@ -48,243 +48,6 @@ def setUp(self):
self.catalog = Catalog.from_json(catalog_dict, spec_version="0.9.1")
self.helper = CatalogSchemaHelper(self.catalog)

def test_round_trip_examples(self):
"""Runs a semantically rigorous round-trip test on real catalog examples."""
compiler = ExpressCompiler(self.catalog)
decompiler = ExpressParser(self.catalog)

example_files = glob.glob(os.path.join(EXAMPLES_DIR, "*.json"))
self.assertTrue(
len(example_files) > 0, "No example files found to run round-trip tests."
)

tested_count = 0
for ex_file in sorted(example_files)[:5]:
with open(ex_file, "r", encoding="utf-8") as f:
ex_data = json.load(f)

messages = ex_data.get("messages", [])
components_list = None
surface_id = "test_surf"
catalog_id = (
"https://a2ui.org/specification/v1_0/catalogs/basic/catalog.json"
)

for msg in messages:
if "updateComponents" in msg:
components_list = msg["updateComponents"].get("components", [])
surface_id = msg["updateComponents"].get("surfaceId", surface_id)
break

if not components_list:
continue

tested_count += 1

original_envelope = {
"version": "v1.0",
"createSurface": {
"surfaceId": surface_id,
"catalogId": catalog_id,
"components": components_list,
},
}

dsl = decompiler.decompile(original_envelope)
compiled_envelope = compiler.compile(
dsl, surface_id=surface_id, catalog_id=catalog_id
)[0]

orig_comps = sorted(
original_envelope["createSurface"]["components"], key=lambda x: x["id"]
)
comp_comps = sorted(
compiled_envelope["createSurface"]["components"], key=lambda x: x["id"]
)

self.assertEqual(len(orig_comps), len(comp_comps))

for idx, orig in enumerate(orig_comps):
comp = comp_comps[idx]
self.assertEqual(orig["id"], comp["id"])
self.assertEqual(orig["component"], comp["component"])

for key, orig_v in orig.items():
if key in ["component", "id", "checks"]:
continue
self.assertIn(key, comp)
comp_v = comp[key]

if (
isinstance(orig_v, dict)
and "call" in orig_v
and "returnType" not in orig_v
):
if (
isinstance(comp_v, dict)
and comp_v.get("call") == orig_v["call"]
):
comp_v = {
k2: v2
for k2, v2 in comp_v.items()
if k2 != "returnType"
}
self.assertEqual(orig_v, comp_v)

def test_examples_conversions_match(self):
"""Verifies that all human-authored .a2ui examples compile to match their JSON counterparts."""
compiler = ExpressCompiler(self.catalog)

a2ui_dir = os.path.join(SPEC_DIR, "..", "proposals", "express", "examples")
a2ui_files = glob.glob(os.path.join(a2ui_dir, "*.a2ui"))
self.assertEqual(
len(a2ui_files), 36, f"Expected 36 a2ui files, found {len(a2ui_files)}"
)

for a2ui_file in sorted(a2ui_files):
base_name = os.path.basename(a2ui_file)
json_name = base_name.replace(".a2ui", ".json")
json_file = os.path.join(EXAMPLES_DIR, json_name)
self.assertTrue(
os.path.exists(json_file),
f"JSON counterpart {json_name} does not exist",
)

with open(a2ui_file, "r", encoding="utf-8") as f:
dsl_content = f.read()

with open(json_file, "r", encoding="utf-8") as f:
json_data = json.load(f)

messages = json_data.get("messages", [])
surface_id = "main"
expected_components = []

for msg in messages:
if "createSurface" in msg:
surface_id = msg["createSurface"].get("surfaceId", surface_id)
if "components" in msg["createSurface"]:
expected_components = msg["createSurface"]["components"]
if "updateComponents" in msg:
expected_components = msg["updateComponents"].get("components", [])

compiled_envelope = compiler.compile(dsl_content, surface_id=surface_id)[0]

def normalize_value(val: Any) -> Any:
if isinstance(val, dict):
if "event" in val and isinstance(val["event"], dict):
evt = val["event"]
if "context" in evt and not evt["context"]:
val["event"] = {
k: v for k, v in evt.items() if k != "context"
}
return {
k: normalize_value(v)
for k, v in val.items()
if k != "returnType"
}
if isinstance(val, list):
return [normalize_value(item) for item in val]
return val

if "deleteSurface" in compiled_envelope:
expected_msg = next((m for m in messages if "deleteSurface" in m), None)
self.assertIsNotNone(expected_msg)
self.assertEqual(
expected_msg["deleteSurface"], compiled_envelope["deleteSurface"]
)
continue

if "callFunction" in compiled_envelope:
expected_msg = next((m for m in messages if "callFunction" in m), None)
self.assertIsNotNone(expected_msg)
self.assertEqual(
expected_msg["callFunction"]["call"],
compiled_envelope["callFunction"]["call"],
)
self.assertEqual(
normalize_value(expected_msg["callFunction"].get("args", {})),
normalize_value(compiled_envelope["callFunction"].get("args", {})),
)
continue

if "updateDataModel" in compiled_envelope:
expected_msg = next(
(
m
for m in messages
if "updateDataModel" in m or "updateData" in m
),
None,
)
self.assertIsNotNone(expected_msg)
expected_val = (
expected_msg.get("updateDataModel", {}).get("value", {})
if "updateDataModel" in expected_msg
else expected_msg.get("updateData", {}).get("data", {})
)
self.assertEqual(
normalize_value(expected_val),
normalize_value(
compiled_envelope["updateDataModel"].get("value", {})
),
)
continue

compiled_components = compiled_envelope["createSurface"]["components"]
self.assertEqual(len(compiled_components), len(expected_components))

expected_sorted = sorted(expected_components, key=lambda x: x["id"])
compiled_sorted = sorted(compiled_components, key=lambda x: x["id"])

for idx, expected in enumerate(expected_sorted):
compiled = compiled_sorted[idx]
self.assertEqual(expected["id"], compiled["id"])
self.assertEqual(expected["component"], compiled["component"])

for key, exp_val in expected.items():
if key in ["id", "component"]:
continue
self.assertIn(key, compiled)
comp_val = normalize_value(compiled[key])
exp_val = normalize_value(exp_val)
self.assertEqual(exp_val, comp_val)

def test_data_model_compilation_and_decompilation(self):
"""Validates compiling and decompiling shared data model assignments in the DSL."""
compiler = ExpressCompiler(self.catalog)
decompiler = ExpressParser(self.catalog)

dsl = """$/icon = "check"
$/title = "Enable notification"
$/user/firstName = "Alice"
$/user/age = 30
root = Card(main_column)
main_column = Column([icon, title], _, "center")
icon = Icon($/icon)
title = Text($/title, "body")"""

envelope = compiler.compile(dsl, surface_id="test_data_surf")[0]
self.assertEqual(envelope["version"], "v1.0")
create_surface = envelope["createSurface"]

data_model = create_surface["dataModel"]
self.assertEqual(data_model["icon"], "check")
self.assertEqual(data_model["title"], "Enable notification")
self.assertEqual(data_model["user"]["firstName"], "Alice")
self.assertEqual(data_model["user"]["age"], 30)

decompiled_dsl = decompiler.decompile(envelope)
self.assertIn('$/icon = "check"', decompiled_dsl)
self.assertIn('$/title = "Enable notification"', decompiled_dsl)
self.assertIn("$/user/age = 30", decompiled_dsl)
self.assertIn('$/user/firstName = "Alice"', decompiled_dsl)

compiled_envelope_2 = compiler.compile(
decompiled_dsl, surface_id="test_data_surf"
)[0]
self.assertEqual(compiled_envelope_2["createSurface"]["dataModel"], data_model)

def test_parser_robustness_and_event_variable_resolution(self):
"""Regression tests for parser fallbacks, empty text parts, and event variable resolution."""
compiler = ExpressCompiler(self.catalog)
Expand Down
Loading
Loading