/*
 * 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 com.datastax.driver.core;

import static com.datastax.driver.core.Assertions.assertThat;
import static com.datastax.driver.core.DataType.CollectionType;
import static com.datastax.driver.core.DataType.blob;
import static com.datastax.driver.core.DataType.cint;
import static com.datastax.driver.core.DataType.list;
import static com.datastax.driver.core.DataType.map;
import static com.datastax.driver.core.DataType.set;
import static com.datastax.driver.core.DataType.text;
import static com.datastax.driver.core.DataType.varchar;
import static com.datastax.driver.core.ProtocolVersion.V4;
import static com.datastax.driver.core.TypeTokens.listOf;
import static com.datastax.driver.core.TypeTokens.mapOf;
import static com.datastax.driver.core.TypeTokens.setOf;
import static com.google.common.reflect.TypeToken.of;
import static java.util.Collections.singleton;
import static java.util.Collections.singletonList;
import static java.util.Collections.singletonMap;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import static org.testng.Assert.fail;

import com.datastax.driver.core.exceptions.CodecNotFoundException;
import com.google.common.reflect.TypeToken;
import java.math.BigDecimal;
import java.math.BigInteger;
import java.net.InetAddress;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import org.apache.log4j.Level;
import org.apache.log4j.Logger;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;

public class CodecRegistryTest {

  @DataProvider
  public static Object[][] cql() {
    return new Object[][] {
      {DataType.blob(), TypeCodec.blob()},
      {DataType.cboolean(), TypeCodec.cboolean()},
      {DataType.smallint(), TypeCodec.smallInt()},
      {DataType.tinyint(), TypeCodec.tinyInt()},
      {DataType.cint(), TypeCodec.cint()},
      {DataType.bigint(), TypeCodec.bigint()},
      {DataType.counter(), TypeCodec.counter()},
      {DataType.cdouble(), TypeCodec.cdouble()},
      {DataType.cfloat(), TypeCodec.cfloat()},
      {DataType.varint(), TypeCodec.varint()},
      {DataType.decimal(), TypeCodec.decimal()},
      {DataType.varchar(), TypeCodec.varchar()},
      {DataType.ascii(), TypeCodec.ascii()},
      {DataType.timestamp(), TypeCodec.timestamp()},
      {DataType.date(), TypeCodec.date()},
      {DataType.time(), TypeCodec.time()},
      {DataType.uuid(), TypeCodec.uuid()},
      {DataType.timeuuid(), TypeCodec.timeUUID()},
      {DataType.inet(), TypeCodec.inet()},
      {DataType.duration(), TypeCodec.duration()}
    };
  }

  @DataProvider
  public static Object[][] cqlAndJava() {
    return new Object[][] {
      {DataType.blob(), ByteBuffer.class, TypeCodec.blob()},
      {DataType.cboolean(), Boolean.class, TypeCodec.cboolean()},
      {DataType.smallint(), Short.class, TypeCodec.smallInt()},
      {DataType.tinyint(), Byte.class, TypeCodec.tinyInt()},
      {DataType.cint(), Integer.class, TypeCodec.cint()},
      {DataType.bigint(), Long.class, TypeCodec.bigint()},
      {DataType.counter(), Long.class, TypeCodec.counter()},
      {DataType.cdouble(), Double.class, TypeCodec.cdouble()},
      {DataType.cfloat(), Float.class, TypeCodec.cfloat()},
      {DataType.varint(), BigInteger.class, TypeCodec.varint()},
      {DataType.decimal(), BigDecimal.class, TypeCodec.decimal()},
      {DataType.varchar(), String.class, TypeCodec.varchar()},
      {DataType.ascii(), String.class, TypeCodec.ascii()},
      {DataType.timestamp(), Date.class, TypeCodec.timestamp()},
      {DataType.date(), LocalDate.class, TypeCodec.date()},
      {DataType.time(), Long.class, TypeCodec.time()},
      {DataType.uuid(), UUID.class, TypeCodec.uuid()},
      {DataType.timeuuid(), UUID.class, TypeCodec.timeUUID()},
      {DataType.inet(), InetAddress.class, TypeCodec.inet()},
      {DataType.duration(), Duration.class, TypeCodec.duration()}
    };
  }

  @DataProvider
  public static Object[][] value() {
    return new Object[][] {
      {ByteBuffer.allocate(0), TypeCodec.blob()},
      {Boolean.TRUE, TypeCodec.cboolean()},
      {(short) 42, TypeCodec.smallInt()},
      {(byte) 42, TypeCodec.tinyInt()},
      {42, TypeCodec.cint()},
      {42L, TypeCodec.bigint()},
      {42D, TypeCodec.cdouble()},
      {42F, TypeCodec.cfloat()},
      {new BigInteger("1234"), TypeCodec.varint()},
      {new BigDecimal("123.45"), TypeCodec.decimal()},
      {"foo", TypeCodec.varchar()},
      {new Date(42), TypeCodec.timestamp()},
      {LocalDate.fromDaysSinceEpoch(42), TypeCodec.date()},
      {UUID.randomUUID(), TypeCodec.uuid()},
      {mock(InetAddress.class), TypeCodec.inet()},
      {Duration.from("1mo2d3h"), TypeCodec.duration()}
    };
  }

  @DataProvider
  public static Object[][] cqlAndValue() {
    return new Object[][] {
      {DataType.blob(), ByteBuffer.allocate(0), TypeCodec.blob()},
      {DataType.cboolean(), true, TypeCodec.cboolean()},
      {DataType.smallint(), (short) 42, TypeCodec.smallInt()},
      {DataType.tinyint(), (byte) 42, TypeCodec.tinyInt()},
      {DataType.cint(), 42, TypeCodec.cint()},
      {DataType.bigint(), 42L, TypeCodec.bigint()},
      {DataType.counter(), 42L, TypeCodec.counter()},
      {DataType.cdouble(), 42D, TypeCodec.cdouble()},
      {DataType.cfloat(), 42F, TypeCodec.cfloat()},
      {DataType.varint(), new BigInteger("1234"), TypeCodec.varint()},
      {DataType.decimal(), new BigDecimal("123.45"), TypeCodec.decimal()},
      {DataType.varchar(), "foo", TypeCodec.varchar()},
      {DataType.ascii(), "foo", TypeCodec.ascii()},
      {DataType.timestamp(), new Date(42), TypeCodec.timestamp()},
      {DataType.date(), LocalDate.fromDaysSinceEpoch(42), TypeCodec.date()},
      {DataType.time(), 42L, TypeCodec.time()},
      {DataType.uuid(), UUID.randomUUID(), TypeCodec.uuid()},
      {DataType.timeuuid(), UUID.randomUUID(), TypeCodec.timeUUID()},
      {DataType.inet(), mock(InetAddress.class), TypeCodec.inet()},
      {DataType.duration(), Duration.from("1mo2d3h"), TypeCodec.duration()}
    };
  }

  @Test(groups = "unit", dataProvider = "cql")
  public void should_find_codec_by_cql_type(DataType cqlType, TypeCodec<?> expected) {
    // given
    CodecRegistry registry = new CodecRegistry();
    // when
    TypeCodec<?> actual = registry.codecFor(cqlType);
    // then
    assertThat(actual).isNotNull().accepts(cqlType).isSameAs(expected);
  }

  @Test(groups = "unit", dataProvider = "cqlAndJava")
  public void should_find_codec_by_cql_type_java_type(
      DataType cqlType, Class<?> javaType, TypeCodec<?> expected) {
    // given
    CodecRegistry registry = new CodecRegistry();
    // when
    TypeCodec<?> actual = registry.codecFor(cqlType, javaType);
    // then
    assertThat(actual).isNotNull().accepts(cqlType).accepts(javaType).isSameAs(expected);
  }

  @Test(groups = "unit", dataProvider = "value")
  public void should_find_codec_by_value(Object value, TypeCodec<?> expected) {
    // given
    CodecRegistry registry = new CodecRegistry();
    // when
    TypeCodec<?> actual = registry.codecFor(value);
    // then
    assertThat(actual).isNotNull().accepts(value).isSameAs(expected);
  }

  @Test(groups = "unit", dataProvider = "cqlAndValue")
  public void should_find_codec_by_cql_type_and_value(
      DataType cqlType, Object value, TypeCodec<?> expected) {
    // given
    CodecRegistry registry = new CodecRegistry();
    // when
    TypeCodec<?> actual = registry.codecFor(cqlType, value);
    // then
    assertThat(actual).isNotNull().accepts(cqlType).accepts(value).isSameAs(expected);
  }

  @Test(groups = "unit")
  public void should_find_newly_registered_codec_by_cql_type() {
    // given
    CodecRegistry registry = new CodecRegistry();
    TypeCodec<?> expected = mockCodec(list(text()), listOf(String.class));
    registry.register(expected);
    // when
    TypeCodec<?> actual = registry.codecFor(list(text()));
    // then
    assertThat(actual).isNotNull().isSameAs(expected);
  }

  @Test(groups = "unit")
  public void should_find_default_codec_if_cql_type_already_registered() {
    // given
    CodecRegistry registry = new CodecRegistry();
    TypeCodec<?> newCodec = mockCodec(text(), of(StringBuilder.class));
    registry.register(newCodec);
    // when
    TypeCodec<?> actual = registry.codecFor(text());
    // then
    assertThat(actual)
        .isNotNull()
        .isNotSameAs(newCodec)
        .accepts(text())
        .accepts(String.class)
        .doesNotAccept(StringBuilder.class);
  }

  @Test(groups = "unit")
  public void should_find_newly_registered_codec_by_cql_type_and_java_type() {
    // given
    CodecRegistry registry = new CodecRegistry();
    TypeCodec<?> expected = mockCodec(list(text()), listOf(String.class));
    registry.register(expected);
    // when
    TypeCodec<?> actual = registry.codecFor(list(text()), listOf(String.class));
    // then
    assertThat(actual).isNotNull().isSameAs(expected);
  }

  @Test(groups = "unit")
  public void should_create_list_codec() {
    CollectionType cqlType = list(cint());
    TypeToken<List<Integer>> javaType = listOf(Integer.class);
    assertThat(new CodecRegistry().codecFor(cqlType))
        .isNotNull()
        .accepts(cqlType)
        .accepts(javaType);
    assertThat(new CodecRegistry().codecFor(cqlType, javaType))
        .isNotNull()
        .accepts(cqlType)
        .accepts(javaType);
    assertThat(new CodecRegistry().codecFor(singletonList(42)))
        .isNotNull()
        .accepts(cqlType)
        .accepts(javaType);
    assertThat(new CodecRegistry().codecFor(cqlType, singletonList(42)))
        .isNotNull()
        .accepts(cqlType)
        .accepts(javaType);
    assertThat(new CodecRegistry().codecFor(new ArrayList<Integer>()))
        .isNotNull()
        // empty collections are mapped to blob codec if no CQL type provided
        .accepts(list(blob()))
        .accepts(listOf(ByteBuffer.class));
    assertThat(new CodecRegistry().codecFor(cqlType, new ArrayList<Integer>()))
        .isNotNull()
        .accepts(cqlType)
        .accepts(javaType);
  }

  @Test(groups = "unit")
  public void should_create_set_codec() {
    CollectionType cqlType = set(cint());
    TypeToken<Set<Integer>> javaType = setOf(Integer.class);
    assertThat(new CodecRegistry().codecFor(cqlType))
        .isNotNull()
        .accepts(cqlType)
        .accepts(javaType);
    assertThat(new CodecRegistry().codecFor(cqlType, javaType))
        .isNotNull()
        .accepts(cqlType)
        .accepts(javaType);
    assertThat(new CodecRegistry().codecFor(singleton(42)))
        .isNotNull()
        .accepts(cqlType)
        .accepts(javaType);
    assertThat(new CodecRegistry().codecFor(cqlType, singleton(42)))
        .isNotNull()
        .accepts(cqlType)
        .accepts(javaType);
    assertThat(new CodecRegistry().codecFor(new HashSet<Integer>()))
        .isNotNull()
        // empty collections are mapped to blob codec if no CQL type provided
        .accepts(set(blob()))
        .accepts(setOf(ByteBuffer.class));
    assertThat(new CodecRegistry().codecFor(cqlType, new HashSet<Integer>()))
        .isNotNull()
        .accepts(cqlType)
        .accepts(javaType);
  }

  @Test(groups = "unit")
  public void should_create_map_codec() {
    CollectionType cqlType = map(cint(), list(varchar()));
    TypeToken<Map<Integer, List<String>>> javaType = mapOf(of(Integer.class), listOf(String.class));
    assertThat(new CodecRegistry().codecFor(cqlType))
        .isNotNull()
        .accepts(cqlType)
        .accepts(javaType);
    assertThat(new CodecRegistry().codecFor(cqlType, javaType))
        .isNotNull()
        .accepts(cqlType)
        .accepts(javaType);
    assertThat(new CodecRegistry().codecFor(singletonMap(42, singletonList("foo"))))
        .isNotNull()
        .accepts(cqlType)
        .accepts(javaType);
    assertThat(new CodecRegistry().codecFor(cqlType, singletonMap(42, singletonList("foo"))))
        .isNotNull()
        .accepts(cqlType)
        .accepts(javaType);
    assertThat(new CodecRegistry().codecFor(new HashMap<Integer, List<String>>()))
        .isNotNull()
        // empty collections are mapped to blob codec if no CQL type provided
        .accepts(map(blob(), blob()))
        .accepts(mapOf(ByteBuffer.class, ByteBuffer.class));
    assertThat(new CodecRegistry().codecFor(cqlType, new HashMap<Integer, List<String>>()))
        .isNotNull()
        .accepts(cqlType)
        .accepts(javaType);
  }

  @Test(groups = "unit")
  public void should_create_tuple_codec() {
    CodecRegistry registry = new CodecRegistry();
    TupleType tupleType = TupleType.of(V4, registry, cint(), varchar());
    assertThat(registry.codecFor(tupleType))
        .isNotNull()
        .accepts(tupleType)
        .accepts(TupleValue.class);
    registry = new CodecRegistry();
    tupleType = TupleType.of(V4, registry, cint(), varchar());
    assertThat(registry.codecFor(tupleType, TupleValue.class))
        .isNotNull()
        .accepts(tupleType)
        .accepts(TupleValue.class);
    registry = new CodecRegistry();
    tupleType = TupleType.of(V4, registry, cint(), varchar());
    assertThat(registry.codecFor(new TupleValue(tupleType)))
        .isNotNull()
        .accepts(tupleType)
        .accepts(TupleValue.class);
    assertThat(registry.codecFor(tupleType, new TupleValue(tupleType)))
        .isNotNull()
        .accepts(tupleType)
        .accepts(TupleValue.class);
  }

  @Test(groups = "unit")
  public void should_create_udt_codec() {
    CodecRegistry registry = new CodecRegistry();
    UserType udt =
        new UserType("ks", "test", false, Collections.<UserType.Field>emptyList(), V4, registry);
    assertThat(registry.codecFor(udt)).isNotNull().accepts(udt).accepts(UDTValue.class);
    registry = new CodecRegistry();
    udt = new UserType("ks", "test", false, Collections.<UserType.Field>emptyList(), V4, registry);
    assertThat(registry.codecFor(udt, UDTValue.class))
        .isNotNull()
        .accepts(udt)
        .accepts(UDTValue.class);
    registry = new CodecRegistry();
    udt = new UserType("ks", "test", false, Collections.<UserType.Field>emptyList(), V4, registry);
    assertThat(registry.codecFor(new UDTValue(udt)))
        .isNotNull()
        .accepts(udt)
        .accepts(UDTValue.class);
    registry = new CodecRegistry();
    udt = new UserType("ks", "test", false, Collections.<UserType.Field>emptyList(), V4, registry);
    assertThat(registry.codecFor(udt, new UDTValue(udt)))
        .isNotNull()
        .accepts(udt)
        .accepts(UDTValue.class);
  }

  @Test(groups = "unit")
  public void should_create_codec_for_custom_cql_type() {
    DataType custom = DataType.custom("foo");
    assertThat(new CodecRegistry().codecFor(custom))
        .isNotNull()
        .accepts(custom)
        .accepts(ByteBuffer.class);
    assertThat(new CodecRegistry().codecFor(custom, ByteBuffer.class))
        .isNotNull()
        .accepts(custom)
        .accepts(ByteBuffer.class);
    assertThat(new CodecRegistry().codecFor(custom, ByteBuffer.allocate(0)))
        .isNotNull()
        .accepts(custom)
        .accepts(ByteBuffer.class);
  }

  @Test(groups = "unit")
  public void should_create_derived_codecs_for_java_type_handled_by_custom_codec() {
    TypeCodec<?> newCodec = mockCodec(varchar(), of(StringBuilder.class));
    CodecRegistry registry = new CodecRegistry().register(newCodec);
    // lookup by CQL type only returns default codec
    assertThat(registry.codecFor(list(varchar()))).doesNotAccept(listOf(StringBuilder.class));
    assertThat(registry.codecFor(list(varchar()), listOf(StringBuilder.class))).isNotNull();
  }

  @Test(groups = "unit")
  public void should_not_find_codec_if_java_type_unknown() {
    try {
      new CodecRegistry().codecFor(StringBuilder.class);
      fail("Should not have found a codec for ANY <-> StringBuilder");
    } catch (CodecNotFoundException e) {
      // expected
    }
    try {
      new CodecRegistry().codecFor(varchar(), StringBuilder.class);
      fail("Should not have found a codec for varchar <-> StringBuilder");
    } catch (CodecNotFoundException e) {
      // expected
    }
    try {
      new CodecRegistry().codecFor(new StringBuilder());
      fail("Should not have found a codec for ANY <-> StringBuilder");
    } catch (CodecNotFoundException e) {
      // expected
    }
    try {
      new CodecRegistry().codecFor(varchar(), new StringBuilder());
      fail("Should not have found a codec for varchar <-> StringBuilder");
    } catch (CodecNotFoundException e) {
      // expected
    }
  }

  @Test(groups = "unit")
  public void should_ignore_codec_colliding_with_already_registered_codec() {
    MemoryAppender logs = startCapturingLogs();

    CodecRegistry registry = new CodecRegistry();

    TypeCodec<?> newCodec = mockCodec(cint(), of(Integer.class));

    registry.register(newCodec);

    assertThat(logs.getNext()).contains("Ignoring codec MockCodec");
    assertThat(registry.codecFor(cint(), Integer.class)).isNotSameAs(newCodec);

    stopCapturingLogs(logs);
  }

  @Test(groups = "unit")
  public void should_ignore_codec_colliding_with_already_generated_codec() {
    MemoryAppender logs = startCapturingLogs();

    CodecRegistry registry = new CodecRegistry();

    // Force generation of a list token from the default token
    registry.codecFor(list(cint()), listOf(Integer.class));

    TypeCodec<?> newCodec = mockCodec(list(cint()), listOf(Integer.class));

    registry.register(newCodec);

    assertThat(logs.getNext()).contains("Ignoring codec MockCodec");
    assertThat(registry.codecFor(list(cint()), listOf(Integer.class))).isNotSameAs(newCodec);

    stopCapturingLogs(logs);
  }

  private MemoryAppender startCapturingLogs() {
    Logger registryLogger = Logger.getLogger(CodecRegistry.class);
    registryLogger.setLevel(Level.WARN);
    MemoryAppender logs = new MemoryAppender();
    registryLogger.addAppender(logs);
    return logs;
  }

  private void stopCapturingLogs(MemoryAppender logs) {
    Logger registryLogger = Logger.getLogger(CodecRegistry.class);
    registryLogger.setLevel(null);
    registryLogger.removeAppender(logs);
  }

  private <T> TypeCodec<T> mockCodec(DataType cqlType, TypeToken<T> javaType) {
    @SuppressWarnings("unchecked")
    TypeCodec<T> newCodec = mock(TypeCodec.class);
    when(newCodec.getCqlType()).thenReturn(cqlType);
    when(newCodec.getJavaType()).thenReturn(javaType);
    when(newCodec.accepts(cqlType)).thenReturn(true);
    when(newCodec.accepts(javaType)).thenReturn(true);
    when(newCodec.toString()).thenReturn(String.format("MockCodec [%s <-> %s]", cqlType, javaType));
    return newCodec;
  }
}
