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.Mockito.doReturn;
16 import static org.mockito.Mockito.mock;
17 import static org.mockito.MockitoAnnotations.initMocks;
18 import static org.opendaylight.netconf.topology.singleton.impl.utils.NetconfTopologyUtils.DEFAULT_SCHEMA_REPOSITORY;
20 import akka.actor.ActorContext;
21 import akka.actor.ActorRef;
22 import akka.actor.ActorSystem;
23 import akka.actor.Props;
24 import akka.pattern.Patterns;
25 import akka.testkit.JavaTestKit;
26 import akka.testkit.TestActorRef;
27 import akka.util.Timeout;
28 import com.google.common.base.MoreObjects;
29 import com.google.common.base.Optional;
30 import com.google.common.collect.Lists;
31 import com.google.common.util.concurrent.CheckedFuture;
32 import com.google.common.util.concurrent.Futures;
33 import java.io.ByteArrayInputStream;
34 import java.io.IOException;
35 import java.io.InputStream;
36 import java.net.InetAddress;
37 import java.net.InetSocketAddress;
38 import java.net.UnknownHostException;
39 import java.util.List;
40 import java.util.concurrent.TimeUnit;
41 import org.junit.After;
42 import org.junit.Before;
43 import org.junit.Rule;
44 import org.junit.Test;
45 import org.junit.rules.ExpectedException;
46 import org.mockito.Mock;
47 import org.opendaylight.controller.cluster.schema.provider.impl.YangTextSchemaSourceSerializationProxy;
48 import org.opendaylight.controller.md.sal.dom.api.DOMDataBroker;
49 import org.opendaylight.controller.md.sal.dom.api.DOMRpcException;
50 import org.opendaylight.controller.md.sal.dom.api.DOMRpcResult;
51 import org.opendaylight.controller.md.sal.dom.api.DOMRpcService;
52 import org.opendaylight.controller.md.sal.dom.spi.DefaultDOMRpcResult;
53 import org.opendaylight.netconf.sal.connect.util.RemoteDeviceId;
54 import org.opendaylight.netconf.topology.singleton.impl.actors.NetconfNodeActor;
55 import org.opendaylight.netconf.topology.singleton.impl.utils.ClusteringRpcException;
56 import org.opendaylight.netconf.topology.singleton.impl.utils.NetconfTopologySetup;
57 import org.opendaylight.netconf.topology.singleton.messages.AskForMasterMountPoint;
58 import org.opendaylight.netconf.topology.singleton.messages.CreateInitialMasterActorData;
59 import org.opendaylight.netconf.topology.singleton.messages.MasterActorDataInitialized;
60 import org.opendaylight.netconf.topology.singleton.messages.RefreshSetupMasterActorData;
61 import org.opendaylight.netconf.topology.singleton.messages.RegisterMountPoint;
62 import org.opendaylight.yangtools.yang.common.QName;
63 import org.opendaylight.yangtools.yang.common.RpcError;
64 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
65 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
66 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
67 import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
68 import org.opendaylight.yangtools.yang.data.impl.schema.builder.impl.ImmutableContainerNodeBuilder;
69 import org.opendaylight.yangtools.yang.model.api.SchemaPath;
70 import org.opendaylight.yangtools.yang.model.repo.api.MissingSchemaSourceException;
71 import org.opendaylight.yangtools.yang.model.repo.api.RevisionSourceIdentifier;
72 import org.opendaylight.yangtools.yang.model.repo.api.SchemaRepository;
73 import org.opendaylight.yangtools.yang.model.repo.api.SchemaSourceException;
74 import org.opendaylight.yangtools.yang.model.repo.api.SourceIdentifier;
75 import org.opendaylight.yangtools.yang.model.repo.api.YangTextSchemaSource;
76 import scala.concurrent.Await;
77 import scala.concurrent.Future;
78 import scala.concurrent.duration.Duration;
80 public class NetconfNodeActorTest {
82 private static final Timeout TIMEOUT = new Timeout(Duration.create(5, "seconds"));
83 private static ActorSystem system;
86 public final ExpectedException exception = ExpectedException.none();
88 private ActorRef masterRef;
89 private RemoteDeviceId remoteDeviceId;
92 private DOMRpcService domRpcService;
95 public void setup() throws UnknownHostException {
98 remoteDeviceId = new RemoteDeviceId("netconf-topology",
99 new InetSocketAddress(InetAddress.getByName("127.0.0.1"), 9999));
100 final NetconfTopologySetup setup = mock(NetconfTopologySetup.class);
102 final Props props = NetconfNodeActor.props(setup, remoteDeviceId, DEFAULT_SCHEMA_REPOSITORY,
103 DEFAULT_SCHEMA_REPOSITORY);
105 system = ActorSystem.create();
107 masterRef = TestActorRef.create(system, props, "master_messages");
111 public void teardown() {
112 JavaTestKit.shutdownActorSystem(system);
117 public void testInitDataMessages() throws Exception {
119 final DOMDataBroker domDataBroker = mock(DOMDataBroker.class);
120 final List<SourceIdentifier> sourceIdentifiers = Lists.newArrayList();
122 /* Test init master data */
124 final Future<Object> initialDataToActor =
125 Patterns.ask(masterRef, new CreateInitialMasterActorData(domDataBroker, sourceIdentifiers,
126 domRpcService), TIMEOUT);
128 final Object success = Await.result(initialDataToActor, TIMEOUT.duration());
129 assertTrue(success instanceof MasterActorDataInitialized);
132 /* Test refresh master data */
134 final RemoteDeviceId remoteDeviceId2 = new RemoteDeviceId("netconf-topology2",
135 new InetSocketAddress(InetAddress.getByName("127.0.0.2"), 9999));
137 final NetconfTopologySetup setup2 = mock(NetconfTopologySetup.class);
139 final Future<Object> refreshDataToActor =
140 Patterns.ask(masterRef, new RefreshSetupMasterActorData(setup2, remoteDeviceId2),
143 final Object success2 = Await.result(refreshDataToActor, TIMEOUT.duration());
144 assertTrue(success2 instanceof MasterActorDataInitialized);
149 public void testRegisterMountPointMessage() throws Exception {
151 final DOMDataBroker domDataBroker = mock(DOMDataBroker.class);
152 final List<SourceIdentifier> sourceIdentifiers =
153 Lists.newArrayList(RevisionSourceIdentifier.create("testID", Optional.absent()));
157 final Future<Object> initialDataToActor =
158 Patterns.ask(masterRef, new CreateInitialMasterActorData(domDataBroker, sourceIdentifiers,
159 domRpcService), TIMEOUT);
161 final Object successInit = Await.result(initialDataToActor, TIMEOUT.duration());
163 assertTrue(successInit instanceof MasterActorDataInitialized);
165 // test if slave get right identifiers from master
167 final Future<Object> registerMountPointFuture =
168 Patterns.ask(masterRef, new AskForMasterMountPoint(),
171 final RegisterMountPoint success =
172 (RegisterMountPoint) Await.result(registerMountPointFuture, TIMEOUT.duration());
174 assertEquals(sourceIdentifiers, success.getSourceIndentifiers());
179 public void testYangTextSchemaSourceRequestMessage() throws Exception {
180 final SchemaRepository schemaRepository = mock(SchemaRepository.class);
181 final SourceIdentifier sourceIdentifier = RevisionSourceIdentifier.create("testID", Optional.absent());
182 final Props props = NetconfNodeActor.props(mock(NetconfTopologySetup.class), remoteDeviceId,
183 DEFAULT_SCHEMA_REPOSITORY, schemaRepository);
185 final ActorRef actorRefSchemaRepo = TestActorRef.create(system, props, "master_mocked_schema_repository");
186 final ActorContext actorContext = mock(ActorContext.class);
187 doReturn(system.dispatcher()).when(actorContext).dispatcher();
189 final ProxyYangTextSourceProvider proxyYang =
190 new ProxyYangTextSourceProvider(actorRefSchemaRepo, actorContext);
191 // test if asking for source is resolved and sended back
193 final YangTextSchemaSource yangTextSchemaSource = new YangTextSchemaSource(sourceIdentifier) {
195 protected MoreObjects.ToStringHelper addToStringAttributes(MoreObjects.ToStringHelper toStringHelper) {
200 public InputStream openStream() throws IOException {
201 return new ByteArrayInputStream("YANG".getBytes());
206 final CheckedFuture<YangTextSchemaSource, SchemaSourceException> result =
207 Futures.immediateCheckedFuture(yangTextSchemaSource);
209 doReturn(result).when(schemaRepository).getSchemaSource(sourceIdentifier, YangTextSchemaSource.class);
211 final Future<YangTextSchemaSourceSerializationProxy> resolvedSchema =
212 proxyYang.getYangTextSchemaSource(sourceIdentifier);
214 final YangTextSchemaSourceSerializationProxy success = Await.result(resolvedSchema, TIMEOUT.duration());
216 assertEquals(sourceIdentifier, success.getRepresentation().getIdentifier());
217 assertEquals("YANG", convertStreamToString(success.getRepresentation().openStream()));
220 // test if asking for source is missing
221 exception.expect(MissingSchemaSourceException.class);
223 final SchemaSourceException schemaSourceException =
224 new MissingSchemaSourceException("Fail", sourceIdentifier);
226 final CheckedFuture<YangTextSchemaSource, SchemaSourceException> resultFail =
227 Futures.immediateFailedCheckedFuture(schemaSourceException);
229 doReturn(resultFail).when(schemaRepository).getSchemaSource(sourceIdentifier, YangTextSchemaSource.class);
231 final Future<YangTextSchemaSourceSerializationProxy> failedSchema =
232 proxyYang.getYangTextSchemaSource(sourceIdentifier);
234 Await.result(failedSchema, TIMEOUT.duration());
239 public void testProxyDOMRpcService() throws Exception {
241 final DOMDataBroker domDataBroker = mock(DOMDataBroker.class);
242 final List<SourceIdentifier> sourceIdentifiers =
243 Lists.newArrayList(RevisionSourceIdentifier.create("testID", Optional.absent()));
247 final Future<Object> initialDataToActor =
248 Patterns.ask(masterRef, new CreateInitialMasterActorData(domDataBroker, sourceIdentifiers,
249 domRpcService), TIMEOUT);
251 final Object successInit = Await.result(initialDataToActor, TIMEOUT.duration());
253 assertTrue(successInit instanceof MasterActorDataInitialized);
255 // test if slave get right identifiers from master
257 final ProxyDOMRpcService slaveDomRPCService = new ProxyDOMRpcService(system, masterRef, remoteDeviceId);
259 final SchemaPath schemaPath = SchemaPath.create(true, QName.create("TestQname"));
260 final NormalizedNode<?, ?> outputNode = ImmutableContainerNodeBuilder.create()
261 .withNodeIdentifier(new YangInstanceIdentifier.NodeIdentifier(QName.create("TestQname")))
262 .withChild(ImmutableNodes.leafNode(QName.create("NodeQname"), "foo")).build();
263 final RpcError rpcError = RpcResultBuilder.newError(RpcError.ErrorType.RPC, null, "Rpc invocation failed.");
264 // EmptyResultResponse message
266 doReturn(Futures.immediateCheckedFuture(null)).when(domRpcService).invokeRpc(any(), any());
268 final CheckedFuture<DOMRpcResult, DOMRpcException> resultFutureEmpty =
269 slaveDomRPCService.invokeRpc(schemaPath, outputNode);
271 final Object resultNull = resultFutureEmpty.checkedGet(2, TimeUnit.SECONDS);
273 assertEquals(null, resultNull);
275 // InvokeRpcMessageReply message
277 doReturn(Futures.immediateCheckedFuture(new DefaultDOMRpcResult(outputNode))).
278 when(domRpcService).invokeRpc(any(), any());
280 final CheckedFuture<DOMRpcResult, DOMRpcException> resultFutureNn =
281 slaveDomRPCService.invokeRpc(schemaPath, outputNode);
283 final DOMRpcResult resultNn = resultFutureNn.checkedGet(2, TimeUnit.SECONDS);
285 assertEquals(outputNode, resultNn.getResult());
286 assertTrue(resultNn.getErrors().isEmpty());
288 // InvokeRpcMessageReply message only error
290 doReturn(Futures.immediateCheckedFuture(new DefaultDOMRpcResult(rpcError)))
291 .when(domRpcService).invokeRpc(any(), any());
293 final CheckedFuture<DOMRpcResult, DOMRpcException> resultFutureError =
294 slaveDomRPCService.invokeRpc(schemaPath, outputNode);
296 final DOMRpcResult resultError = resultFutureError.checkedGet(2, TimeUnit.SECONDS);
298 assertNull(resultError.getResult());
299 assertEquals(rpcError, resultError.getErrors().iterator().next());
301 // InvokeRpcMessageReply message error + result
303 doReturn(Futures.immediateCheckedFuture(new DefaultDOMRpcResult(outputNode, rpcError)))
304 .when(domRpcService).invokeRpc(any(), any());
306 final CheckedFuture<DOMRpcResult, DOMRpcException> resultFutureOutputError =
307 slaveDomRPCService.invokeRpc(schemaPath, outputNode);
309 final DOMRpcResult resultOutputError = resultFutureOutputError.checkedGet(2, TimeUnit.SECONDS);
311 assertEquals(outputNode, resultOutputError.getResult());
312 assertEquals(rpcError, resultOutputError.getErrors().iterator().next());
316 exception.expect(DOMRpcException.class);
318 doReturn(Futures.immediateFailedCheckedFuture(new ClusteringRpcException("")))
319 .when(domRpcService).invokeRpc(any(), any());
321 final CheckedFuture<DOMRpcResult, DOMRpcException> resultFutureThrowable =
322 slaveDomRPCService.invokeRpc(schemaPath, outputNode);
324 resultFutureThrowable.checkedGet(2, TimeUnit.SECONDS);
328 private String convertStreamToString(java.io.InputStream is) {
329 java.util.Scanner s = new java.util.Scanner(is).useDelimiter("\\A");
330 return s.hasNext() ? s.next() : "";