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

import static org.fest.assertions.Assertions.assertThat;

import java.io.IOException;
import java.net.URISyntaxException;
import java.net.URL;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.Map.Entry;

import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.SchemaBuilder;
import org.apache.kafka.connect.data.Struct;
import org.bson.BsonDocument;
import org.bson.BsonValue;
import org.junit.Before;
import org.junit.Test;

import io.debezium.connector.mongodb.transforms.ExtractNewDocumentState.ArrayEncoding;
import io.debezium.doc.FixFor;

/**
 * Unit test for {@code MongoDataConverter}.
 *
 * @author Sairam Polavarapu
 */
public class MongoDataConverterTest {

    private String record;
    private BsonDocument val;
    private SchemaBuilder builder;
    private MongoDataConverter converter;

    @Before
    public void setup() throws Exception {
        record = getFile("restaurants5.json");
        val = BsonDocument.parse(record);
        builder = SchemaBuilder.struct().name("pub");
        converter = new MongoDataConverter(ArrayEncoding.ARRAY);
    }

    @Test
    public void shouldCreateCorrectStructFromInsertJson() {
        for (Entry<String, BsonValue> entry : val.entrySet()) {
            converter.addFieldSchema(entry, builder);
        }

        Schema finalSchema = builder.build();
        Struct struct = new Struct(finalSchema);

        for (Entry<String, BsonValue> entry : val.entrySet()) {
            converter.convertRecord(entry, finalSchema, struct);
        }

        assertThat(struct.toString()).isEqualTo(
                "Struct{"
                        + "address=Struct{"
                        + "building=1007,"
                        + "floor=Struct{"
                        + "level=17,"
                        + "description=level 17"
                        + "},"
                        + "coord=[-73.856077, 40.848447],"
                        + "street=Morris Park Ave,"
                        + "zipcode=10462"
                        + "},"
                        + "borough=Bronx,"
                        + "cuisine=Bakery,"
                        + "grades=["
                        + "Struct{date=1393804800000,grade=A,score=2}, "
                        + "Struct{date=1378857600000,grade=A,score=6}, "
                        + "Struct{date=1358985600000,grade=A,score=10}, "
                        + "Struct{date=1322006400000,grade=A,score=9}, "
                        + "Struct{date=1299715200000,grade=B,score=14}"
                        + "],"
                        + "name=Morris Park Bake Shop,"
                        + "restaurant_id=30075445"
                        + "}");
    }

    @Test
    public void shouldCreateCorrectSchemaFromInsertJson() {
        for (Entry<String, BsonValue> entry : val.entrySet()) {
            converter.addFieldSchema(entry, builder);
        }
        Schema finalSchema = builder.build();

        assertThat(finalSchema).isEqualTo(
                SchemaBuilder.struct().name("pub")
                        .field("address", SchemaBuilder.struct().name("pub.address").optional()
                                .field("building", Schema.OPTIONAL_STRING_SCHEMA)
                                .field("floor", SchemaBuilder.struct().name("pub.address.floor").optional()
                                        .field("level", Schema.OPTIONAL_INT32_SCHEMA)
                                        .field("description", Schema.OPTIONAL_STRING_SCHEMA)
                                        .build())
                                .field("coord", SchemaBuilder.array(Schema.OPTIONAL_FLOAT64_SCHEMA).optional().build())
                                .field("street", Schema.OPTIONAL_STRING_SCHEMA)
                                .field("zipcode", Schema.OPTIONAL_STRING_SCHEMA)
                                .build())
                        .field("borough", Schema.OPTIONAL_STRING_SCHEMA)
                        .field("cuisine", Schema.OPTIONAL_STRING_SCHEMA)
                        .field("grades", SchemaBuilder.array(SchemaBuilder.struct().name("pub.grades").optional()
                                .field("date", Schema.OPTIONAL_INT64_SCHEMA)
                                .field("grade", Schema.OPTIONAL_STRING_SCHEMA)
                                .field("score", Schema.OPTIONAL_INT32_SCHEMA)
                                .build())
                                .optional()
                                .build())
                        .field("name", Schema.OPTIONAL_STRING_SCHEMA)
                        .field("restaurant_id", Schema.OPTIONAL_STRING_SCHEMA)
                        .build());
    }

    private String getFile(String fileName) throws IOException, URISyntaxException {
        URL jsonResource = getClass().getClassLoader().getResource(fileName);
        return new String(
                Files.readAllBytes(Paths.get(jsonResource.toURI())),
                StandardCharsets.UTF_8);
    }

    @Test
    @FixFor("DBZ-928")
    public void shouldProcessNullValue() {
        val = BsonDocument.parse("{\n" +
                "    \"_id\" : ObjectId(\"51e5619ee4b01f9fbdfba9fc\"),\n" +
                "    \"delivery\" : {\n" +
                "        \"hour\" : null,\n" +
                "        \"hourId\" : 10\n" +
                "    }\n" +
                "}");
        builder = SchemaBuilder.struct().name("withnull");
        converter = new MongoDataConverter(ArrayEncoding.ARRAY);

        for (Entry<String, BsonValue> entry : val.entrySet()) {
            converter.addFieldSchema(entry, builder);
        }
        Schema finalSchema = builder.build();
        Struct struct = new Struct(finalSchema);

        for (Entry<String, BsonValue> entry : val.entrySet()) {
            converter.convertRecord(entry, finalSchema, struct);
        }

        assertThat(finalSchema).isEqualTo(
                SchemaBuilder.struct().name("withnull")
                        .field("_id", Schema.OPTIONAL_STRING_SCHEMA)
                        .field("delivery", SchemaBuilder.struct().name("withnull.delivery").optional()
                                .field("hour", Schema.OPTIONAL_STRING_SCHEMA)
                                .field("hourId", Schema.OPTIONAL_INT32_SCHEMA)
                                .build())
                        .build());
        assertThat(struct.toString()).isEqualTo(
                "Struct{"
                        + "_id=51e5619ee4b01f9fbdfba9fc,"
                        + "delivery=Struct{"
                        + "hourId=10"
                        + "}"
                        + "}");
    }

    @Test
    @FixFor("DBZ-1315")
    public void shouldProcessUnsupportedValue() {
        val = BsonDocument.parse("{\n" +
                "    \"_id\" : ObjectId(\"518cc94bc27cfa20d9693e5d\"),\n" +
                "    \"name\" : undefined,\n" +
                "    \"address\" : {\n" +
                "        \"building\" : undefined,\n" +
                "        \"floor\" : 10\n" +
                "    }\n" +
                "}");
        builder = SchemaBuilder.struct().name("withundefined");
        converter = new MongoDataConverter(ArrayEncoding.DOCUMENT);

        for (Entry<String, BsonValue> entry : val.entrySet()) {
            converter.addFieldSchema(entry, builder);
        }
        Schema finalSchema = builder.build();
        Struct struct = new Struct(finalSchema);

        for (Entry<String, BsonValue> entry : val.entrySet()) {
            converter.convertRecord(entry, finalSchema, struct);
        }

        assertThat(finalSchema).isEqualTo(
                SchemaBuilder.struct().name("withundefined")
                        .field("_id", Schema.OPTIONAL_STRING_SCHEMA)
                        .field("address", SchemaBuilder.struct().name("withundefined.address").optional()
                                .field("floor", Schema.OPTIONAL_INT32_SCHEMA)
                                .build())
                        .build());
        assertThat(struct.toString()).isEqualTo(
                "Struct{"
                        + "_id=518cc94bc27cfa20d9693e5d,"
                        + "address=Struct{"
                        + "floor=10"
                        + "}"
                        + "}");

    }
}
