+ @SuppressWarnings("resource")
+ @Test
+ public void testConstructor(){
+ ActorSystem actorSystem = mock(ActorSystem.class);
+
+ new DistributedDataStore(actorSystem, "config",
+ mock(ClusterWrapper.class), mock(Configuration.class),
+ new DatastoreContext());
+
+ verify(actorSystem).actorOf(any(Props.class), eq("shardmanager-config"));
+ }
+
+ @Test
+ public void testRegisterChangeListenerWhenShardIsNotLocal() throws Exception {
+
+ ListenerRegistration registration =
+ distributedDataStore.registerChangeListener(TestModel.TEST_PATH, new AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>>() {
+ @Override
+ public void onDataChanged(AsyncDataChangeEvent<YangInstanceIdentifier, NormalizedNode<?, ?>> change) {
+ throw new UnsupportedOperationException("onDataChanged");
+ }
+ }, AsyncDataBroker.DataChangeScope.BASE);
+
+ // Since we do not expect the shard to be local registration will return a NoOpRegistration
+ assertTrue(registration instanceof NoOpDataChangeListenerRegistration);
+
+ assertNotNull(registration);
+ }
+
+ @Test
+ public void testRegisterChangeListenerWhenShardIsLocal() throws Exception {
+ ActorContext actorContext = mock(ActorContext.class);
+
+ distributedDataStore = new DistributedDataStore(actorContext);
+ distributedDataStore.onGlobalContextUpdated(TestModel.createTestContext());
+
+ Future future = mock(Future.class);
+ when(actorContext.getOperationDuration()).thenReturn(FiniteDuration.apply(5, TimeUnit.SECONDS));
+ when(actorContext.getActorSystem()).thenReturn(getSystem());
+ when(actorContext.findLocalShard(anyString())).thenReturn(Optional.of(doNothingActorRef));
+ when(actorContext
+ .executeOperationAsync(eq(doNothingActorRef), anyObject(), any(Timeout.class))).thenReturn(future);
+
+ ListenerRegistration registration =
+ distributedDataStore.registerChangeListener(TestModel.TEST_PATH,
+ mock(AsyncDataChangeListener.class),
+ AsyncDataBroker.DataChangeScope.BASE);
+
+ assertNotNull(registration);
+
+ assertEquals(DataChangeListenerRegistrationProxy.class, registration.getClass());
+ }
+
+ @Test
+ public void testRegisterChangeListenerWhenSuccessfulReplyReceived() throws Exception {
+ ActorContext actorContext = mock(ActorContext.class);
+
+ distributedDataStore = new DistributedDataStore(actorContext);
+ distributedDataStore.onGlobalContextUpdated(
+ TestModel.createTestContext());
+
+ ExecutionContextExecutor executor = ExecutionContexts.fromExecutor(MoreExecutors.sameThreadExecutor());
+
+ // Make Future successful
+ Future f = Futures.successful(new RegisterChangeListenerReply(doNothingActorRef.path()));
+
+ // Setup the mocks
+ ActorSystem actorSystem = mock(ActorSystem.class);
+ ActorSelection actorSelection = mock(ActorSelection.class);
+
+ when(actorContext.getOperationDuration()).thenReturn(FiniteDuration.apply(5, TimeUnit.SECONDS));
+ when(actorSystem.dispatcher()).thenReturn(executor);
+ when(actorSystem.actorOf(any(Props.class))).thenReturn(doNothingActorRef);
+ when(actorContext.getActorSystem()).thenReturn(actorSystem);
+ when(actorContext.findLocalShard(anyString())).thenReturn(Optional.of(doNothingActorRef));
+ when(actorContext
+ .executeOperationAsync(eq(doNothingActorRef), anyObject(), any(Timeout.class))).thenReturn(f);
+ when(actorContext.actorSelection(any(ActorPath.class))).thenReturn(actorSelection);
+
+ ListenerRegistration registration =
+ distributedDataStore.registerChangeListener(TestModel.TEST_PATH,
+ mock(AsyncDataChangeListener.class),
+ AsyncDataBroker.DataChangeScope.BASE);
+
+ assertNotNull(registration);
+
+ assertEquals(DataChangeListenerRegistrationProxy.class, registration.getClass());
+
+ ActorSelection listenerRegistrationActor =
+ ((DataChangeListenerRegistrationProxy) registration).getListenerRegistrationActor();
+
+ assertNotNull(listenerRegistrationActor);
+
+ assertEquals(actorSelection, listenerRegistrationActor);
+ }
+
+ @Test
+ public void testRegisterChangeListenerWhenSuccessfulReplyFailed() throws Exception {
+ ActorContext actorContext = mock(ActorContext.class);
+
+ distributedDataStore = new DistributedDataStore(actorContext);
+ distributedDataStore.onGlobalContextUpdated(
+ TestModel.createTestContext());
+
+ ExecutionContextExecutor executor = ExecutionContexts.fromExecutor(MoreExecutors.sameThreadExecutor());
+
+ // Make Future fail
+ Future f = Futures.failed(new IllegalArgumentException());
+
+ // Setup the mocks
+ ActorSystem actorSystem = mock(ActorSystem.class);
+ ActorSelection actorSelection = mock(ActorSelection.class);
+
+ when(actorContext.getOperationDuration()).thenReturn(FiniteDuration.apply(5, TimeUnit.SECONDS));
+ when(actorSystem.dispatcher()).thenReturn(executor);
+ when(actorSystem.actorOf(any(Props.class))).thenReturn(doNothingActorRef);
+ when(actorContext.getActorSystem()).thenReturn(actorSystem);
+ when(actorContext.findLocalShard(anyString())).thenReturn(Optional.of(doNothingActorRef));
+ when(actorContext
+ .executeOperationAsync(eq(doNothingActorRef), anyObject(), any(Timeout.class))).thenReturn(f);
+ when(actorContext.actorSelection(any(ActorPath.class))).thenReturn(actorSelection);
+