X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Ftest%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2FTransactionProxyTest.java;h=7407897dfac504872e37d35c374397369a8c5576;hp=f2b849122a3c0fced5f8925b0ab6d7e80080152c;hb=0133fc851489ac7ea5f0ca6413175fc8b7fa485b;hpb=8600573a11d7a0fa1c03e6c3f014473096e6a97c 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 f2b849122a..7407897dfa 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 @@ -5,9 +5,15 @@ import akka.actor.ActorSelection; import akka.actor.ActorSystem; import akka.actor.Props; import akka.dispatch.Futures; +import akka.testkit.JavaTestKit; import com.google.common.base.Optional; +import com.google.common.collect.ImmutableMap; import com.google.common.util.concurrent.CheckedFuture; +import com.typesafe.config.Config; +import com.typesafe.config.ConfigFactory; +import org.junit.AfterClass; import org.junit.Before; +import org.junit.BeforeClass; import org.junit.Test; import org.mockito.ArgumentMatcher; import org.mockito.Mock; @@ -44,6 +50,7 @@ import org.opendaylight.yangtools.yang.model.api.SchemaContext; import scala.concurrent.Await; import scala.concurrent.Future; import scala.concurrent.duration.Duration; +import java.io.IOException; import java.util.List; import java.util.concurrent.TimeUnit; import static org.junit.Assert.assertEquals; @@ -54,7 +61,6 @@ import static org.mockito.Matchers.any; import static org.mockito.Matchers.anyString; import static org.mockito.Mockito.argThat; import static org.mockito.Mockito.doReturn; -import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.eq; import static org.mockito.Mockito.isA; import static org.mockito.Mockito.times; @@ -64,7 +70,7 @@ import static org.opendaylight.controller.cluster.datastore.TransactionProxy.Tra import static org.opendaylight.controller.cluster.datastore.TransactionProxy.TransactionType.WRITE_ONLY; @SuppressWarnings("resource") -public class TransactionProxyTest extends AbstractActorTest { +public class TransactionProxyTest { @SuppressWarnings("serial") static class TestException extends RuntimeException { @@ -74,6 +80,8 @@ public class TransactionProxyTest extends AbstractActorTest { CheckedFuture invoke(TransactionProxy proxy) throws Exception; } + private static ActorSystem system; + private final Configuration configuration = new MockConfiguration(); @Mock @@ -86,20 +94,44 @@ public class TransactionProxyTest extends AbstractActorTest { String memberName = "mock-member"; + @BeforeClass + public static void setUpClass() throws IOException { + + Config config = ConfigFactory.parseMap(ImmutableMap.builder(). + put("akka.actor.default-dispatcher.type", + "akka.testkit.CallingThreadDispatcherConfigurator").build()). + withFallback(ConfigFactory.load()); + system = ActorSystem.create("test", config); + } + + @AfterClass + public static void tearDownClass() throws IOException { + JavaTestKit.shutdownActorSystem(system); + system = null; + } + @Before public void setUp(){ MockitoAnnotations.initMocks(this); schemaContext = TestModel.createTestContext(); + DatastoreContext dataStoreContext = DatastoreContext.newBuilder().build(); + doReturn(getSystem()).when(mockActorContext).getActorSystem(); doReturn(memberName).when(mockActorContext).getCurrentMemberName(); doReturn(schemaContext).when(mockActorContext).getSchemaContext(); doReturn(mockClusterWrapper).when(mockActorContext).getClusterWrapper(); + doReturn(mockClusterWrapper).when(mockActorContext).getClusterWrapper(); + doReturn(dataStoreContext).when(mockActorContext).getDatastoreContext(); ShardStrategyFactory.setConfiguration(configuration); } + private ActorSystem getSystem() { + return system; + } + private CreateTransaction eqCreateTransaction(final String memberName, final TransactionType type) { ArgumentMatcher matcher = new ArgumentMatcher() { @@ -306,29 +338,36 @@ public class TransactionProxyTest extends AbstractActorTest { return getSystem().actorSelection(actorRef.path()); } - private CreateTransactionReply createTransactionReply(ActorRef actorRef){ + private CreateTransactionReply createTransactionReply(ActorRef actorRef, int transactionVersion){ return CreateTransactionReply.newBuilder() .setTransactionActorPath(actorRef.path().toString()) - .setTransactionId("txn-1").build(); + .setTransactionId("txn-1") + .setMessageVersion(transactionVersion) + .build(); } - private ActorRef setupActorContextWithInitialCreateTransaction(ActorSystem actorSystem, TransactionType type) { + private ActorRef setupActorContextWithInitialCreateTransaction(ActorSystem actorSystem, TransactionType type, int transactionVersion) { ActorRef actorRef = actorSystem.actorOf(Props.create(DoNothingActor.class)); doReturn(actorSystem.actorSelection(actorRef.path())). when(mockActorContext).actorSelection(actorRef.path().toString()); - doReturn(Optional.of(actorSystem.actorSelection(actorRef.path()))). - when(mockActorContext).findPrimaryShard(eq(DefaultShardStrategy.DEFAULT_SHARD)); + doReturn(Futures.successful(actorSystem.actorSelection(actorRef.path()))). + when(mockActorContext).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD)); - doReturn(createTransactionReply(actorRef)).when(mockActorContext). - executeOperation(eq(actorSystem.actorSelection(actorRef.path())), + doReturn(Futures.successful(createTransactionReply(actorRef, transactionVersion))).when(mockActorContext). + executeOperationAsync(eq(actorSystem.actorSelection(actorRef.path())), eqCreateTransaction(memberName, type)); - doReturn(false).when(mockActorContext).isLocalPath(actorRef.path().toString()); + doReturn(false).when(mockActorContext).isPathLocal(actorRef.path().toString()); return actorRef; } + private ActorRef setupActorContextWithInitialCreateTransaction(ActorSystem actorSystem, TransactionType type) { + return setupActorContextWithInitialCreateTransaction(actorSystem, type, CreateTransaction.CURRENT_VERSION); + } + + private void propagateReadFailedExceptionCause(CheckedFuture future) throws Throwable { @@ -372,7 +411,7 @@ public class TransactionProxyTest extends AbstractActorTest { ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY); doReturn(Futures.successful(new Object())).when(mockActorContext). - executeOperationAsync(any(ActorSelection.class), any()); + executeOperationAsync(eq(actorSelection(actorRef)), eqSerializedReadData()); TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_ONLY); @@ -385,7 +424,7 @@ public class TransactionProxyTest extends AbstractActorTest { ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY); doReturn(Futures.failed(new TestException())).when(mockActorContext). - executeOperationAsync(any(ActorSelection.class), any()); + executeOperationAsync(eq(actorSelection(actorRef)), eqSerializedReadData()); TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_ONLY); @@ -398,12 +437,14 @@ public class TransactionProxyTest extends AbstractActorTest { ActorRef actorRef = getSystem().actorOf(Props.create(DoNothingActor.class)); if (exToThrow instanceof PrimaryNotFoundException) { - doReturn(Optional.absent()).when(mockActorContext).findPrimaryShard(anyString()); + doReturn(Futures.failed(exToThrow)).when(mockActorContext).findPrimaryShardAsync(anyString()); } else { - doReturn(Optional.of(getSystem().actorSelection(actorRef.path()))). - when(mockActorContext).findPrimaryShard(anyString()); + doReturn(Futures.successful(getSystem().actorSelection(actorRef.path()))). + when(mockActorContext).findPrimaryShardAsync(anyString()); } - doThrow(exToThrow).when(mockActorContext).executeOperation(any(ActorSelection.class), any()); + + doReturn(Futures.failed(exToThrow)).when(mockActorContext).executeOperationAsync( + any(ActorSelection.class), any()); TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_ONLY); @@ -499,6 +540,24 @@ public class TransactionProxyTest extends AbstractActorTest { transactionProxy.read(TestModel.TEST_PATH); } + @Test(expected=IllegalArgumentException.class) + public void testInvalidCreateTransactionReply() throws Throwable { + ActorRef actorRef = getSystem().actorOf(Props.create(DoNothingActor.class)); + + doReturn(getSystem().actorSelection(actorRef.path())).when(mockActorContext). + actorSelection(actorRef.path().toString()); + + doReturn(Futures.successful(getSystem().actorSelection(actorRef.path()))). + when(mockActorContext).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD)); + + doReturn(Futures.successful(new Object())).when(mockActorContext).executeOperationAsync( + eq(getSystem().actorSelection(actorRef.path())), eqCreateTransaction(memberName, READ_ONLY)); + + TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_ONLY); + + propagateReadFailedExceptionCause(transactionProxy.read(TestModel.TEST_PATH)); + } + @Test public void testExists() throws Exception { ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY); @@ -536,7 +595,7 @@ public class TransactionProxyTest extends AbstractActorTest { ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY); doReturn(Futures.successful(new Object())).when(mockActorContext). - executeOperationAsync(any(ActorSelection.class), any()); + executeOperationAsync(eq(actorSelection(actorRef)), eqSerializedDataExists()); TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_ONLY); @@ -549,7 +608,7 @@ public class TransactionProxyTest extends AbstractActorTest { ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY); doReturn(Futures.failed(new TestException())).when(mockActorContext). - executeOperationAsync(any(ActorSelection.class), any()); + executeOperationAsync(eq(actorSelection(actorRef)), eqSerializedDataExists()); TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_ONLY); @@ -610,7 +669,7 @@ public class TransactionProxyTest extends AbstractActorTest { } @Test(expected=IllegalStateException.class) - public void testxistsPreConditionCheck() { + public void testExistsPreConditionCheck() { TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); @@ -693,8 +752,7 @@ public class TransactionProxyTest extends AbstractActorTest { doReturn(mergeSerializedDataReply()).when(mockActorContext).executeOperationAsync( eq(actorSelection(actorRef)), eqSerializedMergeData(nodeToWrite)); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, - WRITE_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); transactionProxy.merge(TestModel.TEST_PATH, nodeToWrite); @@ -749,7 +807,6 @@ public class TransactionProxyTest extends AbstractActorTest { } } - @SuppressWarnings("unchecked") @Test public void testReady() throws Exception { ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE); @@ -784,7 +841,46 @@ public class TransactionProxyTest extends AbstractActorTest { verifyCohortFutures(proxy, getSystem().actorSelection(actorRef.path())); } - @SuppressWarnings("unchecked") + @Test + public void testReadyForwardCompatibility() throws Exception { + ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE, 0); + + NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); + + doReturn(readSerializedDataReply(null)).when(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), eqSerializedReadData()); + + doReturn(writeSerializedDataReply()).when(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), eqSerializedWriteData(nodeToWrite)); + + doReturn(readySerializedTxReply(actorRef.path().toString())).when(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), isA(ReadyTransaction.SERIALIZABLE_CLASS)); + + doReturn(actorRef.path().toString()).when(mockActorContext).resolvePath(eq(actorRef.path().toString()), + eq(actorRef.path().toString())); + + TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, + READ_WRITE); + + transactionProxy.read(TestModel.TEST_PATH); + + transactionProxy.write(TestModel.TEST_PATH, nodeToWrite); + + DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready(); + + assertTrue(ready instanceof ThreePhaseCommitCohortProxy); + + ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready; + + verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), + WriteDataReply.SERIALIZABLE_CLASS); + + verifyCohortFutures(proxy, getSystem().actorSelection(actorRef.path())); + + verify(mockActorContext).resolvePath(eq(actorRef.path().toString()), + eq(actorRef.path().toString())); + } + @Test public void testReadyWithRecordingOperationFailure() throws Exception { ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY); @@ -800,7 +896,7 @@ public class TransactionProxyTest extends AbstractActorTest { doReturn(readySerializedTxReply(actorRef.path().toString())).when(mockActorContext).executeOperationAsync( eq(actorSelection(actorRef)), isA(ReadyTransaction.SERIALIZABLE_CLASS)); - doReturn(false).when(mockActorContext).isLocalPath(actorRef.path().toString()); + doReturn(false).when(mockActorContext).isPathLocal(actorRef.path().toString()); TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); @@ -815,13 +911,12 @@ public class TransactionProxyTest extends AbstractActorTest { ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready; + verifyCohortFutures(proxy, TestException.class); + verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), MergeDataReply.SERIALIZABLE_CLASS, TestException.class); - - verifyCohortFutures(proxy, TestException.class); } - @SuppressWarnings("unchecked") @Test public void testReadyWithReplyFailure() throws Exception { ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY); @@ -855,9 +950,8 @@ public class TransactionProxyTest extends AbstractActorTest { @Test public void testReadyWithInitialCreateTransactionFailure() throws Exception { - doReturn(Optional.absent()).when(mockActorContext).findPrimaryShard(anyString()); -// doThrow(new PrimaryNotFoundException("mock")).when(mockActorContext).executeShardOperation( -// anyString(), any()); + doReturn(Futures.failed(new PrimaryNotFoundException("mock"))).when( + mockActorContext).findPrimaryShardAsync(anyString()); TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); @@ -879,7 +973,6 @@ public class TransactionProxyTest extends AbstractActorTest { verifyCohortFutures(proxy, PrimaryNotFoundException.class); } - @SuppressWarnings("unchecked") @Test public void testReadyWithInvalidReplyMessageType() throws Exception { ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY); @@ -918,7 +1011,6 @@ public class TransactionProxyTest extends AbstractActorTest { assertTrue("Invalid identifier: " + id, id.toString().startsWith(memberName)); } - @SuppressWarnings("unchecked") @Test public void testClose() throws Exception{ ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE); @@ -958,8 +1050,8 @@ public class TransactionProxyTest extends AbstractActorTest { doReturn(actorSystem.actorSelection(shardActorRef.path())). when(mockActorContext).actorSelection(shardActorRef.path().toString()); - doReturn(Optional.of(actorSystem.actorSelection(shardActorRef.path()))). - when(mockActorContext).findPrimaryShard(eq(DefaultShardStrategy.DEFAULT_SHARD)); + doReturn(Futures.successful(actorSystem.actorSelection(shardActorRef.path()))). + when(mockActorContext).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD)); String actorPath = "akka.tcp://system@127.0.0.1:2550/user/tx-actor"; CreateTransactionReply createTransactionReply = CreateTransactionReply.newBuilder() @@ -967,11 +1059,11 @@ public class TransactionProxyTest extends AbstractActorTest { .setTransactionActorPath(actorPath) .build(); - doReturn(createTransactionReply).when(mockActorContext). - executeOperation(eq(actorSystem.actorSelection(shardActorRef.path())), + doReturn(Futures.successful(createTransactionReply)).when(mockActorContext). + executeOperationAsync(eq(actorSystem.actorSelection(shardActorRef.path())), eqCreateTransaction(memberName, READ_ONLY)); - doReturn(true).when(mockActorContext).isLocalPath(actorPath); + doReturn(true).when(mockActorContext).isPathLocal(actorPath); TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,READ_ONLY); @@ -1013,8 +1105,8 @@ public class TransactionProxyTest extends AbstractActorTest { doReturn(actorSystem.actorSelection(shardActorRef.path())). when(mockActorContext).actorSelection(shardActorRef.path().toString()); - doReturn(Optional.of(actorSystem.actorSelection(shardActorRef.path()))). - when(mockActorContext).findPrimaryShard(eq(DefaultShardStrategy.DEFAULT_SHARD)); + doReturn(Futures.successful(actorSystem.actorSelection(shardActorRef.path()))). + when(mockActorContext).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD)); String actorPath = "akka.tcp://system@127.0.0.1:2550/user/tx-actor"; CreateTransactionReply createTransactionReply = CreateTransactionReply.newBuilder() @@ -1022,11 +1114,11 @@ public class TransactionProxyTest extends AbstractActorTest { .setTransactionActorPath(actorPath) .build(); - doReturn(createTransactionReply).when(mockActorContext). - executeOperation(eq(actorSystem.actorSelection(shardActorRef.path())), + doReturn(Futures.successful(createTransactionReply)).when(mockActorContext). + executeOperationAsync(eq(actorSystem.actorSelection(shardActorRef.path())), eqCreateTransaction(memberName, WRITE_ONLY)); - doReturn(true).when(mockActorContext).isLocalPath(actorPath); + doReturn(true).when(mockActorContext).isPathLocal(actorPath); NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);