package com.atlassian.vcache.internal.test;
   
   import com.atlassian.vcache.DirectExternalCache;
   import com.atlassian.vcache.PutPolicy;
   import com.atlassian.vcache.TransactionalExternalCache;
   import com.google.common.collect.Maps;
   import org.junit.Rule;
   import org.junit.Test;
   import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.util.Map;
  import java.util.Optional;
  import java.util.concurrent.CompletionStage;
  import java.util.stream.Collectors;
  
  import static com.atlassian.vcache.VCacheUtils.unsafeJoin;
  import static com.atlassian.vcache.internal.test.CompletionStageSuccessful.successful;
  import static com.atlassian.vcache.internal.test.CompletionStageSuccessful.successfulWith;
  import static org.hamcrest.Matchers.containsInAnyOrder;
  import static org.hamcrest.Matchers.is;
  import static org.hamcrest.Matchers.not;
  import static org.junit.Assert.assertThat;
  
  /**
   * Base test class for the {@link com.atlassian.vcache.TransactionalExternalCache}.
   */
  public abstract class AbstractTransactionalExternalCacheIT {
      protected static final String CACHE_NAME = "fragileSting";
      private static final Logger log = LoggerFactory.getLogger(AbstractStableReadExternalCacheIT.class);
  
      @Rule
      public LoggingTestWatcher watcher = new LoggingTestWatcher(log);
  
      protected abstract TransactionalExternalCache<String> cache();
  
      protected abstract DirectExternalCache<String> directCache();
  
      protected abstract void directCacheRefresh();
  
      protected abstract void cacheTransactionSync();
      protected abstract void cacheTransactionDiscard();
  
      @Test
      public void put_get_sync_successful2() {
          cache().put("claira", "dancing", PutPolicy.PUT_ALWAYS);
  
          final CompletionStage<Optional<String>> get1 = cache().get("claira");
  
          assertThat(get1, successfulWith(is(Optional.of("dancing"))));
  
          // Now use the directCache to verify no data has been written
          directCacheRefresh();
          final CompletionStage<Optional<String>> get2 = directCache().get("claira");
  
          assertThat(get2, successfulWith(is(Optional.empty())));
  
          // Now sync the changes
          cacheTransactionSync();
  
          // And check the changes are now visible
          directCacheRefresh();
          final CompletionStage<Optional<String>> get3 = directCache().get("claira");
  
          assertThat(get3, successfulWith(is(Optional.of("dancing"))));
  
          // Now check another lookup will succeed with the transactional cache
          final CompletionStage<Optional<String>> get4 = cache().get("claira");
  
          assertThat(get4, successfulWith(is(Optional.of("dancing"))));
      }
  
      @Test
      public void put_get_sync_unsuccessful2() throws Exception {
          cache().put("claira", "dancing", PutPolicy.ADD_ONLY); // Must be ADD_ONLY
  
          final CompletionStage<Optional<String>> get1 = cache().get("claira");
  
          assertThat(get1, successfulWith(is(Optional.of("dancing"))));
  
          // Now use the directCache to verify no data has been written
          directCacheRefresh();
          final CompletionStage<Optional<String>> get2 = directCache().get("claira");
  
          assertThat(get2, successfulWith(is(Optional.empty())));
  
          // Now add values to the backing cache using the directCache
          final CompletionStage<Map<String, String>> get3 =
                  directCache().getBulk(keys -> Maps.asMap(keys, k -> k + "-1"), "claira", "josephine");
  
          assertThat(get3, successful());
          assertThat(unsafeJoin(get3).keySet(), containsInAnyOrder("claira", "josephine"));
          assertThat(unsafeJoin(get3).values(), containsInAnyOrder("claira-1", "josephine-1"));
  
          // Now sync the changes
          cacheTransactionSync();
  
          // And check the cache is empty
          directCacheRefresh();
         final CompletionStage<Map<String, Optional<String>>> get4 = directCache().getBulk("claira", "josephine");
 
         assertThat(get4, successful());
         assertThat(unsafeJoin(get4).keySet(), containsInAnyOrder("claira", "josephine"));
         assertThat(unsafeJoin(get4).values(), containsInAnyOrder(Optional.empty(), Optional.empty()));
     }
 
     @Test
     public void getBulk_getBulkFunction_directPut_sync() throws Exception {
         final CompletionStage<Boolean> put1 = directCache().put("claira", "dancing", PutPolicy.PUT_ALWAYS);
 
         assertThat(put1, successful());
 
         final CompletionStage<Map<String, Optional<String>>> get1 = cache().getBulk("claira", "josephine");
 
         assertThat(get1, successful());
         assertThat(unsafeJoin(get1).keySet(), containsInAnyOrder("claira", "josephine"));
         assertThat(unsafeJoin(get1).values(), containsInAnyOrder(Optional.of("dancing"), Optional.empty()));
 
         final CompletionStage<Map<String, String>> get2 =
                 cache().getBulk(keys -> Maps.asMap(keys, k -> "football"), "claira", "josephine");
 
         assertThat(get2, successful());
         assertThat(unsafeJoin(get2).keySet(), containsInAnyOrder("claira", "josephine"));
         assertThat(unsafeJoin(get2).values(), containsInAnyOrder("dancing", "football"));
 
         // Now add an entry to the backend cache for "josephine" and verify one does not exist yet
         directCacheRefresh();
         final CompletionStage<Boolean> put2 = directCache().put("josephine", "soccer", PutPolicy.ADD_ONLY);
 
         assertThat(put2, successfulWith(is(true)));
 
         cacheTransactionSync();
 
         // Now verify nothing in the cache, due to panic when added josephine
         directCacheRefresh();
         final CompletionStage<Map<String, Optional<String>>> get3 = directCache().getBulk("claira", "josephine");
 
         assertThat(get3, successful());
         assertThat(unsafeJoin(get3).keySet(), containsInAnyOrder("claira", "josephine"));
         assertThat(unsafeJoin(get3).values(), containsInAnyOrder(Optional.empty(), Optional.empty()));
     }
 
     @Test
     public void put_get_remove_get() {
         cache().put("nirvana", "shotgun", PutPolicy.PUT_ALWAYS);
 
         final CompletionStage<Optional<String>> get1 = cache().get("nirvana");
 
         assertThat(get1, successfulWith(is(Optional.of("shotgun"))));
 
         cache().remove("a", "b", "nirvana");
 
         final CompletionStage<Optional<String>> get2 = cache().get("nirvana");
 
         assertThat(get2, successfulWith(is(Optional.empty())));
     }
 
     @Test
     public void getSupplier_removeAll_get() {
         final CompletionStage<String> get1 = cache().get("vienna", () -> "sandwiches");
 
         assertThat(get1, successfulWith(is("sandwiches")));
 
         cache().removeAll();
 
         final CompletionStage<Optional<String>> get2 = cache().get("vienna");
 
         assertThat(get2, successfulWith(is(Optional.empty())));
     }
 
     @Test
     public void put_get_discard() {
         cache().put("josie", "surfing", PutPolicy.PUT_ALWAYS);
 
         final CompletionStage<Optional<String>> get1 = cache().get("josie");
 
         assertThat(get1, successfulWith(is(Optional.of("surfing"))));
 
         // Now use the directCache to verify no data has been written
         directCacheRefresh();
         final CompletionStage<Optional<String>> get2 = directCache().get("josie");
 
         assertThat(get2, successfulWith(is(Optional.empty())));
 
         // Now discard the changes
         cacheTransactionDiscard();
 
         // And check the changes are not visible
         directCacheRefresh();
         final CompletionStage<Optional<String>> get3 = directCache().get("josie");
 
         assertThat(get3, successfulWith(is(Optional.empty())));
     }
 
     @Test
     public void check_get_null_detection() {
         assertThat(cache().get("kenny", () -> null), not(successful()));
         assertThat(cache().getBulk(
                 strings -> strings.stream().collect(Collectors.toMap(k -> k, k -> null)),
                 "extra"),
                 not(successful()));
     }
 
     @SuppressWarnings("ConstantConditions")
     @Test(expected = NullPointerException.class)
     public void check_put_null_detection() {
         cache().put("key", null, PutPolicy.ADD_ONLY);
     }
 }