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.both;
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.JavaTestKit;
19 import akka.testkit.TestProbe;
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() throws Exception {
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 protected abstract T createTransaction(ProxyHistory parent, TransactionIdentifier id, DataTreeSnapshot snapshot);
119 public void tearDown() throws Exception {
120 JavaTestKit.shutdownActorSystem(system);
124 public abstract void testExists() throws Exception;
127 public abstract void testRead() throws Exception;
130 public abstract void testWrite() throws Exception;
133 public abstract void testMerge() throws Exception;
136 public abstract void testDelete() throws Exception;
139 public abstract void testDirectCommit() throws Exception;
142 public abstract void testCanCommit() throws Exception;
145 public abstract void testPreCommit() throws Exception;
148 public abstract void testDoCommit() throws Exception;
151 public abstract void testForwardToRemoteAbort() throws Exception;
154 public abstract void testForwardToRemoteCommit() throws Exception;
157 public void testAbortVotingFuture() throws Exception {
158 testRequestResponse(f -> transaction.abort(f), TransactionAbortRequest.class, TransactionAbortSuccess::new);
162 public void testForwardToRemotePurge() throws Exception {
163 final TestProbe probe = new TestProbe(system);
164 final TransactionPurgeRequest request = new TransactionPurgeRequest(TRANSACTION_ID, 0L, probe.ref());
165 testForwardToRemote(request, TransactionPurgeRequest.class);
169 public void testReplayMessages() throws Exception {
170 final TestProbe probe = new TestProbe(system);
171 final List<ConnectionEntry> entries = new ArrayList<>();
172 final Consumer<Response<?, ?>> callback = createCallbackMock();
173 final ReadTransactionRequest request1 =
174 new ReadTransactionRequest(TRANSACTION_ID, 2L, probe.ref(), PATH_2, true);
175 final ExistsTransactionRequest request2 =
176 new ExistsTransactionRequest(TRANSACTION_ID, 3L, probe.ref(), PATH_3, true);
177 entries.add(AccessClientUtil.createConnectionEntry(request1, callback, 0L));
178 entries.add(AccessClientUtil.createConnectionEntry(request2, callback, 0L));
179 final TransactionTester<RemoteProxyTransaction> successor = createRemoteProxyTransactionTester();
180 final AbortLocalTransactionRequest successful1 = new AbortLocalTransactionRequest(TRANSACTION_ID, probe.ref());
181 transaction.recordSuccessfulRequest(successful1);
182 final ReadTransactionRequest successful2 =
183 new ReadTransactionRequest(TRANSACTION_ID, 1L, probe.ref(), PATH_1, true);
184 transaction.recordSuccessfulRequest(successful2);
185 transaction.startReconnect();
187 final ProxyHistory mockSuccessor = mock(ProxyHistory.class);
188 when(mockSuccessor.createTransactionProxy(TRANSACTION_ID, transaction.isSnapshotOnly(), false))
189 .thenReturn(successor.getTransaction());
191 transaction.replayMessages(mockSuccessor, entries);
193 final ModifyTransactionRequest transformed = successor.expectTransactionRequest(ModifyTransactionRequest.class);
194 Assert.assertNotNull(transformed);
195 Assert.assertEquals(successful1.getSequence(), transformed.getSequence());
196 Assert.assertTrue(transformed.getPersistenceProtocol().isPresent());
197 Assert.assertEquals(PersistenceProtocol.ABORT, transformed.getPersistenceProtocol().get());
199 ReadTransactionRequest tmpRead = successor.expectTransactionRequest(ReadTransactionRequest.class);
200 Assert.assertNotNull(tmpRead);
201 Assert.assertEquals(successful2.getTarget(), tmpRead.getTarget());
202 Assert.assertEquals(successful2.getSequence(), tmpRead.getSequence());
203 Assert.assertEquals(successful2.getPath(), tmpRead.getPath());
204 Assert.assertEquals(successor.localActor(), tmpRead.getReplyTo());
206 tmpRead = successor.expectTransactionRequest(ReadTransactionRequest.class);
207 Assert.assertNotNull(tmpRead);
208 Assert.assertEquals(request1.getTarget(), tmpRead.getTarget());
209 Assert.assertEquals(request1.getSequence(), tmpRead.getSequence());
210 Assert.assertEquals(request1.getPath(), tmpRead.getPath());
211 Assert.assertEquals(successor.localActor(), tmpRead.getReplyTo());
213 final ExistsTransactionRequest tmpExist = successor.expectTransactionRequest(ExistsTransactionRequest.class);
214 Assert.assertNotNull(tmpExist);
215 Assert.assertEquals(request2.getTarget(), tmpExist.getTarget());
216 Assert.assertEquals(request2.getSequence(), tmpExist.getSequence());
217 Assert.assertEquals(request2.getPath(), tmpExist.getPath());
218 Assert.assertEquals(successor.localActor(), tmpExist.getReplyTo());
221 protected void checkModifications(final ModifyTransactionRequest modifyRequest) {
222 final List<TransactionModification> modifications = modifyRequest.getModifications();
223 Assert.assertEquals(3, modifications.size());
224 Assert.assertThat(modifications, hasItem(both(isA(TransactionWrite.class)).and(hasPath(PATH_1))));
225 Assert.assertThat(modifications, hasItem(both(isA(TransactionMerge.class)).and(hasPath(PATH_2))));
226 Assert.assertThat(modifications, hasItem(both(isA(TransactionDelete.class)).and(hasPath(PATH_3))));
229 protected void testRequestResponse(final Consumer<VotingFuture> consumer,
230 final Class<? extends TransactionRequest> expectedRequest,
231 final BiFunction<TransactionIdentifier, Long, TransactionSuccess> replySupplier)
233 final TransactionTester<T> tester = getTester();
234 final VotingFuture future = mock(VotingFuture.class);
236 consumer.accept(future);
237 final TransactionRequest req = tester.expectTransactionRequest(expectedRequest);
238 tester.replySuccess(replySupplier.apply(TRANSACTION_ID, req.getSequence()));
239 verify(future).voteYes();
242 protected <T extends TransactionRequest> T testHandleForwardedRemoteRequest(final T request) throws Exception {
243 transaction.handleReplayedRemoteRequest(request, createCallbackMock(), Ticker.systemTicker().read());
244 final RequestEnvelope envelope = backendProbe.expectMsgClass(RequestEnvelope.class);
245 final T received = (T) envelope.getMessage();
246 Assert.assertTrue(received.getClass().equals(request.getClass()));
247 Assert.assertEquals(TRANSACTION_ID, received.getTarget());
248 Assert.assertEquals(clientContextProbe.ref(), received.getReplyTo());
252 protected <T extends TransactionRequest> T testForwardToRemote(final TransactionRequest toForward,
253 final Class<T> expectedMessageClass) {
254 final Consumer<Response<?, ?>> callback = createCallbackMock();
255 final TransactionTester<RemoteProxyTransaction> transactionTester = createRemoteProxyTransactionTester();
256 final RemoteProxyTransaction successor = transactionTester.getTransaction();
257 transaction.forwardToRemote(successor, toForward, callback);
258 return transactionTester.expectTransactionRequest(expectedMessageClass);
261 protected TransactionTester<T> getTester() {
265 @SuppressWarnings("unchecked")
266 protected <T> Consumer<T> createCallbackMock() {
267 return mock(Consumer.class);
270 protected static BaseMatcher<TransactionModification> hasPath(final YangInstanceIdentifier path) {
271 return new BaseMatcher<TransactionModification>() {
274 public boolean matches(final Object item) {
275 return path.equals(((TransactionModification) item).getPath());
279 public void describeTo(final Description description) {
280 description.appendValue(path);
284 public void describeMismatch(final Object item, final Description description) {
285 final TransactionModification modification = (TransactionModification) item;
286 description.appendText("was ").appendValue(modification.getPath());
291 protected TestProbe createProbe() {
292 return new TestProbe(system);
295 protected TransactionTester<LocalReadWriteProxyTransaction> createLocalProxy() {
296 final TestProbe backendProbe = new TestProbe(system, "backend2");
297 final TestProbe clientContextProbe = new TestProbe(system, "clientContext2");
298 final ClientActorContext context =
299 AccessClientUtil.createClientActorContext(system, clientContextProbe.ref(), CLIENT_ID, PERSISTENCE_ID);
300 final ShardBackendInfo backend = new ShardBackendInfo(backendProbe.ref(), 0L, ABIVersion.BORON,
301 "default", UnsignedLong.ZERO, Optional.empty(), 3);
302 final AbstractClientConnection<ShardBackendInfo> connection =
303 AccessClientUtil.createConnectedConnection(context, 0L, backend);
304 final AbstractClientHistory history = mock(AbstractClientHistory.class);
305 final ProxyHistory parent = ProxyHistory.createClient(history, connection, HISTORY_ID);
306 final DataTreeSnapshot snapshot = mock(DataTreeSnapshot.class);
307 when(snapshot.newModification()).thenReturn(mock(CursorAwareDataTreeModification.class));
308 final LocalReadWriteProxyTransaction tx =
309 new LocalReadWriteProxyTransaction(parent, TestUtils.TRANSACTION_ID, snapshot);
310 return new TransactionTester<>(tx, connection, backendProbe);
313 protected TransactionTester<RemoteProxyTransaction> createRemoteProxyTransactionTester() {
314 final TestProbe clientContextProbe = new TestProbe(system, "remoteClientContext");
315 final TestProbe backendProbe = new TestProbe(system, "remoteBackend");
316 final AbstractClientHistory history = mock(AbstractClientHistory.class);
317 final ClientActorContext context =
318 AccessClientUtil.createClientActorContext(system, clientContextProbe.ref(), CLIENT_ID, PERSISTENCE_ID);
319 final ShardBackendInfo backend = new ShardBackendInfo(backendProbe.ref(), 0L, ABIVersion.BORON,
320 "default", UnsignedLong.ZERO, Optional.empty(), 5);
321 final AbstractClientConnection<ShardBackendInfo> connection =
322 AccessClientUtil.createConnectedConnection(context, 0L, backend);
323 final ProxyHistory proxyHistory = ProxyHistory.createClient(history, connection, HISTORY_ID);
324 final RemoteProxyTransaction transaction =
325 new RemoteProxyTransaction(proxyHistory, TRANSACTION_ID, false, false, false);
326 return new TransactionTester<>(transaction, connection, backendProbe);