/*
 * 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.nifi.db.schemaregistry;

import org.apache.nifi.embedded.database.EmbeddedDatabaseConnectionService;
import org.apache.nifi.reporting.InitializationException;
import org.apache.nifi.schema.access.SchemaNotFoundException;
import org.apache.nifi.serialization.record.RecordField;
import org.apache.nifi.serialization.record.RecordFieldType;
import org.apache.nifi.serialization.record.RecordSchema;
import org.apache.nifi.serialization.record.SchemaIdentifier;
import org.apache.nifi.serialization.record.StandardSchemaIdentifier;
import org.apache.nifi.util.NoOpProcessor;
import org.apache.nifi.util.TestRunner;
import org.apache.nifi.util.TestRunners;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;

import java.io.IOException;
import java.nio.file.Path;
import java.sql.Connection;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.Arrays;
import java.util.List;
import java.util.Optional;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;

public class DatabaseTableSchemaRegistryTest {

    private static final List<String> CREATE_TABLE_STATEMENTS = Arrays.asList(
            "CREATE SCHEMA SCHEMA1",
            "CREATE SCHEMA SCHEMA2",
            "CREATE TABLE PERSONS (id integer primary key, name varchar(100)," +
                    " code integer CONSTRAINT CODE_RANGE CHECK (code >= 0 AND code < 1000), dt date)",
            "CREATE TABLE SCHEMA1.PERSONS (id integer primary key, name varchar(100)," +
                    " code integer CONSTRAINT CODE_RANGE CHECK (code >= 0 AND code < 1000), dt date)",
            "CREATE TABLE SCHEMA2.PERSONS (id2 integer primary key, name varchar(100)," +
                    " code integer CONSTRAINT CODE_RANGE CHECK (code >= 0 AND code < 1000), dt date)",
            "CREATE TABLE UUID_TEST (id integer primary key, name VARCHAR(100))",
            "CREATE TABLE LONGVARBINARY_TEST (id integer primary key, name CLOB)",
            "CREATE TABLE CONSTANTS (id integer primary key, created TIMESTAMP DEFAULT CURRENT_TIMESTAMP)"
    );

    private static final String SERVICE_ID = EmbeddedDatabaseConnectionService.class.getName();

    private EmbeddedDatabaseConnectionService dbcpService;

    private TestRunner runner;

    @BeforeEach
    public void setService(@TempDir final Path tempDir) throws InitializationException, SQLException {
        dbcpService = new EmbeddedDatabaseConnectionService(tempDir);

        try (
                Connection connection = dbcpService.getConnection();
                Statement statement = connection.createStatement()
        ) {
            for (String createTableStatement : CREATE_TABLE_STATEMENTS) {
                statement.execute(createTableStatement);
            }
        }

        runner = TestRunners.newTestRunner(NoOpProcessor.class);
        runner.addControllerService(SERVICE_ID, dbcpService);
        runner.enableControllerService(dbcpService);
    }

    @AfterEach
    public void close() throws IOException {
        dbcpService.close();
    }

    @Test
    public void testGetSchemaExists() throws Exception {
        DatabaseTableSchemaRegistry dbSchemaRegistry = new DatabaseTableSchemaRegistry();
        runner.addControllerService("schemaRegistry", dbSchemaRegistry);
        runner.setProperty(dbSchemaRegistry, DatabaseTableSchemaRegistry.DBCP_SERVICE, SERVICE_ID);
        runner.setProperty(dbSchemaRegistry, DatabaseTableSchemaRegistry.SCHEMA_NAME, "SCHEMA1");
        runner.enableControllerService(dbSchemaRegistry);
        SchemaIdentifier schemaIdentifier = new StandardSchemaIdentifier.Builder()
                .name("PERSONS")
                .build();
        RecordSchema recordSchema = dbSchemaRegistry.retrieveSchema(schemaIdentifier);
        assertNotNull(recordSchema);
        Optional<RecordField> recordField = recordSchema.getField("ID");
        assertTrue(recordField.isPresent());
        assertEquals(RecordFieldType.INT.getDataType(), recordField.get().getDataType());
        recordField = recordSchema.getField("NAME");
        assertTrue(recordField.isPresent());
        assertEquals(RecordFieldType.STRING.getDataType(), recordField.get().getDataType());
        recordField = recordSchema.getField("CODE");
        assertTrue(recordField.isPresent());
        assertEquals(RecordFieldType.INT.getDataType(), recordField.get().getDataType());
        recordField = recordSchema.getField("DT");
        assertTrue(recordField.isPresent());
        assertEquals(RecordFieldType.DATE.getDataType(), recordField.get().getDataType());
        // Get nonexistent field
        recordField = recordSchema.getField("NOT_A_FIELD");
        assertFalse(recordField.isPresent());
    }

    @Test
    public void testGetSchemaNotExists() throws Exception {
        DatabaseTableSchemaRegistry dbSchemaRegistry = new DatabaseTableSchemaRegistry();
        runner.addControllerService("schemaRegistry", dbSchemaRegistry);
        runner.setProperty(dbSchemaRegistry, DatabaseTableSchemaRegistry.DBCP_SERVICE, SERVICE_ID);
        runner.setProperty(dbSchemaRegistry, DatabaseTableSchemaRegistry.SCHEMA_NAME, "SCHEMA1");
        runner.enableControllerService(dbSchemaRegistry);
        SchemaIdentifier schemaIdentifier = new StandardSchemaIdentifier.Builder()
                .name("NOT_A_TABLE")
                .build();
        assertThrows(SchemaNotFoundException.class, () -> dbSchemaRegistry.retrieveSchema(schemaIdentifier));
    }

    @Test
    public void testGetSchemaCurrentTimestampIgnored() throws Exception {
        final DatabaseTableSchemaRegistry dbSchemaRegistry = new DatabaseTableSchemaRegistry();
        runner.addControllerService("schemaRegistry", dbSchemaRegistry);

        runner.setProperty(dbSchemaRegistry, DatabaseTableSchemaRegistry.DBCP_SERVICE, SERVICE_ID);

        runner.enableControllerService(dbSchemaRegistry);

        final SchemaIdentifier schemaIdentifier = new StandardSchemaIdentifier.Builder()
                .name("CONSTANTS")
                .build();
        final RecordSchema recordSchema = dbSchemaRegistry.retrieveSchema(schemaIdentifier);
        assertNotNull(recordSchema);

        final Optional<RecordField> idField = recordSchema.getField("ID");
        assertTrue(idField.isPresent());
        assertEquals(RecordFieldType.INT.getDataType(), idField.get().getDataType());

        final Optional<RecordField> createdFieldFound = recordSchema.getField("CREATED");
        assertTrue(createdFieldFound.isPresent());
        final RecordField createdField = createdFieldFound.get();
        assertEquals(RecordFieldType.TIMESTAMP.getDataType(), createdField.getDataType());

        // Default Value of CURRENT_TIMESTAMP ignored
        assertNull(createdField.getDefaultValue());
    }
}
