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

import static com.datastax.driver.core.TypeCodec.bigint;
import static com.datastax.driver.core.TypeCodec.decimal;
import static com.datastax.driver.core.TypeCodec.list;
import static com.datastax.driver.core.TypeCodec.varchar;
import static com.datastax.driver.core.querybuilder.QueryBuilder.bindMarker;
import static com.datastax.driver.core.querybuilder.QueryBuilder.eq;
import static com.datastax.driver.core.querybuilder.QueryBuilder.insertInto;
import static com.datastax.driver.core.querybuilder.QueryBuilder.select;
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.PreparedStatement;
import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.Row;
import com.datastax.driver.core.querybuilder.BuiltStatement;
import com.datastax.driver.core.utils.CassandraVersion;
import com.google.common.base.Optional;
import com.google.common.collect.Lists;
import java.math.BigDecimal;
import java.util.List;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;

public class OptionalCodecTest extends CCMTestsSupport {

  private final OptionalCodec<List<String>> optionalCodec =
      new OptionalCodec<List<String>>(list(varchar()));

  private final CodecRegistry registry = new CodecRegistry().register(optionalCodec);

  private BuiltStatement insertStmt;

  private BuiltStatement selectStmt;

  @Override
  public void onTestContextInitialized() {
    execute(
        "CREATE TABLE foo (c1 text, c2 text, c3 list<text>, c4 bigint, c5 decimal, PRIMARY KEY (c1, c2))");
  }

  @Override
  public Cluster.Builder createClusterBuilder() {
    return Cluster.builder().withCodecRegistry(registry);
  }

  @BeforeMethod(groups = "short")
  public void createBuiltStatements() throws Exception {
    insertStmt =
        insertInto("foo")
            .value("c1", bindMarker())
            .value("c2", bindMarker())
            .value("c3", bindMarker());
    selectStmt = select("c2", "c3").from("foo").where(eq("c1", bindMarker()));
  }

  /**
   * Validates that if a column is unset, that retrieving the value using {@link OptionalCodec}
   * should return an {@link Optional#absent()} value. Since CQL Lists can't differentiate between
   * null and empty lists, the OptionalCodec should be smart enough to map an empty list to absent.
   *
   * @test_category data_types:serialization
   * @expected_result an absent value.
   * @jira_ticket JAVA-846
   * @since 2.2.0
   */
  @Test(groups = "short")
  @CassandraVersion("2.2.0")
  public void should_map_unset_value_to_absent() {
    PreparedStatement insertPrep = session().prepare(this.insertStmt);
    PreparedStatement selectPrep = session().prepare(this.selectStmt);

    BoundStatement bs = insertPrep.bind();
    bs.setString(0, "should_map_unset_value_to_absent");
    bs.setString(1, "1");
    session().execute(bs);

    ResultSet results = session().execute(selectPrep.bind("should_map_unset_value_to_absent"));
    assertThat(results.getAvailableWithoutFetching()).isEqualTo(1);

    Row row = results.one();
    assertThat(row.getString("c2")).isEqualTo("1");
    assertThat(row.get("c3", optionalCodec.getJavaType())).isEqualTo(Optional.absent());
  }

  /**
   * Validates that if a column is set to {@link Optional#absent()} using {@link OptionalCodec} that
   * it should be stored as null and that retrieving it should return {@link Optional#absent()}
   * using {@link OptionalCodec}.
   *
   * @test_category data_types:serialization
   * @expected_result an absent value.
   * @jira_ticket JAVA-846
   * @since 2.2.0
   */
  @Test(groups = "short")
  public void should_map_absent_null_value_to_absent() {
    PreparedStatement insertPrep = session().prepare(this.insertStmt);
    PreparedStatement selectPrep = session().prepare(this.selectStmt);

    BoundStatement bs = insertPrep.bind();
    bs.setString(0, "should_map_absent_null_value_to_absent");
    bs.setString(1, "1");
    bs.set(2, Optional.<List<String>>absent(), optionalCodec.getJavaType());
    session().execute(bs);

    ResultSet results =
        session().execute(selectPrep.bind("should_map_absent_null_value_to_absent"));
    assertThat(results.getAvailableWithoutFetching()).isEqualTo(1);

    Row row = results.one();
    assertThat(row.getString("c2")).isEqualTo("1");
    assertThat(row.getList("c3", String.class)).isEmpty();
    assertThat(row.get("c3", optionalCodec.getJavaType())).isEqualTo(Optional.absent());
  }

  /**
   * Validates that if a column is set to an {@link Optional} value using {@link OptionalCodec} that
   * it should be stored as the option's value and that retrieving it should return an {@link
   * Optional} using {@link OptionalCodec} and its actual value without using it.
   *
   * @test_category data_types:serialization
   * @expected_result The options value is stored appropriately and is retrievable with and without
   *     OptionalCodec.
   * @jira_ticket JAVA-846
   * @since 2.2.0
   */
  @Test(groups = "short")
  public void should_map_some_back_to_itself() {
    PreparedStatement insertPrep = session().prepare(this.insertStmt);
    PreparedStatement selectPrep = session().prepare(this.selectStmt);

    List<String> data = Lists.newArrayList("1", "2", "3");

    BoundStatement bs = insertPrep.bind();
    bs.setString(0, "should_map_some_back_to_itself");
    bs.setString(1, "1");
    bs.set(2, Optional.of(data), optionalCodec.getJavaType());
    session().execute(bs);

    ResultSet results = session().execute(selectPrep.bind("should_map_some_back_to_itself"));
    assertThat(results.getAvailableWithoutFetching()).isEqualTo(1);

    Row row = results.one();
    assertThat(row.getString("c2")).isEqualTo("1");

    // Ensure data stored correctly.
    assertThat(row.getList("c3", String.class)).isEqualTo(data);

    // Ensure data retrievable using Option codec.
    Optional<List<String>> returnData = row.get("c3", optionalCodec.getJavaType());
    assertThat(returnData.isPresent()).isTrue();
    assertThat(returnData.get()).isEqualTo(data);
  }

  @Test(groups = "short")
  public void should_map_a_primitive_type_to_absent() {
    OptionalCodec<Long> optionalLongCodec = new OptionalCodec<Long>(bigint());
    cluster().getConfiguration().getCodecRegistry().register(optionalLongCodec);

    PreparedStatement stmt = session().prepare("insert into foo (c1, c2, c4) values (?,?,?)");

    BoundStatement bs = stmt.bind();
    bs.setString(0, "should_map_a_primitive_type_to_absent");
    bs.setString(1, "1");
    bs.set(2, Optional.<Long>absent(), optionalLongCodec.getJavaType());
    session().execute(bs);

    PreparedStatement selectBigint = session().prepare("select c1, c4 from foo where c1=?");
    ResultSet results =
        session().execute(selectBigint.bind("should_map_a_primitive_type_to_absent"));

    assertThat(results.getAvailableWithoutFetching()).isEqualTo(1);

    Row row = results.one();
    assertThat(row.get("c4", optionalLongCodec.getJavaType())).isEqualTo(Optional.<Long>absent());
    assertThat(row.getLong("c4"))
        .isEqualTo(0L); // This will return a 0L since it returns the primitive value.
    assertThat(row.get("c4", Long.class)).isNull();
  }

  @Test(groups = "short")
  public void should_map_a_nullable_type_to_absent() {
    OptionalCodec<BigDecimal> optionalDecimalCodec = new OptionalCodec<BigDecimal>(decimal());
    cluster().getConfiguration().getCodecRegistry().register(optionalDecimalCodec);

    PreparedStatement stmt = session().prepare("insert into foo (c1, c2, c5) values (?,?,?)");

    BoundStatement bs = stmt.bind();
    bs.setString(0, "should_map_a_nullable_type_to_absent");
    bs.setString(1, "1");
    bs.set(2, Optional.<BigDecimal>absent(), optionalDecimalCodec.getJavaType());
    session().execute(bs);

    PreparedStatement selectDecimal = session().prepare("select c1, c5 from foo where c1=?");
    ResultSet results =
        session().execute(selectDecimal.bind("should_map_a_nullable_type_to_absent"));

    assertThat(results.getAvailableWithoutFetching()).isEqualTo(1);

    Row row = results.one();
    assertThat(row.get("c5", optionalDecimalCodec.getJavaType()))
        .isEqualTo(Optional.<BigDecimal>absent());
    assertThat(row.getDecimal("c5"))
        .isNull(); // Since BigDecimal is not a primitive it is nullable so expect null.
  }
}
