opendaylight/md-sal/cds-access-client/src/test/java/org/opendaylight/controller/cluster/access/client/AbstractClientConnectionTest.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.hasItems;
  11 import static org.mockito.Matchers.isA;
  12 import static org.mockito.Mockito.mock;
  13 import static org.mockito.Mockito.timeout;
  14 import static org.mockito.Mockito.verify;
  15 import static org.opendaylight.controller.cluster.access.client.ConnectionEntryMatcher.entryWithRequest;
  16
  17 import akka.actor.ActorRef;
  18 import akka.actor.ActorSystem;
  19 import akka.testkit.TestProbe;
  20 import akka.testkit.javadsl.TestKit;
  21 import com.google.common.collect.Iterables;
  22 import java.util.Optional;
  23 import java.util.function.Consumer;
  24 import org.junit.After;
  25 import org.junit.Assert;
  26 import org.junit.Before;
  27 import org.junit.Test;
  28 import org.mockito.MockitoAnnotations;
  29 import org.opendaylight.controller.cluster.access.commands.AbortLocalTransactionRequest;
  30 import org.opendaylight.controller.cluster.access.commands.TransactionAbortSuccess;
  31 import org.opendaylight.controller.cluster.access.commands.TransactionFailure;
  32 import org.opendaylight.controller.cluster.access.concepts.ClientIdentifier;
  33 import org.opendaylight.controller.cluster.access.concepts.FrontendIdentifier;
  34 import org.opendaylight.controller.cluster.access.concepts.FrontendType;
  35 import org.opendaylight.controller.cluster.access.concepts.LocalHistoryIdentifier;
  36 import org.opendaylight.controller.cluster.access.concepts.MemberName;
  37 import org.opendaylight.controller.cluster.access.concepts.Request;
  38 import org.opendaylight.controller.cluster.access.concepts.RequestEnvelope;
  39 import org.opendaylight.controller.cluster.access.concepts.RequestSuccess;
  40 import org.opendaylight.controller.cluster.access.concepts.Response;
  41 import org.opendaylight.controller.cluster.access.concepts.ResponseEnvelope;
  42 import org.opendaylight.controller.cluster.access.concepts.RuntimeRequestException;
  43 import org.opendaylight.controller.cluster.access.concepts.SuccessEnvelope;
  44 import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
  45
  46 public abstract class AbstractClientConnectionTest<T extends AbstractClientConnection<U>, U extends BackendInfo> {
  47
  48     protected static final MemberName MEMBER_NAME = MemberName.forName("member-1");
  49     protected static final FrontendType FRONTEND_TYPE =
  50             FrontendType.forName(ClientActorContextTest.class.getSimpleName());
  51     protected static final FrontendIdentifier FRONTEND_ID = FrontendIdentifier.create(MEMBER_NAME, FRONTEND_TYPE);
  52     protected static final ClientIdentifier CLIENT_ID = ClientIdentifier.create(FRONTEND_ID, 0);
  53     protected static final String PERSISTENCE_ID = "per-1";
  54
  55     protected T connection;
  56     protected ClientActorContext context;
  57     protected ActorSystem system;
  58     protected TestProbe backendProbe;
  59     protected TestProbe contextProbe;
  60     protected TestProbe replyToProbe;
  61
  62     @Before
  63     public void setUp() {
  64         MockitoAnnotations.initMocks(this);
  65         system = ActorSystem.apply();
  66         backendProbe = new TestProbe(system);
  67         contextProbe = new TestProbe(system);
  68         context = new ClientActorContext(contextProbe.ref(), PERSISTENCE_ID, system,
  69                 CLIENT_ID, AccessClientUtil.newMockClientActorConfig());
  70         replyToProbe = new TestProbe(system);
  71         connection = createConnection();
  72     }
  73
  74     protected abstract T createConnection();
  75
  76     @Test
  77     public void testLocalActor() {
  78         Assert.assertEquals(contextProbe.ref(), connection.localActor());
  79     }
  80
  81     @Test
  82     public abstract void testReconnectConnection();
  83
  84     @Test
  85     public void testPoison() {
  86         final Consumer<Response<?, ?>> callback = mock(Consumer.class);
  87         final Request<?, ?> request = createRequest(replyToProbe.ref());
  88         final ConnectionEntry entry = new ConnectionEntry(request, callback, 0L);
  89         connection.enqueueEntry(entry, 0L);
  90         connection.poison(new RuntimeRequestException("fail", new RuntimeException("fail")));
  91         verify(callback, timeout(1000)).accept(isA(TransactionFailure.class));
  92     }
  93
  94     @Test
  95     public void testSendRequestReceiveResponse() {
  96         final Consumer<Response<?, ?>> callback = mock(Consumer.class);
  97         final Request<?, ?> request = createRequest(replyToProbe.ref());
  98         connection.sendRequest(request, callback);
  99         final RequestEnvelope requestEnvelope = backendProbe.expectMsgClass(RequestEnvelope.class);
 100         Assert.assertEquals(request, requestEnvelope.getMessage());
 101         final LocalHistoryIdentifier historyId = new LocalHistoryIdentifier(CLIENT_ID, 0L);
 102         final RequestSuccess<?, ?> message = new TransactionAbortSuccess(new TransactionIdentifier(historyId, 0L), 0L);
 103         final ResponseEnvelope<?> envelope = new SuccessEnvelope(message, 0L, 0L, 0L);
 104         connection.receiveResponse(envelope);
 105         verify(callback, timeout(1000)).accept(isA(TransactionAbortSuccess.class));
 106     }
 107
 108     @Test
 109     public void testRun() {
 110         final ClientActorBehavior<U> behavior = mock(ClientActorBehavior.class);
 111         Assert.assertSame(behavior, connection.runTimer(behavior));
 112     }
 113
 114     @Test
 115     public void testCheckTimeoutEmptyQueue() {
 116         final Optional<Long> timeout = connection.checkTimeout(context.ticker().read());
 117         Assert.assertFalse(timeout.isPresent());
 118     }
 119
 120     @Test
 121     public void testCheckTimeout() {
 122         final Consumer<Response<?, ?>> callback = mock(Consumer.class);
 123         connection.sendRequest(createRequest(replyToProbe.ref()), callback);
 124         final long now = context.ticker().read();
 125         final Optional<Long> timeout = connection.checkTimeout(now);
 126         Assert.assertTrue(timeout.isPresent());
 127     }
 128
 129     @Test
 130     public void testReplay() {
 131         final Consumer<Response<?, ?>> callback = mock(Consumer.class);
 132         final Request<?, ?> request1 = createRequest(replyToProbe.ref());
 133         final Request<?, ?> request2 = createRequest(replyToProbe.ref());
 134         connection.sendRequest(request1, callback);
 135         connection.sendRequest(request2, callback);
 136         final Iterable<ConnectionEntry> entries = connection.startReplay();
 137         Assert.assertThat(entries, hasItems(entryWithRequest(request1), entryWithRequest(request2)));
 138         Assert.assertEquals(2, Iterables.size(entries));
 139         Iterables.removeIf(entries, e -> true);
 140         final ReconnectForwarder forwarder = mock(ReconnectForwarder.class);
 141         connection.finishReplay(forwarder);
 142     }
 143
 144     @After
 145     public void tearDown() {
 146         TestKit.shutdownActorSystem(system);
 147     }
 148
 149     protected Request<?, ?> createRequest(final ActorRef replyTo) {
 150         final TransactionIdentifier identifier =
 151                 new TransactionIdentifier(new LocalHistoryIdentifier(CLIENT_ID, 0L), 0L);
 152         return new AbortLocalTransactionRequest(identifier, replyTo);
 153     }
 154
 155 }