2 * Copyright (c) 2018 Inocybe Technologies 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
8 package org.opendaylight.netconf.topology.singleton.impl;
10 import static java.nio.charset.StandardCharsets.UTF_8;
11 import static org.mockito.Matchers.any;
12 import static org.mockito.Matchers.eq;
13 import static org.mockito.Mockito.after;
14 import static org.mockito.Mockito.doNothing;
15 import static org.mockito.Mockito.doReturn;
16 import static org.mockito.Mockito.mock;
17 import static org.mockito.Mockito.reset;
18 import static org.mockito.Mockito.timeout;
19 import static org.mockito.Mockito.verify;
20 import static org.mockito.Mockito.verifyNoMoreInteractions;
21 import static org.mockito.MockitoAnnotations.initMocks;
22 import static org.opendaylight.controller.md.sal.binding.api.DataObjectModification.ModificationType.DELETE;
23 import static org.opendaylight.controller.md.sal.binding.api.DataObjectModification.ModificationType.SUBTREE_MODIFIED;
24 import static org.opendaylight.controller.md.sal.binding.api.DataObjectModification.ModificationType.WRITE;
26 import akka.actor.ActorSystem;
27 import akka.actor.Props;
28 import akka.cluster.Cluster;
29 import akka.dispatch.Dispatchers;
30 import akka.testkit.TestActorRef;
31 import akka.testkit.javadsl.TestKit;
32 import akka.util.Timeout;
33 import com.google.common.collect.ImmutableList;
34 import com.google.common.collect.Iterables;
35 import com.google.common.io.ByteSource;
36 import com.google.common.util.concurrent.Futures;
37 import com.typesafe.config.ConfigFactory;
38 import java.net.InetSocketAddress;
39 import java.util.Collections;
40 import java.util.List;
42 import java.util.concurrent.CompletableFuture;
43 import java.util.concurrent.ConcurrentHashMap;
44 import java.util.concurrent.ExecutionException;
45 import java.util.concurrent.TimeUnit;
46 import java.util.concurrent.TimeoutException;
47 import java.util.stream.Collectors;
48 import org.junit.After;
49 import org.junit.Before;
50 import org.junit.Test;
51 import org.mockito.Mock;
52 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
53 import org.opendaylight.controller.md.sal.binding.api.DataObjectModification;
54 import org.opendaylight.controller.md.sal.binding.api.DataTreeIdentifier;
55 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
56 import org.opendaylight.controller.md.sal.dom.api.DOMDataBroker;
57 import org.opendaylight.controller.md.sal.dom.api.DOMMountPoint;
58 import org.opendaylight.controller.md.sal.dom.api.DOMMountPointService;
59 import org.opendaylight.controller.md.sal.dom.api.DOMNotificationService;
60 import org.opendaylight.controller.md.sal.dom.api.DOMRpcService;
61 import org.opendaylight.netconf.sal.connect.api.NetconfDeviceSchemasResolver;
62 import org.opendaylight.netconf.sal.connect.netconf.NetconfDevice;
63 import org.opendaylight.netconf.sal.connect.util.RemoteDeviceId;
64 import org.opendaylight.netconf.topology.singleton.impl.actors.NetconfNodeActor;
65 import org.opendaylight.netconf.topology.singleton.impl.utils.NetconfTopologySetup;
66 import org.opendaylight.netconf.topology.singleton.impl.utils.NetconfTopologyUtils;
67 import org.opendaylight.netconf.topology.singleton.messages.AskForMasterMountPoint;
68 import org.opendaylight.netconf.topology.singleton.messages.CreateInitialMasterActorData;
69 import org.opendaylight.netconf.topology.singleton.messages.MasterActorDataInitialized;
70 import org.opendaylight.netconf.topology.singleton.messages.YangTextSchemaSourceRequest;
71 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev130715.Host;
72 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev130715.IpAddress;
73 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev130715.Ipv4Address;
74 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev130715.PortNumber;
75 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.NetconfNode;
76 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.NetconfNodeBuilder;
77 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.NetconfNodeConnectionStatus;
78 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.netconf.node.connection.status.ClusteredConnectionStatusBuilder;
79 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.NodeId;
80 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.Node;
81 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.NodeBuilder;
82 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.NodeKey;
83 import org.opendaylight.yangtools.concepts.ListenerRegistration;
84 import org.opendaylight.yangtools.concepts.ObjectRegistration;
85 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
86 import org.opendaylight.yangtools.yang.model.repo.api.RevisionSourceIdentifier;
87 import org.opendaylight.yangtools.yang.model.repo.api.SchemaContextFactory;
88 import org.opendaylight.yangtools.yang.model.repo.api.SourceIdentifier;
89 import org.opendaylight.yangtools.yang.model.repo.api.YangTextSchemaSource;
90 import org.opendaylight.yangtools.yang.model.repo.spi.PotentialSchemaSource;
91 import org.opendaylight.yangtools.yang.parser.repo.SharedSchemaRepository;
92 import org.opendaylight.yangtools.yang.parser.rfc7950.repo.TextToASTTransformer;
95 * Unit tests for NetconfNodeManager.
97 * @author Thomas Pantelis
99 public class NetconfNodeManagerTest {
100 private static final String ACTOR_SYSTEM_NAME = "test";
101 private static final RemoteDeviceId DEVICE_ID = new RemoteDeviceId("device", new InetSocketAddress(65535));
102 private static final List<SourceIdentifier> SOURCE_IDENTIFIERS =
103 ImmutableList.of(RevisionSourceIdentifier.create("testID"));
106 private DOMMountPointService mockMountPointService;
109 private DOMMountPointService.DOMMountPointBuilder mockMountPointBuilder;
112 private ObjectRegistration<DOMMountPoint> mockMountPointReg;
115 private DataBroker mockDataBroker;
118 private DOMDataBroker mockDeviceDataBroker;
121 private DOMRpcService mockRpcService;
124 private NetconfDeviceSchemasResolver mockSchemasResolver;
127 private SchemaContextFactory mockSchemaContextFactory;
129 private ActorSystem slaveSystem;
130 private ActorSystem masterSystem;
131 private TestActorRef<TestMasterActor> testMasterActorRef;
132 private NetconfNodeManager netconfNodeManager;
133 private String masterAddress;
136 public void setup() {
139 final Timeout responseTimeout = Timeout.apply(1, TimeUnit.SECONDS);
141 slaveSystem = ActorSystem.create(ACTOR_SYSTEM_NAME, ConfigFactory.load().getConfig("Slave"));
142 masterSystem = ActorSystem.create(ACTOR_SYSTEM_NAME, ConfigFactory.load().getConfig("Master"));
144 masterAddress = Cluster.get(masterSystem).selfAddress().toString();
146 SharedSchemaRepository masterSchemaRepository = new SharedSchemaRepository("master");
147 masterSchemaRepository.registerSchemaSourceListener(
148 TextToASTTransformer.create(masterSchemaRepository, masterSchemaRepository));
150 String yangTemplate =
152 + " namespace \"ID\";"
156 SOURCE_IDENTIFIERS.stream().map(
157 sourceId -> masterSchemaRepository.registerSchemaSource(
158 id -> Futures.immediateFuture(YangTextSchemaSource.delegateForByteSource(id,
159 ByteSource.wrap(yangTemplate.replaceAll("ID", id.getName()).getBytes(UTF_8)))),
160 PotentialSchemaSource.create(sourceId, YangTextSchemaSource.class, 1)))
161 .collect(Collectors.toList());
163 NetconfTopologySetup masterSetup = new NetconfTopologySetup.NetconfTopologySetupBuilder()
164 .setActorSystem(masterSystem).setDataBroker(mockDataBroker).setSchemaResourceDTO(
165 new NetconfDevice.SchemaResourcesDTO(masterSchemaRepository, masterSchemaRepository,
166 mockSchemaContextFactory, mockSchemasResolver)).build();
168 testMasterActorRef = TestActorRef.create(masterSystem, Props.create(TestMasterActor.class, masterSetup,
169 DEVICE_ID, responseTimeout, mockMountPointService).withDispatcher(Dispatchers.DefaultDispatcherId()),
170 NetconfTopologyUtils.createMasterActorName(DEVICE_ID.getName(), masterAddress));
172 SharedSchemaRepository slaveSchemaRepository = new SharedSchemaRepository("slave");
173 slaveSchemaRepository.registerSchemaSourceListener(
174 TextToASTTransformer.create(slaveSchemaRepository, slaveSchemaRepository));
176 NetconfTopologySetup slaveSetup = new NetconfTopologySetup.NetconfTopologySetupBuilder()
177 .setActorSystem(slaveSystem).setDataBroker(mockDataBroker).setSchemaResourceDTO(
178 new NetconfDevice.SchemaResourcesDTO(slaveSchemaRepository, slaveSchemaRepository,
179 mockSchemaContextFactory, mockSchemasResolver)).build();
181 netconfNodeManager = new NetconfNodeManager(slaveSetup, DEVICE_ID, responseTimeout,
182 mockMountPointService);
184 setupMountPointMocks();
188 public void teardown() {
189 TestKit.shutdownActorSystem(slaveSystem, true);
190 TestKit.shutdownActorSystem(masterSystem, true);
193 @SuppressWarnings("unchecked")
195 public void testSlaveMountPointRegistration() throws InterruptedException, ExecutionException, TimeoutException {
198 ListenerRegistration<?> mockListenerReg = mock(ListenerRegistration.class);
199 doReturn(mockListenerReg).when(mockDataBroker).registerDataTreeChangeListener(any(), any());
201 final NodeId nodeId = new NodeId("device");
202 final NodeKey nodeKey = new NodeKey(nodeId);
203 final String topologyId = "topology-netconf";
204 final InstanceIdentifier<Node> nodeListPath = NetconfTopologyUtils.createTopologyNodeListPath(
205 nodeKey, topologyId);
207 netconfNodeManager.registerDataTreeChangeListener(topologyId, nodeKey);
208 verify(mockDataBroker).registerDataTreeChangeListener(any(), eq(netconfNodeManager));
210 // Invoke onDataTreeChanged with a NetconfNode WRITE to simulate the master writing the operational state to
211 // Connected. Expect the slave mount point created and registered.
213 final NetconfNode netconfNode = newNetconfNode();
214 final Node node = new NodeBuilder().setNodeId(nodeId).addAugmentation(NetconfNode.class, netconfNode).build();
216 DataObjectModification<Node> mockDataObjModification = mock(DataObjectModification.class);
217 doReturn(Iterables.getLast(nodeListPath.getPathArguments())).when(mockDataObjModification).getIdentifier();
218 doReturn(WRITE).when(mockDataObjModification).getModificationType();
219 doReturn(node).when(mockDataObjModification).getDataAfter();
221 netconfNodeManager.onDataTreeChanged(Collections.singletonList(
222 new NetconfTopologyManagerTest.CustomTreeModification(new DataTreeIdentifier<>(
223 LogicalDatastoreType.OPERATIONAL, nodeListPath), mockDataObjModification)));
225 verify(mockMountPointBuilder, timeout(5000)).register();
226 verify(mockMountPointBuilder).addService(eq(DOMDataBroker.class), any());
227 verify(mockMountPointBuilder).addService(eq(DOMRpcService.class), any());
228 verify(mockMountPointBuilder).addService(eq(DOMNotificationService.class), any());
229 verify(mockMountPointService).createMountPoint(DEVICE_ID.getTopologyPath());
231 // Notify that the NetconfNode operational state was deleted. Expect the slave mount point closed.
233 doReturn(DELETE).when(mockDataObjModification).getModificationType();
234 doReturn(node).when(mockDataObjModification).getDataBefore();
235 doReturn(null).when(mockDataObjModification).getDataAfter();
237 netconfNodeManager.onDataTreeChanged(Collections.singletonList(
238 new NetconfTopologyManagerTest.CustomTreeModification(new DataTreeIdentifier<>(
239 LogicalDatastoreType.OPERATIONAL, nodeListPath), mockDataObjModification)));
241 verify(mockMountPointReg, timeout(5000)).close();
243 // Notify with a NetconfNode operational state WRITE. Expect the slave mount point re-created.
245 setupMountPointMocks();
247 doReturn(WRITE).when(mockDataObjModification).getModificationType();
248 doReturn(null).when(mockDataObjModification).getDataBefore();
249 doReturn(node).when(mockDataObjModification).getDataAfter();
251 netconfNodeManager.onDataTreeChanged(Collections.singletonList(
252 new NetconfTopologyManagerTest.CustomTreeModification(new DataTreeIdentifier<>(
253 LogicalDatastoreType.OPERATIONAL, nodeListPath), mockDataObjModification)));
255 verify(mockMountPointBuilder, timeout(5000)).register();
257 // Notify again with a NetconfNode operational state WRITE. Expect the prior slave mount point closed and
258 // and a new one registered.
260 setupMountPointMocks();
262 doReturn(node).when(mockDataObjModification).getDataBefore();
264 netconfNodeManager.onDataTreeChanged(Collections.singletonList(
265 new NetconfTopologyManagerTest.CustomTreeModification(new DataTreeIdentifier<>(
266 LogicalDatastoreType.OPERATIONAL, nodeListPath), mockDataObjModification)));
268 verify(mockMountPointReg, timeout(5000)).close();
269 verify(mockMountPointBuilder, timeout(5000)).register();
271 // Notify that the NetconfNode operational state was changed to UnableToConnect. Expect the slave mount point
274 setupMountPointMocks();
276 final Node updatedNode = new NodeBuilder().setNodeId(nodeId)
277 .addAugmentation(NetconfNode.class, new NetconfNodeBuilder(netconfNode)
278 .setConnectionStatus(NetconfNodeConnectionStatus.ConnectionStatus.UnableToConnect)
281 doReturn(SUBTREE_MODIFIED).when(mockDataObjModification).getModificationType();
282 doReturn(node).when(mockDataObjModification).getDataBefore();
283 doReturn(updatedNode).when(mockDataObjModification).getDataAfter();
285 netconfNodeManager.onDataTreeChanged(Collections.singletonList(
286 new NetconfTopologyManagerTest.CustomTreeModification(new DataTreeIdentifier<>(
287 LogicalDatastoreType.OPERATIONAL, nodeListPath), mockDataObjModification)));
289 verify(mockMountPointReg, timeout(5000)).close();
291 netconfNodeManager.close();
292 verifyNoMoreInteractions(mockMountPointReg);
295 @SuppressWarnings("unchecked")
297 public void testSlaveMountPointRegistrationFailuresAndRetries()
298 throws InterruptedException, ExecutionException, TimeoutException {
299 final NodeId nodeId = new NodeId("device");
300 final NodeKey nodeKey = new NodeKey(nodeId);
301 final String topologyId = "topology-netconf";
302 final InstanceIdentifier<Node> nodeListPath = NetconfTopologyUtils.createTopologyNodeListPath(
303 nodeKey, topologyId);
305 final NetconfNode netconfNode = newNetconfNode();
306 final Node node = new NodeBuilder().setNodeId(nodeId).addAugmentation(NetconfNode.class, netconfNode).build();
308 DataObjectModification<Node> mockDataObjModification = mock(DataObjectModification.class);
309 doReturn(Iterables.getLast(nodeListPath.getPathArguments())).when(mockDataObjModification).getIdentifier();
310 doReturn(WRITE).when(mockDataObjModification).getModificationType();
311 doReturn(node).when(mockDataObjModification).getDataAfter();
313 // First try the registration where the perceived master hasn't been initialized as the master.
315 netconfNodeManager.onDataTreeChanged(Collections.singletonList(
316 new NetconfTopologyManagerTest.CustomTreeModification(new DataTreeIdentifier<>(
317 LogicalDatastoreType.OPERATIONAL, nodeListPath), mockDataObjModification)));
319 verify(mockMountPointBuilder, after(1000).never()).register();
321 // Initialize the master but drop the initial YangTextSchemaSourceRequest message sent to the master so
326 CompletableFuture<AskForMasterMountPoint> yangTextSchemaSourceRequestFuture = new CompletableFuture<>();
327 testMasterActorRef.underlyingActor().messagesToDrop.put(YangTextSchemaSourceRequest.class,
328 yangTextSchemaSourceRequestFuture);
330 netconfNodeManager.onDataTreeChanged(Collections.singletonList(
331 new NetconfTopologyManagerTest.CustomTreeModification(new DataTreeIdentifier<>(
332 LogicalDatastoreType.OPERATIONAL, nodeListPath), mockDataObjModification)));
334 yangTextSchemaSourceRequestFuture.get(5, TimeUnit.SECONDS);
335 verify(mockMountPointBuilder, timeout(5000)).register();
337 // Initiate another registration but drop the initial AskForMasterMountPoint message sent to the master so
340 setupMountPointMocks();
342 CompletableFuture<AskForMasterMountPoint> askForMasterMountPointFuture = new CompletableFuture<>();
343 testMasterActorRef.underlyingActor().messagesToDrop.put(AskForMasterMountPoint.class,
344 askForMasterMountPointFuture);
346 netconfNodeManager.onDataTreeChanged(Collections.singletonList(
347 new NetconfTopologyManagerTest.CustomTreeModification(new DataTreeIdentifier<>(
348 LogicalDatastoreType.OPERATIONAL, nodeListPath), mockDataObjModification)));
350 askForMasterMountPointFuture.get(5, TimeUnit.SECONDS);
351 verify(mockMountPointReg, timeout(5000)).close();
352 verify(mockMountPointBuilder, timeout(5000)).register();
354 setupMountPointMocks();
355 netconfNodeManager.close();
356 verify(mockMountPointReg, timeout(5000)).close();
359 private NetconfNode newNetconfNode() {
360 return new NetconfNodeBuilder()
361 .setHost(new Host(new IpAddress(new Ipv4Address("127.0.0.1"))))
362 .setPort(new PortNumber(9999))
363 .setConnectionStatus(NetconfNodeConnectionStatus.ConnectionStatus.Connected)
364 .setClusteredConnectionStatus(new ClusteredConnectionStatusBuilder()
365 .setNetconfMasterNode(masterAddress).build())
369 private void setupMountPointMocks() {
370 reset(mockMountPointService, mockMountPointBuilder, mockMountPointReg);
372 doNothing().when(mockMountPointReg).close();
374 doReturn(mockMountPointBuilder).when(mockMountPointBuilder).addInitialSchemaContext(any());
375 doReturn(mockMountPointBuilder).when(mockMountPointBuilder).addService(any(), any());
376 doReturn(mockMountPointReg).when(mockMountPointBuilder).register();
378 doReturn(mockMountPointBuilder).when(mockMountPointService).createMountPoint(any());
381 private void initializeMaster() {
382 TestKit kit = new TestKit(masterSystem);
384 testMasterActorRef.tell(new CreateInitialMasterActorData(mockDeviceDataBroker, SOURCE_IDENTIFIERS,
385 mockRpcService), kit.getRef());
387 kit.expectMsgClass(MasterActorDataInitialized.class);
390 private static class TestMasterActor extends NetconfNodeActor {
391 final Map<Class<?>, CompletableFuture<? extends Object>> messagesToDrop = new ConcurrentHashMap<>();
393 TestMasterActor(NetconfTopologySetup setup, RemoteDeviceId deviceId, Timeout actorResponseWaitTime,
394 DOMMountPointService mountPointService) {
395 super(setup, deviceId, setup.getSchemaResourcesDTO().getSchemaRegistry(),
396 setup.getSchemaResourcesDTO().getSchemaRepository(), actorResponseWaitTime, mountPointService);
399 @SuppressWarnings({ "rawtypes", "unchecked" })
401 public void handleReceive(Object message) throws Exception {
402 CompletableFuture dropFuture = messagesToDrop.remove(message.getClass());
403 if (dropFuture != null) {
404 dropFuture.complete(message);
406 super.handleReceive(message);