Fix test case mis-spelling.
[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / test / java / org / opendaylight / controller / cluster / datastore / TransactionProxyTest.java
1 package org.opendaylight.controller.cluster.datastore;
2
3 import akka.actor.ActorRef;
4 import akka.actor.ActorSelection;
5 import akka.actor.ActorSystem;
6 import akka.actor.Props;
7 import akka.dispatch.Futures;
8 import akka.testkit.JavaTestKit;
9 import com.google.common.base.Optional;
10 import com.google.common.collect.ImmutableMap;
11 import com.google.common.util.concurrent.CheckedFuture;
12 import com.typesafe.config.Config;
13 import com.typesafe.config.ConfigFactory;
14 import org.junit.AfterClass;
15 import org.junit.Before;
16 import org.junit.BeforeClass;
17 import org.junit.Test;
18 import org.mockito.ArgumentMatcher;
19 import org.mockito.Mock;
20 import org.mockito.MockitoAnnotations;
21 import org.opendaylight.controller.cluster.datastore.TransactionProxy.TransactionType;
22 import org.opendaylight.controller.cluster.datastore.exceptions.PrimaryNotFoundException;
23 import org.opendaylight.controller.cluster.datastore.exceptions.TimeoutException;
24 import org.opendaylight.controller.cluster.datastore.messages.CloseTransaction;
25 import org.opendaylight.controller.cluster.datastore.messages.CreateTransaction;
26 import org.opendaylight.controller.cluster.datastore.messages.DataExists;
27 import org.opendaylight.controller.cluster.datastore.messages.DataExistsReply;
28 import org.opendaylight.controller.cluster.datastore.messages.DeleteData;
29 import org.opendaylight.controller.cluster.datastore.messages.DeleteDataReply;
30 import org.opendaylight.controller.cluster.datastore.messages.MergeData;
31 import org.opendaylight.controller.cluster.datastore.messages.MergeDataReply;
32 import org.opendaylight.controller.cluster.datastore.messages.ReadData;
33 import org.opendaylight.controller.cluster.datastore.messages.ReadDataReply;
34 import org.opendaylight.controller.cluster.datastore.messages.ReadyTransaction;
35 import org.opendaylight.controller.cluster.datastore.messages.ReadyTransactionReply;
36 import org.opendaylight.controller.cluster.datastore.messages.WriteData;
37 import org.opendaylight.controller.cluster.datastore.messages.WriteDataReply;
38 import org.opendaylight.controller.cluster.datastore.shardstrategy.DefaultShardStrategy;
39 import org.opendaylight.controller.cluster.datastore.shardstrategy.ShardStrategyFactory;
40 import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
41 import org.opendaylight.controller.cluster.datastore.utils.DoNothingActor;
42 import org.opendaylight.controller.cluster.datastore.utils.MockConfiguration;
43 import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
44 import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
45 import org.opendaylight.controller.protobuff.messages.transaction.ShardTransactionMessages.CreateTransactionReply;
46 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
47 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
48 import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
49 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
50 import scala.concurrent.Await;
51 import scala.concurrent.Future;
52 import scala.concurrent.duration.Duration;
53 import java.io.IOException;
54 import java.util.List;
55 import java.util.concurrent.TimeUnit;
56 import static org.junit.Assert.assertEquals;
57 import static org.junit.Assert.assertNotNull;
58 import static org.junit.Assert.assertTrue;
59 import static org.junit.Assert.fail;
60 import static org.mockito.Matchers.any;
61 import static org.mockito.Matchers.anyString;
62 import static org.mockito.Mockito.argThat;
63 import static org.mockito.Mockito.doReturn;
64 import static org.mockito.Mockito.eq;
65 import static org.mockito.Mockito.isA;
66 import static org.mockito.Mockito.times;
67 import static org.mockito.Mockito.verify;
68 import static org.opendaylight.controller.cluster.datastore.TransactionProxy.TransactionType.READ_ONLY;
69 import static org.opendaylight.controller.cluster.datastore.TransactionProxy.TransactionType.READ_WRITE;
70 import static org.opendaylight.controller.cluster.datastore.TransactionProxy.TransactionType.WRITE_ONLY;
71
72 @SuppressWarnings("resource")
73 public class TransactionProxyTest {
74
75     @SuppressWarnings("serial")
76     static class TestException extends RuntimeException {
77     }
78
79     static interface Invoker {
80         CheckedFuture<?, ReadFailedException> invoke(TransactionProxy proxy) throws Exception;
81     }
82
83     private static ActorSystem system;
84
85     private final Configuration configuration = new MockConfiguration();
86
87     @Mock
88     private ActorContext mockActorContext;
89
90     private SchemaContext schemaContext;
91
92     @Mock
93     private ClusterWrapper mockClusterWrapper;
94
95     String memberName = "mock-member";
96
97     @BeforeClass
98     public static void setUpClass() throws IOException {
99
100         Config config = ConfigFactory.parseMap(ImmutableMap.<String, Object>builder().
101                 put("akka.actor.default-dispatcher.type",
102                         "akka.testkit.CallingThreadDispatcherConfigurator").build()).
103                 withFallback(ConfigFactory.load());
104         system = ActorSystem.create("test", config);
105     }
106
107     @AfterClass
108     public static void tearDownClass() throws IOException {
109         JavaTestKit.shutdownActorSystem(system);
110         system = null;
111     }
112
113     @Before
114     public void setUp(){
115         MockitoAnnotations.initMocks(this);
116
117         schemaContext = TestModel.createTestContext();
118
119         DatastoreContext dataStoreContext = DatastoreContext.newBuilder().build();
120
121         doReturn(getSystem()).when(mockActorContext).getActorSystem();
122         doReturn(memberName).when(mockActorContext).getCurrentMemberName();
123         doReturn(schemaContext).when(mockActorContext).getSchemaContext();
124         doReturn(mockClusterWrapper).when(mockActorContext).getClusterWrapper();
125         doReturn(mockClusterWrapper).when(mockActorContext).getClusterWrapper();
126         doReturn(dataStoreContext).when(mockActorContext).getDatastoreContext();
127
128         ShardStrategyFactory.setConfiguration(configuration);
129     }
130
131     private ActorSystem getSystem() {
132         return system;
133     }
134
135     private CreateTransaction eqCreateTransaction(final String memberName,
136             final TransactionType type) {
137         ArgumentMatcher<CreateTransaction> matcher = new ArgumentMatcher<CreateTransaction>() {
138             @Override
139             public boolean matches(Object argument) {
140                 CreateTransaction obj = CreateTransaction.fromSerializable(argument);
141                 return obj.getTransactionId().startsWith(memberName) &&
142                        obj.getTransactionType() == type.ordinal();
143             }
144         };
145
146         return argThat(matcher);
147     }
148
149     private DataExists eqSerializedDataExists() {
150         ArgumentMatcher<DataExists> matcher = new ArgumentMatcher<DataExists>() {
151             @Override
152             public boolean matches(Object argument) {
153                 return DataExists.SERIALIZABLE_CLASS.equals(argument.getClass()) &&
154                        DataExists.fromSerializable(argument).getPath().equals(TestModel.TEST_PATH);
155             }
156         };
157
158         return argThat(matcher);
159     }
160
161     private DataExists eqDataExists() {
162         ArgumentMatcher<DataExists> matcher = new ArgumentMatcher<DataExists>() {
163             @Override
164             public boolean matches(Object argument) {
165                 return (argument instanceof DataExists) &&
166                     ((DataExists)argument).getPath().equals(TestModel.TEST_PATH);
167             }
168         };
169
170         return argThat(matcher);
171     }
172
173     private ReadData eqSerializedReadData() {
174         ArgumentMatcher<ReadData> matcher = new ArgumentMatcher<ReadData>() {
175             @Override
176             public boolean matches(Object argument) {
177                 return ReadData.SERIALIZABLE_CLASS.equals(argument.getClass()) &&
178                        ReadData.fromSerializable(argument).getPath().equals(TestModel.TEST_PATH);
179             }
180         };
181
182         return argThat(matcher);
183     }
184
185     private ReadData eqReadData() {
186         ArgumentMatcher<ReadData> matcher = new ArgumentMatcher<ReadData>() {
187             @Override
188             public boolean matches(Object argument) {
189                 return (argument instanceof ReadData) &&
190                     ((ReadData)argument).getPath().equals(TestModel.TEST_PATH);
191             }
192         };
193
194         return argThat(matcher);
195     }
196
197     private WriteData eqSerializedWriteData(final NormalizedNode<?, ?> nodeToWrite) {
198         ArgumentMatcher<WriteData> matcher = new ArgumentMatcher<WriteData>() {
199             @Override
200             public boolean matches(Object argument) {
201                 if(!WriteData.SERIALIZABLE_CLASS.equals(argument.getClass())) {
202                     return false;
203                 }
204
205                 WriteData obj = WriteData.fromSerializable(argument, schemaContext);
206                 return obj.getPath().equals(TestModel.TEST_PATH) &&
207                        obj.getData().equals(nodeToWrite);
208             }
209         };
210
211         return argThat(matcher);
212     }
213
214     private WriteData eqWriteData(final NormalizedNode<?, ?> nodeToWrite) {
215         ArgumentMatcher<WriteData> matcher = new ArgumentMatcher<WriteData>() {
216             @Override
217             public boolean matches(Object argument) {
218                 if(argument instanceof WriteData) {
219                     WriteData obj = (WriteData) argument;
220                     return obj.getPath().equals(TestModel.TEST_PATH) &&
221                         obj.getData().equals(nodeToWrite);
222                 }
223                 return false;
224             }
225         };
226
227         return argThat(matcher);
228     }
229
230     private MergeData eqSerializedMergeData(final NormalizedNode<?, ?> nodeToWrite) {
231         ArgumentMatcher<MergeData> matcher = new ArgumentMatcher<MergeData>() {
232             @Override
233             public boolean matches(Object argument) {
234                 if(!MergeData.SERIALIZABLE_CLASS.equals(argument.getClass())) {
235                     return false;
236                 }
237
238                 MergeData obj = MergeData.fromSerializable(argument, schemaContext);
239                 return obj.getPath().equals(TestModel.TEST_PATH) &&
240                        obj.getData().equals(nodeToWrite);
241             }
242         };
243
244         return argThat(matcher);
245     }
246
247     private MergeData eqMergeData(final NormalizedNode<?, ?> nodeToWrite) {
248         ArgumentMatcher<MergeData> matcher = new ArgumentMatcher<MergeData>() {
249             @Override
250             public boolean matches(Object argument) {
251                 if(argument instanceof MergeData) {
252                     MergeData obj = ((MergeData) argument);
253                     return obj.getPath().equals(TestModel.TEST_PATH) &&
254                         obj.getData().equals(nodeToWrite);
255                 }
256
257                return false;
258             }
259         };
260
261         return argThat(matcher);
262     }
263
264     private DeleteData eqSerializedDeleteData() {
265         ArgumentMatcher<DeleteData> matcher = new ArgumentMatcher<DeleteData>() {
266             @Override
267             public boolean matches(Object argument) {
268                 return DeleteData.SERIALIZABLE_CLASS.equals(argument.getClass()) &&
269                        DeleteData.fromSerializable(argument).getPath().equals(TestModel.TEST_PATH);
270             }
271         };
272
273         return argThat(matcher);
274     }
275
276         private DeleteData eqDeleteData() {
277         ArgumentMatcher<DeleteData> matcher = new ArgumentMatcher<DeleteData>() {
278             @Override
279             public boolean matches(Object argument) {
280                 return argument instanceof DeleteData &&
281                     ((DeleteData)argument).getPath().equals(TestModel.TEST_PATH);
282             }
283         };
284
285         return argThat(matcher);
286     }
287
288     private Future<Object> readySerializedTxReply(String path) {
289         return Futures.successful((Object)new ReadyTransactionReply(path).toSerializable());
290     }
291
292     private Future<Object> readyTxReply(String path) {
293         return Futures.successful((Object)new ReadyTransactionReply(path));
294     }
295
296
297     private Future<Object> readSerializedDataReply(NormalizedNode<?, ?> data) {
298         return Futures.successful(new ReadDataReply(schemaContext, data).toSerializable());
299     }
300
301     private Future<ReadDataReply> readDataReply(NormalizedNode<?, ?> data) {
302         return Futures.successful(new ReadDataReply(schemaContext, data));
303     }
304
305     private Future<Object> dataExistsSerializedReply(boolean exists) {
306         return Futures.successful(new DataExistsReply(exists).toSerializable());
307     }
308
309     private Future<DataExistsReply> dataExistsReply(boolean exists) {
310         return Futures.successful(new DataExistsReply(exists));
311     }
312
313     private Future<Object> writeSerializedDataReply() {
314         return Futures.successful(new WriteDataReply().toSerializable());
315     }
316
317     private Future<WriteDataReply> writeDataReply() {
318         return Futures.successful(new WriteDataReply());
319     }
320
321     private Future<Object> mergeSerializedDataReply() {
322         return Futures.successful(new MergeDataReply().toSerializable());
323     }
324
325     private Future<MergeDataReply> mergeDataReply() {
326         return Futures.successful(new MergeDataReply());
327     }
328
329     private Future<Object> deleteSerializedDataReply() {
330         return Futures.successful(new DeleteDataReply().toSerializable());
331     }
332
333     private Future<DeleteDataReply> deleteDataReply() {
334         return Futures.successful(new DeleteDataReply());
335     }
336
337     private ActorSelection actorSelection(ActorRef actorRef) {
338         return getSystem().actorSelection(actorRef.path());
339     }
340
341     private CreateTransactionReply createTransactionReply(ActorRef actorRef, int transactionVersion){
342         return CreateTransactionReply.newBuilder()
343             .setTransactionActorPath(actorRef.path().toString())
344             .setTransactionId("txn-1")
345             .setMessageVersion(transactionVersion)
346             .build();
347     }
348
349     private ActorRef setupActorContextWithInitialCreateTransaction(ActorSystem actorSystem, TransactionType type, int transactionVersion) {
350         ActorRef actorRef = actorSystem.actorOf(Props.create(DoNothingActor.class));
351         doReturn(actorSystem.actorSelection(actorRef.path())).
352                 when(mockActorContext).actorSelection(actorRef.path().toString());
353
354         doReturn(Futures.successful(actorSystem.actorSelection(actorRef.path()))).
355                 when(mockActorContext).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD));
356
357         doReturn(Futures.successful(createTransactionReply(actorRef, transactionVersion))).when(mockActorContext).
358                 executeOperationAsync(eq(actorSystem.actorSelection(actorRef.path())),
359                         eqCreateTransaction(memberName, type));
360
361         doReturn(false).when(mockActorContext).isPathLocal(actorRef.path().toString());
362
363         return actorRef;
364     }
365
366     private ActorRef setupActorContextWithInitialCreateTransaction(ActorSystem actorSystem, TransactionType type) {
367         return setupActorContextWithInitialCreateTransaction(actorSystem, type, CreateTransaction.CURRENT_VERSION);
368     }
369
370
371     private void propagateReadFailedExceptionCause(CheckedFuture<?, ReadFailedException> future)
372             throws Throwable {
373
374         try {
375             future.checkedGet(5, TimeUnit.SECONDS);
376             fail("Expected ReadFailedException");
377         } catch(ReadFailedException e) {
378             throw e.getCause();
379         }
380     }
381
382     @Test
383     public void testRead() throws Exception {
384         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY);
385
386         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,
387                 READ_ONLY);
388
389         doReturn(readSerializedDataReply(null)).when(mockActorContext).executeOperationAsync(
390                 eq(actorSelection(actorRef)), eqSerializedReadData());
391
392         Optional<NormalizedNode<?, ?>> readOptional = transactionProxy.read(
393                 TestModel.TEST_PATH).get(5, TimeUnit.SECONDS);
394
395         assertEquals("NormalizedNode isPresent", false, readOptional.isPresent());
396
397         NormalizedNode<?, ?> expectedNode = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
398
399         doReturn(readSerializedDataReply(expectedNode)).when(mockActorContext).executeOperationAsync(
400                 eq(actorSelection(actorRef)), eqSerializedReadData());
401
402         readOptional = transactionProxy.read(TestModel.TEST_PATH).get(5, TimeUnit.SECONDS);
403
404         assertEquals("NormalizedNode isPresent", true, readOptional.isPresent());
405
406         assertEquals("Response NormalizedNode", expectedNode, readOptional.get());
407     }
408
409     @Test(expected = ReadFailedException.class)
410     public void testReadWithInvalidReplyMessageType() throws Exception {
411         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY);
412
413         doReturn(Futures.successful(new Object())).when(mockActorContext).
414                 executeOperationAsync(eq(actorSelection(actorRef)), eqSerializedReadData());
415
416         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,
417                 READ_ONLY);
418
419         transactionProxy.read(TestModel.TEST_PATH).checkedGet(5, TimeUnit.SECONDS);
420     }
421
422     @Test(expected = TestException.class)
423     public void testReadWithAsyncRemoteOperatonFailure() throws Throwable {
424         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY);
425
426         doReturn(Futures.failed(new TestException())).when(mockActorContext).
427                 executeOperationAsync(eq(actorSelection(actorRef)), eqSerializedReadData());
428
429         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,
430                 READ_ONLY);
431
432         propagateReadFailedExceptionCause(transactionProxy.read(TestModel.TEST_PATH));
433     }
434
435     private void testExceptionOnInitialCreateTransaction(Exception exToThrow, Invoker invoker)
436             throws Throwable {
437         ActorRef actorRef = getSystem().actorOf(Props.create(DoNothingActor.class));
438
439         if (exToThrow instanceof PrimaryNotFoundException) {
440             doReturn(Futures.failed(exToThrow)).when(mockActorContext).findPrimaryShardAsync(anyString());
441         } else {
442             doReturn(Futures.successful(getSystem().actorSelection(actorRef.path()))).
443                     when(mockActorContext).findPrimaryShardAsync(anyString());
444         }
445
446         doReturn(Futures.failed(exToThrow)).when(mockActorContext).executeOperationAsync(
447                 any(ActorSelection.class), any());
448
449         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_ONLY);
450
451         propagateReadFailedExceptionCause(invoker.invoke(transactionProxy));
452     }
453
454     private void testReadWithExceptionOnInitialCreateTransaction(Exception exToThrow) throws Throwable {
455         testExceptionOnInitialCreateTransaction(exToThrow, new Invoker() {
456             @Override
457             public CheckedFuture<?, ReadFailedException> invoke(TransactionProxy proxy) throws Exception {
458                 return proxy.read(TestModel.TEST_PATH);
459             }
460         });
461     }
462
463     @Test(expected = PrimaryNotFoundException.class)
464     public void testReadWhenAPrimaryNotFoundExceptionIsThrown() throws Throwable {
465         testReadWithExceptionOnInitialCreateTransaction(new PrimaryNotFoundException("test"));
466     }
467
468     @Test(expected = TimeoutException.class)
469     public void testReadWhenATimeoutExceptionIsThrown() throws Throwable {
470         testReadWithExceptionOnInitialCreateTransaction(new TimeoutException("test",
471                 new Exception("reason")));
472     }
473
474     @Test(expected = TestException.class)
475     public void testReadWhenAnyOtherExceptionIsThrown() throws Throwable {
476         testReadWithExceptionOnInitialCreateTransaction(new TestException());
477     }
478
479     @Test(expected = TestException.class)
480     public void testReadWithPriorRecordingOperationFailure() throws Throwable {
481         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE);
482
483         NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
484
485         doReturn(writeSerializedDataReply()).when(mockActorContext).executeOperationAsync(
486                 eq(actorSelection(actorRef)), eqSerializedWriteData(nodeToWrite));
487
488         doReturn(Futures.failed(new TestException())).when(mockActorContext).
489                 executeOperationAsync(eq(actorSelection(actorRef)), eqSerializedDeleteData());
490
491         doReturn(readSerializedDataReply(null)).when(mockActorContext).executeOperationAsync(
492                 eq(actorSelection(actorRef)), eqSerializedReadData());
493
494         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,
495                 READ_WRITE);
496
497         transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
498
499         transactionProxy.delete(TestModel.TEST_PATH);
500
501         try {
502             propagateReadFailedExceptionCause(transactionProxy.read(TestModel.TEST_PATH));
503         } finally {
504             verify(mockActorContext, times(0)).executeOperationAsync(
505                     eq(actorSelection(actorRef)), eqSerializedReadData());
506         }
507     }
508
509     @Test
510     public void testReadWithPriorRecordingOperationSuccessful() throws Throwable {
511         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE);
512
513         NormalizedNode<?, ?> expectedNode = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
514
515         doReturn(writeSerializedDataReply()).when(mockActorContext).executeOperationAsync(
516                 eq(actorSelection(actorRef)), eqSerializedWriteData(expectedNode));
517
518         doReturn(readSerializedDataReply(expectedNode)).when(mockActorContext).executeOperationAsync(
519                 eq(actorSelection(actorRef)), eqSerializedReadData());
520
521         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,
522                 READ_WRITE);
523
524         transactionProxy.write(TestModel.TEST_PATH, expectedNode);
525
526         Optional<NormalizedNode<?, ?>> readOptional = transactionProxy.read(
527                 TestModel.TEST_PATH).get(5, TimeUnit.SECONDS);
528
529         assertEquals("NormalizedNode isPresent", true, readOptional.isPresent());
530
531         assertEquals("Response NormalizedNode", expectedNode, readOptional.get());
532     }
533
534     @Test(expected=IllegalStateException.class)
535     public void testReadPreConditionCheck() {
536
537         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,
538                 WRITE_ONLY);
539
540         transactionProxy.read(TestModel.TEST_PATH);
541     }
542
543     @Test(expected=IllegalArgumentException.class)
544     public void testInvalidCreateTransactionReply() throws Throwable {
545         ActorRef actorRef = getSystem().actorOf(Props.create(DoNothingActor.class));
546
547         doReturn(getSystem().actorSelection(actorRef.path())).when(mockActorContext).
548             actorSelection(actorRef.path().toString());
549
550         doReturn(Futures.successful(getSystem().actorSelection(actorRef.path()))).
551             when(mockActorContext).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD));
552
553         doReturn(Futures.successful(new Object())).when(mockActorContext).executeOperationAsync(
554             eq(getSystem().actorSelection(actorRef.path())), eqCreateTransaction(memberName, READ_ONLY));
555
556         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_ONLY);
557
558         propagateReadFailedExceptionCause(transactionProxy.read(TestModel.TEST_PATH));
559     }
560
561     @Test
562     public void testExists() throws Exception {
563         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY);
564
565         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,
566                 READ_ONLY);
567
568         doReturn(dataExistsSerializedReply(false)).when(mockActorContext).executeOperationAsync(
569                 eq(actorSelection(actorRef)), eqSerializedDataExists());
570
571         Boolean exists = transactionProxy.exists(TestModel.TEST_PATH).checkedGet();
572
573         assertEquals("Exists response", false, exists);
574
575         doReturn(dataExistsSerializedReply(true)).when(mockActorContext).executeOperationAsync(
576                 eq(actorSelection(actorRef)), eqSerializedDataExists());
577
578         exists = transactionProxy.exists(TestModel.TEST_PATH).checkedGet();
579
580         assertEquals("Exists response", true, exists);
581     }
582
583     @Test(expected = PrimaryNotFoundException.class)
584     public void testExistsWhenAPrimaryNotFoundExceptionIsThrown() throws Throwable {
585         testExceptionOnInitialCreateTransaction(new PrimaryNotFoundException("test"), new Invoker() {
586             @Override
587             public CheckedFuture<?, ReadFailedException> invoke(TransactionProxy proxy) throws Exception {
588                 return proxy.exists(TestModel.TEST_PATH);
589             }
590         });
591     }
592
593     @Test(expected = ReadFailedException.class)
594     public void testExistsWithInvalidReplyMessageType() throws Exception {
595         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY);
596
597         doReturn(Futures.successful(new Object())).when(mockActorContext).
598                 executeOperationAsync(eq(actorSelection(actorRef)), eqSerializedDataExists());
599
600         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,
601                 READ_ONLY);
602
603         transactionProxy.exists(TestModel.TEST_PATH).checkedGet(5, TimeUnit.SECONDS);
604     }
605
606     @Test(expected = TestException.class)
607     public void testExistsWithAsyncRemoteOperatonFailure() throws Throwable {
608         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY);
609
610         doReturn(Futures.failed(new TestException())).when(mockActorContext).
611                 executeOperationAsync(eq(actorSelection(actorRef)), eqSerializedDataExists());
612
613         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,
614                 READ_ONLY);
615
616         propagateReadFailedExceptionCause(transactionProxy.exists(TestModel.TEST_PATH));
617     }
618
619     @Test(expected = TestException.class)
620     public void testExistsWithPriorRecordingOperationFailure() throws Throwable {
621         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE);
622
623         NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
624
625         doReturn(writeSerializedDataReply()).when(mockActorContext).executeOperationAsync(
626                 eq(actorSelection(actorRef)), eqSerializedWriteData(nodeToWrite));
627
628         doReturn(Futures.failed(new TestException())).when(mockActorContext).
629                 executeOperationAsync(eq(actorSelection(actorRef)), eqSerializedDeleteData());
630
631         doReturn(dataExistsSerializedReply(false)).when(mockActorContext).executeOperationAsync(
632                 eq(actorSelection(actorRef)), eqSerializedDataExists());
633
634         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,
635                 READ_WRITE);
636
637         transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
638
639         transactionProxy.delete(TestModel.TEST_PATH);
640
641         try {
642             propagateReadFailedExceptionCause(transactionProxy.exists(TestModel.TEST_PATH));
643         } finally {
644             verify(mockActorContext, times(0)).executeOperationAsync(
645                     eq(actorSelection(actorRef)), eqSerializedDataExists());
646         }
647     }
648
649     @Test
650     public void testExistsWithPriorRecordingOperationSuccessful() throws Throwable {
651         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE);
652
653         NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
654
655         doReturn(writeSerializedDataReply()).when(mockActorContext).executeOperationAsync(
656                 eq(actorSelection(actorRef)), eqSerializedWriteData(nodeToWrite));
657
658         doReturn(dataExistsSerializedReply(true)).when(mockActorContext).executeOperationAsync(
659                 eq(actorSelection(actorRef)), eqSerializedDataExists());
660
661         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,
662                 READ_WRITE);
663
664         transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
665
666         Boolean exists = transactionProxy.exists(TestModel.TEST_PATH).checkedGet();
667
668         assertEquals("Exists response", true, exists);
669     }
670
671     @Test(expected=IllegalStateException.class)
672     public void testExistsPreConditionCheck() {
673
674         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,
675                 WRITE_ONLY);
676
677         transactionProxy.exists(TestModel.TEST_PATH);
678     }
679
680     private void verifyRecordingOperationFutures(List<Future<Object>> futures,
681             Class<?>... expResultTypes) throws Exception {
682         assertEquals("getRecordingOperationFutures size", expResultTypes.length, futures.size());
683
684         int i = 0;
685         for( Future<Object> future: futures) {
686             assertNotNull("Recording operation Future is null", future);
687
688             Class<?> expResultType = expResultTypes[i++];
689             if(Throwable.class.isAssignableFrom(expResultType)) {
690                 try {
691                     Await.result(future, Duration.create(5, TimeUnit.SECONDS));
692                     fail("Expected exception from recording operation Future");
693                 } catch(Exception e) {
694                     // Expected
695                 }
696             } else {
697                 assertEquals("Recording operation Future result type", expResultType,
698                              Await.result(future, Duration.create(5, TimeUnit.SECONDS)).getClass());
699             }
700         }
701     }
702
703     @Test
704     public void testWrite() throws Exception {
705         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY);
706
707         NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
708
709         doReturn(writeSerializedDataReply()).when(mockActorContext).executeOperationAsync(
710                 eq(actorSelection(actorRef)), eqSerializedWriteData(nodeToWrite));
711
712         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,
713                 WRITE_ONLY);
714
715         transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
716
717         verify(mockActorContext).executeOperationAsync(
718                 eq(actorSelection(actorRef)), eqSerializedWriteData(nodeToWrite));
719
720         verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(),
721                 WriteDataReply.SERIALIZABLE_CLASS);
722     }
723
724     @Test(expected=IllegalStateException.class)
725     public void testWritePreConditionCheck() {
726
727         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,
728                 READ_ONLY);
729
730         transactionProxy.write(TestModel.TEST_PATH,
731                 ImmutableNodes.containerNode(TestModel.TEST_QNAME));
732     }
733
734     @Test(expected=IllegalStateException.class)
735     public void testWriteAfterReadyPreConditionCheck() {
736
737         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,
738                 WRITE_ONLY);
739
740         transactionProxy.ready();
741
742         transactionProxy.write(TestModel.TEST_PATH,
743                 ImmutableNodes.containerNode(TestModel.TEST_QNAME));
744     }
745
746     @Test
747     public void testMerge() throws Exception {
748         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY);
749
750         NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
751
752         doReturn(mergeSerializedDataReply()).when(mockActorContext).executeOperationAsync(
753                 eq(actorSelection(actorRef)), eqSerializedMergeData(nodeToWrite));
754
755         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY);
756
757         transactionProxy.merge(TestModel.TEST_PATH, nodeToWrite);
758
759         verify(mockActorContext).executeOperationAsync(
760                 eq(actorSelection(actorRef)), eqSerializedMergeData(nodeToWrite));
761
762         verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(),
763                 MergeDataReply.SERIALIZABLE_CLASS);
764     }
765
766     @Test
767     public void testDelete() throws Exception {
768         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY);
769
770         doReturn(deleteSerializedDataReply()).when(mockActorContext).executeOperationAsync(
771                 eq(actorSelection(actorRef)), eqSerializedDeleteData());
772
773         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,
774                 WRITE_ONLY);
775
776         transactionProxy.delete(TestModel.TEST_PATH);
777
778         verify(mockActorContext).executeOperationAsync(
779                 eq(actorSelection(actorRef)), eqSerializedDeleteData());
780
781         verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(),
782                 DeleteDataReply.SERIALIZABLE_CLASS);
783     }
784
785     private void verifyCohortFutures(ThreePhaseCommitCohortProxy proxy,
786         Object... expReplies) throws Exception {
787         assertEquals("getReadyOperationFutures size", expReplies.length,
788                 proxy.getCohortFutures().size());
789
790         int i = 0;
791         for( Future<ActorSelection> future: proxy.getCohortFutures()) {
792             assertNotNull("Ready operation Future is null", future);
793
794             Object expReply = expReplies[i++];
795             if(expReply instanceof ActorSelection) {
796                 ActorSelection actual = Await.result(future, Duration.create(5, TimeUnit.SECONDS));
797                 assertEquals("Cohort actor path", expReply, actual);
798             } else {
799                 // Expecting exception.
800                 try {
801                     Await.result(future, Duration.create(5, TimeUnit.SECONDS));
802                     fail("Expected exception from ready operation Future");
803                 } catch(Exception e) {
804                     // Expected
805                 }
806             }
807         }
808     }
809
810     @Test
811     public void testReady() throws Exception {
812         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE);
813
814         NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
815
816         doReturn(readSerializedDataReply(null)).when(mockActorContext).executeOperationAsync(
817                 eq(actorSelection(actorRef)), eqSerializedReadData());
818
819         doReturn(writeSerializedDataReply()).when(mockActorContext).executeOperationAsync(
820                 eq(actorSelection(actorRef)), eqSerializedWriteData(nodeToWrite));
821
822         doReturn(readySerializedTxReply(actorRef.path().toString())).when(mockActorContext).executeOperationAsync(
823                 eq(actorSelection(actorRef)), isA(ReadyTransaction.SERIALIZABLE_CLASS));
824
825         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,
826                 READ_WRITE);
827
828         transactionProxy.read(TestModel.TEST_PATH);
829
830         transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
831
832         DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
833
834         assertTrue(ready instanceof ThreePhaseCommitCohortProxy);
835
836         ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready;
837
838         verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(),
839                 WriteDataReply.SERIALIZABLE_CLASS);
840
841         verifyCohortFutures(proxy, getSystem().actorSelection(actorRef.path()));
842     }
843
844     @Test
845     public void testReadyForwardCompatibility() throws Exception {
846         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE, 0);
847
848         NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
849
850         doReturn(readSerializedDataReply(null)).when(mockActorContext).executeOperationAsync(
851                 eq(actorSelection(actorRef)), eqSerializedReadData());
852
853         doReturn(writeSerializedDataReply()).when(mockActorContext).executeOperationAsync(
854                 eq(actorSelection(actorRef)), eqSerializedWriteData(nodeToWrite));
855
856         doReturn(readySerializedTxReply(actorRef.path().toString())).when(mockActorContext).executeOperationAsync(
857                 eq(actorSelection(actorRef)), isA(ReadyTransaction.SERIALIZABLE_CLASS));
858
859         doReturn(actorRef.path().toString()).when(mockActorContext).resolvePath(eq(actorRef.path().toString()),
860                 eq(actorRef.path().toString()));
861
862         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,
863                 READ_WRITE);
864
865         transactionProxy.read(TestModel.TEST_PATH);
866
867         transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
868
869         DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
870
871         assertTrue(ready instanceof ThreePhaseCommitCohortProxy);
872
873         ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready;
874
875         verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(),
876                 WriteDataReply.SERIALIZABLE_CLASS);
877
878         verifyCohortFutures(proxy, getSystem().actorSelection(actorRef.path()));
879
880         verify(mockActorContext).resolvePath(eq(actorRef.path().toString()),
881                 eq(actorRef.path().toString()));
882     }
883
884     @Test
885     public void testReadyWithRecordingOperationFailure() throws Exception {
886         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY);
887
888         NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
889
890         doReturn(mergeSerializedDataReply()).when(mockActorContext).executeOperationAsync(
891                 eq(actorSelection(actorRef)), eqSerializedMergeData(nodeToWrite));
892
893         doReturn(Futures.failed(new TestException())).when(mockActorContext).
894                 executeOperationAsync(eq(actorSelection(actorRef)), eqSerializedWriteData(nodeToWrite));
895
896         doReturn(readySerializedTxReply(actorRef.path().toString())).when(mockActorContext).executeOperationAsync(
897                 eq(actorSelection(actorRef)), isA(ReadyTransaction.SERIALIZABLE_CLASS));
898
899         doReturn(false).when(mockActorContext).isPathLocal(actorRef.path().toString());
900
901         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,
902                 WRITE_ONLY);
903
904         transactionProxy.merge(TestModel.TEST_PATH, nodeToWrite);
905
906         transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
907
908         DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
909
910         assertTrue(ready instanceof ThreePhaseCommitCohortProxy);
911
912         ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready;
913
914         verifyCohortFutures(proxy, TestException.class);
915
916         verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(),
917                 MergeDataReply.SERIALIZABLE_CLASS, TestException.class);
918     }
919
920     @Test
921     public void testReadyWithReplyFailure() throws Exception {
922         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY);
923
924         NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
925
926         doReturn(mergeSerializedDataReply()).when(mockActorContext).executeOperationAsync(
927                 eq(actorSelection(actorRef)), eqSerializedMergeData(nodeToWrite));
928
929         doReturn(Futures.failed(new TestException())).when(mockActorContext).
930                 executeOperationAsync(eq(actorSelection(actorRef)),
931                         isA(ReadyTransaction.SERIALIZABLE_CLASS));
932
933         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,
934                 WRITE_ONLY);
935
936         transactionProxy.merge(TestModel.TEST_PATH, nodeToWrite);
937
938         DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
939
940         assertTrue(ready instanceof ThreePhaseCommitCohortProxy);
941
942         ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready;
943
944         verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(),
945                 MergeDataReply.SERIALIZABLE_CLASS);
946
947         verifyCohortFutures(proxy, TestException.class);
948     }
949
950     @Test
951     public void testReadyWithInitialCreateTransactionFailure() throws Exception {
952
953         doReturn(Futures.failed(new PrimaryNotFoundException("mock"))).when(
954                 mockActorContext).findPrimaryShardAsync(anyString());
955
956         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,
957                 WRITE_ONLY);
958
959         NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
960
961         transactionProxy.merge(TestModel.TEST_PATH, nodeToWrite);
962
963         transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
964
965         transactionProxy.delete(TestModel.TEST_PATH);
966
967         DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
968
969         assertTrue(ready instanceof ThreePhaseCommitCohortProxy);
970
971         ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready;
972
973         verifyCohortFutures(proxy, PrimaryNotFoundException.class);
974     }
975
976     @Test
977     public void testReadyWithInvalidReplyMessageType() throws Exception {
978         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY);
979
980         NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
981
982         doReturn(writeSerializedDataReply()).when(mockActorContext).executeOperationAsync(
983                 eq(actorSelection(actorRef)), eqSerializedWriteData(nodeToWrite));
984
985         doReturn(Futures.successful(new Object())).when(mockActorContext).
986                 executeOperationAsync(eq(actorSelection(actorRef)),
987                         isA(ReadyTransaction.SERIALIZABLE_CLASS));
988
989         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,
990                 WRITE_ONLY);
991
992         transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
993
994         DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
995
996         assertTrue(ready instanceof ThreePhaseCommitCohortProxy);
997
998         ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready;
999
1000         verifyCohortFutures(proxy, IllegalArgumentException.class);
1001     }
1002
1003     @Test
1004     public void testGetIdentifier() {
1005         setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY);
1006         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,
1007                 TransactionProxy.TransactionType.READ_ONLY);
1008
1009         Object id = transactionProxy.getIdentifier();
1010         assertNotNull("getIdentifier returned null", id);
1011         assertTrue("Invalid identifier: " + id, id.toString().startsWith(memberName));
1012     }
1013
1014     @Test
1015     public void testClose() throws Exception{
1016         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE);
1017
1018         doReturn(readSerializedDataReply(null)).when(mockActorContext).executeOperationAsync(
1019                 eq(actorSelection(actorRef)), eqSerializedReadData());
1020
1021         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,
1022                 READ_WRITE);
1023
1024         transactionProxy.read(TestModel.TEST_PATH);
1025
1026         transactionProxy.close();
1027
1028         verify(mockActorContext).sendOperationAsync(
1029                 eq(actorSelection(actorRef)), isA(CloseTransaction.SERIALIZABLE_CLASS));
1030     }
1031
1032
1033     /**
1034      * Method to test a local Tx actor. The Tx paths are matched to decide if the
1035      * Tx actor is local or not. This is done by mocking the Tx actor path
1036      * and the caller paths and ensuring that the paths have the remote-address format
1037      *
1038      * Note: Since the default akka provider for test is not a RemoteActorRefProvider,
1039      * the paths returned for the actors for all the tests are not qualified remote paths.
1040      * Hence are treated as non-local/remote actors. In short, all tests except
1041      * few below run for remote actors
1042      *
1043      * @throws Exception
1044      */
1045     @Test
1046     public void testLocalTxActorRead() throws Exception {
1047         ActorSystem actorSystem = getSystem();
1048         ActorRef shardActorRef = actorSystem.actorOf(Props.create(DoNothingActor.class));
1049
1050         doReturn(actorSystem.actorSelection(shardActorRef.path())).
1051             when(mockActorContext).actorSelection(shardActorRef.path().toString());
1052
1053         doReturn(Futures.successful(actorSystem.actorSelection(shardActorRef.path()))).
1054             when(mockActorContext).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD));
1055
1056         String actorPath = "akka.tcp://system@127.0.0.1:2550/user/tx-actor";
1057         CreateTransactionReply createTransactionReply = CreateTransactionReply.newBuilder()
1058             .setTransactionId("txn-1")
1059             .setTransactionActorPath(actorPath)
1060             .build();
1061
1062         doReturn(Futures.successful(createTransactionReply)).when(mockActorContext).
1063             executeOperationAsync(eq(actorSystem.actorSelection(shardActorRef.path())),
1064                 eqCreateTransaction(memberName, READ_ONLY));
1065
1066         doReturn(true).when(mockActorContext).isPathLocal(actorPath);
1067
1068         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,READ_ONLY);
1069
1070         // negative test case with null as the reply
1071         doReturn(readDataReply(null)).when(mockActorContext).executeOperationAsync(
1072             any(ActorSelection.class), eqReadData());
1073
1074         Optional<NormalizedNode<?, ?>> readOptional = transactionProxy.read(
1075             TestModel.TEST_PATH).get(5, TimeUnit.SECONDS);
1076
1077         assertEquals("NormalizedNode isPresent", false, readOptional.isPresent());
1078
1079         // test case with node as read data reply
1080         NormalizedNode<?, ?> expectedNode = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1081
1082         doReturn(readDataReply(expectedNode)).when(mockActorContext).executeOperationAsync(
1083             any(ActorSelection.class), eqReadData());
1084
1085         readOptional = transactionProxy.read(TestModel.TEST_PATH).get(5, TimeUnit.SECONDS);
1086
1087         assertEquals("NormalizedNode isPresent", true, readOptional.isPresent());
1088
1089         assertEquals("Response NormalizedNode", expectedNode, readOptional.get());
1090
1091         // test for local data exists
1092         doReturn(dataExistsReply(true)).when(mockActorContext).executeOperationAsync(
1093             any(ActorSelection.class), eqDataExists());
1094
1095         boolean exists = transactionProxy.exists(TestModel.TEST_PATH).checkedGet();
1096
1097         assertEquals("Exists response", true, exists);
1098     }
1099
1100     @Test
1101     public void testLocalTxActorWrite() throws Exception {
1102         ActorSystem actorSystem = getSystem();
1103         ActorRef shardActorRef = actorSystem.actorOf(Props.create(DoNothingActor.class));
1104
1105         doReturn(actorSystem.actorSelection(shardActorRef.path())).
1106             when(mockActorContext).actorSelection(shardActorRef.path().toString());
1107
1108         doReturn(Futures.successful(actorSystem.actorSelection(shardActorRef.path()))).
1109             when(mockActorContext).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD));
1110
1111         String actorPath = "akka.tcp://system@127.0.0.1:2550/user/tx-actor";
1112         CreateTransactionReply createTransactionReply = CreateTransactionReply.newBuilder()
1113             .setTransactionId("txn-1")
1114             .setTransactionActorPath(actorPath)
1115             .build();
1116
1117         doReturn(Futures.successful(createTransactionReply)).when(mockActorContext).
1118         executeOperationAsync(eq(actorSystem.actorSelection(shardActorRef.path())),
1119                 eqCreateTransaction(memberName, WRITE_ONLY));
1120
1121         doReturn(true).when(mockActorContext).isPathLocal(actorPath);
1122
1123         NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1124
1125         doReturn(writeDataReply()).when(mockActorContext).executeOperationAsync(
1126             any(ActorSelection.class), eqWriteData(nodeToWrite));
1127
1128         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY);
1129         transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
1130
1131         verify(mockActorContext).executeOperationAsync(
1132             any(ActorSelection.class), eqWriteData(nodeToWrite));
1133
1134         //testing local merge
1135         doReturn(mergeDataReply()).when(mockActorContext).executeOperationAsync(
1136             any(ActorSelection.class), eqMergeData(nodeToWrite));
1137
1138         transactionProxy.merge(TestModel.TEST_PATH, nodeToWrite);
1139
1140         verify(mockActorContext).executeOperationAsync(
1141             any(ActorSelection.class), eqMergeData(nodeToWrite));
1142
1143
1144         //testing local delete
1145         doReturn(deleteDataReply()).when(mockActorContext).executeOperationAsync(
1146             any(ActorSelection.class), eqDeleteData());
1147
1148         transactionProxy.delete(TestModel.TEST_PATH);
1149
1150         verify(mockActorContext).executeOperationAsync(any(ActorSelection.class), eqDeleteData());
1151
1152         verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(),
1153             WriteDataReply.class, MergeDataReply.class, DeleteDataReply.class);
1154
1155         // testing ready
1156         doReturn(readyTxReply(shardActorRef.path().toString())).when(mockActorContext).executeOperationAsync(
1157             any(ActorSelection.class), isA(ReadyTransaction.class));
1158
1159         DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
1160
1161         assertTrue(ready instanceof ThreePhaseCommitCohortProxy);
1162
1163         ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready;
1164
1165         verifyCohortFutures(proxy, getSystem().actorSelection(shardActorRef.path()));
1166     }
1167 }

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