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%2FAbstractTransactionProxyTest.java;h=d3ae761b599198a06b12b63774d51d955c19af7b;hp=a64a5802b8387102bdfac19e4535a440088ccf55;hb=HEAD;hpb=ba3433c7bf94e551a4ea520c95165a9358bf9227 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 deleted file mode 100644 index a64a5802b8..0000000000 --- a/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/AbstractTransactionProxyTest.java +++ /dev/null @@ -1,481 +0,0 @@ -/* - * Copyright (c) 2015 Brocade Communications Systems, Inc. and others. All rights reserved. - * - * This program and the accompanying materials are made available under the - * terms of the Eclipse Public License v1.0 which accompanies this distribution, - * and is available at http://www.eclipse.org/legal/epl-v10.html - */ -package org.opendaylight.controller.cluster.datastore; - -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.fail; -import static org.mockito.Matchers.any; -import static org.mockito.Matchers.argThat; -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.verify; -import akka.actor.ActorRef; -import akka.actor.ActorSelection; -import akka.actor.ActorSystem; -import akka.actor.Props; -import akka.dispatch.Futures; -import akka.testkit.JavaTestKit; -import com.codahale.metrics.MetricRegistry; -import com.codahale.metrics.Timer; -import com.google.common.base.Objects; -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 java.io.IOException; -import java.util.ArrayList; -import java.util.Iterator; -import java.util.List; -import java.util.Map; -import java.util.concurrent.TimeUnit; -import org.junit.AfterClass; -import org.junit.Before; -import org.junit.BeforeClass; -import org.mockito.ArgumentCaptor; -import org.mockito.ArgumentMatcher; -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.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.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.ReadyTransactionReply; -import org.opendaylight.controller.cluster.datastore.modification.AbstractModification; -import org.opendaylight.controller.cluster.datastore.modification.Modification; -import org.opendaylight.controller.cluster.datastore.modification.WriteModification; -import org.opendaylight.controller.cluster.datastore.shardstrategy.DefaultShardStrategy; -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.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; -import org.opendaylight.yangtools.yang.model.api.SchemaContext; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import scala.concurrent.Await; -import scala.concurrent.Future; -import scala.concurrent.duration.Duration; - -/** - * Abstract base class for TransactionProxy unit tests. - * - * @author Thomas Pantelis - */ -public abstract class AbstractTransactionProxyTest { - protected final Logger log = LoggerFactory.getLogger(getClass()); - - private static ActorSystem system; - - private final Configuration configuration = new MockConfiguration() { - @Override - public Map getModuleNameToShardStrategyMap() { - return ImmutableMap.builder().put( - "junk", new ShardStrategy() { - @Override - public String findShard(YangInstanceIdentifier path) { - return "junk"; - } - }).build(); - } - - @Override - public Optional getModuleNameFromNameSpace(String nameSpace) { - return TestModel.JUNK_QNAME.getNamespace().toASCIIString().equals(nameSpace) ? - Optional.of("junk") : Optional.absent(); - } - }; - - @Mock - protected ActorContext mockActorContext; - - private SchemaContext schemaContext; - - @Mock - private ClusterWrapper mockClusterWrapper; - - protected final String memberName = "mock-member"; - - protected final Builder dataStoreContextBuilder = DatastoreContext.newBuilder().operationTimeoutInSeconds(2); - - @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(); - - doReturn(getSystem()).when(mockActorContext).getActorSystem(); - doReturn(getSystem().dispatchers().defaultGlobalDispatcher()).when(mockActorContext).getClientDispatcher(); - doReturn(memberName).when(mockActorContext).getCurrentMemberName(); - doReturn(schemaContext).when(mockActorContext).getSchemaContext(); - doReturn(mockClusterWrapper).when(mockActorContext).getClusterWrapper(); - doReturn(mockClusterWrapper).when(mockActorContext).getClusterWrapper(); - doReturn(dataStoreContextBuilder.build()).when(mockActorContext).getDatastoreContext(); - doReturn(10).when(mockActorContext).getTransactionOutstandingOperationLimit(); - - Timer timer = new MetricRegistry().timer("test"); - doReturn(timer).when(mockActorContext).getOperationTimer(any(String.class)); - - ShardStrategyFactory.setConfiguration(configuration); - } - - protected ActorSystem getSystem() { - return system; - } - - protected CreateTransaction eqCreateTransaction(final String memberName, - final TransactionType type) { - ArgumentMatcher matcher = new ArgumentMatcher() { - @Override - public boolean matches(Object argument) { - if(CreateTransaction.SERIALIZABLE_CLASS.equals(argument.getClass())) { - CreateTransaction obj = CreateTransaction.fromSerializable(argument); - return obj.getTransactionId().startsWith(memberName) && - obj.getTransactionType() == type.ordinal(); - } - - return false; - } - }; - - 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 - public boolean matches(Object argument) { - return (argument instanceof DataExists) && - ((DataExists)argument).getPath().equals(TestModel.TEST_PATH); - } - }; - - 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() { - ArgumentMatcher matcher = new ArgumentMatcher() { - @Override - public boolean matches(Object argument) { - return (argument instanceof ReadData) && - ((ReadData)argument).getPath().equals(TestModel.TEST_PATH); - } - }; - - return argThat(matcher); - } - - protected Future readyTxReply(String path) { - 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)); - } - - protected Future batchedModificationsReply(int count) { - return Futures.successful(new BatchedModificationsReply(count)); - } - - protected Future incompleteFuture() { - return mock(Future.class); - } - - protected ActorSelection actorSelection(ActorRef actorRef) { - return getSystem().actorSelection(actorRef.path()); - } - - protected void expectBatchedModifications(ActorRef actorRef, int count) { - doReturn(batchedModificationsReply(count)).when(mockActorContext).executeOperationAsync( - eq(actorSelection(actorRef)), isA(BatchedModifications.class)); - } - - protected void expectBatchedModificationsReady(ActorRef actorRef) { - expectBatchedModificationsReady(actorRef, false); - } - - protected void expectBatchedModificationsReady(ActorRef actorRef, boolean doCommitOnReady) { - doReturn(doCommitOnReady ? Futures.successful(new CommitTransactionReply().toSerializable()) : - readyTxReply(actorRef.path().toString())).when(mockActorContext).executeOperationAsync( - eq(actorSelection(actorRef)), isA(BatchedModifications.class)); - } - - protected void expectBatchedModifications(int count) { - doReturn(batchedModificationsReply(count)).when(mockActorContext).executeOperationAsync( - any(ActorSelection.class), isA(BatchedModifications.class)); - } - - protected void expectIncompleteBatchedModifications() { - doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync( - any(ActorSelection.class), isA(BatchedModifications.class)); - } - - protected void expectFailedBatchedModifications(ActorRef actorRef) { - doReturn(Futures.failed(new TestException())).when(mockActorContext).executeOperationAsync( - 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 ActorRef setupActorContextWithoutInitialCreateTransaction(ActorSystem actorSystem) { - return setupActorContextWithoutInitialCreateTransaction(actorSystem, DefaultShardStrategy.DEFAULT_SHARD); - } - - protected Future primaryShardInfoReply(ActorSystem actorSystem, ActorRef actorRef) { - return Futures.successful(new PrimaryShardInfo(actorSystem.actorSelection(actorRef.path()), - Optional.absent())); - } - - protected ActorRef setupActorContextWithoutInitialCreateTransaction(ActorSystem actorSystem, String shardName) { - 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)). - 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); - - return setupActorContextWithInitialCreateTransaction(actorSystem, type, transactionVersion, - memberName, shardActorRef); - } - - protected ActorRef setupActorContextWithInitialCreateTransaction(ActorSystem actorSystem, - TransactionType type, int transactionVersion, String prefix, ActorRef shardActorRef) { - - ActorRef txActorRef; - if(type == TransactionType.WRITE_ONLY && transactionVersion >= DataStoreVersions.LITHIUM_VERSION && - dataStoreContextBuilder.build().isWriteOnlyTransactionOptimizationsEnabled()) { - txActorRef = shardActorRef; - } else { - txActorRef = actorSystem.actorOf(Props.create(DoNothingActor.class)); - log.info("Created mock shard Tx actor {}", txActorRef); - - doReturn(actorSystem.actorSelection(txActorRef.path())). - when(mockActorContext).actorSelection(txActorRef.path().toString()); - - doReturn(Futures.successful(createTransactionReply(txActorRef, transactionVersion))).when(mockActorContext). - executeOperationAsync(eq(actorSystem.actorSelection(shardActorRef.path())), - eqCreateTransaction(prefix, type)); - } - - return txActorRef; - } - - protected ActorRef setupActorContextWithInitialCreateTransaction(ActorSystem actorSystem, TransactionType type) { - return setupActorContextWithInitialCreateTransaction(actorSystem, type, DataStoreVersions.CURRENT_VERSION, - DefaultShardStrategy.DEFAULT_SHARD); - } - - protected ActorRef setupActorContextWithInitialCreateTransaction(ActorSystem actorSystem, TransactionType type, - String shardName) { - return setupActorContextWithInitialCreateTransaction(actorSystem, type, DataStoreVersions.CURRENT_VERSION, - shardName); - } - - protected void propagateReadFailedExceptionCause(CheckedFuture future) - throws Throwable { - - try { - future.checkedGet(5, TimeUnit.SECONDS); - fail("Expected ReadFailedException"); - } catch(ReadFailedException e) { - throw e.getCause(); - } - } - - protected List captureBatchedModifications(ActorRef actorRef) { - ArgumentCaptor batchedModificationsCaptor = - ArgumentCaptor.forClass(BatchedModifications.class); - verify(mockActorContext, Mockito.atLeastOnce()).executeOperationAsync( - eq(actorSelection(actorRef)), batchedModificationsCaptor.capture()); - - List batchedModifications = filterCaptured( - batchedModificationsCaptor, BatchedModifications.class); - return batchedModifications; - } - - protected List filterCaptured(ArgumentCaptor captor, Class type) { - List captured = new ArrayList<>(); - for(T c: captor.getAllValues()) { - if(type.isInstance(c)) { - captured.add(c); - } - } - - return captured; - } - - protected void verifyOneBatchedModification(ActorRef actorRef, Modification expected, boolean expIsReady) { - List batchedModifications = captureBatchedModifications(actorRef); - assertEquals("Captured BatchedModifications count", 1, batchedModifications.size()); - - verifyBatchedModifications(batchedModifications.get(0), expIsReady, expIsReady, expected); - } - - protected void verifyBatchedModifications(Object message, boolean expIsReady, Modification... expected) { - verifyBatchedModifications(message, expIsReady, false, expected); - } - - protected void verifyBatchedModifications(Object message, boolean expIsReady, boolean expIsDoCommitOnReady, - Modification... expected) { - assertEquals("Message type", BatchedModifications.class, message.getClass()); - BatchedModifications batchedModifications = (BatchedModifications)message; - assertEquals("BatchedModifications size", expected.length, batchedModifications.getModifications().size()); - assertEquals("isReady", expIsReady, batchedModifications.isReady()); - assertEquals("isDoCommitOnReady", expIsDoCommitOnReady, batchedModifications.isDoCommitOnReady()); - for(int i = 0; i < batchedModifications.getModifications().size(); i++) { - Modification actual = batchedModifications.getModifications().get(i); - assertEquals("Modification type", expected[i].getClass(), actual.getClass()); - assertEquals("getPath", ((AbstractModification)expected[i]).getPath(), - ((AbstractModification)actual).getPath()); - if(actual instanceof WriteModification) { - assertEquals("getData", ((WriteModification)expected[i]).getData(), - ((WriteModification)actual).getData()); - } - } - } - - protected void verifyCohortFutures(AbstractThreePhaseCommitCohort proxy, - Object... expReplies) throws Exception { - assertEquals("getReadyOperationFutures size", expReplies.length, - proxy.getCohortFutures().size()); - - List futureResults = new ArrayList<>(); - for( Future future: proxy.getCohortFutures()) { - assertNotNull("Ready operation Future is null", future); - try { - futureResults.add(Await.result(future, Duration.create(5, TimeUnit.SECONDS))); - } catch(Exception e) { - futureResults.add(e); - } - } - - for(int i = 0; i < expReplies.length; i++) { - Object expReply = expReplies[i]; - boolean found = false; - Iterator iter = futureResults.iterator(); - while(iter.hasNext()) { - Object actual = iter.next(); - if(CommitTransactionReply.SERIALIZABLE_CLASS.isInstance(expReply) && - CommitTransactionReply.SERIALIZABLE_CLASS.isInstance(actual)) { - found = true; - } else if(expReply instanceof ActorSelection && Objects.equal(expReply, actual)) { - found = true; - } else if(expReply instanceof Class && ((Class)expReply).isInstance(actual)) { - found = true; - } - - if(found) { - iter.remove(); - break; - } - } - - if(!found) { - fail(String.format("No cohort Future response found for %s. Actual: %s", expReply, futureResults)); - } - } - } -}