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 org.mockito.ArgumentMatchers.any;
11 import static org.mockito.ArgumentMatchers.eq;
12 import static org.mockito.Mockito.after;
13 import static org.mockito.Mockito.doNothing;
14 import static org.mockito.Mockito.doReturn;
15 import static org.mockito.Mockito.mock;
16 import static org.mockito.Mockito.reset;
17 import static org.mockito.Mockito.timeout;
18 import static org.mockito.Mockito.verify;
19 import static org.mockito.Mockito.verifyNoMoreInteractions;
20 import static org.opendaylight.mdsal.binding.api.DataObjectModification.ModificationType.DELETE;
21 import static org.opendaylight.mdsal.binding.api.DataObjectModification.ModificationType.SUBTREE_MODIFIED;
22 import static org.opendaylight.mdsal.binding.api.DataObjectModification.ModificationType.WRITE;
24 import akka.actor.ActorSystem;
25 import akka.actor.Props;
26 import akka.cluster.Cluster;
27 import akka.dispatch.Dispatchers;
28 import akka.testkit.TestActorRef;
29 import akka.testkit.javadsl.TestKit;
30 import akka.util.Timeout;
31 import com.google.common.collect.Iterables;
32 import com.google.common.io.CharSource;
33 import com.google.common.util.concurrent.Futures;
34 import com.typesafe.config.ConfigFactory;
35 import java.net.InetSocketAddress;
36 import java.util.List;
38 import java.util.concurrent.CompletableFuture;
39 import java.util.concurrent.ConcurrentHashMap;
40 import java.util.concurrent.ExecutionException;
41 import java.util.concurrent.TimeUnit;
42 import java.util.concurrent.TimeoutException;
43 import java.util.stream.Collectors;
44 import org.junit.After;
45 import org.junit.Before;
46 import org.junit.Test;
47 import org.junit.runner.RunWith;
48 import org.mockito.Mock;
49 import org.mockito.junit.MockitoJUnitRunner;
50 import org.opendaylight.mdsal.binding.api.DataBroker;
51 import org.opendaylight.mdsal.binding.api.DataObjectModification;
52 import org.opendaylight.mdsal.binding.api.DataTreeIdentifier;
53 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
54 import org.opendaylight.mdsal.dom.api.DOMDataBroker;
55 import org.opendaylight.mdsal.dom.api.DOMMountPoint;
56 import org.opendaylight.mdsal.dom.api.DOMMountPointService;
57 import org.opendaylight.mdsal.dom.api.DOMNotificationService;
58 import org.opendaylight.mdsal.dom.api.DOMRpcService;
59 import org.opendaylight.netconf.client.mdsal.api.DeviceNetconfSchemaProvider;
60 import org.opendaylight.netconf.client.mdsal.api.NetconfDeviceSchemasResolver;
61 import org.opendaylight.netconf.client.mdsal.api.RemoteDeviceId;
62 import org.opendaylight.netconf.client.mdsal.api.RemoteDeviceServices;
63 import org.opendaylight.netconf.client.mdsal.api.RemoteDeviceServices.Actions;
64 import org.opendaylight.netconf.client.mdsal.api.RemoteDeviceServices.Rpcs;
65 import org.opendaylight.netconf.dom.api.NetconfDataTreeService;
66 import org.opendaylight.netconf.topology.singleton.impl.actors.NetconfNodeActor;
67 import org.opendaylight.netconf.topology.singleton.impl.utils.NetconfTopologySetup;
68 import org.opendaylight.netconf.topology.singleton.impl.utils.NetconfTopologyUtils;
69 import org.opendaylight.netconf.topology.singleton.messages.AskForMasterMountPoint;
70 import org.opendaylight.netconf.topology.singleton.messages.CreateInitialMasterActorData;
71 import org.opendaylight.netconf.topology.singleton.messages.MasterActorDataInitialized;
72 import org.opendaylight.netconf.topology.singleton.messages.YangTextSchemaSourceRequest;
73 import org.opendaylight.netconf.topology.spi.NetconfNodeUtils;
74 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev130715.Host;
75 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev130715.IpAddress;
76 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev130715.Ipv4Address;
77 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev130715.PortNumber;
78 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.device.rev240120.ConnectionOper.ConnectionStatus;
79 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.device.rev240120.connection.oper.ClusteredConnectionStatusBuilder;
80 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev231121.NetconfNode;
81 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev231121.NetconfNodeBuilder;
82 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.NodeId;
83 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.Node;
84 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.NodeBuilder;
85 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.NodeKey;
86 import org.opendaylight.yangtools.concepts.ObjectRegistration;
87 import org.opendaylight.yangtools.concepts.Registration;
88 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
89 import org.opendaylight.yangtools.yang.common.Uint16;
90 import org.opendaylight.yangtools.yang.model.api.source.SourceIdentifier;
91 import org.opendaylight.yangtools.yang.model.api.source.YangTextSource;
92 import org.opendaylight.yangtools.yang.model.repo.api.EffectiveModelContextFactory;
93 import org.opendaylight.yangtools.yang.model.repo.spi.PotentialSchemaSource;
94 import org.opendaylight.yangtools.yang.model.spi.source.DelegatedYangTextSource;
95 import org.opendaylight.yangtools.yang.parser.repo.SharedSchemaRepository;
96 import org.opendaylight.yangtools.yang.parser.rfc7950.repo.TextToIRTransformer;
99 * Unit tests for NetconfNodeManager.
101 * @author Thomas Pantelis
103 @RunWith(MockitoJUnitRunner.StrictStubs.class)
104 public class NetconfNodeManagerTest extends AbstractBaseSchemasTest {
105 private static final String ACTOR_SYSTEM_NAME = "test";
106 private static final RemoteDeviceId DEVICE_ID = new RemoteDeviceId("device", new InetSocketAddress(65535));
107 private static final List<SourceIdentifier> SOURCE_IDENTIFIERS = List.of(new SourceIdentifier("testID"));
110 private DOMMountPointService mockMountPointService;
112 private DOMMountPointService.DOMMountPointBuilder mockMountPointBuilder;
114 private ObjectRegistration<DOMMountPoint> mockMountPointReg;
116 private DataBroker mockDataBroker;
118 private NetconfDataTreeService netconfService;
120 private DOMDataBroker mockDeviceDataBroker;
122 private Rpcs.Normalized mockRpcService;
124 private Actions.Normalized mockActionService;
126 private NetconfDeviceSchemasResolver mockSchemasResolver;
128 private EffectiveModelContextFactory mockSchemaContextFactory;
130 private ActorSystem slaveSystem;
131 private ActorSystem masterSystem;
132 private TestActorRef<TestMasterActor> testMasterActorRef;
133 private NetconfNodeManager netconfNodeManager;
134 private String masterAddress;
137 public void setup() {
138 final Timeout responseTimeout = Timeout.apply(1, TimeUnit.SECONDS);
140 slaveSystem = ActorSystem.create(ACTOR_SYSTEM_NAME, ConfigFactory.load().getConfig("Slave"));
141 masterSystem = ActorSystem.create(ACTOR_SYSTEM_NAME, ConfigFactory.load().getConfig("Master"));
143 masterAddress = Cluster.get(masterSystem).selfAddress().toString();
145 SharedSchemaRepository masterSchemaRepository = new SharedSchemaRepository("master");
146 masterSchemaRepository.registerSchemaSourceListener(
147 TextToIRTransformer.create(masterSchemaRepository, masterSchemaRepository));
149 final String yangTemplate = """
155 SOURCE_IDENTIFIERS.stream().map(
156 sourceId -> masterSchemaRepository.registerSchemaSource(
157 id -> Futures.immediateFuture(new DelegatedYangTextSource(id,
158 CharSource.wrap(yangTemplate.replaceAll("ID", id.name().getLocalName())))),
159 PotentialSchemaSource.create(sourceId, YangTextSource.class, 1)))
160 .collect(Collectors.toList());
162 NetconfTopologySetup masterSetup = NetconfTopologySetup.builder()
163 .setActorSystem(masterSystem)
164 .setDataBroker(mockDataBroker)
165 .setBaseSchemaProvider(BASE_SCHEMAS)
166 .setDeviceSchemaProvider(createDeviceSchemaProvider(masterSchemaRepository))
169 testMasterActorRef = TestActorRef.create(masterSystem, Props.create(TestMasterActor.class, masterSetup,
170 DEVICE_ID, responseTimeout, mockMountPointService).withDispatcher(Dispatchers.DefaultDispatcherId()),
171 NetconfTopologyUtils.createMasterActorName(DEVICE_ID.name(), masterAddress));
173 SharedSchemaRepository slaveSchemaRepository = new SharedSchemaRepository("slave");
174 slaveSchemaRepository.registerSchemaSourceListener(
175 TextToIRTransformer.create(slaveSchemaRepository, slaveSchemaRepository));
177 NetconfTopologySetup slaveSetup = NetconfTopologySetup.builder()
178 .setActorSystem(slaveSystem)
179 .setDataBroker(mockDataBroker)
180 .setBaseSchemaProvider(BASE_SCHEMAS)
181 .setDeviceSchemaProvider(createDeviceSchemaProvider(slaveSchemaRepository))
184 netconfNodeManager = new NetconfNodeManager(slaveSetup, DEVICE_ID, responseTimeout,
185 mockMountPointService);
187 setupMountPointMocks();
190 private static DeviceNetconfSchemaProvider createDeviceSchemaProvider(final SharedSchemaRepository repository) {
191 final var provider = mock(DeviceNetconfSchemaProvider.class);
192 doReturn(repository).when(provider).registry();
193 doReturn(repository).when(provider).repository();
198 public void teardown() {
199 TestKit.shutdownActorSystem(slaveSystem, true);
200 TestKit.shutdownActorSystem(masterSystem, true);
203 @SuppressWarnings("unchecked")
205 public void testSlaveMountPointRegistration() throws InterruptedException, ExecutionException, TimeoutException {
208 Registration mockListenerReg = mock(Registration.class);
209 doReturn(mockListenerReg).when(mockDataBroker).registerTreeChangeListener(any(), any());
211 final NodeId nodeId = new NodeId("device");
212 final NodeKey nodeKey = new NodeKey(nodeId);
213 final String topologyId = "topology-netconf";
214 final InstanceIdentifier<Node> nodeListPath = NetconfTopologyUtils.createTopologyNodeListPath(
215 nodeKey, topologyId);
217 netconfNodeManager.registerDataTreeChangeListener(topologyId, nodeKey);
218 verify(mockDataBroker).registerTreeChangeListener(any(), eq(netconfNodeManager));
220 // Invoke onDataTreeChanged with a NetconfNode WRITE to simulate the master writing the operational state to
221 // Connected. Expect the slave mount point created and registered.
223 final NetconfNode netconfNode = newNetconfNode();
224 final Node node = new NodeBuilder().setNodeId(nodeId).addAugmentation(netconfNode).build();
226 DataObjectModification<Node> mockDataObjModification = mock(DataObjectModification.class);
227 doReturn(Iterables.getLast(nodeListPath.getPathArguments())).when(mockDataObjModification).step();
228 doReturn(WRITE).when(mockDataObjModification).modificationType();
229 doReturn(node).when(mockDataObjModification).dataAfter();
231 netconfNodeManager.onDataTreeChanged(List.of(
232 new NetconfTopologyManagerTest.CustomTreeModification(DataTreeIdentifier.of(
233 LogicalDatastoreType.OPERATIONAL, nodeListPath), mockDataObjModification)));
235 verify(mockMountPointBuilder, timeout(5000)).register();
236 verify(mockMountPointBuilder).addService(eq(DOMDataBroker.class), any());
237 verify(mockMountPointBuilder).addService(eq(DOMRpcService.class), any());
238 verify(mockMountPointBuilder).addService(eq(DOMNotificationService.class), any());
239 verify(mockMountPointService).createMountPoint(NetconfNodeUtils.defaultTopologyMountPath(DEVICE_ID));
241 // Notify that the NetconfNode operational state was deleted. Expect the slave mount point closed.
243 doReturn(DELETE).when(mockDataObjModification).modificationType();
245 netconfNodeManager.onDataTreeChanged(List.of(
246 new NetconfTopologyManagerTest.CustomTreeModification(DataTreeIdentifier.of(
247 LogicalDatastoreType.OPERATIONAL, nodeListPath), mockDataObjModification)));
249 verify(mockMountPointReg, timeout(5000)).close();
251 // Notify with a NetconfNode operational state WRITE. Expect the slave mount point re-created.
253 setupMountPointMocks();
255 doReturn(WRITE).when(mockDataObjModification).modificationType();
256 doReturn(null).when(mockDataObjModification).dataBefore();
257 doReturn(node).when(mockDataObjModification).dataAfter();
259 netconfNodeManager.onDataTreeChanged(List.of(
260 new NetconfTopologyManagerTest.CustomTreeModification(DataTreeIdentifier.of(
261 LogicalDatastoreType.OPERATIONAL, nodeListPath), mockDataObjModification)));
263 verify(mockMountPointBuilder, timeout(5000)).register();
265 // Notify again with a NetconfNode operational state WRITE. Expect the prior slave mount point closed and
266 // and a new one registered.
268 setupMountPointMocks();
270 doReturn(node).when(mockDataObjModification).dataBefore();
272 netconfNodeManager.onDataTreeChanged(List.of(
273 new NetconfTopologyManagerTest.CustomTreeModification(DataTreeIdentifier.of(
274 LogicalDatastoreType.OPERATIONAL, nodeListPath), mockDataObjModification)));
276 verify(mockMountPointReg, timeout(5000)).close();
277 verify(mockMountPointBuilder, timeout(5000)).register();
279 // Notify that the NetconfNode operational state was changed to UnableToConnect. Expect the slave mount point
282 reset(mockMountPointService, mockMountPointBuilder, mockMountPointReg);
283 doNothing().when(mockMountPointReg).close();
285 final Node updatedNode = new NodeBuilder().setNodeId(nodeId)
286 .addAugmentation(new NetconfNodeBuilder(netconfNode)
287 .setConnectionStatus(ConnectionStatus.UnableToConnect)
291 doReturn(SUBTREE_MODIFIED).when(mockDataObjModification).modificationType();
292 doReturn(node).when(mockDataObjModification).dataBefore();
293 doReturn(updatedNode).when(mockDataObjModification).dataAfter();
295 netconfNodeManager.onDataTreeChanged(List.of(
296 new NetconfTopologyManagerTest.CustomTreeModification(DataTreeIdentifier.of(
297 LogicalDatastoreType.OPERATIONAL, nodeListPath), mockDataObjModification)));
299 verify(mockMountPointReg, timeout(5000)).close();
301 netconfNodeManager.close();
302 verifyNoMoreInteractions(mockMountPointReg);
305 @SuppressWarnings("unchecked")
307 public void testSlaveMountPointRegistrationFailuresAndRetries()
308 throws InterruptedException, ExecutionException, TimeoutException {
309 final NodeId nodeId = new NodeId("device");
310 final NodeKey nodeKey = new NodeKey(nodeId);
311 final String topologyId = "topology-netconf";
312 final InstanceIdentifier<Node> nodeListPath = NetconfTopologyUtils.createTopologyNodeListPath(
313 nodeKey, topologyId);
315 final NetconfNode netconfNode = newNetconfNode();
316 final Node node = new NodeBuilder().setNodeId(nodeId).addAugmentation(netconfNode).build();
318 DataObjectModification<Node> mockDataObjModification = mock(DataObjectModification.class);
319 doReturn(Iterables.getLast(nodeListPath.getPathArguments())).when(mockDataObjModification).step();
320 doReturn(WRITE).when(mockDataObjModification).modificationType();
321 doReturn(node).when(mockDataObjModification).dataAfter();
323 // First try the registration where the perceived master hasn't been initialized as the master.
325 netconfNodeManager.onDataTreeChanged(List.of(
326 new NetconfTopologyManagerTest.CustomTreeModification(DataTreeIdentifier.of(
327 LogicalDatastoreType.OPERATIONAL, nodeListPath), mockDataObjModification)));
329 verify(mockMountPointBuilder, after(1000).never()).register();
331 // Initialize the master but drop the initial YangTextSchemaSourceRequest message sent to the master so
336 CompletableFuture<AskForMasterMountPoint> yangTextSchemaSourceRequestFuture = new CompletableFuture<>();
337 testMasterActorRef.underlyingActor().messagesToDrop.put(YangTextSchemaSourceRequest.class,
338 yangTextSchemaSourceRequestFuture);
340 netconfNodeManager.onDataTreeChanged(List.of(
341 new NetconfTopologyManagerTest.CustomTreeModification(DataTreeIdentifier.of(
342 LogicalDatastoreType.OPERATIONAL, nodeListPath), mockDataObjModification)));
344 yangTextSchemaSourceRequestFuture.get(5, TimeUnit.SECONDS);
345 verify(mockMountPointBuilder, timeout(5000)).register();
347 // Initiate another registration but drop the initial AskForMasterMountPoint message sent to the master so
350 setupMountPointMocks();
352 CompletableFuture<AskForMasterMountPoint> askForMasterMountPointFuture = new CompletableFuture<>();
353 testMasterActorRef.underlyingActor().messagesToDrop.put(AskForMasterMountPoint.class,
354 askForMasterMountPointFuture);
356 netconfNodeManager.onDataTreeChanged(List.of(
357 new NetconfTopologyManagerTest.CustomTreeModification(DataTreeIdentifier.of(
358 LogicalDatastoreType.OPERATIONAL, nodeListPath), mockDataObjModification)));
360 askForMasterMountPointFuture.get(5, TimeUnit.SECONDS);
361 verify(mockMountPointReg, timeout(5000)).close();
362 verify(mockMountPointBuilder, timeout(5000)).register();
364 reset(mockMountPointService, mockMountPointBuilder, mockMountPointReg);
365 doNothing().when(mockMountPointReg).close();
366 netconfNodeManager.close();
367 verify(mockMountPointReg, timeout(5000)).close();
370 private NetconfNode newNetconfNode() {
371 return new NetconfNodeBuilder()
372 .setHost(new Host(new IpAddress(new Ipv4Address("127.0.0.1"))))
373 .setPort(new PortNumber(Uint16.valueOf(9999)))
374 .setConnectionStatus(ConnectionStatus.Connected)
375 .setClusteredConnectionStatus(new ClusteredConnectionStatusBuilder()
376 .setNetconfMasterNode(masterAddress).build())
380 private void setupMountPointMocks() {
381 reset(mockMountPointService, mockMountPointBuilder, mockMountPointReg);
382 doNothing().when(mockMountPointReg).close();
383 doReturn(mockMountPointReg).when(mockMountPointBuilder).register();
384 doReturn(mockMountPointBuilder).when(mockMountPointService).createMountPoint(any());
387 private void initializeMaster() {
388 TestKit kit = new TestKit(masterSystem);
389 testMasterActorRef.tell(new CreateInitialMasterActorData(mockDeviceDataBroker, netconfService,
390 SOURCE_IDENTIFIERS, new RemoteDeviceServices(mockRpcService, mockActionService)), kit.getRef());
392 kit.expectMsgClass(MasterActorDataInitialized.class);
395 private static class TestMasterActor extends NetconfNodeActor {
396 final Map<Class<?>, CompletableFuture<? extends Object>> messagesToDrop = new ConcurrentHashMap<>();
398 TestMasterActor(final NetconfTopologySetup setup, final RemoteDeviceId deviceId,
399 final Timeout actorResponseWaitTime, final DOMMountPointService mountPointService) {
400 super(setup, deviceId, actorResponseWaitTime, mountPointService);
403 @SuppressWarnings({ "rawtypes", "unchecked" })
405 public void handleReceive(final Object message) {
406 CompletableFuture dropFuture = messagesToDrop.remove(message.getClass());
407 if (dropFuture != null) {
408 dropFuture.complete(message);
410 super.handleReceive(message);