opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/utils/ActorContextTest.java

   1 package org.opendaylight.controller.cluster.datastore.utils;
   2
   3 import static org.junit.Assert.assertEquals;
   4 import static org.junit.Assert.assertTrue;
   5 import static org.mockito.Mockito.mock;
   6 import akka.actor.ActorRef;
   7 import akka.actor.ActorSelection;
   8 import akka.actor.Address;
   9 import akka.actor.Props;
  10 import akka.actor.UntypedActor;
  11 import akka.japi.Creator;
  12 import akka.testkit.JavaTestKit;
  13 import com.google.common.base.Optional;
  14 import java.util.concurrent.TimeUnit;
  15 import org.junit.Test;
  16 import org.opendaylight.controller.cluster.datastore.AbstractActorTest;
  17 import org.opendaylight.controller.cluster.datastore.ClusterWrapper;
  18 import org.opendaylight.controller.cluster.datastore.Configuration;
  19 import org.opendaylight.controller.cluster.datastore.messages.FindLocalShard;
  20 import org.opendaylight.controller.cluster.datastore.messages.LocalShardFound;
  21 import org.opendaylight.controller.cluster.datastore.messages.LocalShardNotFound;
  22 import scala.concurrent.Await;
  23 import scala.concurrent.Future;
  24 import scala.concurrent.duration.Duration;
  25
  26 public class ActorContextTest extends AbstractActorTest{
  27
  28     private static class MockShardManager extends UntypedActor {
  29
  30         private final boolean found;
  31         private final ActorRef actorRef;
  32
  33         private MockShardManager(boolean found, ActorRef actorRef){
  34
  35             this.found = found;
  36             this.actorRef = actorRef;
  37         }
  38
  39         @Override public void onReceive(Object message) throws Exception {
  40             if(found){
  41                 getSender().tell(new LocalShardFound(actorRef), getSelf());
  42             } else {
  43                 getSender().tell(new LocalShardNotFound(((FindLocalShard) message).getShardName()), getSelf());
  44             }
  45         }
  46
  47         private static Props props(final boolean found, final ActorRef actorRef){
  48             return Props.create(new MockShardManagerCreator(found, actorRef) );
  49         }
  50
  51         @SuppressWarnings("serial")
  52         private static class MockShardManagerCreator implements Creator<MockShardManager> {
  53             final boolean found;
  54             final ActorRef actorRef;
  55
  56             MockShardManagerCreator(boolean found, ActorRef actorRef) {
  57                 this.found = found;
  58                 this.actorRef = actorRef;
  59             }
  60
  61             @Override
  62             public MockShardManager create() throws Exception {
  63                 return new MockShardManager(found, actorRef);
  64             }
  65         }
  66     }
  67
  68     @Test
  69     public void testFindLocalShardWithShardFound(){
  70         new JavaTestKit(getSystem()) {{
  71
  72             new Within(duration("1 seconds")) {
  73                 @Override
  74                 protected void run() {
  75
  76                     ActorRef shardActorRef = getSystem().actorOf(Props.create(EchoActor.class));
  77
  78                     ActorRef shardManagerActorRef = getSystem()
  79                         .actorOf(MockShardManager.props(true, shardActorRef));
  80
  81                     ActorContext actorContext =
  82                         new ActorContext(getSystem(), shardManagerActorRef , mock(ClusterWrapper.class),
  83                             mock(Configuration.class));
  84
  85                     Optional<ActorRef> out = actorContext.findLocalShard("default");
  86
  87                     assertEquals(shardActorRef, out.get());
  88
  89
  90                     expectNoMsg();
  91                 }
  92             };
  93         }};
  94
  95     }
  96
  97     @Test
  98     public void testFindLocalShardWithShardNotFound(){
  99         new JavaTestKit(getSystem()) {{
 100             ActorRef shardManagerActorRef = getSystem()
 101                     .actorOf(MockShardManager.props(false, null));
 102
 103             ActorContext actorContext =
 104                     new ActorContext(getSystem(), shardManagerActorRef , mock(ClusterWrapper.class),
 105                             mock(Configuration.class));
 106
 107             Optional<ActorRef> out = actorContext.findLocalShard("default");
 108             assertTrue(!out.isPresent());
 109         }};
 110
 111     }
 112
 113     @Test
 114     public void testExecuteRemoteOperation() {
 115         new JavaTestKit(getSystem()) {{
 116             ActorRef shardActorRef = getSystem().actorOf(Props.create(EchoActor.class));
 117
 118             ActorRef shardManagerActorRef = getSystem()
 119                     .actorOf(MockShardManager.props(true, shardActorRef));
 120
 121             ActorContext actorContext =
 122                     new ActorContext(getSystem(), shardManagerActorRef , mock(ClusterWrapper.class),
 123                             mock(Configuration.class));
 124
 125             ActorSelection actor = actorContext.actorSelection(shardActorRef.path());
 126
 127             Object out = actorContext.executeOperation(actor, "hello");
 128
 129             assertEquals("hello", out);
 130         }};
 131     }
 132
 133     @Test
 134     public void testExecuteRemoteOperationAsync() {
 135         new JavaTestKit(getSystem()) {{
 136             ActorRef shardActorRef = getSystem().actorOf(Props.create(EchoActor.class));
 137
 138             ActorRef shardManagerActorRef = getSystem()
 139                     .actorOf(MockShardManager.props(true, shardActorRef));
 140
 141             ActorContext actorContext =
 142                     new ActorContext(getSystem(), shardManagerActorRef , mock(ClusterWrapper.class),
 143                             mock(Configuration.class));
 144
 145             ActorSelection actor = actorContext.actorSelection(shardActorRef.path());
 146
 147             Future<Object> future = actorContext.executeOperationAsync(actor, "hello");
 148
 149             try {
 150                 Object result = Await.result(future, Duration.create(3, TimeUnit.SECONDS));
 151                 assertEquals("Result", "hello", result);
 152             } catch(Exception e) {
 153                 throw new AssertionError(e);
 154             }
 155         }};
 156     }
 157
 158     @Test
 159     public void testIsPathLocal() {
 160         MockClusterWrapper clusterWrapper = new MockClusterWrapper();
 161         ActorContext actorContext = null;
 162
 163         actorContext = new ActorContext(getSystem(), null, clusterWrapper, mock(Configuration.class));
 164         assertEquals(false, actorContext.isPathLocal(null));
 165         assertEquals(false, actorContext.isPathLocal(""));
 166
 167         clusterWrapper.setSelfAddress(null);
 168         actorContext = new ActorContext(getSystem(), null, clusterWrapper, mock(Configuration.class));
 169         assertEquals(false, actorContext.isPathLocal(""));
 170
 171         // even if the path is in local format, match the primary path (first 3 elements) and return true
 172         clusterWrapper.setSelfAddress(new Address("akka", "test"));
 173         actorContext = new ActorContext(getSystem(), null, clusterWrapper, mock(Configuration.class));
 174         assertEquals(true, actorContext.isPathLocal("akka://test/user/$a"));
 175
 176         clusterWrapper.setSelfAddress(new Address("akka", "test"));
 177         actorContext = new ActorContext(getSystem(), null, clusterWrapper, mock(Configuration.class));
 178         assertEquals(true, actorContext.isPathLocal("akka://test/user/$a"));
 179
 180         clusterWrapper.setSelfAddress(new Address("akka", "test"));
 181         actorContext = new ActorContext(getSystem(), null, clusterWrapper, mock(Configuration.class));
 182         assertEquals(true, actorContext.isPathLocal("akka://test/user/token2/token3/$a"));
 183
 184         // self address of remote format,but Tx path local format.
 185         clusterWrapper.setSelfAddress(new Address("akka.tcp", "system", "127.0.0.1", 2550));
 186         actorContext = new ActorContext(getSystem(), null, clusterWrapper, mock(Configuration.class));
 187         assertEquals(true, actorContext.isPathLocal(
 188             "akka://system/user/shardmanager/shard/transaction"));
 189
 190         // self address of local format,but Tx path remote format.
 191         clusterWrapper.setSelfAddress(new Address("akka.tcp", "system"));
 192         actorContext = new ActorContext(getSystem(), null, clusterWrapper, mock(Configuration.class));
 193         assertEquals(false, actorContext.isPathLocal(
 194             "akka://system@127.0.0.1:2550/user/shardmanager/shard/transaction"));
 195
 196         //local path but not same
 197         clusterWrapper.setSelfAddress(new Address("akka", "test"));
 198         actorContext = new ActorContext(getSystem(), null, clusterWrapper, mock(Configuration.class));
 199         assertEquals(true, actorContext.isPathLocal("akka://test1/user/$a"));
 200
 201         //ip and port same
 202         clusterWrapper.setSelfAddress(new Address("akka.tcp", "system", "127.0.0.1", 2550));
 203         actorContext = new ActorContext(getSystem(), null, clusterWrapper, mock(Configuration.class));
 204         assertEquals(true, actorContext.isPathLocal("akka.tcp://system@127.0.0.1:2550/"));
 205
 206         // forward-slash missing in address
 207         clusterWrapper.setSelfAddress(new Address("akka.tcp", "system", "127.0.0.1", 2550));
 208         actorContext = new ActorContext(getSystem(), null, clusterWrapper, mock(Configuration.class));
 209         assertEquals(false, actorContext.isPathLocal("akka.tcp://system@127.0.0.1:2550"));
 210
 211         //ips differ
 212         clusterWrapper.setSelfAddress(new Address("akka.tcp", "system", "127.0.0.1", 2550));
 213         actorContext = new ActorContext(getSystem(), null, clusterWrapper, mock(Configuration.class));
 214         assertEquals(false, actorContext.isPathLocal("akka.tcp://system@127.1.0.1:2550/"));
 215
 216         //ports differ
 217         clusterWrapper.setSelfAddress(new Address("akka.tcp", "system", "127.0.0.1", 2550));
 218         actorContext = new ActorContext(getSystem(), null, clusterWrapper, mock(Configuration.class));
 219         assertEquals(false, actorContext.isPathLocal("akka.tcp://system@127.0.0.1:2551/"));
 220     }
 221
 222     @Test
 223     public void testResolvePathForRemoteActor() {
 224         ActorContext actorContext =
 225                 new ActorContext(getSystem(), mock(ActorRef.class), mock(
 226                         ClusterWrapper.class),
 227                         mock(Configuration.class));
 228
 229         String actual = actorContext.resolvePath(
 230                 "akka.tcp://system@127.0.0.1:2550/user/shardmanager/shard",
 231                 "akka://system/user/shardmanager/shard/transaction");
 232
 233         String expected = "akka.tcp://system@127.0.0.1:2550/user/shardmanager/shard/transaction";
 234
 235         assertEquals(expected, actual);
 236     }
 237
 238     @Test
 239     public void testResolvePathForLocalActor() {
 240         ActorContext actorContext =
 241                 new ActorContext(getSystem(), mock(ActorRef.class), mock(ClusterWrapper.class),
 242                         mock(Configuration.class));
 243
 244         String actual = actorContext.resolvePath(
 245                 "akka://system/user/shardmanager/shard",
 246                 "akka://system/user/shardmanager/shard/transaction");
 247
 248         String expected = "akka://system/user/shardmanager/shard/transaction";
 249
 250         assertEquals(expected, actual);
 251     }
 252
 253     @Test
 254     public void testResolvePathForRemoteActorWithProperRemoteAddress() {
 255         ActorContext actorContext =
 256                 new ActorContext(getSystem(), mock(ActorRef.class), mock(ClusterWrapper.class),
 257                         mock(Configuration.class));
 258
 259         String actual = actorContext.resolvePath(
 260                 "akka.tcp://system@7.0.0.1:2550/user/shardmanager/shard",
 261                 "akka.tcp://system@7.0.0.1:2550/user/shardmanager/shard/transaction");
 262
 263         String expected = "akka.tcp://system@7.0.0.1:2550/user/shardmanager/shard/transaction";
 264
 265         assertEquals(expected, actual);
 266     }
 267
 268 }