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.ArgumentMatchers.any;
12 import static org.mockito.ArgumentMatchers.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.opendaylight.mdsal.binding.api.DataObjectModification.ModificationType.DELETE;
22 import static org.opendaylight.mdsal.binding.api.DataObjectModification.ModificationType.SUBTREE_MODIFIED;
23 import static org.opendaylight.mdsal.binding.api.DataObjectModification.ModificationType.WRITE;
25 import akka.actor.ActorSystem;
26 import akka.actor.Props;
27 import akka.cluster.Cluster;
28 import akka.dispatch.Dispatchers;
29 import akka.testkit.TestActorRef;
30 import akka.testkit.javadsl.TestKit;
31 import akka.util.Timeout;
32 import com.google.common.collect.Iterables;
33 import com.google.common.io.ByteSource;
34 import com.google.common.util.concurrent.Futures;
35 import com.typesafe.config.ConfigFactory;
36 import java.net.InetSocketAddress;
37 import java.util.List;
39 import java.util.concurrent.CompletableFuture;
40 import java.util.concurrent.ConcurrentHashMap;
41 import java.util.concurrent.ExecutionException;
42 import java.util.concurrent.TimeUnit;
43 import java.util.concurrent.TimeoutException;
44 import java.util.stream.Collectors;
45 import org.junit.After;
46 import org.junit.Before;
47 import org.junit.Test;
48 import org.junit.runner.RunWith;
49 import org.mockito.Mock;
50 import org.mockito.junit.MockitoJUnitRunner;
51 import org.opendaylight.mdsal.binding.api.DataBroker;
52 import org.opendaylight.mdsal.binding.api.DataObjectModification;
53 import org.opendaylight.mdsal.binding.api.DataTreeIdentifier;
54 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
55 import org.opendaylight.mdsal.dom.api.DOMDataBroker;
56 import org.opendaylight.mdsal.dom.api.DOMMountPoint;
57 import org.opendaylight.mdsal.dom.api.DOMMountPointService;
58 import org.opendaylight.mdsal.dom.api.DOMNotificationService;
59 import org.opendaylight.mdsal.dom.api.DOMRpcService;
60 import org.opendaylight.netconf.client.mdsal.NetconfDevice;
61 import org.opendaylight.netconf.client.mdsal.api.CredentialProvider;
62 import org.opendaylight.netconf.client.mdsal.api.KeyStoreProvider;
63 import org.opendaylight.netconf.client.mdsal.api.NetconfDeviceSchemasResolver;
64 import org.opendaylight.netconf.client.mdsal.api.RemoteDeviceId;
65 import org.opendaylight.netconf.client.mdsal.api.RemoteDeviceServices;
66 import org.opendaylight.netconf.client.mdsal.api.RemoteDeviceServices.Actions;
67 import org.opendaylight.netconf.client.mdsal.api.RemoteDeviceServices.Rpcs;
68 import org.opendaylight.netconf.dom.api.NetconfDataTreeService;
69 import org.opendaylight.netconf.topology.singleton.impl.actors.NetconfNodeActor;
70 import org.opendaylight.netconf.topology.singleton.impl.utils.NetconfTopologySetup;
71 import org.opendaylight.netconf.topology.singleton.impl.utils.NetconfTopologyUtils;
72 import org.opendaylight.netconf.topology.singleton.messages.AskForMasterMountPoint;
73 import org.opendaylight.netconf.topology.singleton.messages.CreateInitialMasterActorData;
74 import org.opendaylight.netconf.topology.singleton.messages.MasterActorDataInitialized;
75 import org.opendaylight.netconf.topology.singleton.messages.YangTextSchemaSourceRequest;
76 import org.opendaylight.netconf.topology.spi.NetconfNodeUtils;
77 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev130715.Host;
78 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev130715.IpAddress;
79 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev130715.Ipv4Address;
80 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev130715.PortNumber;
81 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.device.rev230430.ConnectionOper.ConnectionStatus;
82 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.device.rev230430.connection.oper.ClusteredConnectionStatusBuilder;
83 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev221225.NetconfNode;
84 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev221225.NetconfNodeBuilder;
85 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.NodeId;
86 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.Node;
87 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.NodeBuilder;
88 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.NodeKey;
89 import org.opendaylight.yangtools.concepts.ListenerRegistration;
90 import org.opendaylight.yangtools.concepts.ObjectRegistration;
91 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
92 import org.opendaylight.yangtools.yang.common.Uint16;
93 import org.opendaylight.yangtools.yang.model.repo.api.EffectiveModelContextFactory;
94 import org.opendaylight.yangtools.yang.model.repo.api.SourceIdentifier;
95 import org.opendaylight.yangtools.yang.model.repo.api.YangTextSchemaSource;
96 import org.opendaylight.yangtools.yang.model.repo.spi.PotentialSchemaSource;
97 import org.opendaylight.yangtools.yang.parser.repo.SharedSchemaRepository;
98 import org.opendaylight.yangtools.yang.parser.rfc7950.repo.TextToIRTransformer;
101 * Unit tests for NetconfNodeManager.
103 * @author Thomas Pantelis
105 @RunWith(MockitoJUnitRunner.StrictStubs.class)
106 public class NetconfNodeManagerTest extends AbstractBaseSchemasTest {
107 private static final String ACTOR_SYSTEM_NAME = "test";
108 private static final RemoteDeviceId DEVICE_ID = new RemoteDeviceId("device", new InetSocketAddress(65535));
109 private static final List<SourceIdentifier> SOURCE_IDENTIFIERS = List.of(new SourceIdentifier("testID"));
112 private DOMMountPointService mockMountPointService;
114 private DOMMountPointService.DOMMountPointBuilder mockMountPointBuilder;
116 private ObjectRegistration<DOMMountPoint> mockMountPointReg;
118 private DataBroker mockDataBroker;
120 private NetconfDataTreeService netconfService;
122 private DOMDataBroker mockDeviceDataBroker;
124 private Rpcs.Normalized mockRpcService;
126 private Actions.Normalized mockActionService;
128 private NetconfDeviceSchemasResolver mockSchemasResolver;
130 private EffectiveModelContextFactory mockSchemaContextFactory;
132 private CredentialProvider credentialProvider;
134 private KeyStoreProvider keyStoreProvider;
136 private ActorSystem slaveSystem;
137 private ActorSystem masterSystem;
138 private TestActorRef<TestMasterActor> testMasterActorRef;
139 private NetconfNodeManager netconfNodeManager;
140 private String masterAddress;
143 public void setup() {
144 final Timeout responseTimeout = Timeout.apply(1, TimeUnit.SECONDS);
146 slaveSystem = ActorSystem.create(ACTOR_SYSTEM_NAME, ConfigFactory.load().getConfig("Slave"));
147 masterSystem = ActorSystem.create(ACTOR_SYSTEM_NAME, ConfigFactory.load().getConfig("Master"));
149 masterAddress = Cluster.get(masterSystem).selfAddress().toString();
151 SharedSchemaRepository masterSchemaRepository = new SharedSchemaRepository("master");
152 masterSchemaRepository.registerSchemaSourceListener(
153 TextToIRTransformer.create(masterSchemaRepository, masterSchemaRepository));
155 final String yangTemplate = """
161 SOURCE_IDENTIFIERS.stream().map(
162 sourceId -> masterSchemaRepository.registerSchemaSource(
163 id -> Futures.immediateFuture(YangTextSchemaSource.delegateForByteSource(id,
164 ByteSource.wrap(yangTemplate.replaceAll("ID", id.name().getLocalName()).getBytes(UTF_8)))),
165 PotentialSchemaSource.create(sourceId, YangTextSchemaSource.class, 1)))
166 .collect(Collectors.toList());
168 NetconfTopologySetup masterSetup = new NetconfTopologySetup.NetconfTopologySetupBuilder()
169 .setActorSystem(masterSystem)
170 .setDataBroker(mockDataBroker)
171 .setSchemaResourceDTO(new NetconfDevice.SchemaResourcesDTO(
172 masterSchemaRepository, masterSchemaRepository, mockSchemaContextFactory, mockSchemasResolver))
173 .setBaseSchemas(BASE_SCHEMAS)
174 .setCredentialProvider(credentialProvider)
175 .setKeyStoreProvider(keyStoreProvider)
178 testMasterActorRef = TestActorRef.create(masterSystem, Props.create(TestMasterActor.class, masterSetup,
179 DEVICE_ID, responseTimeout, mockMountPointService).withDispatcher(Dispatchers.DefaultDispatcherId()),
180 NetconfTopologyUtils.createMasterActorName(DEVICE_ID.name(), masterAddress));
182 SharedSchemaRepository slaveSchemaRepository = new SharedSchemaRepository("slave");
183 slaveSchemaRepository.registerSchemaSourceListener(
184 TextToIRTransformer.create(slaveSchemaRepository, slaveSchemaRepository));
186 NetconfTopologySetup slaveSetup = new NetconfTopologySetup.NetconfTopologySetupBuilder()
187 .setActorSystem(slaveSystem)
188 .setDataBroker(mockDataBroker)
189 .setSchemaResourceDTO(new NetconfDevice.SchemaResourcesDTO(
190 slaveSchemaRepository, slaveSchemaRepository, mockSchemaContextFactory, mockSchemasResolver))
191 .setBaseSchemas(BASE_SCHEMAS)
192 .setCredentialProvider(credentialProvider)
193 .setKeyStoreProvider(keyStoreProvider)
196 netconfNodeManager = new NetconfNodeManager(slaveSetup, DEVICE_ID, responseTimeout,
197 mockMountPointService);
199 setupMountPointMocks();
203 public void teardown() {
204 TestKit.shutdownActorSystem(slaveSystem, true);
205 TestKit.shutdownActorSystem(masterSystem, true);
208 @SuppressWarnings("unchecked")
210 public void testSlaveMountPointRegistration() throws InterruptedException, ExecutionException, TimeoutException {
213 ListenerRegistration<?> mockListenerReg = mock(ListenerRegistration.class);
214 doReturn(mockListenerReg).when(mockDataBroker).registerDataTreeChangeListener(any(), any());
216 final NodeId nodeId = new NodeId("device");
217 final NodeKey nodeKey = new NodeKey(nodeId);
218 final String topologyId = "topology-netconf";
219 final InstanceIdentifier<Node> nodeListPath = NetconfTopologyUtils.createTopologyNodeListPath(
220 nodeKey, topologyId);
222 netconfNodeManager.registerDataTreeChangeListener(topologyId, nodeKey);
223 verify(mockDataBroker).registerDataTreeChangeListener(any(), eq(netconfNodeManager));
225 // Invoke onDataTreeChanged with a NetconfNode WRITE to simulate the master writing the operational state to
226 // Connected. Expect the slave mount point created and registered.
228 final NetconfNode netconfNode = newNetconfNode();
229 final Node node = new NodeBuilder().setNodeId(nodeId).addAugmentation(netconfNode).build();
231 DataObjectModification<Node> mockDataObjModification = mock(DataObjectModification.class);
232 doReturn(Iterables.getLast(nodeListPath.getPathArguments())).when(mockDataObjModification).getIdentifier();
233 doReturn(WRITE).when(mockDataObjModification).getModificationType();
234 doReturn(node).when(mockDataObjModification).getDataAfter();
236 netconfNodeManager.onDataTreeChanged(List.of(
237 new NetconfTopologyManagerTest.CustomTreeModification(DataTreeIdentifier.create(
238 LogicalDatastoreType.OPERATIONAL, nodeListPath), mockDataObjModification)));
240 verify(mockMountPointBuilder, timeout(5000)).register();
241 verify(mockMountPointBuilder).addService(eq(DOMDataBroker.class), any());
242 verify(mockMountPointBuilder).addService(eq(DOMRpcService.class), any());
243 verify(mockMountPointBuilder).addService(eq(DOMNotificationService.class), any());
244 verify(mockMountPointService).createMountPoint(NetconfNodeUtils.defaultTopologyMountPath(DEVICE_ID));
246 // Notify that the NetconfNode operational state was deleted. Expect the slave mount point closed.
248 doReturn(DELETE).when(mockDataObjModification).getModificationType();
250 netconfNodeManager.onDataTreeChanged(List.of(
251 new NetconfTopologyManagerTest.CustomTreeModification(DataTreeIdentifier.create(
252 LogicalDatastoreType.OPERATIONAL, nodeListPath), mockDataObjModification)));
254 verify(mockMountPointReg, timeout(5000)).close();
256 // Notify with a NetconfNode operational state WRITE. Expect the slave mount point re-created.
258 setupMountPointMocks();
260 doReturn(WRITE).when(mockDataObjModification).getModificationType();
261 doReturn(null).when(mockDataObjModification).getDataBefore();
262 doReturn(node).when(mockDataObjModification).getDataAfter();
264 netconfNodeManager.onDataTreeChanged(List.of(
265 new NetconfTopologyManagerTest.CustomTreeModification(DataTreeIdentifier.create(
266 LogicalDatastoreType.OPERATIONAL, nodeListPath), mockDataObjModification)));
268 verify(mockMountPointBuilder, timeout(5000)).register();
270 // Notify again with a NetconfNode operational state WRITE. Expect the prior slave mount point closed and
271 // and a new one registered.
273 setupMountPointMocks();
275 doReturn(node).when(mockDataObjModification).getDataBefore();
277 netconfNodeManager.onDataTreeChanged(List.of(
278 new NetconfTopologyManagerTest.CustomTreeModification(DataTreeIdentifier.create(
279 LogicalDatastoreType.OPERATIONAL, nodeListPath), mockDataObjModification)));
281 verify(mockMountPointReg, timeout(5000)).close();
282 verify(mockMountPointBuilder, timeout(5000)).register();
284 // Notify that the NetconfNode operational state was changed to UnableToConnect. Expect the slave mount point
287 reset(mockMountPointService, mockMountPointBuilder, mockMountPointReg);
288 doNothing().when(mockMountPointReg).close();
290 final Node updatedNode = new NodeBuilder().setNodeId(nodeId)
291 .addAugmentation(new NetconfNodeBuilder(netconfNode)
292 .setConnectionStatus(ConnectionStatus.UnableToConnect)
296 doReturn(SUBTREE_MODIFIED).when(mockDataObjModification).getModificationType();
297 doReturn(node).when(mockDataObjModification).getDataBefore();
298 doReturn(updatedNode).when(mockDataObjModification).getDataAfter();
300 netconfNodeManager.onDataTreeChanged(List.of(
301 new NetconfTopologyManagerTest.CustomTreeModification(DataTreeIdentifier.create(
302 LogicalDatastoreType.OPERATIONAL, nodeListPath), mockDataObjModification)));
304 verify(mockMountPointReg, timeout(5000)).close();
306 netconfNodeManager.close();
307 verifyNoMoreInteractions(mockMountPointReg);
310 @SuppressWarnings("unchecked")
312 public void testSlaveMountPointRegistrationFailuresAndRetries()
313 throws InterruptedException, ExecutionException, TimeoutException {
314 final NodeId nodeId = new NodeId("device");
315 final NodeKey nodeKey = new NodeKey(nodeId);
316 final String topologyId = "topology-netconf";
317 final InstanceIdentifier<Node> nodeListPath = NetconfTopologyUtils.createTopologyNodeListPath(
318 nodeKey, topologyId);
320 final NetconfNode netconfNode = newNetconfNode();
321 final Node node = new NodeBuilder().setNodeId(nodeId).addAugmentation(netconfNode).build();
323 DataObjectModification<Node> mockDataObjModification = mock(DataObjectModification.class);
324 doReturn(Iterables.getLast(nodeListPath.getPathArguments())).when(mockDataObjModification).getIdentifier();
325 doReturn(WRITE).when(mockDataObjModification).getModificationType();
326 doReturn(node).when(mockDataObjModification).getDataAfter();
328 // First try the registration where the perceived master hasn't been initialized as the master.
330 netconfNodeManager.onDataTreeChanged(List.of(
331 new NetconfTopologyManagerTest.CustomTreeModification(DataTreeIdentifier.create(
332 LogicalDatastoreType.OPERATIONAL, nodeListPath), mockDataObjModification)));
334 verify(mockMountPointBuilder, after(1000).never()).register();
336 // Initialize the master but drop the initial YangTextSchemaSourceRequest message sent to the master so
341 CompletableFuture<AskForMasterMountPoint> yangTextSchemaSourceRequestFuture = new CompletableFuture<>();
342 testMasterActorRef.underlyingActor().messagesToDrop.put(YangTextSchemaSourceRequest.class,
343 yangTextSchemaSourceRequestFuture);
345 netconfNodeManager.onDataTreeChanged(List.of(
346 new NetconfTopologyManagerTest.CustomTreeModification(DataTreeIdentifier.create(
347 LogicalDatastoreType.OPERATIONAL, nodeListPath), mockDataObjModification)));
349 yangTextSchemaSourceRequestFuture.get(5, TimeUnit.SECONDS);
350 verify(mockMountPointBuilder, timeout(5000)).register();
352 // Initiate another registration but drop the initial AskForMasterMountPoint message sent to the master so
355 setupMountPointMocks();
357 CompletableFuture<AskForMasterMountPoint> askForMasterMountPointFuture = new CompletableFuture<>();
358 testMasterActorRef.underlyingActor().messagesToDrop.put(AskForMasterMountPoint.class,
359 askForMasterMountPointFuture);
361 netconfNodeManager.onDataTreeChanged(List.of(
362 new NetconfTopologyManagerTest.CustomTreeModification(DataTreeIdentifier.create(
363 LogicalDatastoreType.OPERATIONAL, nodeListPath), mockDataObjModification)));
365 askForMasterMountPointFuture.get(5, TimeUnit.SECONDS);
366 verify(mockMountPointReg, timeout(5000)).close();
367 verify(mockMountPointBuilder, timeout(5000)).register();
369 reset(mockMountPointService, mockMountPointBuilder, mockMountPointReg);
370 doNothing().when(mockMountPointReg).close();
371 netconfNodeManager.close();
372 verify(mockMountPointReg, timeout(5000)).close();
375 private NetconfNode newNetconfNode() {
376 return new NetconfNodeBuilder()
377 .setHost(new Host(new IpAddress(new Ipv4Address("127.0.0.1"))))
378 .setPort(new PortNumber(Uint16.valueOf(9999)))
379 .setConnectionStatus(ConnectionStatus.Connected)
380 .setClusteredConnectionStatus(new ClusteredConnectionStatusBuilder()
381 .setNetconfMasterNode(masterAddress).build())
385 private void setupMountPointMocks() {
386 reset(mockMountPointService, mockMountPointBuilder, mockMountPointReg);
387 doNothing().when(mockMountPointReg).close();
388 doReturn(mockMountPointReg).when(mockMountPointBuilder).register();
389 doReturn(mockMountPointBuilder).when(mockMountPointService).createMountPoint(any());
392 private void initializeMaster() {
393 TestKit kit = new TestKit(masterSystem);
394 testMasterActorRef.tell(new CreateInitialMasterActorData(mockDeviceDataBroker, netconfService,
395 SOURCE_IDENTIFIERS, new RemoteDeviceServices(mockRpcService, mockActionService)), kit.getRef());
397 kit.expectMsgClass(MasterActorDataInitialized.class);
400 private static class TestMasterActor extends NetconfNodeActor {
401 final Map<Class<?>, CompletableFuture<? extends Object>> messagesToDrop = new ConcurrentHashMap<>();
403 TestMasterActor(final NetconfTopologySetup setup, final RemoteDeviceId deviceId,
404 final Timeout actorResponseWaitTime, final DOMMountPointService mountPointService) {
405 super(setup, deviceId, actorResponseWaitTime, mountPointService);
408 @SuppressWarnings({ "rawtypes", "unchecked" })
410 public void handleReceive(final Object message) {
411 CompletableFuture dropFuture = messagesToDrop.remove(message.getClass());
412 if (dropFuture != null) {
413 dropFuture.complete(message);
415 super.handleReceive(message);