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.mockito.MockitoAnnotations.initMocks;
  22 import static org.opendaylight.mdsal.binding.api.DataObjectModification.ModificationType.DELETE;
  23 import static org.opendaylight.mdsal.binding.api.DataObjectModification.ModificationType.SUBTREE_MODIFIED;
  24 import static org.opendaylight.mdsal.binding.api.DataObjectModification.ModificationType.WRITE;
  25
  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;
  41 import java.util.Map;
  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.mdsal.binding.api.DataBroker;
  53 import org.opendaylight.mdsal.binding.api.DataObjectModification;
  54 import org.opendaylight.mdsal.binding.api.DataTreeIdentifier;
  55 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
  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.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;
  93
  94 /**
  95  * Unit tests for NetconfNodeManager.
  96  *
  97  * @author Thomas Pantelis
  98  */
  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"));
 104
 105     @Mock
 106     private DOMMountPointService mockMountPointService;
 107
 108     @Mock
 109     private DOMMountPointService.DOMMountPointBuilder mockMountPointBuilder;
 110
 111     @Mock
 112     private ObjectRegistration<DOMMountPoint> mockMountPointReg;
 113
 114     @Mock
 115     private DataBroker mockDataBroker;
 116
 117     @Mock
 118     private DOMDataBroker mockDeviceDataBroker;
 119
 120     @Mock
 121     private DOMRpcService mockRpcService;
 122
 123     @Mock
 124     private NetconfDeviceSchemasResolver mockSchemasResolver;
 125
 126     @Mock
 127     private SchemaContextFactory mockSchemaContextFactory;
 128
 129     private ActorSystem slaveSystem;
 130     private ActorSystem masterSystem;
 131     private TestActorRef<TestMasterActor> testMasterActorRef;
 132     private NetconfNodeManager netconfNodeManager;
 133     private String masterAddress;
 134
 135     @Before
 136     public void setup() {
 137         initMocks(this);
 138
 139         final Timeout responseTimeout = Timeout.apply(1, TimeUnit.SECONDS);
 140
 141         slaveSystem = ActorSystem.create(ACTOR_SYSTEM_NAME, ConfigFactory.load().getConfig("Slave"));
 142         masterSystem = ActorSystem.create(ACTOR_SYSTEM_NAME, ConfigFactory.load().getConfig("Master"));
 143
 144         masterAddress = Cluster.get(masterSystem).selfAddress().toString();
 145
 146         SharedSchemaRepository masterSchemaRepository = new SharedSchemaRepository("master");
 147         masterSchemaRepository.registerSchemaSourceListener(
 148                 TextToASTTransformer.create(masterSchemaRepository, masterSchemaRepository));
 149
 150         String yangTemplate =
 151                   "module ID {"
 152                 + "  namespace \"ID\";"
 153                 + "  prefix ID;"
 154                 + "}";
 155
 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());
 162
 163         NetconfTopologySetup masterSetup = new NetconfTopologySetup.NetconfTopologySetupBuilder()
 164                 .setActorSystem(masterSystem).setDataBroker(mockDataBroker).setSchemaResourceDTO(
 165                         new NetconfDevice.SchemaResourcesDTO(masterSchemaRepository, masterSchemaRepository,
 166                                 mockSchemaContextFactory, mockSchemasResolver)).build();
 167
 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));
 171
 172         SharedSchemaRepository slaveSchemaRepository = new SharedSchemaRepository("slave");
 173         slaveSchemaRepository.registerSchemaSourceListener(
 174                 TextToASTTransformer.create(slaveSchemaRepository, slaveSchemaRepository));
 175
 176         NetconfTopologySetup slaveSetup = new NetconfTopologySetup.NetconfTopologySetupBuilder()
 177                 .setActorSystem(slaveSystem).setDataBroker(mockDataBroker).setSchemaResourceDTO(
 178                         new NetconfDevice.SchemaResourcesDTO(slaveSchemaRepository, slaveSchemaRepository,
 179                                 mockSchemaContextFactory, mockSchemasResolver)).build();
 180
 181         netconfNodeManager = new NetconfNodeManager(slaveSetup, DEVICE_ID, responseTimeout,
 182                 mockMountPointService);
 183
 184         setupMountPointMocks();
 185     }
 186
 187     @After
 188     public void teardown() {
 189         TestKit.shutdownActorSystem(slaveSystem, true);
 190         TestKit.shutdownActorSystem(masterSystem, true);
 191     }
 192
 193     @SuppressWarnings("unchecked")
 194     @Test
 195     public void testSlaveMountPointRegistration() throws InterruptedException, ExecutionException, TimeoutException {
 196         initializeMaster();
 197
 198         ListenerRegistration<?> mockListenerReg = mock(ListenerRegistration.class);
 199         doReturn(mockListenerReg).when(mockDataBroker).registerDataTreeChangeListener(any(), any());
 200
 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);
 206
 207         netconfNodeManager.registerDataTreeChangeListener(topologyId, nodeKey);
 208         verify(mockDataBroker).registerDataTreeChangeListener(any(), eq(netconfNodeManager));
 209
 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.
 212
 213         final NetconfNode netconfNode = newNetconfNode();
 214         final Node node = new NodeBuilder().setNodeId(nodeId).addAugmentation(NetconfNode.class, netconfNode).build();
 215
 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();
 220
 221         netconfNodeManager.onDataTreeChanged(Collections.singletonList(
 222                 new NetconfTopologyManagerTest.CustomTreeModification(DataTreeIdentifier.create(
 223                         LogicalDatastoreType.OPERATIONAL, nodeListPath), mockDataObjModification)));
 224
 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());
 230
 231         // Notify that the NetconfNode operational state was deleted. Expect the slave mount point closed.
 232
 233         doReturn(DELETE).when(mockDataObjModification).getModificationType();
 234         doReturn(node).when(mockDataObjModification).getDataBefore();
 235         doReturn(null).when(mockDataObjModification).getDataAfter();
 236
 237         netconfNodeManager.onDataTreeChanged(Collections.singletonList(
 238                 new NetconfTopologyManagerTest.CustomTreeModification(DataTreeIdentifier.create(
 239                         LogicalDatastoreType.OPERATIONAL, nodeListPath), mockDataObjModification)));
 240
 241         verify(mockMountPointReg, timeout(5000)).close();
 242
 243         // Notify with a NetconfNode operational state WRITE. Expect the slave mount point re-created.
 244
 245         setupMountPointMocks();
 246
 247         doReturn(WRITE).when(mockDataObjModification).getModificationType();
 248         doReturn(null).when(mockDataObjModification).getDataBefore();
 249         doReturn(node).when(mockDataObjModification).getDataAfter();
 250
 251         netconfNodeManager.onDataTreeChanged(Collections.singletonList(
 252                 new NetconfTopologyManagerTest.CustomTreeModification(DataTreeIdentifier.create(
 253                         LogicalDatastoreType.OPERATIONAL, nodeListPath), mockDataObjModification)));
 254
 255         verify(mockMountPointBuilder, timeout(5000)).register();
 256
 257         // Notify again with a NetconfNode operational state WRITE. Expect the prior slave mount point closed and
 258         // and a new one registered.
 259
 260         setupMountPointMocks();
 261
 262         doReturn(node).when(mockDataObjModification).getDataBefore();
 263
 264         netconfNodeManager.onDataTreeChanged(Collections.singletonList(
 265                 new NetconfTopologyManagerTest.CustomTreeModification(DataTreeIdentifier.create(
 266                         LogicalDatastoreType.OPERATIONAL, nodeListPath), mockDataObjModification)));
 267
 268         verify(mockMountPointReg, timeout(5000)).close();
 269         verify(mockMountPointBuilder, timeout(5000)).register();
 270
 271         // Notify that the NetconfNode operational state was changed to UnableToConnect. Expect the slave mount point
 272         // closed.
 273
 274         setupMountPointMocks();
 275
 276         final Node updatedNode = new NodeBuilder().setNodeId(nodeId)
 277                 .addAugmentation(NetconfNode.class, new NetconfNodeBuilder(netconfNode)
 278                         .setConnectionStatus(NetconfNodeConnectionStatus.ConnectionStatus.UnableToConnect)
 279                         .build()).build();
 280
 281         doReturn(SUBTREE_MODIFIED).when(mockDataObjModification).getModificationType();
 282         doReturn(node).when(mockDataObjModification).getDataBefore();
 283         doReturn(updatedNode).when(mockDataObjModification).getDataAfter();
 284
 285         netconfNodeManager.onDataTreeChanged(Collections.singletonList(
 286                 new NetconfTopologyManagerTest.CustomTreeModification(DataTreeIdentifier.create(
 287                         LogicalDatastoreType.OPERATIONAL, nodeListPath), mockDataObjModification)));
 288
 289         verify(mockMountPointReg, timeout(5000)).close();
 290
 291         netconfNodeManager.close();
 292         verifyNoMoreInteractions(mockMountPointReg);
 293     }
 294
 295     @SuppressWarnings("unchecked")
 296     @Test
 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);
 304
 305         final NetconfNode netconfNode = newNetconfNode();
 306         final Node node = new NodeBuilder().setNodeId(nodeId).addAugmentation(NetconfNode.class, netconfNode).build();
 307
 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();
 312
 313         // First try the registration where the perceived master hasn't been initialized as the master.
 314
 315         netconfNodeManager.onDataTreeChanged(Collections.singletonList(
 316                 new NetconfTopologyManagerTest.CustomTreeModification(DataTreeIdentifier.create(
 317                         LogicalDatastoreType.OPERATIONAL, nodeListPath), mockDataObjModification)));
 318
 319         verify(mockMountPointBuilder, after(1000).never()).register();
 320
 321         // Initialize the master but drop the initial YangTextSchemaSourceRequest message sent to the master so
 322         // it retries.
 323
 324         initializeMaster();
 325
 326         CompletableFuture<AskForMasterMountPoint> yangTextSchemaSourceRequestFuture = new CompletableFuture<>();
 327         testMasterActorRef.underlyingActor().messagesToDrop.put(YangTextSchemaSourceRequest.class,
 328                 yangTextSchemaSourceRequestFuture);
 329
 330         netconfNodeManager.onDataTreeChanged(Collections.singletonList(
 331                 new NetconfTopologyManagerTest.CustomTreeModification(DataTreeIdentifier.create(
 332                         LogicalDatastoreType.OPERATIONAL, nodeListPath), mockDataObjModification)));
 333
 334         yangTextSchemaSourceRequestFuture.get(5, TimeUnit.SECONDS);
 335         verify(mockMountPointBuilder, timeout(5000)).register();
 336
 337         // Initiate another registration but drop the initial AskForMasterMountPoint message sent to the master so
 338         // it retries.
 339
 340         setupMountPointMocks();
 341
 342         CompletableFuture<AskForMasterMountPoint> askForMasterMountPointFuture = new CompletableFuture<>();
 343         testMasterActorRef.underlyingActor().messagesToDrop.put(AskForMasterMountPoint.class,
 344                 askForMasterMountPointFuture);
 345
 346         netconfNodeManager.onDataTreeChanged(Collections.singletonList(
 347                 new NetconfTopologyManagerTest.CustomTreeModification(DataTreeIdentifier.create(
 348                         LogicalDatastoreType.OPERATIONAL, nodeListPath), mockDataObjModification)));
 349
 350         askForMasterMountPointFuture.get(5, TimeUnit.SECONDS);
 351         verify(mockMountPointReg, timeout(5000)).close();
 352         verify(mockMountPointBuilder, timeout(5000)).register();
 353
 354         setupMountPointMocks();
 355         netconfNodeManager.close();
 356         verify(mockMountPointReg, timeout(5000)).close();
 357     }
 358
 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())
 366                 .build();
 367     }
 368
 369     private void setupMountPointMocks() {
 370         reset(mockMountPointService, mockMountPointBuilder, mockMountPointReg);
 371
 372         doNothing().when(mockMountPointReg).close();
 373
 374         doReturn(mockMountPointBuilder).when(mockMountPointBuilder).addInitialSchemaContext(any());
 375         doReturn(mockMountPointBuilder).when(mockMountPointBuilder).addService(any(), any());
 376         doReturn(mockMountPointReg).when(mockMountPointBuilder).register();
 377
 378         doReturn(mockMountPointBuilder).when(mockMountPointService).createMountPoint(any());
 379     }
 380
 381     private void initializeMaster() {
 382         TestKit kit = new TestKit(masterSystem);
 383
 384         testMasterActorRef.tell(new CreateInitialMasterActorData(mockDeviceDataBroker, SOURCE_IDENTIFIERS,
 385                 mockRpcService), kit.getRef());
 386
 387         kit.expectMsgClass(MasterActorDataInitialized.class);
 388     }
 389
 390     private static class TestMasterActor extends NetconfNodeActor {
 391         final Map<Class<?>, CompletableFuture<? extends Object>> messagesToDrop = new ConcurrentHashMap<>();
 392
 393         TestMasterActor(final NetconfTopologySetup setup, final RemoteDeviceId deviceId,
 394                 final Timeout actorResponseWaitTime, final DOMMountPointService mountPointService) {
 395             super(setup, deviceId, actorResponseWaitTime, mountPointService);
 396         }
 397
 398         @SuppressWarnings({ "rawtypes", "unchecked" })
 399         @Override
 400         public void handleReceive(final Object message) {
 401             CompletableFuture dropFuture = messagesToDrop.remove(message.getClass());
 402             if (dropFuture != null) {
 403                 dropFuture.complete(message);
 404             } else {
 405                 super.handleReceive(message);
 406             }
 407         }
 408     }
 409 }