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%2FAbstractTransactionProxyTest.java;h=cf5bff0846ba8cc3635f958edac2892b00554493;hb=refs%2Fchanges%2F78%2F33178%2F3;hp=a64a5802b8387102bdfac19e4535a440088ccf55;hpb=9d77fd7cf8aa00f063ed4a7875738f53b92518e9;p=controller.git diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/AbstractTransactionProxyTest.java b/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/AbstractTransactionProxyTest.java index a64a5802b8..cf5bff0846 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/AbstractTransactionProxyTest.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/AbstractTransactionProxyTest.java @@ -23,6 +23,7 @@ import akka.actor.ActorSystem; import akka.actor.Props; import akka.dispatch.Futures; import akka.testkit.JavaTestKit; +import akka.util.Timeout; import com.codahale.metrics.MetricRegistry; import com.codahale.metrics.Timer; import com.google.common.base.Objects; @@ -46,17 +47,19 @@ import org.mockito.Mock; import org.mockito.Mockito; import org.mockito.MockitoAnnotations; import org.opendaylight.controller.cluster.datastore.DatastoreContext.Builder; -import org.opendaylight.controller.cluster.datastore.TransactionProxy.TransactionType; import org.opendaylight.controller.cluster.datastore.TransactionProxyTest.TestException; +import org.opendaylight.controller.cluster.datastore.config.Configuration; import org.opendaylight.controller.cluster.datastore.messages.BatchedModifications; import org.opendaylight.controller.cluster.datastore.messages.BatchedModificationsReply; import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply; import org.opendaylight.controller.cluster.datastore.messages.CreateTransaction; +import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionReply; import org.opendaylight.controller.cluster.datastore.messages.DataExists; import org.opendaylight.controller.cluster.datastore.messages.DataExistsReply; import org.opendaylight.controller.cluster.datastore.messages.PrimaryShardInfo; import org.opendaylight.controller.cluster.datastore.messages.ReadData; import org.opendaylight.controller.cluster.datastore.messages.ReadDataReply; +import org.opendaylight.controller.cluster.datastore.messages.ReadyLocalTransaction; import org.opendaylight.controller.cluster.datastore.messages.ReadyTransactionReply; import org.opendaylight.controller.cluster.datastore.modification.AbstractModification; import org.opendaylight.controller.cluster.datastore.modification.Modification; @@ -65,11 +68,11 @@ import org.opendaylight.controller.cluster.datastore.shardstrategy.DefaultShardS import org.opendaylight.controller.cluster.datastore.shardstrategy.ShardStrategy; import org.opendaylight.controller.cluster.datastore.shardstrategy.ShardStrategyFactory; import org.opendaylight.controller.cluster.datastore.utils.ActorContext; -import org.opendaylight.controller.cluster.datastore.utils.DoNothingActor; import org.opendaylight.controller.cluster.datastore.utils.MockConfiguration; +import org.opendaylight.controller.cluster.raft.utils.DoNothingActor; +import org.opendaylight.controller.md.cluster.datastore.model.CarsModel; 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.yangtools.yang.data.api.YangInstanceIdentifier; import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode; import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree; @@ -91,27 +94,41 @@ public abstract class AbstractTransactionProxyTest { private static ActorSystem system; private final Configuration configuration = new MockConfiguration() { + Map strategyMap = ImmutableMap.builder().put( + "junk", new ShardStrategy() { + @Override + public String findShard(YangInstanceIdentifier path) { + return "junk"; + } + }).put( + "cars", new ShardStrategy() { + @Override + public String findShard(YangInstanceIdentifier path) { + return "cars"; + } + }).build(); + @Override - public Map getModuleNameToShardStrategyMap() { - return ImmutableMap.builder().put( - "junk", new ShardStrategy() { - @Override - public String findShard(YangInstanceIdentifier path) { - return "junk"; - } - }).build(); + public ShardStrategy getStrategyForModule(String moduleName) { + return strategyMap.get(moduleName); } @Override - public Optional getModuleNameFromNameSpace(String nameSpace) { - return TestModel.JUNK_QNAME.getNamespace().toASCIIString().equals(nameSpace) ? - Optional.of("junk") : Optional.absent(); + public String getModuleNameFromNameSpace(String nameSpace) { + if(TestModel.JUNK_QNAME.getNamespace().toASCIIString().equals(nameSpace)) { + return "junk"; + } else if(CarsModel.BASE_QNAME.getNamespace().toASCIIString().equals(nameSpace)){ + return "cars"; + } + return null; } }; @Mock protected ActorContext mockActorContext; + protected TransactionContextFactory mockComponentFactory; + private SchemaContext schemaContext; @Mock @@ -119,7 +136,9 @@ public abstract class AbstractTransactionProxyTest { protected final String memberName = "mock-member"; - protected final Builder dataStoreContextBuilder = DatastoreContext.newBuilder().operationTimeoutInSeconds(2); + private final int operationTimeoutInSeconds = 2; + protected final Builder dataStoreContextBuilder = DatastoreContext.newBuilder() + .operationTimeoutInSeconds(operationTimeoutInSeconds); @BeforeClass public static void setUpClass() throws IOException { @@ -146,16 +165,17 @@ public abstract class AbstractTransactionProxyTest { doReturn(getSystem()).when(mockActorContext).getActorSystem(); doReturn(getSystem().dispatchers().defaultGlobalDispatcher()).when(mockActorContext).getClientDispatcher(); doReturn(memberName).when(mockActorContext).getCurrentMemberName(); + doReturn(new ShardStrategyFactory(configuration)).when(mockActorContext).getShardStrategyFactory(); doReturn(schemaContext).when(mockActorContext).getSchemaContext(); + doReturn(new Timeout(operationTimeoutInSeconds, TimeUnit.SECONDS)).when(mockActorContext).getOperationTimeout(); doReturn(mockClusterWrapper).when(mockActorContext).getClusterWrapper(); doReturn(mockClusterWrapper).when(mockActorContext).getClusterWrapper(); doReturn(dataStoreContextBuilder.build()).when(mockActorContext).getDatastoreContext(); - doReturn(10).when(mockActorContext).getTransactionOutstandingOperationLimit(); + + mockComponentFactory = TransactionContextFactory.create(mockActorContext); Timer timer = new MetricRegistry().timer("test"); doReturn(timer).when(mockActorContext).getOperationTimer(any(String.class)); - - ShardStrategyFactory.setConfiguration(configuration); } protected ActorSystem getSystem() { @@ -167,7 +187,7 @@ public abstract class AbstractTransactionProxyTest { ArgumentMatcher matcher = new ArgumentMatcher() { @Override public boolean matches(Object argument) { - if(CreateTransaction.SERIALIZABLE_CLASS.equals(argument.getClass())) { + if(CreateTransaction.class.equals(argument.getClass())) { CreateTransaction obj = CreateTransaction.fromSerializable(argument); return obj.getTransactionId().startsWith(memberName) && obj.getTransactionType() == type.ordinal(); @@ -180,18 +200,6 @@ public abstract class AbstractTransactionProxyTest { return argThat(matcher); } - protected DataExists eqSerializedDataExists() { - ArgumentMatcher matcher = new ArgumentMatcher() { - @Override - public boolean matches(Object argument) { - return DataExists.SERIALIZABLE_CLASS.equals(argument.getClass()) && - DataExists.fromSerializable(argument).getPath().equals(TestModel.TEST_PATH); - } - }; - - return argThat(matcher); - } - protected DataExists eqDataExists() { ArgumentMatcher matcher = new ArgumentMatcher() { @Override @@ -204,28 +212,15 @@ public abstract class AbstractTransactionProxyTest { return argThat(matcher); } - protected ReadData eqSerializedReadData() { - return eqSerializedReadData(TestModel.TEST_PATH); - } - - protected ReadData eqSerializedReadData(final YangInstanceIdentifier path) { - ArgumentMatcher matcher = new ArgumentMatcher() { - @Override - public boolean matches(Object argument) { - return ReadData.SERIALIZABLE_CLASS.equals(argument.getClass()) && - ReadData.fromSerializable(argument).getPath().equals(path); - } - }; - - return argThat(matcher); + protected ReadData eqReadData() { + return eqReadData(TestModel.TEST_PATH); } - protected ReadData eqReadData() { + protected ReadData eqReadData(final YangInstanceIdentifier path) { ArgumentMatcher matcher = new ArgumentMatcher() { @Override public boolean matches(Object argument) { - return (argument instanceof ReadData) && - ((ReadData)argument).getPath().equals(TestModel.TEST_PATH); + return (argument instanceof ReadData) && ((ReadData)argument).getPath().equals(path); } }; @@ -236,31 +231,20 @@ public abstract class AbstractTransactionProxyTest { return Futures.successful((Object)new ReadyTransactionReply(path)); } - protected Future readSerializedDataReply(NormalizedNode data, - short transactionVersion) { - return Futures.successful(new ReadDataReply(data, transactionVersion).toSerializable()); - } - - protected Future readSerializedDataReply(NormalizedNode data) { - return readSerializedDataReply(data, DataStoreVersions.CURRENT_VERSION); - } protected Future readDataReply(NormalizedNode data) { return Futures.successful(new ReadDataReply(data, DataStoreVersions.CURRENT_VERSION)); } - protected Future dataExistsSerializedReply(boolean exists) { - return Futures.successful(DataExistsReply.create(exists).toSerializable()); - } - protected Future dataExistsReply(boolean exists) { - return Futures.successful(DataExistsReply.create(exists)); + return Futures.successful(new DataExistsReply(exists, DataStoreVersions.CURRENT_VERSION)); } protected Future batchedModificationsReply(int count) { return Futures.successful(new BatchedModificationsReply(count)); } + @SuppressWarnings("unchecked") protected Future incompleteFuture() { return mock(Future.class); } @@ -299,12 +283,14 @@ public abstract class AbstractTransactionProxyTest { eq(actorSelection(actorRef)), isA(BatchedModifications.class)); } - protected CreateTransactionReply createTransactionReply(ActorRef actorRef, int transactionVersion){ - return CreateTransactionReply.newBuilder() - .setTransactionActorPath(actorRef.path().toString()) - .setTransactionId("txn-1") - .setMessageVersion(transactionVersion) - .build(); + protected void expectReadyLocalTransaction(ActorRef actorRef, boolean doCommitOnReady) { + doReturn(doCommitOnReady ? Futures.successful(new CommitTransactionReply().toSerializable()) : + readyTxReply(actorRef.path().toString())).when(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), isA(ReadyLocalTransaction.class), any(Timeout.class)); + } + + protected CreateTransactionReply createTransactionReply(ActorRef actorRef, short transactionVersion){ + return new CreateTransactionReply(actorRef.path().toString(), "txn-1", transactionVersion); } protected ActorRef setupActorContextWithoutInitialCreateTransaction(ActorSystem actorSystem) { @@ -312,40 +298,47 @@ public abstract class AbstractTransactionProxyTest { } protected Future primaryShardInfoReply(ActorSystem actorSystem, ActorRef actorRef) { + return primaryShardInfoReply(actorSystem, actorRef, DataStoreVersions.CURRENT_VERSION); + } + + protected Future primaryShardInfoReply(ActorSystem actorSystem, ActorRef actorRef, + short transactionVersion) { return Futures.successful(new PrimaryShardInfo(actorSystem.actorSelection(actorRef.path()), - Optional.absent())); + transactionVersion, Optional.absent())); } protected ActorRef setupActorContextWithoutInitialCreateTransaction(ActorSystem actorSystem, String shardName) { + return setupActorContextWithoutInitialCreateTransaction(actorSystem, shardName, DataStoreVersions.CURRENT_VERSION); + } + + protected ActorRef setupActorContextWithoutInitialCreateTransaction(ActorSystem actorSystem, String shardName, + short transactionVersion) { ActorRef actorRef = actorSystem.actorOf(Props.create(DoNothingActor.class)); log.info("Created mock shard actor {}", actorRef); doReturn(actorSystem.actorSelection(actorRef.path())). when(mockActorContext).actorSelection(actorRef.path().toString()); - doReturn(primaryShardInfoReply(actorSystem, actorRef)). + doReturn(primaryShardInfoReply(actorSystem, actorRef, transactionVersion)). when(mockActorContext).findPrimaryShardAsync(eq(shardName)); - doReturn(false).when(mockActorContext).isPathLocal(actorRef.path().toString()); - - doReturn(10).when(mockActorContext).getTransactionOutstandingOperationLimit(); - return actorRef; } protected ActorRef setupActorContextWithInitialCreateTransaction(ActorSystem actorSystem, - TransactionType type, int transactionVersion, String shardName) { - ActorRef shardActorRef = setupActorContextWithoutInitialCreateTransaction(actorSystem, shardName); + TransactionType type, short transactionVersion, String shardName) { + ActorRef shardActorRef = setupActorContextWithoutInitialCreateTransaction(actorSystem, shardName, + transactionVersion); return setupActorContextWithInitialCreateTransaction(actorSystem, type, transactionVersion, memberName, shardActorRef); } protected ActorRef setupActorContextWithInitialCreateTransaction(ActorSystem actorSystem, - TransactionType type, int transactionVersion, String prefix, ActorRef shardActorRef) { + TransactionType type, short transactionVersion, String prefix, ActorRef shardActorRef) { ActorRef txActorRef; - if(type == TransactionType.WRITE_ONLY && transactionVersion >= DataStoreVersions.LITHIUM_VERSION && + if(type == TransactionType.WRITE_ONLY && dataStoreContextBuilder.build().isWriteOnlyTransactionOptimizationsEnabled()) { txActorRef = shardActorRef; } else { @@ -353,11 +346,11 @@ public abstract class AbstractTransactionProxyTest { log.info("Created mock shard Tx actor {}", txActorRef); doReturn(actorSystem.actorSelection(txActorRef.path())). - when(mockActorContext).actorSelection(txActorRef.path().toString()); + when(mockActorContext).actorSelection(txActorRef.path().toString()); doReturn(Futures.successful(createTransactionReply(txActorRef, transactionVersion))).when(mockActorContext). - executeOperationAsync(eq(actorSystem.actorSelection(shardActorRef.path())), - eqCreateTransaction(prefix, type)); + executeOperationAsync(eq(actorSystem.actorSelection(shardActorRef.path())), + eqCreateTransaction(prefix, type), any(Timeout.class)); } return txActorRef; @@ -381,7 +374,12 @@ public abstract class AbstractTransactionProxyTest { future.checkedGet(5, TimeUnit.SECONDS); fail("Expected ReadFailedException"); } catch(ReadFailedException e) { - throw e.getCause(); + assertNotNull("Expected a cause", e.getCause()); + if(e.getCause().getCause() != null) { + throw e.getCause().getCause(); + } else { + throw e.getCause(); + } } }