/*
 * 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.DataType.cfloat;
import static org.assertj.core.api.Assertions.assertThat;

import com.datastax.driver.core.utils.CassandraVersion;
import java.util.UUID;
import org.testng.annotations.Test;

@CassandraVersion("2.1.0")
public class TypeCodecTupleIntegrationTest extends CCMTestsSupport {

  private final String insertQuery = "INSERT INTO users (id, name, location) VALUES (?, ?, ?)";
  private final String selectQuery = "SELECT id, name, location FROM users WHERE id = ?";

  private final UUID uuid = UUID.randomUUID();

  private TupleType locationType;
  private TupleValue locationValue;
  private TupleValue partialLocationValueInserted;
  private Location location;
  private Location partialLocation;
  private TupleValue partialLocationValueRetrieved;

  @Override
  public void onTestContextInitialized() {
    execute(
        "CREATE TABLE IF NOT EXISTS \"users\" (id uuid PRIMARY KEY, name text, location frozen<tuple<float,float>>)");
  }

  @Test(groups = "short")
  public void should_handle_tuples_with_default_codecs() {
    setUpTupleTypes(cluster());
    // simple statement
    session().execute(insertQuery, uuid, "John Doe", locationValue);
    ResultSet rows = session().execute(selectQuery, uuid);
    Row row = rows.one();
    assertRow(row);
    // prepared + values
    PreparedStatement ps = session().prepare(insertQuery);
    session().execute(ps.bind(uuid, "John Doe", locationValue));
    rows = session().execute(selectQuery, uuid);
    row = rows.one();
    assertRow(row);
    // bound with setTupleValue
    session()
        .execute(
            ps.bind()
                .setUUID(0, uuid)
                .setString(1, "John Doe")
                .setTupleValue("location", locationValue));
    rows = session().execute(selectQuery, uuid);
    row = rows.one();
    assertRow(row);
  }

  @Test(groups = "short")
  public void should_handle_partial_tuples_with_default_codecs() {
    setUpTupleTypes(cluster());
    // simple statement
    session().execute(insertQuery, uuid, "John Doe", partialLocationValueInserted);
    ResultSet rows = session().execute(selectQuery, uuid);
    Row row = rows.one();
    assertPartialRow(row);
    // prepared + values
    PreparedStatement ps = session().prepare(insertQuery);
    session().execute(ps.bind(uuid, "John Doe", partialLocationValueInserted));
    rows = session().execute(selectQuery, uuid);
    row = rows.one();
    assertPartialRow(row);
    // bound with setTupleValue
    session()
        .execute(
            ps.bind()
                .setUUID(0, uuid)
                .setString(1, "John Doe")
                .setTupleValue("location", partialLocationValueInserted));
    rows = session().execute(selectQuery, uuid);
    row = rows.one();
    assertPartialRow(row);
  }

  @Test(groups = "short")
  public void should_handle_tuples_with_custom_codecs() {
    Cluster cluster = register(createClusterBuilder().build());
    Session session = cluster.connect(keyspace);
    setUpTupleTypes(cluster);
    cluster
        .getConfiguration()
        .getCodecRegistry()
        .register(new LocationCodec(TypeCodec.tuple(locationType)));
    session.execute(insertQuery, uuid, "John Doe", locationValue);
    ResultSet rows = session.execute(selectQuery, uuid);
    Row row = rows.one();
    assertThat(row.getUUID(0)).isEqualTo(uuid);
    assertThat(row.getObject(0)).isEqualTo(uuid);
    assertThat(row.get(0, UUID.class)).isEqualTo(uuid);
    assertThat(row.getString(1)).isEqualTo("John Doe");
    assertThat(row.getObject(1)).isEqualTo("John Doe");
    assertThat(row.get(1, String.class)).isEqualTo("John Doe");
    assertThat(row.getTupleValue(2)).isEqualTo(locationValue);
    // edge case: getObject should use default codecs;
    // but tuple and udt codecs are registered on the fly;
    // so if we have another manually-registered codec
    // that one will be picked up
    assertThat(row.getObject(2)).isEqualTo(location);
    assertThat(row.get(2, TupleValue.class)).isEqualTo(locationValue);
    assertThat(row.get(2, Location.class)).isEqualTo(location);
  }

  @Test(groups = "short")
  public void should_handle_partial_tuples_with_custom_codecs() {
    Cluster cluster = register(createClusterBuilder().build());
    Session session = cluster.connect(keyspace);
    setUpTupleTypes(cluster);
    cluster
        .getConfiguration()
        .getCodecRegistry()
        .register(new LocationCodec(TypeCodec.tuple(locationType)));
    session.execute(insertQuery, uuid, "John Doe", partialLocationValueInserted);
    ResultSet rows = session.execute(selectQuery, uuid);
    Row row = rows.one();
    assertThat(row.getUUID(0)).isEqualTo(uuid);
    assertThat(row.getObject(0)).isEqualTo(uuid);
    assertThat(row.get(0, UUID.class)).isEqualTo(uuid);
    assertThat(row.getString(1)).isEqualTo("John Doe");
    assertThat(row.getObject(1)).isEqualTo("John Doe");
    assertThat(row.get(1, String.class)).isEqualTo("John Doe");
    assertThat(row.getTupleValue(2)).isEqualTo(locationType.newValue(37.387224f, null));
    // corner case: getObject should use default codecs;
    // but tuple and udt codecs are registered on the fly;
    // so if we have another manually-registered codec
    // that one will be picked up :(
    assertThat(row.getObject(2)).isEqualTo(partialLocation);
    assertThat(row.get(2, TupleValue.class)).isEqualTo(locationType.newValue(37.387224f, null));
    assertThat(row.get(2, Location.class)).isEqualTo(partialLocation);
  }

  private void assertRow(Row row) {
    assertThat(row.getUUID(0)).isEqualTo(uuid);
    assertThat(row.getObject(0)).isEqualTo(uuid);
    assertThat(row.get(0, UUID.class)).isEqualTo(uuid);
    assertThat(row.getString(1)).isEqualTo("John Doe");
    assertThat(row.getObject(1)).isEqualTo("John Doe");
    assertThat(row.get(1, String.class)).isEqualTo("John Doe");
    assertThat(row.getTupleValue(2)).isEqualTo(locationValue);
    assertThat(row.getObject(2)).isEqualTo(locationValue);
    assertThat(row.get(2, TupleValue.class)).isEqualTo(locationValue);
  }

  private void assertPartialRow(Row row) {
    assertThat(row.getUUID(0)).isEqualTo(uuid);
    assertThat(row.getObject(0)).isEqualTo(uuid);
    assertThat(row.get(0, UUID.class)).isEqualTo(uuid);
    assertThat(row.getString(1)).isEqualTo("John Doe");
    assertThat(row.getObject(1)).isEqualTo("John Doe");
    assertThat(row.get(1, String.class)).isEqualTo("John Doe");
    assertThat(row.getTupleValue(2)).isEqualTo(partialLocationValueRetrieved);
    assertThat(row.getObject(2)).isEqualTo(partialLocationValueRetrieved);
    assertThat(row.get(2, TupleValue.class)).isEqualTo(partialLocationValueRetrieved);
  }

  private void setUpTupleTypes(Cluster cluster) {
    locationType = cluster.getMetadata().newTupleType(cfloat(), cfloat());
    locationValue = locationType.newValue().setFloat(0, 37.387224f).setFloat(1, -121.9733837f);
    // insert a tuple of a different dimension
    partialLocationValueInserted =
        cluster.getMetadata().newTupleType(cfloat()).newValue().setFloat(0, 37.387224f);
    // retrieve the partial tuple with null missing values
    partialLocationValueRetrieved = locationType.newValue(37.387224f, null);
    location = new Location(37.387224f, -121.9733837f);
    partialLocation = new Location(37.387224f, 0.0f);
  }

  static class LocationCodec extends MappingCodec<Location, TupleValue> {

    private final TupleType tupleType;

    public LocationCodec(TypeCodec<TupleValue> innerCodec) {
      super(innerCodec, Location.class);
      tupleType = (TupleType) innerCodec.getCqlType();
    }

    @Override
    protected Location deserialize(TupleValue value) {
      return value == null ? null : new Location(value.getFloat(0), value.getFloat(1));
    }

    @Override
    protected TupleValue serialize(Location value) {
      return value == null
          ? null
          : tupleType.newValue().setFloat(0, value.latitude).setFloat(1, value.longitude);
    }
  }

  static class Location {

    float latitude;

    float longitude;

    public Location(float latitude, float longitude) {
      this.latitude = latitude;
      this.longitude = longitude;
    }

    @Override
    public boolean equals(Object o) {
      if (this == o) return true;
      if (o == null || getClass() != o.getClass()) return false;

      Location location = (Location) o;

      return Float.compare(location.latitude, latitude) == 0
          && Float.compare(location.longitude, longitude) == 0;
    }

    @Override
    public int hashCode() {
      int result = (latitude != +0.0f ? Float.floatToIntBits(latitude) : 0);
      result = 31 * result + (longitude != +0.0f ? Float.floatToIntBits(longitude) : 0);
      return result;
    }

    @Override
    public String toString() {
      return "[" + latitude + ", " + longitude + "]";
    }
  }
}
