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.base.Optional;
12 import com.google.common.util.concurrent.MoreExecutors;
13 import org.junit.After;
14 import org.junit.Before;
15 import org.junit.Test;
16 import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListenerReply;
17 import org.opendaylight.controller.cluster.datastore.shardstrategy.ShardStrategyFactory;
18 import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
19 import org.opendaylight.controller.cluster.datastore.utils.DoNothingActor;
20 import org.opendaylight.controller.cluster.datastore.utils.MockActorContext;
21 import org.opendaylight.controller.cluster.datastore.utils.MockConfiguration;
22 import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
23 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker;
24 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeEvent;
25 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeListener;
26 import org.opendaylight.controller.protobuff.messages.transaction.ShardTransactionMessages.CreateTransactionReply;
27 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadTransaction;
28 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
29 import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionChain;
30 import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction;
31 import org.opendaylight.yangtools.concepts.ListenerRegistration;
32 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
33 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
34 import scala.concurrent.ExecutionContextExecutor;
35 import scala.concurrent.Future;
36 import scala.concurrent.duration.FiniteDuration;
37 import java.util.concurrent.TimeUnit;
38 import static junit.framework.TestCase.assertEquals;
39 import static junit.framework.TestCase.assertNull;
40 import static org.junit.Assert.assertNotNull;
41 import static org.junit.Assert.assertTrue;
42 import static org.mockito.Matchers.any;
43 import static org.mockito.Matchers.anyObject;
44 import static org.mockito.Matchers.anyString;
45 import static org.mockito.Matchers.eq;
46 import static org.mockito.Mockito.mock;
47 import static org.mockito.Mockito.verify;
48 import static org.mockito.Mockito.when;
50 public class DistributedDataStoreTest extends AbstractActorTest{
52 private DistributedDataStore distributedDataStore;
53 private MockActorContext mockActorContext;
54 private ActorRef doNothingActorRef;
57 public void setUp() throws Exception {
58 ShardStrategyFactory.setConfiguration(new MockConfiguration());
59 final Props props = Props.create(DoNothingActor.class);
61 doNothingActorRef = getSystem().actorOf(props);
63 mockActorContext = new MockActorContext(getSystem(), doNothingActorRef);
64 distributedDataStore = new DistributedDataStore(mockActorContext);
65 distributedDataStore.onGlobalContextUpdated(
66 TestModel.createTestContext());
68 // Make CreateTransactionReply as the default response. Will need to be
69 // tuned if a specific test requires some other response
70 mockActorContext.setExecuteShardOperationResponse(
71 CreateTransactionReply.newBuilder()
72 .setTransactionActorPath(doNothingActorRef.path().toString())
73 .setTransactionId("txn-1 ")
78 public void tearDown() throws Exception {
82 @SuppressWarnings("resource")
84 public void testConstructor(){
85 ActorSystem actorSystem = mock(ActorSystem.class);
87 new DistributedDataStore(actorSystem, "config",
88 mock(ClusterWrapper.class), mock(Configuration.class),
89 DatastoreContext.newBuilder().build());
91 verify(actorSystem).actorOf(any(Props.class), eq("shardmanager-config"));
95 public void testRegisterChangeListenerWhenShardIsNotLocal() throws Exception {
97 ListenerRegistration registration =
98 distributedDataStore.registerChangeListener(TestModel.TEST_PATH, new AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>>() {
100 public void onDataChanged(AsyncDataChangeEvent<YangInstanceIdentifier, NormalizedNode<?, ?>> change) {
101 throw new UnsupportedOperationException("onDataChanged");
103 }, AsyncDataBroker.DataChangeScope.BASE);
105 // Since we do not expect the shard to be local registration will return a NoOpRegistration
106 assertTrue(registration instanceof NoOpDataChangeListenerRegistration);
108 assertNotNull(registration);
112 public void testRegisterChangeListenerWhenShardIsLocal() throws Exception {
113 ActorContext actorContext = mock(ActorContext.class);
115 distributedDataStore = new DistributedDataStore(actorContext);
116 distributedDataStore.onGlobalContextUpdated(TestModel.createTestContext());
118 Future future = mock(Future.class);
119 when(actorContext.getOperationDuration()).thenReturn(FiniteDuration.apply(5, TimeUnit.SECONDS));
120 when(actorContext.getActorSystem()).thenReturn(getSystem());
121 when(actorContext.findLocalShard(anyString())).thenReturn(Optional.of(doNothingActorRef));
123 .executeOperationAsync(eq(doNothingActorRef), 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);
156 when(actorContext.findLocalShard(anyString())).thenReturn(Optional.of(doNothingActorRef));
158 .executeOperationAsync(eq(doNothingActorRef), anyObject(), any(Timeout.class))).thenReturn(f);
159 when(actorContext.actorSelection(any(ActorPath.class))).thenReturn(actorSelection);
161 ListenerRegistration registration =
162 distributedDataStore.registerChangeListener(TestModel.TEST_PATH,
163 mock(AsyncDataChangeListener.class),
164 AsyncDataBroker.DataChangeScope.BASE);
166 assertNotNull(registration);
168 assertEquals(DataChangeListenerRegistrationProxy.class, registration.getClass());
170 ActorSelection listenerRegistrationActor =
171 ((DataChangeListenerRegistrationProxy) registration).getListenerRegistrationActor();
173 assertNotNull(listenerRegistrationActor);
175 assertEquals(actorSelection, listenerRegistrationActor);
179 public void testRegisterChangeListenerWhenSuccessfulReplyFailed() throws Exception {
180 ActorContext actorContext = mock(ActorContext.class);
182 distributedDataStore = new DistributedDataStore(actorContext);
183 distributedDataStore.onGlobalContextUpdated(
184 TestModel.createTestContext());
186 ExecutionContextExecutor executor = ExecutionContexts.fromExecutor(MoreExecutors.sameThreadExecutor());
189 Future f = Futures.failed(new IllegalArgumentException());
192 ActorSystem actorSystem = mock(ActorSystem.class);
193 ActorSelection actorSelection = mock(ActorSelection.class);
195 when(actorContext.getOperationDuration()).thenReturn(FiniteDuration.apply(5, TimeUnit.SECONDS));
196 when(actorSystem.dispatcher()).thenReturn(executor);
197 when(actorSystem.actorOf(any(Props.class))).thenReturn(doNothingActorRef);
198 when(actorContext.getActorSystem()).thenReturn(actorSystem);
199 when(actorContext.findLocalShard(anyString())).thenReturn(Optional.of(doNothingActorRef));
201 .executeOperationAsync(eq(doNothingActorRef), anyObject(), any(Timeout.class))).thenReturn(f);
202 when(actorContext.actorSelection(any(ActorPath.class))).thenReturn(actorSelection);
204 ListenerRegistration registration =
205 distributedDataStore.registerChangeListener(TestModel.TEST_PATH,
206 mock(AsyncDataChangeListener.class),
207 AsyncDataBroker.DataChangeScope.BASE);
209 assertNotNull(registration);
211 assertEquals(DataChangeListenerRegistrationProxy.class, registration.getClass());
213 ActorSelection listenerRegistrationActor =
214 ((DataChangeListenerRegistrationProxy) registration).getListenerRegistrationActor();
216 assertNull(listenerRegistrationActor);
222 public void testCreateTransactionChain() throws Exception {
223 final DOMStoreTransactionChain transactionChain = distributedDataStore.createTransactionChain();
224 assertNotNull(transactionChain);
228 public void testNewReadOnlyTransaction() throws Exception {
229 final DOMStoreReadTransaction transaction = distributedDataStore.newReadOnlyTransaction();
230 assertNotNull(transaction);
234 public void testNewWriteOnlyTransaction() throws Exception {
235 final DOMStoreWriteTransaction transaction = distributedDataStore.newWriteOnlyTransaction();
236 assertNotNull(transaction);
240 public void testNewReadWriteTransaction() throws Exception {
241 final DOMStoreReadWriteTransaction transaction = distributedDataStore.newReadWriteTransaction();
242 assertNotNull(transaction);