1 package org.opendaylight.controller.cluster.datastore;
3 import akka.actor.ActorPath;
4 import akka.actor.ActorRef;
5 import akka.actor.ActorSelection;
6 import akka.actor.ActorSystem;
7 import akka.actor.Props;
8 import akka.dispatch.ExecutionContexts;
9 import akka.dispatch.Futures;
10 import akka.util.Timeout;
11 import com.google.common.util.concurrent.MoreExecutors;
12 import org.junit.After;
13 import org.junit.Before;
14 import org.junit.Test;
15 import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListenerReply;
16 import org.opendaylight.controller.cluster.datastore.shardstrategy.ShardStrategyFactory;
17 import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
18 import org.opendaylight.controller.cluster.datastore.utils.DoNothingActor;
19 import org.opendaylight.controller.cluster.datastore.utils.MockActorContext;
20 import org.opendaylight.controller.cluster.datastore.utils.MockConfiguration;
21 import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
22 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker;
23 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeEvent;
24 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeListener;
25 import org.opendaylight.controller.protobuff.messages.transaction.ShardTransactionMessages.CreateTransactionReply;
26 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadTransaction;
27 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
28 import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionChain;
29 import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction;
30 import org.opendaylight.yangtools.concepts.ListenerRegistration;
31 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
32 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
33 import scala.concurrent.ExecutionContextExecutor;
34 import scala.concurrent.Future;
35 import scala.concurrent.duration.FiniteDuration;
37 import java.util.concurrent.TimeUnit;
39 import static junit.framework.TestCase.assertEquals;
40 import static junit.framework.TestCase.assertNull;
41 import static org.junit.Assert.assertNotNull;
42 import static org.junit.Assert.assertTrue;
43 import static org.mockito.Matchers.any;
44 import static org.mockito.Matchers.anyObject;
45 import static org.mockito.Matchers.anyString;
46 import static org.mockito.Matchers.eq;
47 import static org.mockito.Mockito.mock;
48 import static org.mockito.Mockito.verify;
49 import static org.mockito.Mockito.when;
51 public class DistributedDataStoreTest extends AbstractActorTest{
53 private DistributedDataStore distributedDataStore;
54 private MockActorContext mockActorContext;
55 private ActorRef doNothingActorRef;
58 public void setUp() throws Exception {
59 ShardStrategyFactory.setConfiguration(new MockConfiguration());
60 final Props props = Props.create(DoNothingActor.class);
62 doNothingActorRef = getSystem().actorOf(props);
64 mockActorContext = new MockActorContext(getSystem(), doNothingActorRef);
65 distributedDataStore = new DistributedDataStore(mockActorContext);
66 distributedDataStore.onGlobalContextUpdated(
67 TestModel.createTestContext());
69 // Make CreateTransactionReply as the default response. Will need to be
70 // tuned if a specific test requires some other response
71 mockActorContext.setExecuteShardOperationResponse(
72 CreateTransactionReply.newBuilder()
73 .setTransactionActorPath(doNothingActorRef.path().toString())
74 .setTransactionId("txn-1 ")
79 public void tearDown() throws Exception {
83 @SuppressWarnings("resource")
85 public void testConstructor(){
86 ActorSystem actorSystem = mock(ActorSystem.class);
88 new DistributedDataStore(actorSystem, "config",
89 mock(ClusterWrapper.class), mock(Configuration.class),
90 new DatastoreContext());
92 verify(actorSystem).actorOf(any(Props.class), eq("shardmanager-config"));
96 public void testRegisterChangeListenerWhenShardIsNotLocal() throws Exception {
98 ListenerRegistration registration =
99 distributedDataStore.registerChangeListener(TestModel.TEST_PATH, new AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>>() {
101 public void onDataChanged(AsyncDataChangeEvent<YangInstanceIdentifier, NormalizedNode<?, ?>> change) {
102 throw new UnsupportedOperationException("onDataChanged");
104 }, AsyncDataBroker.DataChangeScope.BASE);
106 // Since we do not expect the shard to be local registration will return a NoOpRegistration
107 assertTrue(registration instanceof NoOpDataChangeListenerRegistration);
109 assertNotNull(registration);
113 public void testRegisterChangeListenerWhenShardIsLocal() throws Exception {
114 ActorContext actorContext = mock(ActorContext.class);
116 distributedDataStore = new DistributedDataStore(actorContext);
117 distributedDataStore.onGlobalContextUpdated(TestModel.createTestContext());
119 Future future = mock(Future.class);
120 when(actorContext.getOperationDuration()).thenReturn(FiniteDuration.apply(5, TimeUnit.SECONDS));
121 when(actorContext.getActorSystem()).thenReturn(getSystem());
123 .executeLocalShardOperationAsync(anyString(), anyObject(), any(Timeout.class))).thenReturn(future);
125 ListenerRegistration registration =
126 distributedDataStore.registerChangeListener(TestModel.TEST_PATH,
127 mock(AsyncDataChangeListener.class),
128 AsyncDataBroker.DataChangeScope.BASE);
130 assertNotNull(registration);
132 assertEquals(DataChangeListenerRegistrationProxy.class, registration.getClass());
136 public void testRegisterChangeListenerWhenSuccessfulReplyReceived() throws Exception {
137 ActorContext actorContext = mock(ActorContext.class);
139 distributedDataStore = new DistributedDataStore(actorContext);
140 distributedDataStore.onGlobalContextUpdated(
141 TestModel.createTestContext());
143 ExecutionContextExecutor executor = ExecutionContexts.fromExecutor(MoreExecutors.sameThreadExecutor());
145 // Make Future successful
146 Future f = Futures.successful(new RegisterChangeListenerReply(doNothingActorRef.path()));
149 ActorSystem actorSystem = mock(ActorSystem.class);
150 ActorSelection actorSelection = mock(ActorSelection.class);
152 when(actorContext.getOperationDuration()).thenReturn(FiniteDuration.apply(5, TimeUnit.SECONDS));
153 when(actorSystem.dispatcher()).thenReturn(executor);
154 when(actorSystem.actorOf(any(Props.class))).thenReturn(doNothingActorRef);
155 when(actorContext.getActorSystem()).thenReturn(actorSystem);
157 .executeLocalShardOperationAsync(anyString(), anyObject(), any(Timeout.class))).thenReturn(f);
158 when(actorContext.actorSelection(any(ActorPath.class))).thenReturn(actorSelection);
160 ListenerRegistration registration =
161 distributedDataStore.registerChangeListener(TestModel.TEST_PATH,
162 mock(AsyncDataChangeListener.class),
163 AsyncDataBroker.DataChangeScope.BASE);
165 assertNotNull(registration);
167 assertEquals(DataChangeListenerRegistrationProxy.class, registration.getClass());
169 ActorSelection listenerRegistrationActor =
170 ((DataChangeListenerRegistrationProxy) registration).getListenerRegistrationActor();
172 assertNotNull(listenerRegistrationActor);
174 assertEquals(actorSelection, listenerRegistrationActor);
178 public void testRegisterChangeListenerWhenSuccessfulReplyFailed() throws Exception {
179 ActorContext actorContext = mock(ActorContext.class);
181 distributedDataStore = new DistributedDataStore(actorContext);
182 distributedDataStore.onGlobalContextUpdated(
183 TestModel.createTestContext());
185 ExecutionContextExecutor executor = ExecutionContexts.fromExecutor(MoreExecutors.sameThreadExecutor());
188 Future f = Futures.failed(new IllegalArgumentException());
191 ActorSystem actorSystem = mock(ActorSystem.class);
192 ActorSelection actorSelection = mock(ActorSelection.class);
194 when(actorContext.getOperationDuration()).thenReturn(FiniteDuration.apply(5, TimeUnit.SECONDS));
195 when(actorSystem.dispatcher()).thenReturn(executor);
196 when(actorSystem.actorOf(any(Props.class))).thenReturn(doNothingActorRef);
197 when(actorContext.getActorSystem()).thenReturn(actorSystem);
199 .executeLocalShardOperationAsync(anyString(), anyObject(), any(Timeout.class))).thenReturn(f);
200 when(actorContext.actorSelection(any(ActorPath.class))).thenReturn(actorSelection);
202 ListenerRegistration registration =
203 distributedDataStore.registerChangeListener(TestModel.TEST_PATH,
204 mock(AsyncDataChangeListener.class),
205 AsyncDataBroker.DataChangeScope.BASE);
207 assertNotNull(registration);
209 assertEquals(DataChangeListenerRegistrationProxy.class, registration.getClass());
211 ActorSelection listenerRegistrationActor =
212 ((DataChangeListenerRegistrationProxy) registration).getListenerRegistrationActor();
214 assertNull(listenerRegistrationActor);
220 public void testCreateTransactionChain() throws Exception {
221 final DOMStoreTransactionChain transactionChain = distributedDataStore.createTransactionChain();
222 assertNotNull(transactionChain);
226 public void testNewReadOnlyTransaction() throws Exception {
227 final DOMStoreReadTransaction transaction = distributedDataStore.newReadOnlyTransaction();
228 assertNotNull(transaction);
232 public void testNewWriteOnlyTransaction() throws Exception {
233 final DOMStoreWriteTransaction transaction = distributedDataStore.newWriteOnlyTransaction();
234 assertNotNull(transaction);
238 public void testNewReadWriteTransaction() throws Exception {
239 final DOMStoreReadWriteTransaction transaction = distributedDataStore.newReadWriteTransaction();
240 assertNotNull(transaction);