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

import static com.datastax.driver.core.CreateCCM.TestMode.PER_METHOD;
import static com.datastax.driver.core.querybuilder.QueryBuilder.eq;
import static com.datastax.driver.core.querybuilder.QueryBuilder.set;
import static com.datastax.driver.core.querybuilder.QueryBuilder.update;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.fail;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertTrue;

import com.datastax.driver.core.CCMTestsSupport;
import com.datastax.driver.core.Cluster;
import com.datastax.driver.core.CodecRegistry;
import com.datastax.driver.core.CreateCCM;
import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.Row;
import com.datastax.driver.core.Session;
import com.datastax.driver.core.TypeCodec;
import com.datastax.driver.core.UDTValue;
import com.datastax.driver.core.VersionNumber;
import com.datastax.driver.core.exceptions.CodecNotFoundException;
import com.datastax.driver.core.exceptions.InvalidQueryException;
import com.datastax.driver.core.querybuilder.BuiltStatement;
import com.datastax.driver.core.utils.CassandraVersion;
import com.datastax.driver.core.utils.MoreObjects;
import com.datastax.driver.core.utils.UUIDs;
import com.datastax.driver.mapping.annotations.Accessor;
import com.datastax.driver.mapping.annotations.Column;
import com.datastax.driver.mapping.annotations.Field;
import com.datastax.driver.mapping.annotations.Frozen;
import com.datastax.driver.mapping.annotations.FrozenValue;
import com.datastax.driver.mapping.annotations.Param;
import com.datastax.driver.mapping.annotations.PartitionKey;
import com.datastax.driver.mapping.annotations.Query;
import com.datastax.driver.mapping.annotations.Table;
import com.datastax.driver.mapping.annotations.UDT;
import com.google.common.collect.Maps;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import org.assertj.core.data.MapEntry;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;

@SuppressWarnings({"unused", "WeakerAccess"})
@CassandraVersion("2.1.0")
@CreateCCM(PER_METHOD)
public class MapperUDTTest extends CCMTestsSupport {

  @BeforeMethod(groups = "short")
  public void createObjects() {
    execute(
        "CREATE TYPE address (street text, city text, \"ZIP_code\" int, phones set<text>)",
        "CREATE TABLE users (user_id uuid PRIMARY KEY, name text, mainaddress frozen<address>, other_addresses map<text,frozen<address>>)");
  }

  @AfterMethod(groups = "short")
  public void deleteObjects() {
    execute("DROP TABLE IF EXISTS users", "DROP TYPE  IF EXISTS address");
  }

  @Table(name = "users", readConsistency = "QUORUM", writeConsistency = "QUORUM")
  public static class User {
    @PartitionKey
    @Column(name = "user_id")
    private UUID userId;

    private String name;

    @Frozen private Address mainAddress;

    @Column(name = "other_addresses")
    @FrozenValue
    private Map<String, Address> otherAddresses = Maps.newHashMap();

    public User() {}

    public User(String name, Address address) {
      this.userId = UUIDs.random();
      this.name = name;
      this.mainAddress = address;
      this.otherAddresses = new HashMap<String, Address>();
    }

    public UUID getUserId() {
      return userId;
    }

    public void setUserId(UUID userId) {
      this.userId = userId;
    }

    public String getName() {
      return name;
    }

    public void setName(String name) {
      this.name = name;
    }

    public Address getMainAddress() {
      return mainAddress;
    }

    public void setMainAddress(Address address) {
      this.mainAddress = address;
    }

    public Map<String, Address> getOtherAddresses() {
      return otherAddresses;
    }

    public void setOtherAddresses(Map<String, Address> otherAddresses) {
      this.otherAddresses = otherAddresses;
    }

    public void addOtherAddress(String name, Address address) {
      this.otherAddresses.put(name, address);
    }

    @Override
    public boolean equals(Object other) {
      if (this == other) return true;
      if (other instanceof User) {
        User that = (User) other;
        return MoreObjects.equal(this.userId, that.userId)
            && MoreObjects.equal(this.name, that.name)
            && MoreObjects.equal(this.mainAddress, that.mainAddress)
            && MoreObjects.equal(this.otherAddresses, that.otherAddresses);
      }
      return false;
    }

    @Override
    public int hashCode() {
      return MoreObjects.hashCode(this.userId, this.name, this.mainAddress, this.otherAddresses);
    }

    @Override
    public String toString() {
      return String.format(
          "User(userId=%s, name=%s, mainAddress=%s, otherAddresses=%s)",
          userId, name, mainAddress, otherAddresses);
    }
  }

  @UDT(name = "address")
  public static class Address {

    // Dummy constant to test that static fields are properly ignored
    public static final int FOO = 1;

    @Field // not strictly required, but we want to check that the annotation works without a name
    private String city;

    // Declared out of order compared to the UDT definition, to make sure that we serialize fields
    // in the correct order (JAVA-884)
    private String street;

    @Field(name = "ZIP_code", caseSensitive = true)
    private int zipCode;

    private Set<String> phones;

    public Address() {}

    public Address(String street, String city, int zipCode, String... phones) {
      this.street = street;
      this.city = city;
      this.zipCode = zipCode;
      this.phones = new HashSet<String>();
      Collections.addAll(this.phones, phones);
    }

    public String getStreet() {
      return street;
    }

    public void setStreet(String street) {
      this.street = street;
    }

    public String getCity() {
      return city;
    }

    public void setCity(String city) {
      this.city = city;
    }

    public int getZipCode() {
      return zipCode;
    }

    public void setZipCode(int zipCode) {
      this.zipCode = zipCode;
    }

    public Set<String> getPhones() {
      return phones;
    }

    public void setPhones(Set<String> phones) {
      this.phones = phones;
    }

    @Override
    public boolean equals(Object other) {
      if (this == other) return true;
      if (other instanceof Address) {
        Address that = (Address) other;
        return MoreObjects.equal(this.street, that.street)
            && MoreObjects.equal(this.city, that.city)
            && MoreObjects.equal(this.zipCode, that.zipCode)
            && MoreObjects.equal(this.phones, that.phones);
      }
      return false;
    }

    @Override
    public int hashCode() {
      return MoreObjects.hashCode(this.street, this.city, this.zipCode, this.phones);
    }

    @Override
    public String toString() {
      return String.format(
          "Address(street=%s, city=%s, zip=%d, phones=%s)", street, city, zipCode, phones);
    }
  }

  @Accessor
  public interface UserAccessor {
    @Query("SELECT * FROM users WHERE user_id=:userId")
    User getOne(@Param("userId") UUID userId);

    @Query("UPDATE users SET other_addresses[:name]=:address WHERE user_id=:id")
    ResultSet addAddress(
        @Param("id") UUID id, @Param("name") String addressName, @Param("address") Address address);

    @Query("UPDATE users SET other_addresses=:addresses where user_id=:id")
    ResultSet setOtherAddresses(
        @Param("id") UUID id, @Param("addresses") Map<String, Address> addresses);

    @Query("SELECT * FROM users")
    Result<User> getAll();
  }

  @Test(groups = "short")
  public void testSimpleEntity() throws Exception {
    Mapper<User> m = new MappingManager(session()).mapper(User.class);

    User u1 =
        new User(
            "Paul", new Address("12 4th Street", "Springfield", 12345, "12341343", "435423245"));
    u1.addOtherAddress("work", new Address("5 Main Street", "Springfield", 12345, "23431342"));
    m.save(u1);

    assertEquals(m.get(u1.getUserId()), u1);
  }

  @Test(groups = "short")
  public void should_handle_null_UDT_value() throws Exception {
    Mapper<User> m = new MappingManager(session()).mapper(User.class);

    User u1 = new User("Paul", null);
    m.save(u1);

    assertNull(m.get(u1.getUserId()).getMainAddress());
  }

  @Test(groups = "short")
  public void testAccessor() throws Exception {
    MappingManager manager = new MappingManager(session());

    Mapper<User> m = new MappingManager(session()).mapper(User.class);
    User u1 =
        new User(
            "Paul", new Address("12 4th Street", "Springfield", 12345, "12341343", "435423245"));
    m.save(u1);

    UserAccessor userAccessor = manager.createAccessor(UserAccessor.class);

    Address workAddress = new Address("5 Main Street", "Springfield", 12345, "23431342");
    userAccessor.addAddress(u1.getUserId(), "work", workAddress);

    User u2 = userAccessor.getOne(u1.getUserId());
    assertEquals(workAddress, u2.getOtherAddresses().get("work"));

    // Adding a null value should remove it from the list.
    userAccessor.addAddress(u1.getUserId(), "work", null);
    User u3 = userAccessor.getOne(u1.getUserId());
    assertThat(u3.getOtherAddresses()).doesNotContainKey("work");

    // Add a bunch of other addresses
    Map<String, Address> otherAddresses = Maps.newHashMap();
    otherAddresses.put("work", workAddress);
    otherAddresses.put(
        "cabin", new Address("42 Middle of Nowhere", "Lake of the Woods", 49553, "8675309"));

    userAccessor.setOtherAddresses(u1.getUserId(), otherAddresses);
    User u4 = userAccessor.getOne(u1.getUserId());
    assertThat(u4.getOtherAddresses()).isEqualTo(otherAddresses);

    // Nullify other addresses
    userAccessor.setOtherAddresses(u1.getUserId(), null);
    User u5 = userAccessor.getOne(u1.getUserId());
    assertThat(u5.getOtherAddresses()).isEmpty();

    // No argument call
    Result<User> u = userAccessor.getAll();
    assertEquals(u.one(), u5);
    assertTrue(u.isExhausted());
  }

  @Test(groups = "short")
  public void should_be_able_to_use_udtCodec_standalone() {
    // Create a separate Cluster/Session to start with a CodecRegistry from scratch (so not already
    // registered).
    Cluster cluster = register(createClusterBuilder().build());
    CodecRegistry registry = cluster.getConfiguration().getCodecRegistry();
    Session session = cluster.connect(keyspace);

    UUID userId = UUIDs.random();

    // Create a user.
    session.execute(
        "update users SET other_addresses['condo']={street: '101 Ocean Ln', city: 'Jacksonville, FL', \"ZIP_code\": 89898, phones: {'8675309'}} "
            + " WHERE user_id="
            + TypeCodec.uuid().format(userId));
    session.execute(
        "update users SET mainaddress={street: '42 Middle of Nowhere', city: 'Lake of the Woods', \"ZIP_code\": 49553, phones: {'8675039'}} "
            + " WHERE user_Id="
            + TypeCodec.uuid().format(userId));

    // Get the user.
    Row row = session.execute("select * from users where user_id=?", userId).one();

    UDTValue udtValue = row.getUDTValue("mainaddress");
    assertThat(udtValue.getString("street")).isEqualTo("42 Middle of Nowhere");

    Object udtObject = row.getObject("mainaddress");
    assertThat(udtObject).isEqualTo(udtValue);

    // There shouldn't be a codec for address.
    try {
      assertThat(registry.codecFor(udtValue.getType(), Address.class));
      fail("Didn't expect to find codec for udtType <-> Address");
    } catch (CodecNotFoundException e) {
      // expected.
    }

    // Expect a this_udt <-> UDTValue codec to exist.  This is a pretty safe bet or else we wouldn't
    // get
    // this value back.
    TypeCodec<UDTValue> udtCodec = registry.codecFor(udtValue.getType());
    assertThat(udtCodec.getCqlType()).isEqualTo(udtValue.getType());
    assertThat(udtCodec.getJavaType().getRawType()).isEqualTo(UDTValue.class);

    // Retrieve codec for Address, if it can be mapped it will be created, if already registered
    // it'll be used.
    TypeCodec<Address> codec = new MappingManager(session).udtCodec(Address.class);

    // The codec should be registered after we call udtCodec.
    assertThat(registry.codecFor(udtValue.getType(), Address.class)).isEqualTo(codec);

    // Should be able to retrieve as an Address.
    Address mainAddress = row.get("mainaddress", Address.class);
    assertThat(mainAddress.getCity()).isEqualTo("Lake of the Woods");
    assertThat(mainAddress.getStreet()).isEqualTo("42 Middle of Nowhere");
    assertThat(mainAddress.getZipCode()).isEqualTo(49553);
    assertThat(mainAddress.getPhones()).containsExactly("8675039");

    // Should be able to retrieve within a Map.
    Address expectedOtherAddress = new Address();
    expectedOtherAddress.setStreet("101 Ocean Ln");
    expectedOtherAddress.setCity("Jacksonville, FL");
    expectedOtherAddress.setZipCode(89898);
    expectedOtherAddress.setPhones(Collections.singleton("8675309"));

    Map<String, Address> otherAddresses =
        row.getMap("other_addresses", String.class, Address.class);
    assertThat(otherAddresses).containsOnly(MapEntry.entry("condo", expectedOtherAddress));
  }

  /**
   * Ensures that if a table is altered after a {@link Mapper} is created that it continues to work
   * as long as the change is compatible. A new column is added to the table in this case, which is
   * a compatible change.
   *
   * <p>It also ensures that after the change is made that requesting a new {@link Mapper} returns a
   * different instance instead of the existing one.
   *
   * @jira_ticket JAVA-1126
   * @test_category object_mapper
   */
  @Test(groups = "short")
  public void should_work_normally_when_when_table_is_altered_but_remains_compatible() {
    MappingManager manager = new MappingManager(session());
    User expected =
        new User(
            "Paul", new Address("12 4th Street", "Springfield", 12345, "12341343", "435423245"));
    Mapper<User> m = manager.mapper(User.class);
    Mapper<User> m1 = manager.mapper(User.class);
    assertThat(m1).isSameAs(m);
    session().execute("ALTER TABLE users ADD foo text");
    m.save(expected);
    User retrieved = m.get(expected.getUserId());
    assertThat(retrieved).isEqualTo(expected);

    // Mapper should be recreated since table changed.
    Mapper<User> m2 = manager.mapper(User.class);
    assertThat(m2).isNotSameAs(m);
  }

  /**
   * Ensures that if a UDT is altered after a {@link Mapper} is created that has a UDT that it
   * continues to work as long as the change is compatible. A new field is added to the table in
   * this case, which is a compatible change.
   *
   * <p>It also ensures that after the change is made that requesting a new {@link TypeCodec} for
   * that UDT and a new {@link Mapper} for a table using that UDT returns different instances
   * instead of the existing ones.
   *
   * @jira_ticket JAVA-1126
   * @test_category object_mapper
   */
  @Test(groups = "short")
  public void should_work_normally_when_udt_is_altered_but_remains_compatible() {
    MappingManager manager = new MappingManager(session());
    User expected =
        new User(
            "Paul", new Address("12 4th Street", "Springfield", 12345, "12341343", "435423245"));
    Mapper<User> m = manager.mapper(User.class);
    TypeCodec c = manager.udtCodec(Address.class);
    TypeCodec c1 = manager.udtCodec(Address.class);
    assertThat(c1).isSameAs(c);
    session().execute("ALTER TYPE address ADD foo text");
    m.save(expected);
    User retrieved = m.get(expected.getUserId());
    assertThat(retrieved).isEqualTo(expected);

    // Codec should be recreated when requested since UDT and thus the table changed.
    Mapper m2 = manager.mapper(User.class);
    assertThat(m2).isNotSameAs(m);

    // Codec should be recreated when requested since UDT changed.
    TypeCodec c2 = manager.udtCodec(Address.class);
    assertThat(c2).isNotSameAs(c);
  }

  /**
   * Ensures that if a table is dropped after a {@link Mapper} is created that the {@link Mapper}
   * can no longer successfully make queries and that attempting to create a new {@link Mapper}
   * throws an {@link IllegalArgumentException}.
   *
   * <p>It also ensures that requesting a {@link Mapper} fails as the previous {@link Mapper} was
   * evicted and a new one cannot be created as the table has been dropped.
   *
   * @jira_ticket JAVA-1126
   * @test_category object_mapper
   */
  @Test(groups = "short")
  public void should_throw_error_when_table_is_dropped() {
    User user =
        new User(
            "Paul", new Address("12 4th Street", "Springfield", 12345, "12341343", "435423245"));
    MappingManager manager = new MappingManager(session());
    Mapper<User> mapper = manager.mapper(User.class);
    session().execute("DROP TABLE users");
    // error assertion depends on C* version
    boolean isCassandra4 =
        ccm().getCassandraVersion().nextStable().compareTo(VersionNumber.parse("4.0")) >= 0;
    // usage of stale mapper
    try {
      mapper.save(user);
      fail("Expected InvalidQueryException");
    } catch (InvalidQueryException e) {
      // error message is different for C* 4.0
      if (isCassandra4) {
        assertThat(e.getMessage()).contains("table users", "does not exist");
      } else {
        assertThat(e.getMessage()).contains("unconfigured", "users");
      }
    }
    try {
      mapper.get(user.getUserId());
      fail("Expected InvalidQueryException");
    } catch (InvalidQueryException e) {
      if (isCassandra4) {
        assertThat(e.getMessage()).contains("table users", "does not exist");
      } else {
        assertThat(e.getMessage()).contains("unconfigured", "users");
      }
    }
    // trying to use a new mapper
    try {
      manager.mapper(User.class);
      fail("Expected IllegalArgumentException");
    } catch (IllegalArgumentException e) {
      assertThat(e.getMessage())
          .isEqualTo(
              "Table or materialized view users does not exist in keyspace \"" + keyspace + "\"");
    }
  }

  /**
   * Ensures that if a table is altered after a {@link Mapper} is created that it no longer
   * continues to work if the change is not compatible. A column is dropped from the table in this
   * case, which is an incompatible change.
   *
   * <p>It also ensures that requesting a {@link Mapper} fails as the previous {@link Mapper} was
   * evicted and a new one cannot be created as a declared column is missing from the schema but is
   * present in the class definition.
   *
   * @jira_ticket JAVA-1126
   * @test_category object_mapper
   */
  @Test(groups = "short")
  public void should_throw_error_when_table_is_altered_and_is_not_compatible_anymore() {
    User user =
        new User(
            "Paul", new Address("12 4th Street", "Springfield", 12345, "12341343", "435423245"));
    MappingManager manager = new MappingManager(session());
    Mapper<User> mapper = manager.mapper(User.class);
    session().execute("ALTER TABLE users DROP mainaddress");
    // usage of stale mapper
    try {
      mapper.save(user);
      fail("Expected InvalidQueryException");
    } catch (InvalidQueryException e) {
      // Error message varies by C* version.
      assertThat(e.getMessage())
          .matches("(Unknown identifier mainaddress|Undefined column name mainaddress.*)");
    }
    try {
      mapper.get(user.getUserId());
      fail("Expected InvalidQueryException");
    } catch (InvalidQueryException e) {
      // Error message varies by C* version.
      assertThat(e.getMessage())
          .matches(
              "(Undefined name mainaddress in selection clause|Undefined column name mainaddress.*)");
    }
    // trying to use a new mapper
    try {
      manager.mapper(User.class);
      fail("Expected IllegalArgumentException");
    } catch (IllegalArgumentException e) {
      assertThat(e.getMessage()).startsWith("Column mainaddress does not exist");
    }
  }

  /**
   * Ensures that if a UDT is altered after a {@link Mapper} is created that it no longer continues
   * to work if the change made to the UDT is not compatible. A field is renamed in the UDT in this
   * case, which is not a compatible change.
   *
   * <p>It also ensures that requesting a {@link Mapper} fails as the previous {@link Mapper} was
   * evicted and a new one cannot be created as a declared field is missing from the schema but is
   * present in the class definition.
   *
   * <p>Also verifies that a {@link TypeCodec} cannot be requested for the UDT as well.
   *
   * @jira_ticket JAVA-1126
   * @test_category object_mapper
   */
  @Test(groups = "short")
  public void should_throw_error_when_udt_is_altered_and_is_not_compatible_anymore() {
    User user =
        new User(
            "Paul", new Address("12 4th Street", "Springfield", 12345, "12341343", "435423245"));
    MappingManager manager = new MappingManager(session());
    Mapper<User> mapper = manager.mapper(User.class);
    session().execute("ALTER TYPE address RENAME \"ZIP_code\" to zip_code");
    // usage of stale mapper
    try {
      mapper.save(user);
      fail("Expected CodecNotFoundException");
    } catch (CodecNotFoundException e) {
      // ok, codec could not be created
    }
    // insert manually to be able to test retrieval
    session()
        .execute(
            "INSERT INTO users (user_id, name, mainaddress) VALUES (?, 'Paul', { street : '12 4th Street', zip_code : 12345 })",
            user.getUserId());
    try {
      mapper.get(user.getUserId());
      fail("Expected CodecNotFoundException");
    } catch (CodecNotFoundException e) {
      // ok, codec could not be created
    }
    // trying to use a new mapper
    try {
      manager.mapper(User.class);
      fail("Expected IllegalArgumentException");
    } catch (IllegalArgumentException e) {
      assertThat(e.getMessage())
          .isEqualTo(
              String.format("Field \"ZIP_code\" does not exist in type \"%s\".address", keyspace));
    }
    try {
      manager.udtCodec(Address.class);
      fail("Expected IllegalArgumentException");
    } catch (IllegalArgumentException e) {
      assertThat(e.getMessage())
          .isEqualTo(
              String.format("Field \"ZIP_code\" does not exist in type \"%s\".address", keyspace));
    }
  }

  @UDT(name = "address")
  public static class AddressUnknownField {

    private String city;

    private String street;

    @Field(name = "ZIP_code", caseSensitive = true)
    private int zipCode;

    private Set<String> phones;

    public String province;

    public AddressUnknownField() {}

    public String getStreet() {
      return street;
    }

    public void setStreet(String street) {
      this.street = street;
    }

    public String getCity() {
      return city;
    }

    public void setCity(String city) {
      this.city = city;
    }

    public int getZipCode() {
      return zipCode;
    }

    public void setZipCode(int zipCode) {
      this.zipCode = zipCode;
    }

    public Set<String> getPhones() {
      return phones;
    }

    public void setPhones(Set<String> phones) {
      this.phones = phones;
    }

    public String getProvince() {
      return province;
    }

    public void setProvince(String province) {
      this.province = province;
    }
  }

  /**
   * Ensures that when attempting to create a {@link TypeCodec} from a class that has a field that
   * does not exist in the UDT that an {@link IllegalArgumentException} is thrown.
   *
   * @jira_ticket JAVA-1126
   * @test_category object_mapper
   */
  @Test(groups = "short", expectedExceptions = IllegalArgumentException.class)
  public void should_fail_to_create_codec_if_class_has_field_not_in_udt() {
    MappingManager manager = new MappingManager(session());
    manager.getUDTCodec(AddressUnknownField.class, keyspace);
  }

  @UDT(name = "nonexistent")
  public static class NonExistentUDT {
    public String name;

    public void setName(String name) {
      this.name = name;
    }

    public String getName() {
      return this.name;
    }
  }

  /**
   * Ensures that when attempting to create a {@link TypeCodec} from a class that has a {@link UDT}
   * annotation with a name that doesn't exist in the current keyspace that an {@link
   * IllegalArgumentException} is thrown.
   *
   * @jira_ticket JAVA-1126
   * @test_category object_mapper
   */
  @Test(groups = "short", expectedExceptions = IllegalArgumentException.class)
  public void should_fail_to_create_codec_if_udt_does_not_exist() {
    MappingManager manager = new MappingManager(session());
    manager.getUDTCodec(NonExistentUDT.class, keyspace);
  }

  @Table(name = "users")
  public static class UserWithAddressUnknownField {
    @PartitionKey
    @Column(name = "user_id")
    private UUID userId;

    private String name;

    @Frozen private AddressUnknownField mainAddress;

    @Column(name = "other_addresses")
    @FrozenValue
    private Map<String, Address> otherAddresses = Maps.newHashMap();

    public UserWithAddressUnknownField() {}

    public UUID getUserId() {
      return userId;
    }

    public void setUserId(UUID userId) {
      this.userId = userId;
    }

    public String getName() {
      return name;
    }

    public void setName(String name) {
      this.name = name;
    }

    public AddressUnknownField getMainAddress() {
      return mainAddress;
    }

    public void setMainAddress(AddressUnknownField address) {
      this.mainAddress = address;
    }

    public Map<String, Address> getOtherAddresses() {
      return otherAddresses;
    }

    public void setOtherAddresses(Map<String, Address> otherAddresses) {
      this.otherAddresses = otherAddresses;
    }

    public void addOtherAddress(String name, Address address) {
      this.otherAddresses.put(name, address);
    }
  }

  /**
   * Ensures that when attempting to create a {@link Mapper} from a class that has a field that is a
   * class mapping to a user type that has a field that does not exist in that UDT that an {@link
   * IllegalArgumentException} is thrown.
   *
   * @jira_ticket JAVA-1126
   * @test_category object_mapper
   */
  @Test(groups = "short", expectedExceptions = IllegalArgumentException.class)
  public void
      should_fail_to_create_mapper_if_class_has_udt_field_class_that_has_field_not_in_udt() {
    MappingManager manager = new MappingManager(session());
    manager.mapper(UserWithAddressUnknownField.class);
  }

  /**
   * Ensures that MappedUDTCodec is able to properly format UDTs when printing the query string of a
   * BuiltStatement.
   *
   * @jira_ticket JAVA-1272
   * @test_category object_mapper
   */
  @Test(groups = "short")
  public void should_format_mapped_udt() throws Exception {
    MappingManager manager = new MappingManager(session());
    UUID uuid = UUIDs.random();
    BuiltStatement update =
        update("users")
            .with(
                set(
                    "mainaddress",
                    new Address("12 4th Street", "Springfield", 12345, "12341343", "435423245")))
            .where(eq("user_id", uuid));
    CodecRegistry codecRegistry = cluster().getConfiguration().getCodecRegistry();
    codecRegistry.register(manager.udtCodec(Address.class));
    String queryString = update.getQueryString(codecRegistry);
    assertThat(queryString).isEqualTo("UPDATE users SET mainaddress=? WHERE user_id=?;");
    update.setForceNoValues(true);
    queryString = update.getQueryString(codecRegistry);
    assertThat(queryString)
        .isEqualTo(
            "UPDATE users "
                + "SET mainaddress={street:'12 4th Street',city:'Springfield',\"ZIP_code\":12345,phones:{'435423245','12341343'}} "
                + "WHERE user_id="
                + uuid
                + ";");
    // check that the query string is valid
    session().execute(queryString);
  }

  /**
   * Ensures that MappedUDTCodec is able to properly parse UDTs.
   *
   * @jira_ticket JAVA-1272
   * @test_category object_mapper
   */
  @Test(groups = "short")
  public void should_parse_mapped_udt() throws Exception {
    MappingManager manager = new MappingManager(session());
    TypeCodec<Address> codec = manager.udtCodec(Address.class);
    Address actual =
        codec.parse(
            "{street:'12 4th Street',city:'Springfield',\"ZIP_code\":12345,phones:{'435423245','12341343'}}");
    assertThat(actual)
        .isEqualTo(new Address("12 4th Street", "Springfield", 12345, "12341343", "435423245"));
  }
}
