Update to handle catalog IT validation(resolve schema from subject)

runtime/engine/src/test/java/io/aklivity/zilla/runtime/engine/test/internal/binding/TestBindingFactory.java

@@ -275,6 +275,18 @@ private void onInitialBegin(
                             {
                                 doInitialReset(traceId);
                             }
+                            if (assertion.resolve)
+                            {
+                                String schema = handler.resolve(id);
+                                if (assertion.schema == null && schema != null)
+                                {
+                                    doInitialReset(traceId);
+                                }
+                                else if (assertion.schema != null && !assertion.schema.equals(schema))
+                                {
+                                    doInitialReset(traceId);
+                                }
+                            }
                         }
                         else
                         {

runtime/engine/src/test/java/io/aklivity/zilla/runtime/engine/test/internal/binding/config/TestBindingOptionsConfig.java

@@ -87,15 +87,18 @@ public static final class CatalogAssertion
         public final int id;
         public final String schema;
         public final long delay;
+        public final boolean resolve;
 
         public CatalogAssertion(
             int id,
             String schema,
-            long delay)
+            long delay,
+            boolean resolve)
         {
             this.id = id;
             this.schema = schema;
             this.delay = delay;
+            this.resolve = resolve;
         }
     }
 }

runtime/engine/src/test/java/io/aklivity/zilla/runtime/engine/test/internal/binding/config/TestBindingOptionsConfigAdapter.java

@@ -44,6 +44,7 @@ public final class TestBindingOptionsConfigAdapter implements OptionsConfigAdapt
     private static final String ID_NAME = "id";
     private static final String SCHEMA_NAME = "schema";
     private static final String DELAY_NAME = "delay";
+    private static final String RESOLVE_NAME = "resolve";
 
     private final SchemaConfigAdapter schema = new SchemaConfigAdapter();
 
@@ -175,7 +176,8 @@ public OptionsConfig adaptFromJson(
                             catalogAssertions.add(new TestBindingOptionsConfig.CatalogAssertion(
                                 c.containsKey(ID_NAME) ? c.getInt(ID_NAME) : 0,
                                 c.containsKey(SCHEMA_NAME) ? !c.isNull(SCHEMA_NAME) ? c.getString(SCHEMA_NAME) : null : null,
-                                c.containsKey(DELAY_NAME) ? c.getJsonNumber(DELAY_NAME).longValue() : 0L));
+                                c.containsKey(DELAY_NAME) ? c.getJsonNumber(DELAY_NAME).longValue() : 0L,
+                                c.containsKey(RESOLVE_NAME) ? c.getBoolean(RESOLVE_NAME) : false));
                         }
                         testOptions.catalogAssertions(catalogName, catalogAssertions);
                     }

specs/engine.spec/src/main/scripts/io/aklivity/zilla/specs/engine/schema/binding/test.schema.patch.json

@@ -83,6 +83,10 @@
                                                     "delay":
                                                     {
                                                         "type": "number"
+                                                    },
+                                                    "resolve":
+                                                    {
+                                                        "type": "boolean"
                                                     }
                                                 }
                                             }