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

   1 /*
   2  * Copyright (c) 2016 Cisco Systems, Inc. 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 org.hamcrest.CoreMatchers.containsString;
  11 import static org.hamcrest.CoreMatchers.instanceOf;
  12 import static org.hamcrest.CoreMatchers.startsWith;
  13 import static org.hamcrest.MatcherAssert.assertThat;
  14 import static org.junit.Assert.assertEquals;
  15 import static org.junit.Assert.assertNull;
  16 import static org.junit.Assert.assertThrows;
  17 import static org.junit.Assert.assertTrue;
  18 import static org.mockito.ArgumentMatchers.any;
  19 import static org.mockito.ArgumentMatchers.anyCollection;
  20 import static org.mockito.ArgumentMatchers.argThat;
  21 import static org.mockito.ArgumentMatchers.eq;
  22 import static org.mockito.Mockito.after;
  23 import static org.mockito.Mockito.doAnswer;
  24 import static org.mockito.Mockito.doNothing;
  25 import static org.mockito.Mockito.doReturn;
  26 import static org.mockito.Mockito.mock;
  27 import static org.mockito.Mockito.reset;
  28 import static org.mockito.Mockito.timeout;
  29 import static org.mockito.Mockito.times;
  30 import static org.mockito.Mockito.verify;
  31 import static org.mockito.Mockito.verifyNoMoreInteractions;
  32 import static org.opendaylight.mdsal.common.api.CommitInfo.emptyFluentFuture;
  33 import static org.opendaylight.yangtools.util.concurrent.FluentFutures.immediateFluentFuture;
  34
  35 import akka.actor.ActorRef;
  36 import akka.actor.ActorSystem;
  37 import akka.actor.Props;
  38 import akka.actor.Status.Failure;
  39 import akka.actor.Status.Success;
  40 import akka.pattern.AskTimeoutException;
  41 import akka.pattern.Patterns;
  42 import akka.testkit.TestActorRef;
  43 import akka.testkit.javadsl.TestKit;
  44 import akka.util.Timeout;
  45 import com.google.common.collect.ImmutableList;
  46 import com.google.common.io.CharSource;
  47 import com.google.common.net.InetAddresses;
  48 import com.google.common.util.concurrent.FluentFuture;
  49 import com.google.common.util.concurrent.Futures;
  50 import com.google.common.util.concurrent.ListenableFuture;
  51 import com.google.common.util.concurrent.SettableFuture;
  52 import java.net.InetSocketAddress;
  53 import java.time.Duration;
  54 import java.util.ArrayList;
  55 import java.util.List;
  56 import java.util.Optional;
  57 import java.util.concurrent.ExecutionException;
  58 import java.util.concurrent.TimeUnit;
  59 import org.junit.After;
  60 import org.junit.Before;
  61 import org.junit.Test;
  62 import org.junit.runner.RunWith;
  63 import org.mockito.ArgumentCaptor;
  64 import org.mockito.Mock;
  65 import org.mockito.junit.MockitoJUnitRunner;
  66 import org.opendaylight.controller.cluster.schema.provider.impl.YangTextSchemaSourceSerializationProxy;
  67 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
  68 import org.opendaylight.mdsal.dom.api.DOMActionException;
  69 import org.opendaylight.mdsal.dom.api.DOMActionResult;
  70 import org.opendaylight.mdsal.dom.api.DOMActionService;
  71 import org.opendaylight.mdsal.dom.api.DOMDataBroker;
  72 import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier;
  73 import org.opendaylight.mdsal.dom.api.DOMDataTreeReadTransaction;
  74 import org.opendaylight.mdsal.dom.api.DOMDataTreeReadWriteTransaction;
  75 import org.opendaylight.mdsal.dom.api.DOMDataTreeWriteTransaction;
  76 import org.opendaylight.mdsal.dom.api.DOMMountPoint;
  77 import org.opendaylight.mdsal.dom.api.DOMMountPointService;
  78 import org.opendaylight.mdsal.dom.api.DOMNotificationService;
  79 import org.opendaylight.mdsal.dom.api.DOMRpcException;
  80 import org.opendaylight.mdsal.dom.api.DOMRpcResult;
  81 import org.opendaylight.mdsal.dom.api.DOMRpcService;
  82 import org.opendaylight.mdsal.dom.api.DOMSchemaService;
  83 import org.opendaylight.mdsal.dom.spi.DefaultDOMRpcResult;
  84 import org.opendaylight.mdsal.dom.spi.SimpleDOMActionResult;
  85 import org.opendaylight.netconf.client.mdsal.NetconfDevice.SchemaResourcesDTO;
  86 import org.opendaylight.netconf.client.mdsal.api.RemoteDeviceId;
  87 import org.opendaylight.netconf.client.mdsal.api.RemoteDeviceServices;
  88 import org.opendaylight.netconf.client.mdsal.api.RemoteDeviceServices.Actions;
  89 import org.opendaylight.netconf.client.mdsal.api.RemoteDeviceServices.Rpcs;
  90 import org.opendaylight.netconf.dom.api.NetconfDataTreeService;
  91 import org.opendaylight.netconf.topology.singleton.impl.actors.NetconfNodeActor;
  92 import org.opendaylight.netconf.topology.singleton.impl.utils.ClusteringActionException;
  93 import org.opendaylight.netconf.topology.singleton.impl.utils.ClusteringRpcException;
  94 import org.opendaylight.netconf.topology.singleton.impl.utils.NetconfTopologySetup;
  95 import org.opendaylight.netconf.topology.singleton.messages.AskForMasterMountPoint;
  96 import org.opendaylight.netconf.topology.singleton.messages.CreateInitialMasterActorData;
  97 import org.opendaylight.netconf.topology.singleton.messages.MasterActorDataInitialized;
  98 import org.opendaylight.netconf.topology.singleton.messages.NotMasterException;
  99 import org.opendaylight.netconf.topology.singleton.messages.RefreshSetupMasterActorData;
 100 import org.opendaylight.netconf.topology.singleton.messages.RegisterMountPoint;
 101 import org.opendaylight.netconf.topology.singleton.messages.UnregisterSlaveMountPoint;
 102 import org.opendaylight.yangtools.concepts.ObjectRegistration;
 103 import org.opendaylight.yangtools.concepts.Registration;
 104 import org.opendaylight.yangtools.util.concurrent.FluentFutures;
 105 import org.opendaylight.yangtools.yang.common.ErrorType;
 106 import org.opendaylight.yangtools.yang.common.QName;
 107 import org.opendaylight.yangtools.yang.common.RpcError;
 108 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
 109 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
 110 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifier;
 111 import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode;
 112 import org.opendaylight.yangtools.yang.data.impl.schema.Builders;
 113 import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
 114 import org.opendaylight.yangtools.yang.model.api.EffectiveModelContext;
 115 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
 116 import org.opendaylight.yangtools.yang.model.api.source.SourceIdentifier;
 117 import org.opendaylight.yangtools.yang.model.api.source.YangTextSource;
 118 import org.opendaylight.yangtools.yang.model.api.stmt.SchemaNodeIdentifier.Absolute;
 119 import org.opendaylight.yangtools.yang.model.repo.api.EffectiveModelContextFactory;
 120 import org.opendaylight.yangtools.yang.model.repo.api.MissingSchemaSourceException;
 121 import org.opendaylight.yangtools.yang.model.repo.api.SchemaRepository;
 122 import org.opendaylight.yangtools.yang.model.repo.api.SchemaResolutionException;
 123 import org.opendaylight.yangtools.yang.model.repo.spi.PotentialSchemaSource;
 124 import org.opendaylight.yangtools.yang.model.repo.spi.SchemaSourceRegistry;
 125 import org.opendaylight.yangtools.yang.model.spi.source.DelegatedYangTextSource;
 126 import org.opendaylight.yangtools.yang.parser.repo.SharedSchemaRepository;
 127 import org.opendaylight.yangtools.yang.parser.rfc7950.repo.TextToIRTransformer;
 128 import scala.concurrent.Await;
 129 import scala.concurrent.Future;
 130
 131 @RunWith(MockitoJUnitRunner.StrictStubs.class)
 132 public class NetconfNodeActorTest extends AbstractBaseSchemasTest {
 133
 134     private static final Timeout TIMEOUT = Timeout.create(Duration.ofSeconds(5));
 135     private static final SourceIdentifier SOURCE_IDENTIFIER1 = new SourceIdentifier("yang1");
 136     private static final SourceIdentifier SOURCE_IDENTIFIER2 = new SourceIdentifier("yang2");
 137
 138     private ActorSystem system = ActorSystem.create();
 139     private final TestKit testKit = new TestKit(system);
 140
 141     private ActorRef masterRef;
 142     private RemoteDeviceId remoteDeviceId;
 143     private final SharedSchemaRepository masterSchemaRepository = new SharedSchemaRepository("master");
 144
 145     @Mock
 146     private Rpcs.Normalized mockRpc;
 147     @Mock
 148     private DOMRpcService mockDOMRpcService;
 149     @Mock
 150     private Actions.Normalized mockDOMActionService;
 151     @Mock
 152     private DOMMountPointService mockMountPointService;
 153     @Mock
 154     private DOMMountPointService.DOMMountPointBuilder mockMountPointBuilder;
 155     @Mock
 156     private ObjectRegistration<DOMMountPoint> mockMountPointReg;
 157     @Mock
 158     private DOMDataBroker mockDOMDataBroker;
 159     @Mock
 160     private NetconfDataTreeService netconfService;
 161     @Mock
 162     private Registration mockSchemaSourceReg1;
 163     @Mock
 164     private Registration mockSchemaSourceReg2;
 165     @Mock
 166     private SchemaSourceRegistry mockRegistry;
 167     @Mock
 168     private EffectiveModelContextFactory mockSchemaContextFactory;
 169     @Mock
 170     private SchemaRepository mockSchemaRepository;
 171     @Mock
 172     private EffectiveModelContext mockSchemaContext;
 173     @Mock
 174     private SchemaResourcesDTO schemaResourceDTO;
 175
 176     @Before
 177     public void setup() {
 178         remoteDeviceId = new RemoteDeviceId("netconf-topology",
 179                 new InetSocketAddress(InetAddresses.forString("127.0.0.1"), 9999));
 180
 181         masterSchemaRepository.registerSchemaSourceListener(
 182                 TextToIRTransformer.create(masterSchemaRepository, masterSchemaRepository));
 183
 184         doReturn(masterSchemaRepository).when(schemaResourceDTO).getSchemaRepository();
 185         doReturn(mockRegistry).when(schemaResourceDTO).getSchemaRegistry();
 186         final NetconfTopologySetup setup = NetconfTopologySetup.builder()
 187             .setActorSystem(system)
 188             .setIdleTimeout(Duration.ofSeconds(1))
 189             .setSchemaResourceDTO(schemaResourceDTO)
 190             .setBaseSchemas(BASE_SCHEMAS)
 191             .build();
 192
 193         final Props props = NetconfNodeActor.props(setup, remoteDeviceId, TIMEOUT, mockMountPointService);
 194
 195         masterRef = TestActorRef.create(system, props, "master_messages");
 196
 197         resetMountPointMocks();
 198
 199         doReturn(mockMountPointBuilder).when(mockMountPointService).createMountPoint(any());
 200
 201         doReturn(mockSchemaSourceReg1).when(mockRegistry).registerSchemaSource(any(), withSourceId(SOURCE_IDENTIFIER1));
 202         doReturn(mockSchemaSourceReg2).when(mockRegistry).registerSchemaSource(any(), withSourceId(SOURCE_IDENTIFIER2));
 203
 204         doReturn(mockSchemaContextFactory).when(mockSchemaRepository).createEffectiveModelContextFactory();
 205
 206         doReturn(mockDOMRpcService).when(mockRpc).domRpcService();
 207
 208     }
 209
 210     @After
 211     public void teardown() {
 212         TestKit.shutdownActorSystem(system, true);
 213         system = null;
 214     }
 215
 216     @Test
 217     public void testInitializeAndRefreshMasterData() {
 218
 219         // Test CreateInitialMasterActorData.
 220
 221         initializeMaster(new ArrayList<>());
 222
 223         // Test RefreshSetupMasterActorData.
 224
 225         final RemoteDeviceId newRemoteDeviceId = new RemoteDeviceId("netconf-topology2",
 226                 new InetSocketAddress(InetAddresses.forString("127.0.0.2"), 9999));
 227
 228         final NetconfTopologySetup newSetup = NetconfTopologySetup.builder()
 229             .setBaseSchemas(BASE_SCHEMAS)
 230             .setSchemaResourceDTO(schemaResourceDTO)
 231             .setActorSystem(system)
 232             .build();
 233
 234         masterRef.tell(new RefreshSetupMasterActorData(newSetup, newRemoteDeviceId), testKit.getRef());
 235
 236         testKit.expectMsgClass(MasterActorDataInitialized.class);
 237     }
 238
 239     @Test
 240     public void testAskForMasterMountPoint() {
 241
 242         // Test with master not setup yet.
 243
 244         final TestKit kit = new TestKit(system);
 245
 246         masterRef.tell(new AskForMasterMountPoint(kit.getRef()), kit.getRef());
 247
 248         final Failure failure = kit.expectMsgClass(Failure.class);
 249         assertTrue(failure.cause() instanceof NotMasterException);
 250
 251         // Now initialize - master should send the RegisterMountPoint message.
 252
 253         List<SourceIdentifier> sourceIdentifiers = List.of(new SourceIdentifier("testID"));
 254         initializeMaster(sourceIdentifiers);
 255
 256         masterRef.tell(new AskForMasterMountPoint(kit.getRef()), kit.getRef());
 257
 258         final RegisterMountPoint registerMountPoint = kit.expectMsgClass(RegisterMountPoint.class);
 259
 260         assertEquals(sourceIdentifiers, registerMountPoint.getSourceIndentifiers());
 261     }
 262
 263     @Test
 264     public void testRegisterAndUnregisterMountPoint() throws Exception {
 265
 266         ActorRef slaveRef = registerSlaveMountPoint();
 267
 268         // Unregister
 269
 270         slaveRef.tell(new UnregisterSlaveMountPoint(), testKit.getRef());
 271
 272         verify(mockMountPointReg, timeout(5000)).close();
 273         verify(mockSchemaSourceReg1, timeout(1000)).close();
 274         verify(mockSchemaSourceReg2, timeout(1000)).close();
 275
 276         // Test registration with another interleaved registration that completes while the first registration
 277         // is resolving the schema context.
 278
 279         reset(mockSchemaSourceReg1, mockRegistry, mockSchemaRepository);
 280         resetMountPointMocks();
 281
 282         doReturn(mockSchemaSourceReg1).when(mockRegistry).registerSchemaSource(any(), withSourceId(SOURCE_IDENTIFIER1));
 283
 284         final var newMockSchemaSourceReg = mock(Registration.class);
 285
 286         final EffectiveModelContextFactory newMockSchemaContextFactory = mock(EffectiveModelContextFactory.class);
 287         doReturn(Futures.immediateFuture(mockSchemaContext))
 288                 .when(newMockSchemaContextFactory).createEffectiveModelContext(anyCollection());
 289
 290         doAnswer(unused -> {
 291             SettableFuture<SchemaContext> future = SettableFuture.create();
 292             new Thread(() -> {
 293                 doReturn(newMockSchemaSourceReg).when(mockRegistry).registerSchemaSource(any(),
 294                         withSourceId(SOURCE_IDENTIFIER1));
 295
 296                 doReturn(newMockSchemaContextFactory).when(mockSchemaRepository)
 297                         .createEffectiveModelContextFactory();
 298
 299                 slaveRef.tell(new RegisterMountPoint(ImmutableList.of(SOURCE_IDENTIFIER1), masterRef),
 300                         testKit.getRef());
 301
 302                 future.set(mockSchemaContext);
 303             }).start();
 304             return future;
 305         }).when(mockSchemaContextFactory).createEffectiveModelContext(anyCollection());
 306
 307         doReturn(mockSchemaContextFactory).when(mockSchemaRepository)
 308                 .createEffectiveModelContextFactory();
 309
 310         slaveRef.tell(new RegisterMountPoint(ImmutableList.of(SOURCE_IDENTIFIER1), masterRef), testKit.getRef());
 311
 312         verify(mockMountPointBuilder, timeout(5000)).register();
 313         verify(mockMountPointBuilder, after(500)).addService(eq(DOMDataBroker.class), any());
 314         verify(mockMountPointBuilder).addService(eq(DOMRpcService.class), any());
 315         verify(mockMountPointBuilder).addService(eq(DOMActionService.class), any());
 316         verify(mockMountPointBuilder).addService(eq(DOMNotificationService.class), any());
 317         verify(mockMountPointBuilder).addService(eq(DOMSchemaService.class), any());
 318         verify(mockSchemaSourceReg1).close();
 319         verify(mockRegistry, times(2)).registerSchemaSource(any(), withSourceId(SOURCE_IDENTIFIER1));
 320         verify(mockSchemaRepository, times(2)).createEffectiveModelContextFactory();
 321         verifyNoMoreInteractions(mockMountPointBuilder, newMockSchemaSourceReg);
 322
 323         // Stop the slave actor and verify schema source registrations are closed.
 324
 325         final Future<Boolean> stopFuture = Patterns.gracefulStop(slaveRef, TIMEOUT.duration());
 326         Await.result(stopFuture, TIMEOUT.duration());
 327
 328         verify(mockMountPointReg).close();
 329         verify(newMockSchemaSourceReg).close();
 330     }
 331
 332     @SuppressWarnings("unchecked")
 333     @Test
 334     public void testRegisterMountPointWithSchemaFailures() throws Exception {
 335         SchemaResourcesDTO schemaResourceDTO2 = mock(SchemaResourcesDTO.class);
 336         doReturn(mockRegistry).when(schemaResourceDTO2).getSchemaRegistry();
 337         doReturn(mockSchemaRepository).when(schemaResourceDTO2).getSchemaRepository();
 338         final NetconfTopologySetup setup = NetconfTopologySetup.builder()
 339                 .setSchemaResourceDTO(schemaResourceDTO2)
 340                 .setBaseSchemas(BASE_SCHEMAS)
 341                 .setActorSystem(system)
 342                 .build();
 343
 344         final ActorRef slaveRef = system.actorOf(NetconfNodeActor.props(setup, remoteDeviceId, TIMEOUT,
 345                 mockMountPointService));
 346
 347         // Test unrecoverable failure.
 348
 349         doReturn(Futures.immediateFailedFuture(
 350             new SchemaResolutionException("mock", new SourceIdentifier("foo"), null)))
 351                 .when(mockSchemaContextFactory).createEffectiveModelContext(anyCollection());
 352
 353         slaveRef.tell(new RegisterMountPoint(ImmutableList.of(SOURCE_IDENTIFIER1, SOURCE_IDENTIFIER2),
 354                 masterRef), testKit.getRef());
 355
 356         testKit.expectMsgClass(Success.class);
 357
 358         verify(mockRegistry, timeout(5000)).registerSchemaSource(any(), withSourceId(SOURCE_IDENTIFIER1));
 359         verify(mockRegistry, timeout(5000)).registerSchemaSource(any(), withSourceId(SOURCE_IDENTIFIER2));
 360
 361         verify(mockMountPointBuilder, after(1000).never()).register();
 362         verify(mockSchemaSourceReg1, timeout(1000)).close();
 363         verify(mockSchemaSourceReg2, timeout(1000)).close();
 364
 365         // Test recoverable AskTimeoutException - schema context resolution should be retried.
 366
 367         reset(mockSchemaSourceReg1, mockSchemaSourceReg2);
 368
 369         doReturn(Futures.immediateFailedFuture(new SchemaResolutionException("mock", new SourceIdentifier("foo"),
 370                 new AskTimeoutException("timeout"))))
 371             .doReturn(Futures.immediateFuture(mockSchemaContext))
 372             .when(mockSchemaContextFactory).createEffectiveModelContext(anyCollection());
 373
 374         slaveRef.tell(new RegisterMountPoint(ImmutableList.of(SOURCE_IDENTIFIER1, SOURCE_IDENTIFIER2),
 375                 masterRef), testKit.getRef());
 376
 377         testKit.expectMsgClass(Success.class);
 378
 379         verify(mockMountPointBuilder, timeout(5000)).register();
 380         verifyNoMoreInteractions(mockSchemaSourceReg1, mockSchemaSourceReg2);
 381
 382         // Test AskTimeoutException with an interleaved successful registration. The first schema context resolution
 383         // attempt should not be retried.
 384
 385         reset(mockSchemaSourceReg1, mockSchemaSourceReg2, mockSchemaRepository, mockSchemaContextFactory);
 386         resetMountPointMocks();
 387
 388         final EffectiveModelContextFactory mockSchemaContextFactorySuccess = mock(EffectiveModelContextFactory.class);
 389         doReturn(Futures.immediateFuture(mockSchemaContext))
 390                 .when(mockSchemaContextFactorySuccess).createEffectiveModelContext(anyCollection());
 391
 392         doAnswer(unused -> {
 393             SettableFuture<SchemaContext> future = SettableFuture.create();
 394             new Thread(() -> {
 395                 doReturn(mockSchemaContextFactorySuccess).when(mockSchemaRepository)
 396                     .createEffectiveModelContextFactory();
 397
 398                 slaveRef.tell(new RegisterMountPoint(ImmutableList.of(SOURCE_IDENTIFIER1, SOURCE_IDENTIFIER2),
 399                         masterRef), testKit.getRef());
 400
 401                 future.setException(new SchemaResolutionException("mock", new SourceIdentifier("foo"),
 402                     new AskTimeoutException("timeout")));
 403             }).start();
 404             return future;
 405         }).when(mockSchemaContextFactory).createEffectiveModelContext(anyCollection());
 406
 407         doReturn(mockSchemaContextFactory).when(mockSchemaRepository).createEffectiveModelContextFactory();
 408
 409         slaveRef.tell(new RegisterMountPoint(ImmutableList.of(SOURCE_IDENTIFIER1, SOURCE_IDENTIFIER2),
 410                 masterRef), testKit.getRef());
 411
 412         verify(mockMountPointBuilder, timeout(5000)).register();
 413         verify(mockSchemaRepository, times(2)).createEffectiveModelContextFactory();
 414     }
 415
 416     @Test(expected = MissingSchemaSourceException.class)
 417     public void testMissingSchemaSourceOnMissingProvider() throws Exception {
 418         final SharedSchemaRepository repository = new SharedSchemaRepository("test");
 419
 420         SchemaResourcesDTO schemaResourceDTO2 = mock(SchemaResourcesDTO.class);
 421         doReturn(repository).when(schemaResourceDTO2).getSchemaRegistry();
 422         doReturn(repository).when(schemaResourceDTO2).getSchemaRepository();
 423         final NetconfTopologySetup setup = NetconfTopologySetup.builder()
 424             .setActorSystem(system)
 425             .setSchemaResourceDTO(schemaResourceDTO2)
 426             .setIdleTimeout(Duration.ofSeconds(1))
 427             .setBaseSchemas(BASE_SCHEMAS)
 428             .build();
 429         final Props props = NetconfNodeActor.props(setup, remoteDeviceId, TIMEOUT, mockMountPointService);
 430         ActorRef actor = TestActorRef.create(system, props, "master_messages_2");
 431
 432         final SourceIdentifier sourceIdentifier = new SourceIdentifier("testID");
 433
 434         final ProxyYangTextSourceProvider proxyYangProvider =
 435                 new ProxyYangTextSourceProvider(actor, system.dispatcher(), TIMEOUT);
 436
 437         final Future<YangTextSchemaSourceSerializationProxy> resolvedSchemaFuture =
 438                 proxyYangProvider.getYangTextSchemaSource(sourceIdentifier);
 439         Await.result(resolvedSchemaFuture, TIMEOUT.duration());
 440     }
 441
 442     @Test
 443     public void testYangTextSchemaSourceRequest() throws Exception {
 444         final SourceIdentifier sourceIdentifier = new SourceIdentifier("testID");
 445
 446         final ProxyYangTextSourceProvider proxyYangProvider =
 447                 new ProxyYangTextSourceProvider(masterRef, system.dispatcher(), TIMEOUT);
 448
 449         final YangTextSource yangTextSchemaSource = new DelegatedYangTextSource(sourceIdentifier,
 450             CharSource.wrap("YANG"));
 451
 452         // Test success.
 453
 454         final var schemaSourceReg = masterSchemaRepository.registerSchemaSource(
 455             id -> Futures.immediateFuture(yangTextSchemaSource),
 456             PotentialSchemaSource.create(sourceIdentifier, YangTextSource.class, 1));
 457
 458         final Future<YangTextSchemaSourceSerializationProxy> resolvedSchemaFuture =
 459                 proxyYangProvider.getYangTextSchemaSource(sourceIdentifier);
 460
 461         final YangTextSchemaSourceSerializationProxy success = Await.result(resolvedSchemaFuture, TIMEOUT.duration());
 462
 463         assertEquals(sourceIdentifier, success.getRepresentation().sourceId());
 464         assertEquals("YANG", success.getRepresentation().read());
 465
 466         // Test missing source failure.
 467
 468         schemaSourceReg.close();
 469
 470         final MissingSchemaSourceException ex = assertThrows(MissingSchemaSourceException.class,
 471             () -> {
 472                 final Future<YangTextSchemaSourceSerializationProxy> failedSchemaFuture =
 473                         proxyYangProvider.getYangTextSchemaSource(sourceIdentifier);
 474                 Await.result(failedSchemaFuture, TIMEOUT.duration());
 475             });
 476         assertThat(ex.getMessage(), startsWith("No providers registered for source"));
 477         assertThat(ex.getMessage(), containsString(sourceIdentifier.toString()));
 478     }
 479
 480     @Test
 481     public void testSlaveInvokeRpc() throws Exception {
 482
 483         initializeMaster(List.of(new SourceIdentifier("testID")));
 484         registerSlaveMountPoint();
 485
 486         ArgumentCaptor<DOMRpcService> domRPCServiceCaptor = ArgumentCaptor.forClass(DOMRpcService.class);
 487         verify(mockMountPointBuilder).addService(eq(DOMRpcService.class), domRPCServiceCaptor.capture());
 488
 489         final DOMRpcService slaveDomRPCService = domRPCServiceCaptor.getValue();
 490         assertTrue(slaveDomRPCService instanceof ProxyDOMRpcService);
 491
 492         final QName testQName = QName.create("", "TestQname");
 493         final ContainerNode outputNode = Builders.containerBuilder()
 494                 .withNodeIdentifier(new NodeIdentifier(testQName))
 495                 .withChild(ImmutableNodes.leafNode(testQName, "foo")).build();
 496         final RpcError rpcError = RpcResultBuilder.newError(ErrorType.RPC, null, "Rpc invocation failed.");
 497
 498         // RPC with no response output.
 499
 500         doReturn(FluentFutures.immediateNullFluentFuture()).when(mockDOMRpcService).invokeRpc(any(), any());
 501
 502         DOMRpcResult result = slaveDomRPCService.invokeRpc(testQName, outputNode).get(2, TimeUnit.SECONDS);
 503
 504         assertEquals(null, result);
 505
 506         // RPC with response output.
 507
 508         doReturn(FluentFutures.immediateFluentFuture(new DefaultDOMRpcResult(outputNode)))
 509                 .when(mockDOMRpcService).invokeRpc(any(), any());
 510
 511         result = slaveDomRPCService.invokeRpc(testQName, outputNode).get(2, TimeUnit.SECONDS);
 512
 513         assertEquals(outputNode, result.value());
 514         assertTrue(result.errors().isEmpty());
 515
 516         // RPC with response error.
 517
 518         doReturn(FluentFutures.immediateFluentFuture(new DefaultDOMRpcResult(rpcError)))
 519                 .when(mockDOMRpcService).invokeRpc(any(), any());
 520
 521         result = slaveDomRPCService.invokeRpc(testQName, outputNode).get(2, TimeUnit.SECONDS);
 522
 523         assertNull(result.value());
 524         assertEquals(rpcError, result.errors().iterator().next());
 525
 526         // RPC with response output and error.
 527
 528         doReturn(FluentFutures.immediateFluentFuture(new DefaultDOMRpcResult(outputNode, rpcError)))
 529                 .when(mockDOMRpcService).invokeRpc(any(), any());
 530
 531         final DOMRpcResult resultOutputError =
 532                 slaveDomRPCService.invokeRpc(testQName, outputNode).get(2, TimeUnit.SECONDS);
 533
 534         assertEquals(outputNode, resultOutputError.value());
 535         assertEquals(rpcError, resultOutputError.errors().iterator().next());
 536
 537         // RPC failure.
 538         doReturn(FluentFutures.immediateFailedFluentFuture(new ClusteringRpcException("mock")))
 539             .when(mockDOMRpcService).invokeRpc(any(), any());
 540         final ListenableFuture<? extends DOMRpcResult> future = slaveDomRPCService.invokeRpc(testQName, outputNode);
 541
 542         final ExecutionException e = assertThrows(ExecutionException.class, () -> future.get(2, TimeUnit.SECONDS));
 543         final Throwable cause = e.getCause();
 544         assertThat(cause, instanceOf(DOMRpcException.class));
 545         assertEquals("mock", cause.getMessage());
 546     }
 547
 548     @Test
 549     public void testSlaveInvokeAction() throws Exception {
 550         initializeMaster(List.of(new SourceIdentifier("testActionID")));
 551         registerSlaveMountPoint();
 552
 553         ArgumentCaptor<DOMActionService> domActionServiceCaptor = ArgumentCaptor.forClass(DOMActionService.class);
 554         verify(mockMountPointBuilder).addService(eq(DOMActionService.class), domActionServiceCaptor.capture());
 555
 556         final DOMActionService slaveDomActionService = domActionServiceCaptor.getValue();
 557         assertTrue(slaveDomActionService instanceof ProxyDOMActionService);
 558
 559         final QName testQName = QName.create("test", "2019-08-16", "TestActionQname");
 560         final Absolute schemaPath = Absolute.of(testQName);
 561
 562         final YangInstanceIdentifier yangIIdPath = YangInstanceIdentifier.of(testQName);
 563
 564         final DOMDataTreeIdentifier domDataTreeIdentifier = new DOMDataTreeIdentifier(LogicalDatastoreType.OPERATIONAL,
 565             yangIIdPath);
 566
 567         final ContainerNode outputNode = Builders.containerBuilder()
 568             .withNodeIdentifier(new NodeIdentifier(testQName))
 569             .withChild(ImmutableNodes.leafNode(testQName, "foo")).build();
 570
 571         // Action with no response output.
 572         doReturn(FluentFutures.immediateNullFluentFuture()).when(mockDOMActionService)
 573             .invokeAction(any(), any(), any());
 574         DOMActionResult result = slaveDomActionService.invokeAction(schemaPath, domDataTreeIdentifier, outputNode)
 575             .get(2, TimeUnit.SECONDS);
 576         assertEquals(null, result);
 577
 578         // Action with response output.
 579         doReturn(FluentFutures.immediateFluentFuture(new SimpleDOMActionResult(outputNode))).when(mockDOMActionService)
 580             .invokeAction(any(), any(), any());
 581         result = slaveDomActionService.invokeAction(schemaPath, domDataTreeIdentifier, outputNode)
 582             .get(2, TimeUnit.SECONDS);
 583
 584         assertEquals(Optional.of(outputNode), result.getOutput());
 585         assertTrue(result.getErrors().isEmpty());
 586
 587         // Action failure.
 588         doReturn(FluentFutures.immediateFailedFluentFuture(new ClusteringActionException("mock")))
 589             .when(mockDOMActionService).invokeAction(any(), any(), any());
 590         final ListenableFuture<? extends DOMActionResult> future = slaveDomActionService.invokeAction(schemaPath,
 591             domDataTreeIdentifier, outputNode);
 592
 593         final ExecutionException e = assertThrows(ExecutionException.class, () -> future.get(2, TimeUnit.SECONDS));
 594         final Throwable cause = e.getCause();
 595         assertThat(cause, instanceOf(DOMActionException.class));
 596         assertEquals("mock", cause.getMessage());
 597     }
 598
 599     @Test
 600     public void testSlaveNewTransactionRequests() {
 601         doReturn(mock(DOMDataTreeReadTransaction.class)).when(mockDOMDataBroker).newReadOnlyTransaction();
 602         doReturn(mock(DOMDataTreeReadWriteTransaction.class)).when(mockDOMDataBroker).newReadWriteTransaction();
 603         doReturn(mock(DOMDataTreeWriteTransaction.class)).when(mockDOMDataBroker).newWriteOnlyTransaction();
 604
 605         initializeMaster(List.of());
 606         registerSlaveMountPoint();
 607
 608         final var domDataBrokerCaptor = ArgumentCaptor.forClass(DOMDataBroker.class);
 609         verify(mockMountPointBuilder).addService(eq(DOMDataBroker.class), domDataBrokerCaptor.capture());
 610
 611         final DOMDataBroker slaveDOMDataBroker = domDataBrokerCaptor.getValue();
 612         assertTrue(slaveDOMDataBroker instanceof ProxyDOMDataBroker);
 613
 614         slaveDOMDataBroker.newReadOnlyTransaction();
 615         verify(mockDOMDataBroker).newReadOnlyTransaction();
 616
 617         slaveDOMDataBroker.newReadWriteTransaction();
 618         verify(mockDOMDataBroker).newReadWriteTransaction();
 619
 620         slaveDOMDataBroker.newWriteOnlyTransaction();
 621         verify(mockDOMDataBroker).newWriteOnlyTransaction();
 622     }
 623
 624     @Test
 625     public void testSlaveNewNetconfDataTreeServiceRequest() {
 626         initializeMaster(List.of());
 627         registerSlaveMountPoint();
 628
 629         ArgumentCaptor<NetconfDataTreeService> netconfCaptor = ArgumentCaptor.forClass(NetconfDataTreeService.class);
 630         verify(mockMountPointBuilder).addService(eq(NetconfDataTreeService.class), netconfCaptor.capture());
 631
 632         final NetconfDataTreeService slaveNetconfService = netconfCaptor.getValue();
 633         assertTrue(slaveNetconfService instanceof ProxyNetconfDataTreeService);
 634
 635         final YangInstanceIdentifier PATH = YangInstanceIdentifier.of();
 636         final LogicalDatastoreType STORE = LogicalDatastoreType.CONFIGURATION;
 637         final ContainerNode NODE = Builders.containerBuilder()
 638             .withNodeIdentifier(new NodeIdentifier(QName.create("", "cont")))
 639             .build();
 640
 641         final FluentFuture<Optional<Object>> result = immediateFluentFuture(Optional.of(NODE));
 642         doReturn(result).when(netconfService).get(PATH);
 643         doReturn(result).when(netconfService).getConfig(PATH);
 644         doReturn(emptyFluentFuture()).when(netconfService).commit();
 645
 646         slaveNetconfService.get(PATH);
 647         slaveNetconfService.getConfig(PATH);
 648         slaveNetconfService.lock();
 649         slaveNetconfService.merge(STORE, PATH, NODE, Optional.empty());
 650         slaveNetconfService.replace(STORE, PATH, NODE, Optional.empty());
 651         slaveNetconfService.create(STORE, PATH, NODE, Optional.empty());
 652         slaveNetconfService.delete(STORE, PATH);
 653         slaveNetconfService.remove(STORE, PATH);
 654         slaveNetconfService.discardChanges();
 655         slaveNetconfService.commit();
 656
 657         verify(netconfService, timeout(1000)).get(PATH);
 658         verify(netconfService, timeout(1000)).getConfig(PATH);
 659         verify(netconfService, timeout(1000)).lock();
 660         verify(netconfService, timeout(1000)).merge(STORE, PATH, NODE, Optional.empty());
 661         verify(netconfService, timeout(1000)).replace(STORE, PATH, NODE, Optional.empty());
 662         verify(netconfService, timeout(1000)).create(STORE, PATH, NODE, Optional.empty());
 663         verify(netconfService, timeout(1000)).delete(STORE, PATH);
 664         verify(netconfService, timeout(1000)).remove(STORE, PATH);
 665         verify(netconfService, timeout(1000)).discardChanges();
 666         verify(netconfService, timeout(1000)).commit();
 667     }
 668
 669     private ActorRef registerSlaveMountPoint() {
 670         SchemaResourcesDTO schemaResourceDTO2 = mock(SchemaResourcesDTO.class);
 671         doReturn(mockRegistry).when(schemaResourceDTO2).getSchemaRegistry();
 672         doReturn(mockSchemaRepository).when(schemaResourceDTO2).getSchemaRepository();
 673         final ActorRef slaveRef = system.actorOf(NetconfNodeActor.props(NetconfTopologySetup.builder()
 674                 .setSchemaResourceDTO(schemaResourceDTO2)
 675                 .setActorSystem(system)
 676                 .setBaseSchemas(BASE_SCHEMAS)
 677                 .build(), remoteDeviceId, TIMEOUT, mockMountPointService));
 678
 679         doReturn(Futures.immediateFuture(mockSchemaContext))
 680                 .when(mockSchemaContextFactory).createEffectiveModelContext(anyCollection());
 681
 682         slaveRef.tell(new RegisterMountPoint(ImmutableList.of(SOURCE_IDENTIFIER1, SOURCE_IDENTIFIER2),
 683                 masterRef), testKit.getRef());
 684
 685         verify(mockMountPointBuilder, timeout(5000)).register();
 686         verify(mockMountPointBuilder).addService(eq(DOMSchemaService.class), any());
 687         verify(mockMountPointBuilder).addService(eq(DOMDataBroker.class), any());
 688         verify(mockMountPointBuilder).addService(eq(NetconfDataTreeService.class), any());
 689         verify(mockMountPointBuilder).addService(eq(DOMRpcService.class), any());
 690         verify(mockMountPointBuilder).addService(eq(DOMNotificationService.class), any());
 691
 692         testKit.expectMsgClass(Success.class);
 693         return slaveRef;
 694     }
 695
 696     private void initializeMaster(final List<SourceIdentifier> sourceIdentifiers) {
 697         masterRef.tell(new CreateInitialMasterActorData(mockDOMDataBroker, netconfService, sourceIdentifiers,
 698                 new RemoteDeviceServices(mockRpc, mockDOMActionService)), testKit.getRef());
 699         testKit.expectMsgClass(MasterActorDataInitialized.class);
 700     }
 701
 702     private void resetMountPointMocks() {
 703         reset(mockMountPointReg, mockMountPointBuilder);
 704
 705         doNothing().when(mockMountPointReg).close();
 706
 707         doReturn(mockMountPointBuilder).when(mockMountPointBuilder).addService(any(), any());
 708         doReturn(mockMountPointReg).when(mockMountPointBuilder).register();
 709     }
 710
 711     private static PotentialSchemaSource<?> withSourceId(final SourceIdentifier identifier) {
 712         return argThat(argument -> identifier.equals(argument.getSourceIdentifier()));
 713     }
 714 }