code: /*
   * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
   */
  
  package io.airbyte.integrations.destination.snowflake;
  
  import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
  import static org.junit.jupiter.api.Assertions.assertEquals;
  import static org.junit.jupiter.api.Assertions.assertThrows;
  
  import com.fasterxml.jackson.databind.JsonNode;
  import com.fasterxml.jackson.databind.node.ObjectNode;
  import io.airbyte.commons.json.Jsons;
  import io.airbyte.commons.string.Strings;
  import io.airbyte.db.factory.DataSourceFactory;
  import io.airbyte.db.jdbc.JdbcDatabase;
  import io.airbyte.protocol.models.AirbyteConnectionStatus;
  import java.io.IOException;
  import java.nio.file.Files;
  import java.nio.file.Paths;
  import java.sql.SQLException;
  import javax.sql.DataSource;
  import org.junit.jupiter.api.Test;
  
  class SnowflakeDestinationIntegrationTest {
  
    private final SnowflakeSQLNameTransformer namingResolver = new SnowflakeSQLNameTransformer();
  
    @Test
    void testCheckFailsWithInvalidPermissions() throws Exception {
      // TODO(sherifnada) this test case is assumes config.json does not have permission to access the
      // schema
      // this connector should be updated with multiple credentials, each with a clear purpose (valid,
      // invalid: insufficient permissions, invalid: wrong password, etc..)
      final JsonNode credentialsJsonString = Jsons.deserialize(Files.readString(Paths.get("secrets/config.json")));
      final AirbyteConnectionStatus check = new SnowflakeDestination().check(credentialsJsonString);
      assertEquals(AirbyteConnectionStatus.Status.FAILED, check.getStatus());
    }
  
    @Test
    public void testInvalidSchemaName() throws Exception {
      final JsonNode config = getConfig();
      final String schema = config.get("schema").asText();
      final DataSource dataSource = SnowflakeDatabase.createDataSource(config);
      try {
        final JdbcDatabase database = SnowflakeDatabase.getDatabase(dataSource);
        assertDoesNotThrow(() -> syncWithNamingResolver(database, schema));
        assertThrows(SQLException.class, () -> syncWithoutNamingResolver(database, schema));
      } finally {
        DataSourceFactory.close(dataSource);
      }
    }
  
    public void syncWithNamingResolver(final JdbcDatabase database, final String schema) throws SQLException {
      final String normalizedSchemaName = namingResolver.getIdentifier(schema);
      try {
        database.execute(String.format("CREATE SCHEMA %s", normalizedSchemaName));
      } finally {
        database.execute(String.format("DROP SCHEMA IF EXISTS %s", normalizedSchemaName));
      }
    }
  
    private void syncWithoutNamingResolver(final JdbcDatabase database, final String schema) throws SQLException {
      try {
        database.execute(String.format("CREATE SCHEMA %s", schema));
      } finally {
        database.execute(String.format("DROP SCHEMA IF EXISTS %s", schema));
      }
    }
  
    private JsonNode getConfig() throws IOException {
      final JsonNode config = Jsons.deserialize(Files.readString(Paths.get("secrets/insert_config.json")));
      final String schemaName = "schemaName with whitespace " + Strings.addRandomSuffix("integration_test", "_", 5);
      ((ObjectNode) config).put("schema", schemaName);
      return config;
    }
  
  }
  