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 static org.junit.Assert.assertEquals;
   4 import static org.junit.Assert.assertNotNull;
   5 import static org.junit.Assert.assertTrue;
   6 import static org.mockito.Matchers.any;
   7 import static org.mockito.Matchers.anyString;
   8 import static org.mockito.Matchers.eq;
   9 import static org.mockito.Matchers.isA;
  10 import static org.mockito.Mockito.doReturn;
  11 import static org.mockito.Mockito.mock;
  12 import static org.mockito.Mockito.never;
  13 import static org.mockito.Mockito.verify;
  14 import static org.opendaylight.controller.cluster.datastore.TransactionType.READ_ONLY;
  15 import static org.opendaylight.controller.cluster.datastore.TransactionType.READ_WRITE;
  16 import static org.opendaylight.controller.cluster.datastore.TransactionType.WRITE_ONLY;
  17 import akka.actor.ActorRef;
  18 import akka.actor.ActorSelection;
  19 import akka.actor.ActorSystem;
  20 import akka.actor.Props;
  21 import akka.dispatch.Futures;
  22 import akka.util.Timeout;
  23 import com.google.common.base.Optional;
  24 import com.google.common.collect.Sets;
  25 import com.google.common.util.concurrent.CheckedFuture;
  26 import com.google.common.util.concurrent.FutureCallback;
  27 import com.google.common.util.concurrent.Uninterruptibles;
  28 import java.util.Collection;
  29 import java.util.List;
  30 import java.util.concurrent.CountDownLatch;
  31 import java.util.concurrent.ExecutionException;
  32 import java.util.concurrent.TimeUnit;
  33 import java.util.concurrent.atomic.AtomicReference;
  34 import org.junit.Assert;
  35 import org.junit.Test;
  36 import org.mockito.InOrder;
  37 import org.mockito.Mockito;
  38 import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderException;
  39 import org.opendaylight.controller.cluster.datastore.exceptions.NotInitializedException;
  40 import org.opendaylight.controller.cluster.datastore.exceptions.PrimaryNotFoundException;
  41 import org.opendaylight.controller.cluster.datastore.exceptions.TimeoutException;
  42 import org.opendaylight.controller.cluster.datastore.messages.BatchedModifications;
  43 import org.opendaylight.controller.cluster.datastore.messages.CloseTransaction;
  44 import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
  45 import org.opendaylight.controller.cluster.datastore.messages.PrimaryShardInfo;
  46 import org.opendaylight.controller.cluster.datastore.messages.ReadyTransaction;
  47 import org.opendaylight.controller.cluster.datastore.modification.DeleteModification;
  48 import org.opendaylight.controller.cluster.datastore.modification.MergeModification;
  49 import org.opendaylight.controller.cluster.datastore.modification.WriteModification;
  50 import org.opendaylight.controller.cluster.datastore.shardstrategy.DefaultShardStrategy;
  51 import org.opendaylight.controller.cluster.datastore.utils.DoNothingActor;
  52 import org.opendaylight.controller.cluster.datastore.utils.NormalizedNodeAggregatorTest;
  53 import org.opendaylight.controller.md.cluster.datastore.model.CarsModel;
  54 import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper;
  55 import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
  56 import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
  57 import org.opendaylight.controller.protobuff.messages.transaction.ShardTransactionMessages.CreateTransactionReply;
  58 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
  59 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  60 import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode;
  61 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  62 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree;
  63 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeModification;
  64 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeSnapshot;
  65 import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
  66 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
  67 import scala.concurrent.Promise;
  68
  69 @SuppressWarnings("resource")
  70 public class TransactionProxyTest extends AbstractTransactionProxyTest {
  71
  72     @SuppressWarnings("serial")
  73     static class TestException extends RuntimeException {
  74     }
  75
  76     static interface Invoker {
  77         CheckedFuture<?, ReadFailedException> invoke(TransactionProxy proxy) throws Exception;
  78     }
  79
  80     @Test
  81     public void testRead() throws Exception {
  82         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY);
  83
  84         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY);
  85
  86         doReturn(readSerializedDataReply(null)).when(mockActorContext).executeOperationAsync(
  87                 eq(actorSelection(actorRef)), eqSerializedReadData());
  88
  89         Optional<NormalizedNode<?, ?>> readOptional = transactionProxy.read(
  90                 TestModel.TEST_PATH).get(5, TimeUnit.SECONDS);
  91
  92         assertEquals("NormalizedNode isPresent", false, readOptional.isPresent());
  93
  94         NormalizedNode<?, ?> expectedNode = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
  95
  96         doReturn(readSerializedDataReply(expectedNode)).when(mockActorContext).executeOperationAsync(
  97                 eq(actorSelection(actorRef)), eqSerializedReadData());
  98
  99         readOptional = transactionProxy.read(TestModel.TEST_PATH).get(5, TimeUnit.SECONDS);
 100
 101         assertEquals("NormalizedNode isPresent", true, readOptional.isPresent());
 102
 103         assertEquals("Response NormalizedNode", expectedNode, readOptional.get());
 104     }
 105
 106     @Test(expected = ReadFailedException.class)
 107     public void testReadWithInvalidReplyMessageType() throws Exception {
 108         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY);
 109
 110         doReturn(Futures.successful(new Object())).when(mockActorContext).
 111                 executeOperationAsync(eq(actorSelection(actorRef)), eqSerializedReadData());
 112
 113         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY);
 114
 115         transactionProxy.read(TestModel.TEST_PATH).checkedGet(5, TimeUnit.SECONDS);
 116     }
 117
 118     @Test(expected = TestException.class)
 119     public void testReadWithAsyncRemoteOperatonFailure() throws Throwable {
 120         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY);
 121
 122         doReturn(Futures.failed(new TestException())).when(mockActorContext).
 123                 executeOperationAsync(eq(actorSelection(actorRef)), eqSerializedReadData());
 124
 125         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY);
 126
 127         propagateReadFailedExceptionCause(transactionProxy.read(TestModel.TEST_PATH));
 128     }
 129
 130     private void testExceptionOnInitialCreateTransaction(Exception exToThrow, Invoker invoker)
 131             throws Throwable {
 132         ActorRef actorRef = getSystem().actorOf(Props.create(DoNothingActor.class));
 133
 134         if (exToThrow instanceof PrimaryNotFoundException) {
 135             doReturn(Futures.failed(exToThrow)).when(mockActorContext).findPrimaryShardAsync(anyString());
 136         } else {
 137             doReturn(primaryShardInfoReply(getSystem(), actorRef)).
 138                     when(mockActorContext).findPrimaryShardAsync(anyString());
 139         }
 140
 141         doReturn(Futures.failed(exToThrow)).when(mockActorContext).executeOperationAsync(
 142                 any(ActorSelection.class), any(), any(Timeout.class));
 143
 144         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY);
 145
 146         propagateReadFailedExceptionCause(invoker.invoke(transactionProxy));
 147     }
 148
 149     private void testReadWithExceptionOnInitialCreateTransaction(Exception exToThrow) throws Throwable {
 150         testExceptionOnInitialCreateTransaction(exToThrow, new Invoker() {
 151             @Override
 152             public CheckedFuture<?, ReadFailedException> invoke(TransactionProxy proxy) throws Exception {
 153                 return proxy.read(TestModel.TEST_PATH);
 154             }
 155         });
 156     }
 157
 158     @Test(expected = PrimaryNotFoundException.class)
 159     public void testReadWhenAPrimaryNotFoundExceptionIsThrown() throws Throwable {
 160         testReadWithExceptionOnInitialCreateTransaction(new PrimaryNotFoundException("test"));
 161     }
 162
 163     @Test(expected = TimeoutException.class)
 164     public void testReadWhenATimeoutExceptionIsThrown() throws Throwable {
 165         testReadWithExceptionOnInitialCreateTransaction(new TimeoutException("test",
 166                 new Exception("reason")));
 167     }
 168
 169     @Test(expected = TestException.class)
 170     public void testReadWhenAnyOtherExceptionIsThrown() throws Throwable {
 171         testReadWithExceptionOnInitialCreateTransaction(new TestException());
 172     }
 173
 174     @Test
 175     public void testReadWithPriorRecordingOperationSuccessful() throws Throwable {
 176         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE);
 177
 178         NormalizedNode<?, ?> expectedNode = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 179
 180         expectBatchedModifications(actorRef, 1);
 181
 182         doReturn(readSerializedDataReply(expectedNode)).when(mockActorContext).executeOperationAsync(
 183                 eq(actorSelection(actorRef)), eqSerializedReadData());
 184
 185         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE);
 186
 187         transactionProxy.write(TestModel.TEST_PATH, expectedNode);
 188
 189         Optional<NormalizedNode<?, ?>> readOptional = transactionProxy.read(
 190                 TestModel.TEST_PATH).get(5, TimeUnit.SECONDS);
 191
 192         assertEquals("NormalizedNode isPresent", true, readOptional.isPresent());
 193         assertEquals("Response NormalizedNode", expectedNode, readOptional.get());
 194
 195         InOrder inOrder = Mockito.inOrder(mockActorContext);
 196         inOrder.verify(mockActorContext).executeOperationAsync(
 197                 eq(actorSelection(actorRef)), isA(BatchedModifications.class));
 198
 199         inOrder.verify(mockActorContext).executeOperationAsync(
 200                 eq(actorSelection(actorRef)), eqSerializedReadData());
 201     }
 202
 203     @Test(expected=IllegalStateException.class)
 204     public void testReadPreConditionCheck() {
 205         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY);
 206         transactionProxy.read(TestModel.TEST_PATH);
 207     }
 208
 209     @Test(expected=IllegalArgumentException.class)
 210     public void testInvalidCreateTransactionReply() throws Throwable {
 211         ActorRef actorRef = getSystem().actorOf(Props.create(DoNothingActor.class));
 212
 213         doReturn(getSystem().actorSelection(actorRef.path())).when(mockActorContext).
 214             actorSelection(actorRef.path().toString());
 215
 216         doReturn(primaryShardInfoReply(getSystem(), actorRef)).
 217             when(mockActorContext).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD));
 218
 219         doReturn(Futures.successful(new Object())).when(mockActorContext).executeOperationAsync(
 220             eq(getSystem().actorSelection(actorRef.path())), eqCreateTransaction(memberName, READ_ONLY),
 221             any(Timeout.class));
 222
 223         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY);
 224
 225         propagateReadFailedExceptionCause(transactionProxy.read(TestModel.TEST_PATH));
 226     }
 227
 228     @Test
 229     public void testExists() throws Exception {
 230         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY);
 231
 232         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY);
 233
 234         doReturn(dataExistsSerializedReply(false)).when(mockActorContext).executeOperationAsync(
 235                 eq(actorSelection(actorRef)), eqSerializedDataExists());
 236
 237         Boolean exists = transactionProxy.exists(TestModel.TEST_PATH).checkedGet();
 238
 239         assertEquals("Exists response", false, exists);
 240
 241         doReturn(dataExistsSerializedReply(true)).when(mockActorContext).executeOperationAsync(
 242                 eq(actorSelection(actorRef)), eqSerializedDataExists());
 243
 244         exists = transactionProxy.exists(TestModel.TEST_PATH).checkedGet();
 245
 246         assertEquals("Exists response", true, exists);
 247     }
 248
 249     @Test(expected = PrimaryNotFoundException.class)
 250     public void testExistsWhenAPrimaryNotFoundExceptionIsThrown() throws Throwable {
 251         testExceptionOnInitialCreateTransaction(new PrimaryNotFoundException("test"), new Invoker() {
 252             @Override
 253             public CheckedFuture<?, ReadFailedException> invoke(TransactionProxy proxy) throws Exception {
 254                 return proxy.exists(TestModel.TEST_PATH);
 255             }
 256         });
 257     }
 258
 259     @Test(expected = ReadFailedException.class)
 260     public void testExistsWithInvalidReplyMessageType() throws Exception {
 261         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY);
 262
 263         doReturn(Futures.successful(new Object())).when(mockActorContext).
 264                 executeOperationAsync(eq(actorSelection(actorRef)), eqSerializedDataExists());
 265
 266         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY);
 267
 268         transactionProxy.exists(TestModel.TEST_PATH).checkedGet(5, TimeUnit.SECONDS);
 269     }
 270
 271     @Test(expected = TestException.class)
 272     public void testExistsWithAsyncRemoteOperatonFailure() throws Throwable {
 273         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY);
 274
 275         doReturn(Futures.failed(new TestException())).when(mockActorContext).
 276                 executeOperationAsync(eq(actorSelection(actorRef)), eqSerializedDataExists());
 277
 278         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY);
 279
 280         propagateReadFailedExceptionCause(transactionProxy.exists(TestModel.TEST_PATH));
 281     }
 282
 283     @Test
 284     public void testExistsWithPriorRecordingOperationSuccessful() throws Throwable {
 285         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE);
 286
 287         NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 288
 289         expectBatchedModifications(actorRef, 1);
 290
 291         doReturn(dataExistsSerializedReply(true)).when(mockActorContext).executeOperationAsync(
 292                 eq(actorSelection(actorRef)), eqSerializedDataExists());
 293
 294         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE);
 295
 296         transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
 297
 298         Boolean exists = transactionProxy.exists(TestModel.TEST_PATH).checkedGet();
 299
 300         assertEquals("Exists response", true, exists);
 301
 302         InOrder inOrder = Mockito.inOrder(mockActorContext);
 303         inOrder.verify(mockActorContext).executeOperationAsync(
 304                 eq(actorSelection(actorRef)), isA(BatchedModifications.class));
 305
 306         inOrder.verify(mockActorContext).executeOperationAsync(
 307                 eq(actorSelection(actorRef)), eqSerializedDataExists());
 308     }
 309
 310     @Test(expected=IllegalStateException.class)
 311     public void testExistsPreConditionCheck() {
 312         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY);
 313         transactionProxy.exists(TestModel.TEST_PATH);
 314     }
 315
 316     @Test
 317     public void testWrite() throws Exception {
 318         dataStoreContextBuilder.shardBatchedModificationCount(1);
 319         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY);
 320
 321         NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 322
 323         expectBatchedModifications(actorRef, 1);
 324
 325         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY);
 326
 327         transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
 328
 329         verifyOneBatchedModification(actorRef, new WriteModification(TestModel.TEST_PATH, nodeToWrite), false);
 330     }
 331
 332     @Test
 333     public void testWriteAfterAsyncRead() throws Throwable {
 334         ActorRef actorRef = setupActorContextWithoutInitialCreateTransaction(getSystem(), DefaultShardStrategy.DEFAULT_SHARD);
 335
 336         Promise<Object> createTxPromise = akka.dispatch.Futures.promise();
 337         doReturn(createTxPromise).when(mockActorContext).executeOperationAsync(
 338                 eq(getSystem().actorSelection(actorRef.path())),
 339                 eqCreateTransaction(memberName, READ_WRITE), any(Timeout.class));
 340
 341         doReturn(readSerializedDataReply(null)).when(mockActorContext).executeOperationAsync(
 342                 eq(actorSelection(actorRef)), eqSerializedReadData());
 343
 344         expectBatchedModificationsReady(actorRef);
 345
 346         final NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 347
 348         final TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE);
 349
 350         final CountDownLatch readComplete = new CountDownLatch(1);
 351         final AtomicReference<Throwable> caughtEx = new AtomicReference<>();
 352         com.google.common.util.concurrent.Futures.addCallback(transactionProxy.read(TestModel.TEST_PATH),
 353                 new  FutureCallback<Optional<NormalizedNode<?, ?>>>() {
 354                     @Override
 355                     public void onSuccess(Optional<NormalizedNode<?, ?>> result) {
 356                         try {
 357                             transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
 358                         } catch (Exception e) {
 359                             caughtEx.set(e);
 360                         } finally {
 361                             readComplete.countDown();
 362                         }
 363                     }
 364
 365                     @Override
 366                     public void onFailure(Throwable t) {
 367                         caughtEx.set(t);
 368                         readComplete.countDown();
 369                     }
 370                 });
 371
 372         createTxPromise.success(createTransactionReply(actorRef, DataStoreVersions.CURRENT_VERSION));
 373
 374         Uninterruptibles.awaitUninterruptibly(readComplete, 5, TimeUnit.SECONDS);
 375
 376         if(caughtEx.get() != null) {
 377             throw caughtEx.get();
 378         }
 379
 380         // This sends the batched modification.
 381         transactionProxy.ready();
 382
 383         verifyOneBatchedModification(actorRef, new WriteModification(TestModel.TEST_PATH, nodeToWrite), true);
 384     }
 385
 386     @Test(expected=IllegalStateException.class)
 387     public void testWritePreConditionCheck() {
 388         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY);
 389         transactionProxy.write(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
 390     }
 391
 392     @Test(expected=IllegalStateException.class)
 393     public void testWriteAfterReadyPreConditionCheck() {
 394         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY);
 395
 396         transactionProxy.ready();
 397
 398         transactionProxy.write(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
 399     }
 400
 401     @Test
 402     public void testMerge() throws Exception {
 403         dataStoreContextBuilder.shardBatchedModificationCount(1);
 404         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY);
 405
 406         NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 407
 408         expectBatchedModifications(actorRef, 1);
 409
 410         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY);
 411
 412         transactionProxy.merge(TestModel.TEST_PATH, nodeToWrite);
 413
 414         verifyOneBatchedModification(actorRef, new MergeModification(TestModel.TEST_PATH, nodeToWrite), false);
 415     }
 416
 417     @Test
 418     public void testDelete() throws Exception {
 419         dataStoreContextBuilder.shardBatchedModificationCount(1);
 420         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY);
 421
 422         expectBatchedModifications(actorRef, 1);
 423
 424         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY);
 425
 426         transactionProxy.delete(TestModel.TEST_PATH);
 427
 428         verifyOneBatchedModification(actorRef, new DeleteModification(TestModel.TEST_PATH), false);
 429     }
 430
 431     @Test
 432     public void testReadWrite() throws Exception {
 433         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE);
 434
 435         NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 436
 437         doReturn(readSerializedDataReply(null)).when(mockActorContext).executeOperationAsync(
 438                 eq(actorSelection(actorRef)), eqSerializedReadData());
 439
 440         expectBatchedModifications(actorRef, 1);
 441
 442         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE);
 443
 444         transactionProxy.read(TestModel.TEST_PATH);
 445
 446         transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
 447
 448         transactionProxy.read(TestModel.TEST_PATH);
 449
 450         transactionProxy.read(TestModel.TEST_PATH);
 451
 452         List<BatchedModifications> batchedModifications = captureBatchedModifications(actorRef);
 453         assertEquals("Captured BatchedModifications count", 1, batchedModifications.size());
 454
 455         verifyBatchedModifications(batchedModifications.get(0), false,
 456                 new WriteModification(TestModel.TEST_PATH, nodeToWrite));
 457     }
 458
 459     @Test
 460     public void testReadyWithReadWrite() throws Exception {
 461         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE);
 462
 463         NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 464
 465         doReturn(readSerializedDataReply(null)).when(mockActorContext).executeOperationAsync(
 466                 eq(actorSelection(actorRef)), eqSerializedReadData());
 467
 468         expectBatchedModificationsReady(actorRef, true);
 469
 470         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE);
 471
 472         transactionProxy.read(TestModel.TEST_PATH);
 473
 474         transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
 475
 476         DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
 477
 478         assertTrue(ready instanceof SingleCommitCohortProxy);
 479
 480         verifyCohortFutures((SingleCommitCohortProxy)ready, new CommitTransactionReply().toSerializable());
 481
 482         List<BatchedModifications> batchedModifications = captureBatchedModifications(actorRef);
 483         assertEquals("Captured BatchedModifications count", 1, batchedModifications.size());
 484
 485         verifyBatchedModifications(batchedModifications.get(0), true, true,
 486                 new WriteModification(TestModel.TEST_PATH, nodeToWrite));
 487
 488         assertEquals("getTotalMessageCount", 1, batchedModifications.get(0).getTotalMessagesSent());
 489     }
 490
 491     @Test
 492     public void testReadyWithNoModifications() throws Exception {
 493         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE);
 494
 495         doReturn(readSerializedDataReply(null)).when(mockActorContext).executeOperationAsync(
 496                 eq(actorSelection(actorRef)), eqSerializedReadData());
 497
 498         expectBatchedModificationsReady(actorRef, true);
 499
 500         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE);
 501
 502         transactionProxy.read(TestModel.TEST_PATH);
 503
 504         DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
 505
 506         assertTrue(ready instanceof SingleCommitCohortProxy);
 507
 508         verifyCohortFutures((SingleCommitCohortProxy)ready, new CommitTransactionReply().toSerializable());
 509
 510         List<BatchedModifications> batchedModifications = captureBatchedModifications(actorRef);
 511         assertEquals("Captured BatchedModifications count", 1, batchedModifications.size());
 512
 513         verifyBatchedModifications(batchedModifications.get(0), true, true);
 514     }
 515
 516     @Test
 517     public void testReadyWithMultipleShardWrites() throws Exception {
 518         ActorRef actorRef1 = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY);
 519
 520         ActorRef actorRef2 = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY, "junk");
 521
 522         expectBatchedModificationsReady(actorRef1);
 523         expectBatchedModificationsReady(actorRef2);
 524
 525         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY);
 526
 527         transactionProxy.write(TestModel.JUNK_PATH, ImmutableNodes.containerNode(TestModel.JUNK_QNAME));
 528         transactionProxy.write(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
 529
 530         DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
 531
 532         assertTrue(ready instanceof ThreePhaseCommitCohortProxy);
 533
 534         verifyCohortFutures((ThreePhaseCommitCohortProxy)ready, actorSelection(actorRef1),
 535                 actorSelection(actorRef2));
 536     }
 537
 538     @Test
 539     public void testReadyWithWriteOnlyAndLastBatchPending() throws Exception {
 540         dataStoreContextBuilder.writeOnlyTransactionOptimizationsEnabled(true);
 541
 542         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY);
 543
 544         NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 545
 546         expectBatchedModificationsReady(actorRef, true);
 547
 548         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY);
 549
 550         transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
 551
 552         DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
 553
 554         assertTrue(ready instanceof SingleCommitCohortProxy);
 555
 556         verifyCohortFutures((SingleCommitCohortProxy)ready, new CommitTransactionReply().toSerializable());
 557
 558         List<BatchedModifications> batchedModifications = captureBatchedModifications(actorRef);
 559         assertEquals("Captured BatchedModifications count", 1, batchedModifications.size());
 560
 561         verifyBatchedModifications(batchedModifications.get(0), true, true,
 562                 new WriteModification(TestModel.TEST_PATH, nodeToWrite));
 563
 564         verify(mockActorContext, never()).executeOperationAsync(eq(actorSelection(actorRef)),
 565                 isA(ReadyTransaction.SERIALIZABLE_CLASS));
 566     }
 567
 568     @Test
 569     public void testReadyWithWriteOnlyAndLastBatchEmpty() throws Exception {
 570         dataStoreContextBuilder.shardBatchedModificationCount(1).writeOnlyTransactionOptimizationsEnabled(true);
 571         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY);
 572
 573         NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 574
 575         expectBatchedModificationsReady(actorRef, true);
 576
 577         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY);
 578
 579         transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
 580
 581         DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
 582
 583         assertTrue(ready instanceof SingleCommitCohortProxy);
 584
 585         verifyCohortFutures((SingleCommitCohortProxy)ready, new CommitTransactionReply().toSerializable());
 586
 587         List<BatchedModifications> batchedModifications = captureBatchedModifications(actorRef);
 588         assertEquals("Captured BatchedModifications count", 2, batchedModifications.size());
 589
 590         verifyBatchedModifications(batchedModifications.get(0), false,
 591                 new WriteModification(TestModel.TEST_PATH, nodeToWrite));
 592
 593         verifyBatchedModifications(batchedModifications.get(1), true, true);
 594
 595         verify(mockActorContext, never()).executeOperationAsync(eq(actorSelection(actorRef)),
 596                 isA(ReadyTransaction.SERIALIZABLE_CLASS));
 597     }
 598
 599     @Test
 600     public void testReadyWithReplyFailure() throws Exception {
 601         dataStoreContextBuilder.writeOnlyTransactionOptimizationsEnabled(true);
 602
 603         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY);
 604
 605         NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 606
 607         expectFailedBatchedModifications(actorRef);
 608
 609         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY);
 610
 611         transactionProxy.merge(TestModel.TEST_PATH, nodeToWrite);
 612
 613         DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
 614
 615         assertTrue(ready instanceof SingleCommitCohortProxy);
 616
 617         verifyCohortFutures((SingleCommitCohortProxy)ready, TestException.class);
 618     }
 619
 620     @Test
 621     public void testReadyWithDebugContextEnabled() throws Exception {
 622         dataStoreContextBuilder.transactionDebugContextEnabled(true);
 623
 624         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE);
 625
 626         expectBatchedModificationsReady(actorRef, true);
 627
 628         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE);
 629
 630         transactionProxy.merge(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
 631
 632         DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
 633
 634         assertTrue(ready instanceof DebugThreePhaseCommitCohort);
 635
 636         verifyCohortFutures((DebugThreePhaseCommitCohort)ready, new CommitTransactionReply().toSerializable());
 637     }
 638
 639     private void testWriteOnlyTxWithFindPrimaryShardFailure(Exception toThrow) throws Exception {
 640         doReturn(Futures.failed(toThrow)).when(mockActorContext).findPrimaryShardAsync(anyString());
 641
 642         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY);
 643
 644         NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 645
 646         transactionProxy.merge(TestModel.TEST_PATH, nodeToWrite);
 647
 648         transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
 649
 650         transactionProxy.delete(TestModel.TEST_PATH);
 651
 652         DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
 653
 654         assertTrue(ready instanceof SingleCommitCohortProxy);
 655
 656         verifyCohortFutures((SingleCommitCohortProxy)ready, toThrow.getClass());
 657     }
 658
 659     @Test
 660     public void testWriteOnlyTxWithPrimaryNotFoundException() throws Exception {
 661         testWriteOnlyTxWithFindPrimaryShardFailure(new PrimaryNotFoundException("mock"));
 662     }
 663
 664     @Test
 665     public void testWriteOnlyTxWithNotInitializedException() throws Exception {
 666         testWriteOnlyTxWithFindPrimaryShardFailure(new NotInitializedException("mock"));
 667     }
 668
 669     @Test
 670     public void testWriteOnlyTxWithNoShardLeaderException() throws Exception {
 671         testWriteOnlyTxWithFindPrimaryShardFailure(new NoShardLeaderException("mock"));
 672     }
 673
 674     @Test
 675     public void testReadyWithInvalidReplyMessageType() throws Exception {
 676         dataStoreContextBuilder.writeOnlyTransactionOptimizationsEnabled(true);
 677         ActorRef actorRef1 = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY);
 678
 679         ActorRef actorRef2 = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY, "junk");
 680
 681         doReturn(Futures.successful(new Object())).when(mockActorContext).
 682                 executeOperationAsync(eq(actorSelection(actorRef1)), isA(BatchedModifications.class));
 683
 684         expectBatchedModificationsReady(actorRef2);
 685
 686         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY);
 687
 688         transactionProxy.write(TestModel.JUNK_PATH, ImmutableNodes.containerNode(TestModel.JUNK_QNAME));
 689         transactionProxy.write(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
 690
 691         DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
 692
 693         assertTrue(ready instanceof ThreePhaseCommitCohortProxy);
 694
 695         verifyCohortFutures((ThreePhaseCommitCohortProxy)ready, actorSelection(actorRef2),
 696                 IllegalArgumentException.class);
 697     }
 698
 699     @Test
 700     public void testGetIdentifier() {
 701         setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY);
 702         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY);
 703
 704         Object id = transactionProxy.getIdentifier();
 705         assertNotNull("getIdentifier returned null", id);
 706         assertTrue("Invalid identifier: " + id, id.toString().startsWith(memberName));
 707     }
 708
 709     @Test
 710     public void testClose() throws Exception{
 711         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE);
 712
 713         doReturn(readSerializedDataReply(null)).when(mockActorContext).executeOperationAsync(
 714                 eq(actorSelection(actorRef)), eqSerializedReadData());
 715
 716         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE);
 717
 718         transactionProxy.read(TestModel.TEST_PATH);
 719
 720         transactionProxy.close();
 721
 722         verify(mockActorContext).sendOperationAsync(
 723                 eq(actorSelection(actorRef)), isA(CloseTransaction.SERIALIZABLE_CLASS));
 724     }
 725
 726
 727     /**
 728      * Method to test a local Tx actor. The Tx paths are matched to decide if the
 729      * Tx actor is local or not. This is done by mocking the Tx actor path
 730      * and the caller paths and ensuring that the paths have the remote-address format
 731      *
 732      * Note: Since the default akka provider for test is not a RemoteActorRefProvider,
 733      * the paths returned for the actors for all the tests are not qualified remote paths.
 734      * Hence are treated as non-local/remote actors. In short, all tests except
 735      * few below run for remote actors
 736      *
 737      * @throws Exception
 738      */
 739     @Test
 740     public void testLocalTxActorRead() throws Exception {
 741         setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY);
 742         doReturn(true).when(mockActorContext).isPathLocal(anyString());
 743
 744         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY);
 745
 746         // negative test case with null as the reply
 747         doReturn(readDataReply(null)).when(mockActorContext).executeOperationAsync(
 748             any(ActorSelection.class), eqReadData());
 749
 750         Optional<NormalizedNode<?, ?>> readOptional = transactionProxy.read(
 751             TestModel.TEST_PATH).get(5, TimeUnit.SECONDS);
 752
 753         assertEquals("NormalizedNode isPresent", false, readOptional.isPresent());
 754
 755         // test case with node as read data reply
 756         NormalizedNode<?, ?> expectedNode = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 757
 758         doReturn(readDataReply(expectedNode)).when(mockActorContext).executeOperationAsync(
 759             any(ActorSelection.class), eqReadData());
 760
 761         readOptional = transactionProxy.read(TestModel.TEST_PATH).get(5, TimeUnit.SECONDS);
 762
 763         assertEquals("NormalizedNode isPresent", true, readOptional.isPresent());
 764
 765         assertEquals("Response NormalizedNode", expectedNode, readOptional.get());
 766
 767         // test for local data exists
 768         doReturn(dataExistsReply(true)).when(mockActorContext).executeOperationAsync(
 769             any(ActorSelection.class), eqDataExists());
 770
 771         boolean exists = transactionProxy.exists(TestModel.TEST_PATH).checkedGet();
 772
 773         assertEquals("Exists response", true, exists);
 774     }
 775
 776     @Test
 777     public void testLocalTxActorReady() throws Exception {
 778         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE);
 779         doReturn(true).when(mockActorContext).isPathLocal(anyString());
 780
 781         expectBatchedModificationsReady(actorRef, true);
 782
 783         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE);
 784
 785         NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 786         transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
 787
 788         DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
 789
 790         assertTrue(ready instanceof SingleCommitCohortProxy);
 791
 792         verifyCohortFutures((SingleCommitCohortProxy)ready, new CommitTransactionReply().toSerializable());
 793     }
 794
 795     private static interface TransactionProxyOperation {
 796         void run(TransactionProxy transactionProxy);
 797     }
 798
 799     private void throttleOperation(TransactionProxyOperation operation) {
 800         throttleOperation(operation, 1, true);
 801     }
 802
 803     private void throttleOperation(TransactionProxyOperation operation, int outstandingOpsLimit, boolean shardFound){
 804         throttleOperation(operation, outstandingOpsLimit, shardFound, TimeUnit.MILLISECONDS.toNanos(
 805                 mockActorContext.getDatastoreContext().getOperationTimeoutInMillis()));
 806     }
 807
 808     private PrimaryShardInfo newPrimaryShardInfo(ActorRef actorRef){
 809         return new PrimaryShardInfo(getSystem().actorSelection(actorRef.path()), DataStoreVersions.CURRENT_VERSION,
 810                 Optional.<DataTree>absent());
 811     }
 812
 813     private PrimaryShardInfo newPrimaryShardInfo(ActorRef actorRef, Optional<DataTree> dataTreeOptional){
 814         return new PrimaryShardInfo(getSystem().actorSelection(actorRef.path()), DataStoreVersions.CURRENT_VERSION,
 815                 dataTreeOptional);
 816     }
 817
 818
 819     private void throttleOperation(TransactionProxyOperation operation, int outstandingOpsLimit, boolean shardFound, long expectedCompletionTime){
 820         ActorSystem actorSystem = getSystem();
 821         ActorRef shardActorRef = actorSystem.actorOf(Props.create(DoNothingActor.class));
 822
 823         // Note that we setting batchedModificationCount to one less than what we need because in TransactionProxy
 824         // we now allow one extra permit to be allowed for ready
 825         doReturn(dataStoreContextBuilder.operationTimeoutInSeconds(2).
 826                 shardBatchedModificationCount(outstandingOpsLimit-1).build()).when(mockActorContext).getDatastoreContext();
 827
 828         doReturn(actorSystem.actorSelection(shardActorRef.path())).
 829                 when(mockActorContext).actorSelection(shardActorRef.path().toString());
 830
 831         if(shardFound) {
 832             doReturn(Futures.successful(newPrimaryShardInfo(shardActorRef))).
 833                     when(mockActorContext).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD));
 834             doReturn(Futures.successful(newPrimaryShardInfo(shardActorRef))).
 835                     when(mockActorContext).findPrimaryShardAsync(eq("cars"));
 836
 837         } else {
 838             doReturn(Futures.failed(new Exception("not found")))
 839                     .when(mockActorContext).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD));
 840         }
 841
 842         String actorPath = "akka.tcp://system@127.0.0.1:2550/user/tx-actor";
 843
 844         doReturn(incompleteFuture()).when(mockActorContext).
 845         executeOperationAsync(eq(actorSystem.actorSelection(shardActorRef.path())),
 846                  eqCreateTransaction(memberName, READ_WRITE), any(Timeout.class));
 847
 848         doReturn(true).when(mockActorContext).isPathLocal(actorPath);
 849
 850         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE);
 851
 852         long start = System.nanoTime();
 853
 854         operation.run(transactionProxy);
 855
 856         long end = System.nanoTime();
 857
 858         Assert.assertTrue(String.format("Expected elapsed time: %s. Actual: %s",
 859                 expectedCompletionTime, (end-start)),
 860                 ((end - start) > expectedCompletionTime) && ((end - start) < expectedCompletionTime*2));
 861
 862     }
 863
 864     private void completeOperation(TransactionProxyOperation operation){
 865         completeOperation(operation, true);
 866     }
 867
 868     private void completeOperation(TransactionProxyOperation operation, boolean shardFound){
 869         ActorSystem actorSystem = getSystem();
 870         ActorRef shardActorRef = actorSystem.actorOf(Props.create(DoNothingActor.class));
 871
 872         doReturn(actorSystem.actorSelection(shardActorRef.path())).
 873                 when(mockActorContext).actorSelection(shardActorRef.path().toString());
 874
 875         if(shardFound) {
 876             doReturn(Futures.successful(newPrimaryShardInfo(shardActorRef))).
 877                     when(mockActorContext).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD));
 878         } else {
 879             doReturn(Futures.failed(new PrimaryNotFoundException("test")))
 880                     .when(mockActorContext).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD));
 881         }
 882
 883         ActorRef txActorRef = actorSystem.actorOf(Props.create(DoNothingActor.class));
 884         String actorPath = txActorRef.path().toString();
 885         CreateTransactionReply createTransactionReply = CreateTransactionReply.newBuilder().
 886                 setTransactionId("txn-1").setTransactionActorPath(actorPath).
 887                 setMessageVersion(DataStoreVersions.CURRENT_VERSION).build();
 888
 889         doReturn(actorSystem.actorSelection(actorPath)).when(mockActorContext).actorSelection(actorPath);
 890
 891         doReturn(Futures.successful(createTransactionReply)).when(mockActorContext).
 892                 executeOperationAsync(eq(actorSystem.actorSelection(shardActorRef.path())),
 893                         eqCreateTransaction(memberName, READ_WRITE), any(Timeout.class));
 894
 895         doReturn(true).when(mockActorContext).isPathLocal(anyString());
 896
 897         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE);
 898
 899         long start = System.nanoTime();
 900
 901         operation.run(transactionProxy);
 902
 903         long end = System.nanoTime();
 904
 905         long expected = TimeUnit.MILLISECONDS.toNanos(mockActorContext.getDatastoreContext().getOperationTimeoutInMillis());
 906         Assert.assertTrue(String.format("Expected elapsed time: %s. Actual: %s",
 907                 expected, (end-start)), (end - start) <= expected);
 908     }
 909
 910     private void completeOperationLocal(TransactionProxyOperation operation, Optional<DataTree> dataTreeOptional){
 911         ActorSystem actorSystem = getSystem();
 912         ActorRef shardActorRef = actorSystem.actorOf(Props.create(DoNothingActor.class));
 913
 914         doReturn(actorSystem.actorSelection(shardActorRef.path())).
 915                 when(mockActorContext).actorSelection(shardActorRef.path().toString());
 916
 917         doReturn(Futures.successful(newPrimaryShardInfo(shardActorRef, dataTreeOptional))).
 918                 when(mockActorContext).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD));
 919
 920         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE);
 921
 922         long start = System.nanoTime();
 923
 924         operation.run(transactionProxy);
 925
 926         long end = System.nanoTime();
 927
 928         long expected = TimeUnit.MILLISECONDS.toNanos(mockActorContext.getDatastoreContext().getOperationTimeoutInMillis());
 929         Assert.assertTrue(String.format("Expected elapsed time: %s. Actual: %s",
 930                 expected, (end-start)), (end - start) <= expected);
 931     }
 932
 933     private Optional<DataTree> createDataTree(){
 934         DataTree dataTree = mock(DataTree.class);
 935         Optional<DataTree> dataTreeOptional = Optional.of(dataTree);
 936         DataTreeSnapshot dataTreeSnapshot = mock(DataTreeSnapshot.class);
 937         DataTreeModification dataTreeModification = mock(DataTreeModification.class);
 938
 939         doReturn(dataTreeSnapshot).when(dataTree).takeSnapshot();
 940         doReturn(dataTreeModification).when(dataTreeSnapshot).newModification();
 941
 942         return dataTreeOptional;
 943     }
 944
 945     private Optional<DataTree> createDataTree(NormalizedNode readResponse){
 946         DataTree dataTree = mock(DataTree.class);
 947         Optional<DataTree> dataTreeOptional = Optional.of(dataTree);
 948         DataTreeSnapshot dataTreeSnapshot = mock(DataTreeSnapshot.class);
 949         DataTreeModification dataTreeModification = mock(DataTreeModification.class);
 950
 951         doReturn(dataTreeSnapshot).when(dataTree).takeSnapshot();
 952         doReturn(dataTreeModification).when(dataTreeSnapshot).newModification();
 953         doReturn(Optional.of(readResponse)).when(dataTreeModification).readNode(any(YangInstanceIdentifier.class));
 954
 955         return dataTreeOptional;
 956     }
 957
 958
 959     @Test
 960     public void testWriteCompletionForLocalShard(){
 961         completeOperationLocal(new TransactionProxyOperation() {
 962             @Override
 963             public void run(TransactionProxy transactionProxy) {
 964                 NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 965
 966                 transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
 967
 968                 transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
 969
 970             }
 971         }, createDataTree());
 972     }
 973
 974     @Test
 975     public void testWriteThrottlingWhenShardFound(){
 976         throttleOperation(new TransactionProxyOperation() {
 977             @Override
 978             public void run(TransactionProxy transactionProxy) {
 979                 NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 980
 981                 expectIncompleteBatchedModifications();
 982
 983                 transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
 984
 985                 transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
 986             }
 987         });
 988     }
 989
 990     @Test
 991     public void testWriteThrottlingWhenShardNotFound(){
 992         // Confirm that there is no throttling when the Shard is not found
 993         completeOperation(new TransactionProxyOperation() {
 994             @Override
 995             public void run(TransactionProxy transactionProxy) {
 996                 NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 997
 998                 expectBatchedModifications(2);
 999
1000                 transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
1001
1002                 transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
1003             }
1004         }, false);
1005
1006     }
1007
1008
1009     @Test
1010     public void testWriteCompletion(){
1011         completeOperation(new TransactionProxyOperation() {
1012             @Override
1013             public void run(TransactionProxy transactionProxy) {
1014                 NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1015
1016                 expectBatchedModifications(2);
1017
1018                 transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
1019
1020                 transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
1021             }
1022         });
1023     }
1024
1025     @Test
1026     public void testMergeThrottlingWhenShardFound(){
1027         throttleOperation(new TransactionProxyOperation() {
1028             @Override
1029             public void run(TransactionProxy transactionProxy) {
1030                 NormalizedNode<?, ?> nodeToMerge = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1031
1032                 expectIncompleteBatchedModifications();
1033
1034                 transactionProxy.merge(TestModel.TEST_PATH, nodeToMerge);
1035
1036                 transactionProxy.merge(TestModel.TEST_PATH, nodeToMerge);
1037             }
1038         });
1039     }
1040
1041     @Test
1042     public void testMergeThrottlingWhenShardNotFound(){
1043         completeOperation(new TransactionProxyOperation() {
1044             @Override
1045             public void run(TransactionProxy transactionProxy) {
1046                 NormalizedNode<?, ?> nodeToMerge = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1047
1048                 expectBatchedModifications(2);
1049
1050                 transactionProxy.merge(TestModel.TEST_PATH, nodeToMerge);
1051
1052                 transactionProxy.merge(TestModel.TEST_PATH, nodeToMerge);
1053             }
1054         }, false);
1055     }
1056
1057     @Test
1058     public void testMergeCompletion(){
1059         completeOperation(new TransactionProxyOperation() {
1060             @Override
1061             public void run(TransactionProxy transactionProxy) {
1062                 NormalizedNode<?, ?> nodeToMerge = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1063
1064                 expectBatchedModifications(2);
1065
1066                 transactionProxy.merge(TestModel.TEST_PATH, nodeToMerge);
1067
1068                 transactionProxy.merge(TestModel.TEST_PATH, nodeToMerge);
1069             }
1070         });
1071
1072     }
1073
1074     @Test
1075     public void testMergeCompletionForLocalShard(){
1076         completeOperationLocal(new TransactionProxyOperation() {
1077             @Override
1078             public void run(TransactionProxy transactionProxy) {
1079                 NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1080
1081                 transactionProxy.merge(TestModel.TEST_PATH, nodeToWrite);
1082
1083                 transactionProxy.merge(TestModel.TEST_PATH, nodeToWrite);
1084
1085             }
1086         }, createDataTree());
1087     }
1088
1089
1090     @Test
1091     public void testDeleteThrottlingWhenShardFound(){
1092
1093         throttleOperation(new TransactionProxyOperation() {
1094             @Override
1095             public void run(TransactionProxy transactionProxy) {
1096                 expectIncompleteBatchedModifications();
1097
1098                 transactionProxy.delete(TestModel.TEST_PATH);
1099
1100                 transactionProxy.delete(TestModel.TEST_PATH);
1101             }
1102         });
1103     }
1104
1105
1106     @Test
1107     public void testDeleteThrottlingWhenShardNotFound(){
1108
1109         completeOperation(new TransactionProxyOperation() {
1110             @Override
1111             public void run(TransactionProxy transactionProxy) {
1112                 expectBatchedModifications(2);
1113
1114                 transactionProxy.delete(TestModel.TEST_PATH);
1115
1116                 transactionProxy.delete(TestModel.TEST_PATH);
1117             }
1118         }, false);
1119     }
1120
1121     @Test
1122     public void testDeleteCompletionForLocalShard(){
1123         completeOperationLocal(new TransactionProxyOperation() {
1124             @Override
1125             public void run(TransactionProxy transactionProxy) {
1126
1127                 transactionProxy.delete(TestModel.TEST_PATH);
1128
1129                 transactionProxy.delete(TestModel.TEST_PATH);
1130             }
1131         }, createDataTree());
1132
1133     }
1134
1135     @Test
1136     public void testDeleteCompletion(){
1137         completeOperation(new TransactionProxyOperation() {
1138             @Override
1139             public void run(TransactionProxy transactionProxy) {
1140                 expectBatchedModifications(2);
1141
1142                 transactionProxy.delete(TestModel.TEST_PATH);
1143
1144                 transactionProxy.delete(TestModel.TEST_PATH);
1145             }
1146         });
1147
1148     }
1149
1150     @Test
1151     public void testReadThrottlingWhenShardFound(){
1152
1153         throttleOperation(new TransactionProxyOperation() {
1154             @Override
1155             public void run(TransactionProxy transactionProxy) {
1156                 doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync(
1157                         any(ActorSelection.class), eqReadData());
1158
1159                 transactionProxy.read(TestModel.TEST_PATH);
1160
1161                 transactionProxy.read(TestModel.TEST_PATH);
1162             }
1163         });
1164     }
1165
1166     @Test
1167     public void testReadThrottlingWhenShardNotFound(){
1168
1169         completeOperation(new TransactionProxyOperation() {
1170             @Override
1171             public void run(TransactionProxy transactionProxy) {
1172                 doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync(
1173                         any(ActorSelection.class), eqReadData());
1174
1175                 transactionProxy.read(TestModel.TEST_PATH);
1176
1177                 transactionProxy.read(TestModel.TEST_PATH);
1178             }
1179         }, false);
1180     }
1181
1182
1183     @Test
1184     public void testReadCompletion(){
1185         completeOperation(new TransactionProxyOperation() {
1186             @Override
1187             public void run(TransactionProxy transactionProxy) {
1188                 NormalizedNode<?, ?> nodeToRead = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1189
1190                 doReturn(readDataReply(nodeToRead)).when(mockActorContext).executeOperationAsync(
1191                         any(ActorSelection.class), eqReadData());
1192
1193                 transactionProxy.read(TestModel.TEST_PATH);
1194
1195                 transactionProxy.read(TestModel.TEST_PATH);
1196             }
1197         });
1198
1199     }
1200
1201     @Test
1202     public void testReadCompletionForLocalShard(){
1203         final NormalizedNode nodeToRead = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1204         completeOperationLocal(new TransactionProxyOperation() {
1205             @Override
1206             public void run(TransactionProxy transactionProxy) {
1207                 transactionProxy.read(TestModel.TEST_PATH);
1208
1209                 transactionProxy.read(TestModel.TEST_PATH);
1210             }
1211         }, createDataTree(nodeToRead));
1212
1213     }
1214
1215     @Test
1216     public void testReadCompletionForLocalShardWhenExceptionOccurs(){
1217         completeOperationLocal(new TransactionProxyOperation() {
1218             @Override
1219             public void run(TransactionProxy transactionProxy) {
1220                 transactionProxy.read(TestModel.TEST_PATH);
1221
1222                 transactionProxy.read(TestModel.TEST_PATH);
1223             }
1224         }, createDataTree());
1225
1226     }
1227
1228     @Test
1229     public void testExistsThrottlingWhenShardFound(){
1230
1231         throttleOperation(new TransactionProxyOperation() {
1232             @Override
1233             public void run(TransactionProxy transactionProxy) {
1234                 doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync(
1235                         any(ActorSelection.class), eqDataExists());
1236
1237                 transactionProxy.exists(TestModel.TEST_PATH);
1238
1239                 transactionProxy.exists(TestModel.TEST_PATH);
1240             }
1241         });
1242     }
1243
1244     @Test
1245     public void testExistsThrottlingWhenShardNotFound(){
1246
1247         completeOperation(new TransactionProxyOperation() {
1248             @Override
1249             public void run(TransactionProxy transactionProxy) {
1250                 doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync(
1251                         any(ActorSelection.class), eqDataExists());
1252
1253                 transactionProxy.exists(TestModel.TEST_PATH);
1254
1255                 transactionProxy.exists(TestModel.TEST_PATH);
1256             }
1257         }, false);
1258     }
1259
1260
1261     @Test
1262     public void testExistsCompletion(){
1263         completeOperation(new TransactionProxyOperation() {
1264             @Override
1265             public void run(TransactionProxy transactionProxy) {
1266                 doReturn(dataExistsReply(true)).when(mockActorContext).executeOperationAsync(
1267                         any(ActorSelection.class), eqDataExists());
1268
1269                 transactionProxy.exists(TestModel.TEST_PATH);
1270
1271                 transactionProxy.exists(TestModel.TEST_PATH);
1272             }
1273         });
1274
1275     }
1276
1277     @Test
1278     public void testExistsCompletionForLocalShard(){
1279         final NormalizedNode nodeToRead = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1280         completeOperationLocal(new TransactionProxyOperation() {
1281             @Override
1282             public void run(TransactionProxy transactionProxy) {
1283                 transactionProxy.exists(TestModel.TEST_PATH);
1284
1285                 transactionProxy.exists(TestModel.TEST_PATH);
1286             }
1287         }, createDataTree(nodeToRead));
1288
1289     }
1290
1291     @Test
1292     public void testExistsCompletionForLocalShardWhenExceptionOccurs(){
1293         completeOperationLocal(new TransactionProxyOperation() {
1294             @Override
1295             public void run(TransactionProxy transactionProxy) {
1296                 transactionProxy.exists(TestModel.TEST_PATH);
1297
1298                 transactionProxy.exists(TestModel.TEST_PATH);
1299             }
1300         }, createDataTree());
1301
1302     }
1303     @Test
1304     public void testReadyThrottling(){
1305
1306         throttleOperation(new TransactionProxyOperation() {
1307             @Override
1308             public void run(TransactionProxy transactionProxy) {
1309                 NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1310
1311                 expectBatchedModifications(1);
1312
1313                 transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
1314
1315                 transactionProxy.ready();
1316             }
1317         });
1318     }
1319
1320     @Test
1321     public void testReadyThrottlingWithTwoTransactionContexts(){
1322         throttleOperation(new TransactionProxyOperation() {
1323             @Override
1324             public void run(TransactionProxy transactionProxy) {
1325                 NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1326                 NormalizedNode<?, ?> carsNode = ImmutableNodes.containerNode(CarsModel.BASE_QNAME);
1327
1328                 expectBatchedModifications(2);
1329
1330                 doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync(
1331                         any(ActorSelection.class), any(ReadyTransaction.class));
1332
1333                 transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
1334
1335                 // Trying to write to Cars will cause another transaction context to get created
1336                 transactionProxy.write(CarsModel.BASE_PATH, carsNode);
1337
1338                 // Now ready should block for both transaction contexts
1339                 transactionProxy.ready();
1340             }
1341         }, 1, true, TimeUnit.MILLISECONDS.toNanos(mockActorContext.getDatastoreContext().getOperationTimeoutInMillis()) * 2);
1342     }
1343
1344     private void testModificationOperationBatching(TransactionType type) throws Exception {
1345         int shardBatchedModificationCount = 3;
1346         dataStoreContextBuilder.shardBatchedModificationCount(shardBatchedModificationCount);
1347
1348         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), type);
1349
1350         expectBatchedModifications(actorRef, shardBatchedModificationCount);
1351
1352         YangInstanceIdentifier writePath1 = TestModel.TEST_PATH;
1353         NormalizedNode<?, ?> writeNode1 = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1354
1355         YangInstanceIdentifier writePath2 = TestModel.OUTER_LIST_PATH;
1356         NormalizedNode<?, ?> writeNode2 = ImmutableNodes.containerNode(TestModel.OUTER_LIST_QNAME);
1357
1358         YangInstanceIdentifier writePath3 = TestModel.INNER_LIST_PATH;
1359         NormalizedNode<?, ?> writeNode3 = ImmutableNodes.containerNode(TestModel.INNER_LIST_QNAME);
1360
1361         YangInstanceIdentifier mergePath1 = TestModel.TEST_PATH;
1362         NormalizedNode<?, ?> mergeNode1 = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1363
1364         YangInstanceIdentifier mergePath2 = TestModel.OUTER_LIST_PATH;
1365         NormalizedNode<?, ?> mergeNode2 = ImmutableNodes.containerNode(TestModel.OUTER_LIST_QNAME);
1366
1367         YangInstanceIdentifier mergePath3 = TestModel.INNER_LIST_PATH;
1368         NormalizedNode<?, ?> mergeNode3 = ImmutableNodes.containerNode(TestModel.INNER_LIST_QNAME);
1369
1370         YangInstanceIdentifier deletePath1 = TestModel.TEST_PATH;
1371         YangInstanceIdentifier deletePath2 = TestModel.OUTER_LIST_PATH;
1372
1373         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, type);
1374
1375         transactionProxy.write(writePath1, writeNode1);
1376         transactionProxy.write(writePath2, writeNode2);
1377         transactionProxy.delete(deletePath1);
1378         transactionProxy.merge(mergePath1, mergeNode1);
1379         transactionProxy.merge(mergePath2, mergeNode2);
1380         transactionProxy.write(writePath3, writeNode3);
1381         transactionProxy.merge(mergePath3, mergeNode3);
1382         transactionProxy.delete(deletePath2);
1383
1384         // This sends the last batch.
1385         transactionProxy.ready();
1386
1387         List<BatchedModifications> batchedModifications = captureBatchedModifications(actorRef);
1388         assertEquals("Captured BatchedModifications count", 3, batchedModifications.size());
1389
1390         verifyBatchedModifications(batchedModifications.get(0), false, new WriteModification(writePath1, writeNode1),
1391                 new WriteModification(writePath2, writeNode2), new DeleteModification(deletePath1));
1392
1393         verifyBatchedModifications(batchedModifications.get(1), false, new MergeModification(mergePath1, mergeNode1),
1394                 new MergeModification(mergePath2, mergeNode2), new WriteModification(writePath3, writeNode3));
1395
1396         verifyBatchedModifications(batchedModifications.get(2), true, true,
1397                 new MergeModification(mergePath3, mergeNode3), new DeleteModification(deletePath2));
1398
1399         assertEquals("getTotalMessageCount", 3, batchedModifications.get(2).getTotalMessagesSent());
1400     }
1401
1402     @Test
1403     public void testReadWriteModificationOperationBatching() throws Throwable {
1404         testModificationOperationBatching(READ_WRITE);
1405     }
1406
1407     @Test
1408     public void testWriteOnlyModificationOperationBatching() throws Throwable {
1409         testModificationOperationBatching(WRITE_ONLY);
1410     }
1411
1412     @Test
1413     public void testOptimizedWriteOnlyModificationOperationBatching() throws Throwable {
1414         dataStoreContextBuilder.writeOnlyTransactionOptimizationsEnabled(true);
1415         testModificationOperationBatching(WRITE_ONLY);
1416     }
1417
1418     @Test
1419     public void testModificationOperationBatchingWithInterleavedReads() throws Throwable {
1420
1421         int shardBatchedModificationCount = 10;
1422         dataStoreContextBuilder.shardBatchedModificationCount(shardBatchedModificationCount);
1423
1424         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE);
1425
1426         expectBatchedModifications(actorRef, shardBatchedModificationCount);
1427
1428         YangInstanceIdentifier writePath1 = TestModel.TEST_PATH;
1429         NormalizedNode<?, ?> writeNode1 = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1430
1431         YangInstanceIdentifier writePath2 = TestModel.OUTER_LIST_PATH;
1432         NormalizedNode<?, ?> writeNode2 = ImmutableNodes.containerNode(TestModel.OUTER_LIST_QNAME);
1433
1434         YangInstanceIdentifier mergePath1 = TestModel.TEST_PATH;
1435         NormalizedNode<?, ?> mergeNode1 = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1436
1437         YangInstanceIdentifier mergePath2 = TestModel.INNER_LIST_PATH;
1438         NormalizedNode<?, ?> mergeNode2 = ImmutableNodes.containerNode(TestModel.INNER_LIST_QNAME);
1439
1440         YangInstanceIdentifier deletePath = TestModel.OUTER_LIST_PATH;
1441
1442         doReturn(readSerializedDataReply(writeNode2)).when(mockActorContext).executeOperationAsync(
1443                 eq(actorSelection(actorRef)), eqSerializedReadData(writePath2));
1444
1445         doReturn(readSerializedDataReply(mergeNode2)).when(mockActorContext).executeOperationAsync(
1446                 eq(actorSelection(actorRef)), eqSerializedReadData(mergePath2));
1447
1448         doReturn(dataExistsSerializedReply(true)).when(mockActorContext).executeOperationAsync(
1449                 eq(actorSelection(actorRef)), eqSerializedDataExists());
1450
1451         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE);
1452
1453         transactionProxy.write(writePath1, writeNode1);
1454         transactionProxy.write(writePath2, writeNode2);
1455
1456         Optional<NormalizedNode<?, ?>> readOptional = transactionProxy.read(writePath2).
1457                 get(5, TimeUnit.SECONDS);
1458
1459         assertEquals("NormalizedNode isPresent", true, readOptional.isPresent());
1460         assertEquals("Response NormalizedNode", writeNode2, readOptional.get());
1461
1462         transactionProxy.merge(mergePath1, mergeNode1);
1463         transactionProxy.merge(mergePath2, mergeNode2);
1464
1465         readOptional = transactionProxy.read(mergePath2).get(5, TimeUnit.SECONDS);
1466
1467         transactionProxy.delete(deletePath);
1468
1469         Boolean exists = transactionProxy.exists(TestModel.TEST_PATH).checkedGet();
1470         assertEquals("Exists response", true, exists);
1471
1472         assertEquals("NormalizedNode isPresent", true, readOptional.isPresent());
1473         assertEquals("Response NormalizedNode", mergeNode2, readOptional.get());
1474
1475         List<BatchedModifications> batchedModifications = captureBatchedModifications(actorRef);
1476         assertEquals("Captured BatchedModifications count", 3, batchedModifications.size());
1477
1478         verifyBatchedModifications(batchedModifications.get(0), false, new WriteModification(writePath1, writeNode1),
1479                 new WriteModification(writePath2, writeNode2));
1480
1481         verifyBatchedModifications(batchedModifications.get(1), false, new MergeModification(mergePath1, mergeNode1),
1482                 new MergeModification(mergePath2, mergeNode2));
1483
1484         verifyBatchedModifications(batchedModifications.get(2), false, new DeleteModification(deletePath));
1485
1486         InOrder inOrder = Mockito.inOrder(mockActorContext);
1487         inOrder.verify(mockActorContext).executeOperationAsync(
1488                 eq(actorSelection(actorRef)), isA(BatchedModifications.class));
1489
1490         inOrder.verify(mockActorContext).executeOperationAsync(
1491                 eq(actorSelection(actorRef)), eqSerializedReadData(writePath2));
1492
1493         inOrder.verify(mockActorContext).executeOperationAsync(
1494                 eq(actorSelection(actorRef)), isA(BatchedModifications.class));
1495
1496         inOrder.verify(mockActorContext).executeOperationAsync(
1497                 eq(actorSelection(actorRef)), eqSerializedReadData(mergePath2));
1498
1499         inOrder.verify(mockActorContext).executeOperationAsync(
1500                 eq(actorSelection(actorRef)), isA(BatchedModifications.class));
1501
1502         inOrder.verify(mockActorContext).executeOperationAsync(
1503                 eq(actorSelection(actorRef)), eqSerializedDataExists());
1504     }
1505
1506     @Test
1507     public void testReadRoot() throws ReadFailedException, InterruptedException, ExecutionException, java.util.concurrent.TimeoutException {
1508
1509         SchemaContext schemaContext = SchemaContextHelper.full();
1510         Configuration configuration = mock(Configuration.class);
1511         doReturn(configuration).when(mockActorContext).getConfiguration();
1512         doReturn(schemaContext).when(mockActorContext).getSchemaContext();
1513         doReturn(Sets.newHashSet("test", "cars")).when(configuration).getAllShardNames();
1514
1515         NormalizedNode<?, ?> expectedNode1 = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1516         NormalizedNode<?, ?> expectedNode2 = ImmutableNodes.containerNode(CarsModel.CARS_QNAME);
1517
1518         setUpReadData("test", NormalizedNodeAggregatorTest.getRootNode(expectedNode1, schemaContext));
1519         setUpReadData("cars", NormalizedNodeAggregatorTest.getRootNode(expectedNode2, schemaContext));
1520
1521         doReturn(memberName).when(mockActorContext).getCurrentMemberName();
1522
1523         doReturn(getSystem().dispatchers().defaultGlobalDispatcher()).when(mockActorContext).getClientDispatcher();
1524
1525         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY);
1526
1527         Optional<NormalizedNode<?, ?>> readOptional = transactionProxy.read(
1528                 YangInstanceIdentifier.builder().build()).get(5, TimeUnit.SECONDS);
1529
1530         assertEquals("NormalizedNode isPresent", true, readOptional.isPresent());
1531
1532         NormalizedNode<?, ?> normalizedNode = readOptional.get();
1533
1534         assertTrue("Expect value to be a Collection", normalizedNode.getValue() instanceof Collection);
1535
1536         Collection<NormalizedNode<?,?>> collection = (Collection<NormalizedNode<?,?>>) normalizedNode.getValue();
1537
1538         for(NormalizedNode<?,?> node : collection){
1539             assertTrue("Expected " + node + " to be a ContainerNode", node instanceof ContainerNode);
1540         }
1541
1542         assertTrue("Child with QName = " + TestModel.TEST_QNAME + " not found",
1543                 NormalizedNodeAggregatorTest.findChildWithQName(collection, TestModel.TEST_QNAME) != null);
1544
1545         assertEquals(expectedNode1, NormalizedNodeAggregatorTest.findChildWithQName(collection, TestModel.TEST_QNAME));
1546
1547         assertTrue("Child with QName = " + CarsModel.BASE_QNAME + " not found",
1548                 NormalizedNodeAggregatorTest.findChildWithQName(collection, CarsModel.BASE_QNAME) != null);
1549
1550         assertEquals(expectedNode2, NormalizedNodeAggregatorTest.findChildWithQName(collection, CarsModel.BASE_QNAME));
1551     }
1552
1553
1554     private void setUpReadData(String shardName, NormalizedNode<?, ?> expectedNode) {
1555         ActorSystem actorSystem = getSystem();
1556         ActorRef shardActorRef = getSystem().actorOf(Props.create(DoNothingActor.class));
1557
1558         doReturn(getSystem().actorSelection(shardActorRef.path())).
1559                 when(mockActorContext).actorSelection(shardActorRef.path().toString());
1560
1561         doReturn(primaryShardInfoReply(getSystem(), shardActorRef)).
1562                 when(mockActorContext).findPrimaryShardAsync(eq(shardName));
1563
1564         doReturn(true).when(mockActorContext).isPathLocal(shardActorRef.path().toString());
1565
1566         ActorRef txActorRef = actorSystem.actorOf(Props.create(DoNothingActor.class));
1567
1568         doReturn(actorSystem.actorSelection(txActorRef.path())).
1569                 when(mockActorContext).actorSelection(txActorRef.path().toString());
1570
1571         doReturn(Futures.successful(createTransactionReply(txActorRef, DataStoreVersions.CURRENT_VERSION))).when(mockActorContext).
1572                 executeOperationAsync(eq(actorSystem.actorSelection(shardActorRef.path())),
1573                         eqCreateTransaction(memberName, TransactionType.READ_ONLY), any(Timeout.class));
1574
1575         doReturn(readSerializedDataReply(expectedNode)).when(mockActorContext).executeOperationAsync(
1576                 eq(actorSelection(txActorRef)), eqSerializedReadData(YangInstanceIdentifier.builder().build()));
1577     }
1578 }