2 * Copyright (c) 2016 Cisco Systems, Inc. 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
9 package org.opendaylight.netconf.topology.singleton.impl;
11 import static java.nio.charset.StandardCharsets.UTF_8;
12 import static org.junit.Assert.assertEquals;
13 import static org.junit.Assert.assertNull;
14 import static org.junit.Assert.assertTrue;
15 import static org.mockito.Matchers.any;
16 import static org.mockito.Matchers.argThat;
17 import static org.mockito.Matchers.eq;
18 import static org.mockito.Mockito.after;
19 import static org.mockito.Mockito.doAnswer;
20 import static org.mockito.Mockito.doNothing;
21 import static org.mockito.Mockito.doReturn;
22 import static org.mockito.Mockito.mock;
23 import static org.mockito.Mockito.reset;
24 import static org.mockito.Mockito.timeout;
25 import static org.mockito.Mockito.times;
26 import static org.mockito.Mockito.verify;
27 import static org.mockito.Mockito.verifyNoMoreInteractions;
28 import static org.mockito.MockitoAnnotations.initMocks;
30 import akka.actor.ActorRef;
31 import akka.actor.ActorSystem;
32 import akka.actor.Props;
33 import akka.actor.Status.Failure;
34 import akka.actor.Status.Success;
35 import akka.pattern.AskTimeoutException;
36 import akka.pattern.Patterns;
37 import akka.testkit.TestActorRef;
38 import akka.testkit.javadsl.TestKit;
39 import akka.util.Timeout;
40 import com.google.common.collect.ImmutableList;
41 import com.google.common.collect.Lists;
42 import com.google.common.io.ByteSource;
43 import com.google.common.net.InetAddresses;
44 import com.google.common.util.concurrent.Futures;
45 import com.google.common.util.concurrent.SettableFuture;
46 import java.io.InputStream;
47 import java.net.InetSocketAddress;
48 import java.util.Collections;
49 import java.util.List;
50 import java.util.Scanner;
51 import java.util.concurrent.ExecutionException;
52 import java.util.concurrent.TimeUnit;
53 import org.junit.After;
54 import org.junit.Before;
55 import org.junit.Rule;
56 import org.junit.Test;
57 import org.junit.rules.ExpectedException;
58 import org.mockito.ArgumentCaptor;
59 import org.mockito.ArgumentMatcher;
60 import org.mockito.Mock;
61 import org.opendaylight.controller.cluster.schema.provider.impl.YangTextSchemaSourceSerializationProxy;
62 import org.opendaylight.controller.md.sal.dom.api.DOMDataBroker;
63 import org.opendaylight.controller.md.sal.dom.api.DOMDataReadOnlyTransaction;
64 import org.opendaylight.controller.md.sal.dom.api.DOMDataReadWriteTransaction;
65 import org.opendaylight.controller.md.sal.dom.api.DOMDataWriteTransaction;
66 import org.opendaylight.controller.md.sal.dom.api.DOMMountPoint;
67 import org.opendaylight.controller.md.sal.dom.api.DOMMountPointService;
68 import org.opendaylight.controller.md.sal.dom.api.DOMNotificationService;
69 import org.opendaylight.controller.md.sal.dom.api.DOMRpcException;
70 import org.opendaylight.controller.md.sal.dom.api.DOMRpcResult;
71 import org.opendaylight.controller.md.sal.dom.api.DOMRpcService;
72 import org.opendaylight.controller.md.sal.dom.spi.DefaultDOMRpcResult;
73 import org.opendaylight.netconf.sal.connect.util.RemoteDeviceId;
74 import org.opendaylight.netconf.topology.singleton.impl.actors.NetconfNodeActor;
75 import org.opendaylight.netconf.topology.singleton.impl.utils.ClusteringRpcException;
76 import org.opendaylight.netconf.topology.singleton.impl.utils.NetconfTopologySetup;
77 import org.opendaylight.netconf.topology.singleton.impl.utils.NetconfTopologySetup.NetconfTopologySetupBuilder;
78 import org.opendaylight.netconf.topology.singleton.messages.AskForMasterMountPoint;
79 import org.opendaylight.netconf.topology.singleton.messages.CreateInitialMasterActorData;
80 import org.opendaylight.netconf.topology.singleton.messages.MasterActorDataInitialized;
81 import org.opendaylight.netconf.topology.singleton.messages.NotMasterException;
82 import org.opendaylight.netconf.topology.singleton.messages.RefreshSetupMasterActorData;
83 import org.opendaylight.netconf.topology.singleton.messages.RegisterMountPoint;
84 import org.opendaylight.netconf.topology.singleton.messages.UnregisterSlaveMountPoint;
85 import org.opendaylight.yangtools.concepts.ObjectRegistration;
86 import org.opendaylight.yangtools.yang.common.QName;
87 import org.opendaylight.yangtools.yang.common.RpcError;
88 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
89 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
90 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
91 import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
92 import org.opendaylight.yangtools.yang.data.impl.schema.builder.impl.ImmutableContainerNodeBuilder;
93 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
94 import org.opendaylight.yangtools.yang.model.api.SchemaPath;
95 import org.opendaylight.yangtools.yang.model.repo.api.MissingSchemaSourceException;
96 import org.opendaylight.yangtools.yang.model.repo.api.RevisionSourceIdentifier;
97 import org.opendaylight.yangtools.yang.model.repo.api.SchemaContextFactory;
98 import org.opendaylight.yangtools.yang.model.repo.api.SchemaRepository;
99 import org.opendaylight.yangtools.yang.model.repo.api.SchemaResolutionException;
100 import org.opendaylight.yangtools.yang.model.repo.api.SchemaSourceFilter;
101 import org.opendaylight.yangtools.yang.model.repo.api.SourceIdentifier;
102 import org.opendaylight.yangtools.yang.model.repo.api.YangTextSchemaSource;
103 import org.opendaylight.yangtools.yang.model.repo.spi.PotentialSchemaSource;
104 import org.opendaylight.yangtools.yang.model.repo.spi.SchemaSourceRegistration;
105 import org.opendaylight.yangtools.yang.model.repo.spi.SchemaSourceRegistry;
106 import org.opendaylight.yangtools.yang.parser.repo.SharedSchemaRepository;
107 import org.opendaylight.yangtools.yang.parser.rfc7950.repo.TextToASTTransformer;
108 import scala.concurrent.Await;
109 import scala.concurrent.Future;
110 import scala.concurrent.duration.Duration;
112 public class NetconfNodeActorTest {
114 private static final Timeout TIMEOUT = new Timeout(Duration.create(5, "seconds"));
115 private static final RevisionSourceIdentifier SOURCE_IDENTIFIER1 = RevisionSourceIdentifier.create("yang1");
116 private static final RevisionSourceIdentifier SOURCE_IDENTIFIER2 = RevisionSourceIdentifier.create("yang2");
118 private ActorSystem system = ActorSystem.create();
119 private final TestKit testKit = new TestKit(system);
122 public final ExpectedException exception = ExpectedException.none();
124 private ActorRef masterRef;
125 private RemoteDeviceId remoteDeviceId;
126 private final SharedSchemaRepository masterSchemaRepository = new SharedSchemaRepository("master");
129 private DOMRpcService mockDOMRpcService;
132 private DOMMountPointService mockMountPointService;
135 private DOMMountPointService.DOMMountPointBuilder mockMountPointBuilder;
138 private ObjectRegistration<DOMMountPoint> mockMountPointReg;
141 private DOMDataBroker mockDOMDataBroker;
144 private SchemaSourceRegistration<?> mockSchemaSourceReg1;
147 private SchemaSourceRegistration<?> mockSchemaSourceReg2;
150 private SchemaSourceRegistry mockRegistry;
153 private SchemaContextFactory mockSchemaContextFactory;
156 private SchemaRepository mockSchemaRepository;
159 private SchemaContext mockSchemaContext;
162 public void setup() {
165 remoteDeviceId = new RemoteDeviceId("netconf-topology",
166 new InetSocketAddress(InetAddresses.forString("127.0.0.1"), 9999));
168 masterSchemaRepository.registerSchemaSourceListener(
169 TextToASTTransformer.create(masterSchemaRepository, masterSchemaRepository));
171 final NetconfTopologySetup setup = NetconfTopologySetupBuilder.create().setActorSystem(system)
172 .setIdleTimeout(Duration.apply(1, TimeUnit.SECONDS)).build();
174 final Props props = NetconfNodeActor.props(setup, remoteDeviceId, masterSchemaRepository,
175 masterSchemaRepository, TIMEOUT, mockMountPointService);
177 masterRef = TestActorRef.create(system, props, "master_messages");
179 resetMountPointMocks();
181 doReturn(mockMountPointBuilder).when(mockMountPointService).createMountPoint(any());
183 doReturn(mockSchemaSourceReg1).when(mockRegistry).registerSchemaSource(any(), withSourceId(SOURCE_IDENTIFIER1));
184 doReturn(mockSchemaSourceReg2).when(mockRegistry).registerSchemaSource(any(), withSourceId(SOURCE_IDENTIFIER2));
186 doReturn(mockSchemaContextFactory).when(mockSchemaRepository)
187 .createSchemaContextFactory(any(SchemaSourceFilter.class));
191 public void teardown() {
192 TestKit.shutdownActorSystem(system, Boolean.TRUE);
197 public void testInitializeAndRefreshMasterData() {
199 // Test CreateInitialMasterActorData.
201 initializeMaster(Lists.newArrayList());
203 // Test RefreshSetupMasterActorData.
205 final RemoteDeviceId newRemoteDeviceId = new RemoteDeviceId("netconf-topology2",
206 new InetSocketAddress(InetAddresses.forString("127.0.0.2"), 9999));
208 final NetconfTopologySetup newSetup = NetconfTopologySetupBuilder.create().setActorSystem(system).build();
210 masterRef.tell(new RefreshSetupMasterActorData(newSetup, newRemoteDeviceId), testKit.getRef());
212 testKit.expectMsgClass(MasterActorDataInitialized.class);
216 public void tesAskForMasterMountPoint() {
218 // Test with master not setup yet.
220 final TestKit kit = new TestKit(system);
222 masterRef.tell(new AskForMasterMountPoint(kit.getRef()), kit.getRef());
224 final Failure failure = kit.expectMsgClass(Failure.class);
225 assertTrue(failure.cause() instanceof NotMasterException);
227 // Now initialize - master should send the RegisterMountPoint message.
229 List<SourceIdentifier> sourceIdentifiers = Lists.newArrayList(RevisionSourceIdentifier.create("testID"));
230 initializeMaster(sourceIdentifiers);
232 masterRef.tell(new AskForMasterMountPoint(kit.getRef()), kit.getRef());
234 final RegisterMountPoint registerMountPoint = kit.expectMsgClass(RegisterMountPoint.class);
236 assertEquals(sourceIdentifiers, registerMountPoint.getSourceIndentifiers());
240 public void testRegisterAndUnregisterMountPoint() throws Exception {
242 ActorRef slaveRef = registerSlaveMountPoint();
246 slaveRef.tell(new UnregisterSlaveMountPoint(), testKit.getRef());
248 verify(mockMountPointReg, timeout(5000)).close();
249 verify(mockSchemaSourceReg1, timeout(1000)).close();
250 verify(mockSchemaSourceReg2, timeout(1000)).close();
252 // Test registration with another interleaved registration that completes while the first registration
253 // is resolving the schema context.
255 reset(mockSchemaSourceReg1, mockRegistry, mockSchemaRepository);
256 resetMountPointMocks();
258 doReturn(mockSchemaSourceReg1).when(mockRegistry).registerSchemaSource(any(), withSourceId(SOURCE_IDENTIFIER1));
260 doReturn(mockSchemaContextFactory).when(mockSchemaRepository)
261 .createSchemaContextFactory(any(SchemaSourceFilter.class));
263 final SchemaSourceRegistration<?> newMockSchemaSourceReg = mock(SchemaSourceRegistration.class);
265 final SchemaContextFactory newMockSchemaContextFactory = mock(SchemaContextFactory.class);
266 doReturn(Futures.immediateFuture(mockSchemaContext))
267 .when(newMockSchemaContextFactory).createSchemaContext(any());
270 SettableFuture<SchemaContext> future = SettableFuture.create();
272 doReturn(newMockSchemaSourceReg).when(mockRegistry).registerSchemaSource(any(),
273 withSourceId(SOURCE_IDENTIFIER1));
275 doReturn(newMockSchemaContextFactory).when(mockSchemaRepository)
276 .createSchemaContextFactory(any(SchemaSourceFilter.class));
278 slaveRef.tell(new RegisterMountPoint(ImmutableList.of(SOURCE_IDENTIFIER1), masterRef),
281 future.set(mockSchemaContext);
284 }).when(mockSchemaContextFactory).createSchemaContext(any());
286 doReturn(mockSchemaContextFactory).when(mockSchemaRepository)
287 .createSchemaContextFactory(any(SchemaSourceFilter.class));
289 slaveRef.tell(new RegisterMountPoint(ImmutableList.of(SOURCE_IDENTIFIER1), masterRef), testKit.getRef());
291 verify(mockMountPointBuilder, timeout(5000)).register();
292 verify(mockMountPointBuilder, after(500)).addInitialSchemaContext(mockSchemaContext);
293 verify(mockMountPointBuilder).addService(eq(DOMDataBroker.class), any());
294 verify(mockMountPointBuilder).addService(eq(DOMRpcService.class), any());
295 verify(mockMountPointBuilder).addService(eq(DOMNotificationService.class), any());
296 verify(mockSchemaSourceReg1).close();
297 verify(mockRegistry, times(2)).registerSchemaSource(any(), withSourceId(SOURCE_IDENTIFIER1));
298 verify(mockSchemaRepository, times(2)).createSchemaContextFactory(any(SchemaSourceFilter.class));
299 verifyNoMoreInteractions(mockMountPointBuilder, newMockSchemaSourceReg);
301 // Stop the slave actor and verify schema source registrations are closed.
303 final Future<Boolean> stopFuture = Patterns.gracefulStop(slaveRef, TIMEOUT.duration());
304 Await.result(stopFuture, TIMEOUT.duration());
306 verify(mockMountPointReg).close();
307 verify(newMockSchemaSourceReg).close();
310 @SuppressWarnings("unchecked")
312 public void testRegisterMountPointWithSchemaFailures() throws Exception {
313 final NetconfTopologySetup setup = NetconfTopologySetupBuilder.create().setActorSystem(system).build();
315 final ActorRef slaveRef = system.actorOf(NetconfNodeActor.props(setup, remoteDeviceId, mockRegistry,
316 mockSchemaRepository, TIMEOUT, mockMountPointService));
318 // Test unrecoverable failure.
320 doReturn(Futures.immediateFailedFuture(new SchemaResolutionException("mock")))
321 .when(mockSchemaContextFactory).createSchemaContext(any());
323 slaveRef.tell(new RegisterMountPoint(ImmutableList.of(SOURCE_IDENTIFIER1, SOURCE_IDENTIFIER2),
324 masterRef), testKit.getRef());
326 testKit.expectMsgClass(Success.class);
328 verify(mockRegistry, timeout(5000)).registerSchemaSource(any(), withSourceId(SOURCE_IDENTIFIER1));
329 verify(mockRegistry, timeout(5000)).registerSchemaSource(any(), withSourceId(SOURCE_IDENTIFIER2));
331 verify(mockMountPointBuilder, after(1000).never()).register();
332 verify(mockSchemaSourceReg1, timeout(1000)).close();
333 verify(mockSchemaSourceReg2, timeout(1000)).close();
335 // Test recoverable AskTimeoutException - schema context resolution should be retried.
337 reset(mockSchemaSourceReg1, mockSchemaSourceReg2);
339 doReturn(Futures.immediateFailedFuture(new SchemaResolutionException("mock",
340 new AskTimeoutException("timeout"))))
341 .doReturn(Futures.immediateFuture(mockSchemaContext))
342 .when(mockSchemaContextFactory).createSchemaContext(any());
344 slaveRef.tell(new RegisterMountPoint(ImmutableList.of(SOURCE_IDENTIFIER1, SOURCE_IDENTIFIER2),
345 masterRef), testKit.getRef());
347 testKit.expectMsgClass(Success.class);
349 verify(mockMountPointBuilder, timeout(5000)).register();
350 verifyNoMoreInteractions(mockSchemaSourceReg1, mockSchemaSourceReg2);
352 // Test AskTimeoutException with an interleaved successful registration. The first schema context resolution
353 // attempt should not be retried.
355 reset(mockSchemaSourceReg1, mockSchemaSourceReg2, mockSchemaRepository, mockSchemaContextFactory);
356 resetMountPointMocks();
358 final SchemaContextFactory mockSchemaContextFactorySuccess = mock(SchemaContextFactory.class);
359 doReturn(Futures.immediateFuture(mockSchemaContext))
360 .when(mockSchemaContextFactorySuccess).createSchemaContext(any());
363 SettableFuture<SchemaContext> future = SettableFuture.create();
365 doReturn(mockSchemaContextFactorySuccess).when(mockSchemaRepository)
366 .createSchemaContextFactory(any(SchemaSourceFilter.class));
368 slaveRef.tell(new RegisterMountPoint(ImmutableList.of(SOURCE_IDENTIFIER1, SOURCE_IDENTIFIER2),
369 masterRef), testKit.getRef());
371 future.setException(new SchemaResolutionException("mock", new AskTimeoutException("timeout")));
374 }).when(mockSchemaContextFactory).createSchemaContext(any());
376 doReturn(mockSchemaContextFactory).when(mockSchemaRepository)
377 .createSchemaContextFactory(any(SchemaSourceFilter.class));
379 slaveRef.tell(new RegisterMountPoint(ImmutableList.of(SOURCE_IDENTIFIER1, SOURCE_IDENTIFIER2),
380 masterRef), testKit.getRef());
382 verify(mockMountPointBuilder, timeout(5000)).register();
383 verify(mockSchemaRepository, times(2)).createSchemaContextFactory(any(SchemaSourceFilter.class));
387 public void testYangTextSchemaSourceRequest() throws Exception {
388 final SourceIdentifier sourceIdentifier = RevisionSourceIdentifier.create("testID");
390 final ProxyYangTextSourceProvider proxyYangProvider =
391 new ProxyYangTextSourceProvider(masterRef, system.dispatcher(), TIMEOUT);
393 final YangTextSchemaSource yangTextSchemaSource = YangTextSchemaSource.delegateForByteSource(sourceIdentifier,
394 ByteSource.wrap("YANG".getBytes(UTF_8)));
398 final SchemaSourceRegistration<YangTextSchemaSource> schemaSourceReg = masterSchemaRepository
399 .registerSchemaSource(id -> Futures.immediateFuture(yangTextSchemaSource),
400 PotentialSchemaSource.create(sourceIdentifier, YangTextSchemaSource.class, 1));
402 final Future<YangTextSchemaSourceSerializationProxy> resolvedSchemaFuture =
403 proxyYangProvider.getYangTextSchemaSource(sourceIdentifier);
405 final YangTextSchemaSourceSerializationProxy success = Await.result(resolvedSchemaFuture, TIMEOUT.duration());
407 assertEquals(sourceIdentifier, success.getRepresentation().getIdentifier());
408 assertEquals("YANG", convertStreamToString(success.getRepresentation().openStream()));
410 // Test missing source failure.
412 exception.expect(MissingSchemaSourceException.class);
414 schemaSourceReg.close();
416 final Future<YangTextSchemaSourceSerializationProxy> failedSchemaFuture =
417 proxyYangProvider.getYangTextSchemaSource(sourceIdentifier);
419 Await.result(failedSchemaFuture, TIMEOUT.duration());
423 @SuppressWarnings({"checkstyle:AvoidHidingCauseException", "checkstyle:IllegalThrows"})
424 public void testSlaveInvokeRpc() throws Throwable {
426 final List<SourceIdentifier> sourceIdentifiers =
427 Lists.newArrayList(RevisionSourceIdentifier.create("testID"));
429 initializeMaster(sourceIdentifiers);
430 registerSlaveMountPoint();
432 ArgumentCaptor<DOMRpcService> domRPCServiceCaptor = ArgumentCaptor.forClass(DOMRpcService.class);
433 verify(mockMountPointBuilder).addService(eq(DOMRpcService.class), domRPCServiceCaptor.capture());
435 final DOMRpcService slaveDomRPCService = domRPCServiceCaptor.getValue();
436 assertTrue(slaveDomRPCService instanceof ProxyDOMRpcService);
438 final QName testQName = QName.create("", "TestQname");
439 final SchemaPath schemaPath = SchemaPath.create(true, testQName);
440 final NormalizedNode<?, ?> outputNode = ImmutableContainerNodeBuilder.create()
441 .withNodeIdentifier(new YangInstanceIdentifier.NodeIdentifier(testQName))
442 .withChild(ImmutableNodes.leafNode(testQName, "foo")).build();
443 final RpcError rpcError = RpcResultBuilder.newError(RpcError.ErrorType.RPC, null, "Rpc invocation failed.");
445 // RPC with no response output.
447 doReturn(Futures.immediateCheckedFuture(null)).when(mockDOMRpcService).invokeRpc(any(), any());
449 DOMRpcResult result = slaveDomRPCService.invokeRpc(schemaPath, outputNode).get(2, TimeUnit.SECONDS);
451 assertEquals(null, result);
453 // RPC with response output.
455 doReturn(Futures.immediateCheckedFuture(new DefaultDOMRpcResult(outputNode)))
456 .when(mockDOMRpcService).invokeRpc(any(), any());
458 result = slaveDomRPCService.invokeRpc(schemaPath, outputNode).get(2, TimeUnit.SECONDS);
460 assertEquals(outputNode, result.getResult());
461 assertTrue(result.getErrors().isEmpty());
463 // RPC with response error.
465 doReturn(Futures.immediateCheckedFuture(new DefaultDOMRpcResult(rpcError)))
466 .when(mockDOMRpcService).invokeRpc(any(), any());
468 result = slaveDomRPCService.invokeRpc(schemaPath, outputNode).get(2, TimeUnit.SECONDS);
470 assertNull(result.getResult());
471 assertEquals(rpcError, result.getErrors().iterator().next());
473 // RPC with response output and error.
475 doReturn(Futures.immediateCheckedFuture(new DefaultDOMRpcResult(outputNode, rpcError)))
476 .when(mockDOMRpcService).invokeRpc(any(), any());
478 final DOMRpcResult resultOutputError =
479 slaveDomRPCService.invokeRpc(schemaPath, outputNode).get(2, TimeUnit.SECONDS);
481 assertEquals(outputNode, resultOutputError.getResult());
482 assertEquals(rpcError, resultOutputError.getErrors().iterator().next());
486 exception.expect(DOMRpcException.class);
488 doReturn(Futures.immediateFailedCheckedFuture(new ClusteringRpcException("mock")))
489 .when(mockDOMRpcService).invokeRpc(any(), any());
492 slaveDomRPCService.invokeRpc(schemaPath, outputNode).get(2, TimeUnit.SECONDS);
493 } catch (ExecutionException e) {
499 public void testSlaveNewTransactionRequests() {
501 doReturn(mock(DOMDataReadOnlyTransaction.class)).when(mockDOMDataBroker).newReadOnlyTransaction();
502 doReturn(mock(DOMDataReadWriteTransaction.class)).when(mockDOMDataBroker).newReadWriteTransaction();
503 doReturn(mock(DOMDataWriteTransaction.class)).when(mockDOMDataBroker).newWriteOnlyTransaction();
505 initializeMaster(Collections.emptyList());
506 registerSlaveMountPoint();
508 ArgumentCaptor<DOMDataBroker> domDataBrokerCaptor = ArgumentCaptor.forClass(DOMDataBroker.class);
509 verify(mockMountPointBuilder).addService(eq(DOMDataBroker.class), domDataBrokerCaptor.capture());
511 final DOMDataBroker slaveDOMDataBroker = domDataBrokerCaptor.getValue();
512 assertTrue(slaveDOMDataBroker instanceof ProxyDOMDataBroker);
514 slaveDOMDataBroker.newReadOnlyTransaction();
515 verify(mockDOMDataBroker).newReadOnlyTransaction();
517 slaveDOMDataBroker.newReadWriteTransaction();
518 verify(mockDOMDataBroker).newReadWriteTransaction();
520 slaveDOMDataBroker.newWriteOnlyTransaction();
521 verify(mockDOMDataBroker).newWriteOnlyTransaction();
524 private ActorRef registerSlaveMountPoint() {
525 final ActorRef slaveRef = system.actorOf(NetconfNodeActor.props(
526 NetconfTopologySetupBuilder.create().setActorSystem(system).build(), remoteDeviceId, mockRegistry,
527 mockSchemaRepository, TIMEOUT, mockMountPointService));
529 doReturn(Futures.immediateFuture(mockSchemaContext))
530 .when(mockSchemaContextFactory).createSchemaContext(any());
532 slaveRef.tell(new RegisterMountPoint(ImmutableList.of(SOURCE_IDENTIFIER1, SOURCE_IDENTIFIER2),
533 masterRef), testKit.getRef());
535 verify(mockMountPointBuilder, timeout(5000)).register();
536 verify(mockMountPointBuilder).addInitialSchemaContext(mockSchemaContext);
537 verify(mockMountPointBuilder).addService(eq(DOMDataBroker.class), any());
538 verify(mockMountPointBuilder).addService(eq(DOMRpcService.class), any());
539 verify(mockMountPointBuilder).addService(eq(DOMNotificationService.class), any());
541 testKit.expectMsgClass(Success.class);
546 private void initializeMaster(List<SourceIdentifier> sourceIdentifiers) {
547 masterRef.tell(new CreateInitialMasterActorData(mockDOMDataBroker, sourceIdentifiers,
548 mockDOMRpcService), testKit.getRef());
550 testKit.expectMsgClass(MasterActorDataInitialized.class);
553 private void resetMountPointMocks() {
554 reset(mockMountPointReg, mockMountPointBuilder);
556 doNothing().when(mockMountPointReg).close();
558 doReturn(mockMountPointBuilder).when(mockMountPointBuilder).addInitialSchemaContext(any());
559 doReturn(mockMountPointBuilder).when(mockMountPointBuilder).addService(any(), any());
560 doReturn(mockMountPointReg).when(mockMountPointBuilder).register();
563 private static PotentialSchemaSource<?> withSourceId(final SourceIdentifier identifier) {
564 return argThat(new ArgumentMatcher<PotentialSchemaSource<?>>() {
566 public boolean matches(final Object argument) {
567 final PotentialSchemaSource<?> potentialSchemaSource = (PotentialSchemaSource<?>) argument;
568 return identifier.equals(potentialSchemaSource.getSourceIdentifier());
573 private static String convertStreamToString(final InputStream is) {
574 try (Scanner scanner = new Scanner(is)) {
575 return scanner.useDelimiter("\\A").hasNext() ? scanner.next() : "";