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