X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Ftest%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2FTransactionProxyTest.java;h=29a5b09c5c8e33fcb8892054564d4264dd2a711d;hb=4e594bb578f687fb24e242e90b918d91fb59c3ac;hp=ac2c07964179192f86b3123f5a9f8a24a8cd2e9d;hpb=edfebf2ecb683a0b16286977c43efa5a67b6744a;p=controller.git diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/TransactionProxyTest.java b/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/TransactionProxyTest.java index ac2c079641..29a5b09c5c 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/TransactionProxyTest.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/TransactionProxyTest.java @@ -9,8 +9,8 @@ import static org.mockito.Matchers.anyString; import static org.mockito.Matchers.eq; import static org.mockito.Matchers.isA; import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; -import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.opendaylight.controller.cluster.datastore.TransactionProxy.TransactionType.READ_ONLY; import static org.opendaylight.controller.cluster.datastore.TransactionProxy.TransactionType.READ_WRITE; @@ -21,11 +21,14 @@ import akka.actor.ActorSystem; import akka.actor.Props; import akka.dispatch.Futures; import com.google.common.base.Optional; +import com.google.common.collect.Sets; import com.google.common.util.concurrent.CheckedFuture; import com.google.common.util.concurrent.FutureCallback; import com.google.common.util.concurrent.Uninterruptibles; +import java.util.Collection; import java.util.List; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; import org.junit.Assert; @@ -33,6 +36,8 @@ import org.junit.Test; import org.mockito.InOrder; import org.mockito.Mockito; import org.opendaylight.controller.cluster.datastore.TransactionProxy.TransactionType; +import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderException; +import org.opendaylight.controller.cluster.datastore.exceptions.NotInitializedException; import org.opendaylight.controller.cluster.datastore.exceptions.PrimaryNotFoundException; import org.opendaylight.controller.cluster.datastore.exceptions.TimeoutException; import org.opendaylight.controller.cluster.datastore.messages.BatchedModifications; @@ -44,14 +49,18 @@ import org.opendaylight.controller.cluster.datastore.modification.MergeModificat import org.opendaylight.controller.cluster.datastore.modification.WriteModification; import org.opendaylight.controller.cluster.datastore.shardstrategy.DefaultShardStrategy; import org.opendaylight.controller.cluster.datastore.utils.DoNothingActor; +import org.opendaylight.controller.cluster.datastore.utils.NormalizedNodeAggregatorTest; import org.opendaylight.controller.md.cluster.datastore.model.CarsModel; +import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper; import org.opendaylight.controller.md.cluster.datastore.model.TestModel; import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException; import org.opendaylight.controller.protobuff.messages.transaction.ShardTransactionMessages.CreateTransactionReply; import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort; import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier; +import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode; import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode; import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes; +import org.opendaylight.yangtools.yang.model.api.SchemaContext; import scala.concurrent.Await; import scala.concurrent.Future; import scala.concurrent.Promise; @@ -162,34 +171,6 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { testReadWithExceptionOnInitialCreateTransaction(new TestException()); } - @Test(expected = TestException.class) - public void testReadWithPriorRecordingOperationFailure() throws Throwable { - doReturn(dataStoreContextBuilder.shardBatchedModificationCount(2).build()). - when(mockActorContext).getDatastoreContext(); - - ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE); - - NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); - - expectFailedBatchedModifications(actorRef); - - doReturn(readSerializedDataReply(null)).when(mockActorContext).executeOperationAsync( - eq(actorSelection(actorRef)), eqSerializedReadData()); - - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); - - transactionProxy.write(TestModel.TEST_PATH, nodeToWrite); - - transactionProxy.delete(TestModel.TEST_PATH); - - try { - propagateReadFailedExceptionCause(transactionProxy.read(TestModel.TEST_PATH)); - } finally { - verify(mockActorContext, times(0)).executeOperationAsync( - eq(actorSelection(actorRef)), eqSerializedReadData()); - } - } - @Test public void testReadWithPriorRecordingOperationSuccessful() throws Throwable { ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE); @@ -299,35 +280,6 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { propagateReadFailedExceptionCause(transactionProxy.exists(TestModel.TEST_PATH)); } - @Test(expected = TestException.class) - public void testExistsWithPriorRecordingOperationFailure() throws Throwable { - doReturn(dataStoreContextBuilder.shardBatchedModificationCount(2).build()). - when(mockActorContext).getDatastoreContext(); - - ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE); - - NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); - - expectFailedBatchedModifications(actorRef); - - doReturn(dataExistsSerializedReply(false)).when(mockActorContext).executeOperationAsync( - eq(actorSelection(actorRef)), eqSerializedDataExists()); - - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, - READ_WRITE); - - transactionProxy.write(TestModel.TEST_PATH, nodeToWrite); - - transactionProxy.delete(TestModel.TEST_PATH); - - try { - propagateReadFailedExceptionCause(transactionProxy.exists(TestModel.TEST_PATH)); - } finally { - verify(mockActorContext, times(0)).executeOperationAsync( - eq(actorSelection(actorRef)), eqSerializedDataExists()); - } - } - @Test public void testExistsWithPriorRecordingOperationSuccessful() throws Throwable { ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE); @@ -659,11 +611,8 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { verifyCohortFutures(proxy, TestException.class); } - @Test - public void testReadyWithInitialCreateTransactionFailure() throws Exception { - - doReturn(Futures.failed(new PrimaryNotFoundException("mock"))).when( - mockActorContext).findPrimaryShardAsync(anyString()); + private void testWriteOnlyTxWithFindPrimaryShardFailure(Exception toThrow) throws Exception { + doReturn(Futures.failed(toThrow)).when(mockActorContext).findPrimaryShardAsync(anyString()); TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); @@ -681,7 +630,22 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready; - verifyCohortFutures(proxy, PrimaryNotFoundException.class); + verifyCohortFutures(proxy, toThrow.getClass()); + } + + @Test + public void testWriteOnlyTxWithPrimaryNotFoundException() throws Exception { + testWriteOnlyTxWithFindPrimaryShardFailure(new PrimaryNotFoundException("mock")); + } + + @Test + public void testWriteOnlyTxWithNotInitializedException() throws Exception { + testWriteOnlyTxWithFindPrimaryShardFailure(new NotInitializedException("mock")); + } + + @Test + public void testWriteOnlyTxWithNoShardLeaderException() throws Exception { + testWriteOnlyTxWithFindPrimaryShardFailure(new NoShardLeaderException("mock")); } @Test @@ -1397,4 +1361,79 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), BatchedModificationsReply.class, BatchedModificationsReply.class, BatchedModificationsReply.class); } + + @Test + public void testReadRoot() throws ReadFailedException, InterruptedException, ExecutionException, java.util.concurrent.TimeoutException { + + SchemaContext schemaContext = SchemaContextHelper.full(); + Configuration configuration = mock(Configuration.class); + doReturn(configuration).when(mockActorContext).getConfiguration(); + doReturn(schemaContext).when(mockActorContext).getSchemaContext(); + doReturn(Sets.newHashSet("test", "cars")).when(configuration).getAllShardNames(); + + NormalizedNode expectedNode1 = ImmutableNodes.containerNode(TestModel.TEST_QNAME); + NormalizedNode expectedNode2 = ImmutableNodes.containerNode(CarsModel.CARS_QNAME); + + setUpReadData("test", NormalizedNodeAggregatorTest.getRootNode(expectedNode1, schemaContext)); + setUpReadData("cars", NormalizedNodeAggregatorTest.getRootNode(expectedNode2, schemaContext)); + + doReturn(memberName).when(mockActorContext).getCurrentMemberName(); + + doReturn(10).when(mockActorContext).getTransactionOutstandingOperationLimit(); + + doReturn(getSystem().dispatchers().defaultGlobalDispatcher()).when(mockActorContext).getClientDispatcher(); + + TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_ONLY); + + Optional> readOptional = transactionProxy.read( + YangInstanceIdentifier.builder().build()).get(5, TimeUnit.SECONDS); + + assertEquals("NormalizedNode isPresent", true, readOptional.isPresent()); + + NormalizedNode normalizedNode = readOptional.get(); + + assertTrue("Expect value to be a Collection", normalizedNode.getValue() instanceof Collection); + + Collection> collection = (Collection>) normalizedNode.getValue(); + + for(NormalizedNode node : collection){ + assertTrue("Expected " + node + " to be a ContainerNode", node instanceof ContainerNode); + } + + assertTrue("Child with QName = " + TestModel.TEST_QNAME + " not found", + NormalizedNodeAggregatorTest.findChildWithQName(collection, TestModel.TEST_QNAME) != null); + + assertEquals(expectedNode1, NormalizedNodeAggregatorTest.findChildWithQName(collection, TestModel.TEST_QNAME)); + + assertTrue("Child with QName = " + CarsModel.BASE_QNAME + " not found", + NormalizedNodeAggregatorTest.findChildWithQName(collection, CarsModel.BASE_QNAME) != null); + + assertEquals(expectedNode2, NormalizedNodeAggregatorTest.findChildWithQName(collection, CarsModel.BASE_QNAME)); + } + + + private void setUpReadData(String shardName, NormalizedNode expectedNode) { + ActorSystem actorSystem = getSystem(); + ActorRef shardActorRef = getSystem().actorOf(Props.create(DoNothingActor.class)); + + doReturn(getSystem().actorSelection(shardActorRef.path())). + when(mockActorContext).actorSelection(shardActorRef.path().toString()); + + doReturn(Futures.successful(getSystem().actorSelection(shardActorRef.path()))). + when(mockActorContext).findPrimaryShardAsync(eq(shardName)); + + doReturn(true).when(mockActorContext).isPathLocal(shardActorRef.path().toString()); + + ActorRef txActorRef = actorSystem.actorOf(Props.create(DoNothingActor.class)); + + doReturn(actorSystem.actorSelection(txActorRef.path())). + when(mockActorContext).actorSelection(txActorRef.path().toString()); + + doReturn(Futures.successful(createTransactionReply(txActorRef, DataStoreVersions.CURRENT_VERSION))).when(mockActorContext). + executeOperationAsync(eq(actorSystem.actorSelection(shardActorRef.path())), + eqCreateTransaction(memberName, TransactionType.READ_ONLY)); + + doReturn(readSerializedDataReply(expectedNode)).when(mockActorContext).executeOperationAsync( + eq(actorSelection(txActorRef)), eqSerializedReadData(YangInstanceIdentifier.builder().build())); + } }