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

   1 /*
   2  * Copyright (c) 2014, 2015 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   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
   7  */
   8
   9 package org.opendaylight.controller.cluster.datastore;
  10
  11 import akka.actor.ActorRef;
  12 import akka.actor.Props;
  13 import akka.pattern.AskTimeoutException;
  14 import akka.testkit.TestActorRef;
  15 import java.util.concurrent.TimeUnit;
  16 import org.junit.Test;
  17 import org.opendaylight.controller.cluster.datastore.identifiers.ShardIdentifier;
  18 import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardStats;
  19 import org.opendaylight.controller.cluster.datastore.node.utils.serialization.NormalizedNodeSerializer;
  20 import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
  21 import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
  22 import org.opendaylight.controller.protobuff.messages.common.NormalizedNodeMessages;
  23 import org.opendaylight.controller.protobuff.messages.transaction.ShardTransactionMessages;
  24 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  25 import org.opendaylight.yangtools.yang.data.impl.schema.Builders;
  26 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
  27 import scala.concurrent.Await;
  28 import scala.concurrent.Future;
  29 import scala.concurrent.duration.Duration;
  30
  31 /**
  32  * Covers negative test cases
  33  *
  34  * @author Basheeruddin Ahmed <syedbahm@cisco.com>
  35  */
  36 public class ShardTransactionFailureTest extends AbstractActorTest {
  37     private static final SchemaContext testSchemaContext =
  38             TestModel.createTestContext();
  39     private static final TransactionType RO = TransactionType.READ_ONLY;
  40     private static final TransactionType RW = TransactionType.READ_WRITE;
  41     private static final TransactionType WO = TransactionType.WRITE_ONLY;
  42
  43     private static final ShardDataTree store = new ShardDataTree(testSchemaContext);
  44
  45     private static final ShardIdentifier SHARD_IDENTIFIER =
  46         ShardIdentifier.builder().memberName("member-1")
  47             .shardName("inventory").type("operational").build();
  48
  49     private final DatastoreContext datastoreContext = DatastoreContext.newBuilder().build();
  50
  51     private final ShardStats shardStats = new ShardStats(SHARD_IDENTIFIER.toString(), "DataStore");
  52
  53     private ActorRef createShard(){
  54         return getSystem().actorOf(Shard.builder().id(SHARD_IDENTIFIER).datastoreContext(datastoreContext).
  55                 schemaContext(TestModel.createTestContext()).props());
  56     }
  57
  58     @Test(expected = ReadFailedException.class)
  59     public void testNegativeReadWithReadOnlyTransactionClosed()
  60         throws Throwable {
  61
  62         final ActorRef shard = createShard();
  63         final Props props = ShardTransaction.props(RO, store.newReadOnlyTransaction("test-txn", null), shard,
  64                 datastoreContext, shardStats, "txn", DataStoreVersions.CURRENT_VERSION);
  65
  66         final TestActorRef<ShardTransaction> subject = TestActorRef
  67             .create(getSystem(), props,
  68                 "testNegativeReadWithReadOnlyTransactionClosed");
  69
  70         ShardTransactionMessages.ReadData readData =
  71             ShardTransactionMessages.ReadData.newBuilder()
  72                 .setInstanceIdentifierPathArguments(
  73                     NormalizedNodeMessages.InstanceIdentifier.newBuilder()
  74                         .build()
  75                 ).build();
  76         Future<Object> future =
  77             akka.pattern.Patterns.ask(subject, readData, 3000);
  78         Await.result(future, Duration.create(3, TimeUnit.SECONDS));
  79
  80         subject.underlyingActor().getDOMStoreTransaction().abort();
  81
  82         future = akka.pattern.Patterns.ask(subject, readData, 3000);
  83         Await.result(future, Duration.create(3, TimeUnit.SECONDS));
  84     }
  85
  86
  87     @Test(expected = ReadFailedException.class)
  88     public void testNegativeReadWithReadWriteTransactionClosed()
  89         throws Throwable {
  90
  91         final ActorRef shard = createShard();
  92         final Props props = ShardTransaction.props(RW, store.newReadWriteTransaction("test-txn", null), shard,
  93                 datastoreContext, shardStats, "txn", DataStoreVersions.CURRENT_VERSION);
  94
  95         final TestActorRef<ShardTransaction> subject = TestActorRef
  96             .create(getSystem(), props,
  97                 "testNegativeReadWithReadWriteTransactionClosed");
  98
  99         ShardTransactionMessages.ReadData readData =
 100             ShardTransactionMessages.ReadData.newBuilder()
 101                 .setInstanceIdentifierPathArguments(
 102                     NormalizedNodeMessages.InstanceIdentifier.newBuilder()
 103                         .build()
 104                 ).build();
 105
 106         Future<Object> future =
 107             akka.pattern.Patterns.ask(subject, readData, 3000);
 108         Await.result(future, Duration.create(3, TimeUnit.SECONDS));
 109
 110         subject.underlyingActor().getDOMStoreTransaction().abort();
 111
 112         future = akka.pattern.Patterns.ask(subject, readData, 3000);
 113         Await.result(future, Duration.create(3, TimeUnit.SECONDS));
 114     }
 115
 116     @Test(expected = ReadFailedException.class)
 117     public void testNegativeExistsWithReadWriteTransactionClosed()
 118         throws Throwable {
 119
 120         final ActorRef shard = createShard();
 121         final Props props = ShardTransaction.props(RW, store.newReadWriteTransaction("test-txn", null), shard,
 122                 datastoreContext, shardStats, "txn", DataStoreVersions.CURRENT_VERSION);
 123
 124         final TestActorRef<ShardTransaction> subject = TestActorRef
 125             .create(getSystem(), props,
 126                 "testNegativeExistsWithReadWriteTransactionClosed");
 127
 128         ShardTransactionMessages.DataExists dataExists =
 129             ShardTransactionMessages.DataExists.newBuilder()
 130                 .setInstanceIdentifierPathArguments(
 131                     NormalizedNodeMessages.InstanceIdentifier.newBuilder()
 132                         .build()
 133                 ).build();
 134
 135         Future<Object> future =
 136             akka.pattern.Patterns.ask(subject, dataExists, 3000);
 137         Await.result(future, Duration.create(3, TimeUnit.SECONDS));
 138
 139         subject.underlyingActor().getDOMStoreTransaction().abort();
 140
 141         future = akka.pattern.Patterns.ask(subject, dataExists, 3000);
 142         Await.result(future, Duration.create(3, TimeUnit.SECONDS));
 143     }
 144
 145     @Test(expected = AskTimeoutException.class)
 146     public void testNegativeWriteWithTransactionReady() throws Exception {
 147
 148
 149         final ActorRef shard = createShard();
 150         final Props props = ShardTransaction.props(WO, store.newReadWriteTransaction("test-txn", null), shard,
 151                 datastoreContext, shardStats, "txn", DataStoreVersions.CURRENT_VERSION);
 152
 153         final TestActorRef<ShardTransaction> subject = TestActorRef
 154             .create(getSystem(), props,
 155                 "testNegativeWriteWithTransactionReady");
 156
 157         ShardTransactionMessages.ReadyTransaction readyTransaction =
 158             ShardTransactionMessages.ReadyTransaction.newBuilder().build();
 159
 160         Future<Object> future =
 161             akka.pattern.Patterns.ask(subject, readyTransaction, 3000);
 162         Await.result(future, Duration.create(3, TimeUnit.SECONDS));
 163
 164         ShardTransactionMessages.WriteData writeData =
 165             ShardTransactionMessages.WriteData.newBuilder()
 166                 .setInstanceIdentifierPathArguments(
 167                     NormalizedNodeMessages.InstanceIdentifier.newBuilder()
 168                         .build()).setNormalizedNode(
 169                 buildNormalizedNode()
 170
 171             ).build();
 172
 173         future = akka.pattern.Patterns.ask(subject, writeData, 3000);
 174         Await.result(future, Duration.create(3, TimeUnit.SECONDS));
 175     }
 176
 177     @Test(expected = AskTimeoutException.class)
 178     public void testNegativeReadWriteWithTransactionReady() throws Exception {
 179
 180
 181         final ActorRef shard = createShard();
 182         final Props props = ShardTransaction.props(RW, store.newReadWriteTransaction("test-txn", null), shard,
 183                 datastoreContext, shardStats, "txn", DataStoreVersions.CURRENT_VERSION);
 184
 185         final TestActorRef<ShardTransaction> subject = TestActorRef
 186             .create(getSystem(), props,
 187                 "testNegativeReadWriteWithTransactionReady");
 188
 189         ShardTransactionMessages.ReadyTransaction readyTransaction =
 190             ShardTransactionMessages.ReadyTransaction.newBuilder().build();
 191
 192         Future<Object> future =
 193             akka.pattern.Patterns.ask(subject, readyTransaction, 3000);
 194         Await.result(future, Duration.create(3, TimeUnit.SECONDS));
 195
 196         ShardTransactionMessages.WriteData writeData =
 197             ShardTransactionMessages.WriteData.newBuilder()
 198                 .setInstanceIdentifierPathArguments(
 199                     NormalizedNodeMessages.InstanceIdentifier.newBuilder()
 200                         .build()
 201                 )
 202                 .setNormalizedNode(buildNormalizedNode())
 203                 .build();
 204
 205         future = akka.pattern.Patterns.ask(subject, writeData, 3000);
 206         Await.result(future, Duration.create(3, TimeUnit.SECONDS));
 207     }
 208
 209     private static NormalizedNodeMessages.Node buildNormalizedNode() {
 210         return NormalizedNodeSerializer
 211             .serialize(Builders.containerBuilder().withNodeIdentifier(new YangInstanceIdentifier.NodeIdentifier(TestModel.TEST_QNAME)).build());
 212     }
 213
 214     @Test(expected = AskTimeoutException.class)
 215     public void testNegativeMergeTransactionReady() throws Exception {
 216
 217
 218         final ActorRef shard = createShard();
 219         final Props props = ShardTransaction.props(RW, store.newReadWriteTransaction("test-txn", null), shard,
 220                 datastoreContext, shardStats, "txn", DataStoreVersions.CURRENT_VERSION);
 221
 222         final TestActorRef<ShardTransaction> subject = TestActorRef
 223             .create(getSystem(), props, "testNegativeMergeTransactionReady");
 224
 225         ShardTransactionMessages.ReadyTransaction readyTransaction =
 226             ShardTransactionMessages.ReadyTransaction.newBuilder().build();
 227
 228         Future<Object> future =
 229             akka.pattern.Patterns.ask(subject, readyTransaction, 3000);
 230         Await.result(future, Duration.create(3, TimeUnit.SECONDS));
 231
 232         ShardTransactionMessages.MergeData mergeData =
 233             ShardTransactionMessages.MergeData.newBuilder()
 234                 .setInstanceIdentifierPathArguments(
 235                     NormalizedNodeMessages.InstanceIdentifier.newBuilder()
 236                         .build()).setNormalizedNode(
 237                 buildNormalizedNode()
 238
 239             ).build();
 240
 241         future = akka.pattern.Patterns.ask(subject, mergeData, 3000);
 242         Await.result(future, Duration.create(3, TimeUnit.SECONDS));
 243     }
 244
 245
 246     @Test(expected = AskTimeoutException.class)
 247     public void testNegativeDeleteDataWhenTransactionReady() throws Exception {
 248
 249
 250         final ActorRef shard = createShard();
 251         final Props props = ShardTransaction.props(RW, store.newReadWriteTransaction("test-txn", null), shard,
 252                 datastoreContext, shardStats, "txn", DataStoreVersions.CURRENT_VERSION);
 253
 254         final TestActorRef<ShardTransaction> subject = TestActorRef
 255             .create(getSystem(), props,
 256                 "testNegativeDeleteDataWhenTransactionReady");
 257
 258         ShardTransactionMessages.ReadyTransaction readyTransaction =
 259             ShardTransactionMessages.ReadyTransaction.newBuilder().build();
 260
 261         Future<Object> future =
 262             akka.pattern.Patterns.ask(subject, readyTransaction, 3000);
 263         Await.result(future, Duration.create(3, TimeUnit.SECONDS));
 264
 265         ShardTransactionMessages.DeleteData deleteData =
 266             ShardTransactionMessages.DeleteData.newBuilder()
 267                 .setInstanceIdentifierPathArguments(
 268                     NormalizedNodeMessages.InstanceIdentifier.newBuilder()
 269                         .build()).build();
 270
 271         future = akka.pattern.Patterns.ask(subject, deleteData, 3000);
 272         Await.result(future, Duration.create(3, TimeUnit.SECONDS));
 273     }
 274 }