/*
 * 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.google.common.collect.Lists.newArrayList;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.fail;

import com.datastax.driver.core.exceptions.InvalidTypeException;
import com.datastax.driver.core.utils.Bytes;
import java.math.BigDecimal;
import java.math.BigInteger;
import java.net.InetAddress;
import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.Date;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import org.testng.annotations.Test;

/** DataType simple unit tests. */
public class DataTypeTest {

  CodecRegistry codecRegistry = new CodecRegistry();

  ProtocolVersion protocolVersion = ProtocolVersion.NEWEST_SUPPORTED;

  static boolean exclude(DataType t) {
    return t.getName() == DataType.Name.COUNTER || t.getName() == DataType.Name.DURATION;
  }

  /** A test value for a primitive data type */
  static class TestValue {
    /** The value as a Java object */
    final Object javaObject;
    /** A CQL string that should parse to the value */
    final String cqlInputString;
    /** How the value should be formatted in CQL */
    final String cqlOutputString;

    TestValue(Object javaObject, String cqlInputString, String cqlOutputString) {
      this.javaObject = javaObject;
      this.cqlInputString = cqlInputString;
      this.cqlOutputString = cqlOutputString;
    }
  }

  private static TestValue[] primitiveTestValues(DataType dt) {
    switch (dt.getName()) {
      case ASCII:
      case TEXT:
      case VARCHAR:
        return new TestValue[] {
          new TestValue("foo", "'foo'", "'foo'"),
          new TestValue("fo'o", "'fo''o'", "'fo''o'"),
          new TestValue(null, null, "NULL"),
          new TestValue(null, "null", "NULL"),
          new TestValue(null, "NULL", "NULL")
        };
      case BIGINT:
        return new TestValue[] {
          new TestValue(42L, "42", "42"),
          new TestValue(91294377723L, "91294377723", "91294377723"),
          new TestValue(-133L, "-133", "-133"),
          new TestValue(null, null, "NULL"),
          new TestValue(null, "null", "NULL"),
          new TestValue(null, "NULL", "NULL")
        };
      case TIMESTAMP:
        // input: single quotes are optional for long literals, mandatory for date patterns
        return new TestValue[] {
          new TestValue(new Date(42L), "42", "42"),
          new TestValue(new Date(91294377723L), "91294377723", "91294377723"),
          new TestValue(new Date(-133L), "-133", "-133"),
          new TestValue(new Date(784041330999L), "'1994-11-05T14:15:30.999+0100'", "784041330999"),
          new TestValue(null, null, "NULL"),
          new TestValue(null, "null", "NULL"),
          new TestValue(null, "NULL", "NULL")
        };
      case DATE:
        // input: single quotes are optional for long literals, mandatory for date patterns
        return new TestValue[] {
          new TestValue(LocalDate.fromDaysSinceEpoch(16071), "'2014-01-01'", "'2014-01-01'"),
          new TestValue(LocalDate.fromDaysSinceEpoch(0), "'1970-01-01'", "'1970-01-01'"),
          new TestValue(
              LocalDate.fromDaysSinceEpoch((int) (2147483648L - (1L << 31))),
              "'2147483648'",
              "'1970-01-01'"),
          new TestValue(
              LocalDate.fromDaysSinceEpoch((int) (0 - (1L << 31))), "0", "'-5877641-06-23'"),
          new TestValue(null, null, "NULL"),
          new TestValue(null, "null", "NULL"),
          new TestValue(null, "NULL", "NULL")
        };
      case TIME:
        // input: all literals must by enclosed in single quotes
        return new TestValue[] {
          new TestValue(54012123450000L, "'54012123450000'", "'15:00:12.123450000'"),
          new TestValue(0L, "'0'", "'00:00:00.000000000'"),
          new TestValue(54012012345000L, "'15:00:12.012345000'", "'15:00:12.012345000'"),
          new TestValue(null, null, "NULL"),
          new TestValue(null, "null", "NULL"),
          new TestValue(null, "NULL", "NULL")
        };
      case BLOB:
        return new TestValue[] {
          new TestValue(Bytes.fromHexString("0x2450"), "0x2450", "0x2450"),
          new TestValue(ByteBuffer.allocate(0), "0x", "0x"),
          new TestValue(null, null, "NULL"),
          new TestValue(null, "null", "NULL"),
          new TestValue(null, "NULL", "NULL")
        };
      case BOOLEAN:
        return new TestValue[] {
          new TestValue(true, "true", "true"),
          new TestValue(false, "false", "false"),
          new TestValue(null, null, "NULL"),
          new TestValue(null, "null", "NULL"),
          new TestValue(null, "NULL", "NULL")
        };
      case DECIMAL:
        return new TestValue[] {
          new TestValue(new BigDecimal("1.23E+8"), "1.23E+8", "1.23E+8"),
          new TestValue(null, null, "NULL"),
          new TestValue(null, "null", "NULL"),
          new TestValue(null, "NULL", "NULL")
        };
      case DOUBLE:
        return new TestValue[] {
          new TestValue(2.39324324, "2.39324324", "2.39324324"),
          new TestValue(-12., "-12.0", "-12.0"),
          new TestValue(null, null, "NULL"),
          new TestValue(null, "null", "NULL"),
          new TestValue(null, "NULL", "NULL")
        };
      case FLOAT:
        return new TestValue[] {
          new TestValue(2.39f, "2.39", "2.39"),
          new TestValue(-12.f, "-12.0", "-12.0"),
          new TestValue(null, null, "NULL"),
          new TestValue(null, "null", "NULL"),
          new TestValue(null, "NULL", "NULL")
        };
      case INET:
        try {
          return new TestValue[] {
            new TestValue(InetAddress.getByName("128.2.12.3"), "'128.2.12.3'", "'128.2.12.3'"),
            new TestValue(null, null, "NULL"),
            new TestValue(null, "null", "NULL"),
            new TestValue(null, "NULL", "NULL")
          };
        } catch (java.net.UnknownHostException e) {
          throw new RuntimeException();
        }
      case TINYINT:
        return new TestValue[] {
          new TestValue((byte) -4, "-4", "-4"),
          new TestValue((byte) 44, "44", "44"),
          new TestValue(null, null, "NULL"),
          new TestValue(null, "null", "NULL"),
          new TestValue(null, "NULL", "NULL")
        };
      case SMALLINT:
        return new TestValue[] {
          new TestValue((short) -3, "-3", "-3"),
          new TestValue((short) 43, "43", "43"),
          new TestValue(null, null, "NULL"),
          new TestValue(null, "null", "NULL"),
          new TestValue(null, "NULL", "NULL")
        };
      case INT:
        return new TestValue[] {
          new TestValue(-2, "-2", "-2"),
          new TestValue(42, "42", "42"),
          new TestValue(null, null, "NULL"),
          new TestValue(null, "null", "NULL"),
          new TestValue(null, "NULL", "NULL")
        };
      case TIMEUUID:
        return new TestValue[] {
          new TestValue(
              UUID.fromString("FE2B4360-28C6-11E2-81C1-0800200C9A66"),
              "fe2b4360-28c6-11e2-81c1-0800200c9a66",
              "fe2b4360-28c6-11e2-81c1-0800200c9a66"),
          new TestValue(null, null, "NULL"),
          new TestValue(null, "null", "NULL"),
          new TestValue(null, "NULL", "NULL")
        };
      case UUID:
        return new TestValue[] {
          new TestValue(
              UUID.fromString("FE2B4360-28C6-11E2-81C1-0800200C9A66"),
              "fe2b4360-28c6-11e2-81c1-0800200c9a66",
              "fe2b4360-28c6-11e2-81c1-0800200c9a66"),
          new TestValue(
              UUID.fromString("067e6162-3b6f-4ae2-a171-2470b63dff00"),
              "067e6162-3b6f-4ae2-a171-2470b63dff00",
              "067e6162-3b6f-4ae2-a171-2470b63dff00"),
          new TestValue(null, null, "NULL"),
          new TestValue(null, "null", "NULL"),
          new TestValue(null, "NULL", "NULL")
        };
      case VARINT:
        return new TestValue[] {
          new TestValue(
              new BigInteger("12387290982347987032483422342432"),
              "12387290982347987032483422342432",
              "12387290982347987032483422342432"),
          new TestValue(null, null, "NULL"),
          new TestValue(null, "null", "NULL"),
          new TestValue(null, "NULL", "NULL")
        };
      default:
        throw new RuntimeException("Missing handling of " + dt);
    }
  }

  @Test(groups = "unit")
  public void parseNativeTest() {
    for (DataType dt : DataType.allPrimitiveTypes()) {
      if (exclude(dt)) continue;

      for (TestValue value : primitiveTestValues(dt))
        assertThat(codecRegistry.codecFor(dt).parse(value.cqlInputString))
            .as("Parsing input %s to a %s", value.cqlInputString, dt)
            .isEqualTo(value.javaObject);
    }
  }

  @Test(groups = "unit")
  public void formatNativeTest() {
    for (DataType dt : DataType.allPrimitiveTypes()) {
      if (exclude(dt)) continue;

      for (TestValue value : primitiveTestValues(dt))
        assertThat(codecRegistry.codecFor(dt).format(value.javaObject))
            .as("Formatting a %s expecting %s", dt, value.cqlOutputString)
            .isEqualTo(value.cqlOutputString);
    }
  }

  @Test(groups = "unit")
  public void parseFormatListTest() {
    String toParse = "['Foo','Bar','Foo''bar']";
    List<String> toFormat = Arrays.asList("Foo", "Bar", "Foo'bar");
    DataType dt = DataType.list(DataType.text());
    assertEquals(codecRegistry.codecFor(dt).parse(toParse), toFormat);
    assertEquals(codecRegistry.codecFor(dt).format(toFormat), toParse);
  }

  @SuppressWarnings("serial")
  @Test(groups = "unit")
  public void parseFormatSetTest() {
    String toParse = "{'Foo','Bar','Foo''bar'}";
    Set<String> toFormat =
        new LinkedHashSet<String>() {
          {
            add("Foo");
            add("Bar");
            add("Foo'bar");
          }
        };
    DataType dt = DataType.set(DataType.text());
    assertEquals(codecRegistry.codecFor(dt).parse(toParse), toFormat);
    assertEquals(codecRegistry.codecFor(dt).format(toFormat), toParse);
  }

  @SuppressWarnings("serial")
  @Test(groups = "unit")
  public void parseFormatMapTest() {
    String toParse = "{'Foo':3,'Bar':42,'Foo''bar':-24}";
    Map<String, Integer> toFormat =
        new LinkedHashMap<String, Integer>() {
          {
            put("Foo", 3);
            put("Bar", 42);
            put("Foo'bar", -24);
          }
        };
    DataType dt = DataType.map(DataType.text(), DataType.cint());
    assertEquals(codecRegistry.codecFor(dt).parse(toParse), toFormat);
    assertEquals(codecRegistry.codecFor(dt).format(toFormat), toParse);
  }

  @SuppressWarnings("serial")
  @Test(groups = "unit")
  public void parseFormatUDTTest() {
    String toParse = "{t:'fo''o',i:3,\"L\":['a','b'],s:{3:{a:0x01}}}";

    final UserType udt1 =
        new UserType(
            "ks",
            "t",
            false,
            Arrays.asList(new UserType.Field("a", DataType.blob())),
            protocolVersion,
            codecRegistry);
    UserType udt2 =
        new UserType(
            "ks",
            "t",
            false,
            Arrays.asList(
                new UserType.Field("t", DataType.text()),
                new UserType.Field("i", DataType.cint()),
                new UserType.Field("L", DataType.list(DataType.text())),
                new UserType.Field("s", DataType.map(DataType.cint(), udt1))),
            protocolVersion,
            codecRegistry);

    UDTValue toFormat = udt2.newValue();
    toFormat.setString("t", "fo'o");
    toFormat.setInt("i", 3);
    toFormat.setList("\"L\"", Arrays.<String>asList("a", "b"));
    toFormat.setMap(
        "s",
        new HashMap<Integer, UDTValue>() {
          {
            put(3, udt1.newValue().setBytes("a", ByteBuffer.wrap(new byte[] {1})));
          }
        });

    assertEquals(codecRegistry.codecFor(udt2).parse(toParse), toFormat);
    assertEquals(codecRegistry.codecFor(udt2).format(toFormat), toParse);
  }

  @SuppressWarnings("deprecation")
  @Test(groups = "unit")
  public void parseFormatTupleTest() {

    String toParse = "(1,'foo',1.0)";
    TupleType t =
        new TupleType(
            newArrayList(DataType.cint(), DataType.text(), DataType.cfloat()),
            protocolVersion,
            codecRegistry);
    TupleValue toFormat = t.newValue(1, "foo", 1.0f);

    assertEquals(codecRegistry.codecFor(t).parse(toParse), toFormat);
    assertEquals(codecRegistry.codecFor(t).format(toFormat), toParse);
  }

  @Test(groups = "unit")
  public void serializeDeserializeTest() {
    for (ProtocolVersion v : ProtocolVersion.values()) serializeDeserializeTest(v);
  }

  public void serializeDeserializeTest(ProtocolVersion version) {

    for (DataType dt : DataType.allPrimitiveTypes()) {
      if (exclude(dt)) continue;

      Object value = TestUtils.getFixedValue(dt);
      TypeCodec<Object> codec = codecRegistry.codecFor(dt);
      assertEquals(codec.deserialize(codec.serialize(value, version), version), value);
    }

    TypeCodec<Long> codec = codecRegistry.codecFor(DataType.bigint());

    try {
      ByteBuffer badValue = ByteBuffer.allocate(4);
      codec.deserialize(badValue, version);
      fail("This should not have worked");
    } catch (InvalidTypeException e) {
      /* That's what we want */
    }
  }

  @Test(groups = "unit")
  public void serializeDeserializeCollectionsTest() {
    for (ProtocolVersion v : ProtocolVersion.values()) serializeDeserializeCollectionsTest(v);
  }

  public void serializeDeserializeCollectionsTest(ProtocolVersion version) {

    List<String> l = Arrays.asList("foo", "bar");

    DataType dt = DataType.list(DataType.text());
    TypeCodec<List<String>> codec = codecRegistry.codecFor(dt);
    assertEquals(codec.deserialize(codec.serialize(l, version), version), l);

    try {
      DataType listOfBigint = DataType.list(DataType.bigint());
      codec = codecRegistry.codecFor(listOfBigint);
      codec.serialize(l, version);
      fail("This should not have worked");
    } catch (InvalidTypeException e) {
      /* That's what we want */
    }
  }
}
