adding test coverage

incubator/validator-avro/pom.xml

@@ -26,7 +26,7 @@
   <properties>
     <maven.compiler.source>11</maven.compiler.source>
     <maven.compiler.target>11</maven.compiler.target>
-    <jacoco.coverage.ratio>0.87</jacoco.coverage.ratio>
+    <jacoco.coverage.ratio>0.92</jacoco.coverage.ratio>
     <jacoco.missed.count>0</jacoco.missed.count>
   </properties>
 

incubator/validator-avro/src/main/java/io/aklivity/zilla/runtime/validator/avro/AvroReadValidator.java

@@ -101,17 +101,14 @@ private int validateComplete(
             {
                 byte[] record = deserializeAvroRecord(schema, payloadBytes, progress, length - progress);
                 valueRO.wrap(record);
+                next.accept(valueRO, 0, valLength);
                 valLength = record.length;
             }
             else if (validate(schema, payloadBytes, progress, length - progress))
             {
-                valueRO.wrap(data);
+                next.accept(data, index, length);
                 valLength = length;
             }
-            if (valLength != -1)
-            {
-                next.accept(valueRO, 0, valLength);
-            }
         }
         return valLength;
     }
@@ -126,7 +123,7 @@ private byte[] deserializeAvroRecord(
         try
         {
             reader = new GenericDatumReader(schema);
-            GenericRecord record = (GenericRecord) reader.read(null,
+            GenericRecord record = reader.read(null,
                 decoder.binaryDecoder(bytes, offset, length, null));
             JsonEncoder jsonEncoder = encoder.jsonEncoder(record.getSchema(), outputStream);
             writer = record instanceof SpecificRecord ?

incubator/validator-avro/src/test/java/io/aklivity/zilla/runtime/validator/avro/AvroValidatorTest.java

@@ -34,6 +34,7 @@
 import io.aklivity.zilla.runtime.engine.config.CatalogConfig;
 import io.aklivity.zilla.runtime.engine.test.internal.catalog.TestCatalog;
 import io.aklivity.zilla.runtime.engine.test.internal.catalog.config.TestCatalogOptionsConfig;
+import io.aklivity.zilla.runtime.engine.validator.function.FragmentConsumer;
 import io.aklivity.zilla.runtime.engine.validator.function.ValueConsumer;
 import io.aklivity.zilla.runtime.validator.avro.config.AvroValidatorConfig;
 
@@ -214,4 +215,45 @@ public void shouldWriteJsonEventExpectAvro()
         int progress = validator.validate(data, 0, data.capacity(), ValueConsumer.NOP);
         assertEquals(expected.capacity(), progress);
     }
+
+    @Test
+    public void shouldWriteValidFragmentAvroEvent()
+    {
+        CatalogConfig catalogConfig = new CatalogConfig("test0", "test", new TestCatalogOptionsConfig(SCHEMA));
+        LongFunction<CatalogHandler> handler = value -> context.attach(catalogConfig);
+        AvroWriteValidator validator = new AvroWriteValidator(avroConfig, handler);
+
+        DirectBuffer data = new UnsafeBuffer();
+
+        byte[] bytes = {0x06, 0x69, 0x64, 0x30, 0x10, 0x70, 0x6f,
+            0x73, 0x69, 0x74, 0x69, 0x76, 0x65};
+        data.wrap(bytes, 0, bytes.length);
+
+        byte[] expectedBytes = {0x00, 0x00, 0x00, 0x00, 0x01, 0x06, 0x69, 0x64,
+            0x30, 0x10, 0x70, 0x6f, 0x73, 0x69, 0x74, 0x69, 0x76, 0x65};
+        DirectBuffer expected = new UnsafeBuffer();
+        expected.wrap(expectedBytes);
+
+        assertEquals(0, validator.validate(0x00, data, 0, data.capacity(), FragmentConsumer.NOP));
+
+        assertEquals(expected.capacity(), validator.validate(0x01, data, 0, data.capacity(), FragmentConsumer.NOP));
+    }
+
+    @Test
+    public void shouldVerifyValidFragmentAvroEvent()
+    {
+        CatalogConfig catalogConfig = new CatalogConfig("test0", "test", new TestCatalogOptionsConfig(SCHEMA));
+        LongFunction<CatalogHandler> handler = value -> context.attach(catalogConfig);
+        AvroReadValidator validator = new AvroReadValidator(avroConfig, handler);
+
+        DirectBuffer data = new UnsafeBuffer();
+
+        byte[] bytes = {0x00, 0x00, 0x00, 0x00, 0x09, 0x06, 0x69, 0x64,
+            0x30, 0x10, 0x70, 0x6f, 0x73, 0x69, 0x74, 0x69, 0x76, 0x65};
+        data.wrap(bytes, 0, bytes.length);
+
+        assertEquals(0, validator.validate(0x00, data, 0, data.capacity(), FragmentConsumer.NOP));
+
+        assertEquals(data.capacity(), validator.validate(0x01, data, 0, data.capacity(), FragmentConsumer.NOP));
+    }
 }

incubator/validator-core/pom.xml

@@ -26,8 +26,8 @@
   <properties>
     <maven.compiler.source>11</maven.compiler.source>
     <maven.compiler.target>11</maven.compiler.target>
-    <jacoco.coverage.ratio>0.70</jacoco.coverage.ratio>
-    <jacoco.missed.count>1</jacoco.missed.count>
+    <jacoco.coverage.ratio>0.82</jacoco.coverage.ratio>
+    <jacoco.missed.count>0</jacoco.missed.count>
   </properties>
 
   <dependencies>

incubator/validator-core/src/test/java/io/aklivity/zilla/runtime/validator/core/IntegerValidatorFactoryTest.java

@@ -24,14 +24,15 @@
 
 import io.aklivity.zilla.runtime.engine.catalog.CatalogHandler;
 import io.aklivity.zilla.runtime.engine.config.ValidatorConfig;
+import io.aklivity.zilla.runtime.engine.validator.FragmentValidator;
 import io.aklivity.zilla.runtime.engine.validator.ValueValidator;
 import io.aklivity.zilla.runtime.validator.core.config.IntegerValidatorConfig;
 
 public class IntegerValidatorFactoryTest
 {
     @Test
     @SuppressWarnings("unchecked")
-    public void shouldCreateReadValidator()
+    public void shouldCreateValueReader()
     {
         // GIVEN
         ValidatorConfig validator = new IntegerValidatorConfig();
@@ -47,7 +48,7 @@ public void shouldCreateReadValidator()
 
     @Test
     @SuppressWarnings("unchecked")
-    public void shouldCreateWriteValidator()
+    public void shouldCreateValueWriter()
     {
         // GIVEN
         ValidatorConfig validator = new IntegerValidatorConfig();
@@ -60,4 +61,36 @@ public void shouldCreateWriteValidator()
         // THEN
         assertThat(writer, instanceOf(IntegerValidator.class));
     }
+
+    @Test
+    @SuppressWarnings("unchecked")
+    public void shouldCreateFragmentReader()
+    {
+        // GIVEN
+        ValidatorConfig validator = new IntegerValidatorConfig();
+        LongFunction<CatalogHandler> supplyCatalog = mock(LongFunction.class);
+        IntegerValidatorFactory factory = new IntegerValidatorFactory();
+
+        // WHEN
+        FragmentValidator reader = factory.createFragmentReader(validator, supplyCatalog);
+
+        // THEN
+        assertThat(reader, instanceOf(IntegerValidator.class));
+    }
+
+    @Test
+    @SuppressWarnings("unchecked")
+    public void shouldCreateFragmentWriter()
+    {
+        // GIVEN
+        ValidatorConfig validator = new IntegerValidatorConfig();
+        LongFunction<CatalogHandler> supplyCatalog = mock(LongFunction.class);
+        IntegerValidatorFactory factory = new IntegerValidatorFactory();
+
+        // WHEN
+        FragmentValidator writer = factory.createFragmentWriter(validator, supplyCatalog);
+
+        // THEN
+        assertThat(writer, instanceOf(IntegerValidator.class));
+    }
 }

incubator/validator-core/src/test/java/io/aklivity/zilla/runtime/validator/core/IntegerValidatorTest.java

@@ -20,6 +20,7 @@
 import org.agrona.concurrent.UnsafeBuffer;
 import org.junit.Test;
 
+import io.aklivity.zilla.runtime.engine.validator.function.FragmentConsumer;
 import io.aklivity.zilla.runtime.engine.validator.function.ValueConsumer;
 import io.aklivity.zilla.runtime.validator.core.config.IntegerValidatorConfig;
 
@@ -47,4 +48,17 @@ public void shouldVerifyInvalidInteger()
         data.wrap(bytes, 0, bytes.length);
         assertEquals(-1, validator.validate(data, 0, data.capacity(), ValueConsumer.NOP));
     }
+
+    @Test
+    public void shouldVerifyValidFragmentInteger()
+    {
+        DirectBuffer data = new UnsafeBuffer();
+
+        byte[] bytes = {0, 0, 0, 42};
+        data.wrap(bytes, 0, bytes.length);
+
+        assertEquals(0, validator.validate(0x00, data, 0, data.capacity(), FragmentConsumer.NOP));
+
+        assertEquals(data.capacity(), validator.validate(0x01, data, 0, data.capacity(), FragmentConsumer.NOP));
+    }
 }

incubator/validator-core/src/test/java/io/aklivity/zilla/runtime/validator/core/LongValidatorFactoryTest.java

@@ -24,14 +24,15 @@
 
 import io.aklivity.zilla.runtime.engine.catalog.CatalogHandler;
 import io.aklivity.zilla.runtime.engine.config.ValidatorConfig;
+import io.aklivity.zilla.runtime.engine.validator.FragmentValidator;
 import io.aklivity.zilla.runtime.engine.validator.ValueValidator;
 import io.aklivity.zilla.runtime.validator.core.config.LongValidatorConfig;
 
 public class LongValidatorFactoryTest
 {
     @Test
     @SuppressWarnings("unchecked")
-    public void shouldCreateReadValidator()
+    public void shouldCreateValueReader()
     {
         // GIVEN
         ValidatorConfig validator = new LongValidatorConfig();
@@ -47,7 +48,7 @@ public void shouldCreateReadValidator()
 
     @Test
     @SuppressWarnings("unchecked")
-    public void shouldCreateWriteValidator()
+    public void shouldCreateValueWriter()
     {
         // GIVEN
         ValidatorConfig validator = new LongValidatorConfig();
@@ -60,4 +61,36 @@ public void shouldCreateWriteValidator()
         // THEN
         assertThat(writer, instanceOf(LongValidator.class));
     }
+
+    @Test
+    @SuppressWarnings("unchecked")
+    public void shouldCreateFragmentReader()
+    {
+        // GIVEN
+        ValidatorConfig validator = new LongValidatorConfig();
+        LongFunction<CatalogHandler> supplyCatalog = mock(LongFunction.class);
+        LongValidatorFactory factory = new LongValidatorFactory();
+
+        // WHEN
+        FragmentValidator reader = factory.createFragmentReader(validator, supplyCatalog);
+
+        // THEN
+        assertThat(reader, instanceOf(LongValidator.class));
+    }
+
+    @Test
+    @SuppressWarnings("unchecked")
+    public void shouldCreateFragmentWriter()
+    {
+        // GIVEN
+        ValidatorConfig validator = new LongValidatorConfig();
+        LongFunction<CatalogHandler> supplyCatalog = mock(LongFunction.class);
+        LongValidatorFactory factory = new LongValidatorFactory();
+
+        // WHEN
+        FragmentValidator writer = factory.createFragmentWriter(validator, supplyCatalog);
+
+        // THEN
+        assertThat(writer, instanceOf(LongValidator.class));
+    }
 }

incubator/validator-core/src/test/java/io/aklivity/zilla/runtime/validator/core/LongValidatorTest.java

@@ -20,6 +20,7 @@
 import org.agrona.concurrent.UnsafeBuffer;
 import org.junit.Test;
 
+import io.aklivity.zilla.runtime.engine.validator.function.FragmentConsumer;
 import io.aklivity.zilla.runtime.engine.validator.function.ValueConsumer;
 import io.aklivity.zilla.runtime.validator.core.config.LongValidatorConfig;
 
@@ -47,4 +48,17 @@ public void shouldVerifyInvalidLong()
         data.wrap(bytes, 0, bytes.length);
         assertEquals(-1, validator.validate(data, 0, data.capacity(), ValueConsumer.NOP));
     }
+
+    @Test
+    public void shouldVerifyValidFragmentLong()
+    {
+        DirectBuffer data = new UnsafeBuffer();
+
+        byte[] bytes = {0, 0, 0, 0, 0, 0, 0, 42};
+        data.wrap(bytes, 0, bytes.length);
+
+        assertEquals(0, validator.validate(0x00, data, 0, data.capacity(), FragmentConsumer.NOP));
+
+        assertEquals(data.capacity(), validator.validate(0x01, data, 0, data.capacity(), FragmentConsumer.NOP));
+    }
 }

incubator/validator-core/src/test/java/io/aklivity/zilla/runtime/validator/core/StringEncodingTest.java

@@ -0,0 +1,57 @@
+/*
+ * Copyright 2021-2023 Aklivity Inc
+ *
+ * Licensed under the Aklivity Community License (the "License"); you may not use
+ * this file except in compliance with the License.  You may obtain a copy of the
+ * License at
+ *
+ *   https://www.aklivity.io/aklivity-community-license/
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+package io.aklivity.zilla.runtime.validator.core;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.nio.charset.StandardCharsets;
+
+import org.agrona.DirectBuffer;
+import org.agrona.concurrent.UnsafeBuffer;
+import org.junit.Test;
+
+public class StringEncodingTest
+{
+    @Test
+    public void shouldVerifyValidUTF8()
+    {
+        DirectBuffer data = new UnsafeBuffer();
+
+        byte[] bytes = "Valid String".getBytes();
+        data.wrap(bytes, 0, bytes.length);
+
+        assertTrue(StringEncoding.UTF_8.validate(data, 0, bytes.length));
+    }
+
+    @Test
+    public void shouldVerifyValidUTF16()
+    {
+        DirectBuffer data = new UnsafeBuffer();
+
+        byte[] bytes = "Valid String".getBytes(StandardCharsets.UTF_8);
+        data.wrap(bytes, 0, bytes.length);
+
+        assertTrue(StringEncoding.UTF_8.validate(data, 0, bytes.length));
+    }
+
+    @Test
+    public void shouldVerifyStringEncodingOf()
+    {
+        assertEquals(StringEncoding.UTF_8, StringEncoding.of("utf_8"));
+        assertEquals(StringEncoding.UTF_16, StringEncoding.of("utf_16"));
+        assertEquals(StringEncoding.INVALID, StringEncoding.of("invalid_encoding"));
+    }
+}

incubator/validator-core/src/test/java/io/aklivity/zilla/runtime/validator/core/StringValidatorTest.java

@@ -179,4 +179,22 @@ public void shouldVerifyIncompleteMessage()
         data.wrap(bytes, 0, bytes.length);
         assertEquals(0, validator.validate(FLAGS_INIT, data, 0, data.capacity(), FragmentConsumer.NOP));
     }
+
+    @Test
+    public void shouldVerifyValidFragmentUTF8()
+    {
+        StringValidatorConfig config = StringValidatorConfig.builder()
+                .encoding("utf_8")
+                .build();
+        StringValidator validator = new StringValidator(config);
+
+        DirectBuffer data = new UnsafeBuffer();
+
+        byte[] bytes = "Valid String".getBytes();
+        data.wrap(bytes, 0, bytes.length);
+
+        assertEquals(0, validator.validate(0x00, data, 0, data.capacity(), FragmentConsumer.NOP));
+
+        assertEquals(data.capacity(), validator.validate(0x01, data, 0, data.capacity(), FragmentConsumer.NOP));
+    }
 }

incubator/validator-json/pom.xml

@@ -24,7 +24,7 @@
 <properties>
     <maven.compiler.source>11</maven.compiler.source>
     <maven.compiler.target>11</maven.compiler.target>
-    <jacoco.coverage.ratio>0.82</jacoco.coverage.ratio>
+    <jacoco.coverage.ratio>0.88</jacoco.coverage.ratio>
     <jacoco.missed.count>0</jacoco.missed.count>
 </properties>
 

incubator/validator-json/src/main/java/io/aklivity/zilla/runtime/validator/json/JsonReadValidator.java

@@ -18,7 +18,6 @@
 
 import org.agrona.BitUtil;
 import org.agrona.DirectBuffer;
-import org.agrona.concurrent.UnsafeBuffer;
 import org.leadpony.justify.api.JsonSchema;
 
 import io.aklivity.zilla.runtime.engine.catalog.CatalogHandler;
@@ -30,8 +29,6 @@
 
 public class JsonReadValidator extends JsonValidator implements ValueValidator, FragmentValidator
 {
-    private final DirectBuffer valueRO = new UnsafeBuffer();
-
     public JsonReadValidator(
         JsonValidatorConfig config,
         LongFunction<CatalogHandler> supplyCatalog)
@@ -86,9 +83,8 @@ private int validateComplete(
         JsonSchema schema = fetchSchema(schemaId);
         if (schema != null && validate(schema, data, progress, length - progress))
         {
-            valueRO.wrap(data);
+            next.accept(data, index, length);
             valLength = length;
-            next.accept(valueRO, 0, length);
         }
         return valLength;
     }