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

©2013 OpenDaylight, A Linux Foundation Collaborative Project. All Rights Reserved.
OpenDaylight is a registered trademark of The OpenDaylight Project, Inc.
Linux Foundation and OpenDaylight are registered trademarks of the Linux Foundation.
Linux is a registered trademark of Linus Torvalds.