/*
 * Copyright Debezium Authors.
 *
 * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
 */
package io.debezium.connector.mongodb.transforms.UpdateOperators;

import static org.assertj.core.api.Assertions.assertThat;

import java.util.Arrays;

import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.SchemaBuilder;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;
import org.junit.jupiter.api.Test;

import io.debezium.connector.mongodb.transforms.ExtractNewDocumentState;
import io.debezium.data.VerifyRecord;

/**
 * Integration test for {@link ExtractNewDocumentState}.
 * <p>
 * This subset of tests cover the Array Update Operator as in the official documentation
 * {@see https://docs.mongodb.com/v3.6/reference/operator/update-array/#update-operators}
 *
 * @author Renato Mefi
 */
public class ExtractNewDocumentStateArrayOperatorTestIT extends AbstractExtractNewDocumentStateUpdateOperatorsTestIT {

    /**
     * @see <a href="https://docs.mongodb.com/v3.6/reference/operator/update/addToSet/#up._S_addToSet">MongoDB operator array update $addToSet</a>
     */
    @Test
    void shouldTransformOperationAddToSet() throws InterruptedException {
        SourceRecord updateRecord = executeSimpleUpdateOperation(
                "{'$addToSet': {dataArrayOfStr: 'b'}}");

        final SourceRecord transformedUpdate = transformation.apply(updateRecord);
        final Struct transformedUpdateValue = (Struct) transformedUpdate.value();
        final Schema valueSchema = transformedUpdate.valueSchema();

        VerifyRecord.assertConnectSchemasAreEqual("_id", valueSchema.field("_id").schema(), Schema.OPTIONAL_INT32_SCHEMA);
        VerifyRecord.assertConnectSchemasAreEqual("dataArrayOfStr", valueSchema.field("dataArrayOfStr").schema(),
                SchemaBuilder.array(Schema.OPTIONAL_STRING_SCHEMA).optional().build());
        assertThat(transformedUpdateValue.get("_id")).isEqualTo(1);
        assertThat(transformedUpdateValue.get("dataArrayOfStr")).isEqualTo(Arrays.asList("a", "c", "e", "b"));
    }

    /**
     * @see <a href="https://docs.mongodb.com/v3.6/reference/operator/update/pop/#up._S_pop">MongoDB operator array update $pop</a>
     */
    @Test
    void shouldTransformOperationPop() throws InterruptedException {
        SourceRecord updateRecord = executeSimpleUpdateOperation(
                "{'$pop': {dataArrayOfStr: -1}}");

        final SourceRecord transformedUpdate = transformation.apply(updateRecord);
        final Struct transformedUpdateValue = (Struct) transformedUpdate.value();
        final Schema valueSchema = transformedUpdate.valueSchema();

        VerifyRecord.assertConnectSchemasAreEqual("_id", valueSchema.field("_id").schema(), Schema.OPTIONAL_INT32_SCHEMA);
        VerifyRecord.assertConnectSchemasAreEqual("dataArrayOfStr", valueSchema.field("dataArrayOfStr").schema(),
                SchemaBuilder.array(Schema.OPTIONAL_STRING_SCHEMA).optional().build());
        assertThat(transformedUpdateValue.get("_id")).isEqualTo(1);
        assertThat(transformedUpdateValue.get("dataArrayOfStr")).isEqualTo(Arrays.asList("c", "e"));
    }

    /**
     * @see <a href="https://docs.mongodb.com/v3.6/reference/operator/update/pull/#pull">MongoDB operator array update $pull</a>
     */
    @Test
    void shouldTransformOperationPull() throws InterruptedException {
        SourceRecord updateRecord = executeSimpleUpdateOperation(
                "{'$pull': {dataArrayOfStr: {$in: ['c']}}}");

        final SourceRecord transformedUpdate = transformation.apply(updateRecord);
        final Struct transformedUpdateValue = (Struct) transformedUpdate.value();
        final Schema valueSchema = transformedUpdate.valueSchema();

        VerifyRecord.assertConnectSchemasAreEqual("_id", valueSchema.field("_id").schema(), Schema.OPTIONAL_INT32_SCHEMA);
        VerifyRecord.assertConnectSchemasAreEqual("dataArrayOfStr", valueSchema.field("dataArrayOfStr").schema(),
                SchemaBuilder.array(Schema.OPTIONAL_STRING_SCHEMA).optional().build());
        assertThat(transformedUpdateValue.get("_id")).isEqualTo(1);
        assertThat(transformedUpdateValue.get("dataArrayOfStr")).isEqualTo(Arrays.asList("a", "e"));
    }

    /**
     * @see <a href="https://docs.mongodb.com/v3.6/reference/operator/update/pullAll/#up._S_pullAll">MongoDB operator array update $pullAll</a>
     */
    @Test
    void shouldTransformOperationPullAll() throws InterruptedException {
        SourceRecord updateRecord = executeSimpleUpdateOperation(
                "{'$pullAll': {dataArrayOfStr: ['c']}}");

        final SourceRecord transformedUpdate = transformation.apply(updateRecord);
        final Struct transformedUpdateValue = (Struct) transformedUpdate.value();
        final Schema valueSchema = transformedUpdate.valueSchema();

        VerifyRecord.assertConnectSchemasAreEqual("_id", valueSchema.field("_id").schema(), Schema.OPTIONAL_INT32_SCHEMA);
        VerifyRecord.assertConnectSchemasAreEqual("dataArrayOfStr", valueSchema.field("dataArrayOfStr").schema(),
                SchemaBuilder.array(Schema.OPTIONAL_STRING_SCHEMA).optional().build());
        assertThat(transformedUpdateValue.get("_id")).isEqualTo(1);
        assertThat(transformedUpdateValue.get("dataArrayOfStr")).isEqualTo(Arrays.asList("a", "e"));
    }

    /**
     * @see <a href="https://docs.mongodb.com/v3.6/reference/operator/update/push/#push">MongoDB operator array update $push</a>
     */
    @Test
    void shouldTransformOperationPush() throws InterruptedException {
        SourceRecord updateRecord = executeSimpleUpdateOperation(
                "{'$push': {dataArrayOfStr: 'g'}}");

        final SourceRecord transformedUpdate = transformation.apply(updateRecord);
        final Struct transformedUpdateValue = (Struct) transformedUpdate.value();
        final Schema valueSchema = transformedUpdate.valueSchema();

        // Operations which include items to arrays result in a new field where the structure looks like "FIELD_NAME.ARRAY_INDEX"
        VerifyRecord.assertConnectSchemasAreEqual("_id", valueSchema.field("_id").schema(), Schema.OPTIONAL_INT32_SCHEMA);
        assertThat(transformedUpdateValue.get("_id")).isEqualTo(1);
        VerifyRecord.assertConnectSchemasAreEqual("dataArrayOfStr", valueSchema.field("dataArrayOfStr").schema(),
                SchemaBuilder.array(Schema.OPTIONAL_STRING_SCHEMA).optional().build());
        assertThat(transformedUpdateValue.get("dataArrayOfStr")).isEqualTo(Arrays.asList("a", "c", "e", "g"));
    }
}
