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){
 342         return CreateTransactionReply.newBuilder()
 343             .setTransactionActorPath(actorRef.path().toString())
 344             .setTransactionId("txn-1").build();
 345     }
 346
 347     private ActorRef setupActorContextWithInitialCreateTransaction(ActorSystem actorSystem, TransactionType type) {
 348         ActorRef actorRef = actorSystem.actorOf(Props.create(DoNothingActor.class));
 349         doReturn(actorSystem.actorSelection(actorRef.path())).
 350                 when(mockActorContext).actorSelection(actorRef.path().toString());
 351
 352         doReturn(Futures.successful(actorSystem.actorSelection(actorRef.path()))).
 353                 when(mockActorContext).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD));
 354
 355         doReturn(Futures.successful(createTransactionReply(actorRef))).when(mockActorContext).
 356                 executeOperationAsync(eq(actorSystem.actorSelection(actorRef.path())),
 357                         eqCreateTransaction(memberName, type));
 358
 359         doReturn(false).when(mockActorContext).isLocalPath(actorRef.path().toString());
 360
 361         return actorRef;
 362     }
 363
 364     private void propagateReadFailedExceptionCause(CheckedFuture<?, ReadFailedException> future)
 365             throws Throwable {
 366
 367         try {
 368             future.checkedGet(5, TimeUnit.SECONDS);
 369             fail("Expected ReadFailedException");
 370         } catch(ReadFailedException e) {
 371             e.printStackTrace();
 372             throw e.getCause();
 373         }
 374     }
 375
 376     @Test
 377     public void testRead() throws Exception {
 378         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY);
 379
 380         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,
 381                 READ_ONLY);
 382
 383         doReturn(readSerializedDataReply(null)).when(mockActorContext).executeOperationAsync(
 384                 eq(actorSelection(actorRef)), eqSerializedReadData());
 385
 386         Optional<NormalizedNode<?, ?>> readOptional = transactionProxy.read(
 387                 TestModel.TEST_PATH).get(5, TimeUnit.SECONDS);
 388
 389         assertEquals("NormalizedNode isPresent", false, readOptional.isPresent());
 390
 391         NormalizedNode<?, ?> expectedNode = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 392
 393         doReturn(readSerializedDataReply(expectedNode)).when(mockActorContext).executeOperationAsync(
 394                 eq(actorSelection(actorRef)), eqSerializedReadData());
 395
 396         readOptional = transactionProxy.read(TestModel.TEST_PATH).get(5, TimeUnit.SECONDS);
 397
 398         assertEquals("NormalizedNode isPresent", true, readOptional.isPresent());
 399
 400         assertEquals("Response NormalizedNode", expectedNode, readOptional.get());
 401     }
 402
 403     @Test(expected = ReadFailedException.class)
 404     public void testReadWithInvalidReplyMessageType() throws Exception {
 405         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY);
 406
 407         doReturn(Futures.successful(new Object())).when(mockActorContext).
 408                 executeOperationAsync(eq(actorSelection(actorRef)), eqSerializedReadData());
 409
 410         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,
 411                 READ_ONLY);
 412
 413         transactionProxy.read(TestModel.TEST_PATH).checkedGet(5, TimeUnit.SECONDS);
 414     }
 415
 416     @Test(expected = TestException.class)
 417     public void testReadWithAsyncRemoteOperatonFailure() throws Throwable {
 418         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY);
 419
 420         doReturn(Futures.failed(new TestException())).when(mockActorContext).
 421                 executeOperationAsync(eq(actorSelection(actorRef)), eqSerializedReadData());
 422
 423         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,
 424                 READ_ONLY);
 425
 426         propagateReadFailedExceptionCause(transactionProxy.read(TestModel.TEST_PATH));
 427     }
 428
 429     private void testExceptionOnInitialCreateTransaction(Exception exToThrow, Invoker invoker)
 430             throws Throwable {
 431         ActorRef actorRef = getSystem().actorOf(Props.create(DoNothingActor.class));
 432
 433         if (exToThrow instanceof PrimaryNotFoundException) {
 434             doReturn(Futures.failed(exToThrow)).when(mockActorContext).findPrimaryShardAsync(anyString());
 435         } else {
 436             doReturn(Futures.successful(getSystem().actorSelection(actorRef.path()))).
 437                     when(mockActorContext).findPrimaryShardAsync(anyString());
 438         }
 439
 440         doReturn(Futures.failed(exToThrow)).when(mockActorContext).executeOperationAsync(
 441                 any(ActorSelection.class), any());
 442
 443         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_ONLY);
 444
 445         propagateReadFailedExceptionCause(invoker.invoke(transactionProxy));
 446     }
 447
 448     private void testReadWithExceptionOnInitialCreateTransaction(Exception exToThrow) throws Throwable {
 449         testExceptionOnInitialCreateTransaction(exToThrow, new Invoker() {
 450             @Override
 451             public CheckedFuture<?, ReadFailedException> invoke(TransactionProxy proxy) throws Exception {
 452                 return proxy.read(TestModel.TEST_PATH);
 453             }
 454         });
 455     }
 456
 457     @Test(expected = PrimaryNotFoundException.class)
 458     public void testReadWhenAPrimaryNotFoundExceptionIsThrown() throws Throwable {
 459         testReadWithExceptionOnInitialCreateTransaction(new PrimaryNotFoundException("test"));
 460     }
 461
 462     @Test(expected = TimeoutException.class)
 463     public void testReadWhenATimeoutExceptionIsThrown() throws Throwable {
 464         testReadWithExceptionOnInitialCreateTransaction(new TimeoutException("test",
 465                 new Exception("reason")));
 466     }
 467
 468     @Test(expected = TestException.class)
 469     public void testReadWhenAnyOtherExceptionIsThrown() throws Throwable {
 470         testReadWithExceptionOnInitialCreateTransaction(new TestException());
 471     }
 472
 473     @Test(expected = TestException.class)
 474     public void testReadWithPriorRecordingOperationFailure() throws Throwable {
 475         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE);
 476
 477         NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 478
 479         doReturn(writeSerializedDataReply()).when(mockActorContext).executeOperationAsync(
 480                 eq(actorSelection(actorRef)), eqSerializedWriteData(nodeToWrite));
 481
 482         doReturn(Futures.failed(new TestException())).when(mockActorContext).
 483                 executeOperationAsync(eq(actorSelection(actorRef)), eqSerializedDeleteData());
 484
 485         doReturn(readSerializedDataReply(null)).when(mockActorContext).executeOperationAsync(
 486                 eq(actorSelection(actorRef)), eqSerializedReadData());
 487
 488         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,
 489                 READ_WRITE);
 490
 491         transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
 492
 493         transactionProxy.delete(TestModel.TEST_PATH);
 494
 495         try {
 496             propagateReadFailedExceptionCause(transactionProxy.read(TestModel.TEST_PATH));
 497         } finally {
 498             verify(mockActorContext, times(0)).executeOperationAsync(
 499                     eq(actorSelection(actorRef)), eqSerializedReadData());
 500         }
 501     }
 502
 503     @Test
 504     public void testReadWithPriorRecordingOperationSuccessful() throws Throwable {
 505         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE);
 506
 507         NormalizedNode<?, ?> expectedNode = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 508
 509         doReturn(writeSerializedDataReply()).when(mockActorContext).executeOperationAsync(
 510                 eq(actorSelection(actorRef)), eqSerializedWriteData(expectedNode));
 511
 512         doReturn(readSerializedDataReply(expectedNode)).when(mockActorContext).executeOperationAsync(
 513                 eq(actorSelection(actorRef)), eqSerializedReadData());
 514
 515         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,
 516                 READ_WRITE);
 517
 518         transactionProxy.write(TestModel.TEST_PATH, expectedNode);
 519
 520         Optional<NormalizedNode<?, ?>> readOptional = transactionProxy.read(
 521                 TestModel.TEST_PATH).get(5, TimeUnit.SECONDS);
 522
 523         assertEquals("NormalizedNode isPresent", true, readOptional.isPresent());
 524
 525         assertEquals("Response NormalizedNode", expectedNode, readOptional.get());
 526     }
 527
 528     @Test(expected=IllegalStateException.class)
 529     public void testReadPreConditionCheck() {
 530
 531         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,
 532                 WRITE_ONLY);
 533
 534         transactionProxy.read(TestModel.TEST_PATH);
 535     }
 536
 537     @Test(expected=IllegalArgumentException.class)
 538     public void testInvalidCreateTransactionReply() throws Throwable {
 539         ActorRef actorRef = getSystem().actorOf(Props.create(DoNothingActor.class));
 540
 541         doReturn(getSystem().actorSelection(actorRef.path())).when(mockActorContext).
 542             actorSelection(actorRef.path().toString());
 543
 544         doReturn(Futures.successful(getSystem().actorSelection(actorRef.path()))).
 545             when(mockActorContext).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD));
 546
 547         doReturn(Futures.successful(new Object())).when(mockActorContext).executeOperationAsync(
 548             eq(getSystem().actorSelection(actorRef.path())), eqCreateTransaction(memberName, READ_ONLY));
 549
 550         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_ONLY);
 551
 552         propagateReadFailedExceptionCause(transactionProxy.read(TestModel.TEST_PATH));
 553     }
 554
 555     @Test
 556     public void testExists() throws Exception {
 557         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY);
 558
 559         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,
 560                 READ_ONLY);
 561
 562         doReturn(dataExistsSerializedReply(false)).when(mockActorContext).executeOperationAsync(
 563                 eq(actorSelection(actorRef)), eqSerializedDataExists());
 564
 565         Boolean exists = transactionProxy.exists(TestModel.TEST_PATH).checkedGet();
 566
 567         assertEquals("Exists response", false, exists);
 568
 569         doReturn(dataExistsSerializedReply(true)).when(mockActorContext).executeOperationAsync(
 570                 eq(actorSelection(actorRef)), eqSerializedDataExists());
 571
 572         exists = transactionProxy.exists(TestModel.TEST_PATH).checkedGet();
 573
 574         assertEquals("Exists response", true, exists);
 575     }
 576
 577     @Test(expected = PrimaryNotFoundException.class)
 578     public void testExistsWhenAPrimaryNotFoundExceptionIsThrown() throws Throwable {
 579         testExceptionOnInitialCreateTransaction(new PrimaryNotFoundException("test"), new Invoker() {
 580             @Override
 581             public CheckedFuture<?, ReadFailedException> invoke(TransactionProxy proxy) throws Exception {
 582                 return proxy.exists(TestModel.TEST_PATH);
 583             }
 584         });
 585     }
 586
 587     @Test(expected = ReadFailedException.class)
 588     public void testExistsWithInvalidReplyMessageType() throws Exception {
 589         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY);
 590
 591         doReturn(Futures.successful(new Object())).when(mockActorContext).
 592                 executeOperationAsync(eq(actorSelection(actorRef)), eqSerializedDataExists());
 593
 594         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,
 595                 READ_ONLY);
 596
 597         transactionProxy.exists(TestModel.TEST_PATH).checkedGet(5, TimeUnit.SECONDS);
 598     }
 599
 600     @Test(expected = TestException.class)
 601     public void testExistsWithAsyncRemoteOperatonFailure() throws Throwable {
 602         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY);
 603
 604         doReturn(Futures.failed(new TestException())).when(mockActorContext).
 605                 executeOperationAsync(eq(actorSelection(actorRef)), eqSerializedDataExists());
 606
 607         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,
 608                 READ_ONLY);
 609
 610         propagateReadFailedExceptionCause(transactionProxy.exists(TestModel.TEST_PATH));
 611     }
 612
 613     @Test(expected = TestException.class)
 614     public void testExistsWithPriorRecordingOperationFailure() throws Throwable {
 615         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE);
 616
 617         NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 618
 619         doReturn(writeSerializedDataReply()).when(mockActorContext).executeOperationAsync(
 620                 eq(actorSelection(actorRef)), eqSerializedWriteData(nodeToWrite));
 621
 622         doReturn(Futures.failed(new TestException())).when(mockActorContext).
 623                 executeOperationAsync(eq(actorSelection(actorRef)), eqSerializedDeleteData());
 624
 625         doReturn(dataExistsSerializedReply(false)).when(mockActorContext).executeOperationAsync(
 626                 eq(actorSelection(actorRef)), eqSerializedDataExists());
 627
 628         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,
 629                 READ_WRITE);
 630
 631         transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
 632
 633         transactionProxy.delete(TestModel.TEST_PATH);
 634
 635         try {
 636             propagateReadFailedExceptionCause(transactionProxy.exists(TestModel.TEST_PATH));
 637         } finally {
 638             verify(mockActorContext, times(0)).executeOperationAsync(
 639                     eq(actorSelection(actorRef)), eqSerializedDataExists());
 640         }
 641     }
 642
 643     @Test
 644     public void testExistsWithPriorRecordingOperationSuccessful() throws Throwable {
 645         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE);
 646
 647         NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 648
 649         doReturn(writeSerializedDataReply()).when(mockActorContext).executeOperationAsync(
 650                 eq(actorSelection(actorRef)), eqSerializedWriteData(nodeToWrite));
 651
 652         doReturn(dataExistsSerializedReply(true)).when(mockActorContext).executeOperationAsync(
 653                 eq(actorSelection(actorRef)), eqSerializedDataExists());
 654
 655         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,
 656                 READ_WRITE);
 657
 658         transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
 659
 660         Boolean exists = transactionProxy.exists(TestModel.TEST_PATH).checkedGet();
 661
 662         assertEquals("Exists response", true, exists);
 663     }
 664
 665     @Test(expected=IllegalStateException.class)
 666     public void testxistsPreConditionCheck() {
 667
 668         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,
 669                 WRITE_ONLY);
 670
 671         transactionProxy.exists(TestModel.TEST_PATH);
 672     }
 673
 674     private void verifyRecordingOperationFutures(List<Future<Object>> futures,
 675             Class<?>... expResultTypes) throws Exception {
 676         assertEquals("getRecordingOperationFutures size", expResultTypes.length, futures.size());
 677
 678         int i = 0;
 679         for( Future<Object> future: futures) {
 680             assertNotNull("Recording operation Future is null", future);
 681
 682             Class<?> expResultType = expResultTypes[i++];
 683             if(Throwable.class.isAssignableFrom(expResultType)) {
 684                 try {
 685                     Await.result(future, Duration.create(5, TimeUnit.SECONDS));
 686                     fail("Expected exception from recording operation Future");
 687                 } catch(Exception e) {
 688                     // Expected
 689                 }
 690             } else {
 691                 assertEquals("Recording operation Future result type", expResultType,
 692                              Await.result(future, Duration.create(5, TimeUnit.SECONDS)).getClass());
 693             }
 694         }
 695     }
 696
 697     @Test
 698     public void testWrite() throws Exception {
 699         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY);
 700
 701         NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 702
 703         doReturn(writeSerializedDataReply()).when(mockActorContext).executeOperationAsync(
 704                 eq(actorSelection(actorRef)), eqSerializedWriteData(nodeToWrite));
 705
 706         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,
 707                 WRITE_ONLY);
 708
 709         transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
 710
 711         verify(mockActorContext).executeOperationAsync(
 712                 eq(actorSelection(actorRef)), eqSerializedWriteData(nodeToWrite));
 713
 714         verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(),
 715                 WriteDataReply.SERIALIZABLE_CLASS);
 716     }
 717
 718     @Test(expected=IllegalStateException.class)
 719     public void testWritePreConditionCheck() {
 720
 721         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,
 722                 READ_ONLY);
 723
 724         transactionProxy.write(TestModel.TEST_PATH,
 725                 ImmutableNodes.containerNode(TestModel.TEST_QNAME));
 726     }
 727
 728     @Test(expected=IllegalStateException.class)
 729     public void testWriteAfterReadyPreConditionCheck() {
 730
 731         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,
 732                 WRITE_ONLY);
 733
 734         transactionProxy.ready();
 735
 736         transactionProxy.write(TestModel.TEST_PATH,
 737                 ImmutableNodes.containerNode(TestModel.TEST_QNAME));
 738     }
 739
 740     @Test
 741     public void testMerge() throws Exception {
 742         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY);
 743
 744         NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 745
 746         doReturn(mergeSerializedDataReply()).when(mockActorContext).executeOperationAsync(
 747                 eq(actorSelection(actorRef)), eqSerializedMergeData(nodeToWrite));
 748
 749         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY);
 750
 751         transactionProxy.merge(TestModel.TEST_PATH, nodeToWrite);
 752
 753         verify(mockActorContext).executeOperationAsync(
 754                 eq(actorSelection(actorRef)), eqSerializedMergeData(nodeToWrite));
 755
 756         verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(),
 757                 MergeDataReply.SERIALIZABLE_CLASS);
 758     }
 759
 760     @Test
 761     public void testDelete() throws Exception {
 762         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY);
 763
 764         doReturn(deleteSerializedDataReply()).when(mockActorContext).executeOperationAsync(
 765                 eq(actorSelection(actorRef)), eqSerializedDeleteData());
 766
 767         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,
 768                 WRITE_ONLY);
 769
 770         transactionProxy.delete(TestModel.TEST_PATH);
 771
 772         verify(mockActorContext).executeOperationAsync(
 773                 eq(actorSelection(actorRef)), eqSerializedDeleteData());
 774
 775         verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(),
 776                 DeleteDataReply.SERIALIZABLE_CLASS);
 777     }
 778
 779     private void verifyCohortFutures(ThreePhaseCommitCohortProxy proxy,
 780         Object... expReplies) throws Exception {
 781         assertEquals("getReadyOperationFutures size", expReplies.length,
 782                 proxy.getCohortFutures().size());
 783
 784         int i = 0;
 785         for( Future<ActorSelection> future: proxy.getCohortFutures()) {
 786             assertNotNull("Ready operation Future is null", future);
 787
 788             Object expReply = expReplies[i++];
 789             if(expReply instanceof ActorSelection) {
 790                 ActorSelection actual = Await.result(future, Duration.create(5, TimeUnit.SECONDS));
 791                 assertEquals("Cohort actor path", expReply, actual);
 792             } else {
 793                 // Expecting exception.
 794                 try {
 795                     Await.result(future, Duration.create(5, TimeUnit.SECONDS));
 796                     fail("Expected exception from ready operation Future");
 797                 } catch(Exception e) {
 798                     // Expected
 799                 }
 800             }
 801         }
 802     }
 803
 804     @SuppressWarnings("unchecked")
 805     @Test
 806     public void testReady() throws Exception {
 807         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE);
 808
 809         NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 810
 811         doReturn(readSerializedDataReply(null)).when(mockActorContext).executeOperationAsync(
 812                 eq(actorSelection(actorRef)), eqSerializedReadData());
 813
 814         doReturn(writeSerializedDataReply()).when(mockActorContext).executeOperationAsync(
 815                 eq(actorSelection(actorRef)), eqSerializedWriteData(nodeToWrite));
 816
 817         doReturn(readySerializedTxReply(actorRef.path().toString())).when(mockActorContext).executeOperationAsync(
 818                 eq(actorSelection(actorRef)), isA(ReadyTransaction.SERIALIZABLE_CLASS));
 819
 820         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,
 821                 READ_WRITE);
 822
 823         transactionProxy.read(TestModel.TEST_PATH);
 824
 825         transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
 826
 827         DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
 828
 829         assertTrue(ready instanceof ThreePhaseCommitCohortProxy);
 830
 831         ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready;
 832
 833         verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(),
 834                 WriteDataReply.SERIALIZABLE_CLASS);
 835
 836         verifyCohortFutures(proxy, getSystem().actorSelection(actorRef.path()));
 837     }
 838
 839     @SuppressWarnings("unchecked")
 840     @Test
 841     public void testReadyWithRecordingOperationFailure() throws Exception {
 842         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY);
 843
 844         NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 845
 846         doReturn(mergeSerializedDataReply()).when(mockActorContext).executeOperationAsync(
 847                 eq(actorSelection(actorRef)), eqSerializedMergeData(nodeToWrite));
 848
 849         doReturn(Futures.failed(new TestException())).when(mockActorContext).
 850                 executeOperationAsync(eq(actorSelection(actorRef)), eqSerializedWriteData(nodeToWrite));
 851
 852         doReturn(readySerializedTxReply(actorRef.path().toString())).when(mockActorContext).executeOperationAsync(
 853                 eq(actorSelection(actorRef)), isA(ReadyTransaction.SERIALIZABLE_CLASS));
 854
 855         doReturn(false).when(mockActorContext).isLocalPath(actorRef.path().toString());
 856
 857         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,
 858                 WRITE_ONLY);
 859
 860         transactionProxy.merge(TestModel.TEST_PATH, nodeToWrite);
 861
 862         transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
 863
 864         DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
 865
 866         assertTrue(ready instanceof ThreePhaseCommitCohortProxy);
 867
 868         ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready;
 869
 870         verifyCohortFutures(proxy, TestException.class);
 871
 872         verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(),
 873                 MergeDataReply.SERIALIZABLE_CLASS, TestException.class);
 874     }
 875
 876     @SuppressWarnings("unchecked")
 877     @Test
 878     public void testReadyWithReplyFailure() throws Exception {
 879         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY);
 880
 881         NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 882
 883         doReturn(mergeSerializedDataReply()).when(mockActorContext).executeOperationAsync(
 884                 eq(actorSelection(actorRef)), eqSerializedMergeData(nodeToWrite));
 885
 886         doReturn(Futures.failed(new TestException())).when(mockActorContext).
 887                 executeOperationAsync(eq(actorSelection(actorRef)),
 888                         isA(ReadyTransaction.SERIALIZABLE_CLASS));
 889
 890         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,
 891                 WRITE_ONLY);
 892
 893         transactionProxy.merge(TestModel.TEST_PATH, nodeToWrite);
 894
 895         DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
 896
 897         assertTrue(ready instanceof ThreePhaseCommitCohortProxy);
 898
 899         ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready;
 900
 901         verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(),
 902                 MergeDataReply.SERIALIZABLE_CLASS);
 903
 904         verifyCohortFutures(proxy, TestException.class);
 905     }
 906
 907     @Test
 908     public void testReadyWithInitialCreateTransactionFailure() throws Exception {
 909
 910         doReturn(Futures.failed(new PrimaryNotFoundException("mock"))).when(
 911                 mockActorContext).findPrimaryShardAsync(anyString());
 912
 913         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,
 914                 WRITE_ONLY);
 915
 916         NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 917
 918         transactionProxy.merge(TestModel.TEST_PATH, nodeToWrite);
 919
 920         transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
 921
 922         transactionProxy.delete(TestModel.TEST_PATH);
 923
 924         DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
 925
 926         assertTrue(ready instanceof ThreePhaseCommitCohortProxy);
 927
 928         ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready;
 929
 930         verifyCohortFutures(proxy, PrimaryNotFoundException.class);
 931     }
 932
 933     @SuppressWarnings("unchecked")
 934     @Test
 935     public void testReadyWithInvalidReplyMessageType() throws Exception {
 936         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY);
 937
 938         NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 939
 940         doReturn(writeSerializedDataReply()).when(mockActorContext).executeOperationAsync(
 941                 eq(actorSelection(actorRef)), eqSerializedWriteData(nodeToWrite));
 942
 943         doReturn(Futures.successful(new Object())).when(mockActorContext).
 944                 executeOperationAsync(eq(actorSelection(actorRef)),
 945                         isA(ReadyTransaction.SERIALIZABLE_CLASS));
 946
 947         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,
 948                 WRITE_ONLY);
 949
 950         transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
 951
 952         DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
 953
 954         assertTrue(ready instanceof ThreePhaseCommitCohortProxy);
 955
 956         ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready;
 957
 958         verifyCohortFutures(proxy, IllegalArgumentException.class);
 959     }
 960
 961     @Test
 962     public void testGetIdentifier() {
 963         setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY);
 964         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,
 965                 TransactionProxy.TransactionType.READ_ONLY);
 966
 967         Object id = transactionProxy.getIdentifier();
 968         assertNotNull("getIdentifier returned null", id);
 969         assertTrue("Invalid identifier: " + id, id.toString().startsWith(memberName));
 970     }
 971
 972     @SuppressWarnings("unchecked")
 973     @Test
 974     public void testClose() throws Exception{
 975         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE);
 976
 977         doReturn(readSerializedDataReply(null)).when(mockActorContext).executeOperationAsync(
 978                 eq(actorSelection(actorRef)), eqSerializedReadData());
 979
 980         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,
 981                 READ_WRITE);
 982
 983         transactionProxy.read(TestModel.TEST_PATH);
 984
 985         transactionProxy.close();
 986
 987         verify(mockActorContext).sendOperationAsync(
 988                 eq(actorSelection(actorRef)), isA(CloseTransaction.SERIALIZABLE_CLASS));
 989     }
 990
 991
 992     /**
 993      * Method to test a local Tx actor. The Tx paths are matched to decide if the
 994      * Tx actor is local or not. This is done by mocking the Tx actor path
 995      * and the caller paths and ensuring that the paths have the remote-address format
 996      *
 997      * Note: Since the default akka provider for test is not a RemoteActorRefProvider,
 998      * the paths returned for the actors for all the tests are not qualified remote paths.
 999      * Hence are treated as non-local/remote actors. In short, all tests except
1000      * few below run for remote actors
1001      *
1002      * @throws Exception
1003      */
1004     @Test
1005     public void testLocalTxActorRead() throws Exception {
1006         ActorSystem actorSystem = getSystem();
1007         ActorRef shardActorRef = actorSystem.actorOf(Props.create(DoNothingActor.class));
1008
1009         doReturn(actorSystem.actorSelection(shardActorRef.path())).
1010             when(mockActorContext).actorSelection(shardActorRef.path().toString());
1011
1012         doReturn(Futures.successful(actorSystem.actorSelection(shardActorRef.path()))).
1013             when(mockActorContext).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD));
1014
1015         String actorPath = "akka.tcp://system@127.0.0.1:2550/user/tx-actor";
1016         CreateTransactionReply createTransactionReply = CreateTransactionReply.newBuilder()
1017             .setTransactionId("txn-1")
1018             .setTransactionActorPath(actorPath)
1019             .build();
1020
1021         doReturn(Futures.successful(createTransactionReply)).when(mockActorContext).
1022             executeOperationAsync(eq(actorSystem.actorSelection(shardActorRef.path())),
1023                 eqCreateTransaction(memberName, READ_ONLY));
1024
1025         doReturn(true).when(mockActorContext).isLocalPath(actorPath);
1026
1027         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,READ_ONLY);
1028
1029         // negative test case with null as the reply
1030         doReturn(readDataReply(null)).when(mockActorContext).executeOperationAsync(
1031             any(ActorSelection.class), eqReadData());
1032
1033         Optional<NormalizedNode<?, ?>> readOptional = transactionProxy.read(
1034             TestModel.TEST_PATH).get(5, TimeUnit.SECONDS);
1035
1036         assertEquals("NormalizedNode isPresent", false, readOptional.isPresent());
1037
1038         // test case with node as read data reply
1039         NormalizedNode<?, ?> expectedNode = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1040
1041         doReturn(readDataReply(expectedNode)).when(mockActorContext).executeOperationAsync(
1042             any(ActorSelection.class), eqReadData());
1043
1044         readOptional = transactionProxy.read(TestModel.TEST_PATH).get(5, TimeUnit.SECONDS);
1045
1046         assertEquals("NormalizedNode isPresent", true, readOptional.isPresent());
1047
1048         assertEquals("Response NormalizedNode", expectedNode, readOptional.get());
1049
1050         // test for local data exists
1051         doReturn(dataExistsReply(true)).when(mockActorContext).executeOperationAsync(
1052             any(ActorSelection.class), eqDataExists());
1053
1054         boolean exists = transactionProxy.exists(TestModel.TEST_PATH).checkedGet();
1055
1056         assertEquals("Exists response", true, exists);
1057     }
1058
1059     @Test
1060     public void testLocalTxActorWrite() throws Exception {
1061         ActorSystem actorSystem = getSystem();
1062         ActorRef shardActorRef = actorSystem.actorOf(Props.create(DoNothingActor.class));
1063
1064         doReturn(actorSystem.actorSelection(shardActorRef.path())).
1065             when(mockActorContext).actorSelection(shardActorRef.path().toString());
1066
1067         doReturn(Futures.successful(actorSystem.actorSelection(shardActorRef.path()))).
1068             when(mockActorContext).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD));
1069
1070         String actorPath = "akka.tcp://system@127.0.0.1:2550/user/tx-actor";
1071         CreateTransactionReply createTransactionReply = CreateTransactionReply.newBuilder()
1072             .setTransactionId("txn-1")
1073             .setTransactionActorPath(actorPath)
1074             .build();
1075
1076         doReturn(Futures.successful(createTransactionReply)).when(mockActorContext).
1077         executeOperationAsync(eq(actorSystem.actorSelection(shardActorRef.path())),
1078                 eqCreateTransaction(memberName, WRITE_ONLY));
1079
1080         doReturn(true).when(mockActorContext).isLocalPath(actorPath);
1081
1082         NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1083
1084         doReturn(writeDataReply()).when(mockActorContext).executeOperationAsync(
1085             any(ActorSelection.class), eqWriteData(nodeToWrite));
1086
1087         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY);
1088         transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
1089
1090         verify(mockActorContext).executeOperationAsync(
1091             any(ActorSelection.class), eqWriteData(nodeToWrite));
1092
1093         //testing local merge
1094         doReturn(mergeDataReply()).when(mockActorContext).executeOperationAsync(
1095             any(ActorSelection.class), eqMergeData(nodeToWrite));
1096
1097         transactionProxy.merge(TestModel.TEST_PATH, nodeToWrite);
1098
1099         verify(mockActorContext).executeOperationAsync(
1100             any(ActorSelection.class), eqMergeData(nodeToWrite));
1101
1102
1103         //testing local delete
1104         doReturn(deleteDataReply()).when(mockActorContext).executeOperationAsync(
1105             any(ActorSelection.class), eqDeleteData());
1106
1107         transactionProxy.delete(TestModel.TEST_PATH);
1108
1109         verify(mockActorContext).executeOperationAsync(any(ActorSelection.class), eqDeleteData());
1110
1111         verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(),
1112             WriteDataReply.class, MergeDataReply.class, DeleteDataReply.class);
1113
1114         // testing ready
1115         doReturn(readyTxReply(shardActorRef.path().toString())).when(mockActorContext).executeOperationAsync(
1116             any(ActorSelection.class), isA(ReadyTransaction.class));
1117
1118         DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
1119
1120         assertTrue(ready instanceof ThreePhaseCommitCohortProxy);
1121
1122         ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready;
1123
1124         verifyCohortFutures(proxy, getSystem().actorSelection(shardActorRef.path()));
1125     }
1126 }