/*
 * 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 org.apache.beam.sdk.extensions.avro.schemas.transforms;

import java.util.Arrays;
import java.util.Map;
import java.util.Objects;
import org.apache.avro.generic.GenericRecord;
import org.apache.beam.sdk.extensions.avro.schemas.utils.AvroUtils;
import org.apache.beam.sdk.schemas.JavaFieldSchema;
import org.apache.beam.sdk.schemas.Schema;
import org.apache.beam.sdk.schemas.Schema.FieldType;
import org.apache.beam.sdk.schemas.annotations.DefaultSchema;
import org.apache.beam.sdk.schemas.transforms.Convert;
import org.apache.beam.sdk.testing.NeedsRunner;
import org.apache.beam.sdk.testing.PAssert;
import org.apache.beam.sdk.testing.TestPipeline;
import org.apache.beam.sdk.testing.UsesSchema;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.Row;
import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList;
import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
import org.checkerframework.checker.nullness.qual.Nullable;
import org.junit.Rule;
import org.junit.Test;
import org.junit.experimental.categories.Category;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;

/** Tests for the {@link Convert} class. */
@RunWith(JUnit4.class)
@Category(UsesSchema.class)
public class ConvertTest {
  @Rule public final transient TestPipeline pipeline = TestPipeline.create();

  /** Test outer POJO. * */
  @DefaultSchema(JavaFieldSchema.class)
  public static class POJO1 {
    public String field1 = "field1";
    public long field2 = 42;
    public POJO1Nested field3 = new POJO1Nested();
    public POJO1Nested[] field4 = new POJO1Nested[] {new POJO1Nested(), new POJO1Nested()};
    public Map<String, POJO1Nested> field5 =
        ImmutableMap.of(
            "first", new POJO1Nested(),
            "second", new POJO1Nested());

    @Override
    public boolean equals(@Nullable Object o) {
      if (this == o) {
        return true;
      }
      if (!(o instanceof POJO1)) {
        return false;
      }
      POJO1 pojo1 = (POJO1) o;
      return field2 == pojo1.field2
          && Objects.equals(field1, pojo1.field1)
          && Objects.equals(field3, pojo1.field3)
          && Arrays.equals(field4, pojo1.field4)
          && Objects.equals(field5, pojo1.field5);
    }

    @Override
    public int hashCode() {
      int result = Objects.hash(field1, field2, field3, field5);
      result = 31 * result + Arrays.hashCode(field4);
      return result;
    }
  }

  /** Test inner POJO. * */
  @DefaultSchema(JavaFieldSchema.class)
  public static class POJO1Nested {
    public String yard1 = "yard2";
    public long yard2 = 43;

    @Override
    public boolean equals(@Nullable Object o) {
      if (this == o) {
        return true;
      }
      if (!(o instanceof POJO1Nested)) {
        return false;
      }
      POJO1Nested that = (POJO1Nested) o;
      return yard2 == that.yard2 && Objects.equals(yard1, that.yard1);
    }

    @Override
    public int hashCode() {
      return Objects.hash(yard1, yard2);
    }
  }

  private static final Schema EXPECTED_SCHEMA1_NESTED =
      Schema.builder().addStringField("yard1").addInt64Field("yard2").build();

  private static final Schema EXPECTED_SCHEMA1 =
      Schema.builder()
          .addStringField("field1")
          .addInt64Field("field2")
          .addRowField("field3", EXPECTED_SCHEMA1_NESTED)
          .addArrayField("field4", FieldType.row(EXPECTED_SCHEMA1_NESTED))
          .addMapField("field5", FieldType.STRING, FieldType.row(EXPECTED_SCHEMA1_NESTED))
          .build();

  private static final Row EXPECTED_ROW1_NESTED =
      Row.withSchema(EXPECTED_SCHEMA1_NESTED).addValues("yard2", 43L).build();

  private static final Row EXPECTED_ROW1 =
      Row.withSchema(EXPECTED_SCHEMA1)
          .addValue("field1")
          .addValue(42L)
          .addValue(EXPECTED_ROW1_NESTED)
          .addArray(ImmutableList.of(EXPECTED_ROW1_NESTED, EXPECTED_ROW1_NESTED))
          .addValue(ImmutableMap.of("first", EXPECTED_ROW1_NESTED, "second", EXPECTED_ROW1_NESTED))
          .build();

  private static final GenericRecord EXPECTED_GENERICRECORD1 =
      AvroUtils.toGenericRecord(EXPECTED_ROW1, AvroUtils.toAvroSchema(EXPECTED_SCHEMA1));

  @Test
  @Category(NeedsRunner.class)
  public void testToGenericRecords() {
    PCollection<GenericRecord> records =
        pipeline.apply(Create.of(new POJO1())).apply(Convert.to(GenericRecord.class));
    PAssert.that(records).containsInAnyOrder(EXPECTED_GENERICRECORD1);
    pipeline.run();
  }
}
