/*
 * 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.serialization;

import org.apache.kafka.common.errors.SerializationException;
import org.apache.kafka.common.utils.Bytes;

import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;

import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Stack;
import java.util.UUID;

import static org.apache.kafka.common.utils.Utils.wrapNullable;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;

public class SerializationTest {

    private final String topic = "testTopic";
    private final Map<Class<?>, List<Object>> testData = new HashMap<>() {
        {
            put(String.class, Arrays.asList(null, "my string"));
            put(Short.class, Arrays.asList(null, (short) 32767, (short) -32768));
            put(Integer.class, Arrays.asList(null, 423412424, -41243432));
            put(Long.class, Arrays.asList(null, 922337203685477580L, -922337203685477581L));
            put(Float.class, Arrays.asList(null, 5678567.12312f, -5678567.12341f));
            put(Double.class, Arrays.asList(null, 5678567.12312d, -5678567.12341d));
            put(byte[].class, Arrays.asList(null, "my string".getBytes()));
            put(ByteBuffer.class, Arrays.asList(
                    null,
                    ByteBuffer.wrap("my string".getBytes()),
                    ByteBuffer.allocate(10).put("my string".getBytes()),
                    ByteBuffer.allocateDirect(10).put("my string".getBytes())));
            put(Bytes.class, Arrays.asList(null, new Bytes("my string".getBytes())));
            put(UUID.class, Arrays.asList(null, UUID.randomUUID()));
        }
    };

    private static class DummyClass {
    }

    @SuppressWarnings("unchecked")
    @Test
    public void allSerdesShouldRoundtripInput() {
        for (Map.Entry<Class<?>, List<Object>> test : testData.entrySet()) {
            try (Serde<Object> serde = Serdes.serdeFrom((Class<Object>) test.getKey())) {
                for (Object value : test.getValue()) {
                    final byte[] serialized = serde.serializer().serialize(topic, value);
                    assertEquals(value, serde.deserializer().deserialize(topic, serialized),
                        "Should get the original " + test.getKey().getSimpleName() + " after serialization and deserialization");

                    if (value instanceof byte[]) {
                        assertArrayEquals((byte[]) value, (byte[]) serde.deserializer().deserialize(topic, null, (byte[]) value),
                                "Should get the original " + test.getKey().getSimpleName() + " after serialization and deserialization");
                    } else {
                        assertEquals(value, serde.deserializer().deserialize(topic, null, wrapNullable(serialized)),
                                "Should get the original " + test.getKey().getSimpleName() + " after serialization and deserialization");
                    }
                }
            }
        }
    }

    @Test
    public void allSerdesShouldSupportNull() {
        for (Class<?> cls : testData.keySet()) {
            try (Serde<?> serde = Serdes.serdeFrom(cls)) {
                assertNull(serde.serializer().serialize(topic, null),
                    "Should support null in " + cls.getSimpleName() + " serialization");
                assertNull(serde.deserializer().deserialize(topic, null),
                    "Should support null in " + cls.getSimpleName() + " deserialization");
                assertNull(serde.deserializer().deserialize(topic, null, (ByteBuffer) null),
                        "Should support null in " + cls.getSimpleName() + " deserialization");
            }
        }
    }

    @Test
    public void testSerdeFromUnknown() {
        assertThrows(IllegalArgumentException.class, () -> Serdes.serdeFrom(DummyClass.class));
    }

    @Test
    public void testSerdeFromNotNull() {
        try (Serde<Long> serde = Serdes.Long()) {
            assertThrows(IllegalArgumentException.class, () -> Serdes.serdeFrom(null, serde.deserializer()));
        }
    }

    @Test
    public void stringSerdeShouldSupportDifferentEncodings() {
        String str = "my string";
        List<String> encodings = Arrays.asList(StandardCharsets.UTF_8.name(), StandardCharsets.UTF_16.name());

        for (String encoding : encodings) {
            try (Serde<String> serDeser = getStringSerde(encoding)) {

                Serializer<String> serializer = serDeser.serializer();
                Deserializer<String> deserializer = serDeser.deserializer();
                assertEquals(str, deserializer.deserialize(topic, serializer.serialize(topic, str)),
                    "Should get the original string after serialization and deserialization with encoding " + encoding);
            }
        }
    }


    @Test
    public void stringSerdeConfigureThrowsOnUnknownEncoding() {
        String encoding = "encoding-does-not-exist";
        try (Serde<String> serDeser = Serdes.String()) {
            Map<String, Object> serializerConfigs = new HashMap<>();
            serializerConfigs.put("key.serializer.encoding", encoding);
            assertThrows(SerializationException.class, () -> serDeser.serializer().configure(serializerConfigs, true));

            Map<String, Object> deserializerConfigs = new HashMap<>();
            deserializerConfigs.put("key.deserializer.encoding", encoding);
            assertThrows(SerializationException.class, () -> serDeser.deserializer().configure(deserializerConfigs, true));
        }
    }

    @SuppressWarnings("unchecked")
    @Test
    public void listSerdeShouldReturnEmptyCollection() {
        List<Integer> testData = Collections.emptyList();
        Serde<List<Integer>> listSerde = Serdes.ListSerde(ArrayList.class, Serdes.Integer());
        assertEquals(testData,
            listSerde.deserializer().deserialize(topic, listSerde.serializer().serialize(topic, testData)),
            "Should get empty collection after serialization and deserialization on an empty list");
    }

    @SuppressWarnings("unchecked")
    @Test
    public void listSerdeShouldReturnNull() {
        List<Integer> testData = null;
        Serde<List<Integer>> listSerde = Serdes.ListSerde(ArrayList.class, Serdes.Integer());
        assertEquals(testData,
            listSerde.deserializer().deserialize(topic, listSerde.serializer().serialize(topic, testData)),
            "Should get null after serialization and deserialization on an empty list");
    }

    @SuppressWarnings("unchecked")
    @Test
    public void listSerdeShouldRoundtripIntPrimitiveInput() {
        List<Integer> testData = Arrays.asList(1, 2, 3);
        Serde<List<Integer>> listSerde = Serdes.ListSerde(ArrayList.class, Serdes.Integer());
        assertEquals(testData,
            listSerde.deserializer().deserialize(topic, listSerde.serializer().serialize(topic, testData)),
            "Should get the original collection of integer primitives after serialization and deserialization");
    }

    @SuppressWarnings("unchecked")
    @Test
    public void listSerdeSerializerShouldReturnByteArrayOfFixedSizeForIntPrimitiveInput() {
        List<Integer> testData = Arrays.asList(1, 2, 3);
        Serde<List<Integer>> listSerde = Serdes.ListSerde(ArrayList.class, Serdes.Integer());
        assertEquals(21, listSerde.serializer().serialize(topic, testData).length,
            "Should get length of 21 bytes after serialization");
    }

    @SuppressWarnings("unchecked")
    @Test
    public void listSerdeShouldRoundtripShortPrimitiveInput() {
        List<Short> testData = Arrays.asList((short) 1, (short) 2, (short) 3);
        Serde<List<Short>> listSerde = Serdes.ListSerde(ArrayList.class, Serdes.Short());
        assertEquals(testData,
            listSerde.deserializer().deserialize(topic, listSerde.serializer().serialize(topic, testData)),
            "Should get the original collection of short primitives after serialization and deserialization");
    }

    @SuppressWarnings("unchecked")
    @Test
    public void listSerdeSerializerShouldReturnByteArrayOfFixedSizeForShortPrimitiveInput() {
        List<Short> testData = Arrays.asList((short) 1, (short) 2, (short) 3);
        Serde<List<Short>> listSerde = Serdes.ListSerde(ArrayList.class, Serdes.Short());
        assertEquals(15, listSerde.serializer().serialize(topic, testData).length,
            "Should get length of 15 bytes after serialization");
    }

    @SuppressWarnings("unchecked")
    @Test
    public void listSerdeShouldRoundtripFloatPrimitiveInput() {
        List<Float> testData = Arrays.asList((float) 1, (float) 2, (float) 3);
        Serde<List<Float>> listSerde = Serdes.ListSerde(ArrayList.class, Serdes.Float());
        assertEquals(testData,
            listSerde.deserializer().deserialize(topic, listSerde.serializer().serialize(topic, testData)),
            "Should get the original collection of float primitives after serialization and deserialization");
    }

    @SuppressWarnings("unchecked")
    @Test
    public void listSerdeSerializerShouldReturnByteArrayOfFixedSizeForFloatPrimitiveInput() {
        List<Float> testData = Arrays.asList((float) 1, (float) 2, (float) 3);
        Serde<List<Float>> listSerde = Serdes.ListSerde(ArrayList.class, Serdes.Float());
        assertEquals(21, listSerde.serializer().serialize(topic, testData).length,
            "Should get length of 21 bytes after serialization");
    }

    @SuppressWarnings("unchecked")
    @Test
    public void listSerdeShouldRoundtripLongPrimitiveInput() {
        List<Long> testData = Arrays.asList((long) 1, (long) 2, (long) 3);
        Serde<List<Long>> listSerde = Serdes.ListSerde(ArrayList.class, Serdes.Long());
        assertEquals(testData,
            listSerde.deserializer().deserialize(topic, listSerde.serializer().serialize(topic, testData)),
            "Should get the original collection of long primitives after serialization and deserialization");
    }

    @SuppressWarnings("unchecked")
    @Test
    public void listSerdeSerializerShouldReturnByteArrayOfFixedSizeForLongPrimitiveInput() {
        List<Long> testData = Arrays.asList((long) 1, (long) 2, (long) 3);
        Serde<List<Long>> listSerde = Serdes.ListSerde(ArrayList.class, Serdes.Long());
        assertEquals(33, listSerde.serializer().serialize(topic, testData).length,
            "Should get length of 33 bytes after serialization");
    }

    @SuppressWarnings("unchecked")
    @Test
    public void listSerdeShouldRoundtripDoublePrimitiveInput() {
        List<Double> testData = Arrays.asList((double) 1, (double) 2, (double) 3);
        Serde<List<Double>> listSerde = Serdes.ListSerde(ArrayList.class, Serdes.Double());
        assertEquals(testData,
            listSerde.deserializer().deserialize(topic, listSerde.serializer().serialize(topic, testData)),
            "Should get the original collection of double primitives after serialization and deserialization");
    }

    @SuppressWarnings("unchecked")
    @Test
    public void listSerdeSerializerShouldReturnByteArrayOfFixedSizeForDoublePrimitiveInput() {
        List<Double> testData = Arrays.asList((double) 1, (double) 2, (double) 3);
        Serde<List<Double>> listSerde = Serdes.ListSerde(ArrayList.class, Serdes.Double());
        assertEquals(33, listSerde.serializer().serialize(topic, testData).length,
            "Should get length of 33 bytes after serialization");
    }

    @SuppressWarnings("unchecked")
    @Test
    public void listSerdeShouldRoundtripUUIDInput() {
        List<UUID> testData = Arrays.asList(UUID.randomUUID(), UUID.randomUUID(), UUID.randomUUID());
        Serde<List<UUID>> listSerde = Serdes.ListSerde(ArrayList.class, Serdes.UUID());
        assertEquals(testData,
            listSerde.deserializer().deserialize(topic, listSerde.serializer().serialize(topic, testData)),
            "Should get the original collection of UUID after serialization and deserialization");
    }

    @SuppressWarnings("unchecked")
    @Test
    public void listSerdeSerializerShouldReturnByteArrayOfFixedSizeForUUIDInput() {
        List<UUID> testData = Arrays.asList(UUID.randomUUID(), UUID.randomUUID(), UUID.randomUUID());
        Serde<List<UUID>> listSerde = Serdes.ListSerde(ArrayList.class, Serdes.UUID());
        assertEquals(117, listSerde.serializer().serialize(topic, testData).length,
            "Should get length of 117 bytes after serialization");
    }

    @SuppressWarnings("unchecked")
    @Test
    public void listSerdeShouldRoundtripNonPrimitiveInput() {
        List<String> testData = Arrays.asList("A", "B", "C");
        Serde<List<String>> listSerde = Serdes.ListSerde(ArrayList.class, Serdes.String());
        assertEquals(testData,
            listSerde.deserializer().deserialize(topic, listSerde.serializer().serialize(topic, testData)),
            "Should get the original collection of strings list after serialization and deserialization");
    }

    @SuppressWarnings("unchecked")
    @Test
    public void listSerdeShouldRoundtripPrimitiveInputWithNullEntries() {
        List<Integer> testData = Arrays.asList(1, null, 3);
        Serde<List<Integer>> listSerde = Serdes.ListSerde(ArrayList.class, Serdes.Integer());
        assertEquals(testData,
            listSerde.deserializer().deserialize(topic, listSerde.serializer().serialize(topic, testData)),
            "Should get the original collection of integer primitives with null entries "
                + "after serialization and deserialization");
    }

    @SuppressWarnings("unchecked")
    @Test
    public void listSerdeShouldRoundtripNonPrimitiveInputWithNullEntries() {
        List<String> testData = Arrays.asList("A", null, "C");
        Serde<List<String>> listSerde = Serdes.ListSerde(ArrayList.class, Serdes.String());
        assertEquals(testData,
            listSerde.deserializer().deserialize(topic, listSerde.serializer().serialize(topic, testData)),
            "Should get the original collection of strings list with null entries "
                + "after serialization and deserialization");
    }

    @SuppressWarnings("unchecked")
    @Test
    public void listSerdeShouldReturnLinkedList() {
        List<Integer> testData = new LinkedList<>();
        Serde<List<Integer>> listSerde = Serdes.ListSerde(LinkedList.class, Serdes.Integer());
        assertInstanceOf(LinkedList.class, listSerde.deserializer().deserialize(topic, listSerde.serializer().serialize(topic, testData)), "Should return List instance of type LinkedList");
    }

    @SuppressWarnings("unchecked")
    @Test
    public void listSerdeShouldReturnStack() {
        List<Integer> testData = new Stack<>();
        Serde<List<Integer>> listSerde = Serdes.ListSerde(Stack.class, Serdes.Integer());
        assertInstanceOf(Stack.class, listSerde.deserializer().deserialize(topic, listSerde.serializer().serialize(topic, testData)), "Should return List instance of type Stack");
    }

    @Test
    public void floatDeserializerShouldThrowSerializationExceptionOnZeroBytes() {
        try (Serde<Float> serde = Serdes.Float()) {
            assertThrows(SerializationException.class, () -> serde.deserializer().deserialize(topic, new byte[0]));
        }
    }

    @Test
    public void floatDeserializerShouldThrowSerializationExceptionOnTooFewBytes() {
        try (Serde<Float> serde = Serdes.Float()) {
            assertThrows(SerializationException.class, () -> serde.deserializer().deserialize(topic, new byte[3]));
        }
    }


    @Test
    public void floatDeserializerShouldThrowSerializationExceptionOnTooManyBytes() {
        try (Serde<Float> serde = Serdes.Float()) {
            assertThrows(SerializationException.class, () -> serde.deserializer().deserialize(topic, new byte[5]));
        }
    }

    @Test
    public void floatSerdeShouldPreserveNaNValues() {
        int someNaNAsIntBits = 0x7f800001;
        float someNaN = Float.intBitsToFloat(someNaNAsIntBits);
        int anotherNaNAsIntBits = 0x7f800002;
        float anotherNaN = Float.intBitsToFloat(anotherNaNAsIntBits);

        try (Serde<Float> serde = Serdes.Float()) {
            // Because of NaN semantics we must assert based on the raw int bits.
            Float roundtrip = serde.deserializer().deserialize(topic,
                    serde.serializer().serialize(topic, someNaN));
            assertEquals(someNaNAsIntBits, Float.floatToRawIntBits(roundtrip));
            Float otherRoundtrip = serde.deserializer().deserialize(topic,
                    serde.serializer().serialize(topic, anotherNaN));
            assertEquals(anotherNaNAsIntBits, Float.floatToRawIntBits(otherRoundtrip));
        }
    }

    @Test
    public void testSerializeVoid() {
        try (Serde<Void> serde = Serdes.Void()) {
            serde.serializer().serialize(topic, null);
        }
    }

    @Test
    public void testDeserializeVoid() {
        try (Serde<Void> serde = Serdes.Void()) {
            serde.deserializer().deserialize(topic, null);
        }
    }

    @Test
    public void voidDeserializerShouldThrowOnNotNullValues() {
        try (Serde<Void> serde = Serdes.Void()) {
            assertThrows(IllegalArgumentException.class, () -> serde.deserializer().deserialize(topic, new byte[5]));
        }
    }

    @Test
    public void stringDeserializerSupportByteBuffer() {
        final String data = "Hello, ByteBuffer!";
        try (Serde<String> serde = Serdes.String()) {
            final Serializer<String> serializer = serde.serializer();
            final Deserializer<String> deserializer = serde.deserializer();
            final byte[] serializedBytes = serializer.serialize(topic, data);
            final ByteBuffer heapBuff = ByteBuffer.allocate(serializedBytes.length << 1).put(serializedBytes);
            heapBuff.flip();
            assertEquals(data, deserializer.deserialize(topic, null, heapBuff));

            final ByteBuffer directBuff = ByteBuffer.allocateDirect(serializedBytes.length << 2).put(serializedBytes);
            directBuff.flip();
            assertEquals(data, deserializer.deserialize(topic, null, directBuff));
        }
    }

    private Serde<String> getStringSerde(String encoder) {
        Map<String, Object> serializerConfigs = new HashMap<>();
        serializerConfigs.put("key.serializer.encoding", encoder);
        Serializer<String> serializer = Serdes.String().serializer();
        serializer.configure(serializerConfigs, true);

        Map<String, Object> deserializerConfigs = new HashMap<>();
        deserializerConfigs.put("key.deserializer.encoding", encoder);
        Deserializer<String> deserializer = Serdes.String().deserializer();
        deserializer.configure(deserializerConfigs, true);

        return Serdes.serdeFrom(serializer, deserializer);
    }

    @ParameterizedTest
    @ValueSource(booleans = { true, false })
    public void testBooleanSerializer(Boolean dataToSerialize) {
        byte[] testData = new byte[1];
        testData[0] = (byte) (dataToSerialize ? 1 : 0);

        Serde<Boolean> booleanSerde = Serdes.Boolean();
        assertArrayEquals(testData, booleanSerde.serializer().serialize(topic, dataToSerialize));
    }

    @ParameterizedTest
    @ValueSource(booleans = { true, false })
    public void testBooleanDeserializer(Boolean dataToDeserialize) {
        byte[] testData = new byte[1];
        testData[0] = (byte) (dataToDeserialize ? 1 : 0);

        Serde<Boolean> booleanSerde = Serdes.Boolean();
        assertEquals(dataToDeserialize, booleanSerde.deserializer().deserialize(topic, testData));
    }

    @Test
    public void booleanDeserializerShouldThrowOnEmptyInput() {
        try (Serde<Boolean> serde = Serdes.Boolean()) {
            assertThrows(SerializationException.class, () -> serde.deserializer().deserialize(topic, new byte[0]));
        }
    }
}
