CDS: Implement front-end support for local transactions
[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.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(mockComponentFactory, 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(mockComponentFactory, 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 }

©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.