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

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