opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/TransactionProxyTest.java

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