/*
 * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
 */

package io.airbyte.integrations.destination.s3;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verifyNoMoreInteractions;
import static org.mockito.Mockito.when;

import com.amazonaws.services.s3.AmazonS3;
import com.amazonaws.services.s3.model.AmazonS3Exception;
import com.amazonaws.services.s3.model.InitiateMultipartUploadRequest;
import com.amazonaws.services.s3.model.InitiateMultipartUploadResult;
import com.amazonaws.services.s3.model.ListObjectsRequest;
import com.amazonaws.services.s3.model.UploadPartRequest;
import com.amazonaws.services.s3.model.UploadPartResult;
import com.fasterxml.jackson.databind.JsonNode;
import io.airbyte.protocol.models.AirbyteConnectionStatus;
import io.airbyte.protocol.models.AirbyteConnectionStatus.Status;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import org.mockito.InOrder;
import org.mockito.Mockito;

public class S3DestinationTest {

  private AmazonS3 s3;
  private S3DestinationConfig config;

  @BeforeEach
  public void setup() {
    s3 = mock(AmazonS3.class);
    final InitiateMultipartUploadResult uploadResult = mock(InitiateMultipartUploadResult.class);
    final UploadPartResult uploadPartResult = mock(UploadPartResult.class);
    when(s3.uploadPart(any(UploadPartRequest.class))).thenReturn(uploadPartResult);
    when(s3.initiateMultipartUpload(any(InitiateMultipartUploadRequest.class))).thenReturn(uploadResult);

    config = S3DestinationConfig.create("fake-bucket", "fake-bucketPath", "fake-region")
        .withEndpoint("fake-endpoint")
        .withAccessKeyCredential("fake-accessKeyId", "fake-secretAccessKey")
        .withS3Client(s3)
        .get();
  }

  @Test
  /**
   * Test that check will fail if IAM user does not have listObjects permission
   */
  public void checksS3WithoutListObjectPermission() {
    final S3Destination destinationFail = new S3Destination(new S3DestinationConfigFactory() {

      public S3DestinationConfig getS3DestinationConfig(final JsonNode config) {
        return S3DestinationConfig.create("fake-bucket", "fake-bucketPath", "fake-region")
            .withEndpoint("fake-endpoint")
            .withAccessKeyCredential("fake-accessKeyId", "fake-secretAccessKey")
            .withS3Client(s3)
            .get();
      }

    });
    doThrow(new AmazonS3Exception("Access Denied")).when(s3).listObjects(any(ListObjectsRequest.class));
    final AirbyteConnectionStatus status = destinationFail.check(null);
    assertEquals(Status.FAILED, status.getStatus(), "Connection check should have failed");
    assertTrue(status.getMessage().indexOf("Access Denied") > 0, "Connection check returned wrong failure message");
  }

  @Test
  /**
   * Test that check will succeed when IAM user has all required permissions
   */
  public void checksS3WithListObjectPermission() {
    final S3Destination destinationSuccess = new S3Destination(new S3DestinationConfigFactory() {

      public S3DestinationConfig getS3DestinationConfig(final JsonNode config) {
        return S3DestinationConfig.create("fake-bucket", "fake-bucketPath", "fake-region")
            .withEndpoint("fake-endpoint")
            .withAccessKeyCredential("fake-accessKeyId", "fake-secretAccessKey")
            .withS3Client(s3)
            .get();
      }

    });
    final AirbyteConnectionStatus status = destinationSuccess.check(null);
    assertEquals(Status.SUCCEEDED, status.getStatus(), "Connection check should have succeeded");
  }

  @Test
  public void createsThenDeletesTestFile() {
    S3Destination.attemptS3WriteAndDelete(mock(S3StorageOperations.class), config, "fake-fileToWriteAndDelete", s3);

    // We want to enforce that putObject happens before deleteObject, so use inOrder.verify()
    final InOrder inOrder = Mockito.inOrder(s3);

    final ArgumentCaptor<String> testFileCaptor = ArgumentCaptor.forClass(String.class);
    inOrder.verify(s3).putObject(eq("fake-bucket"), testFileCaptor.capture(), anyString());

    final String testFile = testFileCaptor.getValue();
    assertTrue(testFile.startsWith("fake-fileToWriteAndDelete/_airbyte_connection_test_"), "testFile was actually " + testFile);

    inOrder.verify(s3).listObjects(any(ListObjectsRequest.class));
    inOrder.verify(s3).deleteObject("fake-bucket", testFile);

    verifyNoMoreInteractions(s3);
  }

}
