opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/compat/PreLithiumTransactionProxyTest.java

   1 /*
   2  * Copyright (c) 2015 Brocade Communications Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8 package org.opendaylight.controller.cluster.datastore.compat;
   9
  10 import static org.junit.Assert.assertEquals;
  11 import static org.junit.Assert.assertTrue;
  12 import static org.mockito.Matchers.any;
  13 import static org.mockito.Matchers.argThat;
  14 import static org.mockito.Matchers.eq;
  15 import static org.mockito.Matchers.isA;
  16 import static org.mockito.Mockito.doReturn;
  17 import static org.mockito.Mockito.verify;
  18 import static org.opendaylight.controller.cluster.datastore.TransactionType.READ_WRITE;
  19 import static org.opendaylight.controller.cluster.datastore.TransactionType.WRITE_ONLY;
  20 import akka.actor.ActorRef;
  21 import akka.dispatch.Futures;
  22 import akka.util.Timeout;
  23 import com.google.common.base.Optional;
  24 import java.util.concurrent.TimeUnit;
  25 import org.junit.Ignore;
  26 import org.junit.Test;
  27 import org.mockito.ArgumentMatcher;
  28 import org.mockito.Mockito;
  29 import org.opendaylight.controller.cluster.datastore.AbstractThreePhaseCommitCohort;
  30 import org.opendaylight.controller.cluster.datastore.AbstractTransactionProxyTest;
  31 import org.opendaylight.controller.cluster.datastore.DataStoreVersions;
  32 import org.opendaylight.controller.cluster.datastore.ThreePhaseCommitCohortProxy;
  33 import org.opendaylight.controller.cluster.datastore.TransactionProxy;
  34 import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransaction;
  35 import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransactionReply;
  36 import org.opendaylight.controller.cluster.datastore.messages.CommitTransaction;
  37 import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
  38 import org.opendaylight.controller.cluster.datastore.messages.DeleteData;
  39 import org.opendaylight.controller.cluster.datastore.messages.DeleteDataReply;
  40 import org.opendaylight.controller.cluster.datastore.messages.MergeData;
  41 import org.opendaylight.controller.cluster.datastore.messages.MergeDataReply;
  42 import org.opendaylight.controller.cluster.datastore.messages.ReadyTransaction;
  43 import org.opendaylight.controller.cluster.datastore.messages.ReadyTransactionReply;
  44 import org.opendaylight.controller.cluster.datastore.messages.WriteData;
  45 import org.opendaylight.controller.cluster.datastore.messages.WriteDataReply;
  46 import org.opendaylight.controller.cluster.datastore.shardstrategy.DefaultShardStrategy;
  47 import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
  48 import org.opendaylight.controller.protobuff.messages.cohort3pc.ThreePhaseCommitCohortMessages;
  49 import org.opendaylight.controller.protobuff.messages.transaction.ShardTransactionMessages;
  50 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
  51 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  52 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  53 import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
  54 import scala.concurrent.Future;
  55
  56 /**
  57  * Unit tests for backwards compatibility with pre-Lithium versions.
  58  *
  59  * @author Thomas Pantelis
  60  */
  61 public class PreLithiumTransactionProxyTest extends AbstractTransactionProxyTest {
  62
  63     private WriteData eqLegacyWriteData(final NormalizedNode<?, ?> nodeToWrite) {
  64         ArgumentMatcher<WriteData> matcher = new ArgumentMatcher<WriteData>() {
  65             @Override
  66             public boolean matches(Object argument) {
  67                 if(ShardTransactionMessages.WriteData.class.equals(argument.getClass())) {
  68                     WriteData obj = WriteData.fromSerializable(argument);
  69                     return obj.getPath().equals(TestModel.TEST_PATH) && obj.getData().equals(nodeToWrite);
  70                 }
  71
  72                 return false;
  73             }
  74         };
  75
  76         return argThat(matcher);
  77     }
  78
  79     private MergeData eqLegacyMergeData(final NormalizedNode<?, ?> nodeToWrite) {
  80         ArgumentMatcher<MergeData> matcher = new ArgumentMatcher<MergeData>() {
  81             @Override
  82             public boolean matches(Object argument) {
  83                 if(ShardTransactionMessages.MergeData.class.equals(argument.getClass())) {
  84                     MergeData obj = MergeData.fromSerializable(argument);
  85                     return obj.getPath().equals(TestModel.TEST_PATH) && obj.getData().equals(nodeToWrite);
  86                 }
  87
  88                 return false;
  89             }
  90         };
  91
  92         return argThat(matcher);
  93     }
  94
  95     private DeleteData eqLegacyDeleteData(final YangInstanceIdentifier expPath) {
  96         ArgumentMatcher<DeleteData> matcher = new ArgumentMatcher<DeleteData>() {
  97             @Override
  98             public boolean matches(Object argument) {
  99                 return ShardTransactionMessages.DeleteData.class.equals(argument.getClass()) &&
 100                        DeleteData.fromSerializable(argument).getPath().equals(expPath);
 101             }
 102         };
 103
 104         return argThat(matcher);
 105     }
 106
 107     private CanCommitTransaction eqCanCommitTransaction(final String transactionID) {
 108         ArgumentMatcher<CanCommitTransaction> matcher = new ArgumentMatcher<CanCommitTransaction>() {
 109             @Override
 110             public boolean matches(Object argument) {
 111                 return ThreePhaseCommitCohortMessages.CanCommitTransaction.class.equals(argument.getClass()) &&
 112                         CanCommitTransaction.fromSerializable(argument).getTransactionID().equals(transactionID);
 113             }
 114         };
 115
 116         return argThat(matcher);
 117     }
 118
 119     private CommitTransaction eqCommitTransaction(final String transactionID) {
 120         ArgumentMatcher<CommitTransaction> matcher = new ArgumentMatcher<CommitTransaction>() {
 121             @Override
 122             public boolean matches(Object argument) {
 123                 return ThreePhaseCommitCohortMessages.CommitTransaction.class.equals(argument.getClass()) &&
 124                         CommitTransaction.fromSerializable(argument).getTransactionID().equals(transactionID);
 125             }
 126         };
 127
 128         return argThat(matcher);
 129     }
 130
 131     private Future<Object> readySerializedTxReply(String path, short version) {
 132         return Futures.successful(new ReadyTransactionReply(path, version).toSerializable());
 133     }
 134
 135     private ActorRef testCompatibilityWithHeliumVersion(short version) throws Exception {
 136         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE, version,
 137                 DefaultShardStrategy.DEFAULT_SHARD);
 138
 139         NormalizedNode<?, ?> testNode = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 140
 141         doReturn(readSerializedDataReply(testNode, version)).when(mockActorContext).executeOperationAsync(
 142                 eq(actorSelection(actorRef)), eqSerializedReadData(TestModel.TEST_PATH));
 143
 144         doReturn(Futures.successful(new WriteDataReply().toSerializable(version))).when(mockActorContext).
 145                 executeOperationAsync(eq(actorSelection(actorRef)), eqLegacyWriteData(testNode));
 146
 147         doReturn(Futures.successful(new MergeDataReply().toSerializable(version))).when(mockActorContext).
 148                 executeOperationAsync(eq(actorSelection(actorRef)), eqLegacyMergeData(testNode));
 149
 150         doReturn(Futures.successful(new DeleteDataReply().toSerializable(version))).when(mockActorContext).
 151                 executeOperationAsync(eq(actorSelection(actorRef)), eqLegacyDeleteData(TestModel.TEST_PATH));
 152
 153         doReturn(readySerializedTxReply(actorRef.path().toString(), version)).when(mockActorContext).executeOperationAsync(
 154                 eq(actorSelection(actorRef)), isA(ReadyTransaction.SERIALIZABLE_CLASS));
 155
 156         doReturn(actorRef.path().toString()).when(mockActorContext).resolvePath(eq(actorRef.path().toString()),
 157                 eq(actorRef.path().toString()));
 158
 159         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE);
 160
 161         Optional<NormalizedNode<?, ?>> readOptional = transactionProxy.read(TestModel.TEST_PATH).
 162                 get(5, TimeUnit.SECONDS);
 163
 164         assertEquals("NormalizedNode isPresent", true, readOptional.isPresent());
 165         assertEquals("Response NormalizedNode", testNode, readOptional.get());
 166
 167         transactionProxy.write(TestModel.TEST_PATH, testNode);
 168
 169         transactionProxy.merge(TestModel.TEST_PATH, testNode);
 170
 171         transactionProxy.delete(TestModel.TEST_PATH);
 172
 173         AbstractThreePhaseCommitCohort<?> proxy = transactionProxy.ready();
 174
 175         verifyCohortFutures(proxy, getSystem().actorSelection(actorRef.path()));
 176
 177         doReturn(Futures.successful(CanCommitTransactionReply.YES.toSerializable())).when(mockActorContext).
 178                 executeOperationAsync(eq(actorSelection(actorRef)),
 179                     eqCanCommitTransaction(transactionProxy.getIdentifier().toString()), any(Timeout.class));
 180
 181         doReturn(Futures.successful(new CommitTransactionReply().toSerializable())).when(mockActorContext).
 182                 executeOperationAsync(eq(actorSelection(actorRef)),
 183                    eqCommitTransaction(transactionProxy.getIdentifier().toString()), any(Timeout.class));
 184
 185         Boolean canCommit = proxy.canCommit().get(3, TimeUnit.SECONDS);
 186         assertEquals("canCommit", true, canCommit.booleanValue());
 187
 188         proxy.preCommit().get(3, TimeUnit.SECONDS);
 189
 190         proxy.commit().get(3, TimeUnit.SECONDS);
 191
 192         return actorRef;
 193     }
 194
 195     @Test
 196     public void testCompatibilityWithBaseHeliumVersion() throws Exception {
 197         ActorRef actorRef = testCompatibilityWithHeliumVersion(DataStoreVersions.BASE_HELIUM_VERSION);
 198
 199         verify(mockActorContext).resolvePath(eq(actorRef.path().toString()),
 200                 eq(actorRef.path().toString()));
 201     }
 202
 203     @Test
 204     public void testCompatibilityWithHeliumR1Version() throws Exception {
 205         ActorRef actorRef = testCompatibilityWithHeliumVersion(DataStoreVersions.HELIUM_1_VERSION);
 206
 207         verify(mockActorContext, Mockito.never()).resolvePath(eq(actorRef.path().toString()),
 208                 eq(actorRef.path().toString()));
 209     }
 210
 211     @Test
 212     @Ignore
 213     // FIXME: disabled until we can get the primary shard version from the ShardManager as we now skip
 214     // creating transaction actors for write-only Tx's.
 215     public void testWriteOnlyCompatibilityWithHeliumR2Version() throws Exception {
 216         short version = DataStoreVersions.HELIUM_2_VERSION;
 217         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY, version,
 218                 DefaultShardStrategy.DEFAULT_SHARD);
 219
 220         NormalizedNode<?, ?> testNode = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 221
 222         doReturn(Futures.successful(new WriteDataReply().toSerializable(version))).when(mockActorContext).
 223                 executeOperationAsync(eq(actorSelection(actorRef)), eqLegacyWriteData(testNode));
 224
 225         doReturn(readySerializedTxReply(actorRef.path().toString(), version)).when(mockActorContext).executeOperationAsync(
 226                 eq(actorSelection(actorRef)), isA(ReadyTransaction.SERIALIZABLE_CLASS));
 227
 228         doReturn(actorRef.path().toString()).when(mockActorContext).resolvePath(eq(actorRef.path().toString()),
 229                 eq(actorRef.path().toString()));
 230
 231         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY);
 232
 233         transactionProxy.write(TestModel.TEST_PATH, testNode);
 234
 235         DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
 236
 237         assertTrue(ready instanceof ThreePhaseCommitCohortProxy);
 238
 239         ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready;
 240
 241         verifyCohortFutures(proxy, getSystem().actorSelection(actorRef.path()));
 242     }
 243 }