/*
 * 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.extras.codecs.arrays;

import static org.assertj.core.api.Assertions.assertThat;

import com.datastax.driver.core.BoundStatement;
import com.datastax.driver.core.CCMTestsSupport;
import com.datastax.driver.core.Cluster;
import com.datastax.driver.core.CodecRegistry;
import com.datastax.driver.core.DataType;
import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.Row;
import com.datastax.driver.core.TypeCodec;
import com.datastax.driver.mapping.Mapper;
import com.datastax.driver.mapping.MappingManager;
import com.datastax.driver.mapping.annotations.Column;
import com.datastax.driver.mapping.annotations.PartitionKey;
import com.datastax.driver.mapping.annotations.Table;
import com.google.common.primitives.Primitives;
import java.util.List;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;

public class ArrayCodecsTest extends CCMTestsSupport {

  private static final ObjectArrayCodec<String> stringArrayCodec =
      new ObjectArrayCodec<String>(
          DataType.list(DataType.varchar()), String[].class, TypeCodec.varchar());

  @Override
  public void onTestContextInitialized() {
    execute(
        "CREATE TABLE lists ("
            + "pk int PRIMARY KEY, "
            + "l_int list<int>, "
            + "l_bigint list<bigint>, "
            + "l_float list<float>, "
            + "l_double list<double>, "
            + "l_string list<text> "
            + ")",
        "INSERT INTO lists (pk, l_int, l_bigint, l_float, l_double, l_string) VALUES (1, "
            + "[1, 2, 3], "
            + "[4, 5, 6], "
            + "[1.0, 2.0, 3.0], "
            + "[4.0, 5.0, 6.0], "
            + "['a', 'b', 'c'] "
            + ")");
  }

  @Override
  public Cluster.Builder createClusterBuilder() {
    return super.createClusterBuilder()
        .withCodecRegistry(
            new CodecRegistry()
                .register(IntArrayCodec.instance)
                .register(LongArrayCodec.instance)
                .register(FloatArrayCodec.instance)
                .register(DoubleArrayCodec.instance)
                .register(stringArrayCodec));
  }

  @DataProvider(name = "ArrayCodecsTest-serializing")
  public static Object[][] parametersForSerializationTests() {
    return new Object[][] {
      {"l_int", int[].class, new int[] {1, 2, 3}},
      {"l_bigint", long[].class, new long[] {4, 5, 6}},
      {"l_float", float[].class, new float[] {1, 2, 3}},
      {"l_double", double[].class, new double[] {4, 5, 6}},
      {"l_string", String[].class, new String[] {"a", "b", "c"}}
    };
  }

  @DataProvider(name = "ArrayCodecsTest-formatting")
  public static Object[][] parametersForFormattingTests() {
    return new Object[][] {
      {new IntArrayCodec(), new int[] {1, 2, 3}, new int[0], "[1,2,3]"},
      {new LongArrayCodec(), new long[] {4, 5, 6}, new long[0], "[4,5,6]"},
      {new FloatArrayCodec(), new float[] {1, 2, 3}, new float[0], "[1.0,2.0,3.0]"},
      {new DoubleArrayCodec(), new double[] {4, 5, 6}, new double[0], "[4.0,5.0,6.0]"},
      {stringArrayCodec, new String[] {"a", "b", "c"}, new String[0], "['a','b','c']"}
    };
  }

  @Test(groups = "short", dataProvider = "ArrayCodecsTest-serializing")
  public <A> void should_read_list_column_as_array(
      String columnName, Class<A> arrayClass, A expected) {
    Row row =
        session().execute(String.format("SELECT %s FROM lists WHERE pk = 1", columnName)).one();
    A actual = row.get(columnName, arrayClass);
    assertThat(actual).isEqualTo(expected);
  }

  @Test(groups = "short", dataProvider = "ArrayCodecsTest-serializing")
  public <A> void should_set_list_column_with_array(
      String columnName, Class<A> arrayClass, A expected) {
    PreparedStatement ps =
        session().prepare(String.format("INSERT INTO lists (pk, %s) VALUES (?, ?)", columnName));
    BoundStatement bs = ps.bind().setInt(0, 2).set(columnName, expected, arrayClass);
    session().execute(bs);
    Row row =
        session().execute(String.format("SELECT %s FROM lists WHERE pk = 2", columnName)).one();
    List<?> list = row.getList(columnName, Primitives.wrap(arrayClass.getComponentType()));
    assertThat(list.toArray()).isEqualTo(expected);
  }

  @Test(groups = "unit", dataProvider = "ArrayCodecsTest-formatting")
  public <A> void should_format_array_as_cql_literal(
      TypeCodec<A> codec, A array, A emptyArray, String cql) {
    assertThat(codec.format(array)).isEqualTo(cql);
    assertThat(codec.format(emptyArray)).isEqualTo("[]");
    assertThat(codec.format(null)).isEqualToIgnoringCase("NULL");
  }

  @Test(groups = "unit", dataProvider = "ArrayCodecsTest-formatting")
  public <A> void should_parse_array_from_cql_literal(
      TypeCodec<A> codec, A array, A emptyArray, String cql) {
    assertThat(codec.parse(cql)).isEqualTo(array);
    assertThat(codec.parse("[]")).isEqualTo(emptyArray);
    assertThat(codec.parse("NULL")).isNull();
    assertThat(codec.parse("")).isNull();
    assertThat(codec.parse(null)).isNull();
  }

  @Test(groups = "short")
  public void should_use_mapper_to_store_and_retrieve_values_with_custom_joda_codecs() {
    // given
    MappingManager manager = new MappingManager(session());
    Mapper<Mapped> mapper = manager.mapper(Mapped.class);
    // when
    Mapped pojo = new Mapped();
    mapper.save(pojo);
    Mapped actual = mapper.get(42);
    // then
    assertThat(actual).isEqualToComparingFieldByField(pojo);
  }

  @SuppressWarnings("unused")
  @Table(name = "lists")
  public static class Mapped {

    @PartitionKey private int pk;

    @Column(name = "l_string")
    private String[] strings;

    @Column(name = "l_int")
    private int[] ints;

    @Column(name = "l_bigint")
    private long[] longs;

    @Column(name = "l_float")
    private float[] floats;

    @Column(name = "l_double")
    private double[] doubles;

    public Mapped() {
      pk = 42;
      strings = new String[] {"a", "b", "c"};
      ints = new int[] {1, 2, 3};
      longs = new long[] {4, 5, 6};
      floats = new float[] {1, 2, 3};
      doubles = new double[] {4, 5, 6};
    }

    public int getPk() {
      return pk;
    }

    public void setPk(int pk) {
      this.pk = pk;
    }

    public String[] getStrings() {
      return strings;
    }

    public void setStrings(String[] strings) {
      this.strings = strings;
    }

    public int[] getInts() {
      return ints;
    }

    public void setInts(int[] ints) {
      this.ints = ints;
    }

    public long[] getLongs() {
      return longs;
    }

    public void setLongs(long[] longs) {
      this.longs = longs;
    }

    public float[] getFloats() {
      return floats;
    }

    public void setFloats(float[] floats) {
      this.floats = floats;
    }

    public double[] getDoubles() {
      return doubles;
    }

    public void setDoubles(double[] doubles) {
      this.doubles = doubles;
    }
  }
}
