/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements. See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.kafka.common.protocol.types;

import org.apache.kafka.common.utils.ByteUtils;

import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;

import java.nio.ByteBuffer;
import java.util.Arrays;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;

public class ProtocolSerializationTest {

    private Schema schema;
    private Struct struct;

    @BeforeEach
    public void setup() {
        this.schema = new Schema(new Field("boolean", Type.BOOLEAN),
                                 new Field("int8", Type.INT8),
                                 new Field("int16", Type.INT16),
                                 new Field("int32", Type.INT32),
                                 new Field("int64", Type.INT64),
                                 new Field("varint", Type.VARINT),
                                 new Field("varlong", Type.VARLONG),
                                 new Field("float64", Type.FLOAT64),
                                 new Field("string", Type.STRING),
                                 new Field("compact_string", Type.COMPACT_STRING),
                                 new Field("nullable_string", Type.NULLABLE_STRING),
                                 new Field("compact_nullable_string", Type.COMPACT_NULLABLE_STRING),
                                 new Field("bytes", Type.BYTES),
                                 new Field("compact_bytes", Type.COMPACT_BYTES),
                                 new Field("nullable_bytes", Type.NULLABLE_BYTES),
                                 new Field("compact_nullable_bytes", Type.COMPACT_NULLABLE_BYTES),
                                 new Field("array", new ArrayOf(Type.INT32)),
                                 new Field("compact_array", new CompactArrayOf(Type.INT32)),
                                 new Field("null_array", ArrayOf.nullable(Type.INT32)),
                                 new Field("compact_null_array", CompactArrayOf.nullable(Type.INT32)),
                                 new Field("struct", new Schema(new Field("field", new ArrayOf(Type.INT32)))));
        this.struct = new Struct(this.schema).set("boolean", true)
                                             .set("int8", (byte) 1)
                                             .set("int16", (short) 1)
                                             .set("int32", 1)
                                             .set("int64", 1L)
                                             .set("varint", 300)
                                             .set("varlong", 500L)
                                             .set("float64", 0.5D)
                                             .set("string", "1")
                                             .set("compact_string", "1")
                                             .set("nullable_string", null)
                                             .set("compact_nullable_string", null)
                                             .set("bytes", ByteBuffer.wrap("1".getBytes()))
                                             .set("compact_bytes", ByteBuffer.wrap("1".getBytes()))
                                             .set("nullable_bytes", null)
                                             .set("compact_nullable_bytes", null)
                                             .set("array", new Object[] {1})
                                             .set("compact_array", new Object[] {1})
                                             .set("null_array", null)
                                             .set("compact_null_array", null);
        this.struct.set("struct", this.struct.instance("struct").set("field", new Object[] {1, 2, 3}));
    }

    @Test
    public void testSimple() {
        check(Type.BOOLEAN, false, "BOOLEAN");
        check(Type.BOOLEAN, true, "BOOLEAN");
        check(Type.INT8, (byte) -111, "INT8");
        check(Type.INT16, (short) -11111, "INT16");
        check(Type.INT32, -11111111, "INT32");
        check(Type.INT64, -11111111111L, "INT64");
        check(Type.FLOAT64, 2.5, "FLOAT64");
        check(Type.FLOAT64, -0.5, "FLOAT64");
        check(Type.FLOAT64, 1e300, "FLOAT64");
        check(Type.FLOAT64, 0.0, "FLOAT64");
        check(Type.FLOAT64, -0.0, "FLOAT64");
        check(Type.FLOAT64, Double.MAX_VALUE, "FLOAT64");
        check(Type.FLOAT64, Double.MIN_VALUE, "FLOAT64");
        check(Type.FLOAT64, Double.NaN, "FLOAT64");
        check(Type.FLOAT64, Double.NEGATIVE_INFINITY, "FLOAT64");
        check(Type.FLOAT64, Double.POSITIVE_INFINITY, "FLOAT64");
        check(Type.STRING, "", "STRING");
        check(Type.STRING, "hello", "STRING");
        check(Type.STRING, "A\u00ea\u00f1\u00fcC", "STRING");
        check(Type.COMPACT_STRING, "", "COMPACT_STRING");
        check(Type.COMPACT_STRING, "hello", "COMPACT_STRING");
        check(Type.COMPACT_STRING, "A\u00ea\u00f1\u00fcC", "COMPACT_STRING");
        check(Type.NULLABLE_STRING, null, "NULLABLE_STRING");
        check(Type.NULLABLE_STRING, "", "NULLABLE_STRING");
        check(Type.NULLABLE_STRING, "hello", "NULLABLE_STRING");
        check(Type.COMPACT_NULLABLE_STRING, null, "COMPACT_NULLABLE_STRING");
        check(Type.COMPACT_NULLABLE_STRING, "", "COMPACT_NULLABLE_STRING");
        check(Type.COMPACT_NULLABLE_STRING, "hello", "COMPACT_NULLABLE_STRING");
        check(Type.BYTES, ByteBuffer.allocate(0), "BYTES");
        check(Type.BYTES, ByteBuffer.wrap("abcd".getBytes()), "BYTES");
        check(Type.COMPACT_BYTES, ByteBuffer.allocate(0), "COMPACT_BYTES");
        check(Type.COMPACT_BYTES, ByteBuffer.wrap("abcd".getBytes()), "COMPACT_BYTES");
        check(Type.NULLABLE_BYTES, null, "NULLABLE_BYTES");
        check(Type.NULLABLE_BYTES, ByteBuffer.allocate(0), "NULLABLE_BYTES");
        check(Type.NULLABLE_BYTES, ByteBuffer.wrap("abcd".getBytes()), "NULLABLE_BYTES");
        check(Type.COMPACT_NULLABLE_BYTES, null, "COMPACT_NULLABLE_BYTES");
        check(Type.COMPACT_NULLABLE_BYTES, ByteBuffer.allocate(0), "COMPACT_NULLABLE_BYTES");
        check(Type.COMPACT_NULLABLE_BYTES, ByteBuffer.wrap("abcd".getBytes()),
                "COMPACT_NULLABLE_BYTES");
        check(Type.VARINT, Integer.MAX_VALUE, "VARINT");
        check(Type.VARINT, Integer.MIN_VALUE, "VARINT");
        check(Type.VARLONG, Long.MAX_VALUE, "VARLONG");
        check(Type.VARLONG, Long.MIN_VALUE, "VARLONG");
        check(new ArrayOf(Type.INT32), new Object[] {1, 2, 3, 4}, "ARRAY(INT32)");
        check(new ArrayOf(Type.STRING), new Object[] {}, "ARRAY(STRING)");
        check(new ArrayOf(Type.STRING), new Object[] {"hello", "there", "beautiful"},
                "ARRAY(STRING)");
        check(new CompactArrayOf(Type.INT32), new Object[] {1, 2, 3, 4},
                "COMPACT_ARRAY(INT32)");
        check(new CompactArrayOf(Type.COMPACT_STRING), new Object[] {},
                "COMPACT_ARRAY(COMPACT_STRING)");
        check(new CompactArrayOf(Type.COMPACT_STRING),
                new Object[] {"hello", "there", "beautiful"},
                "COMPACT_ARRAY(COMPACT_STRING)");
        check(ArrayOf.nullable(Type.STRING), null, "ARRAY(STRING)");
        check(CompactArrayOf.nullable(Type.COMPACT_STRING), null,
                "COMPACT_ARRAY(COMPACT_STRING)");
    }

    @Test
    public void testNulls() {
        for (BoundField f : this.schema.fields()) {
            Object o = this.struct.get(f);
            try {
                this.struct.set(f, null);
                this.struct.validate();
                if (!f.def.type.isNullable())
                    fail("Should not allow serialization of null value.");
            } catch (SchemaException e) {
                assertFalse(f.def.type.isNullable(), f + " should not be nullable");
            } finally {
                this.struct.set(f, o);
            }
        }
    }

    @Test
    public void testDefault() {
        Schema schema = new Schema(new Field("field", Type.INT32, "doc", 42));
        Struct struct = new Struct(schema);
        assertEquals(42, struct.get("field"), "Should get the default value");
        struct.validate(); // should be valid even with missing value
    }

    @Test
    public void testNullableDefault() {
        checkNullableDefault(Type.NULLABLE_BYTES, ByteBuffer.allocate(0));
        checkNullableDefault(Type.COMPACT_NULLABLE_BYTES, ByteBuffer.allocate(0));
        checkNullableDefault(Type.NULLABLE_STRING, "default");
        checkNullableDefault(Type.COMPACT_NULLABLE_STRING, "default");
    }

    private void checkNullableDefault(Type type, Object defaultValue) {
        // Should use default even if the field allows null values
        Schema schema = new Schema(new Field("field", type, "doc", defaultValue));
        Struct struct = new Struct(schema);
        assertEquals(defaultValue, struct.get("field"), "Should get the default value");
        struct.validate(); // should be valid even with missing value
    }

    @Test
    public void testReadArraySizeTooLarge() {
        Type type = new ArrayOf(Type.INT8);
        int size = 10;
        ByteBuffer invalidBuffer = ByteBuffer.allocate(4 + size);
        invalidBuffer.putInt(Integer.MAX_VALUE);
        for (int i = 0; i < size; i++)
            invalidBuffer.put((byte) i);
        invalidBuffer.rewind();
        try {
            type.read(invalidBuffer);
            fail("Array size not validated");
        } catch (SchemaException e) {
            // Expected exception
        }
    }

    @Test
    public void testReadCompactArraySizeTooLarge() {
        Type type = new CompactArrayOf(Type.INT8);
        int size = 10;
        ByteBuffer invalidBuffer = ByteBuffer.allocate(
            ByteUtils.sizeOfUnsignedVarint(Integer.MAX_VALUE) + size);
        ByteUtils.writeUnsignedVarint(Integer.MAX_VALUE, invalidBuffer);
        for (int i = 0; i < size; i++)
            invalidBuffer.put((byte) i);
        invalidBuffer.rewind();
        try {
            type.read(invalidBuffer);
            fail("Array size not validated");
        } catch (SchemaException e) {
            // Expected exception
        }
    }

    @Test
    public void testReadTaggedFieldsSizeTooLarge() {
        int tag = 1;
        Type type = TaggedFields.of(tag, new Field("field", Type.NULLABLE_STRING));
        int size = 10;
        ByteBuffer buffer = ByteBuffer.allocate(size);
        int numTaggedFields = 1;
        // write the number of tagged fields
        ByteUtils.writeUnsignedVarint(numTaggedFields, buffer);
        // write the tag of the first tagged fields
        ByteUtils.writeUnsignedVarint(tag, buffer);
        // write the size of tagged fields for this tag, using a large number for testing
        ByteUtils.writeUnsignedVarint(Integer.MAX_VALUE, buffer);
        int expectedRemaining = buffer.remaining();
        buffer.rewind();

        // should throw SchemaException while reading the buffer, instead of OOM
        Throwable e = assertThrows(SchemaException.class, () -> type.read(buffer));
        assertEquals("Error reading field of size " + Integer.MAX_VALUE + ", only " + expectedRemaining + " bytes available",
            e.getMessage());
    }

    @Test
    public void testReadNegativeArraySize() {
        Type type = new ArrayOf(Type.INT8);
        int size = 10;
        ByteBuffer invalidBuffer = ByteBuffer.allocate(4 + size);
        invalidBuffer.putInt(-1);
        for (int i = 0; i < size; i++)
            invalidBuffer.put((byte) i);
        invalidBuffer.rewind();
        try {
            type.read(invalidBuffer);
            fail("Array size not validated");
        } catch (SchemaException e) {
            // Expected exception
        }
    }

    @Test
    public void testReadZeroCompactArraySize() {
        Type type = new CompactArrayOf(Type.INT8);
        int size = 10;
        ByteBuffer invalidBuffer = ByteBuffer.allocate(
            ByteUtils.sizeOfUnsignedVarint(0) + size);
        ByteUtils.writeUnsignedVarint(0, invalidBuffer);
        for (int i = 0; i < size; i++)
            invalidBuffer.put((byte) i);
        invalidBuffer.rewind();
        try {
            type.read(invalidBuffer);
            fail("Array size not validated");
        } catch (SchemaException e) {
            // Expected exception
        }
    }

    @Test
    public void testReadStringSizeTooLarge() {
        byte[] stringBytes = "foo".getBytes();
        ByteBuffer invalidBuffer = ByteBuffer.allocate(2 + stringBytes.length);
        invalidBuffer.putShort((short) (stringBytes.length * 5));
        invalidBuffer.put(stringBytes);
        invalidBuffer.rewind();
        try {
            Type.STRING.read(invalidBuffer);
            fail("String size not validated");
        } catch (SchemaException e) {
            // Expected exception
        }
        invalidBuffer.rewind();
        try {
            Type.NULLABLE_STRING.read(invalidBuffer);
            fail("String size not validated");
        } catch (SchemaException e) {
            // Expected exception
        }
    }

    @Test
    public void testReadNegativeStringSize() {
        byte[] stringBytes = "foo".getBytes();
        ByteBuffer invalidBuffer = ByteBuffer.allocate(2 + stringBytes.length);
        invalidBuffer.putShort((short) -1);
        invalidBuffer.put(stringBytes);
        invalidBuffer.rewind();
        try {
            Type.STRING.read(invalidBuffer);
            fail("String size not validated");
        } catch (SchemaException e) {
            // Expected exception
        }
    }

    @Test
    public void testReadBytesSizeTooLarge() {
        byte[] stringBytes = "foo".getBytes();
        ByteBuffer invalidBuffer = ByteBuffer.allocate(4 + stringBytes.length);
        invalidBuffer.putInt(stringBytes.length * 5);
        invalidBuffer.put(stringBytes);
        invalidBuffer.rewind();
        try {
            Type.BYTES.read(invalidBuffer);
            fail("Bytes size not validated");
        } catch (SchemaException e) {
            // Expected exception
        }
        invalidBuffer.rewind();
        try {
            Type.NULLABLE_BYTES.read(invalidBuffer);
            fail("Bytes size not validated");
        } catch (SchemaException e) {
            // Expected exception
        }
    }

    @Test
    public void testReadNegativeBytesSize() {
        byte[] stringBytes = "foo".getBytes();
        ByteBuffer invalidBuffer = ByteBuffer.allocate(4 + stringBytes.length);
        invalidBuffer.putInt(-20);
        invalidBuffer.put(stringBytes);
        invalidBuffer.rewind();
        try {
            Type.BYTES.read(invalidBuffer);
            fail("Bytes size not validated");
        } catch (SchemaException e) {
            // Expected exception
        }
    }

    @Test
    public void testToString() {
        String structStr = this.struct.toString();
        assertNotNull(structStr, "Struct string should not be null.");
        assertFalse(structStr.isEmpty(), "Struct string should not be empty.");
    }

    private Object roundtrip(Type type, Object obj) {
        ByteBuffer buffer = ByteBuffer.allocate(type.sizeOf(obj));
        type.write(buffer, obj);
        assertFalse(buffer.hasRemaining(), "The buffer should now be full.");
        buffer.rewind();
        Object read = type.read(buffer);
        assertFalse(buffer.hasRemaining(), "All bytes should have been read.");
        return read;
    }

    private void check(Type type, Object obj, String expectedTypeName) {
        Object result = roundtrip(type, obj);
        if (obj instanceof Object[]) {
            obj = Arrays.asList((Object[]) obj);
            result = Arrays.asList((Object[]) result);
        }
        assertEquals(expectedTypeName, type.toString());
        assertEquals(obj, result, "The object read back should be the same as what was written.");
    }

    @Test
    public void testStructEquals() {
        Schema schema = new Schema(new Field("field1", Type.NULLABLE_STRING), new Field("field2", Type.NULLABLE_STRING));
        Struct emptyStruct1 = new Struct(schema);
        Struct emptyStruct2 = new Struct(schema);
        assertEquals(emptyStruct1, emptyStruct2);

        Struct mostlyEmptyStruct = new Struct(schema).set("field1", "foo");
        assertNotEquals(emptyStruct1, mostlyEmptyStruct);
        assertNotEquals(mostlyEmptyStruct, emptyStruct1);
    }

    @Test
    public void testReadIgnoringExtraDataAtTheEnd() {
        Schema oldSchema = new Schema(new Field("field1", Type.NULLABLE_STRING), new Field("field2", Type.NULLABLE_STRING));
        Schema newSchema = new Schema(new Field("field1", Type.NULLABLE_STRING));
        String value = "foo bar baz";
        Struct oldFormat = new Struct(oldSchema).set("field1", value).set("field2", "fine to ignore");
        ByteBuffer buffer = ByteBuffer.allocate(oldSchema.sizeOf(oldFormat));
        oldFormat.writeTo(buffer);
        buffer.flip();
        Struct newFormat = newSchema.read(buffer);
        assertEquals(value, newFormat.get("field1"));
    }

    @Test
    public void testReadWhenOptionalDataMissingAtTheEndIsTolerated() {
        Schema oldSchema = new Schema(new Field("field1", Type.NULLABLE_STRING));
        Schema newSchema = new Schema(
                true,
                new Field("field1", Type.NULLABLE_STRING),
                new Field("field2", Type.NULLABLE_STRING, "", true, "default"),
                new Field("field3", Type.NULLABLE_STRING, "", true, null),
                new Field("field4", Type.NULLABLE_BYTES, "", true, ByteBuffer.allocate(0)),
                new Field("field5", Type.INT64, "doc", true, Long.MAX_VALUE));
        String value = "foo bar baz";
        Struct oldFormat = new Struct(oldSchema).set("field1", value);
        ByteBuffer buffer = ByteBuffer.allocate(oldSchema.sizeOf(oldFormat));
        oldFormat.writeTo(buffer);
        buffer.flip();
        Struct newFormat = newSchema.read(buffer);
        assertEquals(value, newFormat.get("field1"));
        assertEquals("default", newFormat.get("field2"));
        assertNull(newFormat.get("field3"));
        assertEquals(ByteBuffer.allocate(0), newFormat.get("field4"));
        assertEquals(Long.MAX_VALUE, newFormat.get("field5"));
    }

    @Test
    public void testReadWhenOptionalDataMissingAtTheEndIsNotTolerated() {
        Schema oldSchema = new Schema(new Field("field1", Type.NULLABLE_STRING));
        Schema newSchema = new Schema(
                new Field("field1", Type.NULLABLE_STRING),
                new Field("field2", Type.NULLABLE_STRING, "", true, "default"));
        String value = "foo bar baz";
        Struct oldFormat = new Struct(oldSchema).set("field1", value);
        ByteBuffer buffer = ByteBuffer.allocate(oldSchema.sizeOf(oldFormat));
        oldFormat.writeTo(buffer);
        buffer.flip();
        SchemaException e = assertThrows(SchemaException.class, () -> newSchema.read(buffer));
        assertTrue(e.getMessage().contains("Error reading field 'field2':"));
    }

    @Test
    public void testReadWithMissingNonOptionalExtraDataAtTheEnd() {
        Schema oldSchema = new Schema(new Field("field1", Type.NULLABLE_STRING));
        Schema newSchema = new Schema(
                true,
                new Field("field1", Type.NULLABLE_STRING),
                new Field("field2", Type.NULLABLE_STRING));
        String value = "foo bar baz";
        Struct oldFormat = new Struct(oldSchema).set("field1", value);
        ByteBuffer buffer = ByteBuffer.allocate(oldSchema.sizeOf(oldFormat));
        oldFormat.writeTo(buffer);
        buffer.flip();
        SchemaException e = assertThrows(SchemaException.class, () -> newSchema.read(buffer));
        assertTrue(e.getMessage().contains("Missing value for field 'field2' which has no default value"));
    }

    @Test
    public void testReadBytesBeyondItsSize() {
        Type[] types = new Type[]{
            Type.BYTES,
            Type.COMPACT_BYTES,
            Type.NULLABLE_BYTES,
            Type.COMPACT_NULLABLE_BYTES
        };
        for (Type type : types) {
            ByteBuffer buffer = ByteBuffer.allocate(20);
            type.write(buffer, ByteBuffer.allocate(4));
            buffer.rewind();
            ByteBuffer bytes = (ByteBuffer) type.read(buffer);
            assertThrows(IllegalArgumentException.class, () -> bytes.limit(bytes.limit() + 1));
        }
    }
}
