netconf/netconf-topology-singleton/src/test/java/org/opendaylight/netconf/topology/singleton/impl/NetconfNodeActorTest.java

   1 /*
   2  * Copyright (c) 2016 Cisco Systems, Inc. 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
   9 package org.opendaylight.netconf.topology.singleton.impl;
  10
  11 import static org.junit.Assert.assertEquals;
  12 import static org.junit.Assert.assertNull;
  13 import static org.junit.Assert.assertTrue;
  14 import static org.mockito.Matchers.any;
  15 import static org.mockito.Matchers.argThat;
  16 import static org.mockito.Mockito.doReturn;
  17 import static org.mockito.Mockito.mock;
  18 import static org.mockito.Mockito.timeout;
  19 import static org.mockito.Mockito.verify;
  20 import static org.mockito.MockitoAnnotations.initMocks;
  21 import static org.opendaylight.netconf.topology.singleton.impl.utils.NetconfTopologyUtils.DEFAULT_SCHEMA_REPOSITORY;
  22
  23 import akka.actor.ActorContext;
  24 import akka.actor.ActorRef;
  25 import akka.actor.ActorSystem;
  26 import akka.actor.Props;
  27 import akka.pattern.Patterns;
  28 import akka.testkit.JavaTestKit;
  29 import akka.testkit.TestActorRef;
  30 import akka.util.Timeout;
  31 import com.google.common.base.MoreObjects;
  32 import com.google.common.collect.ImmutableList;
  33 import com.google.common.collect.Lists;
  34 import com.google.common.net.InetAddresses;
  35 import com.google.common.util.concurrent.CheckedFuture;
  36 import com.google.common.util.concurrent.Futures;
  37 import com.google.common.util.concurrent.ListenableFuture;
  38 import com.google.common.util.concurrent.SettableFuture;
  39 import java.io.ByteArrayInputStream;
  40 import java.io.IOException;
  41 import java.io.InputStream;
  42 import java.net.InetSocketAddress;
  43 import java.util.List;
  44 import java.util.concurrent.TimeUnit;
  45 import org.junit.After;
  46 import org.junit.Before;
  47 import org.junit.Rule;
  48 import org.junit.Test;
  49 import org.junit.rules.ExpectedException;
  50 import org.mockito.ArgumentMatcher;
  51 import org.mockito.Mock;
  52 import org.opendaylight.controller.cluster.schema.provider.impl.YangTextSchemaSourceSerializationProxy;
  53 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
  54 import org.opendaylight.controller.md.sal.dom.api.DOMDataBroker;
  55 import org.opendaylight.controller.md.sal.dom.api.DOMMountPointService;
  56 import org.opendaylight.controller.md.sal.dom.api.DOMRpcException;
  57 import org.opendaylight.controller.md.sal.dom.api.DOMRpcResult;
  58 import org.opendaylight.controller.md.sal.dom.api.DOMRpcService;
  59 import org.opendaylight.controller.md.sal.dom.spi.DefaultDOMRpcResult;
  60 import org.opendaylight.netconf.sal.connect.util.RemoteDeviceId;
  61 import org.opendaylight.netconf.topology.singleton.impl.actors.NetconfNodeActor;
  62 import org.opendaylight.netconf.topology.singleton.impl.utils.ClusteringRpcException;
  63 import org.opendaylight.netconf.topology.singleton.impl.utils.NetconfTopologySetup;
  64 import org.opendaylight.netconf.topology.singleton.messages.AskForMasterMountPoint;
  65 import org.opendaylight.netconf.topology.singleton.messages.CreateInitialMasterActorData;
  66 import org.opendaylight.netconf.topology.singleton.messages.MasterActorDataInitialized;
  67 import org.opendaylight.netconf.topology.singleton.messages.RefreshSetupMasterActorData;
  68 import org.opendaylight.netconf.topology.singleton.messages.RegisterMountPoint;
  69 import org.opendaylight.yangtools.yang.common.QName;
  70 import org.opendaylight.yangtools.yang.common.RpcError;
  71 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
  72 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  73 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  74 import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
  75 import org.opendaylight.yangtools.yang.data.impl.schema.builder.impl.ImmutableContainerNodeBuilder;
  76 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
  77 import org.opendaylight.yangtools.yang.model.api.SchemaPath;
  78 import org.opendaylight.yangtools.yang.model.repo.api.MissingSchemaSourceException;
  79 import org.opendaylight.yangtools.yang.model.repo.api.RevisionSourceIdentifier;
  80 import org.opendaylight.yangtools.yang.model.repo.api.SchemaContextFactory;
  81 import org.opendaylight.yangtools.yang.model.repo.api.SchemaRepository;
  82 import org.opendaylight.yangtools.yang.model.repo.api.SchemaResolutionException;
  83 import org.opendaylight.yangtools.yang.model.repo.api.SchemaSourceException;
  84 import org.opendaylight.yangtools.yang.model.repo.api.SchemaSourceFilter;
  85 import org.opendaylight.yangtools.yang.model.repo.api.SourceIdentifier;
  86 import org.opendaylight.yangtools.yang.model.repo.api.YangTextSchemaSource;
  87 import org.opendaylight.yangtools.yang.model.repo.spi.PotentialSchemaSource;
  88 import org.opendaylight.yangtools.yang.model.repo.spi.SchemaSourceRegistration;
  89 import org.opendaylight.yangtools.yang.model.repo.spi.SchemaSourceRegistry;
  90 import scala.concurrent.Await;
  91 import scala.concurrent.Future;
  92 import scala.concurrent.duration.Duration;
  93
  94 public class NetconfNodeActorTest {
  95
  96     private static final Timeout TIMEOUT = new Timeout(Duration.create(5, "seconds"));
  97     private static ActorSystem system;
  98
  99     @Rule
 100     public final ExpectedException exception = ExpectedException.none();
 101
 102     private ActorRef masterRef;
 103     private RemoteDeviceId remoteDeviceId;
 104
 105     @Mock
 106     private DOMRpcService domRpcService;
 107     @Mock
 108     private DOMMountPointService mountPointService;
 109     @Mock
 110     private DataBroker dataBroker;
 111
 112     @Before
 113     public void setup() {
 114         initMocks(this);
 115         remoteDeviceId = new RemoteDeviceId("netconf-topology",
 116                 new InetSocketAddress(InetAddresses.forString("127.0.0.1"), 9999));
 117         final NetconfTopologySetup setup = mock(NetconfTopologySetup.class);
 118
 119         final Props props = NetconfNodeActor.props(setup, remoteDeviceId, DEFAULT_SCHEMA_REPOSITORY,
 120                 DEFAULT_SCHEMA_REPOSITORY, TIMEOUT, mountPointService);
 121
 122         system = ActorSystem.create();
 123
 124         masterRef = TestActorRef.create(system, props, "master_messages");
 125     }
 126
 127     @After
 128     public void teardown() {
 129         JavaTestKit.shutdownActorSystem(system);
 130         system = null;
 131     }
 132
 133     @Test
 134     public void testInitDataMessages() throws Exception {
 135
 136         final DOMDataBroker domDataBroker = mock(DOMDataBroker.class);
 137         final List<SourceIdentifier> sourceIdentifiers = Lists.newArrayList();
 138
 139         /* Test init master data */
 140
 141         final Future<Object> initialDataToActor =
 142                 Patterns.ask(masterRef, new CreateInitialMasterActorData(domDataBroker, sourceIdentifiers,
 143                                 domRpcService), TIMEOUT);
 144
 145         final Object success = Await.result(initialDataToActor, TIMEOUT.duration());
 146         assertTrue(success instanceof MasterActorDataInitialized);
 147
 148
 149         /* Test refresh master data */
 150
 151         final RemoteDeviceId remoteDeviceId2 = new RemoteDeviceId("netconf-topology2",
 152                 new InetSocketAddress(InetAddresses.forString("127.0.0.2"), 9999));
 153
 154         final NetconfTopologySetup setup2 = mock(NetconfTopologySetup.class);
 155
 156         final Future<Object> refreshDataToActor =
 157                 Patterns.ask(masterRef, new RefreshSetupMasterActorData(setup2, remoteDeviceId2),
 158                         TIMEOUT);
 159
 160         final Object success2 = Await.result(refreshDataToActor, TIMEOUT.duration());
 161         assertTrue(success2 instanceof MasterActorDataInitialized);
 162
 163     }
 164
 165     @Test
 166     public void testRegisterMountPointMessage() throws Exception {
 167
 168         final DOMDataBroker domDataBroker = mock(DOMDataBroker.class);
 169         final List<SourceIdentifier> sourceIdentifiers =
 170                 Lists.newArrayList(RevisionSourceIdentifier.create("testID"));
 171
 172         // init master data
 173
 174         final Future<Object> initialDataToActor =
 175                 Patterns.ask(masterRef, new CreateInitialMasterActorData(domDataBroker, sourceIdentifiers,
 176                                 domRpcService), TIMEOUT);
 177
 178         final Object successInit = Await.result(initialDataToActor, TIMEOUT.duration());
 179
 180         assertTrue(successInit instanceof MasterActorDataInitialized);
 181
 182         // test if slave get right identifiers from master
 183
 184         final Future<Object> registerMountPointFuture =
 185                 Patterns.ask(masterRef, new AskForMasterMountPoint(),
 186                         TIMEOUT);
 187
 188         final RegisterMountPoint success =
 189                 (RegisterMountPoint) Await.result(registerMountPointFuture, TIMEOUT.duration());
 190
 191         assertEquals(sourceIdentifiers, success.getSourceIndentifiers());
 192
 193     }
 194
 195     @Test
 196     public void testReceiveRegisterMountpoint() throws Exception {
 197         final NetconfTopologySetup setup = mock(NetconfTopologySetup.class);
 198         final RevisionSourceIdentifier yang1 = RevisionSourceIdentifier.create("yang1");
 199         final RevisionSourceIdentifier yang2 = RevisionSourceIdentifier.create("yang2");
 200         final SchemaSourceRegistry registry = mock(SchemaSourceRegistry.class);
 201         final SchemaRepository schemaRepository = mock(SchemaRepository.class);
 202         final SchemaSourceRegistration<?> regYang1 = mock(SchemaSourceRegistration.class);
 203         final SchemaSourceRegistration<?> regYang2 = mock(SchemaSourceRegistration.class);
 204         doReturn(regYang1).when(registry).registerSchemaSource(any(), withSourceId(yang1));
 205         doReturn(regYang2).when(registry).registerSchemaSource(any(), withSourceId(yang2));
 206         final SchemaContextFactory schemaContextFactory = mock(SchemaContextFactory.class);
 207         doReturn(schemaContextFactory).when(schemaRepository).createSchemaContextFactory(any(SchemaSourceFilter.class));
 208         final SettableFuture<SchemaContext> schemaContextFuture = SettableFuture.create();
 209         final CheckedFuture<SchemaContext, SchemaResolutionException> checkedFuture =
 210                 Futures.makeChecked(schemaContextFuture, e -> new SchemaResolutionException("fail", e));
 211         doReturn(checkedFuture).when(schemaContextFactory).createSchemaContext(any());
 212         final ActorRef slaveRef =
 213                 system.actorOf(NetconfNodeActor.props(setup, remoteDeviceId, registry, schemaRepository, TIMEOUT,
 214                         mountPointService));
 215         final List<SourceIdentifier> sources = ImmutableList.of(yang1, yang2);
 216         slaveRef.tell(new RegisterMountPoint(sources), masterRef);
 217
 218         verify(registry, timeout(1000)).registerSchemaSource(any(), withSourceId(yang1));
 219         verify(registry, timeout(1000)).registerSchemaSource(any(), withSourceId(yang2));
 220         //stop actor
 221         final Future<Boolean> stopFuture = Patterns.gracefulStop(slaveRef, TIMEOUT.duration());
 222         Await.result(stopFuture, TIMEOUT.duration());
 223         //provider should be deregistered
 224         verify(regYang1).close();
 225         verify(regYang2).close();
 226     }
 227
 228     @Test
 229     public void testYangTextSchemaSourceRequestMessage() throws Exception {
 230         final SchemaRepository schemaRepository = mock(SchemaRepository.class);
 231         final SourceIdentifier sourceIdentifier = RevisionSourceIdentifier.create("testID");
 232         final Props props = NetconfNodeActor.props(mock(NetconfTopologySetup.class), remoteDeviceId,
 233                 DEFAULT_SCHEMA_REPOSITORY, schemaRepository, TIMEOUT, mountPointService);
 234
 235         final ActorRef actorRefSchemaRepo = TestActorRef.create(system, props, "master_mocked_schema_repository");
 236         final ActorContext actorContext = mock(ActorContext.class);
 237         doReturn(system.dispatcher()).when(actorContext).dispatcher();
 238
 239         final ProxyYangTextSourceProvider proxyYang =
 240                 new ProxyYangTextSourceProvider(actorRefSchemaRepo, actorContext, TIMEOUT);
 241         // test if asking for source is resolved and sended back
 242
 243         final YangTextSchemaSource yangTextSchemaSource = new YangTextSchemaSource(sourceIdentifier) {
 244             @Override
 245             protected MoreObjects.ToStringHelper addToStringAttributes(
 246                     final MoreObjects.ToStringHelper toStringHelper) {
 247                 return null;
 248             }
 249
 250             @Override
 251             public InputStream openStream() throws IOException {
 252                 return new ByteArrayInputStream("YANG".getBytes());
 253             }
 254         };
 255
 256
 257         final ListenableFuture<YangTextSchemaSource> result = Futures.immediateFuture(yangTextSchemaSource);
 258
 259         doReturn(result).when(schemaRepository).getSchemaSource(sourceIdentifier, YangTextSchemaSource.class);
 260
 261         final Future<YangTextSchemaSourceSerializationProxy> resolvedSchema =
 262                 proxyYang.getYangTextSchemaSource(sourceIdentifier);
 263
 264         final YangTextSchemaSourceSerializationProxy success = Await.result(resolvedSchema, TIMEOUT.duration());
 265
 266         assertEquals(sourceIdentifier, success.getRepresentation().getIdentifier());
 267         assertEquals("YANG", convertStreamToString(success.getRepresentation().openStream()));
 268
 269
 270         // test if asking for source is missing
 271         exception.expect(MissingSchemaSourceException.class);
 272
 273         final SchemaSourceException schemaSourceException =
 274                 new MissingSchemaSourceException("Fail", sourceIdentifier);
 275
 276         final CheckedFuture<YangTextSchemaSource, SchemaSourceException> resultFail =
 277                 Futures.immediateFailedCheckedFuture(schemaSourceException);
 278
 279         doReturn(resultFail).when(schemaRepository).getSchemaSource(sourceIdentifier, YangTextSchemaSource.class);
 280
 281         final Future<YangTextSchemaSourceSerializationProxy> failedSchema =
 282                 proxyYang.getYangTextSchemaSource(sourceIdentifier);
 283
 284         Await.result(failedSchema, TIMEOUT.duration());
 285
 286     }
 287
 288     @Test
 289     public void testProxyDOMRpcService() throws Exception {
 290
 291         final DOMDataBroker domDataBroker = mock(DOMDataBroker.class);
 292         final List<SourceIdentifier> sourceIdentifiers =
 293                 Lists.newArrayList(RevisionSourceIdentifier.create("testID"));
 294
 295         // init master data
 296
 297         final Future<Object> initialDataToActor =
 298                 Patterns.ask(masterRef, new CreateInitialMasterActorData(domDataBroker, sourceIdentifiers,
 299                         domRpcService), TIMEOUT);
 300
 301         final Object successInit = Await.result(initialDataToActor, TIMEOUT.duration());
 302
 303         assertTrue(successInit instanceof MasterActorDataInitialized);
 304
 305         // test if slave get right identifiers from master
 306
 307         final ProxyDOMRpcService slaveDomRPCService =
 308                 new ProxyDOMRpcService(system, masterRef, remoteDeviceId, TIMEOUT);
 309
 310         final QName testQName = QName.create("", "TestQname");
 311         final SchemaPath schemaPath = SchemaPath.create(true, testQName);
 312         final NormalizedNode<?, ?> outputNode = ImmutableContainerNodeBuilder.create()
 313                 .withNodeIdentifier(new YangInstanceIdentifier.NodeIdentifier(testQName))
 314                 .withChild(ImmutableNodes.leafNode(testQName, "foo")).build();
 315         final RpcError rpcError = RpcResultBuilder.newError(RpcError.ErrorType.RPC, null, "Rpc invocation failed.");
 316         // EmptyResultResponse message
 317
 318         doReturn(Futures.immediateCheckedFuture(null)).when(domRpcService).invokeRpc(any(), any());
 319
 320         final CheckedFuture<DOMRpcResult, DOMRpcException> resultFutureEmpty =
 321                 slaveDomRPCService.invokeRpc(schemaPath, outputNode);
 322
 323         final Object resultNull = resultFutureEmpty.checkedGet(2, TimeUnit.SECONDS);
 324
 325         assertEquals(null, resultNull);
 326
 327         // InvokeRpcMessageReply message
 328
 329         doReturn(Futures.immediateCheckedFuture(new DefaultDOMRpcResult(outputNode)))
 330                 .when(domRpcService).invokeRpc(any(), any());
 331
 332         final CheckedFuture<DOMRpcResult, DOMRpcException> resultFutureNn =
 333                 slaveDomRPCService.invokeRpc(schemaPath, outputNode);
 334
 335         final DOMRpcResult resultNn = resultFutureNn.checkedGet(2, TimeUnit.SECONDS);
 336
 337         assertEquals(outputNode, resultNn.getResult());
 338         assertTrue(resultNn.getErrors().isEmpty());
 339
 340         // InvokeRpcMessageReply message only error
 341
 342         doReturn(Futures.immediateCheckedFuture(new DefaultDOMRpcResult(rpcError)))
 343                 .when(domRpcService).invokeRpc(any(), any());
 344
 345         final CheckedFuture<DOMRpcResult, DOMRpcException> resultFutureError =
 346                 slaveDomRPCService.invokeRpc(schemaPath, outputNode);
 347
 348         final DOMRpcResult resultError = resultFutureError.checkedGet(2, TimeUnit.SECONDS);
 349
 350         assertNull(resultError.getResult());
 351         assertEquals(rpcError, resultError.getErrors().iterator().next());
 352
 353         // InvokeRpcMessageReply message error + result
 354
 355         doReturn(Futures.immediateCheckedFuture(new DefaultDOMRpcResult(outputNode, rpcError)))
 356                 .when(domRpcService).invokeRpc(any(), any());
 357
 358         final CheckedFuture<DOMRpcResult, DOMRpcException> resultFutureOutputError =
 359                 slaveDomRPCService.invokeRpc(schemaPath, outputNode);
 360
 361         final DOMRpcResult resultOutputError = resultFutureOutputError.checkedGet(2, TimeUnit.SECONDS);
 362
 363         assertEquals(outputNode, resultOutputError.getResult());
 364         assertEquals(rpcError, resultOutputError.getErrors().iterator().next());
 365
 366         // Throwable message
 367
 368         exception.expect(DOMRpcException.class);
 369
 370         doReturn(Futures.immediateFailedCheckedFuture(new ClusteringRpcException("")))
 371                 .when(domRpcService).invokeRpc(any(), any());
 372
 373         final CheckedFuture<DOMRpcResult, DOMRpcException> resultFutureThrowable =
 374                 slaveDomRPCService.invokeRpc(schemaPath, outputNode);
 375
 376         resultFutureThrowable.checkedGet(2, TimeUnit.SECONDS);
 377
 378     }
 379
 380     private PotentialSchemaSource<?> withSourceId(final SourceIdentifier identifier) {
 381         return argThat(new ArgumentMatcher<PotentialSchemaSource>() {
 382             @Override
 383             public boolean matches(final Object argument) {
 384                 final PotentialSchemaSource potentialSchemaSource = (PotentialSchemaSource) argument;
 385                 return identifier.equals(potentialSchemaSource.getSourceIdentifier());
 386             }
 387         });
 388     }
 389
 390     private String convertStreamToString(final java.io.InputStream is) {
 391         final java.util.Scanner s = new java.util.Scanner(is).useDelimiter("\\A");
 392         return s.hasNext() ? s.next() : "";
 393     }
 394
 395 }