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

import static org.apache.geode.cache.Region.SEPARATOR;
import static org.apache.geode.distributed.ConfigurationProperties.ROLES;
import static org.apache.geode.internal.AvailablePortHelper.getRandomAvailableTCPPort;
import static org.apache.geode.test.dunit.IgnoredException.addIgnoredException;
import static org.apache.geode.test.util.ResourceUtils.createTempFileFromResource;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNotSame;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStreamWriter;
import java.io.PrintWriter;
import java.io.Serializable;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import org.junit.Ignore;
import org.junit.Test;
import org.xml.sax.SAXException;

import org.apache.geode.DataSerializable;
import org.apache.geode.DataSerializer;
import org.apache.geode.cache.AttributesFactory;
import org.apache.geode.cache.AttributesMutator;
import org.apache.geode.cache.Cache;
import org.apache.geode.cache.CacheException;
import org.apache.geode.cache.CacheFactory;
import org.apache.geode.cache.CacheListener;
import org.apache.geode.cache.CacheLoader;
import org.apache.geode.cache.CacheLoaderException;
import org.apache.geode.cache.CacheTransactionManager;
import org.apache.geode.cache.CacheWriter;
import org.apache.geode.cache.CacheXmlException;
import org.apache.geode.cache.CustomExpiry;
import org.apache.geode.cache.DataPolicy;
import org.apache.geode.cache.Declarable;
import org.apache.geode.cache.DiskStore;
import org.apache.geode.cache.DiskStoreFactory;
import org.apache.geode.cache.DiskWriteAttributesFactory;
import org.apache.geode.cache.DynamicRegionFactory;
import org.apache.geode.cache.EvictionAction;
import org.apache.geode.cache.EvictionAlgorithm;
import org.apache.geode.cache.EvictionAttributes;
import org.apache.geode.cache.ExpirationAction;
import org.apache.geode.cache.ExpirationAttributes;
import org.apache.geode.cache.FixedPartitionAttributes;
import org.apache.geode.cache.InterestPolicy;
import org.apache.geode.cache.LoaderHelper;
import org.apache.geode.cache.LossAction;
import org.apache.geode.cache.MembershipAttributes;
import org.apache.geode.cache.MirrorType;
import org.apache.geode.cache.PartitionAttributes;
import org.apache.geode.cache.PartitionAttributesFactory;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.Region.Entry;
import org.apache.geode.cache.RegionAttributes;
import org.apache.geode.cache.RegionExistsException;
import org.apache.geode.cache.RegionShortcut;
import org.apache.geode.cache.ResumptionAction;
import org.apache.geode.cache.Scope;
import org.apache.geode.cache.SubscriptionAttributes;
import org.apache.geode.cache.TransactionListener;
import org.apache.geode.cache.TransactionWriter;
import org.apache.geode.cache.client.ClientCache;
import org.apache.geode.cache.client.ClientRegionShortcut;
import org.apache.geode.cache.client.Pool;
import org.apache.geode.cache.client.PoolFactory;
import org.apache.geode.cache.client.PoolManager;
import org.apache.geode.cache.execute.Function;
import org.apache.geode.cache.execute.FunctionService;
import org.apache.geode.cache.server.CacheServer;
import org.apache.geode.cache.server.ClientSubscriptionConfig;
import org.apache.geode.cache.server.ServerLoad;
import org.apache.geode.cache.server.ServerLoadProbeAdapter;
import org.apache.geode.cache.server.ServerMetrics;
import org.apache.geode.cache.util.ObjectSizer;
import org.apache.geode.cache.util.TransactionListenerAdapter;
import org.apache.geode.internal.AvailablePortHelper;
import org.apache.geode.internal.InternalDataSerializer;
import org.apache.geode.internal.InternalInstantiator;
import org.apache.geode.internal.cache.DiskWriteAttributesImpl;
import org.apache.geode.internal.cache.DistributedRegion;
import org.apache.geode.internal.cache.FixedPartitionAttributesImpl;
import org.apache.geode.internal.cache.GemFireCacheImpl;
import org.apache.geode.internal.cache.LocalRegion;
import org.apache.geode.internal.cache.PartitionedRegion;
import org.apache.geode.internal.cache.PoolFactoryImpl;
import org.apache.geode.internal.cache.functions.TestFunction;
import org.apache.geode.internal.cache.partitioned.fixed.QuarterPartitionResolver;
import org.apache.geode.internal.cache.persistence.DefaultDiskDirs;
import org.apache.geode.internal.cache.xmlcache.CacheCreation;
import org.apache.geode.internal.cache.xmlcache.CacheTransactionManagerCreation;
import org.apache.geode.internal.cache.xmlcache.CacheXml;
import org.apache.geode.internal.cache.xmlcache.CacheXmlGenerator;
import org.apache.geode.internal.cache.xmlcache.CacheXmlParser;
import org.apache.geode.internal.cache.xmlcache.ClientCacheCreation;
import org.apache.geode.internal.cache.xmlcache.Declarable2;
import org.apache.geode.internal.cache.xmlcache.FunctionServiceCreation;
import org.apache.geode.internal.cache.xmlcache.RegionAttributesCreation;
import org.apache.geode.internal.cache.xmlcache.RegionCreation;
import org.apache.geode.internal.cache.xmlcache.ResourceManagerCreation;
import org.apache.geode.internal.cache.xmlcache.SerializerCreation;
import org.apache.geode.test.dunit.Assert;
import org.apache.geode.test.dunit.Host;
import org.apache.geode.test.dunit.IgnoredException;
import org.apache.geode.test.dunit.LogWriterUtils;
import org.apache.geode.test.dunit.NetworkUtils;
import org.apache.geode.test.dunit.SerializableCallable;
import org.apache.geode.test.dunit.VM;

/**
 * Tests 7.0 cache.xml feature : Fixed Partitioning.
 *
 * @since GemFire 6.6
 */
public abstract class CacheXml66DUnitTest extends CacheXmlTestCase {

  @Override
  protected abstract String getGemFireVersion();

  private static final String ALIAS1;

  private static final String ALIAS2;

  static {
    String tmp_alias1 = "localhost";
    String tmp_alias2 = "localhost";
    ALIAS1 = tmp_alias1;
    ALIAS2 = tmp_alias2;
  }

  private static InetSocketAddress createINSA(String host, int port) {
    try {
      InetAddress hostAddr = InetAddress.getByName(host);
      return new InetSocketAddress(hostAddr, port);
    } catch (UnknownHostException cause) {
      throw new IllegalArgumentException("Unknown host " + host, cause);
    }
  }

  /**
   * test for checking default value of PR_Single_Hop feature. Test for checking default value of
   * multiuser-authentication attribute.
   */
  @Test
  public void testDefaultConnectionPool() throws Exception {
    getSystem();
    CacheCreation cache = new CacheCreation();
    PoolFactory f = cache.createPoolFactory();
    f.addLocator(ALIAS2, 3777);
    f.create("mypool");
    RegionAttributesCreation attrs = new RegionAttributesCreation(cache);
    attrs.setPoolName("mypool");
    cache.createVMRegion("rootNORMAL", attrs);
    testXml(cache);
    Cache c = getCache();
    assertNotNull(c);
    Region r = c.getRegion("rootNORMAL");
    assertNotNull(r);
    assertEquals("mypool", r.getAttributes().getPoolName());
    Pool cp = PoolManager.find("mypool");
    assertNotNull(cp);
    assertEquals(1, cp.getLocators().size());
    assertEquals(0, cp.getServers().size());
    assertEquals(createINSA(ALIAS2, 3777), cp.getLocators().get(0));
    assertEquals(PoolFactory.DEFAULT_FREE_CONNECTION_TIMEOUT, cp.getFreeConnectionTimeout());
    assertEquals(PoolFactory.DEFAULT_SERVER_CONNECTION_TIMEOUT, cp.getServerConnectionTimeout());
    assertEquals(PoolFactory.DEFAULT_LOAD_CONDITIONING_INTERVAL, cp.getLoadConditioningInterval());
    assertEquals(PoolFactory.DEFAULT_SOCKET_BUFFER_SIZE, cp.getSocketBufferSize());
    assertEquals(PoolFactory.DEFAULT_THREAD_LOCAL_CONNECTIONS, cp.getThreadLocalConnections());
    assertEquals(PoolFactory.DEFAULT_READ_TIMEOUT, cp.getReadTimeout());
    assertEquals(PoolFactory.DEFAULT_MIN_CONNECTIONS, cp.getMinConnections());
    assertEquals(PoolFactory.DEFAULT_MAX_CONNECTIONS, cp.getMaxConnections());
    assertEquals(PoolFactory.DEFAULT_RETRY_ATTEMPTS, cp.getRetryAttempts());
    assertEquals(PoolFactory.DEFAULT_IDLE_TIMEOUT, cp.getIdleTimeout());
    assertEquals(PoolFactory.DEFAULT_PING_INTERVAL, cp.getPingInterval());
    assertEquals(PoolFactory.DEFAULT_STATISTIC_INTERVAL, cp.getStatisticInterval());
    assertEquals(PoolFactory.DEFAULT_SERVER_GROUP, cp.getServerGroup());
    assertEquals(PoolFactory.DEFAULT_SUBSCRIPTION_ENABLED, cp.getSubscriptionEnabled());
    assertEquals(PoolFactory.DEFAULT_PR_SINGLE_HOP_ENABLED, cp.getPRSingleHopEnabled());
    assertEquals(PoolFactory.DEFAULT_SUBSCRIPTION_REDUNDANCY, cp.getSubscriptionRedundancy());
    assertEquals(PoolFactory.DEFAULT_SUBSCRIPTION_MESSAGE_TRACKING_TIMEOUT,
        cp.getSubscriptionMessageTrackingTimeout());
    assertEquals(PoolFactory.DEFAULT_SUBSCRIPTION_ACK_INTERVAL, cp.getSubscriptionAckInterval());
    assertEquals(PoolFactory.DEFAULT_MULTIUSER_AUTHENTICATION, cp.getMultiuserAuthentication());
  }

  @Test
  public void testDiskStore() throws Exception {
    CacheCreation cache = new CacheCreation();
    DiskStoreFactory dsf = cache.createDiskStoreFactory();
    File[] dirs1 = new File[] {new File("").getAbsoluteFile()};
    DiskStore ds1 =
        dsf.setAllowForceCompaction(true).setAutoCompact(true).setCompactionThreshold(100)
            .setMaxOplogSize(2).setTimeInterval(10).setWriteBufferSize(15).setQueueSize(12)
            .setDiskDirsAndSizes(dirs1, new int[] {1024 * 20}).create(getUniqueName() + 1);
    File[] dirs2 = new File[] {new File("").getAbsoluteFile()};
    DiskStore ds2 =
        dsf.setAllowForceCompaction(false).setAutoCompact(false).setCompactionThreshold(99)
            .setMaxOplogSize(1).setTimeInterval(9).setWriteBufferSize(14).setQueueSize(11)
            .setDiskDirsAndSizes(dirs2, new int[] {1024 * 40}).create(getUniqueName() + 2);

    RegionAttributesCreation attrs = new RegionAttributesCreation(cache);
    attrs.setScope(Scope.DISTRIBUTED_ACK);
    attrs.setDataPolicy(DataPolicy.PERSISTENT_REPLICATE);
    attrs.setDiskStoreName(getUniqueName() + 1);
    attrs.setDiskSynchronous(true);
    RegionCreation root = (RegionCreation) cache.createRegion("root", attrs);
    {
      attrs = new RegionAttributesCreation(cache);
      attrs.setScope(Scope.DISTRIBUTED_ACK);
      attrs.setDataPolicy(DataPolicy.PERSISTENT_REPLICATE);
      attrs.setDiskStoreName(getUniqueName() + 2);
      Region subwithdiskstore = root.createSubregion("subwithdiskstore", attrs);
    }

    {
      attrs = new RegionAttributesCreation(cache);
      attrs.setScope(Scope.DISTRIBUTED_ACK);
      attrs.setDataPolicy(DataPolicy.PERSISTENT_REPLICATE);
      Region subwithdefaultdiskstore = root.createSubregion("subwithdefaultdiskstore", attrs);
    }

    testXml(cache);
  }

  /**
   * test for enabling PRsingleHop feature. Test for enabling multiuser-authentication attribute.
   */
  @Test
  public void testExplicitConnectionPool() throws Exception {
    getSystem();
    CacheCreation cache = new CacheCreation();
    PoolFactory f = cache.createPoolFactory();
    f.addServer(ALIAS2, 3777).addServer(ALIAS1, 3888);
    f.setFreeConnectionTimeout(12345).setServerConnectionTimeout(111)
        .setLoadConditioningInterval(12345).setSocketBufferSize(12345)
        .setThreadLocalConnections(true).setPRSingleHopEnabled(true).setReadTimeout(12345)
        .setMinConnections(12346).setMaxConnections(12347).setRetryAttempts(12348)
        .setIdleTimeout(12349).setPingInterval(12350).setStatisticInterval(12351)
        .setServerGroup("mygroup")
        // commented this out until queues are implemented
        // .setQueueEnabled(true)
        .setSubscriptionRedundancy(12345).setSubscriptionMessageTrackingTimeout(12345)
        .setSubscriptionAckInterval(333).setMultiuserAuthentication(true);
    f.create("mypool");
    RegionAttributesCreation attrs = new RegionAttributesCreation(cache);
    attrs.setPoolName("mypool");
    attrs.setDataPolicy(DataPolicy.EMPTY); // required for multiuser mode
    cache.createVMRegion("rootNORMAL", attrs);
    addIgnoredException("Connection refused: connect");
    testXml(cache);
    Cache c = getCache();
    assertNotNull(c);
    Region r = c.getRegion("rootNORMAL");
    assertNotNull(r);
    assertEquals("mypool", r.getAttributes().getPoolName());
    Pool cp = PoolManager.find("mypool");
    assertNotNull(cp);
    assertEquals(0, cp.getLocators().size());
    assertEquals(2, cp.getServers().size());
    assertEquals(createINSA(ALIAS2, 3777), cp.getServers().get(0));
    assertEquals(createINSA(ALIAS1, 3888), cp.getServers().get(1));
    assertEquals(12345, cp.getFreeConnectionTimeout());
    assertEquals(111, cp.getServerConnectionTimeout());
    assertEquals(12345, cp.getLoadConditioningInterval());
    assertEquals(12345, cp.getSocketBufferSize());
    assertEquals(true, cp.getThreadLocalConnections());
    assertEquals(true, cp.getPRSingleHopEnabled());
    assertEquals(12345, cp.getReadTimeout());
    assertEquals(12346, cp.getMinConnections());
    assertEquals(12347, cp.getMaxConnections());
    assertEquals(12348, cp.getRetryAttempts());
    assertEquals(12349, cp.getIdleTimeout());
    assertEquals(12350, cp.getPingInterval());
    assertEquals(12351, cp.getStatisticInterval());
    assertEquals("mygroup", cp.getServerGroup());
    // TODO: commented this out until queues are implemented
    // assertIndexDetailsEquals(true, cp.getQueueEnabled());
    assertEquals(12345, cp.getSubscriptionRedundancy());
    assertEquals(12345, cp.getSubscriptionMessageTrackingTimeout());
    assertEquals(333, cp.getSubscriptionAckInterval());
    assertEquals(true, cp.getMultiuserAuthentication());
  }

  @Test
  public void testDiskStoreValidation() throws Exception {
    CacheCreation cache = new CacheCreation();
    DiskStoreFactory dsf = cache.createDiskStoreFactory();
    DiskStore ds1 = dsf.create(getUniqueName());

    RegionAttributesCreation attrs = new RegionAttributesCreation(cache);
    attrs.setScope(Scope.DISTRIBUTED_ACK);
    attrs.setDataPolicy(DataPolicy.REPLICATE);
    attrs.setDiskStoreName(getUniqueName());
    RegionCreation root;
    try {
      root = (RegionCreation) cache.createRegion("root", attrs);
      fail("Expected IllegalStateException to be thrown");
    } catch (IllegalStateException e) {
      assertTrue(e.getMessage().contains(
          "Only regions with persistence or overflow to disk can specify DiskStore"));
    }

    EvictionAttributes ea =
        EvictionAttributes.createLRUEntryAttributes(1000, EvictionAction.OVERFLOW_TO_DISK);
    attrs.setEvictionAttributes(ea);
    root = (RegionCreation) cache.createRegion("root", attrs);

    File dir = new File("testDiskStoreValidation");
    dir.mkdir();
    dir.deleteOnExit();

    File[] dirs2 = new File[] {dir, new File("").getAbsoluteFile()};
    try {
      AttributesFactory factory = new AttributesFactory();
      factory.setDiskDirs(dirs2);
      factory.setDiskStoreName(getUniqueName());
      RegionAttributes ra = factory.create();
      fail("Expected IllegalStateException to be thrown");
    } catch (IllegalStateException e) {
      assertTrue(e.getMessage()
          .contains(String.format("Deprecated API %s cannot be used with DiskStore %s",
              "setDiskDirs or setDiskWriteAttributes", getUniqueName())));
    }

    try {
      AttributesFactory factory = new AttributesFactory();
      factory.setDiskStoreName(getUniqueName());
      factory.setDiskDirs(dirs2);
      RegionAttributes ra = factory.create();
      fail("Expected IllegalStateException to be thrown");
    } catch (IllegalStateException e) {
      assertTrue(e.getMessage()
          .contains(String.format("Deprecated API %s cannot be used with DiskStore %s",
              "setDiskDirs", getUniqueName())));
    }

    testXml(cache);
  }

  @Test
  public void testDiskStoreFactory() throws Exception {
    CacheCreation cache = new CacheCreation();
    DiskStoreFactory dsf = cache.createDiskStoreFactory();

    dsf.setDiskDirs(new File[] {new File("non_exist_dir")});
    DiskStore ds1 = dsf.create(getUniqueName());

    dsf.setDiskDirs(new File[] {new File(".")});
    ds1 = dsf.create(getUniqueName());

    RegionAttributesCreation attrs = new RegionAttributesCreation(cache);
    attrs.setScope(Scope.DISTRIBUTED_ACK);
    attrs.setDataPolicy(DataPolicy.PERSISTENT_REPLICATE);
    attrs.setDiskStoreName(getUniqueName());
    AttributesFactory factory = new AttributesFactory(attrs);
    RegionAttributes ra = factory.create();

    RegionCreation root = (RegionCreation) cache.createRegion("root", ra);

    factory = new AttributesFactory();
    factory.setDiskStoreName(getUniqueName());
    factory.setScope(Scope.DISTRIBUTED_ACK);
    factory.setDataPolicy(DataPolicy.PERSISTENT_REPLICATE);

    ra = factory.create();

    RegionCreation root2 = (RegionCreation) cache.createRegion("root2", ra);

    factory = new AttributesFactory();
    factory.setDiskStoreName(null);
    factory.setScope(Scope.DISTRIBUTED_ACK);
    factory.setDataPolicy(DataPolicy.PERSISTENT_REPLICATE);

    ra = factory.create();

    RegionCreation root3 = (RegionCreation) cache.createRegion("root3", ra);

    testXml(cache);
  }

  @Test
  public void testRedefineOfDefaultDiskStore() throws Exception {
    CacheCreation cache = new CacheCreation();
    DiskStoreFactory dsf = cache.createDiskStoreFactory();
    dsf.setAutoCompact(!DiskStoreFactory.DEFAULT_AUTO_COMPACT);
    DiskStore ds1 = dsf.create(DiskStoreFactory.DEFAULT_DISK_STORE_NAME);

    RegionAttributesCreation attrs = new RegionAttributesCreation(cache);
    attrs.setScope(Scope.DISTRIBUTED_ACK);
    attrs.setDataPolicy(DataPolicy.PERSISTENT_REPLICATE);
    AttributesFactory factory = new AttributesFactory(attrs);
    RegionAttributes ra = factory.create();

    RegionCreation root = (RegionCreation) cache.createRegion("root", ra);

    testXml(cache);

    Cache c = getCache();
    assertNotNull(c);

    DiskStore ds2 = c.findDiskStore(DiskStoreFactory.DEFAULT_DISK_STORE_NAME);
    assertNotNull(ds2);
    assertEquals(ds1.getAutoCompact(), ds2.getAutoCompact());
  }

  /**
   * Make sure you can create a persistent partitioned region from xml.
   */
  @Test
  public void testPersistentPartition() throws Exception {
    CacheCreation cache = new CacheCreation();
    RegionAttributesCreation attrs = new RegionAttributesCreation(cache);
    attrs.setDataPolicy(DataPolicy.PERSISTENT_PARTITION);

    cache.createRegion("parRoot", attrs);

    Region r = cache.getRegion("parRoot");
    assertEquals(DataPolicy.PERSISTENT_PARTITION, r.getAttributes().getDataPolicy());

    testXml(cache);

    Cache c = getCache();
    assertNotNull(c);

    Region region = c.getRegion("parRoot");
    assertNotNull(region);

    assertEquals(DataPolicy.PERSISTENT_PARTITION, region.getAttributes().getDataPolicy());
    // since CacheTestCase remoteTearDown does not destroy PartitionedRegion
    region.localDestroyRegion();
  }

  @Test
  public void testBridgeAttributesRelatedToHAOverFlow65() throws Exception {
    CacheCreation cache = new CacheCreation();
    cache.setMessageSyncInterval(3445);
    CacheServer bs = cache.addCacheServer();
    ClientSubscriptionConfig csc = bs.getClientSubscriptionConfig();
    csc.setEvictionPolicy("entry");
    cache.getLogger().config("EvictionPolicy : " + csc.getEvictionPolicy());
    csc.setCapacity(501);
    cache.getLogger().config("EvictionCapacity : " + csc.getCapacity());
    File overflowDirectory = new File("overFlow");
    overflowDirectory.mkdirs();
    DiskStoreFactory dsf = cache.createDiskStoreFactory();
    File[] dirs1 = new File[] {overflowDirectory};
    DiskStore ds1 = dsf.setDiskDirs(dirs1).create(getUniqueName());
    csc.setDiskStoreName(getUniqueName());
    try {
      csc.setOverflowDirectory("overFlow");
    } catch (IllegalStateException e) {
      assertTrue(e.getMessage()
          .contains(String.format("Deprecated API %s cannot be used with DiskStore %s",
              "setOverflowDirectory", getUniqueName())));
    }

    cache.getLogger().config("Eviction disk store : " + csc.getDiskStoreName());
    bs.setPort(AvailablePortHelper.getRandomAvailableTCPPort());
    RegionAttributesCreation attrs = new RegionAttributesCreation(cache);
    attrs.setDataPolicy(DataPolicy.NORMAL);
    cache.createVMRegion("rootNORMAL", attrs);
    testXml(cache);
    Cache c = getCache();
    assertNotNull(c);
    CacheServer server = cache.getCacheServers().iterator().next();
    assertNotNull(server);
    ClientSubscriptionConfig chaqf = server.getClientSubscriptionConfig();
    assertEquals("entry", chaqf.getEvictionPolicy());
    assertEquals(501, chaqf.getCapacity());
    DiskStore dsi = cache.findDiskStore(chaqf.getDiskStoreName());
    assertEquals("overFlow", dsi.getDiskDirs()[0].toString());
  }

  @Test
  public void testClientSubscriptionQueueUsingDefaultDS() throws Exception {
    CacheCreation cache = new CacheCreation();
    cache.setMessageSyncInterval(3445);
    CacheServer bs = cache.addCacheServer();
    ClientSubscriptionConfig csc = bs.getClientSubscriptionConfig();
    csc.setEvictionPolicy("entry");
    cache.getLogger().config("EvictionPolicy : " + csc.getEvictionPolicy());
    csc.setCapacity(501);
    // don't set diskstore or overflowdir
    cache.getLogger().config("EvictionCapacity : " + csc.getCapacity());
    cache.getLogger().config("Eviction disk store : " + csc.getDiskStoreName());
    bs.setPort(AvailablePortHelper.getRandomAvailableTCPPort());
    RegionAttributesCreation attrs = new RegionAttributesCreation(cache);
    attrs.setDataPolicy(DataPolicy.NORMAL);
    cache.createVMRegion("rootNORMAL", attrs);
    testXml(cache);
    Cache c = getCache();
    assertNotNull(c);
    CacheServer server = cache.getCacheServers().iterator().next();
    assertNotNull(server);
    ClientSubscriptionConfig chaqf = server.getClientSubscriptionConfig();
    assertEquals("entry", chaqf.getEvictionPolicy());
    assertEquals(501, chaqf.getCapacity());
    File defaultDiskDir = DefaultDiskDirs.getDefaultDiskDirs()[0].getAbsoluteFile();
    File lockFile =
        new File(defaultDiskDir, "DRLK_IF" + GemFireCacheImpl.getDefaultDiskStoreName() + ".lk");
    assertTrue(lockFile.exists());
  }

  /**
   * Tests that a region created with a named attributes set programmatically for delta propogation
   * has the correct attributes.
   */
  @Test
  public void testTransactionWriter() throws Exception {
    CacheCreation creation = new CacheCreation();
    CacheTransactionManagerCreation ctmc = new CacheTransactionManagerCreation();
    ctmc.setWriter(new TestTransactionWriter());
    creation.addCacheTransactionManagerCreation(ctmc);
    testXml(creation);

    Cache c = getCache();
    assertTrue(c instanceof GemFireCacheImpl);
    c.loadCacheXml(generate(creation));

    TransactionWriter tw = c.getCacheTransactionManager().getWriter();
    assertTrue("tw should be TransactionWriter, but it is:" + tw,
        tw instanceof TestTransactionWriter);
  }

  /**
   * Tests that a region created with a named attributes with diskstore
   */
  @Test
  public void testDiskStoreInTemplates() throws Exception {
    File dir = new File("west");
    dir.mkdir();
    dir.deleteOnExit();

    dir = new File("east");
    dir.mkdir();
    dir.deleteOnExit();

    setXmlFile(findFile("ewtest.xml"));

    String regionName_west = "orders" + SEPARATOR + "west";
    String regionName_east = "orders" + SEPARATOR + "east";

    Cache cache = getCache();

    // verify diskstores
    DiskStore ds = cache.findDiskStore("persistentDiskStore1");
    assertNotNull(ds);
    assertEquals(500, ds.getQueueSize());
    File[] dirs = ds.getDiskDirs();
    assertEquals("west", dirs[0].getPath());

    ds = cache.findDiskStore("persistentDiskStore2");
    assertNotNull(ds);
    assertEquals(500, ds.getQueueSize());
    dirs = ds.getDiskDirs();
    assertEquals("east", dirs[0].getPath());

    // verify templates
    assertNotNull(cache.getRegionAttributes("nack"));
    RegionAttributes attrs = cache.getRegionAttributes("persistent");
    assertEquals(DataPolicy.PERSISTENT_REPLICATE, attrs.getDataPolicy());
    assertEquals(false, attrs.isDiskSynchronous());
    assertEquals("persistentDiskStore1", attrs.getDiskStoreName());

    Region region = cache.getRegion(regionName_west);
    assertNotNull(region);

    attrs = region.getAttributes();
    assertEquals(DataPolicy.PERSISTENT_REPLICATE, attrs.getDataPolicy());
    assertEquals(false, attrs.isDiskSynchronous());
    assertEquals("persistentDiskStore1", attrs.getDiskStoreName());

    region = cache.getRegion(regionName_east);
    assertNotNull(region);

    // Make sure that attributes can be "overridden"
    attrs = region.getAttributes();
    assertEquals(DataPolicy.PERSISTENT_REPLICATE, attrs.getDataPolicy());
    assertEquals(false, attrs.isDiskSynchronous());
    assertEquals("persistentDiskStore2", attrs.getDiskStoreName());

    // bug 41934
    String regionName_datap = "data-p";
    region = cache.getRegion(regionName_datap);
    assertNotNull(region);
    attrs = region.getAttributes();
    PartitionAttributes pa = attrs.getPartitionAttributes();
    assertEquals(1, pa.getRedundantCopies());
    assertEquals(3, pa.getTotalNumBuckets());
    assertEquals(DataPolicy.PERSISTENT_PARTITION, attrs.getDataPolicy());
  }

  @Test
  public void testBackupFiles() throws Exception {
    CacheCreation cache = new CacheCreation();
    File backup1 = new File("/back/me/up");
    File backup2 = new File("/me/too/please");
    cache.addBackup(backup1);
    cache.addBackup(backup2);
    testXml(cache);
    GemFireCacheImpl c = (GemFireCacheImpl) getCache();
    assertEquals(Arrays.asList(backup1, backup2), c.getBackupFiles());
  }

  @Test
  public void testClientCache() throws Exception {
    ClientCacheCreation cache = new ClientCacheCreation();
    cache.setCopyOnRead(true);
    testXml(cache);
    GemFireCacheImpl c = (GemFireCacheImpl) getCache();
    assertEquals(true, c.getCopyOnRead());
    assertEquals(true, c.isClient());
    for (ClientRegionShortcut pra : ClientRegionShortcut.values()) {
      assertNotNull(c.getRegionAttributes(pra.name()));
    }
    assertEquals(ClientRegionShortcut.values().length, c.listRegionAttributes().size());
  }

  @Test
  public void testNormalCache() throws Exception {
    CacheCreation cache = new CacheCreation();
    cache.setCopyOnRead(true);
    testXml(cache);
    GemFireCacheImpl c = (GemFireCacheImpl) getCache();
    assertEquals(true, c.getCopyOnRead());
    assertEquals(false, c.isClient());
    for (RegionShortcut pra : RegionShortcut.values()) {
      assertNotNull(c.getRegionAttributes(pra.name()));
    }
    assertEquals(RegionShortcut.values().length, c.listRegionAttributes().size());
  }

  @Test
  public void testPARTITION() throws Exception {
    CacheCreation cache = new CacheCreation();
    RegionCreation root = (RegionCreation) cache.createRegion("partition", "PARTITION");
    testXml(cache);
    GemFireCacheImpl c = (GemFireCacheImpl) getCache();
    Region r = c.getRegion("partition");
    assertNotNull(r);
    RegionAttributes ra = r.getAttributes();
    assertEquals(DataPolicy.PARTITION, ra.getDataPolicy());
    assertNotNull(ra.getPartitionAttributes());
    assertEquals(0, ra.getPartitionAttributes().getRedundantCopies());
  }

  @Test
  public void testPARTITION_REDUNDANT() throws Exception {
    CacheCreation cache = new CacheCreation();
    RegionCreation root = (RegionCreation) cache.createRegion("rpartition", "PARTITION_REDUNDANT");
    testXml(cache);
    GemFireCacheImpl c = (GemFireCacheImpl) getCache();
    Region r = c.getRegion("rpartition");
    assertNotNull(r);
    RegionAttributes ra = r.getAttributes();
    assertEquals(DataPolicy.PARTITION, ra.getDataPolicy());
    assertNotNull(ra.getPartitionAttributes());
    assertEquals(1, ra.getPartitionAttributes().getRedundantCopies());
  }

  @Test
  public void testPARTITION_PERSISTENT() throws Exception {
    CacheCreation cache = new CacheCreation();
    RegionCreation root = (RegionCreation) cache.createRegion("ppartition", "PARTITION_PERSISTENT");
    testXml(cache);
    GemFireCacheImpl c = (GemFireCacheImpl) getCache();
    Region r = c.getRegion("ppartition");
    assertNotNull(r);
    RegionAttributes ra = r.getAttributes();
    assertEquals(DataPolicy.PERSISTENT_PARTITION, ra.getDataPolicy());
    assertNotNull(ra.getPartitionAttributes());
    assertEquals(0, ra.getPartitionAttributes().getRedundantCopies());
  }

  @Test
  public void testPARTITION_REDUNDANT_PERSISTENT() throws Exception {
    CacheCreation cache = new CacheCreation();
    RegionCreation root =
        (RegionCreation) cache.createRegion("prpartition", "PARTITION_REDUNDANT_PERSISTENT");
    testXml(cache);
    GemFireCacheImpl c = (GemFireCacheImpl) getCache();
    Region r = c.getRegion("prpartition");
    assertNotNull(r);
    RegionAttributes ra = r.getAttributes();
    assertEquals(DataPolicy.PERSISTENT_PARTITION, ra.getDataPolicy());
    assertNotNull(ra.getPartitionAttributes());
    assertEquals(1, ra.getPartitionAttributes().getRedundantCopies());
  }

  @Test
  public void testPARTITION_OVERFLOW() throws Exception {
    CacheCreation cache = new CacheCreation();
    ResourceManagerCreation rmc = new ResourceManagerCreation();
    rmc.setEvictionHeapPercentage(55.0f);
    rmc.setCriticalHeapPercentage(80.0f);
    cache.setResourceManagerCreation(rmc);
    RegionCreation root =
        (RegionCreation) cache.createRegion("partitionoverflow", "PARTITION_OVERFLOW");
    testXml(cache);
    GemFireCacheImpl c = (GemFireCacheImpl) getCache();
    Region r = c.getRegion("partitionoverflow");
    assertNotNull(r);
    RegionAttributes ra = r.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(55.0f, c.getResourceManager().getEvictionHeapPercentage(), 0);
    assertEquals(80.0f, c.getResourceManager().getCriticalHeapPercentage(), 0);
  }

  @Test
  public void testPARTITION_REDUNDANT_OVERFLOW() throws Exception {
    CacheCreation cache = new CacheCreation();
    RegionCreation root =
        (RegionCreation) cache.createRegion("rpartitionoverflow", "PARTITION_REDUNDANT_OVERFLOW");
    testXml(cache);
    GemFireCacheImpl c = (GemFireCacheImpl) getCache();
    Region r = c.getRegion("rpartitionoverflow");
    assertNotNull(r);
    RegionAttributes ra = r.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 {
    CacheCreation cache = new CacheCreation();
    ResourceManagerCreation rmc = new ResourceManagerCreation();
    rmc.setCriticalHeapPercentage(80.0f);
    cache.setResourceManagerCreation(rmc);
