b1b3ff4b2742075c5ca2cc0f4e009dfc13232f6b
[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / test / java / org / opendaylight / controller / cluster / datastore / DataTreeCohortActorTest.java
1 /*
2  * Copyright (c) 2018 Inocybe Technologies and others.  All rights reserved.
3  *
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
7  */
8 package org.opendaylight.controller.cluster.datastore;
9
10 import static org.junit.Assert.assertEquals;
11 import static org.junit.Assert.assertTrue;
12 import static org.mockito.Matchers.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;
17
18 import akka.actor.ActorRef;
19 import akka.pattern.Patterns;
20 import akka.util.Timeout;
21 import com.google.common.util.concurrent.Futures;
22 import com.google.common.util.concurrent.JdkFutureAdapters;
23 import com.google.common.util.concurrent.ListenableFuture;
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.ExecutorService;
29 import java.util.concurrent.Executors;
30 import java.util.concurrent.TimeUnit;
31 import org.junit.After;
32 import org.junit.Before;
33 import org.junit.Test;
34 import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
35 import org.opendaylight.controller.cluster.datastore.DataTreeCohortActor.Abort;
36 import org.opendaylight.controller.cluster.datastore.DataTreeCohortActor.CanCommit;
37 import org.opendaylight.controller.cluster.datastore.DataTreeCohortActor.Commit;
38 import org.opendaylight.controller.cluster.datastore.DataTreeCohortActor.CommitProtocolCommand;
39 import org.opendaylight.controller.cluster.datastore.DataTreeCohortActor.PreCommit;
40 import org.opendaylight.controller.cluster.datastore.DataTreeCohortActor.Success;
41 import org.opendaylight.controller.cluster.raft.TestActorFactory;
42 import org.opendaylight.mdsal.common.api.PostCanCommitStep;
43 import org.opendaylight.mdsal.common.api.PostPreCommitStep;
44 import org.opendaylight.mdsal.common.api.ThreePhaseCommitStep;
45 import org.opendaylight.mdsal.dom.api.DOMDataTreeCandidate;
46 import org.opendaylight.mdsal.dom.api.DOMDataTreeCommitCohort;
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;
51
52 /**
53  * Unit tests for DataTreeCohortActor.
54  *
55  * @author Thomas Pantelis
56  */
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);
64
65     @Before
66     public void setup() {
67         resetMockCohort();
68     }
69
70     @After
71     public void tearDown() {
72         actorFactory.close();
73     }
74
75     @Test
76     public void testSuccessfulThreePhaseCommit() throws Exception {
77         ActorRef cohortActor = newCohortActor("testSuccessfulThreePhaseCommit");
78
79         TransactionIdentifier txId = nextTransactionId();
80         askAndAwait(cohortActor, new CanCommit(txId, CANDIDATES, MOCK_SCHEMA, cohortActor));
81         verify(mockCohort).canCommit(txId, CANDIDATES, MOCK_SCHEMA);
82
83         askAndAwait(cohortActor, new PreCommit(txId));
84         verify(mockPostCanCommit).preCommit();
85
86         askAndAwait(cohortActor, new Commit(txId));
87         verify(mockPostPreCommit).commit();
88
89         resetMockCohort();
90         askAndAwait(cohortActor, new CanCommit(txId, CANDIDATES, MOCK_SCHEMA, cohortActor));
91         verify(mockCohort).canCommit(txId, CANDIDATES, MOCK_SCHEMA);
92     }
93
94     @Test
95     public void testMultipleThreePhaseCommits() throws Exception {
96         ActorRef cohortActor = newCohortActor("testMultipleThreePhaseCommits");
97
98         TransactionIdentifier txId1 = nextTransactionId();
99         TransactionIdentifier txId2 = nextTransactionId();
100
101         askAndAwait(cohortActor, new CanCommit(txId1, CANDIDATES, MOCK_SCHEMA, cohortActor));
102         askAndAwait(cohortActor, new CanCommit(txId2, CANDIDATES, MOCK_SCHEMA, cohortActor));
103
104         askAndAwait(cohortActor, new PreCommit(txId1));
105         askAndAwait(cohortActor, new PreCommit(txId2));
106
107         askAndAwait(cohortActor, new Commit(txId1));
108         askAndAwait(cohortActor, new Commit(txId2));
109     }
110
111     @SuppressWarnings("unchecked")
112     @Test
113     public void testAsyncCohort() throws Exception {
114         ExecutorService executor = Executors.newSingleThreadExecutor();
115
116         doReturn(Futures.makeChecked(executeWithDelay(executor, mockPostCanCommit),
117             ex -> new DataValidationFailedException(YangInstanceIdentifier.EMPTY, "mock")))
118                 .when(mockCohort).canCommit(any(Object.class), any(Collection.class), any(SchemaContext.class));
119
120         doReturn(JdkFutureAdapters.listenInPoolThread(executor.submit(() ->
121             mockPostPreCommit), MoreExecutors.directExecutor())).when(mockPostCanCommit).preCommit();
122
123         doReturn(JdkFutureAdapters.listenInPoolThread(executor.submit(() ->
124             null), MoreExecutors.directExecutor())).when(mockPostPreCommit).commit();
125
126         ActorRef cohortActor = newCohortActor("testAsyncCohort");
127
128         TransactionIdentifier txId = nextTransactionId();
129         askAndAwait(cohortActor, new CanCommit(txId, CANDIDATES, MOCK_SCHEMA, cohortActor));
130         verify(mockCohort).canCommit(txId, CANDIDATES, MOCK_SCHEMA);
131
132         askAndAwait(cohortActor, new PreCommit(txId));
133         verify(mockPostCanCommit).preCommit();
134
135         askAndAwait(cohortActor, new Commit(txId));
136         verify(mockPostPreCommit).commit();
137
138         executor.shutdownNow();
139     }
140
141     @SuppressWarnings("unchecked")
142     @Test
143     public void testFailureOnCanCommit() throws Exception {
144         DataValidationFailedException failure = new DataValidationFailedException(YangInstanceIdentifier.EMPTY, "mock");
145         doReturn(Futures.immediateFailedCheckedFuture(failure)).when(mockCohort).canCommit(any(Object.class),
146                 any(Collection.class), any(SchemaContext.class));
147
148         ActorRef cohortActor = newCohortActor("testFailureOnCanCommit");
149
150         TransactionIdentifier txId = nextTransactionId();
151         try {
152             askAndAwait(cohortActor, new CanCommit(txId, CANDIDATES, MOCK_SCHEMA, cohortActor));
153         } catch (DataValidationFailedException e) {
154             assertEquals("DataValidationFailedException", failure, e);
155         }
156
157         resetMockCohort();
158         askAndAwait(cohortActor, new CanCommit(txId, CANDIDATES, MOCK_SCHEMA, cohortActor));
159         verify(mockCohort).canCommit(txId, CANDIDATES, MOCK_SCHEMA);
160     }
161
162     @Test
163     public void testAbortAfterCanCommit() throws Exception {
164         ActorRef cohortActor = newCohortActor("testAbortAfterCanCommit");
165
166         TransactionIdentifier txId = nextTransactionId();
167         askAndAwait(cohortActor, new CanCommit(txId, CANDIDATES, MOCK_SCHEMA, cohortActor));
168         verify(mockCohort).canCommit(txId, CANDIDATES, MOCK_SCHEMA);
169
170         askAndAwait(cohortActor, new Abort(txId));
171         verify(mockPostCanCommit).abort();
172
173         resetMockCohort();
174         askAndAwait(cohortActor, new CanCommit(txId, CANDIDATES, MOCK_SCHEMA, cohortActor));
175         verify(mockCohort).canCommit(txId, CANDIDATES, MOCK_SCHEMA);
176     }
177
178     @Test
179     public void testAbortAfterPreCommit() throws Exception {
180         ActorRef cohortActor = newCohortActor("testAbortAfterPreCommit");
181
182         TransactionIdentifier txId = nextTransactionId();
183         askAndAwait(cohortActor, new CanCommit(txId, CANDIDATES, MOCK_SCHEMA, cohortActor));
184         verify(mockCohort).canCommit(txId, CANDIDATES, MOCK_SCHEMA);
185
186         askAndAwait(cohortActor, new PreCommit(txId));
187         verify(mockPostCanCommit).preCommit();
188
189         askAndAwait(cohortActor, new Abort(txId));
190         verify(mockPostPreCommit).abort();
191     }
192
193     private <T> ListenableFuture<T> executeWithDelay(ExecutorService executor, T result) {
194         return JdkFutureAdapters.listenInPoolThread(executor.submit(() -> {
195             Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
196             return result;
197         }), MoreExecutors.directExecutor());
198     }
199
200     private ActorRef newCohortActor(String name) {
201         return actorFactory.createActor(DataTreeCohortActor.props(mockCohort, YangInstanceIdentifier.EMPTY), name);
202     }
203
204     @SuppressWarnings("unchecked")
205     private void resetMockCohort() {
206         reset(mockCohort);
207         doReturn(ThreePhaseCommitStep.NOOP_ABORT_FUTURE).when(mockPostCanCommit).abort();
208         doReturn(Futures.immediateFuture(mockPostPreCommit)).when(mockPostCanCommit).preCommit();
209         doReturn(Futures.immediateCheckedFuture(mockPostCanCommit)).when(mockCohort).canCommit(any(Object.class),
210                 any(Collection.class), any(SchemaContext.class));
211
212         doReturn(ThreePhaseCommitStep.NOOP_ABORT_FUTURE).when(mockPostPreCommit).abort();
213         doReturn(Futures.immediateFuture(null)).when(mockPostPreCommit).commit();
214     }
215
216     private static void askAndAwait(ActorRef actor, CommitProtocolCommand<?> message) throws Exception {
217         Timeout timeout = new Timeout(5, TimeUnit.SECONDS);
218         Object result = Await.result(Patterns.ask(actor, message, timeout), timeout.duration());
219         assertTrue("Expected Success but was " + result, result instanceof Success);
220         assertEquals("Success", message.getTxId(), ((Success)result).getTxId());
221     }
222 }

©2013 OpenDaylight, A Linux Foundation Collaborative Project. All Rights Reserved.
OpenDaylight is a registered trademark of The OpenDaylight Project, Inc.
Linux Foundation and OpenDaylight are registered trademarks of the Linux Foundation.
Linux is a registered trademark of Linus Torvalds.