/*
 * 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 java.lang.System.currentTimeMillis;
import static java.lang.System.getProperties;
import static java.lang.System.setProperty;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static org.apache.geode.cache.ExpirationAction.DESTROY;
import static org.apache.geode.cache.ExpirationAction.INVALIDATE;
import static org.apache.geode.cache.Region.SEPARATOR;
import static org.apache.geode.cache.Region.SEPARATOR_CHAR;
import static org.apache.geode.internal.cache.ExpiryTask.permitExpiration;
import static org.apache.geode.internal.cache.ExpiryTask.suspendExpiration;
import static org.apache.geode.internal.cache.LocalRegion.EXPIRY_MS_PROPERTY;
import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
import static org.apache.geode.test.dunit.Invoke.invokeInEveryVM;
import static org.apache.geode.test.dunit.Wait.waitForExpiryClockToChange;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.text.SimpleDateFormat;
import java.util.Arrays;
import java.util.Collection;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.NoSuchElementException;
import java.util.Properties;
import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.atomic.AtomicInteger;

import org.junit.Test;

import org.apache.geode.cache.AttributesFactory;
import org.apache.geode.cache.AttributesMutator;
import org.apache.geode.cache.CacheException;
import org.apache.geode.cache.CacheStatistics;
import org.apache.geode.cache.CustomExpiry;
import org.apache.geode.cache.Declarable;
import org.apache.geode.cache.EntryDestroyedException;
import org.apache.geode.cache.EntryEvent;
import org.apache.geode.cache.EntryExistsException;
import org.apache.geode.cache.EntryNotFoundException;
import org.apache.geode.cache.ExpirationAction;
import org.apache.geode.cache.ExpirationAttributes;
import org.apache.geode.cache.LoaderHelper;
import org.apache.geode.cache.Operation;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.Region.Entry;
import org.apache.geode.cache.RegionAttributes;
import org.apache.geode.cache.RegionDestroyedException;
import org.apache.geode.cache.RegionEvent;
import org.apache.geode.cache.RegionFactory;
import org.apache.geode.cache.RegionReinitializedException;
import org.apache.geode.internal.Assert;
import org.apache.geode.internal.cache.EntryExpiryTask;
import org.apache.geode.internal.cache.EntrySnapshot;
import org.apache.geode.internal.cache.ExpiryTask;
import org.apache.geode.internal.cache.ExpiryTask.ExpiryTaskListener;
import org.apache.geode.internal.cache.LocalRegion;
import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.test.dunit.VM;
import org.apache.geode.test.dunit.Wait;
import org.apache.geode.test.dunit.cache.internal.JUnit4CacheTestCase;

/**
 * An abstract class whose test methods test the functionality of a region regardless of its scope.
 *
 * <P>
 *
 * This class also contains functionality that is used by subclasses. See
 * {@link #getRegionAttributes}.
 *
 * TODO:davidw: Test {@link CacheStatistics}
 *
 * @since GemFire 3.0
 */
public abstract class RegionTestCase extends JUnit4CacheTestCase {

  @Override
  public final void postTearDownCacheTestCase() throws Exception {
    postTearDownRegionTestCase();
  }

  protected void postTearDownRegionTestCase() throws Exception {}

  /**
   * Returns a region with the given name and the attributes for this test.
   *
   * @param <K> the type of keys in the region
   * @param <V> the type of values in the region
   * @param name the name of the region
   * @return the region
   * @see #getRegionAttributes
   */
  protected <K, V> Region<K, V> createRegion(String name) throws CacheException {
    RegionFactory<K, V> regionFactory = getCache().createRegionFactory(getRegionAttributes());
    return createRegion(name, regionFactory);
  }

  protected <K, V> Region<K, V> createRootRegion() throws CacheException {
    return createRootRegion(getRegionAttributes());
  }

  /*
   * Returns the attributes of a region to be tested by this test. Note that the decision as to
   * which attributes are used is left up to the concrete subclass.
   */
  protected abstract <K, V> RegionAttributes<K, V> getRegionAttributes();

  /** pauses only if no ack */
  protected void pauseIfNecessary() {}

  protected void pauseIfNecessary(int ms) {}

  /*
   * Make sure all messages done on region r have been processed on the remote side.
   */
  protected void flushIfNecessary(Region r) {
    // Only needed for no-ack regions
  }

  /**
   * Tests that creating an entry in a region actually creates it
   *
   * @see Region#containsKey
   * @see Region#containsValueForKey
   */
  @Test
  public void testContainsKey() throws CacheException {
    String key = getUniqueName();
    Region<Object, Object> region = createRegion(key);
    Object value = 42;

    assertFalse(region.containsKey(key));
    region.create(key, null);
    assertFalse(region.containsValueForKey(key));

    Region.Entry entry = region.getEntry(key);
    assertNotNull(entry);
    assertEquals(entry.getKey(), key);
    assertNull(entry.getValue());

    region.put(key, value);
    assertTrue(region.containsValueForKey(key));
    assertEquals(entry, region.getEntry(key));
    if (entry.isLocal()) {
      assertEquals(value, entry.getValue());
    } else {
      assertEquals(value, region.getEntry(key).getValue());
    }
  }

  /**
   * Tests that creating or getting entries at an improper time throws exceptions.
   *
   * @see Region#get
   * @see Region#getEntry
   * @see Region#create
   */
  @Test
  public void testBadRegionAccess() throws CacheException {
    String key = getUniqueName();
    Region<Object, Object> region = createRegion(key);

    assertNull(region.get(key));
    assertNull(region.getEntry(key));

    Integer value = 42;
    region.create(key, value);

    try {
      // partitioned regions are logging the EntryExistsException, so emit
      // a directive to ignore it
      logger.info("<ExpectedException action=add>"
          + "org.apache.geode.cache.EntryExistsException" + "</ExpectedException>");
      region.create(key, value);
      fail("Should have thrown an EntryExistsException");

    } catch (EntryExistsException ex) {
      // okay...
    } finally {
      logger.info("<ExpectedException action=remove>"
          + "org.apache.geode.cache.EntryExistsException" + "</ExpectedException>");
    }
  }

  /**
   * Tests that {@link Region#put} on a previously non-existent region entry creates it.
   */
  @Test
  public void testPutNonExistentEntry() throws CacheException {
    String key = getUniqueName();
    Region<Object, Object> region = createRegion(key);

    assertNull(region.getEntry(key));

    Object value = 42;
    region.put(key, value);

    Region.Entry entry = region.getEntry(key);
    assertNotNull(entry);
    assertEquals(key, entry.getKey());
    assertEquals(value, entry.getValue());
    assertEquals(value, region.get(key));

    try {
      Collection values = region.values();
      assertEquals(1, values.size());
      assertEquals(value, values.iterator().next());
    } catch (UnsupportedOperationException uoe) {
      logger
          .info("Region.values() reported UnsupportedOperation");
    }
  }

  /**
   * Indicate whether subregions are supported
   *
   * @return whether subregions are supported
   */
  protected boolean supportsSubregions() {
    return true;
  }

  /**
   * Indicate whether localDestroy and localInvalidate are supported
   *
   * @return true if they are supported
   */
  protected boolean supportsLocalDestroyAndLocalInvalidate() {
    return true;
  }

  /**
   * Tests that sending <code>null</code> to various APIs throws the appropriate exception.
   */
  @Test
  public void testNulls() throws CacheException {
    if (!supportsSubregions()) {
      return;
    }
    String key = getUniqueName();
    Region<Object, Object> region = createRegion(key);

    try {
      region.getSubregion(null);
      fail("Should have thrown an IllegalArgumentException");

    } catch (IllegalArgumentException ex) {
      // pass..
    }

    try {
      region.createSubregion(null, region.getAttributes());
      fail("Should have thrown an IllegalArgumentException");

    } catch (IllegalArgumentException ex) {
      // pass..
    }

    try {
      region.createSubregion("TEST", null);
      fail("Should have thrown an IllegalArgumentException");

    } catch (IllegalArgumentException ex) {
      // pass..
    }

    assertEquals("/", SEPARATOR);
    assertEquals('/', SEPARATOR_CHAR);
    try {
      region.createSubregion("BAD" + SEPARATOR + "TEST", region.getAttributes());
      fail("Should have thrown an IllegalArgumentException");

    } catch (IllegalArgumentException ex) {
      // pass..
    }

    try {
      region.createSubregion("", region.getAttributes());
      fail("Should have thrown an IllegalArgumentException");

    } catch (IllegalArgumentException ex) {
      // pass..
    }

    try {
      region.getEntry(null);
      fail("Should have thrown a NullPointerException");

    } catch (NullPointerException ex) {
      // pass..
    }

    try {
      region.get(null);
      fail("Should have thrown a NullPointerException");

    } catch (NullPointerException ex) {
      // pass..
    }

    try {
      region.get(null, null);
      fail("Should have thrown a NullPointerException");

    } catch (NullPointerException ex) {
      // pass..
    }

    try {
      region.put(null, 42);
      fail("Should have thrown a NullPointerException");

    } catch (NullPointerException ex) {
      // pass..
    }

    try {
      region.put(key, null);
      fail("Should have thrown a NullPointerException");

    } catch (NullPointerException ex) {
      // pass..
    }

    try {
      region.destroy(null);
      fail("Should have thrown a NullPointerException");

    } catch (NullPointerException ex) {
      // pass..
    }
  }

  /**
   * Tests creating subregions. Note that this tests accesses the Region's
   * {@link Region#getStatistics statistics}, so the region must have been created with statistics
   * enabled.
   */
  @Test
  public void testCreateSubRegions() throws CacheException {
    if (!supportsSubregions()) {
      return;
    }
    String name = getUniqueName();

    RegionAttributes<Object, Object> attrs = getRegionAttributes();
    RegionFactory<Object, Object> regionFactory = getCache().createRegionFactory(attrs);
    regionFactory.setStatisticsEnabled(true);

    Region<Object, Object> region = createRegion(name, regionFactory);


    attrs = region.getAttributes();

    CacheStatistics stats = region.getStatistics();
    long lastAccessed = stats.getLastAccessedTime();
    long lastModified = stats.getLastModifiedTime();

    try {
      region.createSubregion(name + SEPARATOR + "BAD", attrs);
      fail("Should have thrown an IllegalArgumentException");

    } catch (IllegalArgumentException ex) {
      CacheStatistics stats2 = region.getStatistics();
      assertEquals(lastAccessed, stats2.getLastAccessedTime());
      assertEquals(lastModified, stats2.getLastModifiedTime());
    }

    Region subregion = region.createSubregion(name, attrs);
    assertTrue(attrs != subregion.getAttributes());

    Set subregions = region.subregions(false);
    assertEquals(1, subregions.size());
    assertEquals(subregion, subregions.iterator().next());
  }

  public static boolean entryIsLocal(Region.Entry re) {
    if (re instanceof EntrySnapshot) {
      return ((EntrySnapshot) re).wasInitiallyLocal();
    } else {
      return re.isLocal();
    }
  }

  /**
   * Tests {@link Region#destroy destroying} an entry and attempting to access it afterwards.
   */
  @Test
  public void testDestroyEntry() throws CacheException {
    String key = getUniqueName();
    Object value = 42;

    Region<Object, Object> region = createRegion(key);

    try {
      region.destroy(key);
      fail("Should have thrown an EntryNotFoundException");

    } catch (EntryNotFoundException ex) {
      // pass...
    }

    region.put(key, value);

    Region.Entry entry = region.getEntry(key);
    assertNotNull(entry);

    region.destroy(key);

    Region.Entry entry2 = region.getEntry(key);
    logger
        .info("Found entry for destroyed key: " + entry2);
    assertNull(entry2);
    if (entry.isLocal()) {
      assertTrue(entry.isDestroyed());
    } else {
      assertFalse(entry.isDestroyed());
    }
    assertEquals(0, region.keySet().size());

    if (entry.isLocal()) {
      try {
        entry.getKey();
        fail("Should have thrown an EntryDestroyedException");
      } catch (EntryDestroyedException ex) {
        // pass...
      }
      try {
        entry.getRegion();
        fail("Should have thrown an EntryDestroyedException");

      } catch (EntryDestroyedException ex) {
        // pass...
      }

      try {
        entry.getStatistics();
        fail("Should have thrown an EntryDestroyedException");

      } catch (EntryDestroyedException ex) {
        // pass...
      }

      try {
        entry.getUserAttribute();
        fail("Should have thrown an EntryDestroyedException");

      } catch (EntryDestroyedException ex) {
        // pass...
      }

      try {
        entry.setUserAttribute("blah");
        fail("Should have thrown an EntryDestroyedException");

      } catch (EntryDestroyedException ex) {
        // pass...
      }

      try {
        entry.getValue();
        fail("Should have thrown an EntryDestroyedException");

      } catch (EntryDestroyedException ex) {
        // pass...
      }
    }
  }

  /**
   * Tests destroying an entire region and that accessing it after it has been destory causes a
   * {@link RegionDestroyedException}.
   *
   * @see Region#destroyRegion
   */
  @Test
  public void testDestroyRegion() throws CacheException {
    if (!supportsSubregions()) {
      return;
    }
    String name = getUniqueName();
    Object key = "KEY";
    Object value = "VALUE";

    Region<Object, Object> region = createRegion(name);
    region.put(key, value);

    Region.Entry entry = region.getEntry(key);
    assertNotNull(entry);

    region.createSubregion("SUB", region.getAttributes());
    region.destroyRegion();

    assertTrue(entry.isDestroyed());
    assertTrue(region.isDestroyed());

    try {
      region.containsKey(key);
      fail("Should have thrown a RegionDestroyedException");

    } catch (RegionDestroyedException ex) {
      // pass..
    }

    try {
      region.containsValueForKey(key);
      fail("Should have thrown a RegionDestroyedException");

    } catch (RegionDestroyedException ex) {
      // pass..
    }

    try {
      region.create(key, value);
      fail("Should have thrown a RegionDestroyedException");

    } catch (RegionDestroyedException ex) {
      // pass..
    }

    try {
      region.create(key, value, "BLAH");
      fail("Should have thrown a RegionDestroyedException");

    } catch (RegionDestroyedException ex) {
      // pass..
    }

    try {
      region.createSubregion("SUB", getRegionAttributes());
      fail("Should have thrown a RegionDestroyedException");

    } catch (RegionDestroyedException ex) {
      // pass..
    }

    try {
      region.destroy(key);
      fail("Should have thrown a RegionDestroyedException");

    } catch (RegionDestroyedException ex) {
      // pass..
    }

    try {
      region.destroy(key, "BLAH");
      fail("Should have thrown a RegionDestroyedException");

    } catch (RegionDestroyedException ex) {
      // pass..
    }

    try {
      region.destroyRegion();
      fail("Should have thrown a RegionDestroyedException");

    } catch (RegionDestroyedException ex) {
      // pass..
    }

    try {
      region.destroyRegion("ARG");
      fail("Should have thrown a RegionDestroyedException");

    } catch (RegionDestroyedException ex) {
      // pass..
    }

    try {
      region.entrySet(false);
      fail("Should have thrown a RegionDestroyedException");

    } catch (RegionDestroyedException ex) {
      // pass..
    }

    try {
      region.get(key);
      fail("Should have thrown a RegionDestroyedException");

    } catch (RegionDestroyedException ex) {
      // pass..
    }

    try {
      region.get(key, "ARG");
      fail("Should have thrown a RegionDestroyedException");

    } catch (RegionDestroyedException ex) {
      // pass..
    }

    try {
      region.containsKey(key);
      fail("Should have thrown a RegionDestroyedException");

    } catch (RegionDestroyedException ex) {
      // pass..
    }

    region.getAttributes();

    try {
      region.getAttributesMutator();
      fail("Should have thrown a RegionDestroyedException");

    } catch (RegionDestroyedException ex) {
      // pass..
    }

    try {
      region.containsKey(key);
      fail("Should have thrown a RegionDestroyedException");

    } catch (RegionDestroyedException ex) {
      // pass..
    }

    try {
      region.getCache();

    } catch (RegionDestroyedException ex) {
      fail("getCache() shouldn't have thrown a RegionDestroyedException");
    }

    try {
      region.getEntry(key);
      fail("Should have thrown a RegionDestroyedException");

    } catch (RegionDestroyedException ex) {
      // pass..
    }

    try {
      region.getDistributedLock(key);
      fail("Should have thrown a RegionDestroyedException");

    } catch (RegionDestroyedException ex) {
      // pass..
    }

    assertEquals(name, region.getName());


    region.getParentRegion();

    assertEquals(SEPARATOR + "root" + SEPARATOR + name, region.getFullPath());
    assertEquals(name, region.getName());

    try {
      region.getRegionDistributedLock();
      fail("Should have thrown a RegionDestroyedException");

    } catch (RegionDestroyedException ex) {
      // pass..
    }

    try {
      region.getStatistics();
      fail("Should have thrown a RegionDestroyedException");

    } catch (RegionDestroyedException ex) {
      // pass..
    }

    try {
      region.getSubregion("SUB");
      fail("Should have thrown a RegionDestroyedException");

    } catch (RegionDestroyedException ex) {
      // pass..
    }

    region.getUserAttribute();

    try {
      region.invalidate(key);
      fail("Should have thrown a RegionDestroyedException");

    } catch (RegionDestroyedException ex) {
      // pass..
    }

    try {
      region.invalidateRegion();
      fail("Should have thrown a RegionDestroyedException");

    } catch (RegionDestroyedException ex) {
      // pass..
    }

    try {
      region.keySet();
      fail("Should have thrown a RegionDestroyedException");

    } catch (RegionDestroyedException ex) {
      // pass..
    }

    try {
      region.localDestroy(key);
      fail("Should have thrown a RegionDestroyedException");

    } catch (RegionDestroyedException ex) {
      // pass..
    }

    try {
      region.localDestroyRegion();
      fail("Should have thrown a RegionDestroyedException");

    } catch (RegionDestroyedException ex) {
      // pass..
    }

    try {
      region.localInvalidate(key);
      fail("Should have thrown a RegionDestroyedException");

    } catch (RegionDestroyedException ex) {
      // pass..
    }

    try {
      region.localInvalidateRegion();
      fail("Should have thrown a RegionDestroyedException");

    } catch (RegionDestroyedException ex) {
      // pass..
    }

    try {
      region.put(key, value);
      fail("Should have thrown a RegionDestroyedException");

    } catch (RegionDestroyedException ex) {
      // pass..
    }

    try {
      region.put(key, value, "ARG");
      fail("Should have thrown a RegionDestroyedException");

    } catch (RegionDestroyedException ex) {
      // pass..
    }

    try {
      region.setUserAttribute("ATTR");
      fail("Should have thrown a RegionDestroyedException");

    } catch (RegionDestroyedException ex) {
      // pass..
    }

    try {
      region.subregions(true);
      fail("Should have thrown a RegionDestroyedException");

    } catch (RegionDestroyedException ex) {
      // pass..
    }

    try {
      region.values();
      fail("Should have thrown a RegionDestroyedException");

    } catch (RegionDestroyedException ex) {
      // pass..
    }
  }

  /**
   * Tests the {@link Region#entrySet(boolean)} method without recursion
   */
  @Test
  public void testEntries() throws CacheException {
    String name = getUniqueName();
    Region<Object, Object> region = createRegion(name);
    assertEquals(0, region.entrySet(true).size());
    assertEquals(0, region.entrySet(false).size());

    region.put("A", "a");
    region.put("B", "b");
    region.put("C", "c");

    {
      Set entries = region.entrySet(false);
      assertEquals(3, entries.size());

      Set<Object> keys = new HashSet<>(Arrays.asList("A", "B", "C"));
      Iterator iter = entries.iterator();
      for (int i = 0; i < 3; i++) {
        assertTrue(iter.hasNext());
        assertTrue(keys.remove(((Region.Entry) iter.next()).getKey()));
      }
      assertFalse(iter.hasNext());
    }

    {
      Set entries = region.entrySet(true);
      assertEquals(3, entries.size());

      Set<Object> keys = new HashSet<>(Arrays.asList("A", "B", "C"));
      Iterator iter = entries.iterator();
      for (int i = 0; i < 3; i++) {
        assertTrue(iter.hasNext());
        assertTrue(keys.remove(((Region.Entry) iter.next()).getKey()));
      }
      assertFalse(iter.hasNext());
    }
    {
      Iterator iter = region.entrySet(false).iterator();
      Region.Entry entry = (Region.Entry) iter.next();
      region.destroy(entry.getKey());
      if (entry.isLocal()) {
        assertTrue(entry.isDestroyed());
      } else {
        assertFalse(entry.isDestroyed());
      }
    }

  }

  /**
   * Tests the {@link Region#entrySet} method with recursion
   */
  @Test
  public void testEntriesRecursive() throws CacheException {
    if (!supportsSubregions()) {
      return;
    }
    String name = getUniqueName();
    Region<String, String> region = createRegion(name);

    region.put("A", "a");
    region.put("B", "b");
    region.put("C", "c");
    Region<String, String> sub = region.createSubregion("SUB", region.getAttributes());
    sub.put("D", "d");
    sub.put("E", "e");
    sub.put("F", "f");

    {
      Set entries = region.entrySet(true);
      assertEquals(6, entries.size());


      Set<Object> keys = new HashSet<>(Arrays.asList("A", "B", "C", "D", "E", "F"));
      Iterator iter = entries.iterator();
      for (int i = 0; i < 6; i++) {
        assertTrue("!hasNext, i=" + i, iter.hasNext());
        assertTrue("remove returned false, i=" + i,
            keys.remove(((Region.Entry) iter.next()).getKey()));
      }
      assertFalse(iter.hasNext());
    }

    {
      Set<Region.Entry<?, ?>> entries = region.entrySet(true);
      assertEquals(6, entries.size());


      Set<Object> keys = new HashSet<>(Arrays.asList("A", "B", "C", "D", "E", "F"));
      Iterator<Region.Entry<?, ?>> iter = entries.iterator();
      for (int i = 0; i < 6; i++) {
        assertTrue("!hasNext, i=" + i, iter.hasNext());
        assertTrue("remove returned false, i=" + i,
            keys.remove(iter.next().getKey()));
      }
      assertFalse(iter.hasNext());
    }

    {
      Iterator iter = region.entrySet(true).iterator();
      Region.Entry<String, String> entry = (Region.Entry) iter.next();
      String key = entry.getKey();
      region.destroy(key);
      assertFalse(region.containsKey(key));
      assertTrue(entry.isDestroyed());
    }

  }


  /**
   * Tests the {@link Region#getCache} method (for what it's worth)
   */
  @Test
  public void testGetCache() throws CacheException {
    String name = getUniqueName();
    Region<Object, Object> region = createRegion(name);
    assertSame(getCache(), region.getCache());
  }

  /**
   * Tests the {@link Region#getName} method
   */
  @Test
  public void testGetName() throws CacheException {
    String name = getUniqueName();
    Region<Object, Object> region = createRegion(name);
    assertEquals(name, region.getName());

    assertEquals("root", region.getParentRegion().getName());
  }

  /**
   * Tests the {@link Region#getFullPath} method
   */
  @Test
  public void testGetPathFromRoot() throws CacheException {
    if (!supportsSubregions()) {
      return;
    }
    String name = getUniqueName();

    Region<Object, Object> region = createRegion(name);
    String fullPath = SEPARATOR + "root" + SEPARATOR + name;
    assertEquals(fullPath, region.getFullPath());
    assertEquals(SEPARATOR + "root", region.getParentRegion().getFullPath());
    RegionFactory<Object, Object> regionFactory = cache.createRegionFactory(region.getAttributes());
    Region sub = regionFactory.createSubregion(region, "SUB");
    assertEquals(fullPath + SEPARATOR + "SUB", sub.getFullPath());
  }

  /**
   * Tests the {@link Region#getParentRegion} method
   */
  @Test
  public void testGetParentRegion() throws CacheException {
    if (!supportsSubregions()) {
      return;
    }
    String name = getUniqueName();

    Region<Object, Object> region = createRegion(name);
    assertEquals(getRootRegion(), region.getParentRegion());

    RegionFactory<Object, Object> regionFactory = cache.createRegionFactory(region.getAttributes());
    Region sub = regionFactory.createSubregion(region, "SUB");
    assertEquals(region, sub.getParentRegion());
    assertSame(sub, region.getSubregion("SUB"));
    assertNotNull(sub.getAttributes());
  }

  /**
   * Tests a <code>Region</code>'s user attribute
   *
   * @see Region#setUserAttribute
   */
  @Test
  public void testRegionUserAttribute() throws CacheException {
    String name = getUniqueName();
    Object value = "USER_ATTRIBUTE";

    Region<Object, Object> region = createRegion(name);
    assertNull(region.getUserAttribute());

    region.setUserAttribute(value);
    assertEquals(value, region.getUserAttribute());
  }

  /**
   * Tests a region entry's user attribute
   */
  @Test
  public void testEntryUserAttribute() throws CacheException {
    String name = getUniqueName();
    String key = "KEY";

    String attr = "USER_ATTRIBUTE";

    Region<Object, Object> region = createRegion(name);
    region.create(key, null);

    Region.Entry entry = region.getEntry(key);
    entry.setUserAttribute(attr);
    assertEquals(attr, entry.getUserAttribute());

    entry = region.getEntry(key);
    assertEquals(attr, entry.getUserAttribute());
  }

  /**
   * Tests invalidating a region entry
   */
  @Test
  public void testInvalidateEntry() throws CacheException {
    String name = getUniqueName();
    Object key = "KEY";
    Object value = "VALUE";

    Region<Object, Object> region = createRegion(name);
    region.put(key, value);

    long beforeInvalidates = getCache()
        .getCachePerfStats().getInvalidates();
    Region.Entry entry = region.getEntry(key);
    region.invalidate(key);
    if (entry.isLocal()) {
      assertNull(entry.getValue());
    }
    assertNull(region.get(key));
    long afterInvalidates = getCache()
        .getCachePerfStats().getInvalidates();
    assertEquals("Invalidate CachePerfStats incorrect", beforeInvalidates + 1, afterInvalidates);
  }

  /**
   * Tests invalidating an entire region
   */
  @Test
  public void testInvalidateRegion() throws CacheException {
    String name = getUniqueName();

    Region<Object, Object> region = createRegion(name);
    region.put("A", "a");
    region.put("B", "b");
    region.put("C", "c");

    for (int i = 0; i < 50; i++) {
      region.put("Key=" + i, "Value-" + i);
    }

    region.invalidateRegion();

    Region.Entry entry;
    entry = region.getEntry("A");
    assertNotNull(entry);
    assertNull(entry.getValue());

    entry = region.getEntry("B");
    assertNotNull(entry);
    assertNull(entry.getValue());

    entry = region.getEntry("C");
    assertNotNull(entry);
    assertNull(entry.getValue());

    for (int i = 0; i < 50; i++) {
      String key = "Key=" + i;
      assertFalse("containsValueForKey returned true for key " + key,
          region.containsValueForKey(key));
      assertTrue("containsKey returned false for key " + key, region.containsKey(key));
    }
  }

  /**
   * Tests the {@link Region#keySet()} method.
   */
  @Test
  public void testKeys() throws CacheException {
    String name = getUniqueName();

    Region<Object, Object> region = createRegion(name);
    assertEquals(0, region.keySet().size());

    region.put("A", "a");
    region.put("B", "b");
    region.put("C", "c");

    {
      Set keys = region.keySet();
      assertEquals(3, keys.size());

      assertTrue(keys.contains("A"));
      assertTrue(keys.contains("B"));
      assertTrue(keys.contains("C"));
    }
  }

  /**
   * Tests {@link Region#localDestroy locally destroying} an entry and attempting to access it
   * afterwards. (Not too useful with a <code>LOCAL</code> region.)
   */
  @Test
  public void testLocalDestroyEntry() throws CacheException {
    if (!supportsLocalDestroyAndLocalInvalidate()) {
      return;
    }
    String key = getUniqueName();
    Object value = 42;

    Region<Object, Object> region = createRegion(key);

    boolean isMirrored = getRegionAttributes().getMirrorType().isMirrored();

    try {
      region.localDestroy(key);
      if (isMirrored) {
        fail("Should have thrown an IllegalStateException");
      }
      fail("Should have thrown an EntryNotFoundException");
    } catch (EntryNotFoundException ex) {
      // pass...
    } catch (IllegalStateException ex) {
      if (!isMirrored) {
        throw ex;
      } else {
        return; // abort test
      }
    }

    region.put(key, value);

    Region.Entry entry = region.getEntry(key);
    assertNotNull(entry);

    region.localDestroy(key);

    assertNull(region.getEntry(key));
    assertTrue(entry.isDestroyed());
    assertEquals(0, region.keySet().size());

    try {
      entry.getKey();
      fail("Should have thrown an EntryDestroyedException");

    } catch (EntryDestroyedException ex) {
      // pass...
    }

    try {
      entry.getRegion();
      fail("Should have thrown an EntryDestroyedException");

    } catch (EntryDestroyedException ex) {
      // pass...
    }

    try {
      entry.getStatistics();
      fail("Should have thrown an EntryDestroyedException");

    } catch (EntryDestroyedException ex) {
      // pass...
    }

    try {
      entry.getUserAttribute();
      fail("Should have thrown an EntryDestroyedException");

    } catch (EntryDestroyedException ex) {
      // pass...
    }

    try {
      entry.setUserAttribute("blah");
      fail("Should have thrown an EntryDestroyedException");

    } catch (EntryDestroyedException ex) {
      // pass...
    }

    try {
      entry.getValue();
      fail("Should have thrown an EntryDestroyedException");

    } catch (EntryDestroyedException ex) {
      // pass...
    }
  }

  /**
   * Tests locally destroying an entire region and that accessing it after it has been destory
   * causes a {@link RegionDestroyedException}.
   *
   * @see Region#localDestroyRegion
   */
  @Test
  public void testLocalDestroyRegion() throws CacheException {
    String name = getUniqueName();
    Object key = "KEY";
    Object value = "VALUE";

    Region<Object, Object> region = createRegion(name);
