/*
 * 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.cache;

import static org.apache.geode.cache.RegionShortcut.LOCAL;
import static org.apache.geode.cache.RegionShortcut.LOCAL_HEAP_LRU;
import static org.apache.geode.cache.RegionShortcut.LOCAL_OVERFLOW;
import static org.apache.geode.cache.RegionShortcut.LOCAL_PERSISTENT;
import static org.apache.geode.cache.RegionShortcut.LOCAL_PERSISTENT_OVERFLOW;
import static org.apache.geode.cache.RegionShortcut.PARTITION;
import static org.apache.geode.cache.RegionShortcut.PARTITION_HEAP_LRU;
import static org.apache.geode.cache.RegionShortcut.PARTITION_OVERFLOW;
import static org.apache.geode.cache.RegionShortcut.PARTITION_PERSISTENT;
import static org.apache.geode.cache.RegionShortcut.PARTITION_PERSISTENT_OVERFLOW;
import static org.apache.geode.cache.RegionShortcut.PARTITION_PROXY;
import static org.apache.geode.cache.RegionShortcut.PARTITION_PROXY_REDUNDANT;
import static org.apache.geode.cache.RegionShortcut.PARTITION_REDUNDANT;
import static org.apache.geode.cache.RegionShortcut.PARTITION_REDUNDANT_HEAP_LRU;
import static org.apache.geode.cache.RegionShortcut.PARTITION_REDUNDANT_OVERFLOW;
import static org.apache.geode.cache.RegionShortcut.PARTITION_REDUNDANT_PERSISTENT;
import static org.apache.geode.cache.RegionShortcut.PARTITION_REDUNDANT_PERSISTENT_OVERFLOW;
import static org.apache.geode.cache.RegionShortcut.REPLICATE;
import static org.apache.geode.cache.RegionShortcut.REPLICATE_HEAP_LRU;
import static org.apache.geode.cache.RegionShortcut.REPLICATE_OVERFLOW;
import static org.apache.geode.cache.RegionShortcut.REPLICATE_PERSISTENT;
import static org.apache.geode.cache.RegionShortcut.REPLICATE_PERSISTENT_OVERFLOW;
import static org.apache.geode.cache.RegionShortcut.REPLICATE_PROXY;
import static org.apache.geode.distributed.ConfigurationProperties.CACHE_XML_FILE;
import static org.apache.geode.distributed.ConfigurationProperties.LOCATORS;
import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT;
import static org.apache.geode.distributed.ConfigurationProperties.MCAST_TTL;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

import java.io.File;
import java.io.FileWriter;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Properties;

import org.junit.After;
import org.junit.Ignore;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TestName;

import org.apache.geode.CancelException;
import org.apache.geode.cache.client.PoolManager;
import org.apache.geode.cache.util.CacheListenerAdapter;
import org.apache.geode.cache.util.CacheWriterAdapter;
import org.apache.geode.distributed.DistributedSystem;
import org.apache.geode.distributed.internal.DistributionConfig;
import org.apache.geode.internal.cache.LocalRegion;

/**
 * Unit test for the RegionFactory class
 *
 * @since GemFire 5.0
 */
public class RegionFactoryJUnitTest {

  @Rule
  public TestName testName = new TestName();

  private final String key = "key";
  private final Integer val = 1;
  private final String r1Name = "r1";
  private final String r2Name = "r2";
  private final String r3Name = "r3";
  private final String r1sr1Name = "sr1";
  private Cache cache = null;

  private DistributedSystem distSys = null;

  private Region r1;
  private Region r2;
  private Region r3;
  private Region r1sr1;

  @After
  public void tearDown() throws Exception {
    try {
      if (r1 != null) {
        cleanUpRegion(r1);
      }
      if (r2 != null) {
        cleanUpRegion(r2);
      }
      if (r3 != null) {
        cleanUpRegion(r3);
      }
      if (r1sr1 != null) {
        cleanUpRegion(r1sr1);
      }
      Cache c = cache;
      DistributedSystem d = distSys;
      if (c != null && !c.isClosed()) {
        d = c.getDistributedSystem();
        c.close();
      }
      if (d != null && d.isConnected()) {
        d.disconnect();
      }
    } finally {
      r1 = null;
      r2 = null;
      r3 = null;
      r1sr1 = null;
      cache = null;
      distSys = null;
    }
  }

  /**
   * Tests create, destroy and recreate Region
   */
  @Test
  public void testCreateDestroyCreateRegions() throws Exception {
    // Assert basic region creation when no DistributedSystem or Cache exists
    RegionFactory factory = new RegionFactory(createGemFireProperties());
    r1 = factory.create(r1Name);
    assertBasicRegionFunctionality(r1, r1Name);

    // Assert duplicate creation failure
    try {
      factory.create(r1Name);
      fail("Expected RegionExistsException");
    } catch (RegionExistsException ignored) {
    }

    r1sr1 = factory.createSubregion(r1, r1sr1Name);
    assertBasicRegionFunctionality(r1sr1, r1sr1Name);
    try {
      factory.createSubregion(r1, r1sr1Name);
      fail("Expected RegionExistsException");
    } catch (RegionExistsException ignored) {
    }
    r1sr1.destroyRegion();

    r2 = factory.create(r2Name);
    assertBasicRegionFunctionality(r2, r2Name);
    r2.destroyRegion();

    try {
      factory.createSubregion(r2, "shouldNotBePossible");
      fail("Expected a RegionDestroyedException");
    } catch (RegionDestroyedException ignored) {
    }

    r1.destroyRegion();
    r1 = factory.create(r1Name);
    assertBasicRegionFunctionality(r1, r1Name);

    r1.destroyRegion();
    r1 = factory.create(r1Name);
    assertBasicRegionFunctionality(r1, r1Name);

    r1.destroyRegion();
    r1 = factory.create(r1Name);
    assertBasicRegionFunctionality(r1, r1Name);
  }

  /**
   * Test RegionFactory with Cache close and DistributedSystem disconnect
   */
  @Test
  public void testRegionFactoryAndCacheClose() throws Exception {
    // Assert basic region creation when no DistributedSystem or Cache exists
    RegionFactory factory = new RegionFactory(createGemFireProperties());
    r1 = factory.create(r1Name);
    assertBasicRegionFunctionality(r1, r1Name);

    // Assert duplicate creation failure
    try {
      factory.create(r1Name);
      fail("Expected RegionExistsException");
    } catch (RegionExistsException ignored) {
    }

    r2 = factory.create(r2Name);
    assertBasicRegionFunctionality(r2, r2Name);
    r2.destroyRegion();

    // as of 6.5 if the cache that was used to create a regionFactory is closed
    // then the factory is out of business
    Cache c = r1.getCache();
    r1.destroyRegion();
    c.close();
    try {
      r1 = factory.create(r1Name);
      fail("Use of RegionFactory after cache close should throw CancelException");
    } catch (CancelException expected) {
      // passed
    }

    factory = new RegionFactory(createGemFireProperties());
    r1 = factory.create(r1Name);
    assertBasicRegionFunctionality(r1, r1Name);

    // as of 6.5 if the ds that was used to create a regionFactory is disconnected
    // then the factory is out of business
    // Assert we can handle a disconnected disributed system
    DistributedSystem d = r1.getCache().getDistributedSystem();
    r1.destroyRegion();
    d.disconnect();
    try {
      r1 = factory.create(r1Name);
    } catch (CancelException expected) {
      // passed
    }

    factory = new RegionFactory(createGemFireProperties());
    r1 = factory.create(r1Name);
    assertBasicRegionFunctionality(r1, r1Name);

    // as of 6.5 if the ds that was used to create a regionFactory is disconnected
    // then the factory is out of business
    // Assert we can handle both a closed Cache and a disconnected system
    c = r1.getCache();
    d = c.getDistributedSystem();
    r1.destroyRegion();
    c.close();
    d.disconnect();
    r1 = new RegionFactory(createGemFireProperties()).create(r1Name);
    assertBasicRegionFunctionality(r1, r1Name);
  }

  @Test
  public void testAfterConnect() throws Exception {
    // Assert basic region creation when a Distributed system exists
    distSys = DistributedSystem.connect(createGemFireProperties()); // for teardown

    RegionFactory factory = new RegionFactory();
    r1 = factory.create(r1Name);
    cache = r1.getCache(); // for teardown
    assertBasicRegionFunctionality(r1, r1Name);
  }

  @Test
  public void testAfterConnectWithDifferentProperties() throws Exception {
    // Assert failure when a Distributed system exists but with different properties
    distSys = DistributedSystem.connect(createGemFireProperties()); // for teardown
    final Properties failed = new Properties();
    failed.put(MCAST_TTL, "64");

    try {
      new RegionFactory(failed);
      fail("Expected exception");
    } catch (IllegalStateException expected) {
      // passed
    }
  }

  @Test
  public void testAfterCacheCreate() throws Exception {
    // Assert basic region creation when a Distributed and Cache exist
    DistributedSystem ds = DistributedSystem.connect(createGemFireProperties());
    CacheFactory.create(ds);

    RegionFactory factory = new RegionFactory();
    r1 = factory.create(r1Name);
    assertBasicRegionFunctionality(r1, r1Name);
  }

  @Test
  public void testAfterCacheClosed() throws Exception {
    // Assert basic region creation when a Distributed and Cache exist but the cache is closed
    Properties gemfireProps = createGemFireProperties();
    DistributedSystem ds = DistributedSystem.connect(gemfireProps);
    cache = CacheFactory.create(ds);
    cache.close();

    RegionFactory factory = new RegionFactory(gemfireProps);
    r1 = factory.create(r1Name);
    assertBasicRegionFunctionality(r1, r1Name);
  }

  /**
   * Test method for 'org.apache.geode.cache.RegionFactory.RegionFactory(RegionAttributes)'
   */
  @Test
  public void testRegionFactoryRegionAttributes() throws Exception {
    Properties gemfireProps = createGemFireProperties();
    r1 = new RegionFactory(gemfireProps).setScope(Scope.LOCAL).setConcurrencyLevel(1)
        .setLoadFactor(0.8F).setKeyConstraint(String.class).setStatisticsEnabled(true)
        .create(r1Name);
    assertBasicRegionFunctionality(r1, r1Name);

    final RegionFactory factory = new RegionFactory(gemfireProps, r1.getAttributes());
    r2 = factory.create(r2Name);
    assertBasicRegionFunctionality(r2, r2Name);
    assertRegionAttributes(r1.getAttributes(), r2.getAttributes());

    r3 = factory.create(r3Name);
    assertRegionAttributes(r2.getAttributes(), r3.getAttributes());
  }

  /**
   * Test method for 'org.apache.geode.cache.RegionFactory.RegionFactory(String)'
   */
  @Test
  public void testRegionFactoryString() throws Exception {
    // TODO: create subdir using getName() and create cache.xml there instead of using
    // DEFAULT_CACHE_XML_FILE
    DistributionConfig.DEFAULT_CACHE_XML_FILE.delete();
    try {
      DistributionConfig.DEFAULT_CACHE_XML_FILE.createNewFile();
      FileWriter f = new FileWriter(DistributionConfig.DEFAULT_CACHE_XML_FILE);
      f.write("<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n\n"
          + "<!DOCTYPE cache PUBLIC\n  \"-//GemStone Systems, Inc.//GemFire Declarative Caching 7.0//EN\"\n"
          + "  \"http://www.gemstone.com/dtd/cache7_0.dtd\">\n" + "<cache>\n"
          + " <region-attributes id=\"" + getName()
          + "\" statistics-enabled=\"true\" scope=\"distributed-ack\">\n" + "  <key-constraint>"
          + String.class.getName() + "</key-constraint>\n" + "  <value-constraint>"
          + Integer.class.getName() + "</value-constraint>\n"
          + "    <entry-idle-time><expiration-attributes timeout=\"60\"/></entry-idle-time>\n"
          + " </region-attributes>\n" + "</cache>");
      f.close();

      RegionFactory factory = new RegionFactory(createGemFireProperties(), getName());
      r1 = factory.create(r1Name);
      assertBasicRegionFunctionality(r1, r1Name);
      RegionAttributes ra = r1.getAttributes();
      assertEquals(ra.getStatisticsEnabled(), true);
      assertEquals(ra.getScope().isDistributedAck(), true);
      assertEquals(ra.getValueConstraint(), Integer.class);
      assertEquals(ra.getKeyConstraint(), String.class);
      assertEquals(ra.getEntryIdleTimeout().getTimeout(), 60);
    } finally {
      DistributionConfig.DEFAULT_CACHE_XML_FILE.delete();
    }
  }

  /**
   * Test method for 'org.apache.geode.cache.RegionFactory.RegionFactory(Properties)'
   */
  @Test
  public void testRegionFactoryProperties() throws Exception {
    final Properties gemfireProperties = createGemFireProperties();
    gemfireProperties.put(MCAST_TTL, "64");
    RegionFactory factory = new RegionFactory(gemfireProperties);
    r1 = factory.create(r1Name);
    assertBasicRegionFunctionality(r1, r1Name);
    assertEquals(gemfireProperties.get(MCAST_TTL),
        r1.getCache().getDistributedSystem().getProperties().get(MCAST_TTL));
  }

  /**
   * Test method for 'org.apache.geode.cache.RegionFactory.RegionFactory(Properties,
   * RegionAttributes)'
   */
  @Ignore
  @Test
  public void testRegionFactoryPropertiesRegionAttributes() {
    // TODO: implement test
  }

  /**
   * Test method for 'org.apache.geode.cache.RegionFactory.RegionFactory(Properties, String)'
   */
  @Test
  public void testRegionFactoryPropertiesString() throws Exception {
    File xmlFile = null;
    try {
      final Properties gemfireProperties = createGemFireProperties();
      gemfireProperties.put(MCAST_TTL, "64");
      final String xmlFileName = getName() + "-cache.xml";
      gemfireProperties.put(CACHE_XML_FILE, xmlFileName);
      xmlFile = new File(xmlFileName);
      xmlFile.delete();
      xmlFile.createNewFile();
      FileWriter f = new FileWriter(xmlFile);
      final String attrsId = getName() + "-attrsId";
      f.write("<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n\n"
          + "<!DOCTYPE cache PUBLIC\n  \"-//GemStone Systems, Inc.//GemFire Declarative Caching 7.0//EN\"\n"
          + "  \"http://www.gemstone.com/dtd/cache7_0.dtd\">\n" + "<cache>\n"
          + " <region-attributes id=\"" + attrsId
          + "\" statistics-enabled=\"true\" scope=\"distributed-ack\">\n" + "  <key-constraint>"
          + String.class.getName() + "</key-constraint>\n" + "  <value-constraint>"
          + Integer.class.getName() + "</value-constraint>\n"
          + "    <entry-idle-time><expiration-attributes timeout=\"60\"/></entry-idle-time>\n"
          + " </region-attributes>\n" + "</cache>");
      f.close();

      RegionFactory factory = new RegionFactory(gemfireProperties, attrsId);
      r1 = factory.create(r1Name);
      assertBasicRegionFunctionality(r1, r1Name);
      assertEquals(gemfireProperties.get(MCAST_TTL),
          r1.getCache().getDistributedSystem().getProperties().get(MCAST_TTL));
      assertEquals(gemfireProperties.get(CACHE_XML_FILE),
          r1.getCache().getDistributedSystem().getProperties().get(CACHE_XML_FILE));
      RegionAttributes ra = r1.getAttributes();
      assertEquals(ra.getStatisticsEnabled(), true);
      assertEquals(ra.getScope().isDistributedAck(), true);
      assertEquals(ra.getValueConstraint(), Integer.class);
      assertEquals(ra.getKeyConstraint(), String.class);
      assertEquals(ra.getEntryIdleTimeout().getTimeout(), 60);
    } finally {
      if (xmlFile != null) {
        xmlFile.delete();
      }
    }
  }


  /**
   * Ensure that the RegionFactory set methods mirrors those found in RegionAttributes
   *
   */
  @Test
  public void testAttributesFactoryConformance() throws Exception {
    Method[] af = AttributesFactory.class.getDeclaredMethods();
    Method[] rf = RegionFactory.class.getDeclaredMethods();
    Method am, rm;

    ArrayList afDeprected = new ArrayList(); // hack to ignore deprecated methods
    afDeprected.add("setCacheListener");
    afDeprected.add("setMirrorType");
    afDeprected.add("setPersistBackup");
    afDeprected.add("setBucketRegion");
    afDeprected.add("setEnableWAN");
    afDeprected.add("setEnableBridgeConflation");
    afDeprected.add("setEnableConflation");
    ArrayList methodsToImplement = new ArrayList();

    // Since the RegionFactory has an AttributesFactory member,
    // we only need to make sure the RegionFactory class adds proxies for the
    // 'set' and 'add' methods added to the AttributesFactory. The java complier
    // will notify the
    // developer if a method is removed from AttributesFactory.
    String amName;
    boolean hasMethod = false;
    assertTrue(af.length != 0);
    for (final Method value : af) {
      am = value;
      amName = am.getName();
      if (!afDeprected.contains(amName) && (amName.startsWith("set") || amName.startsWith("add"))) {
        for (final Method method : rf) {
          rm = method;
          if (rm.getName().equals(am.getName())) {
            Class[] rparams = rm.getParameterTypes();
            Class[] aparams = am.getParameterTypes();
            if (rparams.length == aparams.length) {
              boolean hasAllParams = true;
              for (int k = 0; k < rparams.length; k++) {
                if (aparams[k] != rparams[k]) {
                  hasAllParams = false;
                  break;
                }
              } // parameter check
              if (hasAllParams) {
                hasMethod = true;
              }
            }
          }
        } // region factory methods
        if (!hasMethod) {
          methodsToImplement.add(am);
        } else {
          hasMethod = false;
        }
      }
    } // attributes methods

    if (methodsToImplement.size() > 0) {
      fail("RegionFactory does not conform to AttributesFactory, its should proxy these methods "
          + methodsToImplement);
    }
  }

  @Test
  public void testPARTITION() throws Exception {
    Cache c = createCache();
    RegionFactory factory = c.createRegionFactory(PARTITION);
    r1 = factory.create(r1Name);
    RegionAttributes ra = r1.getAttributes();
    assertEquals(DataPolicy.PARTITION, ra.getDataPolicy());
    assertNotNull(ra.getPartitionAttributes());
    assertEquals(0, ra.getPartitionAttributes().getRedundantCopies());
  }

  @Test
  public void testPARTITION_REDUNDANT() throws Exception {
    Cache c = createCache();
    RegionFactory factory = c.createRegionFactory(PARTITION_REDUNDANT);
    r1 = factory.create(r1Name);
    RegionAttributes ra = r1.getAttributes();
    assertEquals(DataPolicy.PARTITION, ra.getDataPolicy());
    assertNotNull(ra.getPartitionAttributes());
    assertEquals(1, ra.getPartitionAttributes().getRedundantCopies());
  }

  @Test
  public void testPARTITION_PERSISTENT() throws Exception {
    Cache c = createCache();
    RegionFactory factory = c.createRegionFactory(PARTITION_PERSISTENT);
    r1 = factory.create(r1Name);
    RegionAttributes ra = r1.getAttributes();
    assertEquals(DataPolicy.PERSISTENT_PARTITION, ra.getDataPolicy());
    assertNotNull(ra.getPartitionAttributes());
    assertEquals(0, ra.getPartitionAttributes().getRedundantCopies());
  }

  @Test
  public void testPARTITION_REDUNDANT_PERSISTENT() throws Exception {
    Cache c = createCache();
    RegionFactory factory = c.createRegionFactory(PARTITION_REDUNDANT_PERSISTENT);
    r1 = factory.create(r1Name);
    RegionAttributes ra = r1.getAttributes();
    assertEquals(DataPolicy.PERSISTENT_PARTITION, ra.getDataPolicy());
    assertNotNull(ra.getPartitionAttributes());
    assertEquals(1, ra.getPartitionAttributes().getRedundantCopies());
  }

  @Test
  public void testPARTITION_OVERFLOW() throws Exception {
    Cache c = createCache();
    RegionFactory factory = c.createRegionFactory(PARTITION_OVERFLOW);
    r1 = factory.create(r1Name);
    RegionAttributes ra = r1.getAttributes();
    assertEquals(DataPolicy.PARTITION, ra.getDataPolicy());
    assertNotNull(ra.getPartitionAttributes());
    assertEquals(0, ra.getPartitionAttributes().getRedundantCopies());
    assertEquals(EvictionAttributes.createLRUHeapAttributes(null, EvictionAction.OVERFLOW_TO_DISK),
        ra.getEvictionAttributes());
    assertEquals(LocalRegion.DEFAULT_HEAPLRU_EVICTION_HEAP_PERCENTAGE,
        c.getResourceManager().getEvictionHeapPercentage(), 0);
  }

  @Test
  public void testPARTITION_REDUNDANT_OVERFLOW() throws Exception {
    Cache c = createCache();
    RegionFactory factory = c.createRegionFactory(PARTITION_REDUNDANT_OVERFLOW);
    r1 = factory.create(r1Name);
    RegionAttributes ra = r1.getAttributes();
    assertEquals(DataPolicy.PARTITION, ra.getDataPolicy());
    assertNotNull(ra.getPartitionAttributes());
    assertEquals(1, ra.getPartitionAttributes().getRedundantCopies());
    assertEquals(EvictionAttributes.createLRUHeapAttributes(null, EvictionAction.OVERFLOW_TO_DISK),
        ra.getEvictionAttributes());
    assertEquals(LocalRegion.DEFAULT_HEAPLRU_EVICTION_HEAP_PERCENTAGE,
        c.getResourceManager().getEvictionHeapPercentage(), 0);
  }

  @Test
  public void testPARTITION_PERSISTENT_OVERFLOW() throws Exception {
    Cache c = createCache();
    RegionFactory factory = c.createRegionFactory(PARTITION_PERSISTENT_OVERFLOW);
    r1 = factory.create(r1Name);
    RegionAttributes ra = r1.getAttributes();
    assertEquals(DataPolicy.PERSISTENT_PARTITION, ra.getDataPolicy());
    assertNotNull(ra.getPartitionAttributes());
    assertEquals(0, ra.getPartitionAttributes().getRedundantCopies());
    assertEquals(EvictionAttributes.createLRUHeapAttributes(null, EvictionAction.OVERFLOW_TO_DISK),
        ra.getEvictionAttributes());
    assertEquals(LocalRegion.DEFAULT_HEAPLRU_EVICTION_HEAP_PERCENTAGE,
        c.getResourceManager().getEvictionHeapPercentage(), 0);
  }

  @Test
  public void testPARTITION_REDUNDANT_PERSISTENT_OVERFLOW() throws Exception {
    Cache c = createCache();
    RegionFactory factory = c.createRegionFactory(PARTITION_REDUNDANT_PERSISTENT_OVERFLOW);
    r1 = factory.create(r1Name);
    RegionAttributes ra = r1.getAttributes();
    assertEquals(DataPolicy.PERSISTENT_PARTITION, ra.getDataPolicy());
    assertNotNull(ra.getPartitionAttributes());
    assertEquals(1, ra.getPartitionAttributes().getRedundantCopies());
    assertEquals(EvictionAttributes.createLRUHeapAttributes(null, EvictionAction.OVERFLOW_TO_DISK),
        ra.getEvictionAttributes());
    assertEquals(LocalRegion.DEFAULT_HEAPLRU_EVICTION_HEAP_PERCENTAGE,
        c.getResourceManager().getEvictionHeapPercentage(), 0);
  }

  @Test
  public void testPARTITION_HEAP_LRU() throws Exception {
    Cache c = createCache();
    RegionFactory factory = c.createRegionFactory(PARTITION_HEAP_LRU);
    r1 = factory.create(r1Name);
    RegionAttributes ra = r1.getAttributes();
    assertEquals(DataPolicy.PARTITION, ra.getDataPolicy());
    assertNotNull(ra.getPartitionAttributes());
    assertEquals(0, ra.getPartitionAttributes().getRedundantCopies());
    assertEquals(EvictionAttributes.createLRUHeapAttributes(), ra.getEvictionAttributes());
    assertEquals(LocalRegion.DEFAULT_HEAPLRU_EVICTION_HEAP_PERCENTAGE,
        c.getResourceManager().getEvictionHeapPercentage(), 0);
  }

  @Test
  public void testPARTITION_REDUNDANT_HEAP_LRU() throws Exception {
    Cache c = createCache();
    RegionFactory factory = c.createRegionFactory(PARTITION_REDUNDANT_HEAP_LRU);
    r1 = factory.create(r1Name);
    RegionAttributes ra = r1.getAttributes();
    assertEquals(DataPolicy.PARTITION, ra.getDataPolicy());
    assertNotNull(ra.getPartitionAttributes());
    assertEquals(1, ra.getPartitionAttributes().getRedundantCopies());
    assertEquals(EvictionAttributes.createLRUHeapAttributes(), ra.getEvictionAttributes());
    assertEquals(LocalRegion.DEFAULT_HEAPLRU_EVICTION_HEAP_PERCENTAGE,
        c.getResourceManager().getEvictionHeapPercentage(), 0);
  }

  @Test
  public void testREPLICATE() throws Exception {
    Cache c = createCache();
    RegionFactory factory = c.createRegionFactory(REPLICATE);
    r1 = factory.create(r1Name);
    RegionAttributes ra = r1.getAttributes();
    assertEquals(DataPolicy.REPLICATE, ra.getDataPolicy());
    assertEquals(Scope.DISTRIBUTED_ACK, ra.getScope());
  }

  @Test
  public void testREPLICATE_PERSISTENT() throws Exception {
    Cache c = createCache();
    RegionFactory factory = c.createRegionFactory(REPLICATE_PERSISTENT);
    r1 = factory.create(r1Name);
    RegionAttributes ra = r1.getAttributes();
    assertEquals(DataPolicy.PERSISTENT_REPLICATE, ra.getDataPolicy());
    assertEquals(Scope.DISTRIBUTED_ACK, ra.getScope());
  }

  @Test
  public void testREPLICATE_OVERFLOW() throws Exception {
    Cache c = createCache();
    RegionFactory factory = c.createRegionFactory(REPLICATE_OVERFLOW);
    r1 = factory.create(r1Name);
    RegionAttributes ra = r1.getAttributes();
    assertEquals(DataPolicy.REPLICATE, ra.getDataPolicy());
    assertEquals(Scope.DISTRIBUTED_ACK, ra.getScope());
    assertEquals(EvictionAttributes.createLRUHeapAttributes(null, EvictionAction.OVERFLOW_TO_DISK),
        ra.getEvictionAttributes());
    assertEquals(LocalRegion.DEFAULT_HEAPLRU_EVICTION_HEAP_PERCENTAGE,
        c.getResourceManager().getEvictionHeapPercentage(), 0);
  }

  @Test
  public void testREPLICATE_PERSISTENT_OVERFLOW() throws Exception {
    Cache c = createCache();
    RegionFactory factory = c.createRegionFactory(REPLICATE_PERSISTENT_OVERFLOW);
    r1 = factory.create(r1Name);
    RegionAttributes ra = r1.getAttributes();
    assertEquals(DataPolicy.PERSISTENT_REPLICATE, ra.getDataPolicy());
    assertEquals(Scope.DISTRIBUTED_ACK, ra.getScope());
    assertEquals(EvictionAttributes.createLRUHeapAttributes(null, EvictionAction.OVERFLOW_TO_DISK),
        ra.getEvictionAttributes());
    assertEquals(LocalRegion.DEFAULT_HEAPLRU_EVICTION_HEAP_PERCENTAGE,
        c.getResourceManager().getEvictionHeapPercentage(), 0);
  }

  @Test
  public void testREPLICATE_HEAP_LRU() throws Exception {
    Cache c = createCache();
    RegionFactory factory = c.createRegionFactory(REPLICATE_HEAP_LRU);
    r1 = factory.create(r1Name);
    RegionAttributes ra = r1.getAttributes();
    assertEquals(DataPolicy.PRELOADED, ra.getDataPolicy());
    assertEquals(new SubscriptionAttributes(InterestPolicy.ALL), ra.getSubscriptionAttributes());
    assertEquals(Scope.DISTRIBUTED_ACK, ra.getScope());
    assertEquals(EvictionAttributes.createLRUHeapAttributes(), ra.getEvictionAttributes());
    assertEquals(LocalRegion.DEFAULT_HEAPLRU_EVICTION_HEAP_PERCENTAGE,
        c.getResourceManager().getEvictionHeapPercentage(), 0);
  }

  @Test
  public void testLOCAL() throws Exception {
    Cache c = createCache();
    RegionFactory factory = c.createRegionFactory(LOCAL);
    r1 = factory.create(r1Name);
    RegionAttributes ra = r1.getAttributes();
    assertEquals(DataPolicy.NORMAL, ra.getDataPolicy());
    assertEquals(Scope.LOCAL, ra.getScope());
  }

  @Test
  public void testLOCAL_PERSISTENT() throws Exception {
    Cache c = createCache();
    RegionFactory factory = c.createRegionFactory(LOCAL_PERSISTENT);
    r1 = factory.create(r1Name);
    RegionAttributes ra = r1.getAttributes();
    assertEquals(DataPolicy.PERSISTENT_REPLICATE, ra.getDataPolicy());
    assertEquals(Scope.LOCAL, ra.getScope());
  }

  @Test
  public void testLOCAL_HEAP_LRU() throws Exception {
    Cache c = new CacheFactory(createGemFireProperties()).create();
    RegionFactory factory = c.createRegionFactory(LOCAL_HEAP_LRU);
    r1 = factory.create(r1Name);
    RegionAttributes ra = r1.getAttributes();
    assertEquals(DataPolicy.NORMAL, ra.getDataPolicy());
    assertEquals(Scope.LOCAL, ra.getScope());
    assertEquals(EvictionAttributes.createLRUHeapAttributes(), ra.getEvictionAttributes());
    assertEquals(LocalRegion.DEFAULT_HEAPLRU_EVICTION_HEAP_PERCENTAGE,
        c.getResourceManager().getEvictionHeapPercentage(), 0);
  }

  @Test
  public void testLOCAL_OVERFLOW() throws Exception {
    Cache c = new CacheFactory(createGemFireProperties()).create();
    RegionFactory factory = c.createRegionFactory(LOCAL_OVERFLOW);
    r1 = factory.create(r1Name);
    RegionAttributes ra = r1.getAttributes();
    assertEquals(DataPolicy.NORMAL, ra.getDataPolicy());
    assertEquals(Scope.LOCAL, ra.getScope());
    assertEquals(EvictionAttributes.createLRUHeapAttributes(null, EvictionAction.OVERFLOW_TO_DISK),
        ra.getEvictionAttributes());
    assertEquals(LocalRegion.DEFAULT_HEAPLRU_EVICTION_HEAP_PERCENTAGE,
        c.getResourceManager().getEvictionHeapPercentage(), 0);
  }

  @Test
  public void testLOCAL_PERSISTENT_OVERFLOW() throws Exception {
    Cache c = new CacheFactory(createGemFireProperties()).create();
    RegionFactory factory = c.createRegionFactory(LOCAL_PERSISTENT_OVERFLOW);
    r1 = factory.create(r1Name);
    RegionAttributes ra = r1.getAttributes();
    assertEquals(DataPolicy.PERSISTENT_REPLICATE, ra.getDataPolicy());
    assertEquals(Scope.LOCAL, ra.getScope());
    assertEquals(EvictionAttributes.createLRUHeapAttributes(null, EvictionAction.OVERFLOW_TO_DISK),
        ra.getEvictionAttributes());
    assertEquals(LocalRegion.DEFAULT_HEAPLRU_EVICTION_HEAP_PERCENTAGE,
        c.getResourceManager().getEvictionHeapPercentage(), 0);
  }

  @Test
  public void testPARTITION_PROXY() throws Exception {
    Cache c = createCache();
    RegionFactory factory = c.createRegionFactory(PARTITION_PROXY);
    r1 = factory.create(r1Name);
    RegionAttributes ra = r1.getAttributes();
    assertEquals(DataPolicy.PARTITION, ra.getDataPolicy());
    assertNotNull(ra.getPartitionAttributes());
    assertEquals(0, ra.getPartitionAttributes().getRedundantCopies());
    assertEquals(0, ra.getPartitionAttributes().getLocalMaxMemory());
  }

  @Test
  public void testPARTITION_PROXY_REDUNDANT() throws Exception {
    Cache c = createCache();
    RegionFactory factory = c.createRegionFactory(PARTITION_PROXY_REDUNDANT);
    r1 = factory.create(r1Name);
    RegionAttributes ra = r1.getAttributes();
    assertEquals(DataPolicy.PARTITION, ra.getDataPolicy());
    assertNotNull(ra.getPartitionAttributes());
    assertEquals(1, ra.getPartitionAttributes().getRedundantCopies());
    assertEquals(0, ra.getPartitionAttributes().getLocalMaxMemory());
  }

  @Test
  public void testREPLICATE_PROXY() throws Exception {
    Cache c = createCache();
    RegionFactory factory = c.createRegionFactory(REPLICATE_PROXY);
    r1 = factory.create(r1Name);
    RegionAttributes ra = r1.getAttributes();
    assertEquals(DataPolicy.EMPTY, ra.getDataPolicy());
    assertEquals(Scope.DISTRIBUTED_ACK, ra.getScope());
  }

  @Test
  public void testSetCacheLoader() throws Exception {
    Cache c = createCache();
    RegionFactory factory = c.createRegionFactory(REPLICATE_PROXY);
    CacheLoader cl = new MyCacheLoader();
    r1 = factory.setCacheLoader(cl).create(r1Name);
    RegionAttributes ra = r1.getAttributes();
    assertEquals(cl, ra.getCacheLoader());
  }

  @Test
  public void testSetCacheWriter() throws Exception {
    Cache c = createCache();
    RegionFactory factory = c.createRegionFactory(REPLICATE_PROXY);
    CacheWriter cw = new MyCacheWriter();
    r1 = factory.setCacheWriter(cw).create(r1Name);
    RegionAttributes ra = r1.getAttributes();
    assertEquals(cw, ra.getCacheWriter());
  }

  @Test
  public void testAddCacheListener() throws Exception {
    Cache c = createCache();
    RegionFactory factory = c.createRegionFactory(REPLICATE_PROXY);
    CacheListener cl = new MyCacheListener();
    r1 = factory.addCacheListener(cl).create(r1Name);
    RegionAttributes ra = r1.getAttributes();
    assertEquals(cl, ra.getCacheListener());
  }

  @Test
  public void testInitCacheListener() throws Exception {
    Cache c = createCache();
    RegionFactory factory = c.createRegionFactory(REPLICATE_PROXY);
    CacheListener cl1 = new MyCacheListener();
    CacheListener cl2 = new MyCacheListener();
    r1 = factory.initCacheListeners(new CacheListener[] {cl1, cl2}).create(r1Name);
    RegionAttributes ra = r1.getAttributes();
    assertEquals(true, Arrays.equals(new CacheListener[] {cl1, cl2}, ra.getCacheListeners()));
  }

  @Test
  public void testSetEvictionAttributes() throws Exception {
    Cache c = createCache();
    RegionFactory factory = c.createRegionFactory(LOCAL);
