/*
 * 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.geode.management.internal.cli.functions;

import static org.apache.geode.cache.Region.SEPARATOR;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.Mockito.atLeastOnce;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;

import java.io.File;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;
import java.util.UUID;

import org.assertj.core.api.AssertionsForClassTypes;
import org.junit.Before;
import org.junit.Test;

import org.apache.geode.cache.Cache;
import org.apache.geode.cache.DataPolicy;
import org.apache.geode.cache.DiskStore;
import org.apache.geode.cache.EvictionAction;
import org.apache.geode.cache.EvictionAttributes;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionAttributes;
import org.apache.geode.cache.asyncqueue.AsyncEventQueue;
import org.apache.geode.cache.execute.FunctionContext;
import org.apache.geode.cache.execute.ResultSender;
import org.apache.geode.cache.server.CacheServer;
import org.apache.geode.cache.server.ClientSubscriptionConfig;
import org.apache.geode.cache.wan.GatewaySender;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.lang.ObjectUtils;
import org.apache.geode.internal.util.CollectionUtils;
import org.apache.geode.management.internal.cli.domain.DiskStoreDetails;
import org.apache.geode.management.internal.exceptions.EntityNotFoundException;

/**
 * The DescribeDiskStoreFunctionJUnitTest test suite class tests the contract and functionality of
 * the DescribeDiskStoreFunction class.
 *
 * @see org.apache.geode.cache.DiskStore
 * @see org.apache.geode.management.internal.cli.domain.DiskStoreDetails
 * @see org.apache.geode.management.internal.cli.functions.DescribeDiskStoreFunction
 * @see org.junit.Test
 * @since GemFire 7.0
 */
public class DescribeDiskStoreFunctionJUnitTest {
  private InternalCache mockCache;

  @Before
  public void setup() {
    mockCache = mock(InternalCache.class, "Cache");
  }

  private void assertAsyncEventQueueDetails(
      final Set<DiskStoreDetails.AsyncEventQueueDetails> expectedAsyncEventQueueDetailsSet,
      final DiskStoreDetails diskStoreDetails) {
    int actualCount = 0;

    for (final DiskStoreDetails.AsyncEventQueueDetails actualAsyncEventQueueDetails : diskStoreDetails
        .iterateAsyncEventQueues()) {
      final DiskStoreDetails.AsyncEventQueueDetails expectedAsyncEventQueueDetails = CollectionUtils
          .findBy(expectedAsyncEventQueueDetailsSet, asyncEventQueueDetails -> ObjectUtils
              .equals(asyncEventQueueDetails.getId(), actualAsyncEventQueueDetails.getId()));

      assertThat(expectedAsyncEventQueueDetails).isNotNull();
      actualCount++;
    }

    assertThat(actualCount).isEqualTo(expectedAsyncEventQueueDetailsSet.size());
  }

  private void assertCacheServerDetails(
      final Set<DiskStoreDetails.CacheServerDetails> expectedCacheServerDetailsSet,
      final DiskStoreDetails diskStoreDetails) {
    int actualCount = 0;

    for (final DiskStoreDetails.CacheServerDetails actualCacheServerDetails : diskStoreDetails
        .iterateCacheServers()) {
      final DiskStoreDetails.CacheServerDetails expectedCacheServerDetails =
          CollectionUtils.findBy(expectedCacheServerDetailsSet,
              cacheServerDetails -> ObjectUtils.equals(cacheServerDetails.getBindAddress(),
                  actualCacheServerDetails.getBindAddress())
                  && ObjectUtils.equals(cacheServerDetails.getPort(),
                      actualCacheServerDetails.getPort()));

      assertThat(expectedCacheServerDetails).isNotNull();
      assertThat(actualCacheServerDetails.getHostName())
          .isEqualTo(expectedCacheServerDetails.getHostName());
      actualCount++;
    }

    assertThat(actualCount).isEqualTo(expectedCacheServerDetailsSet.size());
  }

  private void assertGatewayDetails(
      final Set<DiskStoreDetails.GatewayDetails> expectedGatewayDetailsSet,
      final DiskStoreDetails diskStoreDetails) {
    int actualCount = 0;

    for (final DiskStoreDetails.GatewayDetails actualGatewayDetails : diskStoreDetails
        .iterateGateways()) {
      DiskStoreDetails.GatewayDetails expectedGatewayDetails =
          CollectionUtils.findBy(expectedGatewayDetailsSet, gatewayDetails -> ObjectUtils
              .equals(gatewayDetails.getId(), actualGatewayDetails.getId()));

      assertThat(expectedGatewayDetails).isNotNull();
      assertThat(actualGatewayDetails.isPersistent())
          .isEqualTo(expectedGatewayDetails.isPersistent());
      actualCount++;
    }

    assertThat(actualCount).isEqualTo(expectedGatewayDetailsSet.size());
  }

  private void assertRegionDetails(
      final Set<DiskStoreDetails.RegionDetails> expectedRegionDetailsSet,
      final DiskStoreDetails diskStoreDetails) {
    int actualCount = 0;

    for (final DiskStoreDetails.RegionDetails actualRegionDetails : diskStoreDetails
        .iterateRegions()) {
      final DiskStoreDetails.RegionDetails expectedRegionDetails =
          CollectionUtils.findBy(expectedRegionDetailsSet, regionDetails -> ObjectUtils
              .equals(regionDetails.getFullPath(), actualRegionDetails.getFullPath()));

      assertThat(expectedRegionDetails).isNotNull();
      assertThat(actualRegionDetails.getName()).isEqualTo(expectedRegionDetails.getName());
      assertThat(actualRegionDetails.isPersistent())
          .isEqualTo(expectedRegionDetails.isPersistent());
      assertThat(actualRegionDetails.isOverflowToDisk())
          .isEqualTo(expectedRegionDetails.isOverflowToDisk());
      actualCount++;
    }

    assertThat(actualCount).isEqualTo(expectedRegionDetailsSet.size());
  }

  private DiskStoreDetails.AsyncEventQueueDetails createAsyncEventQueueDetails(final String id) {
    return new DiskStoreDetails.AsyncEventQueueDetails(id);
  }

  private DiskStoreDetails.CacheServerDetails createCacheServerDetails(final String bindAddress,
      final int port, final String hostname) {
    final DiskStoreDetails.CacheServerDetails cacheServerDetails =
        new DiskStoreDetails.CacheServerDetails(bindAddress, port);
    cacheServerDetails.setHostName(hostname);

    return cacheServerDetails;
  }

  private File[] createFileArray(final String... locations) {
    assert locations != null : "The locations argument cannot be null!";

    final File[] directories = new File[locations.length];
    int index = 0;

    for (final String location : locations) {
      directories[index++] = new File(location);
    }

    return directories;
  }

  private DiskStoreDetails.GatewayDetails createGatewayDetails(final String id,
      final boolean persistent) {
    DiskStoreDetails.GatewayDetails gatewayDetails = new DiskStoreDetails.GatewayDetails(id);
    gatewayDetails.setPersistent(persistent);
    return gatewayDetails;
  }

  private int[] createIntArray(final int... array) {
    assert array != null : "The array of int values cannot be null!";
    return array;
  }

  private DiskStore createMockDiskStore(final UUID diskStoreId, final String name,
      final boolean allowForceCompaction, final boolean autoCompact, final int compactionThreshold,
      final long maxOplogSize, final int queueSize, final long timeInterval,
      final int writeBufferSize, final File[] diskDirs, final int[] diskDirSizes,
      final float warningPercentage, final float criticalPercentage) {
    final DiskStore mockDiskStore = mock(DiskStore.class, name);
    when(mockDiskStore.getAllowForceCompaction()).thenReturn(allowForceCompaction);
    when(mockDiskStore.getAutoCompact()).thenReturn(autoCompact);
    when(mockDiskStore.getCompactionThreshold()).thenReturn(compactionThreshold);
    when(mockDiskStore.getDiskStoreUUID()).thenReturn(diskStoreId);
    when(mockDiskStore.getMaxOplogSize()).thenReturn(maxOplogSize);
    when(mockDiskStore.getName()).thenReturn(name);
    when(mockDiskStore.getQueueSize()).thenReturn(queueSize);
    when(mockDiskStore.getTimeInterval()).thenReturn(timeInterval);
    when(mockDiskStore.getWriteBufferSize()).thenReturn(writeBufferSize);
    when(mockDiskStore.getDiskDirs()).thenReturn(diskDirs);
    when(mockDiskStore.getDiskDirSizes()).thenReturn(diskDirSizes);
    when(mockDiskStore.getDiskUsageWarningPercentage()).thenReturn(warningPercentage);
    when(mockDiskStore.getDiskUsageCriticalPercentage()).thenReturn(criticalPercentage);

    return mockDiskStore;
  }

  private DiskStoreDetails.RegionDetails createRegionDetails(final String fullPath,
      final String name, final boolean persistent, final boolean overflow) {
    final DiskStoreDetails.RegionDetails regionDetails =
        new DiskStoreDetails.RegionDetails(fullPath, name);
    regionDetails.setPersistent(persistent);
    regionDetails.setOverflowToDisk(overflow);

    return regionDetails;
  }

  @SuppressWarnings("unchecked")
  private Set<DiskStoreDetails.RegionDetails> setupRegionsForTestExecute(
      final InternalCache mockCache, final String diskStoreName) {
    final Region<Object, Object> mockUserRegion = mock(Region.class, SEPARATOR + "UserRegion");
    final Region<Object, Object> mockGuestRegion = mock(Region.class, SEPARATOR + "GuestRegion");
    final Region<Object, Object> mockSessionRegion =
        mock(Region.class, SEPARATOR + "UserRegion" + SEPARATOR + "SessionRegion");
    final RegionAttributes<Object, Object> mockUserRegionAttributes =
        mock(RegionAttributes.class, "UserRegionAttributes");
    final RegionAttributes<Object, Object> mockSessionRegionAttributes =
        mock(RegionAttributes.class, "SessionRegionAttributes");
    final RegionAttributes<Object, Object> mockGuestRegionAttributes =
        mock(RegionAttributes.class, "GuestRegionAttributes");
    final EvictionAttributes mockUserEvictionAttributes =
        mock(EvictionAttributes.class, "UserEvictionAttributes");
    final EvictionAttributes mockSessionEvictionAttributes =
        mock(EvictionAttributes.class, "SessionEvictionAttributes");
    final EvictionAttributes mockGuestEvictionAttributes =
        mock(EvictionAttributes.class, "GuestEvictionAttributes");

    when(mockCache.rootRegions())
        .thenReturn(CollectionUtils.asSet(mockUserRegion, mockGuestRegion));
    when(mockUserRegion.getAttributes()).thenReturn(mockUserRegionAttributes);
    when(mockUserRegion.getFullPath()).thenReturn(SEPARATOR + "UserRegion");
    when(mockUserRegion.getName()).thenReturn("UserRegion");
    when(mockUserRegion.subregions(false)).thenReturn(CollectionUtils.asSet(mockSessionRegion));
    when(mockUserRegionAttributes.getDataPolicy()).thenReturn(DataPolicy.PERSISTENT_PARTITION);
    when(mockUserRegionAttributes.getDiskStoreName()).thenReturn(diskStoreName);
    when(mockUserRegionAttributes.getEvictionAttributes()).thenReturn(mockUserEvictionAttributes);
    when(mockUserEvictionAttributes.getAction()).thenReturn(EvictionAction.LOCAL_DESTROY);
    when(mockSessionRegion.getAttributes()).thenReturn(mockSessionRegionAttributes);
    when(mockSessionRegion.getFullPath())
        .thenReturn(SEPARATOR + "UserRegion" + SEPARATOR + "SessionRegion");
    when(mockSessionRegion.getName()).thenReturn("SessionRegion");
    when(mockSessionRegion.subregions(false)).thenReturn(Collections.emptySet());
    when(mockSessionRegionAttributes.getDataPolicy()).thenReturn(DataPolicy.REPLICATE);
    when(mockSessionRegionAttributes.getDiskStoreName()).thenReturn(diskStoreName);
    when(mockSessionRegionAttributes.getEvictionAttributes())
        .thenReturn(mockSessionEvictionAttributes);
    when(mockSessionEvictionAttributes.getAction()).thenReturn(EvictionAction.OVERFLOW_TO_DISK);
    when(mockGuestRegion.getAttributes()).thenReturn(mockGuestRegionAttributes);
    when(mockGuestRegion.subregions(false)).thenReturn(Collections.emptySet());
    when(mockGuestRegionAttributes.getDataPolicy()).thenReturn(DataPolicy.REPLICATE);
    when(mockGuestRegionAttributes.getDiskStoreName())
        .thenReturn(DiskStoreDetails.DEFAULT_DISK_STORE_NAME);
    when(mockGuestRegionAttributes.getEvictionAttributes()).thenReturn(mockGuestEvictionAttributes);
    when(mockGuestEvictionAttributes.getAction()).thenReturn(EvictionAction.OVERFLOW_TO_DISK);

    return CollectionUtils.asSet(
        createRegionDetails(SEPARATOR + "UserRegion", "UserRegion", true, false),
        createRegionDetails(SEPARATOR + "UserRegion" + SEPARATOR + "SessionRegion", "SessionRegion",
            false, true));
  }

  private Set<DiskStoreDetails.GatewayDetails> setupGatewaysForTestExecute(
      final InternalCache mockCache, final String diskStoreName) {
    final GatewaySender mockGatewaySender = mock(GatewaySender.class, "GatewaySender");
    when(mockCache.getGatewaySenders()).thenReturn(CollectionUtils.asSet(mockGatewaySender));
    when(mockGatewaySender.getDiskStoreName()).thenReturn(diskStoreName);
    when(mockGatewaySender.getId()).thenReturn("0123456789");
    when(mockGatewaySender.isPersistenceEnabled()).thenReturn(true);

    return CollectionUtils.asSet(createGatewayDetails("0123456789", true));
  }

  private Set<DiskStoreDetails.CacheServerDetails> setupCacheServersForTestExecute(
      final InternalCache mockCache, final String diskStoreName) {
    final CacheServer mockCacheServer1 = mock(CacheServer.class, "CacheServer1");
    final CacheServer mockCacheServer2 = mock(CacheServer.class, "CacheServer2");
    final CacheServer mockCacheServer3 = mock(CacheServer.class, "CacheServer3");
    final ClientSubscriptionConfig cacheServer1ClientSubscriptionConfig =
        mock(ClientSubscriptionConfig.class, "cacheServer1ClientSubscriptionConfig");
    final ClientSubscriptionConfig cacheServer3ClientSubscriptionConfig =
        mock(ClientSubscriptionConfig.class, "cacheServer3ClientSubscriptionConfig");

    when(mockCache.getCacheServers())
        .thenReturn(Arrays.asList(mockCacheServer1, mockCacheServer2, mockCacheServer3));
    when(mockCacheServer1.getClientSubscriptionConfig())
        .thenReturn(cacheServer1ClientSubscriptionConfig);
    when(cacheServer1ClientSubscriptionConfig.getDiskStoreName()).thenReturn(diskStoreName);
    when(mockCacheServer2.getClientSubscriptionConfig()).thenReturn(null);
    when(mockCacheServer3.getClientSubscriptionConfig())
        .thenReturn(cacheServer3ClientSubscriptionConfig);
    when(cacheServer3ClientSubscriptionConfig.getDiskStoreName()).thenReturn("");
    when(mockCacheServer1.getBindAddress()).thenReturn("10.127.0.1");
    when(mockCacheServer1.getPort()).thenReturn(10123);
    when(mockCacheServer1.getHostnameForClients()).thenReturn("rodan");

    return CollectionUtils.asSet(createCacheServerDetails("10.127.0.1", 10123, "rodan"));
  }

  private Set<DiskStoreDetails.AsyncEventQueueDetails> setupAsyncEventQueuesForTestExecute(
      final InternalCache mockCache, final String diskStoreName) {
    final AsyncEventQueue mockAsyncEventQueue1 = mock(AsyncEventQueue.class, "AsyncEventQueue1");
    final AsyncEventQueue mockAsyncEventQueue2 = mock(AsyncEventQueue.class, "AsyncEventQueue2");
    final AsyncEventQueue mockAsyncEventQueue3 = mock(AsyncEventQueue.class, "AsyncEventQueue3");

    when(mockCache.getAsyncEventQueues()).thenReturn(
        CollectionUtils.asSet(mockAsyncEventQueue1, mockAsyncEventQueue2, mockAsyncEventQueue3));
    when(mockAsyncEventQueue1.isPersistent()).thenReturn(true);
    when(mockAsyncEventQueue1.getDiskStoreName()).thenReturn(diskStoreName);
    when(mockAsyncEventQueue1.getId()).thenReturn("9876543210");
    when(mockAsyncEventQueue2.isPersistent()).thenReturn(false);
    when(mockAsyncEventQueue3.isPersistent()).thenReturn(true);
    when(mockAsyncEventQueue3.getDiskStoreName()).thenReturn("memSto");

    return CollectionUtils.asSet(createAsyncEventQueueDetails("9876543210"));
  }

  @Test
  public void testAssertState() {
    DescribeDiskStoreFunction.assertState(true, "null");
  }

  @Test
  public void testAssertStateThrowsIllegalStateException() {
    assertThatThrownBy(
        () -> DescribeDiskStoreFunction.assertState(false, "Expected (%1$s) message!", "test"))
            .isInstanceOf(IllegalStateException.class).hasMessage("Expected (test) message!");
  }

  @Test
  public void testExecute() throws Throwable {
    // Prepare Mocks
    final UUID diskStoreId = UUID.randomUUID();
    final String diskStoreName = "mockDiskStore";
    final String memberId = "mockMemberId";
    final String memberName = "mockMemberName";
    final InternalDistributedMember mockMember =
        mock(InternalDistributedMember.class, "DistributedMember");
    @SuppressWarnings("unchecked")
    final FunctionContext<String> mockFunctionContext =
        mock(FunctionContext.class, "testExecute$FunctionContext");
    final DiskStore mockDiskStore =
        createMockDiskStore(diskStoreId, diskStoreName, true, false,
            75, 8192L, 500, 120L, 10240, createFileArray("/export/disk/backup",
                "/export/disk/overflow", "/export/disk/persistence"),
            createIntArray(10240, 204800, 4096000), 50, 75);
    final TestResultSender testResultSender = new TestResultSender();
    when(mockCache.getMyId()).thenReturn(mockMember);
    when(mockCache.findDiskStore(diskStoreName)).thenReturn(mockDiskStore);
    when(mockCache.getPdxPersistent()).thenReturn(true);
    when(mockCache.getPdxDiskStore()).thenReturn("memoryStore");
    when(mockMember.getId()).thenReturn(memberId);
    when(mockMember.getName()).thenReturn(memberName);
    when(mockFunctionContext.getCache()).thenReturn(mockCache);
    when(mockFunctionContext.getArguments()).thenReturn(diskStoreName);
    when(mockFunctionContext.getResultSender()).thenReturn(testResultSender);

    // Expected Results
    final Set<DiskStoreDetails.RegionDetails> expectedRegionDetails =
        setupRegionsForTestExecute(mockCache, diskStoreName);
    final Set<DiskStoreDetails.GatewayDetails> expectedGatewayDetails =
        setupGatewaysForTestExecute(mockCache, diskStoreName);
    final Set<DiskStoreDetails.CacheServerDetails> expectedCacheServerDetails =
        setupCacheServersForTestExecute(mockCache, diskStoreName);
    final Set<DiskStoreDetails.AsyncEventQueueDetails> expectedAsyncEventQueueDetails =
        setupAsyncEventQueuesForTestExecute(mockCache, diskStoreName);

    // Execute Function and assert results
    final DescribeDiskStoreFunction function = new DescribeDiskStoreFunction();
    function.execute(mockFunctionContext);

    final List<?> results = testResultSender.getResults();
    assertThat(results).isNotNull();
    assertThat(results.size()).isEqualTo(1);

    final DiskStoreDetails diskStoreDetails = (DiskStoreDetails) results.get(0);
    AssertionsForClassTypes.assertThat(diskStoreDetails).isNotNull();
    assertThat(diskStoreDetails.getId()).isEqualTo(diskStoreId);
    assertThat(diskStoreDetails.getName()).isEqualTo(diskStoreName);
    assertThat(diskStoreDetails.getMemberId()).isEqualTo(memberId);
    assertThat(diskStoreDetails.getMemberName()).isEqualTo(memberName);
    assertThat(diskStoreDetails.getAllowForceCompaction()).isTrue();
    assertThat(diskStoreDetails.getAutoCompact()).isFalse();
    assertThat(diskStoreDetails.getCompactionThreshold().intValue()).isEqualTo(75);
    assertThat(diskStoreDetails.getMaxOplogSize().longValue()).isEqualTo(8192L);
    assertThat(diskStoreDetails.isPdxSerializationMetaDataStored()).isFalse();
    assertThat(diskStoreDetails.getQueueSize().intValue()).isEqualTo(500);
    assertThat(diskStoreDetails.getTimeInterval().longValue()).isEqualTo(120L);
    assertThat(diskStoreDetails.getWriteBufferSize().intValue()).isEqualTo(10240);
    assertThat(diskStoreDetails.getDiskUsageWarningPercentage()).isEqualTo(50.0f);
    assertThat(diskStoreDetails.getDiskUsageCriticalPercentage()).isEqualTo(75.0f);

    final List<Integer> expectedDiskDirSizes = Arrays.asList(10240, 204800, 4096000);
    final List<String> expectedDiskDirs =
        Arrays.asList(new File("/export/disk/backup").getAbsolutePath(),
            new File("/export/disk/overflow").getAbsolutePath(),
            new File("/export/disk/persistence").getAbsolutePath());
    int count = 0;

    for (final DiskStoreDetails.DiskDirDetails diskDirDetails : diskStoreDetails) {
      assertThat(expectedDiskDirSizes.contains(diskDirDetails.getSize())).isTrue();
      assertThat(expectedDiskDirs.contains(diskDirDetails.getAbsolutePath())).isTrue();
      count++;
    }

    verify(mockDiskStore, atLeastOnce()).getName();
    verify(mockDiskStore, atLeastOnce()).getDiskStoreUUID();
    assertThat(count).isEqualTo(expectedDiskDirs.size());
    assertRegionDetails(expectedRegionDetails, diskStoreDetails);
    assertCacheServerDetails(expectedCacheServerDetails, diskStoreDetails);
    assertGatewayDetails(expectedGatewayDetails, diskStoreDetails);
    assertAsyncEventQueueDetails(expectedAsyncEventQueueDetails, diskStoreDetails);
  }

  @Test
  public void testExecuteOnMemberHavingANonGemFireCache() throws Throwable {
    final Cache mockNonGemCache = mock(Cache.class, "NonGemCache");
    @SuppressWarnings("unchecked")
    final FunctionContext<String> mockFunctionContext =
        mock(FunctionContext.class, "FunctionContext");
    final TestResultSender testResultSender = new TestResultSender();
    when(mockFunctionContext.getCache()).thenReturn(mockNonGemCache);
    when(mockFunctionContext.getResultSender()).thenReturn(testResultSender);

    final DescribeDiskStoreFunction function = new DescribeDiskStoreFunction();
    function.execute(mockFunctionContext);

    final List<?> results = testResultSender.getResults();
    assertThat(results).isNotNull();
    assertThat(results.isEmpty()).isTrue();
  }

  @Test
  public void testExecuteThrowingEntityNotFoundException() {
    final String memberId = "mockMemberId";
    final String memberName = "mockMemberName";
    final String diskStoreName = "testDiskStore";
    final InternalDistributedMember mockMember =
        mock(InternalDistributedMember.class, "DistributedMember");
    @SuppressWarnings("unchecked")
    final FunctionContext<String> mockFunctionContext =
        mock(FunctionContext.class, "FunctionContext");
    final TestResultSender testResultSender = new TestResultSender();
    when(mockCache.getMyId()).thenReturn(mockMember);
    when(mockCache.findDiskStore(diskStoreName)).thenReturn(null);
    when(mockMember.getId()).thenReturn(memberId);
    when(mockMember.getName()).thenReturn(memberName);
    when(mockFunctionContext.getCache()).thenReturn(mockCache);
    when(mockFunctionContext.getArguments()).thenReturn(diskStoreName);
    when(mockFunctionContext.getResultSender()).thenReturn(testResultSender);

    final DescribeDiskStoreFunction function = new DescribeDiskStoreFunction();
    function.execute(mockFunctionContext);
    String expected = String.format("A disk store with name '%1$s' was not found on member '%2$s'.",
        diskStoreName, memberName);
    assertThatThrownBy(testResultSender::getResults).isInstanceOf(EntityNotFoundException.class)
        .hasMessage(expected);
  }

  @Test
  public void testExecuteThrowingRuntimeException() {
    final String diskStoreName = "testDiskStore";
    final String memberId = "mockMemberId";
    final String memberName = "mockMemberName";
    @SuppressWarnings("unchecked")
    final FunctionContext<String> mockFunctionContext =
        mock(FunctionContext.class, "FunctionContext");
    final InternalDistributedMember mockMember =
        mock(InternalDistributedMember.class, "DistributedMember");
    final TestResultSender testResultSender = new TestResultSender();
    when(mockCache.getMyId()).thenReturn(mockMember);
    when(mockCache.findDiskStore(diskStoreName)).thenThrow(new RuntimeException("ExpectedStrings"));
    when(mockMember.getId()).thenReturn(memberId);
    when(mockMember.getName()).thenReturn(memberName);
    when(mockFunctionContext.getCache()).thenReturn(mockCache);
    when(mockFunctionContext.getArguments()).thenReturn(diskStoreName);
    when(mockFunctionContext.getResultSender()).thenReturn(testResultSender);

    final DescribeDiskStoreFunction function = new DescribeDiskStoreFunction();
    function.execute(mockFunctionContext);
    assertThatThrownBy(testResultSender::getResults).isInstanceOf(RuntimeException.class)
        .hasMessage("ExpectedStrings");
  }

  @Test
  public void testExecuteWithDiskDirsAndDiskSizesMismatch() {
    final String diskStoreName = "mockDiskStore";
    final String memberId = "mockMemberId";
    final String memberName = "mockMemberName";
    final UUID diskStoreId = UUID.randomUUID();
    @SuppressWarnings("unchecked")
    final FunctionContext<String> mockFunctionContext =
        mock(FunctionContext.class, "FunctionContext");
    final InternalDistributedMember mockMember =
        mock(InternalDistributedMember.class, "DistributedMember");
    final DiskStore mockDiskStore =
        createMockDiskStore(diskStoreId, diskStoreName, false, true, 70, 8192000L, 1000, 300L, 8192,
            createFileArray("/export/disk0/gemfire/backup"), new int[0], 50, 75);
    final TestResultSender testResultSender = new TestResultSender();
    when(mockCache.getMyId()).thenReturn(mockMember);
    when(mockCache.findDiskStore(diskStoreName)).thenReturn(mockDiskStore);
    when(mockMember.getId()).thenReturn(memberId);
    when(mockMember.getName()).thenReturn(memberName);
    when(mockFunctionContext.getCache()).thenReturn(mockCache);
    when(mockFunctionContext.getArguments()).thenReturn(diskStoreName);
    when(mockFunctionContext.getResultSender()).thenReturn(testResultSender);

    final DescribeDiskStoreFunction function = new DescribeDiskStoreFunction();
    function.execute(mockFunctionContext);
    String expected =
        "The number of disk directories with a specified size (0) does not match the number of disk directories (1)!";
    assertThatThrownBy(testResultSender::getResults).hasMessage(expected);
    verify(mockDiskStore, atLeastOnce()).getName();
    verify(mockDiskStore, atLeastOnce()).getDiskStoreUUID();
  }

  @Test
  public void testGetRegionDiskStoreName() {
    final String expectedDiskStoreName = "testDiskStore";
    @SuppressWarnings("unchecked")
    final Region<Object, Object> mockRegion = mock(Region.class, "Region");
    @SuppressWarnings("unchecked")
    final RegionAttributes<Object, Object> mockRegionAttributes =
        mock(RegionAttributes.class, "RegionAttributes");
    when(mockRegion.getAttributes()).thenReturn(mockRegionAttributes);
    when(mockRegionAttributes.getDiskStoreName()).thenReturn(expectedDiskStoreName);

    final DescribeDiskStoreFunction function = new DescribeDiskStoreFunction();
    assertThat(function.getDiskStoreName(mockRegion)).isEqualTo(expectedDiskStoreName);
  }

  @Test
  public void testGetRegionDiskStoreNameWhenUnspecified() {
    @SuppressWarnings("unchecked")
    final Region<Object, Object> mockRegion = mock(Region.class, "Region");
    @SuppressWarnings("unchecked")
    final RegionAttributes<Object, Object> mockRegionAttributes =
        mock(RegionAttributes.class, "RegionAttributes");
    when(mockRegion.getAttributes()).thenReturn(mockRegionAttributes);
    when(mockRegionAttributes.getDiskStoreName()).thenReturn(null);

    final DescribeDiskStoreFunction function = new DescribeDiskStoreFunction();
    assertThat(function.getDiskStoreName(mockRegion))
        .isEqualTo(DiskStoreDetails.DEFAULT_DISK_STORE_NAME);
  }

  @Test
  public void testIsRegionOverflowToDiskWhenEvictionActionIsLocalDestroy() {
    @SuppressWarnings("unchecked")
    final Region<Object, Object> mockRegion = mock(Region.class, "Region");
    @SuppressWarnings("unchecked")
    final RegionAttributes<Object, Object> mockRegionAttributes =
        mock(RegionAttributes.class, "RegionAttributes");
    final EvictionAttributes mockEvictionAttributes =
        mock(EvictionAttributes.class, "EvictionAttributes");
    when(mockRegion.getAttributes()).thenReturn(mockRegionAttributes);
    when(mockRegionAttributes.getEvictionAttributes()).thenReturn(mockEvictionAttributes);
    when(mockEvictionAttributes.getAction()).thenReturn(EvictionAction.LOCAL_DESTROY);

    final DescribeDiskStoreFunction function = new DescribeDiskStoreFunction();
    assertThat(function.isOverflowToDisk(mockRegion)).isFalse();
    verify(mockRegion, times(2)).getAttributes();
    verify(mockRegionAttributes, times(2)).getEvictionAttributes();
  }

  @Test
  public void testIsRegionOverflowToDiskWhenEvictionActionIsOverflowToDisk() {
    @SuppressWarnings("unchecked")
    final Region<Object, Object> mockRegion = mock(Region.class, "Region");
    @SuppressWarnings("unchecked")
    final RegionAttributes<Object, Object> mockRegionAttributes =
        mock(RegionAttributes.class, "RegionAttributes");
    final EvictionAttributes mockEvictionAttributes =
        mock(EvictionAttributes.class, "EvictionAttributes");
    when(mockRegion.getAttributes()).thenReturn(mockRegionAttributes);
    when(mockRegionAttributes.getEvictionAttributes()).thenReturn(mockEvictionAttributes);
    when(mockEvictionAttributes.getAction()).thenReturn(EvictionAction.OVERFLOW_TO_DISK);

    final DescribeDiskStoreFunction function = new DescribeDiskStoreFunction();
    assertThat(function.isOverflowToDisk(mockRegion)).isTrue();
    verify(mockRegion, times(2)).getAttributes();
    verify(mockRegionAttributes, times(2)).getEvictionAttributes();
  }

  @Test
  public void testIsRegionOverflowToDiskWithNullEvictionAttributes() {
    @SuppressWarnings("unchecked")
    final Region<Object, Object> mockRegion = mock(Region.class, "Region");
    @SuppressWarnings("unchecked")
    final RegionAttributes<Object, Object> mockRegionAttributes =
        mock(RegionAttributes.class, "RegionAttributes");
    when(mockRegion.getAttributes()).thenReturn(mockRegionAttributes);
    when(mockRegionAttributes.getEvictionAttributes()).thenReturn(null);

    final DescribeDiskStoreFunction function = new DescribeDiskStoreFunction();
    assertThat(function.isOverflowToDisk(mockRegion)).isFalse();
  }

  @Test
  public void testIsRegionPersistentWhenDataPolicyIsPersistentPartition() {
    @SuppressWarnings("unchecked")
    final Region<Object, Object> mockRegion = mock(Region.class, "Region");
    @SuppressWarnings("unchecked")
    final RegionAttributes<Object, Object> mockRegionAttributes =
        mock(RegionAttributes.class, "RegionAttributes");
    when(mockRegion.getAttributes()).thenReturn(mockRegionAttributes);
    when(mockRegionAttributes.getDataPolicy()).thenReturn(DataPolicy.PERSISTENT_PARTITION);

    final DescribeDiskStoreFunction function = new DescribeDiskStoreFunction();
    assertThat(function.isPersistent(mockRegion)).isTrue();
  }

  @Test
  public void testIsRegionPersistentWhenDataPolicyIsPersistentReplicate() {
    @SuppressWarnings("unchecked")
    final Region<Object, Object> mockRegion = mock(Region.class, "Region");
    @SuppressWarnings("unchecked")
    final RegionAttributes<Object, Object> mockRegionAttributes =
        mock(RegionAttributes.class, "RegionAttributes");
    when(mockRegion.getAttributes()).thenReturn(mockRegionAttributes);
    when(mockRegionAttributes.getDataPolicy()).thenReturn(DataPolicy.PERSISTENT_REPLICATE);

    final DescribeDiskStoreFunction function = new DescribeDiskStoreFunction();
    assertThat(function.isPersistent(mockRegion)).isTrue();
  }

  @Test
  public void testIsRegionPersistentWhenDataPolicyIsNormal() {
    @SuppressWarnings("unchecked")
    final Region<Object, Object> mockRegion = mock(Region.class, "Region");
