2 * Copyright (c) 2017 Pantheon Technologies s.r.o. 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.databroker.actors.dds;
10 import static org.hamcrest.CoreMatchers.allOf;
11 import static org.hamcrest.CoreMatchers.hasItem;
12 import static org.hamcrest.core.Is.isA;
13 import static org.mockito.Mockito.mock;
14 import static org.mockito.Mockito.verify;
15 import static org.mockito.Mockito.when;
17 import akka.actor.ActorSystem;
18 import akka.testkit.TestProbe;
19 import akka.testkit.javadsl.TestKit;
20 import com.google.common.base.Ticker;
21 import com.google.common.primitives.UnsignedLong;
22 import java.util.ArrayList;
23 import java.util.List;
24 import java.util.Optional;
25 import java.util.function.BiFunction;
26 import java.util.function.Consumer;
27 import org.hamcrest.BaseMatcher;
28 import org.hamcrest.Description;
29 import org.junit.After;
30 import org.junit.Assert;
31 import org.junit.Before;
32 import org.junit.Test;
33 import org.mockito.Mock;
34 import org.mockito.MockitoAnnotations;
35 import org.opendaylight.controller.cluster.access.ABIVersion;
36 import org.opendaylight.controller.cluster.access.client.AbstractClientConnection;
37 import org.opendaylight.controller.cluster.access.client.AccessClientUtil;
38 import org.opendaylight.controller.cluster.access.client.ClientActorContext;
39 import org.opendaylight.controller.cluster.access.client.ConnectionEntry;
40 import org.opendaylight.controller.cluster.access.commands.AbortLocalTransactionRequest;
41 import org.opendaylight.controller.cluster.access.commands.ExistsTransactionRequest;
42 import org.opendaylight.controller.cluster.access.commands.ModifyTransactionRequest;
43 import org.opendaylight.controller.cluster.access.commands.PersistenceProtocol;
44 import org.opendaylight.controller.cluster.access.commands.ReadTransactionRequest;
45 import org.opendaylight.controller.cluster.access.commands.TransactionAbortRequest;
46 import org.opendaylight.controller.cluster.access.commands.TransactionAbortSuccess;
47 import org.opendaylight.controller.cluster.access.commands.TransactionDelete;
48 import org.opendaylight.controller.cluster.access.commands.TransactionMerge;
49 import org.opendaylight.controller.cluster.access.commands.TransactionModification;
50 import org.opendaylight.controller.cluster.access.commands.TransactionPurgeRequest;
51 import org.opendaylight.controller.cluster.access.commands.TransactionRequest;
52 import org.opendaylight.controller.cluster.access.commands.TransactionSuccess;
53 import org.opendaylight.controller.cluster.access.commands.TransactionWrite;
54 import org.opendaylight.controller.cluster.access.concepts.ClientIdentifier;
55 import org.opendaylight.controller.cluster.access.concepts.LocalHistoryIdentifier;
56 import org.opendaylight.controller.cluster.access.concepts.RequestEnvelope;
57 import org.opendaylight.controller.cluster.access.concepts.Response;
58 import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
59 import org.opendaylight.yangtools.yang.common.QName;
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.tree.CursorAwareDataTreeModification;
63 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeSnapshot;
64 import org.opendaylight.yangtools.yang.data.impl.schema.Builders;
66 public abstract class AbstractProxyTransactionTest<T extends AbstractProxyTransaction> {
67 protected static final TransactionIdentifier TRANSACTION_ID = TestUtils.TRANSACTION_ID;
68 private static final ClientIdentifier CLIENT_ID = TestUtils.CLIENT_ID;
69 private static final LocalHistoryIdentifier HISTORY_ID = TestUtils.HISTORY_ID;
71 protected static final YangInstanceIdentifier PATH_1 = YangInstanceIdentifier.builder()
72 .node(QName.create("ns-1", "node-1"))
74 protected static final YangInstanceIdentifier PATH_2 = YangInstanceIdentifier.builder()
75 .node(QName.create("ns-1", "node-2"))
77 protected static final YangInstanceIdentifier PATH_3 = YangInstanceIdentifier.builder()
78 .node(QName.create("ns-1", "node-3"))
80 protected static final ContainerNode DATA_1 = Builders.containerBuilder()
81 .withNodeIdentifier(new YangInstanceIdentifier.NodeIdentifier(PATH_1.getLastPathArgument().getNodeType()))
83 protected static final ContainerNode DATA_2 = Builders.containerBuilder()
84 .withNodeIdentifier(new YangInstanceIdentifier.NodeIdentifier(PATH_2.getLastPathArgument().getNodeType()))
86 protected static final String PERSISTENCE_ID = "per-1";
89 private DataTreeSnapshot snapshot;
91 private AbstractClientHistory history;
92 private ActorSystem system;
93 private TestProbe backendProbe;
94 private TestProbe clientContextProbe;
95 private TransactionTester<T> tester;
96 protected ClientActorContext context;
97 protected T transaction;
100 public void setUp() {
101 MockitoAnnotations.initMocks(this);
102 system = ActorSystem.apply();
103 clientContextProbe = new TestProbe(system, "clientContext");
104 backendProbe = new TestProbe(system, "backend");
105 context = AccessClientUtil.createClientActorContext(system, clientContextProbe.ref(), CLIENT_ID,
107 final ShardBackendInfo backend = new ShardBackendInfo(backendProbe.ref(), 0L, ABIVersion.BORON,
108 "default", UnsignedLong.ZERO, Optional.empty(), 3);
109 final AbstractClientConnection<ShardBackendInfo> connection =
110 AccessClientUtil.createConnectedConnection(context, 0L, backend);
111 final ProxyHistory parent = ProxyHistory.createClient(history, connection, HISTORY_ID);
112 transaction = createTransaction(parent, TestUtils.TRANSACTION_ID, snapshot);
113 tester = new TransactionTester<>(transaction, connection, backendProbe);
116 @SuppressWarnings("checkstyle:hiddenField")
117 protected abstract T createTransaction(ProxyHistory parent, TransactionIdentifier id, DataTreeSnapshot snapshot);
120 public void tearDown() {
121 TestKit.shutdownActorSystem(system);
125 public abstract void testExists() throws Exception;
128 public abstract void testRead() throws Exception;
131 public abstract void testWrite();
134 public abstract void testMerge();
137 public abstract void testDelete();
140 public abstract void testDirectCommit() throws Exception;
143 public abstract void testCanCommit();
146 public abstract void testPreCommit();
149 public abstract void testDoCommit();
152 public abstract void testForwardToRemoteAbort();
155 public abstract void testForwardToRemoteCommit();
158 public void testAbortVotingFuture() {
159 testRequestResponse(f -> transaction.abort(f), TransactionAbortRequest.class, TransactionAbortSuccess::new);
163 public void testForwardToRemotePurge() {
164 final TestProbe probe = new TestProbe(system);
165 final TransactionPurgeRequest request = new TransactionPurgeRequest(TRANSACTION_ID, 0L, probe.ref());
166 testForwardToRemote(request, TransactionPurgeRequest.class);
170 public void testReplayMessages() {
171 final TestProbe probe = new TestProbe(system);
172 final List<ConnectionEntry> entries = new ArrayList<>();
173 final Consumer<Response<?, ?>> callback = createCallbackMock();
174 final ReadTransactionRequest request1 =
175 new ReadTransactionRequest(TRANSACTION_ID, 2L, probe.ref(), PATH_2, true);
176 final ExistsTransactionRequest request2 =
177 new ExistsTransactionRequest(TRANSACTION_ID, 3L, probe.ref(), PATH_3, true);
178 entries.add(AccessClientUtil.createConnectionEntry(request1, callback, 0L));
179 entries.add(AccessClientUtil.createConnectionEntry(request2, callback, 0L));
180 final TransactionTester<RemoteProxyTransaction> successor = createRemoteProxyTransactionTester();
181 final AbortLocalTransactionRequest successful1 = new AbortLocalTransactionRequest(TRANSACTION_ID, probe.ref());
182 transaction.recordSuccessfulRequest(successful1);
183 final ReadTransactionRequest successful2 =
184 new ReadTransactionRequest(TRANSACTION_ID, 1L, probe.ref(), PATH_1, true);
185 transaction.recordSuccessfulRequest(successful2);
186 transaction.startReconnect();
188 final ProxyHistory mockSuccessor = mock(ProxyHistory.class);
189 when(mockSuccessor.createTransactionProxy(TRANSACTION_ID, transaction.isSnapshotOnly(), false))
190 .thenReturn(successor.getTransaction());
192 transaction.replayMessages(mockSuccessor, entries);
194 final ModifyTransactionRequest transformed = successor.expectTransactionRequest(ModifyTransactionRequest.class);
195 Assert.assertNotNull(transformed);
196 Assert.assertEquals(successful1.getSequence(), transformed.getSequence());
197 Assert.assertTrue(transformed.getPersistenceProtocol().isPresent());
198 Assert.assertEquals(PersistenceProtocol.ABORT, transformed.getPersistenceProtocol().get());
200 ReadTransactionRequest tmpRead = successor.expectTransactionRequest(ReadTransactionRequest.class);
201 Assert.assertNotNull(tmpRead);
202 Assert.assertEquals(successful2.getTarget(), tmpRead.getTarget());
203 Assert.assertEquals(successful2.getSequence(), tmpRead.getSequence());
204 Assert.assertEquals(successful2.getPath(), tmpRead.getPath());
205 Assert.assertEquals(successor.localActor(), tmpRead.getReplyTo());
207 tmpRead = successor.expectTransactionRequest(ReadTransactionRequest.class);
208 Assert.assertNotNull(tmpRead);
209 Assert.assertEquals(request1.getTarget(), tmpRead.getTarget());
210 Assert.assertEquals(request1.getSequence(), tmpRead.getSequence());
211 Assert.assertEquals(request1.getPath(), tmpRead.getPath());
212 Assert.assertEquals(successor.localActor(), tmpRead.getReplyTo());
214 final ExistsTransactionRequest tmpExist = successor.expectTransactionRequest(ExistsTransactionRequest.class);
215 Assert.assertNotNull(tmpExist);
216 Assert.assertEquals(request2.getTarget(), tmpExist.getTarget());
217 Assert.assertEquals(request2.getSequence(), tmpExist.getSequence());
218 Assert.assertEquals(request2.getPath(), tmpExist.getPath());
219 Assert.assertEquals(successor.localActor(), tmpExist.getReplyTo());
222 protected void checkModifications(final ModifyTransactionRequest modifyRequest) {
223 final List<TransactionModification> modifications = modifyRequest.getModifications();
224 Assert.assertEquals(3, modifications.size());
225 Assert.assertThat(modifications, hasItem(allOf(isA(TransactionWrite.class), hasPath(PATH_1))));
226 Assert.assertThat(modifications, hasItem(allOf(isA(TransactionMerge.class), hasPath(PATH_2))));
227 Assert.assertThat(modifications, hasItem(allOf(isA(TransactionDelete.class), hasPath(PATH_3))));
230 @SuppressWarnings("checkstyle:hiddenField")
231 protected <R extends TransactionRequest<R>> void testRequestResponse(final Consumer<VotingFuture<Void>> consumer,
232 final Class<R> expectedRequest,
233 final BiFunction<TransactionIdentifier, Long, TransactionSuccess<?>> replySupplier) {
234 final TransactionTester<T> tester = getTester();
235 final VotingFuture<Void> future = mock(VotingFuture.class);
237 consumer.accept(future);
238 final TransactionRequest<?> req = tester.expectTransactionRequest(expectedRequest);
239 tester.replySuccess(replySupplier.apply(TRANSACTION_ID, req.getSequence()));
240 verify(future).voteYes();
243 protected <R extends TransactionRequest<R>> R testHandleForwardedRemoteRequest(final R request) {
244 transaction.handleReplayedRemoteRequest(request, createCallbackMock(), Ticker.systemTicker().read());
245 final RequestEnvelope envelope = backendProbe.expectMsgClass(RequestEnvelope.class);
246 final R received = (R) envelope.getMessage();
247 Assert.assertTrue(received.getClass().equals(request.getClass()));
248 Assert.assertEquals(TRANSACTION_ID, received.getTarget());
249 Assert.assertEquals(clientContextProbe.ref(), received.getReplyTo());
253 protected <R extends TransactionRequest<R>> R testForwardToRemote(final TransactionRequest<?> toForward,
254 final Class<R> expectedMessageClass) {
255 final Consumer<Response<?, ?>> callback = createCallbackMock();
256 final TransactionTester<RemoteProxyTransaction> transactionTester = createRemoteProxyTransactionTester();
257 final RemoteProxyTransaction successor = transactionTester.getTransaction();
258 transaction.forwardToRemote(successor, toForward, callback);
259 return transactionTester.expectTransactionRequest(expectedMessageClass);
262 protected TransactionTester<T> getTester() {
266 @SuppressWarnings("unchecked")
267 protected static <T> Consumer<T> createCallbackMock() {
268 return mock(Consumer.class);
271 protected static BaseMatcher<TransactionModification> hasPath(final YangInstanceIdentifier path) {
272 return new BaseMatcher<TransactionModification>() {
275 public boolean matches(final Object item) {
276 return path.equals(((TransactionModification) item).getPath());
280 public void describeTo(final Description description) {
281 description.appendValue(path);
285 public void describeMismatch(final Object item, final Description description) {
286 final TransactionModification modification = (TransactionModification) item;
287 description.appendText("was ").appendValue(modification.getPath());
292 protected TestProbe createProbe() {
293 return new TestProbe(system);
296 @SuppressWarnings("checkstyle:hiddenField")
297 protected TransactionTester<LocalReadWriteProxyTransaction> createLocalProxy() {
298 final TestProbe backendProbe = new TestProbe(system, "backend2");
299 final TestProbe clientContextProbe = new TestProbe(system, "clientContext2");
300 final ClientActorContext context =
301 AccessClientUtil.createClientActorContext(system, clientContextProbe.ref(), CLIENT_ID, PERSISTENCE_ID);
302 final ShardBackendInfo backend = new ShardBackendInfo(backendProbe.ref(), 0L, ABIVersion.BORON,
303 "default", UnsignedLong.ZERO, Optional.empty(), 3);
304 final AbstractClientConnection<ShardBackendInfo> connection =
305 AccessClientUtil.createConnectedConnection(context, 0L, backend);
306 final AbstractClientHistory history = mock(AbstractClientHistory.class);
307 final ProxyHistory parent = ProxyHistory.createClient(history, connection, HISTORY_ID);
308 final DataTreeSnapshot snapshot = mock(DataTreeSnapshot.class);
309 when(snapshot.newModification()).thenReturn(mock(CursorAwareDataTreeModification.class));
310 final LocalReadWriteProxyTransaction tx =
311 new LocalReadWriteProxyTransaction(parent, TestUtils.TRANSACTION_ID, snapshot);
312 return new TransactionTester<>(tx, connection, backendProbe);
315 @SuppressWarnings("checkstyle:hiddenField")
316 protected TransactionTester<RemoteProxyTransaction> createRemoteProxyTransactionTester() {
317 final TestProbe clientContextProbe = new TestProbe(system, "remoteClientContext");
318 final TestProbe backendProbe = new TestProbe(system, "remoteBackend");
319 final AbstractClientHistory history = mock(AbstractClientHistory.class);
320 final ClientActorContext context =
321 AccessClientUtil.createClientActorContext(system, clientContextProbe.ref(), CLIENT_ID, PERSISTENCE_ID);
322 final ShardBackendInfo backend = new ShardBackendInfo(backendProbe.ref(), 0L, ABIVersion.BORON,
323 "default", UnsignedLong.ZERO, Optional.empty(), 5);
324 final AbstractClientConnection<ShardBackendInfo> connection =
325 AccessClientUtil.createConnectedConnection(context, 0L, backend);
326 final ProxyHistory proxyHistory = ProxyHistory.createClient(history, connection, HISTORY_ID);
327 final RemoteProxyTransaction transaction =
328 new RemoteProxyTransaction(proxyHistory, TRANSACTION_ID, false, false, false);
329 return new TransactionTester<>(transaction, connection, backendProbe);