extras/rya.export/export.accumulo/src/test/java/org/apache/rya/export/accumulo/AccumuloRyaStatementStoreTest.java - rya - Git at Google

 /*
  * 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.rya.export.accumulo;

 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;

 import java.util.Date;
 import java.util.Iterator;

 import org.apache.hadoop.conf.Configuration;
 import org.apache.log4j.LogManager;
 import org.apache.log4j.Logger;
 import org.apache.rya.api.domain.RyaStatement;
 import org.apache.rya.export.InstanceType;
 import org.apache.rya.export.MergePolicy;
 import org.apache.rya.export.accumulo.util.AccumuloInstanceDriver;
 import org.apache.rya.export.api.MergerException;
 import org.apache.rya.export.api.conf.AccumuloMergeConfiguration;
 import org.apache.rya.export.api.store.AddStatementException;
 import org.apache.rya.export.api.store.FetchStatementException;
 import org.apache.rya.export.api.store.RemoveStatementException;
 import org.apache.rya.export.api.store.RyaStatementStore;
 import org.apache.rya.export.api.store.UpdateStatementException;
 import org.apache.rya.indexing.accumulo.ConfigUtils;
 import org.junit.After;
 import org.junit.AfterClass;
 import org.junit.Before;
 import org.junit.BeforeClass;
 import org.junit.Test;

 import com.google.common.collect.ImmutableList;

 /**
  * Tests the methods of {@link AccumuloRyaStatementStore}.
  */
 public class AccumuloRyaStatementStoreTest {
     private static final Logger log = LogManager.getLogger(AccumuloRyaStatementStoreTest.class);
     private static final InstanceType INSTANCE_TYPE = InstanceType.MOCK;

     private static final boolean IS_MOCK = INSTANCE_TYPE == InstanceType.MOCK;
     private static final String USER_NAME = IS_MOCK ? "test_user" : AccumuloInstanceDriver.ROOT_USER_NAME;
     private static final String PASSWORD = "password";
     private static final String INSTANCE_NAME = "test_instance";
     private static final String AUTHS = "test_auth";
     private static final String RYA_TABLE_PREFIX = "test_";
     private static final String ZOOKEEPERS = "localhost";

     // Rya data store and connections.
     private static AccumuloInstanceDriver accumuloInstanceDriver = null;

     private static final Date DATE = new Date();

     private static final ImmutableList<RyaStatement> RYA_STATEMENTS = ImmutableList.of(
         TestUtils.createRyaStatement("Adam", "analyzes", "apple", DATE),
         TestUtils.createRyaStatement("Bob", "bites", "burger", DATE),
         TestUtils.createRyaStatement("Charlie", "checks", "chores", DATE),
         TestUtils.createRyaStatement("Debbie", "drives", "deal", DATE),
         TestUtils.createRyaStatement("Emma", "eats", "everything", DATE)
     );

     @BeforeClass
     public static void setupResources() throws Exception {
         // Initialize the Accumulo instance that will be used to store Triples and get a connection to it.
         accumuloInstanceDriver = startAccumuloInstanceDriver();
     }

     @Before
     public void setUpPerTest() throws Exception {
         accumuloInstanceDriver.setUpTables();
         accumuloInstanceDriver.setUpDao();
         accumuloInstanceDriver.setUpConfig();
     }

     @After
     public void tearDownPerTest() throws Exception {
         log.info("tearDownPerTest(): tearing down now.");
         accumuloInstanceDriver.tearDownTables();
         accumuloInstanceDriver.tearDownDao();
     }

     @AfterClass
     public static void tearDownPerClass() throws Exception {
         log.info("tearDownPerClass(): tearing down now.");
         accumuloInstanceDriver.tearDown();
     }

     @Test
     public void testFetchStatements() throws MergerException {
         final AccumuloRyaStatementStore accumuloRyaStatementStore = createAccumuloRyaStatementStore();

         for (final RyaStatement ryaStatement : RYA_STATEMENTS) {
             accumuloRyaStatementStore.addStatement(ryaStatement);
         }

         accumuloRyaStatementStore.fetchStatements();
     }

     @Test
     public void testRemoveAddStatements() throws MergerException {
         final AccumuloRyaStatementStore accumuloRyaStatementStore = createAccumuloRyaStatementStore();

         for (final RyaStatement ryaStatement : RYA_STATEMENTS) {
             accumuloRyaStatementStore.addStatement(ryaStatement);
         }
         final RyaStatement stmnt = RYA_STATEMENTS.get(0);

         assertTrue(accumuloRyaStatementStore.containsStatement(stmnt));
         accumuloRyaStatementStore.removeStatement(stmnt);
         assertFalse(accumuloRyaStatementStore.containsStatement(stmnt));

         accumuloRyaStatementStore.addStatement(stmnt);
         assertTrue(accumuloRyaStatementStore.containsStatement(stmnt));
     }

     @Test (expected = FetchStatementException.class)
     public void testFetchStatements_FetchWrongInstance() throws MergerException {
         final AccumuloRyaStatementStore accumuloRyaStatementStore = createAccumuloRyaStatementStore();

         for (final RyaStatement ryaStatement : RYA_STATEMENTS) {
             accumuloRyaStatementStore.addStatement(ryaStatement);
         }

         final Configuration config = accumuloInstanceDriver.getDao().getConf();

         config.set(ConfigUtils.CLOUDBASE_INSTANCE, "wrong instance");

         accumuloRyaStatementStore.fetchStatements();
     }

     @Test
     public void testAddStatement() throws MergerException {
         final AccumuloRyaStatementStore accumuloRyaStatementStore = createAccumuloRyaStatementStore();

         for (final RyaStatement ryaStatement : RYA_STATEMENTS) {
             accumuloRyaStatementStore.addStatement(ryaStatement);
         }
     }

     @Test (expected = AddStatementException.class)
     public void testAddStatement_AddNull() throws Exception {
         final AccumuloRyaStatementStore accumuloRyaStatementStore = createAccumuloRyaStatementStore();

         accumuloRyaStatementStore.addStatement(null);
     }

     @Test
     public void testAddRemoveAddStatement() throws Exception {
         final AccumuloRyaStatementStore accumuloRyaStatementStore = createAccumuloRyaStatementStore();
         final RyaStatement stmnt = RYA_STATEMENTS.get(0);
         accumuloRyaStatementStore.addStatement(stmnt);
         assertTrue(accumuloRyaStatementStore.containsStatement(stmnt));
         assertEquals(1, count(accumuloRyaStatementStore));

         accumuloRyaStatementStore.removeStatement(stmnt);
         assertFalse(accumuloRyaStatementStore.containsStatement(stmnt));
         assertEquals(0, count(accumuloRyaStatementStore));

         accumuloRyaStatementStore.addStatement(stmnt);
         assertTrue(accumuloRyaStatementStore.containsStatement(stmnt));
         assertEquals(1, count(accumuloRyaStatementStore));
     }

     private int count(final RyaStatementStore store) throws FetchStatementException {
         final Iterator<RyaStatement> statements = store.fetchStatements();
         int count = 0;
         while(statements.hasNext()) {
             final RyaStatement statement = statements.next();
             System.out.println(statement.getObject().getData() + "     " + statement.getTimestamp());
             count++;
         }
         return count;
     }

     @Test
     public void testRemoveStatement() throws MergerException {
         final AccumuloRyaStatementStore accumuloRyaStatementStore = createAccumuloRyaStatementStore();

         // Add one then remove it right away
         for (final RyaStatement ryaStatement : RYA_STATEMENTS) {
             accumuloRyaStatementStore.addStatement(ryaStatement);
             accumuloRyaStatementStore.removeStatement(ryaStatement);
             assertTrue(isStatementStoreEmpty(accumuloRyaStatementStore));
         }

         // Add all then remove all
         for (final RyaStatement ryaStatement : RYA_STATEMENTS) {
             accumuloRyaStatementStore.addStatement(ryaStatement);
         }
         for (final RyaStatement ryaStatement : RYA_STATEMENTS) {
             accumuloRyaStatementStore.removeStatement(ryaStatement);
         }
         assertTrue(isStatementStoreEmpty(accumuloRyaStatementStore));

         // Add all then remove all in reverse order
         for (final RyaStatement ryaStatement : RYA_STATEMENTS) {
             accumuloRyaStatementStore.addStatement(ryaStatement);
         }
         final ImmutableList<RyaStatement> reverseList = RYA_STATEMENTS.reverse();
         for (final RyaStatement ryaStatement : reverseList) {
             accumuloRyaStatementStore.removeStatement(ryaStatement);
         }
         assertTrue(isStatementStoreEmpty(accumuloRyaStatementStore));

         // Add all then remove one from middle follow by another before and
         // after the first removed one
         for (final RyaStatement ryaStatement : RYA_STATEMENTS) {
             accumuloRyaStatementStore.addStatement(ryaStatement);
         }

         final RyaStatement firstToRemove = RYA_STATEMENTS.get(2);
         final RyaStatement before = RYA_STATEMENTS.get(1);
         final RyaStatement after = RYA_STATEMENTS.get(3);

         accumuloRyaStatementStore.removeStatement(firstToRemove);
         accumuloRyaStatementStore.removeStatement(before);
         accumuloRyaStatementStore.removeStatement(after);
     }

     @Test (expected = RemoveStatementException.class)
     public void testRemoveStatement_RemoveNull() throws MergerException {
         final AccumuloRyaStatementStore accumuloRyaStatementStore = createAccumuloRyaStatementStore();

         for (final RyaStatement ryaStatement : RYA_STATEMENTS) {
             accumuloRyaStatementStore.addStatement(ryaStatement);
         }

         accumuloRyaStatementStore.removeStatement(null);
     }

     @Test
     public void testRemoveStatement_RemoveStatementNotFound() throws MergerException {
         final AccumuloRyaStatementStore accumuloRyaStatementStore = createAccumuloRyaStatementStore();

         for (final RyaStatement ryaStatement : RYA_STATEMENTS) {
             accumuloRyaStatementStore.addStatement(ryaStatement);
         }

         final RyaStatement notFoundStatement = TestUtils.createRyaStatement("Statement", "not found", "here", DATE);
         accumuloRyaStatementStore.removeStatement(notFoundStatement);
     }

     @Test
     public void testUpdateStatement() throws MergerException {
         final AccumuloRyaStatementStore accumuloRyaStatementStore = createAccumuloRyaStatementStore();

         for (final RyaStatement ryaStatement : RYA_STATEMENTS) {
             accumuloRyaStatementStore.addStatement(ryaStatement);
         }

         final RyaStatement firstRyaStatement = RYA_STATEMENTS.get(0);
         final RyaStatement updatedRyaStatement = TestUtils.copyRyaStatement(firstRyaStatement);

         assertEquals(firstRyaStatement, updatedRyaStatement);

         final String subject = TestUtils.convertRyaIriToString(updatedRyaStatement.getSubject());
         final String predicate = TestUtils.convertRyaIriToString(updatedRyaStatement.getPredicate());
         updatedRyaStatement.setSubject(TestUtils.createRyaIri(subject + "_UPDATED"));
         updatedRyaStatement.setPredicate(TestUtils.createRyaIri(predicate + "_UPDATED"));

         accumuloRyaStatementStore.updateStatement(firstRyaStatement, updatedRyaStatement);

         final Iterator<RyaStatement> ryaStatementsIterator = accumuloRyaStatementStore.fetchStatements();
         int originalCount = 0;
         int updatedCount = 0;
         int totalCount = 0;
         while (ryaStatementsIterator.hasNext()) {
             final RyaStatement ryaStatement = ryaStatementsIterator.next();
             if (ryaStatement.equals(firstRyaStatement)) {
                 originalCount++;
             }
             if (ryaStatement.equals(updatedRyaStatement)) {
                 updatedCount++;
             }
             totalCount++;
         }

         assertEquals(0, originalCount);
         assertEquals(1, updatedCount);
         assertEquals(RYA_STATEMENTS.size(), totalCount);
     }

     @Test (expected = UpdateStatementException.class)
     public void testUpdateStatement_UpdateNull() throws MergerException {
         final AccumuloRyaStatementStore accumuloRyaStatementStore = createAccumuloRyaStatementStore();

         for (final RyaStatement ryaStatement : RYA_STATEMENTS) {
             accumuloRyaStatementStore.addStatement(ryaStatement);
         }

         final RyaStatement firstRyaStatement = RYA_STATEMENTS.get(0);
         final RyaStatement updatedRyaStatement = TestUtils.copyRyaStatement(firstRyaStatement);

         assertEquals(firstRyaStatement, updatedRyaStatement);

         final String subject = TestUtils.convertRyaIriToString(updatedRyaStatement.getSubject());
         final String predicate = TestUtils.convertRyaIriToString(updatedRyaStatement.getPredicate());
         updatedRyaStatement.setSubject(TestUtils.createRyaIri(subject + "_UPDATED"));
         updatedRyaStatement.setPredicate(TestUtils.createRyaIri(predicate + "_UPDATED"));

         accumuloRyaStatementStore.updateStatement(firstRyaStatement, null);
     }

     @Test
     public void testUpdateStatement_OriginalNotFound() throws MergerException {
         final AccumuloRyaStatementStore accumuloRyaStatementStore = createAccumuloRyaStatementStore();

         for (final RyaStatement ryaStatement : RYA_STATEMENTS) {
             accumuloRyaStatementStore.addStatement(ryaStatement);
         }

         final RyaStatement notFoundStatement = TestUtils.createRyaStatement("Statement", "not found", "here", DATE);
         final RyaStatement updatedRyaStatement = TestUtils.copyRyaStatement(notFoundStatement);

         assertEquals(notFoundStatement, updatedRyaStatement);

         final String subject = TestUtils.convertRyaIriToString(updatedRyaStatement.getSubject());
         final String predicate = TestUtils.convertRyaIriToString(updatedRyaStatement.getPredicate());
         updatedRyaStatement.setSubject(TestUtils.createRyaIri(subject + "_UPDATED"));
         updatedRyaStatement.setPredicate(TestUtils.createRyaIri(predicate + "_UPDATED"));

         accumuloRyaStatementStore.updateStatement(notFoundStatement, updatedRyaStatement);

         final Iterator<RyaStatement> ryaStatementsIterator = accumuloRyaStatementStore.fetchStatements();
         int originalCount = 0;
         int updatedCount = 0;
         int totalCount = 0;
         while (ryaStatementsIterator.hasNext()) {
             final RyaStatement ryaStatement = ryaStatementsIterator.next();
             if (ryaStatement.equals(notFoundStatement)) {
                 originalCount++;
             }
             if (ryaStatement.equals(updatedRyaStatement)) {
                 updatedCount++;
             }
             totalCount++;
         }

         assertEquals(0, originalCount);
         assertEquals(1, updatedCount);
         assertEquals(RYA_STATEMENTS.size() + 1, totalCount);
     }


     @After
     public void shutdownMiniResources() {
         if(accumuloInstanceDriver != null) {
             try {
                 log.info("Shutting down the Mini Accumulo being used as a Rya store.");
                 accumuloInstanceDriver.tearDown();
                 log.info("Mini Accumulo being used as a Rya store shut down.");
             } catch(final Exception e) {
                 log.error("Could not shut down the Mini Accumulo.", e);
             }
         }
     }

     private static boolean isStatementStoreEmpty(final AccumuloRyaStatementStore accumuloRyaStatementStore) throws MergerException {
         final Iterator<RyaStatement> iterator = accumuloRyaStatementStore.fetchStatements();
         return !iterator.hasNext();
     }

     /**
      * Setup a Accumulo instance driver to run the test. Establishes the
      * connector to the Accumulo instance.
      * @return an {@link AccumuloInstanceDriver}.
      * @throws Exception
      */
     private static AccumuloInstanceDriver startAccumuloInstanceDriver() throws Exception {
         final AccumuloInstanceDriver accumuloInstanceDriver = new AccumuloInstanceDriver("Test Driver", INSTANCE_TYPE, true, false, true, USER_NAME, PASSWORD, INSTANCE_NAME, RYA_TABLE_PREFIX, AUTHS, ZOOKEEPERS);
         accumuloInstanceDriver.setUp();

         return accumuloInstanceDriver;
     }

     private static AccumuloMergeConfiguration createAccumuloMergeConfiguration() {
         final AccumuloMergeConfiguration accumuloMergeConfiguration = mock(AccumuloMergeConfiguration.class);

         when(accumuloMergeConfiguration.getParentRyaInstanceName()).thenReturn(INSTANCE_NAME);
         when(accumuloMergeConfiguration.getParentUsername()).thenReturn(USER_NAME);
         when(accumuloMergeConfiguration.getParentPassword()).thenReturn(PASSWORD);
         when(accumuloMergeConfiguration.getParentInstanceType()).thenReturn(INSTANCE_TYPE);
         when(accumuloMergeConfiguration.getParentTablePrefix()).thenReturn(RYA_TABLE_PREFIX);
         when(accumuloMergeConfiguration.getParentAuths()).thenReturn(AUTHS);

         // Other
         when(accumuloMergeConfiguration.getMergePolicy()).thenReturn(MergePolicy.TIMESTAMP);

         return accumuloMergeConfiguration;
     }

     private static AccumuloRyaStatementStore createAccumuloRyaStatementStore() throws MergerException {
         final AccumuloMergeConfiguration accumuloMergeConfiguration = createAccumuloMergeConfiguration();
         return createAccumuloRyaStatementStore(accumuloMergeConfiguration);
     }

     private static AccumuloRyaStatementStore createAccumuloRyaStatementStore(final AccumuloMergeConfiguration accumuloMergeConfiguration) throws MergerException {
         final String instance = accumuloMergeConfiguration.getParentRyaInstanceName();
         final String tablePrefix = accumuloMergeConfiguration.getParentTablePrefix();

         return new AccumuloRyaStatementStore(accumuloInstanceDriver.getDao(), tablePrefix, instance);
     }
 }
	/*
	* 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.rya.export.accumulo;

	import static org.junit.Assert.assertEquals;
	import static org.junit.Assert.assertFalse;
	import static org.junit.Assert.assertTrue;
	import static org.mockito.Mockito.mock;
	import static org.mockito.Mockito.when;

	import java.util.Date;
	import java.util.Iterator;

	import org.apache.hadoop.conf.Configuration;
	import org.apache.log4j.LogManager;
	import org.apache.log4j.Logger;
	import org.apache.rya.api.domain.RyaStatement;
	import org.apache.rya.export.InstanceType;
	import org.apache.rya.export.MergePolicy;
	import org.apache.rya.export.accumulo.util.AccumuloInstanceDriver;
	import org.apache.rya.export.api.MergerException;
	import org.apache.rya.export.api.conf.AccumuloMergeConfiguration;
	import org.apache.rya.export.api.store.AddStatementException;
	import org.apache.rya.export.api.store.FetchStatementException;
	import org.apache.rya.export.api.store.RemoveStatementException;
	import org.apache.rya.export.api.store.RyaStatementStore;
	import org.apache.rya.export.api.store.UpdateStatementException;
	import org.apache.rya.indexing.accumulo.ConfigUtils;
	import org.junit.After;
	import org.junit.AfterClass;
	import org.junit.Before;
	import org.junit.BeforeClass;
	import org.junit.Test;

	import com.google.common.collect.ImmutableList;

	/**
	* Tests the methods of {@link AccumuloRyaStatementStore}.
	*/
	public class AccumuloRyaStatementStoreTest {
	private static final Logger log = LogManager.getLogger(AccumuloRyaStatementStoreTest.class);
	private static final InstanceType INSTANCE_TYPE = InstanceType.MOCK;

	private static final boolean IS_MOCK = INSTANCE_TYPE == InstanceType.MOCK;
	private static final String USER_NAME = IS_MOCK ? "test_user" : AccumuloInstanceDriver.ROOT_USER_NAME;
	private static final String PASSWORD = "password";
	private static final String INSTANCE_NAME = "test_instance";
	private static final String AUTHS = "test_auth";
	private static final String RYA_TABLE_PREFIX = "test_";
	private static final String ZOOKEEPERS = "localhost";

	// Rya data store and connections.
	private static AccumuloInstanceDriver accumuloInstanceDriver = null;

	private static final Date DATE = new Date();

	private static final ImmutableList<RyaStatement> RYA_STATEMENTS = ImmutableList.of(
	TestUtils.createRyaStatement("Adam", "analyzes", "apple", DATE),
	TestUtils.createRyaStatement("Bob", "bites", "burger", DATE),
	TestUtils.createRyaStatement("Charlie", "checks", "chores", DATE),
	TestUtils.createRyaStatement("Debbie", "drives", "deal", DATE),
	TestUtils.createRyaStatement("Emma", "eats", "everything", DATE)
	);

	@BeforeClass
	public static void setupResources() throws Exception {
	// Initialize the Accumulo instance that will be used to store Triples and get a connection to it.
	accumuloInstanceDriver = startAccumuloInstanceDriver();
	}

	@Before
	public void setUpPerTest() throws Exception {
	accumuloInstanceDriver.setUpTables();
	accumuloInstanceDriver.setUpDao();
	accumuloInstanceDriver.setUpConfig();
	}

	@After
	public void tearDownPerTest() throws Exception {
	log.info("tearDownPerTest(): tearing down now.");
	accumuloInstanceDriver.tearDownTables();
	accumuloInstanceDriver.tearDownDao();
	}

	@AfterClass
	public static void tearDownPerClass() throws Exception {
	log.info("tearDownPerClass(): tearing down now.");
	accumuloInstanceDriver.tearDown();
	}

	@Test
	public void testFetchStatements() throws MergerException {
	final AccumuloRyaStatementStore accumuloRyaStatementStore = createAccumuloRyaStatementStore();

	for (final RyaStatement ryaStatement : RYA_STATEMENTS) {
	accumuloRyaStatementStore.addStatement(ryaStatement);
	}

	accumuloRyaStatementStore.fetchStatements();
	}

	@Test
	public void testRemoveAddStatements() throws MergerException {
	final AccumuloRyaStatementStore accumuloRyaStatementStore = createAccumuloRyaStatementStore();

	for (final RyaStatement ryaStatement : RYA_STATEMENTS) {
	accumuloRyaStatementStore.addStatement(ryaStatement);
	}
	final RyaStatement stmnt = RYA_STATEMENTS.get(0);

	assertTrue(accumuloRyaStatementStore.containsStatement(stmnt));
	accumuloRyaStatementStore.removeStatement(stmnt);
	assertFalse(accumuloRyaStatementStore.containsStatement(stmnt));

	accumuloRyaStatementStore.addStatement(stmnt);
	assertTrue(accumuloRyaStatementStore.containsStatement(stmnt));
	}

	@Test (expected = FetchStatementException.class)
	public void testFetchStatements_FetchWrongInstance() throws MergerException {
	final AccumuloRyaStatementStore accumuloRyaStatementStore = createAccumuloRyaStatementStore();

	for (final RyaStatement ryaStatement : RYA_STATEMENTS) {
	accumuloRyaStatementStore.addStatement(ryaStatement);
	}

	final Configuration config = accumuloInstanceDriver.getDao().getConf();

	config.set(ConfigUtils.CLOUDBASE_INSTANCE, "wrong instance");

	accumuloRyaStatementStore.fetchStatements();
	}

	@Test
	public void testAddStatement() throws MergerException {
	final AccumuloRyaStatementStore accumuloRyaStatementStore = createAccumuloRyaStatementStore();

	for (final RyaStatement ryaStatement : RYA_STATEMENTS) {
	accumuloRyaStatementStore.addStatement(ryaStatement);
	}
	}

	@Test (expected = AddStatementException.class)
	public void testAddStatement_AddNull() throws Exception {
	final AccumuloRyaStatementStore accumuloRyaStatementStore = createAccumuloRyaStatementStore();

	accumuloRyaStatementStore.addStatement(null);
	}

	@Test
	public void testAddRemoveAddStatement() throws Exception {
	final AccumuloRyaStatementStore accumuloRyaStatementStore = createAccumuloRyaStatementStore();
	final RyaStatement stmnt = RYA_STATEMENTS.get(0);
	accumuloRyaStatementStore.addStatement(stmnt);
	assertTrue(accumuloRyaStatementStore.containsStatement(stmnt));
	assertEquals(1, count(accumuloRyaStatementStore));

	accumuloRyaStatementStore.removeStatement(stmnt);
	assertFalse(accumuloRyaStatementStore.containsStatement(stmnt));
	assertEquals(0, count(accumuloRyaStatementStore));

	accumuloRyaStatementStore.addStatement(stmnt);
	assertTrue(accumuloRyaStatementStore.containsStatement(stmnt));
	assertEquals(1, count(accumuloRyaStatementStore));
	}

	private int count(final RyaStatementStore store) throws FetchStatementException {
	final Iterator<RyaStatement> statements = store.fetchStatements();
	int count = 0;
	while(statements.hasNext()) {
	final RyaStatement statement = statements.next();
	System.out.println(statement.getObject().getData() + " " + statement.getTimestamp());
	count++;
	}
	return count;
	}

	@Test
	public void testRemoveStatement() throws MergerException {
	final AccumuloRyaStatementStore accumuloRyaStatementStore = createAccumuloRyaStatementStore();

	// Add one then remove it right away
	for (final RyaStatement ryaStatement : RYA_STATEMENTS) {
	accumuloRyaStatementStore.addStatement(ryaStatement);
	accumuloRyaStatementStore.removeStatement(ryaStatement);
	assertTrue(isStatementStoreEmpty(accumuloRyaStatementStore));
	}

	// Add all then remove all
	for (final RyaStatement ryaStatement : RYA_STATEMENTS) {
	accumuloRyaStatementStore.addStatement(ryaStatement);
	}
	for (final RyaStatement ryaStatement : RYA_STATEMENTS) {
	accumuloRyaStatementStore.removeStatement(ryaStatement);
	}
	assertTrue(isStatementStoreEmpty(accumuloRyaStatementStore));

	// Add all then remove all in reverse order
	for (final RyaStatement ryaStatement : RYA_STATEMENTS) {
	accumuloRyaStatementStore.addStatement(ryaStatement);
	}
	final ImmutableList<RyaStatement> reverseList = RYA_STATEMENTS.reverse();
	for (final RyaStatement ryaStatement : reverseList) {
	accumuloRyaStatementStore.removeStatement(ryaStatement);
	}
	assertTrue(isStatementStoreEmpty(accumuloRyaStatementStore));

	// Add all then remove one from middle follow by another before and
	// after the first removed one
	for (final RyaStatement ryaStatement : RYA_STATEMENTS) {
	accumuloRyaStatementStore.addStatement(ryaStatement);
	}

	final RyaStatement firstToRemove = RYA_STATEMENTS.get(2);
	final RyaStatement before = RYA_STATEMENTS.get(1);
	final RyaStatement after = RYA_STATEMENTS.get(3);

	accumuloRyaStatementStore.removeStatement(firstToRemove);
	accumuloRyaStatementStore.removeStatement(before);
	accumuloRyaStatementStore.removeStatement(after);
	}

	@Test (expected = RemoveStatementException.class)
	public void testRemoveStatement_RemoveNull() throws MergerException {
	final AccumuloRyaStatementStore accumuloRyaStatementStore = createAccumuloRyaStatementStore();

	for (final RyaStatement ryaStatement : RYA_STATEMENTS) {
	accumuloRyaStatementStore.addStatement(ryaStatement);
	}

	accumuloRyaStatementStore.removeStatement(null);
	}

	@Test
	public void testRemoveStatement_RemoveStatementNotFound() throws MergerException {
	final AccumuloRyaStatementStore accumuloRyaStatementStore = createAccumuloRyaStatementStore();

	for (final RyaStatement ryaStatement : RYA_STATEMENTS) {
	accumuloRyaStatementStore.addStatement(ryaStatement);
	}

	final RyaStatement notFoundStatement = TestUtils.createRyaStatement("Statement", "not found", "here", DATE);
	accumuloRyaStatementStore.removeStatement(notFoundStatement);
	}

	@Test
	public void testUpdateStatement() throws MergerException {
	final AccumuloRyaStatementStore accumuloRyaStatementStore = createAccumuloRyaStatementStore();

	for (final RyaStatement ryaStatement : RYA_STATEMENTS) {
	accumuloRyaStatementStore.addStatement(ryaStatement);
	}

	final RyaStatement firstRyaStatement = RYA_STATEMENTS.get(0);
	final RyaStatement updatedRyaStatement = TestUtils.copyRyaStatement(firstRyaStatement);

	assertEquals(firstRyaStatement, updatedRyaStatement);

	final String subject = TestUtils.convertRyaIriToString(updatedRyaStatement.getSubject());
	final String predicate = TestUtils.convertRyaIriToString(updatedRyaStatement.getPredicate());
	updatedRyaStatement.setSubject(TestUtils.createRyaIri(subject + "_UPDATED"));
	updatedRyaStatement.setPredicate(TestUtils.createRyaIri(predicate + "_UPDATED"));

	accumuloRyaStatementStore.updateStatement(firstRyaStatement, updatedRyaStatement);

	final Iterator<RyaStatement> ryaStatementsIterator = accumuloRyaStatementStore.fetchStatements();
	int originalCount = 0;
	int updatedCount = 0;
	int totalCount = 0;
	while (ryaStatementsIterator.hasNext()) {
	final RyaStatement ryaStatement = ryaStatementsIterator.next();
	if (ryaStatement.equals(firstRyaStatement)) {
	originalCount++;
	}
	if (ryaStatement.equals(updatedRyaStatement)) {
	updatedCount++;
	}
	totalCount++;
	}

	assertEquals(0, originalCount);
	assertEquals(1, updatedCount);
	assertEquals(RYA_STATEMENTS.size(), totalCount);
	}

	@Test (expected = UpdateStatementException.class)
	public void testUpdateStatement_UpdateNull() throws MergerException {
	final AccumuloRyaStatementStore accumuloRyaStatementStore = createAccumuloRyaStatementStore();

	for (final RyaStatement ryaStatement : RYA_STATEMENTS) {
	accumuloRyaStatementStore.addStatement(ryaStatement);
	}

	final RyaStatement firstRyaStatement = RYA_STATEMENTS.get(0);
	final RyaStatement updatedRyaStatement = TestUtils.copyRyaStatement(firstRyaStatement);

	assertEquals(firstRyaStatement, updatedRyaStatement);

	final String subject = TestUtils.convertRyaIriToString(updatedRyaStatement.getSubject());
	final String predicate = TestUtils.convertRyaIriToString(updatedRyaStatement.getPredicate());
	updatedRyaStatement.setSubject(TestUtils.createRyaIri(subject + "_UPDATED"));
	updatedRyaStatement.setPredicate(TestUtils.createRyaIri(predicate + "_UPDATED"));

	accumuloRyaStatementStore.updateStatement(firstRyaStatement, null);
	}

	@Test
	public void testUpdateStatement_OriginalNotFound() throws MergerException {
	final AccumuloRyaStatementStore accumuloRyaStatementStore = createAccumuloRyaStatementStore();

	for (final RyaStatement ryaStatement : RYA_STATEMENTS) {
	accumuloRyaStatementStore.addStatement(ryaStatement);
	}

	final RyaStatement notFoundStatement = TestUtils.createRyaStatement("Statement", "not found", "here", DATE);
	final RyaStatement updatedRyaStatement = TestUtils.copyRyaStatement(notFoundStatement);

	assertEquals(notFoundStatement, updatedRyaStatement);

	final String subject = TestUtils.convertRyaIriToString(updatedRyaStatement.getSubject());
	final String predicate = TestUtils.convertRyaIriToString(updatedRyaStatement.getPredicate());
	updatedRyaStatement.setSubject(TestUtils.createRyaIri(subject + "_UPDATED"));
	updatedRyaStatement.setPredicate(TestUtils.createRyaIri(predicate + "_UPDATED"));

	accumuloRyaStatementStore.updateStatement(notFoundStatement, updatedRyaStatement);

	final Iterator<RyaStatement> ryaStatementsIterator = accumuloRyaStatementStore.fetchStatements();
	int originalCount = 0;
	int updatedCount = 0;
	int totalCount = 0;
	while (ryaStatementsIterator.hasNext()) {
	final RyaStatement ryaStatement = ryaStatementsIterator.next();
	if (ryaStatement.equals(notFoundStatement)) {
	originalCount++;
	}
	if (ryaStatement.equals(updatedRyaStatement)) {
	updatedCount++;
	}
	totalCount++;
	}

	assertEquals(0, originalCount);
	assertEquals(1, updatedCount);
	assertEquals(RYA_STATEMENTS.size() + 1, totalCount);
	}



	@After
	public void shutdownMiniResources() {
	if(accumuloInstanceDriver != null) {
	try {
	log.info("Shutting down the Mini Accumulo being used as a Rya store.");
	accumuloInstanceDriver.tearDown();
	log.info("Mini Accumulo being used as a Rya store shut down.");
	} catch(final Exception e) {
	log.error("Could not shut down the Mini Accumulo.", e);
	}
	}
	}

	private static boolean isStatementStoreEmpty(final AccumuloRyaStatementStore accumuloRyaStatementStore) throws MergerException {
	final Iterator<RyaStatement> iterator = accumuloRyaStatementStore.fetchStatements();
	return !iterator.hasNext();
	}

	/**
	* Setup a Accumulo instance driver to run the test. Establishes the
	* connector to the Accumulo instance.
	* @return an {@link AccumuloInstanceDriver}.
	* @throws Exception
	*/
	private static AccumuloInstanceDriver startAccumuloInstanceDriver() throws Exception {
	final AccumuloInstanceDriver accumuloInstanceDriver = new AccumuloInstanceDriver("Test Driver", INSTANCE_TYPE, true, false, true, USER_NAME, PASSWORD, INSTANCE_NAME, RYA_TABLE_PREFIX, AUTHS, ZOOKEEPERS);
	accumuloInstanceDriver.setUp();

	return accumuloInstanceDriver;
	}

	private static AccumuloMergeConfiguration createAccumuloMergeConfiguration() {
	final AccumuloMergeConfiguration accumuloMergeConfiguration = mock(AccumuloMergeConfiguration.class);

	when(accumuloMergeConfiguration.getParentRyaInstanceName()).thenReturn(INSTANCE_NAME);
	when(accumuloMergeConfiguration.getParentUsername()).thenReturn(USER_NAME);
	when(accumuloMergeConfiguration.getParentPassword()).thenReturn(PASSWORD);
	when(accumuloMergeConfiguration.getParentInstanceType()).thenReturn(INSTANCE_TYPE);
	when(accumuloMergeConfiguration.getParentTablePrefix()).thenReturn(RYA_TABLE_PREFIX);
	when(accumuloMergeConfiguration.getParentAuths()).thenReturn(AUTHS);

	// Other
	when(accumuloMergeConfiguration.getMergePolicy()).thenReturn(MergePolicy.TIMESTAMP);

	return accumuloMergeConfiguration;
	}

	private static AccumuloRyaStatementStore createAccumuloRyaStatementStore() throws MergerException {
	final AccumuloMergeConfiguration accumuloMergeConfiguration = createAccumuloMergeConfiguration();
	return createAccumuloRyaStatementStore(accumuloMergeConfiguration);
	}

	private static AccumuloRyaStatementStore createAccumuloRyaStatementStore(final AccumuloMergeConfiguration accumuloMergeConfiguration) throws MergerException {
	final String instance = accumuloMergeConfiguration.getParentRyaInstanceName();
	final String tablePrefix = accumuloMergeConfiguration.getParentTablePrefix();

	return new AccumuloRyaStatementStore(accumuloInstanceDriver.getDao(), tablePrefix, instance);
	}
	}