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