opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/databroker/actors/dds/ModuleShardBackendResolverTest.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.databroker.actors.dds;
   9
  10 import static org.mockito.Mockito.mock;
  11 import static org.mockito.Mockito.when;
  12
  13 import akka.actor.ActorRef;
  14 import akka.actor.ActorSelection;
  15 import akka.actor.ActorSystem;
  16 import akka.testkit.TestProbe;
  17 import akka.testkit.javadsl.TestKit;
  18 import java.util.Collections;
  19 import java.util.concurrent.CompletionStage;
  20 import java.util.concurrent.ExecutionException;
  21 import org.junit.After;
  22 import org.junit.Assert;
  23 import org.junit.Before;
  24 import org.junit.Test;
  25 import org.mockito.Mock;
  26 import org.mockito.MockitoAnnotations;
  27 import org.opendaylight.controller.cluster.access.commands.ConnectClientFailure;
  28 import org.opendaylight.controller.cluster.access.commands.ConnectClientRequest;
  29 import org.opendaylight.controller.cluster.access.commands.ConnectClientSuccess;
  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.MemberName;
  34 import org.opendaylight.controller.cluster.access.concepts.RuntimeRequestException;
  35 import org.opendaylight.controller.cluster.datastore.messages.PrimaryShardInfo;
  36 import org.opendaylight.controller.cluster.datastore.shardstrategy.ShardStrategy;
  37 import org.opendaylight.controller.cluster.datastore.shardstrategy.ShardStrategyFactory;
  38 import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
  39 import org.opendaylight.controller.cluster.datastore.utils.PrimaryShardInfoFutureCache;
  40 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  41 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree;
  42 import scala.concurrent.Promise;
  43
  44 public class ModuleShardBackendResolverTest {
  45
  46     private static final MemberName MEMBER_NAME = MemberName.forName("member-1");
  47     private static final FrontendType FRONTEND_TYPE = FrontendType.forName("type-1");
  48     private static final FrontendIdentifier FRONTEND_ID = FrontendIdentifier.create(MEMBER_NAME, FRONTEND_TYPE);
  49     private static final ClientIdentifier CLIENT_ID = ClientIdentifier.create(FRONTEND_ID, 0);
  50
  51     private ActorSystem system;
  52     private ModuleShardBackendResolver moduleShardBackendResolver;
  53     private TestProbe contextProbe;
  54
  55     @Mock
  56     private ShardStrategyFactory shardStrategyFactory;
  57     @Mock
  58     private ShardStrategy shardStrategy;
  59     @Mock
  60     private DataTree dataTree;
  61
  62     @Before
  63     public void setUp() {
  64         MockitoAnnotations.initMocks(this);
  65         system = ActorSystem.apply();
  66         contextProbe = new TestProbe(system, "context");
  67         final ActorContext actorContext = createActorContextMock(system, contextProbe.ref());
  68         moduleShardBackendResolver = new ModuleShardBackendResolver(CLIENT_ID, actorContext);
  69         when(actorContext.getShardStrategyFactory()).thenReturn(shardStrategyFactory);
  70         when(shardStrategyFactory.getStrategy(YangInstanceIdentifier.EMPTY)).thenReturn(shardStrategy);
  71         final PrimaryShardInfoFutureCache cache = new PrimaryShardInfoFutureCache();
  72         when(actorContext.getPrimaryShardInfoCache()).thenReturn(cache);
  73     }
  74
  75     @After
  76     public void tearDown() {
  77         TestKit.shutdownActorSystem(system);
  78     }
  79
  80     @Test
  81     public void testResolveShardForPathNonNullCookie() {
  82         when(shardStrategy.findShard(YangInstanceIdentifier.EMPTY)).thenReturn("default");
  83         final Long cookie = moduleShardBackendResolver.resolveShardForPath(YangInstanceIdentifier.EMPTY);
  84         Assert.assertEquals(0L, cookie.longValue());
  85     }
  86
  87     @Test
  88     public void testResolveShardForPathNullCookie() {
  89         when(shardStrategy.findShard(YangInstanceIdentifier.EMPTY)).thenReturn("foo");
  90         final Long cookie = moduleShardBackendResolver.resolveShardForPath(YangInstanceIdentifier.EMPTY);
  91         Assert.assertEquals(1L, cookie.longValue());
  92     }
  93
  94     @Test
  95     public void testGetBackendInfo() throws Exception {
  96         final CompletionStage<ShardBackendInfo> i = moduleShardBackendResolver.getBackendInfo(0L);
  97         contextProbe.expectMsgClass(ConnectClientRequest.class);
  98         final TestProbe backendProbe = new TestProbe(system, "backend");
  99         final ConnectClientSuccess msg = new ConnectClientSuccess(CLIENT_ID, 0L, backendProbe.ref(),
 100                 Collections.emptyList(), dataTree, 3);
 101         contextProbe.reply(msg);
 102         final CompletionStage<ShardBackendInfo> stage = moduleShardBackendResolver.getBackendInfo(0L);
 103         final ShardBackendInfo shardBackendInfo = TestUtils.getWithTimeout(stage.toCompletableFuture());
 104         Assert.assertEquals(0L, shardBackendInfo.getCookie().longValue());
 105         Assert.assertEquals(dataTree, shardBackendInfo.getDataTree().get());
 106         Assert.assertEquals("default", shardBackendInfo.getShardName());
 107     }
 108
 109     @Test
 110     public void testGetBackendInfoFail() throws Exception {
 111         final CompletionStage<ShardBackendInfo> i = moduleShardBackendResolver.getBackendInfo(0L);
 112         final ConnectClientRequest req = contextProbe.expectMsgClass(ConnectClientRequest.class);
 113         final RuntimeException cause = new RuntimeException();
 114         final ConnectClientFailure response = req.toRequestFailure(new RuntimeRequestException("fail", cause));
 115         contextProbe.reply(response);
 116         final CompletionStage<ShardBackendInfo> stage = moduleShardBackendResolver.getBackendInfo(0L);
 117         final ExecutionException caught =
 118                 TestUtils.assertOperationThrowsException(() -> TestUtils.getWithTimeout(stage.toCompletableFuture()),
 119                         ExecutionException.class);
 120         Assert.assertEquals(cause, caught.getCause());
 121     }
 122
 123     @Test
 124     public void testRefreshBackendInfo() throws Exception {
 125         final CompletionStage<ShardBackendInfo> backendInfo = moduleShardBackendResolver.getBackendInfo(0L);
 126         //handle first connect
 127         contextProbe.expectMsgClass(ConnectClientRequest.class);
 128         final TestProbe staleBackendProbe = new TestProbe(system, "staleBackend");
 129         final ConnectClientSuccess msg = new ConnectClientSuccess(CLIENT_ID, 0L, staleBackendProbe.ref(),
 130                 Collections.emptyList(), dataTree, 3);
 131         contextProbe.reply(msg);
 132         //get backend info
 133         final ShardBackendInfo staleBackendInfo = TestUtils.getWithTimeout(backendInfo.toCompletableFuture());
 134         //refresh
 135         final CompletionStage<ShardBackendInfo> refreshed =
 136                 moduleShardBackendResolver.refreshBackendInfo(0L, staleBackendInfo);
 137         //stale backend info should be removed and new connect request issued to the context
 138         contextProbe.expectMsgClass(ConnectClientRequest.class);
 139         final TestProbe refreshedBackendProbe = new TestProbe(system, "refreshedBackend");
 140         final ConnectClientSuccess msg2 = new ConnectClientSuccess(CLIENT_ID, 1L, refreshedBackendProbe.ref(),
 141                 Collections.emptyList(), dataTree, 3);
 142         contextProbe.reply(msg2);
 143         final ShardBackendInfo refreshedBackendInfo = TestUtils.getWithTimeout(refreshed.toCompletableFuture());
 144         Assert.assertEquals(staleBackendInfo.getCookie(), refreshedBackendInfo.getCookie());
 145         Assert.assertEquals(refreshedBackendProbe.ref(), refreshedBackendInfo.getActor());
 146     }
 147
 148     private static ActorContext createActorContextMock(final ActorSystem system, final ActorRef actor) {
 149         final ActorContext mock = mock(ActorContext.class);
 150         final Promise<PrimaryShardInfo> promise = new scala.concurrent.impl.Promise.DefaultPromise<>();
 151         final ActorSelection selection = system.actorSelection(actor.path());
 152         final PrimaryShardInfo shardInfo = new PrimaryShardInfo(selection, (short) 0);
 153         promise.success(shardInfo);
 154         when(mock.findPrimaryShardAsync("default")).thenReturn(promise.future());
 155         return mock;
 156     }
 157 }