2 * Copyright (c) 2018 Inocybe Technologies and others. All rights reserved.
4 * This program and the accompanying materials are made available under the
5 * terms of the Eclipse Public License v1.0 which accompanies this distribution,
6 * and is available at http://www.eclipse.org/legal/epl-v10.html
8 package org.opendaylight.controller.cluster.datastore;
10 import static org.junit.Assert.assertEquals;
11 import static org.junit.Assert.assertTrue;
12 import static org.mockito.ArgumentMatchers.any;
13 import static org.mockito.Mockito.doReturn;
14 import static org.mockito.Mockito.mock;
15 import static org.mockito.Mockito.reset;
16 import static org.mockito.Mockito.verify;
18 import akka.actor.ActorRef;
19 import akka.pattern.Patterns;
20 import akka.util.Timeout;
21 import com.google.common.util.concurrent.FluentFuture;
22 import com.google.common.util.concurrent.Futures;
23 import com.google.common.util.concurrent.ListeningExecutorService;
24 import com.google.common.util.concurrent.MoreExecutors;
25 import com.google.common.util.concurrent.Uninterruptibles;
26 import java.util.ArrayList;
27 import java.util.Collection;
28 import java.util.concurrent.Executors;
29 import java.util.concurrent.TimeUnit;
30 import org.junit.After;
31 import org.junit.Before;
32 import org.junit.Test;
33 import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
34 import org.opendaylight.controller.cluster.datastore.DataTreeCohortActor.Abort;
35 import org.opendaylight.controller.cluster.datastore.DataTreeCohortActor.CanCommit;
36 import org.opendaylight.controller.cluster.datastore.DataTreeCohortActor.Commit;
37 import org.opendaylight.controller.cluster.datastore.DataTreeCohortActor.CommitProtocolCommand;
38 import org.opendaylight.controller.cluster.datastore.DataTreeCohortActor.PreCommit;
39 import org.opendaylight.controller.cluster.datastore.DataTreeCohortActor.Success;
40 import org.opendaylight.controller.cluster.raft.TestActorFactory;
41 import org.opendaylight.mdsal.common.api.PostCanCommitStep;
42 import org.opendaylight.mdsal.common.api.PostPreCommitStep;
43 import org.opendaylight.mdsal.common.api.ThreePhaseCommitStep;
44 import org.opendaylight.mdsal.dom.api.DOMDataTreeCandidate;
45 import org.opendaylight.mdsal.dom.api.DOMDataTreeCommitCohort;
46 import org.opendaylight.yangtools.util.concurrent.FluentFutures;
47 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
48 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataValidationFailedException;
49 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
50 import scala.concurrent.Await;
53 * Unit tests for DataTreeCohortActor.
55 * @author Thomas Pantelis
57 public class DataTreeCohortActorTest extends AbstractActorTest {
58 private static final Collection<DOMDataTreeCandidate> CANDIDATES = new ArrayList<>();
59 private static final SchemaContext MOCK_SCHEMA = mock(SchemaContext.class);
60 private final TestActorFactory actorFactory = new TestActorFactory(getSystem());
61 private final DOMDataTreeCommitCohort mockCohort = mock(DOMDataTreeCommitCohort.class);
62 private final PostCanCommitStep mockPostCanCommit = mock(PostCanCommitStep.class);
63 private final PostPreCommitStep mockPostPreCommit = mock(PostPreCommitStep.class);
71 public void tearDown() {
76 public void testSuccessfulThreePhaseCommit() throws Exception {
77 ActorRef cohortActor = newCohortActor("testSuccessfulThreePhaseCommit");
79 TransactionIdentifier txId = nextTransactionId();
80 askAndAwait(cohortActor, new CanCommit(txId, CANDIDATES, MOCK_SCHEMA, cohortActor));
81 verify(mockCohort).canCommit(txId, MOCK_SCHEMA, CANDIDATES);
83 askAndAwait(cohortActor, new PreCommit(txId));
84 verify(mockPostCanCommit).preCommit();
86 askAndAwait(cohortActor, new Commit(txId));
87 verify(mockPostPreCommit).commit();
90 askAndAwait(cohortActor, new CanCommit(txId, CANDIDATES, MOCK_SCHEMA, cohortActor));
91 verify(mockCohort).canCommit(txId, MOCK_SCHEMA, CANDIDATES);
95 public void testMultipleThreePhaseCommits() throws Exception {
96 ActorRef cohortActor = newCohortActor("testMultipleThreePhaseCommits");
98 TransactionIdentifier txId1 = nextTransactionId();
99 TransactionIdentifier txId2 = nextTransactionId();
101 askAndAwait(cohortActor, new CanCommit(txId1, CANDIDATES, MOCK_SCHEMA, cohortActor));
102 askAndAwait(cohortActor, new CanCommit(txId2, CANDIDATES, MOCK_SCHEMA, cohortActor));
104 askAndAwait(cohortActor, new PreCommit(txId1));
105 askAndAwait(cohortActor, new PreCommit(txId2));
107 askAndAwait(cohortActor, new Commit(txId1));
108 askAndAwait(cohortActor, new Commit(txId2));
111 @SuppressWarnings("unchecked")
113 public void testAsyncCohort() throws Exception {
114 ListeningExecutorService executor = MoreExecutors.listeningDecorator(Executors.newSingleThreadExecutor());
116 doReturn(executeWithDelay(executor, mockPostCanCommit))
117 .when(mockCohort).canCommit(any(Object.class), any(SchemaContext.class), any(Collection.class));
119 doReturn(executor.submit(() -> mockPostPreCommit)).when(mockPostCanCommit).preCommit();
121 doReturn(executor.submit(() -> null)).when(mockPostPreCommit).commit();
123 ActorRef cohortActor = newCohortActor("testAsyncCohort");
125 TransactionIdentifier txId = nextTransactionId();
126 askAndAwait(cohortActor, new CanCommit(txId, CANDIDATES, MOCK_SCHEMA, cohortActor));
127 verify(mockCohort).canCommit(txId, MOCK_SCHEMA, CANDIDATES);
129 askAndAwait(cohortActor, new PreCommit(txId));
130 verify(mockPostCanCommit).preCommit();
132 askAndAwait(cohortActor, new Commit(txId));
133 verify(mockPostPreCommit).commit();
135 executor.shutdownNow();
138 @SuppressWarnings("unchecked")
140 public void testFailureOnCanCommit() throws Exception {
141 DataValidationFailedException failure = new DataValidationFailedException(YangInstanceIdentifier.empty(),
143 doReturn(FluentFutures.immediateFailedFluentFuture(failure)).when(mockCohort).canCommit(any(Object.class),
144 any(SchemaContext.class), any(Collection.class));
146 ActorRef cohortActor = newCohortActor("testFailureOnCanCommit");
148 TransactionIdentifier txId = nextTransactionId();
150 askAndAwait(cohortActor, new CanCommit(txId, CANDIDATES, MOCK_SCHEMA, cohortActor));
151 } catch (DataValidationFailedException e) {
152 assertEquals("DataValidationFailedException", failure, e);
156 askAndAwait(cohortActor, new CanCommit(txId, CANDIDATES, MOCK_SCHEMA, cohortActor));
157 verify(mockCohort).canCommit(txId, MOCK_SCHEMA, CANDIDATES);
161 public void testAbortAfterCanCommit() throws Exception {
162 ActorRef cohortActor = newCohortActor("testAbortAfterCanCommit");
164 TransactionIdentifier txId = nextTransactionId();
165 askAndAwait(cohortActor, new CanCommit(txId, CANDIDATES, MOCK_SCHEMA, cohortActor));
166 verify(mockCohort).canCommit(txId, MOCK_SCHEMA, CANDIDATES);
168 askAndAwait(cohortActor, new Abort(txId));
169 verify(mockPostCanCommit).abort();
172 askAndAwait(cohortActor, new CanCommit(txId, CANDIDATES, MOCK_SCHEMA, cohortActor));
173 verify(mockCohort).canCommit(txId, MOCK_SCHEMA, CANDIDATES);
177 public void testAbortAfterPreCommit() throws Exception {
178 ActorRef cohortActor = newCohortActor("testAbortAfterPreCommit");
180 TransactionIdentifier txId = nextTransactionId();
181 askAndAwait(cohortActor, new CanCommit(txId, CANDIDATES, MOCK_SCHEMA, cohortActor));
182 verify(mockCohort).canCommit(txId, MOCK_SCHEMA, CANDIDATES);
184 askAndAwait(cohortActor, new PreCommit(txId));
185 verify(mockPostCanCommit).preCommit();
187 askAndAwait(cohortActor, new Abort(txId));
188 verify(mockPostPreCommit).abort();
191 private static <T> FluentFuture<T> executeWithDelay(final ListeningExecutorService executor, final T result) {
192 return FluentFuture.from(executor.submit(() -> {
193 Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
198 private ActorRef newCohortActor(final String name) {
199 return actorFactory.createActor(DataTreeCohortActor.props(mockCohort, YangInstanceIdentifier.empty()), name);
202 @SuppressWarnings("unchecked")
203 private void resetMockCohort() {
205 doReturn(ThreePhaseCommitStep.NOOP_ABORT_FUTURE).when(mockPostCanCommit).abort();
206 doReturn(Futures.immediateFuture(mockPostPreCommit)).when(mockPostCanCommit).preCommit();
207 doReturn(FluentFutures.immediateFluentFuture(mockPostCanCommit)).when(mockCohort).canCommit(any(Object.class),
208 any(SchemaContext.class), any(Collection.class));
210 doReturn(ThreePhaseCommitStep.NOOP_ABORT_FUTURE).when(mockPostPreCommit).abort();
211 doReturn(Futures.immediateFuture(null)).when(mockPostPreCommit).commit();
214 private static void askAndAwait(final ActorRef actor, final CommitProtocolCommand<?> message) throws Exception {
215 Timeout timeout = new Timeout(5, TimeUnit.SECONDS);
216 Object result = Await.result(Patterns.ask(actor, message, timeout), timeout.duration());
217 assertTrue("Expected Success but was " + result, result instanceof Success);
218 assertEquals("Success", message.getTxId(), ((Success)result).getTxId());