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

   1 /*
   2  *
   3  *  Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
   4  *
   5  *  This program and the accompanying materials are made available under the
   6  *  terms of the Eclipse Public License v1.0 which accompanies this distribution,
   7  *  and is available at http://www.eclipse.org/legal/epl-v10.html
   8  *
   9  */
  10
  11 package org.opendaylight.controller.cluster.datastore;
  12
  13 import akka.actor.ActorRef;
  14 import akka.actor.Props;
  15 import akka.testkit.TestActorRef;
  16
  17 import com.google.common.util.concurrent.Futures;
  18 import com.google.common.util.concurrent.ListeningExecutorService;
  19 import com.google.common.util.concurrent.MoreExecutors;
  20
  21 import org.junit.BeforeClass;
  22 import org.junit.Test;
  23 import org.mockito.Mockito;
  24 import org.opendaylight.controller.cluster.datastore.identifiers.ShardIdentifier;
  25 import org.opendaylight.controller.cluster.datastore.messages.ForwardedCommitTransaction;
  26 import org.opendaylight.controller.cluster.datastore.modification.CompositeModification;
  27 import org.opendaylight.controller.cluster.datastore.modification.Modification;
  28 import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
  29 import org.opendaylight.controller.md.sal.common.api.data.OptimisticLockFailedException;
  30 import org.opendaylight.controller.md.sal.dom.store.impl.InMemoryDOMDataStore;
  31 import org.opendaylight.controller.protobuff.messages.cohort3pc.ThreePhaseCommitCohortMessages;
  32 import org.opendaylight.controller.protobuff.messages.common.NormalizedNodeMessages;
  33 import org.opendaylight.controller.protobuff.messages.persistent.PersistentMessages;
  34 import org.opendaylight.controller.protobuff.messages.transaction.ShardTransactionMessages;
  35 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
  36 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
  37
  38 import scala.concurrent.Await;
  39 import scala.concurrent.Future;
  40 import scala.concurrent.duration.Duration;
  41 import scala.concurrent.duration.FiniteDuration;
  42
  43 import java.util.Collections;
  44 import java.util.concurrent.TimeUnit;
  45
  46 import static org.junit.Assert.assertTrue;
  47 import static org.mockito.Mockito.when;
  48
  49
  50 public class ThreePhaseCommitCohortFailureTest extends AbstractActorTest {
  51
  52     private static ListeningExecutorService storeExecutor =
  53         MoreExecutors.listeningDecorator(MoreExecutors.sameThreadExecutor());
  54
  55     private static final InMemoryDOMDataStore store =
  56         new InMemoryDOMDataStore("OPER", storeExecutor,
  57             MoreExecutors.sameThreadExecutor());
  58
  59     private static final SchemaContext testSchemaContext =
  60         TestModel.createTestContext();
  61
  62     private static final ShardIdentifier SHARD_IDENTIFIER =
  63         ShardIdentifier.builder().memberName("member-1")
  64             .shardName("inventory").type("config").build();
  65
  66     private final ShardContext shardContext = new ShardContext();
  67
  68     @BeforeClass
  69     public static void staticSetup() {
  70         store.onGlobalContextUpdated(testSchemaContext);
  71     }
  72
  73     private final FiniteDuration ASK_RESULT_DURATION = Duration.create(5000, TimeUnit.MILLISECONDS);
  74
  75
  76     @Test(expected = TestException.class)
  77     public void testNegativeAbortResultsInException() throws Exception {
  78
  79         final ActorRef shard = getSystem().actorOf(Shard.props(SHARD_IDENTIFIER,
  80                 Collections.EMPTY_MAP, shardContext));
  81         final DOMStoreThreePhaseCommitCohort mockCohort = Mockito
  82             .mock(DOMStoreThreePhaseCommitCohort.class);
  83         final CompositeModification mockComposite =
  84             Mockito.mock(CompositeModification.class);
  85         final Props props =
  86             ThreePhaseCommitCohort.props(mockCohort, shard, mockComposite);
  87
  88         final TestActorRef<ThreePhaseCommitCohort> subject = TestActorRef
  89             .create(getSystem(), props,
  90                 "testNegativeAbortResultsInException");
  91
  92         when(mockCohort.abort()).thenReturn(
  93             Futures.<Void>immediateFailedFuture(new TestException()));
  94
  95         Future<Object> future =
  96             akka.pattern.Patterns.ask(subject,
  97                 ThreePhaseCommitCohortMessages.AbortTransaction.newBuilder()
  98                     .build(), 3000);
  99         assertTrue(future.isCompleted());
 100
 101         Await.result(future, ASK_RESULT_DURATION);
 102     }
 103
 104
 105     @Test(expected = OptimisticLockFailedException.class)
 106     public void testNegativeCanCommitResultsInException() throws Exception {
 107
 108         final ActorRef shard = getSystem().actorOf(Shard.props(SHARD_IDENTIFIER,
 109                 Collections.EMPTY_MAP, shardContext));
 110         final DOMStoreThreePhaseCommitCohort mockCohort = Mockito
 111             .mock(DOMStoreThreePhaseCommitCohort.class);
 112         final CompositeModification mockComposite =
 113             Mockito.mock(CompositeModification.class);
 114         final Props props =
 115             ThreePhaseCommitCohort.props(mockCohort, shard, mockComposite);
 116
 117         final TestActorRef<ThreePhaseCommitCohort> subject = TestActorRef
 118             .create(getSystem(), props,
 119                 "testNegativeCanCommitResultsInException");
 120
 121         when(mockCohort.canCommit()).thenReturn(
 122             Futures
 123                 .<Boolean>immediateFailedFuture(
 124                     new OptimisticLockFailedException("some exception")));
 125
 126         Future<Object> future =
 127             akka.pattern.Patterns.ask(subject,
 128                 ThreePhaseCommitCohortMessages.CanCommitTransaction.newBuilder()
 129                     .build(), 3000);
 130
 131
 132         Await.result(future, ASK_RESULT_DURATION);
 133
 134     }
 135
 136
 137     @Test(expected = TestException.class)
 138     public void testNegativePreCommitResultsInException() throws Exception {
 139
 140         final ActorRef shard = getSystem().actorOf(Shard.props(SHARD_IDENTIFIER,
 141                 Collections.EMPTY_MAP, shardContext));
 142         final DOMStoreThreePhaseCommitCohort mockCohort = Mockito
 143             .mock(DOMStoreThreePhaseCommitCohort.class);
 144         final CompositeModification mockComposite =
 145             Mockito.mock(CompositeModification.class);
 146         final Props props =
 147             ThreePhaseCommitCohort.props(mockCohort, shard, mockComposite);
 148
 149         final TestActorRef<ThreePhaseCommitCohort> subject = TestActorRef
 150             .create(getSystem(), props,
 151                 "testNegativePreCommitResultsInException");
 152
 153         when(mockCohort.preCommit()).thenReturn(
 154             Futures
 155                 .<Void>immediateFailedFuture(
 156                     new TestException()));
 157
 158         Future<Object> future =
 159             akka.pattern.Patterns.ask(subject,
 160                 ThreePhaseCommitCohortMessages.PreCommitTransaction.newBuilder()
 161                     .build(), 3000);
 162
 163         Await.result(future, ASK_RESULT_DURATION);
 164
 165     }
 166
 167     @Test(expected = TestException.class)
 168     public void testNegativeCommitResultsInException() throws Exception {
 169
 170         final TestActorRef<Shard> subject = TestActorRef.create(getSystem(),
 171                 Shard.props(SHARD_IDENTIFIER, Collections.EMPTY_MAP, shardContext),
 172                 "testNegativeCommitResultsInException");
 173
 174         final ActorRef shardTransaction =
 175             getSystem().actorOf(ShardTransaction.props(store.newReadWriteTransaction(), subject,
 176                     testSchemaContext, shardContext));
 177
 178         ShardTransactionMessages.WriteData writeData =
 179             ShardTransactionMessages.WriteData.newBuilder()
 180                 .setInstanceIdentifierPathArguments(
 181                     NormalizedNodeMessages.InstanceIdentifier.newBuilder()
 182                         .build()).setNormalizedNode(
 183                 NormalizedNodeMessages.Node.newBuilder().build()
 184
 185             ).build();
 186
 187         //This is done so that Modification list is updated which is used during commit
 188         Future future =
 189             akka.pattern.Patterns.ask(shardTransaction, writeData, 3000);
 190
 191         //ready transaction creates the cohort so that we get into the
 192         //block where in commmit is done
 193         ShardTransactionMessages.ReadyTransaction readyTransaction =
 194             ShardTransactionMessages.ReadyTransaction.newBuilder().build();
 195
 196         future =
 197             akka.pattern.Patterns.ask(shardTransaction, readyTransaction, 3000);
 198
 199         //but when the message is sent it will have the MockCommit object
 200         //so that we can simulate throwing of exception
 201         ForwardedCommitTransaction mockForwardCommitTransaction =
 202             Mockito.mock(ForwardedCommitTransaction.class);
 203         DOMStoreThreePhaseCommitCohort mockThreePhaseCommitTransaction =
 204             Mockito.mock(DOMStoreThreePhaseCommitCohort.class);
 205         when(mockForwardCommitTransaction.getCohort())
 206             .thenReturn(mockThreePhaseCommitTransaction);
 207         when(mockThreePhaseCommitTransaction.commit()).thenReturn(Futures
 208             .<Void>immediateFailedFuture(
 209                 new TestException()));
 210         Modification mockModification = Mockito.mock(
 211             Modification.class);
 212         when(mockForwardCommitTransaction.getModification())
 213             .thenReturn(mockModification);
 214
 215         when(mockModification.toSerializable()).thenReturn(
 216             PersistentMessages.CompositeModification.newBuilder().build());
 217
 218         future =
 219             akka.pattern.Patterns.ask(subject,
 220                 mockForwardCommitTransaction
 221                 , 3000);
 222         Await.result(future, ASK_RESULT_DURATION);
 223     }
 224
 225     private class TestException extends Exception {
 226     }
 227 }