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

   1 package org.opendaylight.controller.cluster.datastore;
   2
   3 import akka.actor.ActorPath;
   4 import akka.actor.ActorSelection;
   5 import akka.actor.Props;
   6 import akka.dispatch.Futures;
   7
   8 import com.google.common.collect.Lists;
   9 import com.google.common.util.concurrent.ListenableFuture;
  10 import static org.junit.Assert.assertEquals;
  11 import static org.junit.Assert.assertNotNull;
  12 import static org.mockito.Mockito.doReturn;
  13 import static org.mockito.Mockito.any;
  14 import static org.mockito.Mockito.isA;
  15 import static org.mockito.Mockito.mock;
  16 import static org.mockito.Mockito.verify;
  17 import static org.mockito.Mockito.times;
  18
  19 import org.junit.Before;
  20 import org.junit.Test;
  21 import org.mockito.Mock;
  22 import org.mockito.MockitoAnnotations;
  23 import org.mockito.stubbing.Stubber;
  24 import org.opendaylight.controller.cluster.datastore.messages.AbortTransaction;
  25 import org.opendaylight.controller.cluster.datastore.messages.AbortTransactionReply;
  26 import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransaction;
  27 import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransactionReply;
  28 import org.opendaylight.controller.cluster.datastore.messages.CommitTransaction;
  29 import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
  30 import org.opendaylight.controller.cluster.datastore.messages.PreCommitTransaction;
  31 import org.opendaylight.controller.cluster.datastore.messages.PreCommitTransactionReply;
  32 import org.opendaylight.controller.cluster.datastore.messages.SerializableMessage;
  33 import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
  34 import org.opendaylight.controller.cluster.datastore.utils.MessageCollectorActor;
  35 import scala.concurrent.duration.FiniteDuration;
  36
  37 import java.util.List;
  38 import java.util.concurrent.ExecutionException;
  39
  40 public class ThreePhaseCommitCohortProxyTest extends AbstractActorTest {
  41
  42     @Mock
  43     private ActorContext actorContext;
  44
  45     @Before
  46     public void setUp() {
  47         MockitoAnnotations.initMocks(this);
  48
  49         doReturn(getSystem()).when(actorContext).getActorSystem();
  50     }
  51
  52     private ThreePhaseCommitCohortProxy setupProxy(int nCohorts) {
  53         List<ActorPath> cohorts = Lists.newArrayList();
  54         for(int i = 1; i <= nCohorts; i++) {
  55             ActorPath path = getSystem().actorOf(Props.create(MessageCollectorActor.class)).path();
  56             cohorts.add(path);
  57             doReturn(mock(ActorSelection.class)).when(actorContext).actorSelection(path);
  58         }
  59
  60         return new ThreePhaseCommitCohortProxy(actorContext, cohorts, "txn-1");
  61     }
  62
  63     private void setupMockActorContext(Class<?> requestType, Object... responses) {
  64         Stubber stubber = doReturn(responses[0] instanceof Throwable ? Futures
  65                 .failed((Throwable) responses[0]) : Futures
  66                 .successful(((SerializableMessage) responses[0]).toSerializable()));
  67
  68         for(int i = 1; i < responses.length; i++) {
  69             stubber = stubber.doReturn(responses[i] instanceof Throwable ? Futures
  70                     .failed((Throwable) responses[i]) : Futures
  71                     .successful(((SerializableMessage) responses[i]).toSerializable()));
  72         }
  73
  74         stubber.when(actorContext).executeRemoteOperationAsync(any(ActorSelection.class),
  75                 isA(requestType), any(FiniteDuration.class));
  76     }
  77
  78     private void verifyCohortInvocations(int nCohorts, Class<?> requestType) {
  79         verify(actorContext, times(nCohorts)).executeRemoteOperationAsync(
  80                 any(ActorSelection.class), isA(requestType), any(FiniteDuration.class));
  81     }
  82
  83     @Test
  84     public void testCanCommitWithOneCohort() throws Exception {
  85
  86         ThreePhaseCommitCohortProxy proxy = setupProxy(1);
  87
  88         setupMockActorContext(CanCommitTransaction.SERIALIZABLE_CLASS,
  89                 new CanCommitTransactionReply(true));
  90
  91         ListenableFuture<Boolean> future = proxy.canCommit();
  92
  93         assertEquals("canCommit", true, future.get());
  94
  95         setupMockActorContext(CanCommitTransaction.SERIALIZABLE_CLASS,
  96                 new CanCommitTransactionReply(false));
  97
  98         future = proxy.canCommit();
  99
 100         assertEquals("canCommit", false, future.get());
 101
 102         verifyCohortInvocations(2, CanCommitTransaction.SERIALIZABLE_CLASS);
 103     }
 104
 105     @Test
 106     public void testCanCommitWithMultipleCohorts() throws Exception {
 107
 108         ThreePhaseCommitCohortProxy proxy = setupProxy(2);
 109
 110         setupMockActorContext(CanCommitTransaction.SERIALIZABLE_CLASS,
 111                 new CanCommitTransactionReply(true), new CanCommitTransactionReply(true));
 112
 113         ListenableFuture<Boolean> future = proxy.canCommit();
 114
 115         assertEquals("canCommit", true, future.get());
 116
 117         verifyCohortInvocations(2, CanCommitTransaction.SERIALIZABLE_CLASS);
 118     }
 119
 120     @Test
 121     public void testCanCommitWithMultipleCohortsAndOneFailure() throws Exception {
 122
 123         ThreePhaseCommitCohortProxy proxy = setupProxy(3);
 124
 125         setupMockActorContext(CanCommitTransaction.SERIALIZABLE_CLASS,
 126                 new CanCommitTransactionReply(true), new CanCommitTransactionReply(false),
 127                 new CanCommitTransactionReply(true));
 128
 129         ListenableFuture<Boolean> future = proxy.canCommit();
 130
 131         assertEquals("canCommit", false, future.get());
 132
 133         verifyCohortInvocations(3, CanCommitTransaction.SERIALIZABLE_CLASS);
 134     }
 135
 136     @Test(expected = ExecutionException.class)
 137     public void testCanCommitWithExceptionFailure() throws Exception {
 138
 139         ThreePhaseCommitCohortProxy proxy = setupProxy(1);
 140
 141         setupMockActorContext(CanCommitTransaction.SERIALIZABLE_CLASS, new RuntimeException("mock"));
 142
 143         proxy.canCommit().get();
 144     }
 145
 146     @Test(expected = ExecutionException.class)
 147     public void testCanCommitWithInvalidResponseType() throws Exception {
 148
 149         ThreePhaseCommitCohortProxy proxy = setupProxy(1);
 150
 151         setupMockActorContext(CanCommitTransaction.SERIALIZABLE_CLASS,
 152                 new PreCommitTransactionReply());
 153
 154         proxy.canCommit().get();
 155     }
 156
 157     @Test
 158     public void testPreCommit() throws Exception {
 159         ThreePhaseCommitCohortProxy proxy = setupProxy(1);
 160
 161         setupMockActorContext(PreCommitTransaction.SERIALIZABLE_CLASS,
 162                 new PreCommitTransactionReply());
 163
 164         proxy.preCommit().get();
 165
 166         verifyCohortInvocations(1, PreCommitTransaction.SERIALIZABLE_CLASS);
 167     }
 168
 169     @Test(expected = ExecutionException.class)
 170     public void testPreCommitWithFailure() throws Exception {
 171         ThreePhaseCommitCohortProxy proxy = setupProxy(2);
 172
 173         setupMockActorContext(PreCommitTransaction.SERIALIZABLE_CLASS,
 174                 new PreCommitTransactionReply(), new RuntimeException("mock"));
 175
 176         proxy.preCommit().get();
 177     }
 178
 179     @Test
 180     public void testAbort() throws Exception {
 181         ThreePhaseCommitCohortProxy proxy = setupProxy(1);
 182
 183         setupMockActorContext(AbortTransaction.SERIALIZABLE_CLASS, new AbortTransactionReply());
 184
 185         proxy.abort().get();
 186
 187         verifyCohortInvocations(1, AbortTransaction.SERIALIZABLE_CLASS);
 188     }
 189
 190     @Test
 191     public void testAbortWithFailure() throws Exception {
 192         ThreePhaseCommitCohortProxy proxy = setupProxy(1);
 193
 194         setupMockActorContext(AbortTransaction.SERIALIZABLE_CLASS, new RuntimeException("mock"));
 195
 196         // The exception should not get propagated.
 197         proxy.abort().get();
 198
 199         verifyCohortInvocations(1, AbortTransaction.SERIALIZABLE_CLASS);
 200     }
 201
 202     @Test
 203     public void testCommit() throws Exception {
 204
 205         ThreePhaseCommitCohortProxy proxy = setupProxy(2);
 206
 207         setupMockActorContext(CommitTransaction.SERIALIZABLE_CLASS, new CommitTransactionReply(),
 208                 new CommitTransactionReply());
 209
 210         proxy.commit().get();
 211
 212         verifyCohortInvocations(2, CommitTransaction.SERIALIZABLE_CLASS);
 213     }
 214
 215     @Test(expected = ExecutionException.class)
 216     public void testCommitWithFailure() throws Exception {
 217
 218         ThreePhaseCommitCohortProxy proxy = setupProxy(2);
 219
 220         setupMockActorContext(CommitTransaction.SERIALIZABLE_CLASS, new CommitTransactionReply(),
 221                 new RuntimeException("mock"));
 222
 223         proxy.commit().get();
 224     }
 225
 226     @Test(expected = ExecutionException.class)
 227     public void teseCommitWithInvalidResponseType() throws Exception {
 228
 229         ThreePhaseCommitCohortProxy proxy = setupProxy(1);
 230
 231         setupMockActorContext(CommitTransaction.SERIALIZABLE_CLASS, new PreCommitTransactionReply());
 232
 233         proxy.commit().get();
 234     }
 235
 236     @Test
 237     public void testGetCohortPaths() {
 238
 239         ThreePhaseCommitCohortProxy proxy = setupProxy(2);
 240
 241         List<ActorPath> paths = proxy.getCohortPaths();
 242         assertNotNull("getCohortPaths returned null", paths);
 243         assertEquals("getCohortPaths size", 2, paths.size());
 244     }
 245 }