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