/*
 * 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.datastax.driver.core.DataType.cint;
import static com.datastax.driver.core.DataType.text;
import static com.datastax.driver.core.TestUtils.serializeForDynamicCompositeType;

import com.datastax.driver.core.utils.CassandraVersion;
import org.testng.SkipException;
import org.testng.annotations.Test;

@CassandraVersion("2.2.0")
@CCMConfig(config = "enable_user_defined_functions:true")
public class AggregateMetadataTest extends CCMTestsSupport {

  @Test(groups = "short")
  public void should_parse_and_format_aggregate_with_initcond_and_no_finalfunc() {
    // given
    String cqlFunction =
        String.format(
            "CREATE FUNCTION %s.cat(s text,v int) RETURNS NULL ON NULL INPUT RETURNS text LANGUAGE java AS 'return s+v;';",
            keyspace);
    String cqlAggregate =
        String.format(
            "CREATE AGGREGATE %s.cat_tos(int) SFUNC cat STYPE text INITCOND '0';", keyspace);
    // when
    session().execute(cqlFunction);
    session().execute(cqlAggregate);
    // then
    KeyspaceMetadata keyspace = cluster().getMetadata().getKeyspace(this.keyspace);
    FunctionMetadata stateFunc = keyspace.getFunction("cat", text(), cint());
    AggregateMetadata aggregate = keyspace.getAggregate("cat_tos", cint());
    assertThat(aggregate).isNotNull();
    assertThat(aggregate.getSignature()).isEqualTo("cat_tos(int)");
    assertThat(aggregate.getSimpleName()).isEqualTo("cat_tos");
    assertThat(aggregate.getArgumentTypes()).containsExactly(cint());
    assertThat(aggregate.getFinalFunc()).isNull();
    assertThat(aggregate.getInitCond()).isEqualTo("0");
    assertThat(aggregate.getReturnType()).isEqualTo(text());
    assertThat(aggregate.getStateFunc()).isEqualTo(stateFunc);
    assertThat(aggregate.getStateType()).isEqualTo(text());
    assertThat(aggregate.toString()).isEqualTo(cqlAggregate);
    assertThat(aggregate.exportAsString())
        .isEqualTo(
            String.format(
                "CREATE AGGREGATE %s.cat_tos(int)\n"
                    + "    SFUNC cat\n"
                    + "    STYPE text\n"
                    + "    INITCOND '0';",
                this.keyspace));
  }

  @Test(groups = "short")
  public void should_parse_and_format_aggregate_with_no_arguments() {
    // given
    String cqlFunction =
        String.format(
            "CREATE FUNCTION %s.inc(i int) RETURNS NULL ON NULL INPUT RETURNS int LANGUAGE java AS 'return i+1;';",
            keyspace);
    String cqlAggregate =
        String.format("CREATE AGGREGATE %s.mycount() SFUNC inc STYPE int INITCOND 0;", keyspace);
    // when
    session().execute(cqlFunction);
    session().execute(cqlAggregate);
    // then
    KeyspaceMetadata keyspace = cluster().getMetadata().getKeyspace(this.keyspace);
    FunctionMetadata stateFunc = keyspace.getFunction("inc", cint());
    AggregateMetadata aggregate = keyspace.getAggregate("mycount");
    assertThat(aggregate).isNotNull();
    assertThat(aggregate.getSignature()).isEqualTo("mycount()");
    assertThat(aggregate.getSimpleName()).isEqualTo("mycount");
    assertThat(aggregate.getArgumentTypes()).isEmpty();
    assertThat(aggregate.getFinalFunc()).isNull();
    assertThat(aggregate.getInitCond()).isEqualTo(0);
    assertThat(aggregate.getReturnType()).isEqualTo(cint());
    assertThat(aggregate.getStateFunc()).isEqualTo(stateFunc);
    assertThat(aggregate.getStateType()).isEqualTo(cint());
    assertThat(aggregate.toString()).isEqualTo(cqlAggregate);
    assertThat(aggregate.exportAsString())
        .isEqualTo(
            String.format(
                "CREATE AGGREGATE %s.mycount()\n"
                    + "    SFUNC inc\n"
                    + "    STYPE int\n"
                    + "    INITCOND 0;",
                this.keyspace));
  }

  @Test(groups = "short")
  public void should_parse_and_format_aggregate_with_final_function() {
    // given
    String cqlFunction1 =
        String.format(
            "CREATE FUNCTION %s.plus(i int, j int) RETURNS NULL ON NULL INPUT RETURNS int LANGUAGE java AS 'return i+j;';",
            keyspace);
    String cqlFunction2 =
        String.format(
            "CREATE FUNCTION %s.announce(i int) RETURNS NULL ON NULL INPUT RETURNS int LANGUAGE java AS 'return i;';",
            keyspace);
    String cqlAggregate =
        String.format(
            "CREATE AGGREGATE %s.prettysum(int) SFUNC plus STYPE int FINALFUNC announce INITCOND 0;",
            keyspace);
    // when
    session().execute(cqlFunction1);
    session().execute(cqlFunction2);
    session().execute(cqlAggregate);
    // then
    KeyspaceMetadata keyspace = cluster().getMetadata().getKeyspace(this.keyspace);
    FunctionMetadata stateFunc = keyspace.getFunction("plus", cint(), cint());
    FunctionMetadata finalFunc = keyspace.getFunction("announce", cint());
    AggregateMetadata aggregate = keyspace.getAggregate("prettysum", cint());
    assertThat(aggregate).isNotNull();
    assertThat(aggregate.getSignature()).isEqualTo("prettysum(int)");
    assertThat(aggregate.getSimpleName()).isEqualTo("prettysum");
    assertThat(aggregate.getArgumentTypes()).containsExactly(cint());
    assertThat(aggregate.getFinalFunc()).isEqualTo(finalFunc);
    assertThat(aggregate.getInitCond()).isEqualTo(0);
    assertThat(aggregate.getReturnType()).isEqualTo(cint());
    assertThat(aggregate.getStateFunc()).isEqualTo(stateFunc);
    assertThat(aggregate.getStateType()).isEqualTo(cint());
    assertThat(aggregate.toString()).isEqualTo(cqlAggregate);
    assertThat(aggregate.exportAsString())
        .isEqualTo(
            String.format(
                "CREATE AGGREGATE %s.prettysum(int)\n"
                    + "    SFUNC plus\n"
                    + "    STYPE int\n"
                    + "    FINALFUNC announce\n"
                    + "    INITCOND 0;",
                this.keyspace));
  }

  @Test(groups = "short")
  public void should_parse_and_format_aggregate_with_no_initcond() {
    // given
    String cqlFunction =
        String.format(
            "CREATE FUNCTION %s.plus2(i int, j int) CALLED ON NULL INPUT RETURNS int LANGUAGE java AS 'return i+j;';",
            keyspace);
    String cqlAggregate =
        String.format("CREATE AGGREGATE %s.sum(int) SFUNC plus2 STYPE int;", keyspace);
    // when
    session().execute(cqlFunction);
    session().execute(cqlAggregate);
    // then
    KeyspaceMetadata keyspace = cluster().getMetadata().getKeyspace(this.keyspace);
    FunctionMetadata stateFunc = keyspace.getFunction("plus2", cint(), cint());
    AggregateMetadata aggregate = keyspace.getAggregate("sum", cint());
    assertThat(aggregate).isNotNull();
    assertThat(aggregate.getSignature()).isEqualTo("sum(int)");
    assertThat(aggregate.getSimpleName()).isEqualTo("sum");
    assertThat(aggregate.getArgumentTypes()).containsExactly(cint());
    assertThat(aggregate.getFinalFunc()).isNull();
    assertThat(aggregate.getInitCond()).isNull();
    assertThat(aggregate.getReturnType()).isEqualTo(cint());
    assertThat(aggregate.getStateFunc()).isEqualTo(stateFunc);
    assertThat(aggregate.getStateType()).isEqualTo(cint());
    assertThat(aggregate.toString()).isEqualTo(cqlAggregate);
    assertThat(aggregate.exportAsString())
        .isEqualTo(
            String.format(
                "CREATE AGGREGATE %s.sum(int)\n" + "    SFUNC plus2\n" + "    STYPE int;",
                this.keyspace));
  }

  @Test(groups = "short")
  public void should_parse_and_format_aggregate_with_udts() {
    // given
    String cqlFunction =
        String.format(
            "CREATE FUNCTION %s.\"MY_FUNC\"(address1 \"Address\", address2 \"Address\") "
                + "CALLED ON NULL INPUT "
                + "RETURNS \"Address\" "
                + "LANGUAGE java "
                + "AS 'return address1;'",
            keyspace);
    String cqlAggregate =
        String.format(
            "CREATE AGGREGATE %s.\"MY_AGGREGATE\"(\"Address\") "
                + "SFUNC \"MY_FUNC\" "
                + "STYPE \"Address\";",
            keyspace);
    // when
    session().execute(cqlFunction);
    session().execute(cqlAggregate);
    // then
    KeyspaceMetadata keyspace = cluster().getMetadata().getKeyspace(this.keyspace);
    UserType addressType = keyspace.getUserType("\"Address\"");
    FunctionMetadata stateFunc = keyspace.getFunction("\"MY_FUNC\"", addressType, addressType);
    AggregateMetadata aggregate = keyspace.getAggregate("\"MY_AGGREGATE\"", addressType);
    assertThat(aggregate).isNotNull();
    assertThat(aggregate.getSignature()).isEqualTo("\"MY_AGGREGATE\"(\"Address\")");
    assertThat(aggregate.getSimpleName()).isEqualTo("MY_AGGREGATE");
    assertThat(aggregate.getArgumentTypes()).containsExactly(addressType);
    assertThat(aggregate.getFinalFunc()).isNull();
    assertThat(aggregate.getInitCond()).isNull();
    assertThat(aggregate.getReturnType()).isEqualTo(addressType);
    assertThat(aggregate.getStateFunc()).isEqualTo(stateFunc);
    assertThat(aggregate.getStateType()).isEqualTo(addressType);
    assertThat(aggregate.toString()).isEqualTo(cqlAggregate);
  }

  /**
   * Validates aggregates with DynamicCompositeType state types and an initcond value that is a
   * literal, i.e.: 's@foo:i@32' can be appropriately parsed and generate a CQL string with the init
   * cond as a hex string, i.e.: 0x80730003666f6f00806900040000002000.
   *
   * @jira_ticket JAVA-1046
   * @test_category metadata
   * @since 3.0.1
   */
  @Test(groups = "short")
  @CassandraVersion("2.2.0")
  public void should_parse_and_format_aggregate_with_composite_type_literal_initcond() {
    VersionNumber ver = ccm().getCassandraVersion();
    if (ver.getMajor() == 3) {
      if ((ver.getMinor() >= 1 && ver.getMinor() < 4)
          || (ver.getMinor() == 0 && ver.getPatch() < 4)) {
        throw new SkipException("Requires C* 2.2.X, 3.0.4+ or 3.4.X+");
      }
    }
    parse_and_format_aggregate_with_composite_type("ag0", "'s@foo:i@32'");
  }

  /**
   * Validates aggregates with DynamicCompositeType state types and an initcond value that is a hex
   * string representing the bytes for the type, i.e.: 0x80730003666f6f00806900040000002000' can be
   * appropriately parsed and generates an equivalent CQL string.
   *
   * @jira_ticket JAVA-1046
   * @test_category metadata
   * @since 3.0.1
   */
  @Test(groups = "short")
  @CassandraVersion("3.4")
  public void should_parse_and_format_aggregate_with_composite_type_hex_initcond() {
    VersionNumber ver = ccm().getCassandraVersion();
    if ((ver.getMinor() >= 1 && ver.getMinor() < 4)) {
      throw new SkipException("Requires 3.0.4+ or 3.4.X+");
    }
    parse_and_format_aggregate_with_composite_type("ag1", "0x80730003666f6f00806900040000002000");
  }

  public void parse_and_format_aggregate_with_composite_type(
      String aggregateName, String initCond) {
    // given
    DataType custom =
        DataType.custom(
            "org.apache.cassandra.db.marshal.DynamicCompositeType("
                + "s=>org.apache.cassandra.db.marshal.UTF8Type,"
                + "i=>org.apache.cassandra.db.marshal.Int32Type)");
    String cqlFunction =
        String.format(
            "CREATE FUNCTION %s.%s_id(i %s) "
                + "RETURNS NULL ON NULL INPUT "
                + "RETURNS %s "
                + "LANGUAGE java "
                + "AS 'return i;';",
            keyspace, aggregateName, custom, custom);
    session().execute(cqlFunction);

    String cqlAggregate =
        String.format(
            "CREATE AGGREGATE %s.%s() " + "SFUNC %s_id " + "STYPE %s " + "INITCOND %s;",
            keyspace, aggregateName, aggregateName, custom, initCond);

    String expectedAggregate =
        String.format(
            "CREATE AGGREGATE %s.%s() "
                + "SFUNC %s_id "
                + "STYPE %s "
                + "INITCOND 0x80730003666f6f00806900040000002000;",
            keyspace, aggregateName, aggregateName, custom);

    int agCount = 0;

    // when
    session().execute(cqlAggregate);
    // then
    KeyspaceMetadata keyspace = cluster().getMetadata().getKeyspace(this.keyspace);
    FunctionMetadata stateFunc = keyspace.getFunction(aggregateName + "_id", custom);
    AggregateMetadata aggregate = keyspace.getAggregate(aggregateName);
    assertThat(aggregate).isNotNull();
    assertThat(aggregate.getSignature()).isEqualTo(aggregateName + "()");
    assertThat(aggregate.getSimpleName()).isEqualTo(aggregateName);
    assertThat(aggregate.getArgumentTypes()).isEmpty();
    assertThat(aggregate.getFinalFunc()).isNull();
    assertThat(aggregate.getInitCond()).isEqualTo(serializeForDynamicCompositeType("foo", 32));
    assertThat(aggregate.getReturnType()).isEqualTo(custom);
    assertThat(aggregate.getStateFunc()).isEqualTo(stateFunc);
    assertThat(aggregate.getStateType()).isEqualTo(custom);

    assertThat(aggregate.toString()).isEqualTo(expectedAggregate);
  }

  @Override
  public void onTestContextInitialized() {
    execute(
        String.format("CREATE TYPE IF NOT EXISTS %s.phone (number text)", keyspace),
        String.format(
            "CREATE TYPE IF NOT EXISTS %s.\"Address\" ("
                + "    street text,"
                + "    city text,"
                + "    zip int,"
                + "    phones frozen<set<frozen<phone>>>,"
                + "    location frozen<tuple<float, float>>"
                + ")",
            keyspace));
  }
}
