Add TransmitQueue unit tests
[controller.git] / opendaylight / md-sal / cds-access-client / src / test / java / org / opendaylight / controller / cluster / access / client / AbstractTransmitQueueTest.java
1 /*
2  * Copyright (c) 2017 Pantheon Technologies s.r.o. 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.access.client;
9
10 import static org.hamcrest.CoreMatchers.everyItem;
11 import static org.mockito.Matchers.any;
12 import static org.mockito.Mockito.mock;
13 import static org.mockito.Mockito.verify;
14 import static org.opendaylight.controller.cluster.access.client.ConnectionEntryMatcher.entryWithRequest;
15
16 import akka.actor.ActorSystem;
17 import akka.testkit.JavaTestKit;
18 import akka.testkit.TestProbe;
19 import com.google.common.base.Ticker;
20 import com.google.common.collect.Iterables;
21 import java.util.Optional;
22 import java.util.function.Consumer;
23 import org.junit.After;
24 import org.junit.Assert;
25 import org.junit.Before;
26 import org.junit.Test;
27 import org.opendaylight.controller.cluster.access.commands.TransactionFailure;
28 import org.opendaylight.controller.cluster.access.commands.TransactionPurgeRequest;
29 import org.opendaylight.controller.cluster.access.commands.TransactionPurgeResponse;
30 import org.opendaylight.controller.cluster.access.concepts.ClientIdentifier;
31 import org.opendaylight.controller.cluster.access.concepts.FrontendIdentifier;
32 import org.opendaylight.controller.cluster.access.concepts.FrontendType;
33 import org.opendaylight.controller.cluster.access.concepts.LocalHistoryIdentifier;
34 import org.opendaylight.controller.cluster.access.concepts.MemberName;
35 import org.opendaylight.controller.cluster.access.concepts.Request;
36 import org.opendaylight.controller.cluster.access.concepts.RequestSuccess;
37 import org.opendaylight.controller.cluster.access.concepts.Response;
38 import org.opendaylight.controller.cluster.access.concepts.RuntimeRequestException;
39 import org.opendaylight.controller.cluster.access.concepts.SuccessEnvelope;
40 import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
41
42 public abstract class AbstractTransmitQueueTest<T extends TransmitQueue> {
43
44     private static final FrontendIdentifier FRONTEND =
45             FrontendIdentifier.create(MemberName.forName("test"), FrontendType.forName("type-1"));
46     private static final ClientIdentifier CLIENT = ClientIdentifier.create(FRONTEND, 0);
47     protected static final LocalHistoryIdentifier HISTORY = new LocalHistoryIdentifier(CLIENT, 0);
48     protected static final TransactionIdentifier TRANSACTION_IDENTIFIER = new TransactionIdentifier(HISTORY, 0);
49     protected T queue;
50     protected ActorSystem system;
51     protected TestProbe probe;
52
53     protected abstract int getMaxInFlightMessages();
54
55     protected abstract T createQueue();
56
57     @Before
58     public void setUp() throws Exception {
59         system = ActorSystem.apply();
60         probe = new TestProbe(system);
61         queue = createQueue();
62     }
63
64     @After
65     public void tearDown() throws Exception {
66         JavaTestKit.shutdownActorSystem(system);
67     }
68
69     @Test
70     public abstract void testCanTransmitCount() throws Exception;
71
72     @Test(expected = UnsupportedOperationException.class)
73     public abstract void testTransmit() throws Exception;
74
75     @Test
76     public void testAsIterable() throws Exception {
77         final Request request = new TransactionPurgeRequest(TRANSACTION_IDENTIFIER, 0L, probe.ref());
78         final Consumer<Response<?, ?>> callback = createConsumerMock();
79         final long now = Ticker.systemTicker().read();
80         final int sentMessages = getMaxInFlightMessages() + 1;
81         for (int i = 0; i < sentMessages; i++) {
82             queue.enqueue(new ConnectionEntry(request, callback, now), now);
83         }
84         final Iterable<ConnectionEntry> entries = queue.asIterable();
85         Assert.assertEquals(sentMessages, Iterables.size(entries));
86         Assert.assertThat(entries, everyItem(entryWithRequest(request)));
87     }
88
89     @Test
90     public void testTicksStalling() throws Exception {
91         final long now = Ticker.systemTicker().read();
92         Assert.assertEquals(0, queue.ticksStalling(now));
93     }
94
95     @Test
96     public void testCompleteReponseNotMatchingRequest() throws Exception {
97         final long requestSequence = 0L;
98         final long txSequence = 0L;
99         final long sessionId = 0L;
100         final Request request = new TransactionPurgeRequest(TRANSACTION_IDENTIFIER, requestSequence, probe.ref());
101         final Consumer<Response<?, ?>> callback = createConsumerMock();
102         final long now = Ticker.systemTicker().read();
103         queue.enqueue(new ConnectionEntry(request, callback, now), now);
104         //different transaction id
105         final TransactionIdentifier anotherTxId = new TransactionIdentifier(HISTORY, 1L);
106         final RequestSuccess<?, ?> success1 = new TransactionPurgeResponse(anotherTxId, requestSequence);
107         final Optional<TransmittedConnectionEntry> completed1 =
108                 queue.complete(new SuccessEnvelope(success1, sessionId, txSequence, 1L), now);
109         Assert.assertFalse(completed1.isPresent());
110         //different response sequence
111         final long differentResponseSequence = 1L;
112         final RequestSuccess<?, ?> success2 =
113                 new TransactionPurgeResponse(TRANSACTION_IDENTIFIER, differentResponseSequence);
114         final Optional<TransmittedConnectionEntry> completed2 =
115                 queue.complete(new SuccessEnvelope(success2, sessionId, txSequence, 1L), now);
116         Assert.assertFalse(completed2.isPresent());
117         //different tx sequence
118         final long differentTxSequence = 1L;
119         final RequestSuccess<?, ?> success3 =
120                 new TransactionPurgeResponse(TRANSACTION_IDENTIFIER, requestSequence);
121         final Optional<TransmittedConnectionEntry> completed3 =
122                 queue.complete(new SuccessEnvelope(success3, sessionId, differentTxSequence, 1L), now);
123         Assert.assertFalse(completed3.isPresent());
124         //different session id
125         final long differentSessionId = 1L;
126         final RequestSuccess<?, ?> success4 =
127                 new TransactionPurgeResponse(TRANSACTION_IDENTIFIER, requestSequence);
128         final Optional<TransmittedConnectionEntry> completed4 =
129                 queue.complete(new SuccessEnvelope(success4, differentSessionId, differentTxSequence, 1L), now);
130         Assert.assertFalse(completed4.isPresent());
131     }
132
133     @Test
134     public void testIsEmpty() throws Exception {
135         Assert.assertTrue(queue.isEmpty());
136         final Request request = new TransactionPurgeRequest(TRANSACTION_IDENTIFIER, 0L, probe.ref());
137         final Consumer<Response<?, ?>> callback = createConsumerMock();
138         final long now = Ticker.systemTicker().read();
139         queue.enqueue(new ConnectionEntry(request, callback, now), now);
140         Assert.assertFalse(queue.isEmpty());
141     }
142
143     @Test
144     public void testPeek() throws Exception {
145         final Request request1 = new TransactionPurgeRequest(TRANSACTION_IDENTIFIER, 0L, probe.ref());
146         final Request request2 = new TransactionPurgeRequest(TRANSACTION_IDENTIFIER, 1L, probe.ref());
147         final Consumer<Response<?, ?>> callback = createConsumerMock();
148         final long now = Ticker.systemTicker().read();
149         final ConnectionEntry entry1 = new ConnectionEntry(request1, callback, now);
150         final ConnectionEntry entry2 = new ConnectionEntry(request2, callback, now);
151         queue.enqueue(entry1, now);
152         queue.enqueue(entry2, now);
153         Assert.assertEquals(entry1.getRequest(), queue.peek().getRequest());
154     }
155
156     @Test
157     public void testPoison() throws Exception {
158         final Request request = new TransactionPurgeRequest(TRANSACTION_IDENTIFIER, 0L, probe.ref());
159         final Consumer<Response<?, ?>> callback = createConsumerMock();
160         final long now = Ticker.systemTicker().read();
161         queue.enqueue(new ConnectionEntry(request, callback, now), now);
162         queue.poison(new RuntimeRequestException("fail", new RuntimeException("fail")));
163         verify(callback).accept(any(TransactionFailure.class));
164         Assert.assertTrue(queue.isEmpty());
165     }
166
167     @SuppressWarnings("unchecked")
168     protected static Consumer<Response<?, ?>> createConsumerMock() {
169         return mock(Consumer.class);
170     }
171 }