text: /**
   * Copyright (c) Codice Foundation
   *
   * <p>This is free software: you can redistribute it and/or modify it under the terms of the GNU
   * Lesser General Public License as published by the Free Software Foundation, either version 3 of
   * the License, or any later version.
   *
   * <p>This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
   * without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
   * GNU Lesser General Public License for more details. A copy of the GNU Lesser General Public
   * License is distributed along with this program and can be found at
   * <http://www.gnu.org/licenses/lgpl.html>.
   */
  package ddf.catalog.resource.download;
  
  import static org.hamcrest.MatcherAssert.assertThat;
  import static org.hamcrest.Matchers.instanceOf;
  import static org.hamcrest.Matchers.is;
  import static org.hamcrest.Matchers.notNullValue;
  import static org.junit.Assert.assertEquals;
  import static org.junit.Assert.assertNotNull;
  import static org.junit.Assert.assertTrue;
  import static org.mockito.ArgumentMatchers.anyLong;
  import static org.mockito.Mockito.mock;
  import static org.mockito.Mockito.reset;
  import static org.mockito.Mockito.timeout;
  import static org.mockito.Mockito.verify;
  import static org.mockito.Mockito.when;
  
  import ddf.catalog.cache.MockInputStream;
  import ddf.catalog.cache.impl.CacheKey;
  import ddf.catalog.cache.impl.ResourceCacheImpl;
  import ddf.catalog.data.Metacard;
  import ddf.catalog.data.impl.MetacardImpl;
  import ddf.catalog.event.retrievestatus.DownloadStatusInfo;
  import ddf.catalog.event.retrievestatus.DownloadStatusInfoImpl;
  import ddf.catalog.event.retrievestatus.DownloadsStatusEventListener;
  import ddf.catalog.event.retrievestatus.DownloadsStatusEventPublisher;
  import ddf.catalog.operation.ResourceRequest;
  import ddf.catalog.operation.ResourceResponse;
  import ddf.catalog.resource.Resource;
  import ddf.catalog.resource.ResourceNotFoundException;
  import ddf.catalog.resource.ResourceNotSupportedException;
  import ddf.catalog.resource.data.ReliableResource;
  import ddf.catalog.resource.download.DownloadManagerState.DownloadState;
  import ddf.catalog.resourceretriever.ResourceRetriever;
  import ddf.security.service.impl.SubjectUtils;
  import java.io.ByteArrayOutputStream;
  import java.io.File;
  import java.io.IOException;
  import java.io.InputStream;
  import java.io.Serializable;
  import java.util.ArrayList;
  import java.util.HashMap;
  import java.util.List;
  import java.util.Map;
  import java.util.concurrent.ExecutorService;
  import java.util.concurrent.Executors;
  import java.util.concurrent.Future;
  import java.util.concurrent.TimeUnit;
  import javax.activation.MimeType;
  import javax.activation.MimeTypeParseException;
  import org.apache.commons.io.FileUtils;
  import org.apache.commons.io.IOUtils;
  import org.codice.ddf.catalog.resource.download.DownloadException;
  import org.junit.Before;
  import org.junit.BeforeClass;
  import org.junit.Ignore;
  import org.junit.Rule;
  import org.junit.Test;
  import org.junit.rules.MethodRule;
  import org.junit.rules.TestWatchman;
  import org.junit.runners.model.FrameworkMethod;
  import org.mockito.ArgumentCaptor;
  import org.mockito.invocation.InvocationOnMock;
  import org.mockito.stubbing.Answer;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  public class ReliableResourceDownloadManagerTest {
    public static final int MONITOR_PERIOD = 5;
  
    public static final String EXPECTED_METACARD_ID = "abc123";
  
    public static final String EXPECTED_METACARD_SOURCE_ID = "ddf-1";
  
    public static final String EXPECTED_CACHE_KEY =
        EXPECTED_METACARD_SOURCE_ID + "-" + EXPECTED_METACARD_ID;
  
    private static final Logger LOGGER =
        LoggerFactory.getLogger(ReliableResourceDownloadManagerTest.class);
  
    private static String productCacheDirectory;
  
    private static String productInputFilename;
  
    private static long expectedFileSize;
  
    private static String expectedFileContents;
  
    private static final String DOWNLOAD_ID_KEY = "downloadId";
  
    private static final String FILE_NAME_KEY = "fileName";
  
    private static final String BYTES_DOWNLOADED_KEY = "bytesDownloaded";
  
    private static final String PERCENT_KEY = "percent";
  
    private static final String USER_KEY = "user";
  
    private static final String STATUS_KEY = "status";
  
    @Rule
    public MethodRule watchman =
        new TestWatchman() {
          public void starting(FrameworkMethod method) {
            LOGGER.debug(
                "***************************  STARTING: {}  **************************\n",
                method.getName());
          }
  
          public void finished(FrameworkMethod method) {
            LOGGER.debug(
                "***************************  END: {}  **************************\n",
                method.getName());
          }
        };
  
    private ResourceCacheImpl resourceCache;
  
    private DownloadsStatusEventPublisher eventPublisher;
  
    private DownloadsStatusEventListener eventListener;
  
    private ReliableResourceDownloadManager downloadMgr;
  
    private ResourceRequest resourceRequest;
  
    private ResourceResponse resourceResponse;
  
    private Resource resource;
  
    private MockInputStream mis;
  
    private InputStream productInputStream;
  
    private ExecutorService executor;
  
    private Future<ByteArrayOutputStream> future;
  
    private DownloadStatusInfoImpl downloadStatusInfo;
  
    @BeforeClass
    public static void oneTimeSetup() throws IOException {
      String workingDir = System.getProperty("user.dir");
      productCacheDirectory = workingDir + "/target/tests/product-cache";
      productInputFilename = workingDir + "/src/test/resources/foo_10_lines.txt";
      File productInputFile = new File(productInputFilename);
      expectedFileSize = productInputFile.length();
      expectedFileContents = FileUtils.readFileToString(productInputFile);
    }
  
    @Before
    public void setup() {
      resourceCache = mock(ResourceCacheImpl.class);
      when(resourceCache.getProductCacheDirectory()).thenReturn(productCacheDirectory);
      eventPublisher = mock(DownloadsStatusEventPublisher.class);
      eventListener = mock(DownloadsStatusEventListener.class);
      downloadStatusInfo = new DownloadStatusInfoImpl();
      downloadStatusInfo.setSubjectOperations(new SubjectUtils());
      downloadMgr =
          new ReliableResourceDownloadManager(
              getDownloaderConfig(), downloadStatusInfo, Executors.newSingleThreadExecutor());
    }
  
    @Test(expected = DownloadException.class)
    public void testDownloadWithNullMetacard() throws Exception {
      resourceRequest = mock(ResourceRequest.class);
      ResourceRetriever retriever = mock(ResourceRetriever.class);
  
      downloadMgr.download(resourceRequest, null, retriever);
    }
  
    @Test(expected = DownloadException.class)
    public void testDownloadWithEmptyMetacardId() throws Exception {
      Metacard metacard = getMockMetacard("", EXPECTED_METACARD_SOURCE_ID);
      resourceRequest = mock(ResourceRequest.class);
      ResourceRetriever retriever = mock(ResourceRetriever.class);
  
      downloadMgr.download(resourceRequest, metacard, retriever);
    }
  
    @Test(expected = DownloadException.class)
    public void testDownloadWithNullResourceRetriever() throws Exception {
      Metacard metacard = getMockMetacard(EXPECTED_METACARD_ID, EXPECTED_METACARD_SOURCE_ID);
      resourceRequest = mock(ResourceRequest.class);
  
      downloadMgr.download(resourceRequest, metacard, null);
    }
  
    @Test(expected = DownloadException.class)
    public void testDownloadWithNullResourceRequest() throws Exception {
      Metacard metacard = getMockMetacard(EXPECTED_METACARD_ID, EXPECTED_METACARD_SOURCE_ID);
      ResourceRetriever retriever = mock(ResourceRetriever.class);
  
      downloadMgr.download(null, metacard, retriever);
    }
  
    @Test(expected = DownloadException.class)
    public void testDownloadResourceNotFound() throws Exception {
      Metacard metacard = getMockMetacard(EXPECTED_METACARD_ID, EXPECTED_METACARD_SOURCE_ID);
      resourceRequest = mock(ResourceRequest.class);
      ResourceRetriever retriever = mock(ResourceRetriever.class);
      when(retriever.retrieveResource()).thenThrow(new ResourceNotFoundException());
  
      downloadMgr.download(resourceRequest, metacard, retriever);
    }
  
    @Test(expected = DownloadException.class)
    public void testDownloadResourceNotSupported() throws Exception {
      Metacard metacard = getMockMetacard(EXPECTED_METACARD_ID, EXPECTED_METACARD_SOURCE_ID);
      resourceRequest = mock(ResourceRequest.class);
      ResourceRetriever retriever = mock(ResourceRetriever.class);
      when(retriever.retrieveResource()).thenThrow(new ResourceNotSupportedException());
  
      downloadMgr.download(resourceRequest, metacard, retriever);
    }
  
    @Test(expected = DownloadException.class)
    public void testDownloadIOException() throws Exception {
      Metacard metacard = getMockMetacard(EXPECTED_METACARD_ID, EXPECTED_METACARD_SOURCE_ID);
      resourceRequest = mock(ResourceRequest.class);
      ResourceRetriever retriever = mock(ResourceRetriever.class);
      when(retriever.retrieveResource()).thenThrow(new IOException());
  
      downloadMgr.download(resourceRequest, metacard, retriever);
    }
  
    @Test
    // @Ignore
    public void testDownloadWithoutCaching() throws Exception {
      mis = new MockInputStream(productInputFilename);
      Metacard metacard = getMockMetacard(EXPECTED_METACARD_ID, EXPECTED_METACARD_SOURCE_ID);
      resourceResponse = getMockResourceResponse();
  
      ResourceRetriever retriever = mock(ResourceRetriever.class);
      when(retriever.retrieveResource()).thenReturn(resourceResponse);
  
      int chunkSize = 50;
      downloadMgr.setChunkSize(chunkSize);
  
      ResourceResponse newResourceResponse =
          downloadMgr.download(resourceRequest, metacard, retriever);
      assertThat(newResourceResponse, is(notNullValue()));
      productInputStream = newResourceResponse.getResource().getInputStream();
      assertThat(productInputStream, is(instanceOf(ReliableResourceInputStream.class)));
  
      ByteArrayOutputStream clientBytesRead = clientRead(chunkSize, productInputStream);
  
      verifyClientBytesRead(clientBytesRead);
  
      cleanup();
    }
  
    @Test
    // @Ignore
    public void testDownloadWithCaching() throws Exception {
      mis = new MockInputStream(productInputFilename);
      Metacard metacard = getMockMetacard(EXPECTED_METACARD_ID, EXPECTED_METACARD_SOURCE_ID);
      resourceResponse = getMockResourceResponse();
  
      ResourceRetriever retriever = mock(ResourceRetriever.class);
      when(retriever.retrieveResource()).thenReturn(resourceResponse);
  
      CacheKey cacheKey = new CacheKey(metacard, resourceResponse.getRequest());
      String key = cacheKey.generateKey();
      when(resourceCache.isPending(key)).thenReturn(false);
  
      int chunkSize = 50;
      startDownload(true, chunkSize, false, metacard, retriever);
  
      ByteArrayOutputStream clientBytesRead = clientRead(chunkSize, productInputStream);
  
      // Captures the ReliableResource object that should have been put in the ResourceCacheImpl's map
      ArgumentCaptor<ReliableResource> argument = ArgumentCaptor.forClass(ReliableResource.class);
      verify(resourceCache).put(argument.capture());
  
      verifyCaching(argument.getValue(), EXPECTED_CACHE_KEY);
  
      verifyClientBytesRead(clientBytesRead);
  
      cleanup();
    }
  
    /**
     * Verifies that if client is reading from @ReliableResourceInputStream slower than {@link
     * ReliableResourceCallable} is reading from product InputStream and writing to
     * FileBackedOutputStream, that complete product is still successfully downloaded by the client.
     * (This will be the case with CXF and @ReliableResourceCallable)
     *
     * @throws Exception
     */
    @Test
    // @Ignore
    public void testDownloadWithCachingDifferentChunkSizes() throws Exception {
      mis = new MockInputStream(productInputFilename);
      Metacard metacard = getMockMetacard(EXPECTED_METACARD_ID, EXPECTED_METACARD_SOURCE_ID);
      resourceResponse = getMockResourceResponse();
  
      ResourceRetriever retriever = mock(ResourceRetriever.class);
      when(retriever.retrieveResource()).thenReturn(resourceResponse);
  
      CacheKey cacheKey = new CacheKey(metacard, resourceResponse.getRequest());
      String key = cacheKey.generateKey();
      when(resourceCache.isPending(key)).thenReturn(false);
  
      int chunkSize = 50;
      startDownload(true, chunkSize, false, metacard, retriever);
  
      int clientChunkSize = 2;
      ByteArrayOutputStream clientBytesRead = clientRead(clientChunkSize, productInputStream);
  
      // Captures the ReliableResource object that should have been put in the ResourceCacheImpl's map
      ArgumentCaptor<ReliableResource> argument = ArgumentCaptor.forClass(ReliableResource.class);
      verify(resourceCache).put(argument.capture());
  
      verifyCaching(argument.getValue(), EXPECTED_CACHE_KEY);
  
      verifyClientBytesRead(clientBytesRead);
  
      cleanup();
    }
  
    /**
     * Test that if an Exception is thrown while reading the product's InputStream that download is
     * interrupted, retried and successfully completes on the second attempt.
     *
     * @throws Exception
     */
    @Test
    @Ignore
    public void testStoreWithInputStreamRecoverableErrorCachingDisabled() throws Exception {
  
      mis = new MockInputStream(productInputFilename);
  
      Metacard metacard = getMockMetacard(EXPECTED_METACARD_ID, EXPECTED_METACARD_SOURCE_ID);
      resourceResponse = getMockResourceResponse();
      ResourceRetriever retriever =
          getMockResourceRetrieverWithRetryCapability(RetryType.INPUT_STREAM_IO_EXCEPTION);
  
      int chunkSize = 50;
      startDownload(false, chunkSize, false, metacard, retriever);
  
      ByteArrayOutputStream clientBytesRead = clientRead(chunkSize, productInputStream);
  
      // Verifies client read same contents as product input file
      verifyClientBytesRead(clientBytesRead);
  
      cleanup();
    }
  
    /**
     * Test that if an Exception is thrown while reading the product's InputStream that download and
     * caching is interrupted, retried and both successfully complete on the second attempt.
     *
     * @throws Exception
     */
    @Test
    @Ignore
    public void testStoreWithInputStreamRecoverableErrorCachingEnabled() throws Exception {
  
      mis = new MockInputStream(productInputFilename);
  
      Metacard metacard = getMockMetacard(EXPECTED_METACARD_ID, EXPECTED_METACARD_SOURCE_ID);
      resourceResponse = getMockResourceResponse();
      ResourceRetriever retriever =
          getMockResourceRetrieverWithRetryCapability(RetryType.INPUT_STREAM_IO_EXCEPTION);
  
      int chunkSize = 50;
      startDownload(true, chunkSize, false, metacard, retriever);
  
      ByteArrayOutputStream clientBytesRead = clientRead(chunkSize, productInputStream);
  
      // Captures the ReliableResource object that should have been put in the ResourceCacheImpl's map
      ArgumentCaptor<ReliableResource> argument = ArgumentCaptor.forClass(ReliableResource.class);
      verify(resourceCache).put(argument.capture());
  
      verifyCaching(argument.getValue(), EXPECTED_CACHE_KEY);
  
      // Verifies client read same contents as product input file
      verifyClientBytesRead(clientBytesRead);
  
      cleanup();
    }
  
    /**
     * Test storing product in cache and one of the chunks being stored takes too long, triggering the
     * CacheMonitor to interrupt the caching. Verify that caching is retried and successfully
     * completes on the second attempt.
     *
     * @throws Exception
     */
    @Test
    // @Ignore
    public void testStoreWithTimeoutExceptionCachingEnabled() throws Exception {
  
      mis = new MockInputStream(productInputFilename);
      Metacard metacard = getMockMetacard(EXPECTED_METACARD_ID, EXPECTED_METACARD_SOURCE_ID);
      resourceResponse = getMockResourceResponse();
      ResourceRetriever retriever =
          getMockResourceRetrieverWithRetryCapability(RetryType.TIMEOUT_EXCEPTION);
  
      int chunkSize = 50;
      startDownload(true, chunkSize, false, metacard, retriever);
  
      ByteArrayOutputStream clientBytesRead = clientRead(chunkSize, productInputStream);
  
      // Captures the ReliableResource object that should have been put in the ResourceCacheImpl's map
      ArgumentCaptor<ReliableResource> argument = ArgumentCaptor.forClass(ReliableResource.class);
      verify(resourceCache).put(argument.capture());
  
      verifyCaching(argument.getValue(), EXPECTED_CACHE_KEY);
  
      verifyClientBytesRead(clientBytesRead);
  
      cleanup();
    }
  
    /**
     * Tests that if user/client cancels a product retrieval that is in progress and actively being
     * cached, and the admin has configured caching to continue, that the caching continues and cached
     * file is placed in the cache map.
     *
     * @throws Exception
     */
    @Test
    @Ignore
    public void testClientCancelProductDownloadCachingContinues() throws Exception {
  
      mis = new MockInputStream(productInputFilename);
      Metacard metacard = getMockMetacard(EXPECTED_METACARD_ID, EXPECTED_METACARD_SOURCE_ID);
      resourceResponse = getMockResourceResponse();
  
      ResourceRetriever retriever =
          getMockResourceRetrieverWithRetryCapability(RetryType.CLIENT_CANCELS_DOWNLOAD);
  
      int chunkSize = 50;
      startDownload(true, chunkSize, true, metacard, retriever);
  
      // On second read of ReliableResourceInputStream, client will close the stream simulating a
      // cancel
      // of the product download
      clientRead(chunkSize, productInputStream, 2);
  
      // Captures the ReliableResource object that should have been put in the ResourceCacheImpl's map
      ArgumentCaptor<ReliableResource> argument = ArgumentCaptor.forClass(ReliableResource.class);
      verify(resourceCache, timeout(3000)).put(argument.capture());
  
      verifyCaching(argument.getValue(), EXPECTED_CACHE_KEY);
  
      cleanup();
    }
  
    /**
     * Tests that if user/client cancels a product retrieval that is in progress and actively being
     * cached, and the admin has not configured caching to continue, that the (partially) cached file
     * is deleted and not placed in the cache map.
     *
     * @throws Exception
     */
    @Test
    @Ignore
    public void testClientCancelProductDownloadCachingStops() throws Exception {
  
      mis = new MockInputStream(productInputFilename, true);
      mis.setReadDelay(MONITOR_PERIOD - 2, TimeUnit.MILLISECONDS);
  
      Metacard metacard = getMockMetacard(EXPECTED_METACARD_ID, EXPECTED_METACARD_SOURCE_ID);
      resourceResponse = getMockResourceResponse();
  
      // Need the product InputStream (MockInputStream) to read slower so that client has time to
      // start reading from the ReliableResourceInputStream and close it, simulating a cancel of
      // the product download
      ResourceRetriever retriever =
          getMockResourceRetrieverWithRetryCapability(RetryType.CLIENT_CANCELS_DOWNLOAD, true);
  
      int chunkSize = 50;
      startDownload(true, chunkSize, false, metacard, retriever);
  
      // On second read of ReliableResourceInputStream, client will close the stream simulating a
      // cancel
      // of the product download
      clientRead(chunkSize, productInputStream, 2);
  
      // Verify product was not cached, i.e., its pending caching entry was removed
      String cacheKey = new CacheKey(metacard, resourceResponse.getRequest()).generateKey();
      verify(resourceCache, timeout(3000)).removePendingCacheEntry(cacheKey);
  
      cleanup();
    }
  
    /**
     * Tests that if network connection drops repeatedly during a product retrieval such that all of
     * the retry attempts are used up before entire product is downloaded, then the (partially) cached
     * file is deleted and not placed in the cache map, and product download fails.
     *
     * @throws Exception
     */
    @Test
    @Ignore
    public void testRetryAttemptsExhaustedDuringProductDownload() throws Exception {
  
      mis = new MockInputStream(productInputFilename);
      Metacard metacard = getMockMetacard(EXPECTED_METACARD_ID, EXPECTED_METACARD_SOURCE_ID);
      resourceResponse = getMockResourceResponse();
  
      ResourceRetriever retriever =
          getMockResourceRetrieverWithRetryCapability(RetryType.NETWORK_CONNECTION_UP_AND_DOWN);
  
      int chunkSize = 50;
      startDownload(true, chunkSize, false, metacard, retriever);
  
      ByteArrayOutputStream clientBytesRead = clientRead(chunkSize, productInputStream);
  
      // Verify client did not receive entire product download
      assertTrue(clientBytesRead.size() < expectedFileSize);
  
      // Verify product was not cached, i.e., its pending caching entry was removed
      String cacheKey = new CacheKey(metacard, resourceResponse.getRequest()).generateKey();
      verify(resourceCache, timeout(3000)).removePendingCacheEntry(cacheKey);
  
      cleanup();
    }
  
    /**
     * Tests that if network connection dropped during a product retrieval that is in progress and
     * actively being cached, that the (partially) cached file is deleted and not placed in the cache
     * map.
     *
     * @throws Exception
     */
    @Test
    @Ignore
    public void testNetworkConnectionDroppedDuringProductDownload() throws Exception {
  
      mis = new MockInputStream(productInputFilename);
      Metacard metacard = getMockMetacard(EXPECTED_METACARD_ID, EXPECTED_METACARD_SOURCE_ID);
      resourceResponse = getMockResourceResponse();
  
      ResourceRetriever retriever =
          getMockResourceRetrieverWithRetryCapability(RetryType.NETWORK_CONNECTION_DROPPED);
  
      int chunkSize = 50;
      startDownload(true, chunkSize, false, metacard, retriever);
  
      ByteArrayOutputStream clientBytesRead = clientRead(chunkSize, productInputStream);
  
      // Verify client did not receive entire product download
      assertTrue(clientBytesRead.size() < expectedFileSize);
  
      // Verify product was not cached, i.e., its pending caching entry was removed
      String cacheKey = new CacheKey(metacard, resourceResponse.getRequest()).generateKey();
      verify(resourceCache, timeout(3000)).removePendingCacheEntry(cacheKey);
  
      cleanup();
    }
  
    /**
     * Tests that if exception with file being cached to occurs during a product retrieval, then the
     * (partially) cached file is deleted and it is not placed in the cache map, but the product
     * continues to be streamed to the client until the EOF is detected.
     *
     * @throws Exception
     */
    @Test
    @Ignore
    public void testCacheFileExceptionDuringProductDownload() throws Exception {
  
      // Need the product InputStream (MockInputStream) to read slower so that client has time to
      // start reading from the ReliableResourceInputStream and close the FileOutputStream the
      // download manager is writing to, simulating a cache file exception during
      // the product download
      mis = new MockInputStream(productInputFilename, true);
      mis.setReadDelay(MONITOR_PERIOD - 2, TimeUnit.MILLISECONDS);
  
      Metacard metacard = getMockMetacard(EXPECTED_METACARD_ID, EXPECTED_METACARD_SOURCE_ID);
      resourceResponse = getMockResourceResponse();
  
      ResourceRetriever retriever = mock(ResourceRetriever.class);
      when(retriever.retrieveResource()).thenReturn(resourceResponse);
  
      int chunkSize = 2;
      startDownload(true, chunkSize, false, metacard, retriever);
  
      // On second chunk read by client it will close the download manager's cache file output stream
      // to simulate a cache file exception that should be detected by the ReliableResourceCallable
      executor = Executors.newCachedThreadPool();
      ProductDownloadClient productDownloadClient =
          new ProductDownloadClient(productInputStream, chunkSize);
      productDownloadClient.setSimulateCacheFileException(2, downloadMgr);
      future = executor.submit(productDownloadClient);
      ByteArrayOutputStream clientBytesRead = future.get();
  
      verifyClientBytesRead(clientBytesRead);
  
      // Verify product was not cached, i.e., its pending caching entry was removed
      String cacheKey = new CacheKey(metacard, resourceResponse.getRequest()).generateKey();
      verify(resourceCache, timeout(3000)).removePendingCacheEntry(cacheKey);
  
      cleanup();
    }
  
    /**
     * Tests that if exception with the FileBackedOutputStream being written to and concurrently read
     * by the client occurs during a product retrieval, then the product download to the client is
     * stopped, but the caching of the file continues.
     *
     * @throws Exception
     */
    @Test
    @Ignore
    // Currently Ignored because cannot figure out how to get FileBackedOutputStream (FBOS) to throw
    // exception
    // during product download - this test successfully closes the FBOS, but the
    // ReliableResourceCallable
    // does not seem to detect this and continues to stream successfully to the client.
    public void testStreamToClientExceptionDuringProductDownloadCachingEnabled() throws Exception {
  
      mis = new MockInputStream(productInputFilename);
  
      Metacard metacard = getMockMetacard(EXPECTED_METACARD_ID, EXPECTED_METACARD_SOURCE_ID);
      resourceResponse = getMockResourceResponse();
  
      downloadMgr =
          new ReliableResourceDownloadManager(
              getDownloaderConfig(), downloadStatusInfo, Executors.newSingleThreadExecutor());
  
      // Use small chunk size so download takes long enough for client
      // to have time to simulate FileBackedOutputStream exception
      int chunkSize = 2;
      downloadMgr.setChunkSize(chunkSize);
  
      ResourceRetriever retriever = mock(ResourceRetriever.class);
      when(retriever.retrieveResource()).thenReturn(resourceResponse);
  
      ArgumentCaptor<ReliableResource> argument = ArgumentCaptor.forClass(ReliableResource.class);
  
      ResourceResponse newResourceResponse =
          downloadMgr.download(resourceRequest, metacard, retriever);
      assertThat(newResourceResponse, is(notNullValue()));
      productInputStream = newResourceResponse.getResource().getInputStream();
      assertThat(productInputStream, is(instanceOf(ReliableResourceInputStream.class)));
  
      // On second chunk read by client it will close the download manager's cache file output stream
      // to simulate a cache file exception that should be detected by the ReliableResourceCallable
      executor = Executors.newCachedThreadPool();
      ProductDownloadClient productDownloadClient =
          new ProductDownloadClient(productInputStream, chunkSize);
      productDownloadClient.setSimulateFbosException(chunkSize, downloadMgr);
      future = executor.submit(productDownloadClient);
      ByteArrayOutputStream clientBytesRead = future.get();
  
      // Verify client did not receive entire product download
      assertTrue(clientBytesRead.size() < expectedFileSize);
  
      // Captures the ReliableResource object that should have been put in the ResourceCacheImpl's map
      verify(resourceCache, timeout(3000)).put(argument.capture());
  
      verifyCaching(argument.getValue(), EXPECTED_CACHE_KEY);
  
      cleanup();
    }
  
    @Test
    public void testGetDownloadsInProgress() {
      List<String> downloadIds = new ArrayList<>();
      downloadIds.add("03e3f850-240c-4cc6-a5a3-318dc34ad4bd");
      downloadIds.add("5tygH67-345t-3er5-r86y-5tyZHU7UGD092");
      downloadIds.add("9tY75f4-345t-4er8-jj87-Y67hJJK098yaq");
      downloadIds.add("56lJOJl-45gg-3wf5-ww23-tldf7HewfhweJ");
      downloadIds.add("rRdlefj-ggal-erty-rr6e-ZZefoeje546kL");
  
      List<Map<String, String>> downloadStatusMaps = getDownloadStatusMaps(downloadIds);
  
      DownloadStatusInfo downloadStatusInfo = mock(DownloadStatusInfo.class);
      when(downloadStatusInfo.getAllDownloads()).thenReturn(downloadIds);
      when(downloadStatusInfo.getDownloadStatus(downloadIds.get(0)))
          .thenReturn(downloadStatusMaps.get(0));
      when(downloadStatusInfo.getDownloadStatus(downloadIds.get(1)))
          .thenReturn(downloadStatusMaps.get(1));
      when(downloadStatusInfo.getDownloadStatus(downloadIds.get(2)))
          .thenReturn(downloadStatusMaps.get(2));
      when(downloadStatusInfo.getDownloadStatus(downloadIds.get(3)))
          .thenReturn(downloadStatusMaps.get(3));
      when(downloadStatusInfo.getDownloadStatus(downloadIds.get(4)))
          .thenReturn(downloadStatusMaps.get(4));
  
      ReliableResourceDownloadManager reliableResorceDownloadManager =
          new ReliableResourceDownloadManager(null, downloadStatusInfo, null);
  
      List<DownloadInfo> downloadInfoList = reliableResorceDownloadManager.getDownloadsInProgress();
  
      assertThat(downloadInfoList.size(), is(2));
      for (DownloadInfo downloadInfo : downloadInfoList) {
        assertThat(downloadInfo.isDownloadInState(DownloadState.IN_PROGRESS), is(true));
      }
    }
  
    private List<Map<String, String>> getDownloadStatusMaps(List<String> downloadIds) {
      List<Map<String, String>> downloads = new ArrayList<>();
      Map<String, String> downloadStatus1 =
          createDownloadStatusMap(
              downloadIds.get(0),
              "image1.jpg",
              DownloadState.IN_PROGRESS.name(),
              "862978048",
              "76",
              "admin");
      downloads.add(downloadStatus1);
      Map<String, String> downloadStatus2 =
          createDownloadStatusMap(
              downloadIds.get(1),
              "image2.jpg",
              DownloadState.IN_PROGRESS.name(),
              "456212",
              "21",
              "localhost");
      downloads.add(downloadStatus2);
      Map<String, String> downloadStatus3 =
          createDownloadStatusMap(
              downloadIds.get(2),
              "image3.jpg",
              DownloadState.COMPLETED.name(),
              "3487",
              "11",
              "andrewreynolds");
      downloads.add(downloadStatus3);
      Map<String, String> downloadStatus4 =
          createDownloadStatusMap(
              downloadIds.get(3),
              "image4.jpg",
              DownloadState.FAILED.name(),
              "6567544543",
              "11",
              "markjohnson");
      downloads.add(downloadStatus4);
      Map<String, String> downloadStatus5 =
          createDownloadStatusMap(
              downloadIds.get(4),
              "image5.jpg",
              DownloadState.CANCELED.name(),
              "243",
              "1",
              "chriscole");
      downloads.add(downloadStatus5);
  
      return downloads;
    }
  
    private Map<String, String> createDownloadStatusMap(
        String downloadId,
        String fileName,
        String status,
        String bytesDownloaded,
        String percent,
        String user) {
      Map<String, String> downloadStatus1 = new HashMap<>();
      downloadStatus1.put(DOWNLOAD_ID_KEY, downloadId);
      downloadStatus1.put(FILE_NAME_KEY, fileName);
      downloadStatus1.put(STATUS_KEY, status);
      downloadStatus1.put(BYTES_DOWNLOADED_KEY, bytesDownloaded);
      downloadStatus1.put(PERCENT_KEY, percent);
      downloadStatus1.put(USER_KEY, user);
      return downloadStatus1;
    }
  
    private void startDownload(
        boolean cacheEnabled,
        int chunkSize,
        boolean cacheWhenCanceled,
        Metacard metacard,
        ResourceRetriever retriever)
        throws Exception {
      //        downloadMgr = new ReliableResourceDownloadManager(getDownloaderConfig());
      downloadMgr.setCacheEnabled(cacheEnabled);
      downloadMgr.setChunkSize(chunkSize);
      downloadMgr.setCacheWhenCanceled(cacheWhenCanceled);
  
      ResourceResponse newResourceResponse =
          downloadMgr.download(resourceRequest, metacard, retriever);
      assertThat(newResourceResponse, is(notNullValue()));
      productInputStream = newResourceResponse.getResource().getInputStream();
      assertThat(productInputStream, is(instanceOf(ReliableResourceInputStream.class)));
    }
  
    /////////////////////////////////////////////////////////////////////////////////////////////////////
  
    private void verifyCaching(ReliableResource reliableResource, String expectedCacheKey)
        throws IOException {
      assertEquals(expectedCacheKey, reliableResource.getKey());
      byte[] cachedData = reliableResource.getByteArray();
      assertNotNull(cachedData);
  
      // Verifies cached data read in was same contents as product input file
      assertTrue(cachedData.length == expectedFileSize);
      assertEquals(expectedFileContents, new String(cachedData));
  
      // Verifies cached file on disk has same contents as product input file
      assertEquals(expectedFileContents, IOUtils.toString(reliableResource.getInputStream()));
    }
  
    private void verifyClientBytesRead(ByteArrayOutputStream clientBytesRead) {
  
      // Verifies client read same contents as product input file
      if (clientBytesRead != null) {
        assertThat(clientBytesRead.size(), is(Long.valueOf(expectedFileSize).intValue()));
        assertEquals(expectedFileContents, new String(clientBytesRead.toByteArray()));
      }
    }
  
    private Metacard getMockMetacard(String id, String source) {
  
      Metacard metacard = mock(Metacard.class);
  
      when(metacard.getId()).thenReturn(id);
  
      when(metacard.getSourceId()).thenReturn(source);
  
      when(metacard.getMetacardType()).thenReturn(MetacardImpl.BASIC_METACARD);
  
      return metacard;
    }
  
    private ResourceResponse getMockResourceResponse() throws Exception {
      resourceRequest = mock(ResourceRequest.class);
      Map<String, Serializable> requestProperties = new HashMap<String, Serializable>();
      when(resourceRequest.getPropertyNames()).thenReturn(requestProperties.keySet());
  
      resource = mock(Resource.class);
      when(resource.getInputStream()).thenReturn(mis);
      when(resource.getName()).thenReturn("test-resource");
      when(resource.getMimeType()).thenReturn(new MimeType("text/plain"));
  
      resourceResponse = mock(ResourceResponse.class);
      when(resourceResponse.getRequest()).thenReturn(resourceRequest);
      when(resourceResponse.getResource()).thenReturn(resource);
      Map<String, Serializable> responseProperties = new HashMap<String, Serializable>();
      when(resourceResponse.getProperties()).thenReturn(responseProperties);
  
      return resourceResponse;
    }
  
    private ResourceRetriever getMockResourceRetrieverWithRetryCapability(final RetryType retryType)
        throws Exception {
      return getMockResourceRetrieverWithRetryCapability(retryType, false);
    }
  
    private ResourceRetriever getMockResourceRetrieverWithRetryCapability(
        final RetryType retryType, final boolean readSlow) throws Exception {
  
      // Mocking to support re-retrieval of product when error encountered
      // during caching.
      ResourceRetriever retriever = mock(ResourceRetriever.class);
      when(retriever.retrieveResource())
          .thenAnswer(
              new Answer<Object>() {
                int invocationCount = 0;
  
                public Object answer(InvocationOnMock invocation) throws ResourceNotFoundException {
                  // Create new InputStream for retrieving the same product. This
                  // simulates re-retrieving the product from the remote source.
                  invocationCount++;
                  if (readSlow) {
                    mis = new MockInputStream(productInputFilename, true);
                    mis.setReadDelay(MONITOR_PERIOD - 2, TimeUnit.MILLISECONDS);
                  } else {
                    mis = new MockInputStream(productInputFilename);
                  }
  
                  if (retryType == RetryType.INPUT_STREAM_IO_EXCEPTION) {
                    if (invocationCount == 1) {
                      mis.setInvocationCountToThrowIOException(5);
                    } else {
                      mis.setInvocationCountToThrowIOException(-1);
                    }
                  } else if (retryType == RetryType.TIMEOUT_EXCEPTION) {
                    if (invocationCount == 1) {
                      mis.setInvocationCountToTimeout(3);
                      mis.setReadDelay(MONITOR_PERIOD * 2, TimeUnit.SECONDS);
                    } else {
                      mis.setInvocationCountToTimeout(-1);
                      mis.setReadDelay(0, TimeUnit.SECONDS);
                    }
                  } else if (retryType == RetryType.NETWORK_CONNECTION_UP_AND_DOWN) {
                    mis.setInvocationCountToThrowIOException(2);
                  } else if (retryType == RetryType.NETWORK_CONNECTION_DROPPED) {
                    if (invocationCount == 1) {
                      mis.setInvocationCountToThrowIOException(2);
                    } else {
                      throw new ResourceNotFoundException();
                    }
                  }
  
                  // Reset the mock Resource so that it can be reconfigured to return
                  // the new InputStream
                  reset(resource);
                  when(resource.getInputStream()).thenReturn(mis);
                  when(resource.getName()).thenReturn("test-resource");
                  try {
                    when(resource.getMimeType()).thenReturn(new MimeType("text/plain"));
                  } catch (MimeTypeParseException e) {
                  }
  
                  // Reset the mock ResourceResponse so that it can be reconfigured to return
                  // the new Resource
                  reset(resourceResponse);
                  when(resourceResponse.getRequest()).thenReturn(resourceRequest);
                  when(resourceResponse.getResource()).thenReturn(resource);
                  when(resourceResponse.getProperties())
                      .thenReturn(new HashMap<String, Serializable>());
  
                  return resourceResponse;
                }
              });
  
      ArgumentCaptor<Long> bytesReadArg = ArgumentCaptor.forClass(Long.class);
  
      // Mocking to support re-retrieval of product when error encountered
      // during caching. This resource retriever supports skipping.
      when(retriever.retrieveResource(anyLong()))
          .thenAnswer(
              new Answer<Object>() {
                int invocationCount = 0;
  
                public Object answer(InvocationOnMock invocation)
                    throws ResourceNotFoundException, IOException {
                  // Create new InputStream for retrieving the same product. This
                  // simulates re-retrieving the product from the remote source.
                  invocationCount++;
                  if (readSlow) {
                    mis = new MockInputStream(productInputFilename, true);
                    mis.setReadDelay(MONITOR_PERIOD - 2, TimeUnit.MILLISECONDS);
                  } else {
                    mis = new MockInputStream(productInputFilename);
                  }
  
                  // Skip the number of bytes that have already been read
                  Object[] args = invocation.getArguments();
                  long bytesToSkip = (Long) args[0];
  
                  mis.skip(bytesToSkip);
  
                  if (retryType == RetryType.INPUT_STREAM_IO_EXCEPTION) {
                    if (invocationCount == 1) {
                      mis.setInvocationCountToThrowIOException(5);
                    } else {
                      mis.setInvocationCountToThrowIOException(-1);
                    }
                  } else if (retryType == RetryType.TIMEOUT_EXCEPTION) {
                    if (invocationCount == 1) {
                      mis.setInvocationCountToTimeout(3);
                      mis.setReadDelay(MONITOR_PERIOD * 2, TimeUnit.SECONDS);
                    } else {
                      mis.setInvocationCountToTimeout(-1);
                      mis.setReadDelay(0, TimeUnit.MILLISECONDS);
                    }
                  } else if (retryType == RetryType.NETWORK_CONNECTION_UP_AND_DOWN) {
                    mis.setInvocationCountToThrowIOException(2);
                  } else if (retryType == RetryType.NETWORK_CONNECTION_DROPPED) {
                    if (invocationCount == 1) {
                      mis.setInvocationCountToThrowIOException(2);
                    } else {
                      throw new ResourceNotFoundException();
                    }
                  }
  
                  // Reset the mock Resource so that it can be reconfigured to return
                  // the new InputStream
                  reset(resource);
                  when(resource.getInputStream()).thenReturn(mis);
                  when(resource.getName()).thenReturn("test-resource");
                  try {
                    when(resource.getMimeType()).thenReturn(new MimeType("text/plain"));
                  } catch (MimeTypeParseException e) {
                  }
  
                  // Reset the mock ResourceResponse so that it can be reconfigured to return
                  // the new Resource
                  reset(resourceResponse);
                  when(resourceResponse.getRequest()).thenReturn(resourceRequest);
                  when(resourceResponse.getResource()).thenReturn(resource);
                  Map<String, Serializable> responseProperties = new HashMap<>();
                  responseProperties.put("BytesSkipped", true);
                  when(resourceResponse.getProperties()).thenReturn(responseProperties);
                  when(resourceResponse.containsPropertyName("BytesSkipped")).thenReturn(true);
                  when(resourceResponse.getPropertyValue("BytesSkipped")).thenReturn(true);
  
                  return resourceResponse;
                }
              });
  
      return retriever;
    }
  
    public ByteArrayOutputStream clientRead(int chunkSize, InputStream is) throws Exception {
      return clientRead(chunkSize, is, -1);
    }
  
    public ByteArrayOutputStream clientRead(
        int chunkSize, InputStream is, int simulatedCancelChunkCount) throws Exception {
      executor = Executors.newCachedThreadPool();
      ProductDownloadClient productDownloadClient =
          new ProductDownloadClient(is, chunkSize, simulatedCancelChunkCount);
      future = executor.submit(productDownloadClient);
      ByteArrayOutputStream clientBytesRead = future.get();
  
      return clientBytesRead;
    }
  
    private void cleanup() {
      try {
        IOUtils.closeQuietly(productInputStream);
        FileUtils.deleteDirectory(new File(productCacheDirectory));
      } catch (IOException e) {
      }
      future.cancel(true);
      executor.shutdownNow();
    }
  
    private ReliableResourceDownloaderConfig getDownloaderConfig() {
      ReliableResourceDownloaderConfig downloaderConfig = new ReliableResourceDownloaderConfig();
      downloaderConfig.setResourceCache(resourceCache);
      downloaderConfig.setEventPublisher(eventPublisher);
      downloaderConfig.setEventListener(eventListener);
      return downloaderConfig;
    }
  
    private enum RetryType {
      INPUT_STREAM_IO_EXCEPTION,
      TIMEOUT_EXCEPTION,
      NETWORK_CONNECTION_UP_AND_DOWN,
      NETWORK_CONNECTION_DROPPED,
      CLIENT_CANCELS_DOWNLOAD,
      CACHE_FILE_EXCEPTION
    }
  }
  