netconf/netconf-topology-singleton/src/test/java/org/opendaylight/netconf/topology/singleton/impl/NetconfNodeManagerTest.java

   1 /*
   2  * Copyright (c) 2018 Inocybe Technologies and others.  All rights reserved.
   3  *
   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
   7  */
   8 package org.opendaylight.netconf.topology.singleton.impl;
   9
  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;
  24
  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;
  38 import java.util.Map;
  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.DOMActionService;
  56 import org.opendaylight.mdsal.dom.api.DOMDataBroker;
  57 import org.opendaylight.mdsal.dom.api.DOMMountPoint;
  58 import org.opendaylight.mdsal.dom.api.DOMMountPointService;
  59 import org.opendaylight.mdsal.dom.api.DOMNotificationService;
  60 import org.opendaylight.mdsal.dom.api.DOMRpcService;
  61 import org.opendaylight.netconf.dom.api.NetconfDataTreeService;
  62 import org.opendaylight.netconf.sal.connect.api.NetconfDeviceSchemasResolver;
  63 import org.opendaylight.netconf.sal.connect.netconf.NetconfDevice;
  64 import org.opendaylight.netconf.sal.connect.util.RemoteDeviceId;
  65 import org.opendaylight.netconf.topology.singleton.impl.actors.NetconfNodeActor;
  66 import org.opendaylight.netconf.topology.singleton.impl.utils.NetconfTopologySetup;
  67 import org.opendaylight.netconf.topology.singleton.impl.utils.NetconfTopologyUtils;
  68 import org.opendaylight.netconf.topology.singleton.messages.AskForMasterMountPoint;
  69 import org.opendaylight.netconf.topology.singleton.messages.CreateInitialMasterActorData;
  70 import org.opendaylight.netconf.topology.singleton.messages.MasterActorDataInitialized;
  71 import org.opendaylight.netconf.topology.singleton.messages.YangTextSchemaSourceRequest;
  72 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev130715.Host;
  73 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev130715.IpAddress;
  74 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev130715.Ipv4Address;
  75 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev130715.PortNumber;
  76 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.NetconfNode;
  77 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.NetconfNodeBuilder;
  78 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.NetconfNodeConnectionStatus;
  79 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.netconf.node.connection.status.ClusteredConnectionStatusBuilder;
  80 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.NodeId;
  81 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.Node;
  82 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.NodeBuilder;
  83 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.NodeKey;
  84 import org.opendaylight.yangtools.concepts.ListenerRegistration;
  85 import org.opendaylight.yangtools.concepts.ObjectRegistration;
  86 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  87 import org.opendaylight.yangtools.yang.common.Uint16;
  88 import org.opendaylight.yangtools.yang.model.repo.api.EffectiveModelContextFactory;
  89 import org.opendaylight.yangtools.yang.model.repo.api.SourceIdentifier;
  90 import org.opendaylight.yangtools.yang.model.repo.api.YangTextSchemaSource;
  91 import org.opendaylight.yangtools.yang.model.repo.spi.PotentialSchemaSource;
  92 import org.opendaylight.yangtools.yang.parser.repo.SharedSchemaRepository;
  93 import org.opendaylight.yangtools.yang.parser.rfc7950.repo.TextToIRTransformer;
  94
  95 /**
  96  * Unit tests for NetconfNodeManager.
  97  *
  98  * @author Thomas Pantelis
  99  */
 100 @RunWith(MockitoJUnitRunner.StrictStubs.class)
 101 public class NetconfNodeManagerTest extends AbstractBaseSchemasTest {
 102     private static final String ACTOR_SYSTEM_NAME = "test";
 103     private static final RemoteDeviceId DEVICE_ID = new RemoteDeviceId("device", new InetSocketAddress(65535));
 104     private static final List<SourceIdentifier> SOURCE_IDENTIFIERS = List.of(new SourceIdentifier("testID"));
 105
 106     @Mock
 107     private DOMMountPointService mockMountPointService;
 108
 109     @Mock
 110     private DOMMountPointService.DOMMountPointBuilder mockMountPointBuilder;
 111
 112     @Mock
 113     private ObjectRegistration<DOMMountPoint> mockMountPointReg;
 114
 115     @Mock
 116     private DataBroker mockDataBroker;
 117
 118     @Mock
 119     private NetconfDataTreeService netconfService;
 120
 121     @Mock
 122     private DOMDataBroker mockDeviceDataBroker;
 123
 124     @Mock
 125     private DOMRpcService mockRpcService;
 126
 127     @Mock
 128     private DOMActionService mockActionService;
 129
 130     @Mock
 131     private NetconfDeviceSchemasResolver mockSchemasResolver;
 132
 133     @Mock
 134     private EffectiveModelContextFactory mockSchemaContextFactory;
 135
 136     private ActorSystem slaveSystem;
 137     private ActorSystem masterSystem;
 138     private TestActorRef<TestMasterActor> testMasterActorRef;
 139     private NetconfNodeManager netconfNodeManager;
 140     private String masterAddress;
 141
 142     @Before
 143     public void setup() {
 144         final Timeout responseTimeout = Timeout.apply(1, TimeUnit.SECONDS);
 145
 146         slaveSystem = ActorSystem.create(ACTOR_SYSTEM_NAME, ConfigFactory.load().getConfig("Slave"));
 147         masterSystem = ActorSystem.create(ACTOR_SYSTEM_NAME, ConfigFactory.load().getConfig("Master"));
 148
 149         masterAddress = Cluster.get(masterSystem).selfAddress().toString();
 150
 151         SharedSchemaRepository masterSchemaRepository = new SharedSchemaRepository("master");
 152         masterSchemaRepository.registerSchemaSourceListener(
 153                 TextToIRTransformer.create(masterSchemaRepository, masterSchemaRepository));
 154
 155         final String yangTemplate = """
 156             module ID {\
 157               namespace "ID";\
 158               prefix ID;\
 159             }""";
 160
 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());
 167
 168         NetconfTopologySetup masterSetup = new NetconfTopologySetup.NetconfTopologySetupBuilder()
 169                 .setActorSystem(masterSystem).setDataBroker(mockDataBroker).setSchemaResourceDTO(
 170                         new NetconfDevice.SchemaResourcesDTO(masterSchemaRepository, masterSchemaRepository,
 171                                 mockSchemaContextFactory, mockSchemasResolver)).setBaseSchemas(BASE_SCHEMAS).build();
 172
 173         testMasterActorRef = TestActorRef.create(masterSystem, Props.create(TestMasterActor.class, masterSetup,
 174                 DEVICE_ID, responseTimeout, mockMountPointService).withDispatcher(Dispatchers.DefaultDispatcherId()),
 175                 NetconfTopologyUtils.createMasterActorName(DEVICE_ID.getName(), masterAddress));
 176
 177         SharedSchemaRepository slaveSchemaRepository = new SharedSchemaRepository("slave");
 178         slaveSchemaRepository.registerSchemaSourceListener(
 179                 TextToIRTransformer.create(slaveSchemaRepository, slaveSchemaRepository));
 180
 181         NetconfTopologySetup slaveSetup = new NetconfTopologySetup.NetconfTopologySetupBuilder()
 182                 .setActorSystem(slaveSystem).setDataBroker(mockDataBroker).setSchemaResourceDTO(
 183                         new NetconfDevice.SchemaResourcesDTO(slaveSchemaRepository, slaveSchemaRepository,
 184                                 mockSchemaContextFactory, mockSchemasResolver)).setBaseSchemas(BASE_SCHEMAS).build();
 185
 186         netconfNodeManager = new NetconfNodeManager(slaveSetup, DEVICE_ID, responseTimeout,
 187                 mockMountPointService);
 188
 189         setupMountPointMocks();
 190     }
 191
 192     @After
 193     public void teardown() {
 194         TestKit.shutdownActorSystem(slaveSystem, true);
 195         TestKit.shutdownActorSystem(masterSystem, true);
 196     }
 197
 198     @SuppressWarnings("unchecked")
 199     @Test
 200     public void testSlaveMountPointRegistration() throws InterruptedException, ExecutionException, TimeoutException {
 201         initializeMaster();
 202
 203         ListenerRegistration<?> mockListenerReg = mock(ListenerRegistration.class);
 204         doReturn(mockListenerReg).when(mockDataBroker).registerDataTreeChangeListener(any(), any());
 205
 206         final NodeId nodeId = new NodeId("device");
 207         final NodeKey nodeKey = new NodeKey(nodeId);
 208         final String topologyId = "topology-netconf";
 209         final InstanceIdentifier<Node> nodeListPath = NetconfTopologyUtils.createTopologyNodeListPath(
 210                 nodeKey, topologyId);
 211
 212         netconfNodeManager.registerDataTreeChangeListener(topologyId, nodeKey);
 213         verify(mockDataBroker).registerDataTreeChangeListener(any(), eq(netconfNodeManager));
 214
 215         // Invoke onDataTreeChanged with a NetconfNode WRITE to simulate the master writing the operational state to
 216         // Connected. Expect the slave mount point created and registered.
 217
 218         final NetconfNode netconfNode = newNetconfNode();
 219         final Node node = new NodeBuilder().setNodeId(nodeId).addAugmentation(netconfNode).build();
 220
 221         DataObjectModification<Node> mockDataObjModification = mock(DataObjectModification.class);
 222         doReturn(Iterables.getLast(nodeListPath.getPathArguments())).when(mockDataObjModification).getIdentifier();
 223         doReturn(WRITE).when(mockDataObjModification).getModificationType();
 224         doReturn(node).when(mockDataObjModification).getDataAfter();
 225
 226         netconfNodeManager.onDataTreeChanged(List.of(
 227                 new NetconfTopologyManagerTest.CustomTreeModification(DataTreeIdentifier.create(
 228                         LogicalDatastoreType.OPERATIONAL, nodeListPath), mockDataObjModification)));
 229
 230         verify(mockMountPointBuilder, timeout(5000)).register();
 231         verify(mockMountPointBuilder).addService(eq(DOMDataBroker.class), any());
 232         verify(mockMountPointBuilder).addService(eq(DOMRpcService.class), any());
 233         verify(mockMountPointBuilder).addService(eq(DOMNotificationService.class), any());
 234         verify(mockMountPointService).createMountPoint(DEVICE_ID.getTopologyPath());
 235
 236         // Notify that the NetconfNode operational state was deleted. Expect the slave mount point closed.
 237
 238         doReturn(DELETE).when(mockDataObjModification).getModificationType();
 239
 240         netconfNodeManager.onDataTreeChanged(List.of(
 241                 new NetconfTopologyManagerTest.CustomTreeModification(DataTreeIdentifier.create(
 242                         LogicalDatastoreType.OPERATIONAL, nodeListPath), mockDataObjModification)));
 243
 244         verify(mockMountPointReg, timeout(5000)).close();
 245
 246         // Notify with a NetconfNode operational state WRITE. Expect the slave mount point re-created.
 247
 248         setupMountPointMocks();
 249
 250         doReturn(WRITE).when(mockDataObjModification).getModificationType();
 251         doReturn(null).when(mockDataObjModification).getDataBefore();
 252         doReturn(node).when(mockDataObjModification).getDataAfter();
 253
 254         netconfNodeManager.onDataTreeChanged(List.of(
 255                 new NetconfTopologyManagerTest.CustomTreeModification(DataTreeIdentifier.create(
 256                         LogicalDatastoreType.OPERATIONAL, nodeListPath), mockDataObjModification)));
 257
 258         verify(mockMountPointBuilder, timeout(5000)).register();
 259
 260         // Notify again with a NetconfNode operational state WRITE. Expect the prior slave mount point closed and
 261         // and a new one registered.
 262
 263         setupMountPointMocks();
 264
 265         doReturn(node).when(mockDataObjModification).getDataBefore();
 266
 267         netconfNodeManager.onDataTreeChanged(List.of(
 268                 new NetconfTopologyManagerTest.CustomTreeModification(DataTreeIdentifier.create(
 269                         LogicalDatastoreType.OPERATIONAL, nodeListPath), mockDataObjModification)));
 270
 271         verify(mockMountPointReg, timeout(5000)).close();
 272         verify(mockMountPointBuilder, timeout(5000)).register();
 273
 274         // Notify that the NetconfNode operational state was changed to UnableToConnect. Expect the slave mount point
 275         // closed.
 276
 277         reset(mockMountPointService, mockMountPointBuilder, mockMountPointReg);
 278         doNothing().when(mockMountPointReg).close();
 279
 280         final Node updatedNode = new NodeBuilder().setNodeId(nodeId)
 281                 .addAugmentation(new NetconfNodeBuilder(netconfNode)
 282                     .setConnectionStatus(NetconfNodeConnectionStatus.ConnectionStatus.UnableToConnect)
 283                     .build())
 284                 .build();
 285
 286         doReturn(SUBTREE_MODIFIED).when(mockDataObjModification).getModificationType();
 287         doReturn(node).when(mockDataObjModification).getDataBefore();
 288         doReturn(updatedNode).when(mockDataObjModification).getDataAfter();
 289
 290         netconfNodeManager.onDataTreeChanged(List.of(
 291                 new NetconfTopologyManagerTest.CustomTreeModification(DataTreeIdentifier.create(
 292                         LogicalDatastoreType.OPERATIONAL, nodeListPath), mockDataObjModification)));
 293
 294         verify(mockMountPointReg, timeout(5000)).close();
 295
 296         netconfNodeManager.close();
 297         verifyNoMoreInteractions(mockMountPointReg);
 298     }
 299
 300     @SuppressWarnings("unchecked")
 301     @Test
 302     public void testSlaveMountPointRegistrationFailuresAndRetries()
 303             throws InterruptedException, ExecutionException, TimeoutException {
 304         final NodeId nodeId = new NodeId("device");
 305         final NodeKey nodeKey = new NodeKey(nodeId);
 306         final String topologyId = "topology-netconf";
 307         final InstanceIdentifier<Node> nodeListPath = NetconfTopologyUtils.createTopologyNodeListPath(
 308                 nodeKey, topologyId);
 309
 310         final NetconfNode netconfNode = newNetconfNode();
 311         final Node node = new NodeBuilder().setNodeId(nodeId).addAugmentation(netconfNode).build();
 312
 313         DataObjectModification<Node> mockDataObjModification = mock(DataObjectModification.class);
 314         doReturn(Iterables.getLast(nodeListPath.getPathArguments())).when(mockDataObjModification).getIdentifier();
 315         doReturn(WRITE).when(mockDataObjModification).getModificationType();
 316         doReturn(node).when(mockDataObjModification).getDataAfter();
 317
 318         // First try the registration where the perceived master hasn't been initialized as the master.
 319
 320         netconfNodeManager.onDataTreeChanged(List.of(
 321                 new NetconfTopologyManagerTest.CustomTreeModification(DataTreeIdentifier.create(
 322                         LogicalDatastoreType.OPERATIONAL, nodeListPath), mockDataObjModification)));
 323
 324         verify(mockMountPointBuilder, after(1000).never()).register();
 325
 326         // Initialize the master but drop the initial YangTextSchemaSourceRequest message sent to the master so
 327         // it retries.
 328
 329         initializeMaster();
 330
 331         CompletableFuture<AskForMasterMountPoint> yangTextSchemaSourceRequestFuture = new CompletableFuture<>();
 332         testMasterActorRef.underlyingActor().messagesToDrop.put(YangTextSchemaSourceRequest.class,
 333                 yangTextSchemaSourceRequestFuture);
 334
 335         netconfNodeManager.onDataTreeChanged(List.of(
 336                 new NetconfTopologyManagerTest.CustomTreeModification(DataTreeIdentifier.create(
 337                         LogicalDatastoreType.OPERATIONAL, nodeListPath), mockDataObjModification)));
 338
 339         yangTextSchemaSourceRequestFuture.get(5, TimeUnit.SECONDS);
 340         verify(mockMountPointBuilder, timeout(5000)).register();
 341
 342         // Initiate another registration but drop the initial AskForMasterMountPoint message sent to the master so
 343         // it retries.
 344
 345         setupMountPointMocks();
 346
 347         CompletableFuture<AskForMasterMountPoint> askForMasterMountPointFuture = new CompletableFuture<>();
 348         testMasterActorRef.underlyingActor().messagesToDrop.put(AskForMasterMountPoint.class,
 349                 askForMasterMountPointFuture);
 350
 351         netconfNodeManager.onDataTreeChanged(List.of(
 352                 new NetconfTopologyManagerTest.CustomTreeModification(DataTreeIdentifier.create(
 353                         LogicalDatastoreType.OPERATIONAL, nodeListPath), mockDataObjModification)));
 354
 355         askForMasterMountPointFuture.get(5, TimeUnit.SECONDS);
 356         verify(mockMountPointReg, timeout(5000)).close();
 357         verify(mockMountPointBuilder, timeout(5000)).register();
 358
 359         reset(mockMountPointService, mockMountPointBuilder, mockMountPointReg);
 360         doNothing().when(mockMountPointReg).close();
 361         netconfNodeManager.close();
 362         verify(mockMountPointReg, timeout(5000)).close();
 363     }
 364
 365     private NetconfNode newNetconfNode() {
 366         return new NetconfNodeBuilder()
 367                 .setHost(new Host(new IpAddress(new Ipv4Address("127.0.0.1"))))
 368                 .setPort(new PortNumber(Uint16.valueOf(9999)))
 369                 .setConnectionStatus(NetconfNodeConnectionStatus.ConnectionStatus.Connected)
 370                 .setClusteredConnectionStatus(new ClusteredConnectionStatusBuilder()
 371                         .setNetconfMasterNode(masterAddress).build())
 372                 .build();
 373     }
 374
 375     private void setupMountPointMocks() {
 376         reset(mockMountPointService, mockMountPointBuilder, mockMountPointReg);
 377         doNothing().when(mockMountPointReg).close();
 378         doReturn(mockMountPointReg).when(mockMountPointBuilder).register();
 379         doReturn(mockMountPointBuilder).when(mockMountPointService).createMountPoint(any());
 380     }
 381
 382     private void initializeMaster() {
 383         TestKit kit = new TestKit(masterSystem);
 384         testMasterActorRef.tell(new CreateInitialMasterActorData(mockDeviceDataBroker, netconfService,
 385             SOURCE_IDENTIFIERS,
 386                 mockRpcService, mockActionService), kit.getRef());
 387
 388         kit.expectMsgClass(MasterActorDataInitialized.class);
 389     }
 390
 391     private static class TestMasterActor extends NetconfNodeActor {
 392         final Map<Class<?>, CompletableFuture<? extends Object>> messagesToDrop = new ConcurrentHashMap<>();
 393
 394         TestMasterActor(final NetconfTopologySetup setup, final RemoteDeviceId deviceId,
 395                 final Timeout actorResponseWaitTime, final DOMMountPointService mountPointService) {
 396             super(setup, deviceId, actorResponseWaitTime, mountPointService);
 397         }
 398
 399         @SuppressWarnings({ "rawtypes", "unchecked" })
 400         @Override
 401         public void handleReceive(final Object message) {
 402             CompletableFuture dropFuture = messagesToDrop.remove(message.getClass());
 403             if (dropFuture != null) {
 404                 dropFuture.complete(message);
 405             } else {
 406                 super.handleReceive(message);
 407             }
 408         }
 409     }
 410 }