/*
 * 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.hadoop.fs.s3a;

import java.util.concurrent.atomic.AtomicLong;

import com.amazonaws.AmazonServiceException;
import com.amazonaws.SdkBaseException;
import com.amazonaws.services.s3.Headers;
import com.amazonaws.services.s3.model.CopyObjectRequest;
import com.amazonaws.services.s3.model.GetObjectRequest;
import com.amazonaws.services.s3.model.ObjectMetadata;
import com.amazonaws.services.s3.model.S3Object;
import com.amazonaws.services.s3.transfer.model.CopyResult;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.apache.hadoop.fs.PathIOException;
import org.apache.hadoop.fs.s3a.impl.ChangeDetectionPolicy;
import org.apache.hadoop.fs.s3a.impl.ChangeTracker;
import org.apache.hadoop.test.HadoopTestBase;
import org.apache.http.HttpStatus;

import static org.apache.hadoop.fs.s3a.impl.ChangeDetectionPolicy.CHANGE_DETECTED;
import static org.apache.hadoop.fs.s3a.impl.ChangeDetectionPolicy.createPolicy;
import static org.apache.hadoop.fs.s3a.impl.ChangeTracker.CHANGE_REPORTED_BY_S3;
import static org.apache.hadoop.test.LambdaTestUtils.intercept;

/**
 * Test {@link ChangeTracker}.
 */
public class TestStreamChangeTracker extends HadoopTestBase {

  private static final Logger LOG =
      LoggerFactory.getLogger(TestStreamChangeTracker.class);

  public static final String BUCKET = "bucket";

  public static final String OBJECT = "object";

  public static final String DEST_OBJECT = "new_object";

  public static final String URI = "s3a://" + BUCKET + "/" + OBJECT;

  @Test
  public void testVersionCheckingHandlingNoVersions() throws Throwable {
    LOG.info("If an endpoint doesn't return versions, that's OK");
    ChangeTracker tracker = newTracker(
        ChangeDetectionPolicy.Mode.Client,
        ChangeDetectionPolicy.Source.VersionId,
        false);
    assertFalse("Tracker should not have applied contraints " + tracker,
        tracker.maybeApplyConstraint(newGetObjectRequest()));
    tracker.processResponse(
        newResponse(null, null),
        "", 0);
    assertTrackerMismatchCount(tracker, 0);
  }

  @Test
  public void testVersionCheckingHandlingNoVersionsVersionRequired()
      throws Throwable {
    LOG.info("If an endpoint doesn't return versions but we are configured to"
        + "require them");
    ChangeTracker tracker = newTracker(
        ChangeDetectionPolicy.Mode.Client,
        ChangeDetectionPolicy.Source.VersionId,
        true);
    expectNoVersionAttributeException(tracker, newResponse(null, null),
        "policy requires VersionId");
  }

  @Test
  public void testEtagCheckingWarn() throws Throwable {
    LOG.info("If an endpoint doesn't return errors, that's OK");
    ChangeTracker tracker = newTracker(
        ChangeDetectionPolicy.Mode.Warn,
        ChangeDetectionPolicy.Source.ETag,
        false);
    assertFalse("Tracker should not have applied constraints " + tracker,
        tracker.maybeApplyConstraint(newGetObjectRequest()));
    tracker.processResponse(
        newResponse("e1", null),
        "", 0);
    tracker.processResponse(
        newResponse("e1", null),
        "", 0);
    tracker.processResponse(
        newResponse("e2", null),
        "", 0);
    assertTrackerMismatchCount(tracker, 1);
    // subsequent error triggers doesn't trigger another warning
    tracker.processResponse(
        newResponse("e2", null),
        "", 0);
    assertTrackerMismatchCount(tracker, 1);
  }

  @Test
  public void testVersionCheckingOnClient() throws Throwable {
    LOG.info("Verify the client-side version checker raises exceptions");
    ChangeTracker tracker = newTracker(
        ChangeDetectionPolicy.Mode.Client,
        ChangeDetectionPolicy.Source.VersionId,
        false);
    assertFalse("Tracker should not have applied constraints " + tracker,
        tracker.maybeApplyConstraint(newGetObjectRequest()));
    tracker.processResponse(
        newResponse(null, "rev1"),
        "", 0);
    assertTrackerMismatchCount(tracker, 0);
    assertRevisionId(tracker, "rev1");
    GetObjectRequest request = newGetObjectRequest();
    expectChangeException(tracker,
        newResponse(null, "rev2"), "change detected");
    // mismatch was noted (so gets to FS stats)
    assertTrackerMismatchCount(tracker, 1);

    // another read causes another exception
    expectChangeException(tracker,
        newResponse(null, "rev2"), "change detected");
    // mismatch was noted again
    assertTrackerMismatchCount(tracker, 2);
  }

  @Test
  public void testVersionCheckingOnServer() throws Throwable {
    LOG.info("Verify the client-side version checker handles null-ness");
    ChangeTracker tracker = newTracker(
        ChangeDetectionPolicy.Mode.Server,
        ChangeDetectionPolicy.Source.VersionId,
        false);
    assertFalse("Tracker should not have applied contraints " + tracker,
        tracker.maybeApplyConstraint(newGetObjectRequest()));
    tracker.processResponse(
        newResponse(null, "rev1"),
        "", 0);
    assertTrackerMismatchCount(tracker, 0);
    assertRevisionId(tracker, "rev1");
    GetObjectRequest request = newGetObjectRequest();
    assertConstraintApplied(tracker, request);
    // now, the tracker expects a null response
    expectChangeException(tracker, null, CHANGE_REPORTED_BY_S3);
    assertTrackerMismatchCount(tracker, 1);

    // now, imagine the server doesn't trigger a failure due to some
    // bug in its logic
    // we should still react to the reported value
    expectChangeException(tracker,
        newResponse(null, "rev2"),
        CHANGE_DETECTED);
  }

  @Test
  public void testVersionCheckingUpfrontETag() throws Throwable {
    ChangeTracker tracker = newTracker(
        ChangeDetectionPolicy.Mode.Server,
        ChangeDetectionPolicy.Source.ETag,
        false,
        objectAttributes("etag1", "versionid1"));

    assertEquals("etag1", tracker.getRevisionId());
  }

  @Test
  public void testVersionCheckingUpfrontVersionId() throws Throwable {
    ChangeTracker tracker = newTracker(
        ChangeDetectionPolicy.Mode.Server,
        ChangeDetectionPolicy.Source.VersionId,
        false,
        objectAttributes("etag1", "versionid1"));

    assertEquals("versionid1", tracker.getRevisionId());
  }

  @Test
  public void testVersionCheckingETagCopyServer() throws Throwable {
    ChangeTracker tracker = newTracker(
        ChangeDetectionPolicy.Mode.Server,
        ChangeDetectionPolicy.Source.VersionId,
        false,
        objectAttributes("etag1", "versionid1"));
    assertConstraintApplied(tracker, newCopyObjectRequest());
  }

  @Test
  public void testVersionCheckingETagCopyClient() throws Throwable {
    ChangeTracker tracker = newTracker(
        ChangeDetectionPolicy.Mode.Client,
        ChangeDetectionPolicy.Source.VersionId,
        false,
        objectAttributes("etag1", "versionid1"));
    assertFalse("Tracker should not have applied contraints " + tracker,
        tracker.maybeApplyConstraint(newCopyObjectRequest()));
  }

  @Test
  public void testCopyVersionIdRequired() throws Throwable {
    ChangeTracker tracker = newTracker(
        ChangeDetectionPolicy.Mode.Client,
        ChangeDetectionPolicy.Source.VersionId,
        true,
        objectAttributes("etag1", "versionId"));

    expectNoVersionAttributeException(tracker, newCopyResult("etag1",
        null),
        "policy requires VersionId");
  }

  @Test
  public void testCopyETagRequired() throws Throwable {
    ChangeTracker tracker = newTracker(
        ChangeDetectionPolicy.Mode.Client,
        ChangeDetectionPolicy.Source.ETag,
        true,
        objectAttributes("etag1", "versionId"));

    expectNoVersionAttributeException(tracker, newCopyResult(null,
        "versionId"),
        "policy requires ETag");
  }

  @Test
  public void testCopyVersionMismatch() throws Throwable {
    ChangeTracker tracker = newTracker(
        ChangeDetectionPolicy.Mode.Server,
        ChangeDetectionPolicy.Source.ETag,
        true,
        objectAttributes("etag", "versionId"));

    // 412 is translated to RemoteFileChangedException
    // note: this scenario is never currently hit due to
    // https://github.com/aws/aws-sdk-java/issues/1644
    AmazonServiceException awsException =
        new AmazonServiceException("aws exception");
    awsException.setStatusCode(HttpStatus.SC_PRECONDITION_FAILED);
    expectChangeException(tracker, awsException, "copy",
        RemoteFileChangedException.PRECONDITIONS_FAILED);

    // processing another type of exception does nothing
    tracker.processException(new SdkBaseException("foo"), "copy");
  }

  protected void assertConstraintApplied(final ChangeTracker tracker,
      final GetObjectRequest request) {
    assertTrue("Tracker should have applied contraints " + tracker,
        tracker.maybeApplyConstraint(request));
  }

  protected void assertConstraintApplied(final ChangeTracker tracker,
      final CopyObjectRequest request) throws PathIOException {
    assertTrue("Tracker should have applied contraints " + tracker,
        tracker.maybeApplyConstraint(request));
  }

  protected RemoteFileChangedException expectChangeException(
      final ChangeTracker tracker,
      final S3Object response,
      final String message) throws Exception {
    return expectException(tracker, response, message,
        RemoteFileChangedException.class);
  }

  protected RemoteFileChangedException expectChangeException(
      final ChangeTracker tracker,
      final SdkBaseException exception,
      final String operation,
      final String message) throws Exception {
    return expectException(tracker, exception, operation, message,
        RemoteFileChangedException.class);
  }

  protected PathIOException expectNoVersionAttributeException(
      final ChangeTracker tracker,
      final S3Object response,
      final String message) throws Exception {
    return expectException(tracker, response, message,
        NoVersionAttributeException.class);
  }

  protected PathIOException expectNoVersionAttributeException(
      final ChangeTracker tracker,
      final CopyResult response,
      final String message) throws Exception {
    return expectException(tracker, response, message,
        NoVersionAttributeException.class);
  }

  protected <T extends Exception> T expectException(
      final ChangeTracker tracker,
      final S3Object response,
      final String message,
      final Class<T> clazz) throws Exception {
    return intercept(
        clazz,
        message,
        () -> {
          tracker.processResponse(response, "", 0);
          return tracker;
        });
  }

  protected <T extends Exception> T expectException(
      final ChangeTracker tracker,
      final CopyResult response,
      final String message,
      final Class<T> clazz) throws Exception {
    return intercept(
        clazz,
        message,
        () -> {
          tracker.processResponse(response);
          return tracker;
        });
  }

  protected <T extends Exception> T expectException(
      final ChangeTracker tracker,
      final SdkBaseException exception,
      final String operation,
      final String message,
      final Class<T> clazz) throws Exception {
    return intercept(
        clazz,
        message,
        () -> {
          tracker.processException(exception, operation);
          return tracker;
        });
  }

  protected void assertRevisionId(final ChangeTracker tracker,
      final String revId) {
    assertEquals("Wrong revision ID in " + tracker,
        revId, tracker.getRevisionId());
  }


  protected void assertTrackerMismatchCount(
      final ChangeTracker tracker,
      final int expectedCount) {
    assertEquals("counter in tracker " + tracker,
        expectedCount, tracker.getVersionMismatches().get());
  }

  /**
   * Create tracker.
   * Contains standard assertions(s).
   * @return the tracker.
   */
  protected ChangeTracker newTracker(final ChangeDetectionPolicy.Mode mode,
      final ChangeDetectionPolicy.Source source, boolean requireVersion) {
    return newTracker(mode, source, requireVersion,
        objectAttributes(null, null));
  }

  /**
   * Create tracker.
   * Contains standard assertions(s).
   * @return the tracker.
   */
  protected ChangeTracker newTracker(final ChangeDetectionPolicy.Mode mode,
      final ChangeDetectionPolicy.Source source, boolean requireVersion,
      S3ObjectAttributes objectAttributes) {
    ChangeDetectionPolicy policy = createPolicy(
        mode,
        source,
        requireVersion);
    ChangeTracker tracker = new ChangeTracker(URI, policy,
        new AtomicLong(0), objectAttributes);
    if (objectAttributes.getVersionId() == null
        && objectAttributes.getETag() == null) {
      assertFalse("Tracker should not have applied constraints " + tracker,
          tracker.maybeApplyConstraint(newGetObjectRequest()));
    }
    return tracker;
  }

  private GetObjectRequest newGetObjectRequest() {
    return new GetObjectRequest(BUCKET, OBJECT);
  }

  private CopyObjectRequest newCopyObjectRequest() {
    return new CopyObjectRequest(BUCKET, OBJECT, BUCKET, DEST_OBJECT);
  }

  private CopyResult newCopyResult(String eTag, String versionId) {
    CopyResult copyResult = new CopyResult();
    copyResult.setSourceBucketName(BUCKET);
    copyResult.setSourceKey(OBJECT);
    copyResult.setDestinationBucketName(BUCKET);
    copyResult.setDestinationKey(DEST_OBJECT);
    copyResult.setETag(eTag);
    copyResult.setVersionId(versionId);
    return copyResult;
  }

  private S3Object newResponse(String etag, String versionId) {
    ObjectMetadata md = new ObjectMetadata();
    if (etag != null) {
      md.setHeader(Headers.ETAG, etag);
    }
    if (versionId != null) {
      md.setHeader(Headers.S3_VERSION_ID, versionId);
    }
    S3Object response = emptyResponse();
    response.setObjectMetadata(md);
    return response;
  }

  private S3Object emptyResponse() {
    S3Object response = new S3Object();
    response.setBucketName(BUCKET);
    response.setKey(OBJECT);
    return response;
  }

  private S3ObjectAttributes objectAttributes(
      String etag, String versionId) {
    return new S3ObjectAttributes(BUCKET,
        OBJECT,
        null,
        null,
        etag,
        versionId);
  }
}
