2 * Copyright (c) 2016 Cisco Systems, Inc. and others. All rights reserved.
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
9 package org.opendaylight.netconf.topology.singleton.impl;
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;
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.base.Optional;
33 import com.google.common.collect.ImmutableList;
34 import com.google.common.collect.Lists;
35 import com.google.common.util.concurrent.CheckedFuture;
36 import com.google.common.util.concurrent.Futures;
37 import com.google.common.util.concurrent.SettableFuture;
38 import java.io.ByteArrayInputStream;
39 import java.io.IOException;
40 import java.io.InputStream;
41 import java.net.InetAddress;
42 import java.net.InetSocketAddress;
43 import java.net.UnknownHostException;
44 import java.util.List;
45 import java.util.concurrent.TimeUnit;
46 import org.junit.After;
47 import org.junit.Before;
48 import org.junit.Rule;
49 import org.junit.Test;
50 import org.junit.rules.ExpectedException;
51 import org.mockito.ArgumentMatcher;
52 import org.mockito.Mock;
53 import org.opendaylight.controller.cluster.schema.provider.impl.YangTextSchemaSourceSerializationProxy;
54 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
55 import org.opendaylight.controller.md.sal.dom.api.DOMDataBroker;
56 import org.opendaylight.controller.md.sal.dom.api.DOMMountPointService;
57 import org.opendaylight.controller.md.sal.dom.api.DOMRpcException;
58 import org.opendaylight.controller.md.sal.dom.api.DOMRpcResult;
59 import org.opendaylight.controller.md.sal.dom.api.DOMRpcService;
60 import org.opendaylight.controller.md.sal.dom.spi.DefaultDOMRpcResult;
61 import org.opendaylight.netconf.sal.connect.util.RemoteDeviceId;
62 import org.opendaylight.netconf.topology.singleton.impl.actors.NetconfNodeActor;
63 import org.opendaylight.netconf.topology.singleton.impl.utils.ClusteringRpcException;
64 import org.opendaylight.netconf.topology.singleton.impl.utils.NetconfTopologySetup;
65 import org.opendaylight.netconf.topology.singleton.messages.AskForMasterMountPoint;
66 import org.opendaylight.netconf.topology.singleton.messages.CreateInitialMasterActorData;
67 import org.opendaylight.netconf.topology.singleton.messages.MasterActorDataInitialized;
68 import org.opendaylight.netconf.topology.singleton.messages.RefreshSetupMasterActorData;
69 import org.opendaylight.netconf.topology.singleton.messages.RegisterMountPoint;
70 import org.opendaylight.yangtools.yang.common.QName;
71 import org.opendaylight.yangtools.yang.common.RpcError;
72 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
73 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
74 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
75 import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
76 import org.opendaylight.yangtools.yang.data.impl.schema.builder.impl.ImmutableContainerNodeBuilder;
77 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
78 import org.opendaylight.yangtools.yang.model.api.SchemaPath;
79 import org.opendaylight.yangtools.yang.model.repo.api.MissingSchemaSourceException;
80 import org.opendaylight.yangtools.yang.model.repo.api.RevisionSourceIdentifier;
81 import org.opendaylight.yangtools.yang.model.repo.api.SchemaContextFactory;
82 import org.opendaylight.yangtools.yang.model.repo.api.SchemaRepository;
83 import org.opendaylight.yangtools.yang.model.repo.api.SchemaResolutionException;
84 import org.opendaylight.yangtools.yang.model.repo.api.SchemaSourceException;
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;
94 public class NetconfNodeActorTest {
96 private static final Timeout TIMEOUT = new Timeout(Duration.create(5, "seconds"));
97 private static ActorSystem system;
100 public final ExpectedException exception = ExpectedException.none();
102 private ActorRef masterRef;
103 private RemoteDeviceId remoteDeviceId;
106 private DOMRpcService domRpcService;
108 private DOMMountPointService mountPointService;
110 private DataBroker dataBroker;
113 public void setup() throws UnknownHostException {
116 remoteDeviceId = new RemoteDeviceId("netconf-topology",
117 new InetSocketAddress(InetAddress.getByName("127.0.0.1"), 9999));
118 final NetconfTopologySetup setup = mock(NetconfTopologySetup.class);
120 final Props props = NetconfNodeActor.props(setup, remoteDeviceId, DEFAULT_SCHEMA_REPOSITORY,
121 DEFAULT_SCHEMA_REPOSITORY, TIMEOUT, mountPointService);
123 system = ActorSystem.create();
125 masterRef = TestActorRef.create(system, props, "master_messages");
129 public void teardown() {
130 JavaTestKit.shutdownActorSystem(system);
135 public void testInitDataMessages() throws Exception {
137 final DOMDataBroker domDataBroker = mock(DOMDataBroker.class);
138 final List<SourceIdentifier> sourceIdentifiers = Lists.newArrayList();
140 /* Test init master data */
142 final Future<Object> initialDataToActor =
143 Patterns.ask(masterRef, new CreateInitialMasterActorData(domDataBroker, sourceIdentifiers,
144 domRpcService), TIMEOUT);
146 final Object success = Await.result(initialDataToActor, TIMEOUT.duration());
147 assertTrue(success instanceof MasterActorDataInitialized);
150 /* Test refresh master data */
152 final RemoteDeviceId remoteDeviceId2 = new RemoteDeviceId("netconf-topology2",
153 new InetSocketAddress(InetAddress.getByName("127.0.0.2"), 9999));
155 final NetconfTopologySetup setup2 = mock(NetconfTopologySetup.class);
157 final Future<Object> refreshDataToActor =
158 Patterns.ask(masterRef, new RefreshSetupMasterActorData(setup2, remoteDeviceId2),
161 final Object success2 = Await.result(refreshDataToActor, TIMEOUT.duration());
162 assertTrue(success2 instanceof MasterActorDataInitialized);
167 public void testRegisterMountPointMessage() throws Exception {
169 final DOMDataBroker domDataBroker = mock(DOMDataBroker.class);
170 final List<SourceIdentifier> sourceIdentifiers =
171 Lists.newArrayList(RevisionSourceIdentifier.create("testID", Optional.absent()));
175 final Future<Object> initialDataToActor =
176 Patterns.ask(masterRef, new CreateInitialMasterActorData(domDataBroker, sourceIdentifiers,
177 domRpcService), TIMEOUT);
179 final Object successInit = Await.result(initialDataToActor, TIMEOUT.duration());
181 assertTrue(successInit instanceof MasterActorDataInitialized);
183 // test if slave get right identifiers from master
185 final Future<Object> registerMountPointFuture =
186 Patterns.ask(masterRef, new AskForMasterMountPoint(),
189 final RegisterMountPoint success =
190 (RegisterMountPoint) Await.result(registerMountPointFuture, TIMEOUT.duration());
192 assertEquals(sourceIdentifiers, success.getSourceIndentifiers());
197 public void testReceiveRegisterMountpoint() throws Exception {
198 final NetconfTopologySetup setup = mock(NetconfTopologySetup.class);
199 final RevisionSourceIdentifier yang1 = RevisionSourceIdentifier.create("yang1");
200 final RevisionSourceIdentifier yang2 = RevisionSourceIdentifier.create("yang2");
201 final SchemaSourceRegistry registry = mock(SchemaSourceRegistry.class);
202 final SchemaRepository schemaRepository = mock(SchemaRepository.class);
203 final SchemaSourceRegistration regYang1 = mock(SchemaSourceRegistration.class);
204 final SchemaSourceRegistration regYang2 = mock(SchemaSourceRegistration.class);
205 doReturn(regYang1).when(registry).registerSchemaSource(any(), withSourceId(yang1));
206 doReturn(regYang2).when(registry).registerSchemaSource(any(), withSourceId(yang2));
207 final SchemaContextFactory schemaContextFactory = mock(SchemaContextFactory.class);
208 doReturn(schemaContextFactory).when(schemaRepository).createSchemaContextFactory(any());
209 final SettableFuture<SchemaContext> schemaContextFuture = SettableFuture.create();
210 final CheckedFuture<SchemaContext, SchemaResolutionException> checkedFuture =
211 Futures.makeChecked(schemaContextFuture, e -> new SchemaResolutionException("fail", e));
212 doReturn(checkedFuture).when(schemaContextFactory).createSchemaContext(any());
213 final ActorRef slaveRef =
214 system.actorOf(NetconfNodeActor.props(setup, remoteDeviceId, registry, schemaRepository, TIMEOUT,
216 final List<SourceIdentifier> sources = ImmutableList.of(yang1, yang2);
217 slaveRef.tell(new RegisterMountPoint(sources), masterRef);
219 verify(registry, timeout(1000)).registerSchemaSource(any(), withSourceId(yang1));
220 verify(registry, timeout(1000)).registerSchemaSource(any(), withSourceId(yang2));
222 final Future<Boolean> stopFuture = Patterns.gracefulStop(slaveRef, TIMEOUT.duration());
223 Await.result(stopFuture, TIMEOUT.duration());
224 //provider should be deregistered
225 verify(regYang1).close();
226 verify(regYang2).close();
230 public void testYangTextSchemaSourceRequestMessage() throws Exception {
231 final SchemaRepository schemaRepository = mock(SchemaRepository.class);
232 final SourceIdentifier sourceIdentifier = RevisionSourceIdentifier.create("testID", Optional.absent());
233 final Props props = NetconfNodeActor.props(mock(NetconfTopologySetup.class), remoteDeviceId,
234 DEFAULT_SCHEMA_REPOSITORY, schemaRepository, TIMEOUT, mountPointService);
236 final ActorRef actorRefSchemaRepo = TestActorRef.create(system, props, "master_mocked_schema_repository");
237 final ActorContext actorContext = mock(ActorContext.class);
238 doReturn(system.dispatcher()).when(actorContext).dispatcher();
240 final ProxyYangTextSourceProvider proxyYang =
241 new ProxyYangTextSourceProvider(actorRefSchemaRepo, actorContext, TIMEOUT);
242 // test if asking for source is resolved and sended back
244 final YangTextSchemaSource yangTextSchemaSource = new YangTextSchemaSource(sourceIdentifier) {
246 protected MoreObjects.ToStringHelper addToStringAttributes(final MoreObjects.ToStringHelper toStringHelper) {
251 public InputStream openStream() throws IOException {
252 return new ByteArrayInputStream("YANG".getBytes());
257 final CheckedFuture<YangTextSchemaSource, SchemaSourceException> result =
258 Futures.immediateCheckedFuture(yangTextSchemaSource);
260 doReturn(result).when(schemaRepository).getSchemaSource(sourceIdentifier, YangTextSchemaSource.class);
262 final Future<YangTextSchemaSourceSerializationProxy> resolvedSchema =
263 proxyYang.getYangTextSchemaSource(sourceIdentifier);
265 final YangTextSchemaSourceSerializationProxy success = Await.result(resolvedSchema, TIMEOUT.duration());
267 assertEquals(sourceIdentifier, success.getRepresentation().getIdentifier());
268 assertEquals("YANG", convertStreamToString(success.getRepresentation().openStream()));
271 // test if asking for source is missing
272 exception.expect(MissingSchemaSourceException.class);
274 final SchemaSourceException schemaSourceException =
275 new MissingSchemaSourceException("Fail", sourceIdentifier);
277 final CheckedFuture<YangTextSchemaSource, SchemaSourceException> resultFail =
278 Futures.immediateFailedCheckedFuture(schemaSourceException);
280 doReturn(resultFail).when(schemaRepository).getSchemaSource(sourceIdentifier, YangTextSchemaSource.class);
282 final Future<YangTextSchemaSourceSerializationProxy> failedSchema =
283 proxyYang.getYangTextSchemaSource(sourceIdentifier);
285 Await.result(failedSchema, TIMEOUT.duration());
290 public void testProxyDOMRpcService() throws Exception {
292 final DOMDataBroker domDataBroker = mock(DOMDataBroker.class);
293 final List<SourceIdentifier> sourceIdentifiers =
294 Lists.newArrayList(RevisionSourceIdentifier.create("testID", Optional.absent()));
298 final Future<Object> initialDataToActor =
299 Patterns.ask(masterRef, new CreateInitialMasterActorData(domDataBroker, sourceIdentifiers,
300 domRpcService), TIMEOUT);
302 final Object successInit = Await.result(initialDataToActor, TIMEOUT.duration());
304 assertTrue(successInit instanceof MasterActorDataInitialized);
306 // test if slave get right identifiers from master
308 final ProxyDOMRpcService slaveDomRPCService = new ProxyDOMRpcService(system, masterRef, remoteDeviceId, TIMEOUT);
310 final SchemaPath schemaPath = SchemaPath.create(true, QName.create("TestQname"));
311 final NormalizedNode<?, ?> outputNode = ImmutableContainerNodeBuilder.create()
312 .withNodeIdentifier(new YangInstanceIdentifier.NodeIdentifier(QName.create("TestQname")))
313 .withChild(ImmutableNodes.leafNode(QName.create("NodeQname"), "foo")).build();
314 final RpcError rpcError = RpcResultBuilder.newError(RpcError.ErrorType.RPC, null, "Rpc invocation failed.");
315 // EmptyResultResponse message
317 doReturn(Futures.immediateCheckedFuture(null)).when(domRpcService).invokeRpc(any(), any());
319 final CheckedFuture<DOMRpcResult, DOMRpcException> resultFutureEmpty =
320 slaveDomRPCService.invokeRpc(schemaPath, outputNode);
322 final Object resultNull = resultFutureEmpty.checkedGet(2, TimeUnit.SECONDS);
324 assertEquals(null, resultNull);
326 // InvokeRpcMessageReply message
328 doReturn(Futures.immediateCheckedFuture(new DefaultDOMRpcResult(outputNode))).
329 when(domRpcService).invokeRpc(any(), any());
331 final CheckedFuture<DOMRpcResult, DOMRpcException> resultFutureNn =
332 slaveDomRPCService.invokeRpc(schemaPath, outputNode);
334 final DOMRpcResult resultNn = resultFutureNn.checkedGet(2, TimeUnit.SECONDS);
336 assertEquals(outputNode, resultNn.getResult());
337 assertTrue(resultNn.getErrors().isEmpty());
339 // InvokeRpcMessageReply message only error
341 doReturn(Futures.immediateCheckedFuture(new DefaultDOMRpcResult(rpcError)))
342 .when(domRpcService).invokeRpc(any(), any());
344 final CheckedFuture<DOMRpcResult, DOMRpcException> resultFutureError =
345 slaveDomRPCService.invokeRpc(schemaPath, outputNode);
347 final DOMRpcResult resultError = resultFutureError.checkedGet(2, TimeUnit.SECONDS);
349 assertNull(resultError.getResult());
350 assertEquals(rpcError, resultError.getErrors().iterator().next());
352 // InvokeRpcMessageReply message error + result
354 doReturn(Futures.immediateCheckedFuture(new DefaultDOMRpcResult(outputNode, rpcError)))
355 .when(domRpcService).invokeRpc(any(), any());
357 final CheckedFuture<DOMRpcResult, DOMRpcException> resultFutureOutputError =
358 slaveDomRPCService.invokeRpc(schemaPath, outputNode);
360 final DOMRpcResult resultOutputError = resultFutureOutputError.checkedGet(2, TimeUnit.SECONDS);
362 assertEquals(outputNode, resultOutputError.getResult());
363 assertEquals(rpcError, resultOutputError.getErrors().iterator().next());
367 exception.expect(DOMRpcException.class);
369 doReturn(Futures.immediateFailedCheckedFuture(new ClusteringRpcException("")))
370 .when(domRpcService).invokeRpc(any(), any());
372 final CheckedFuture<DOMRpcResult, DOMRpcException> resultFutureThrowable =
373 slaveDomRPCService.invokeRpc(schemaPath, outputNode);
375 resultFutureThrowable.checkedGet(2, TimeUnit.SECONDS);
379 private PotentialSchemaSource<?> withSourceId(final SourceIdentifier identifier) {
380 return argThat(new ArgumentMatcher<PotentialSchemaSource>() {
382 public boolean matches(final Object argument) {
383 final PotentialSchemaSource potentialSchemaSource = (PotentialSchemaSource) argument;
384 return identifier.equals(potentialSchemaSource.getSourceIdentifier());
389 private String convertStreamToString(final java.io.InputStream is) {
390 final java.util.Scanner s = new java.util.Scanner(is).useDelimiter("\\A");
391 return s.hasNext() ? s.next() : "";