opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/TransactionProxyTest.java

   1 package org.opendaylight.controller.cluster.datastore;
   2
   3 import akka.actor.ActorRef;
   4 import akka.actor.Props;
   5 import com.google.common.base.Optional;
   6 import com.google.common.util.concurrent.ListenableFuture;
   7 import junit.framework.Assert;
   8 import org.junit.Test;
   9 import org.opendaylight.controller.cluster.datastore.messages.CloseTransaction;
  10 import org.opendaylight.controller.cluster.datastore.messages.DeleteData;
  11 import org.opendaylight.controller.cluster.datastore.messages.MergeData;
  12 import org.opendaylight.controller.cluster.datastore.messages.ReadDataReply;
  13 import org.opendaylight.controller.cluster.datastore.messages.ReadyTransactionReply;
  14 import org.opendaylight.controller.cluster.datastore.messages.WriteData;
  15 import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
  16 import org.opendaylight.controller.cluster.datastore.utils.DoNothingActor;
  17 import org.opendaylight.controller.cluster.datastore.utils.MessageCollectorActor;
  18 import org.opendaylight.controller.cluster.datastore.utils.MockActorContext;
  19 import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
  20 import org.opendaylight.controller.protobuff.messages.transaction.ShardTransactionMessages.CreateTransactionReply;
  21 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
  22 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  23 import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
  24
  25 import java.util.List;
  26 import java.util.concurrent.ExecutorService;
  27 import java.util.concurrent.Executors;
  28
  29 public class TransactionProxyTest extends AbstractActorTest {
  30
  31     private ExecutorService transactionExecutor =
  32         Executors.newSingleThreadExecutor();
  33
  34     @Test
  35     public void testRead() throws Exception {
  36         final Props props = Props.create(DoNothingActor.class);
  37         final ActorRef actorRef = getSystem().actorOf(props);
  38
  39         final MockActorContext actorContext = new MockActorContext(this.getSystem());
  40         actorContext.setExecuteShardOperationResponse(createTransactionReply(actorRef));
  41         actorContext.setExecuteRemoteOperationResponse("message");
  42
  43
  44         TransactionProxy transactionProxy =
  45             new TransactionProxy(actorContext,
  46                 TransactionProxy.TransactionType.READ_ONLY, transactionExecutor, TestModel.createTestContext());
  47
  48
  49         ListenableFuture<Optional<NormalizedNode<?, ?>>> read =
  50             transactionProxy.read(TestModel.TEST_PATH);
  51
  52         Optional<NormalizedNode<?, ?>> normalizedNodeOptional = read.get();
  53
  54         Assert.assertFalse(normalizedNodeOptional.isPresent());
  55
  56         actorContext.setExecuteRemoteOperationResponse(new ReadDataReply(
  57             TestModel.createTestContext(),ImmutableNodes.containerNode(TestModel.TEST_QNAME)).toSerializable());
  58
  59         read = transactionProxy.read(TestModel.TEST_PATH);
  60
  61         normalizedNodeOptional = read.get();
  62
  63         Assert.assertTrue(normalizedNodeOptional.isPresent());
  64     }
  65
  66     @Test
  67     public void testReadWhenANullIsReturned() throws Exception {
  68         final Props props = Props.create(DoNothingActor.class);
  69         final ActorRef actorRef = getSystem().actorOf(props);
  70
  71         final MockActorContext actorContext = new MockActorContext(this.getSystem());
  72         actorContext.setExecuteShardOperationResponse(createTransactionReply(actorRef));
  73         actorContext.setExecuteRemoteOperationResponse("message");
  74
  75         TransactionProxy transactionProxy =
  76             new TransactionProxy(actorContext,
  77                 TransactionProxy.TransactionType.READ_ONLY, transactionExecutor, TestModel.createTestContext());
  78
  79
  80         ListenableFuture<Optional<NormalizedNode<?, ?>>> read =
  81             transactionProxy.read(TestModel.TEST_PATH);
  82
  83         Optional<NormalizedNode<?, ?>> normalizedNodeOptional = read.get();
  84
  85         Assert.assertFalse(normalizedNodeOptional.isPresent());
  86
  87         actorContext.setExecuteRemoteOperationResponse(new ReadDataReply(
  88            TestModel.createTestContext(), null).toSerializable());
  89
  90         read = transactionProxy.read(TestModel.TEST_PATH);
  91
  92         normalizedNodeOptional = read.get();
  93
  94         Assert.assertFalse(normalizedNodeOptional.isPresent());
  95     }
  96
  97     @Test
  98     public void testWrite() throws Exception {
  99         final Props props = Props.create(MessageCollectorActor.class);
 100         final ActorRef actorRef = getSystem().actorOf(props);
 101
 102         final MockActorContext actorContext = new MockActorContext(this.getSystem());
 103         actorContext.setExecuteShardOperationResponse(createTransactionReply(actorRef));
 104         actorContext.setExecuteRemoteOperationResponse("message");
 105
 106         TransactionProxy transactionProxy =
 107             new TransactionProxy(actorContext,
 108                 TransactionProxy.TransactionType.READ_ONLY, transactionExecutor, TestModel.createTestContext());
 109
 110         transactionProxy.write(TestModel.TEST_PATH,
 111             ImmutableNodes.containerNode(TestModel.NAME_QNAME));
 112
 113         ActorContext testContext = new ActorContext(getSystem(), getSystem().actorOf(Props.create(DoNothingActor.class)));
 114         Object messages = testContext
 115             .executeLocalOperation(actorRef, "messages",
 116                 ActorContext.ASK_DURATION);
 117
 118         Assert.assertNotNull(messages);
 119
 120         Assert.assertTrue(messages instanceof List);
 121
 122         List<Object> listMessages = (List<Object>) messages;
 123
 124         Assert.assertEquals(1, listMessages.size());
 125
 126         Assert.assertEquals(WriteData.SERIALIZABLE_CLASS, listMessages.get(0).getClass());
 127     }
 128
 129     @Test
 130     public void testMerge() throws Exception {
 131         final Props props = Props.create(MessageCollectorActor.class);
 132         final ActorRef actorRef = getSystem().actorOf(props);
 133
 134         final MockActorContext actorContext = new MockActorContext(this.getSystem());
 135         actorContext.setExecuteShardOperationResponse(createTransactionReply(actorRef));
 136         actorContext.setExecuteRemoteOperationResponse("message");
 137
 138         TransactionProxy transactionProxy =
 139             new TransactionProxy(actorContext,
 140                 TransactionProxy.TransactionType.READ_ONLY, transactionExecutor, TestModel.createTestContext());
 141
 142         transactionProxy.merge(TestModel.TEST_PATH,
 143             ImmutableNodes.containerNode(TestModel.NAME_QNAME));
 144
 145         ActorContext testContext = new ActorContext(getSystem(), getSystem().actorOf(Props.create(DoNothingActor.class)));
 146         Object messages = testContext
 147             .executeLocalOperation(actorRef, "messages",
 148                 ActorContext.ASK_DURATION);
 149
 150         Assert.assertNotNull(messages);
 151
 152         Assert.assertTrue(messages instanceof List);
 153
 154         List<Object> listMessages = (List<Object>) messages;
 155
 156         Assert.assertEquals(1, listMessages.size());
 157
 158         Assert.assertEquals(MergeData.SERIALIZABLE_CLASS, listMessages.get(0).getClass());
 159     }
 160
 161     @Test
 162     public void testDelete() throws Exception {
 163         final Props props = Props.create(MessageCollectorActor.class);
 164         final ActorRef actorRef = getSystem().actorOf(props);
 165
 166         final MockActorContext actorContext = new MockActorContext(this.getSystem());
 167         actorContext.setExecuteShardOperationResponse(createTransactionReply(actorRef));
 168         actorContext.setExecuteRemoteOperationResponse("message");
 169
 170         TransactionProxy transactionProxy =
 171             new TransactionProxy(actorContext,
 172                 TransactionProxy.TransactionType.READ_ONLY, transactionExecutor, TestModel.createTestContext());
 173
 174         transactionProxy.delete(TestModel.TEST_PATH);
 175
 176         ActorContext testContext = new ActorContext(getSystem(), getSystem().actorOf(Props.create(DoNothingActor.class)));
 177         Object messages = testContext
 178             .executeLocalOperation(actorRef, "messages",
 179                 ActorContext.ASK_DURATION);
 180
 181         Assert.assertNotNull(messages);
 182
 183         Assert.assertTrue(messages instanceof List);
 184
 185         List<Object> listMessages = (List<Object>) messages;
 186
 187         Assert.assertEquals(1, listMessages.size());
 188
 189         Assert.assertEquals(DeleteData.SERIALIZABLE_CLASS, listMessages.get(0).getClass());
 190     }
 191
 192     @Test
 193     public void testReady() throws Exception {
 194         final Props props = Props.create(DoNothingActor.class);
 195         final ActorRef doNothingActorRef = getSystem().actorOf(props);
 196
 197         final MockActorContext actorContext = new MockActorContext(this.getSystem());
 198         actorContext.setExecuteShardOperationResponse(createTransactionReply(doNothingActorRef));
 199         actorContext.setExecuteRemoteOperationResponse(new ReadyTransactionReply(doNothingActorRef.path()));
 200
 201         TransactionProxy transactionProxy =
 202             new TransactionProxy(actorContext,
 203                 TransactionProxy.TransactionType.READ_ONLY, transactionExecutor, TestModel.createTestContext());
 204
 205
 206         DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
 207
 208         Assert.assertTrue(ready instanceof ThreePhaseCommitCohortProxy);
 209
 210         ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready;
 211
 212         Assert.assertTrue("No cohort paths returned", proxy.getCohortPaths().size() > 0);
 213
 214     }
 215
 216     @Test
 217     public void testGetIdentifier(){
 218         final Props props = Props.create(DoNothingActor.class);
 219         final ActorRef doNothingActorRef = getSystem().actorOf(props);
 220
 221         final MockActorContext actorContext = new MockActorContext(this.getSystem());
 222         actorContext.setExecuteShardOperationResponse( createTransactionReply(doNothingActorRef) );
 223
 224         TransactionProxy transactionProxy =
 225             new TransactionProxy(actorContext,
 226                 TransactionProxy.TransactionType.READ_ONLY, transactionExecutor, TestModel.createTestContext());
 227
 228         Assert.assertNotNull(transactionProxy.getIdentifier());
 229     }
 230
 231     @Test
 232     public void testClose(){
 233         final Props props = Props.create(MessageCollectorActor.class);
 234         final ActorRef actorRef = getSystem().actorOf(props);
 235
 236         final MockActorContext actorContext = new MockActorContext(this.getSystem());
 237         actorContext.setExecuteShardOperationResponse(createTransactionReply(actorRef));
 238         actorContext.setExecuteRemoteOperationResponse("message");
 239
 240         TransactionProxy transactionProxy =
 241             new TransactionProxy(actorContext,
 242                 TransactionProxy.TransactionType.READ_ONLY, transactionExecutor, TestModel.createTestContext());
 243
 244         transactionProxy.close();
 245
 246         ActorContext testContext = new ActorContext(getSystem(), getSystem().actorOf(Props.create(DoNothingActor.class)));
 247         Object messages = testContext
 248             .executeLocalOperation(actorRef, "messages",
 249                 ActorContext.ASK_DURATION);
 250
 251         Assert.assertNotNull(messages);
 252
 253         Assert.assertTrue(messages instanceof List);
 254
 255         List<Object> listMessages = (List<Object>) messages;
 256
 257         Assert.assertEquals(1, listMessages.size());
 258
 259         Assert.assertTrue(listMessages.get(0) instanceof CloseTransaction);
 260     }
 261
 262     private CreateTransactionReply createTransactionReply(ActorRef actorRef){
 263         return CreateTransactionReply.newBuilder()
 264             .setTransactionActorPath(actorRef.path().toString())
 265             .setTransactionId("txn-1")
 266             .build();
 267     }
 268 }