common/src/test/java/org/opendaylight/transportpce/common/device/DeviceTransactionManagerTest.java

   1 /*
   2  * Copyright © 2017 Orange, 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.transportpce.common.device;
  10
  11 import static org.mockito.Matchers.any;
  12
  13 import com.google.common.base.Optional;
  14 import com.google.common.util.concurrent.Futures;
  15 import com.google.common.util.concurrent.ListenableFuture;
  16 import com.google.common.util.concurrent.ListeningExecutorService;
  17 import com.google.common.util.concurrent.MoreExecutors;
  18
  19 import java.util.LinkedList;
  20 import java.util.List;
  21 import java.util.concurrent.ExecutionException;
  22 import java.util.concurrent.Executors;
  23 import java.util.concurrent.Future;
  24 import java.util.concurrent.TimeUnit;
  25 import java.util.concurrent.TimeoutException;
  26
  27 import org.junit.After;
  28 import org.junit.Assert;
  29 import org.junit.Before;
  30 import org.junit.Test;
  31 import org.junit.runner.RunWith;
  32 import org.mockito.Mock;
  33 import org.mockito.Mockito;
  34 import org.mockito.runners.MockitoJUnitRunner;
  35 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
  36 import org.opendaylight.controller.md.sal.binding.api.MountPoint;
  37 import org.opendaylight.controller.md.sal.binding.api.MountPointService;
  38 import org.opendaylight.controller.md.sal.binding.api.ReadWriteTransaction;
  39 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  40 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.network.rev150608.Network;
  41 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.network.rev150608.NetworkBuilder;
  42 import org.opendaylight.yangtools.yang.binding.DataObject;
  43 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  44
  45
  46 @RunWith(MockitoJUnitRunner.class)
  47 public class DeviceTransactionManagerTest {
  48
  49     @Mock private MountPointService mountPointServiceMock;
  50     @Mock private MountPoint mountPointMock;
  51     @Mock private DataBroker dataBrokerMock;
  52     @Mock private ReadWriteTransaction rwTransactionMock;
  53
  54     private DeviceTransactionManagerImpl transactionManager;
  55     private String defaultDeviceId = "device-id";
  56     private LogicalDatastoreType defaultDatastore = LogicalDatastoreType.OPERATIONAL;
  57     private InstanceIdentifier<Network> defaultIid = InstanceIdentifier.create(Network.class);
  58     private Network defaultData = new NetworkBuilder().build();
  59     private long defaultTimeout = 1000;
  60     private TimeUnit defaultTimeUnit = TimeUnit.MILLISECONDS;
  61
  62     @Before
  63     public void before() {
  64         Mockito.when(mountPointServiceMock.getMountPoint(any())).thenReturn(Optional.of(mountPointMock));
  65         Mockito.when(mountPointMock.getService(any())).thenReturn(Optional.of(dataBrokerMock));
  66         Mockito.when(dataBrokerMock.newReadWriteTransaction()).thenReturn(rwTransactionMock);
  67         Mockito.when(rwTransactionMock.submit()).thenReturn(Futures.immediateCheckedFuture(null));
  68
  69         this.transactionManager = new DeviceTransactionManagerImpl(mountPointServiceMock, 3000);
  70     }
  71
  72     @After
  73     public void after() {
  74         transactionManager.preDestroy();
  75     }
  76
  77     @Test
  78     public void basicPositiveTransactionTest() {
  79         try {
  80             putAndSubmit(transactionManager, defaultDeviceId, defaultDatastore, defaultIid, defaultData);
  81         } catch (InterruptedException | ExecutionException e) {
  82             Assert.fail("Exception catched! " + e);
  83             return;
  84         }
  85
  86         Mockito.verify(rwTransactionMock, Mockito.times(1)).put(defaultDatastore, defaultIid, defaultData);
  87         Mockito.verify(rwTransactionMock, Mockito.times(1)).submit();
  88     }
  89
  90     @Test
  91     public void advancedPositiveTransactionTest() {
  92         try {
  93             Future<java.util.Optional<DeviceTransaction>> firstDeviceTxFuture =
  94                     transactionManager.getDeviceTransaction(defaultDeviceId);
  95             DeviceTransaction firstDeviceTx = firstDeviceTxFuture.get().get();
  96
  97             Future<java.util.Optional<DeviceTransaction>> secondDeviceTxFuture =
  98                     transactionManager.getDeviceTransaction(defaultDeviceId);
  99             Assert.assertFalse(secondDeviceTxFuture.isDone());
 100
 101             Future<java.util.Optional<DeviceTransaction>> thirdDeviceTxFuture =
 102                     transactionManager.getDeviceTransaction(defaultDeviceId);
 103             Assert.assertFalse(thirdDeviceTxFuture.isDone());
 104
 105             firstDeviceTx.put(defaultDatastore, defaultIid, defaultData);
 106             Assert.assertFalse(secondDeviceTxFuture.isDone());
 107             Assert.assertFalse(thirdDeviceTxFuture.isDone());
 108             Thread.sleep(200);
 109             Assert.assertFalse(secondDeviceTxFuture.isDone());
 110             Assert.assertFalse(thirdDeviceTxFuture.isDone());
 111
 112             Future<java.util.Optional<DeviceTransaction>> anotherDeviceTxFuture =
 113                     transactionManager.getDeviceTransaction("another-id");
 114             Assert.assertTrue(anotherDeviceTxFuture.isDone());
 115             anotherDeviceTxFuture.get().get().submit(defaultTimeout, defaultTimeUnit);
 116
 117             firstDeviceTx.submit(defaultTimeout, defaultTimeUnit);
 118             Thread.sleep(200);
 119             Assert.assertTrue(secondDeviceTxFuture.isDone());
 120             Assert.assertFalse(thirdDeviceTxFuture.isDone());
 121
 122             DeviceTransaction secondDeviceTx = secondDeviceTxFuture.get().get();
 123             secondDeviceTx.put(defaultDatastore, defaultIid, defaultData);
 124             Assert.assertFalse(thirdDeviceTxFuture.isDone());
 125
 126             secondDeviceTx.submit(defaultTimeout, defaultTimeUnit);
 127             Thread.sleep(200);
 128             Assert.assertTrue(thirdDeviceTxFuture.isDone());
 129
 130             DeviceTransaction thirdDeviceTx = thirdDeviceTxFuture.get().get();
 131             thirdDeviceTx.put(defaultDatastore, defaultIid, defaultData);
 132             thirdDeviceTx.submit(defaultTimeout, defaultTimeUnit);
 133
 134             Mockito.verify(rwTransactionMock, Mockito.times(3)).put(defaultDatastore, defaultIid, defaultData);
 135             Mockito.verify(rwTransactionMock, Mockito.times(4)).submit();
 136         } catch (InterruptedException | ExecutionException e) {
 137             Assert.fail("Exception catched! " + e);
 138         }
 139     }
 140
 141     @Test
 142     public void bigAmountOfTransactionsOnSameDeviceTest() {
 143         int numberOfTxs = 100;
 144         List<Future<java.util.Optional<DeviceTransaction>>> deviceTransactionFutures = new LinkedList<>();
 145         List<DeviceTransaction> deviceTransactions = new LinkedList<>();
 146
 147         for (int i = 0; i < numberOfTxs; i++) {
 148             deviceTransactionFutures.add(transactionManager.getDeviceTransaction(defaultDeviceId));
 149         }
 150
 151         try {
 152             for (Future<java.util.Optional<DeviceTransaction>> futureTx : deviceTransactionFutures) {
 153                 DeviceTransaction deviceTx = futureTx.get().get();
 154                 deviceTx.submit(defaultTimeout, defaultTimeUnit);
 155                 deviceTransactions.add(deviceTx);
 156             }
 157         } catch (InterruptedException | ExecutionException e) {
 158             Assert.fail("Exception catched! " + e);
 159         }
 160
 161         for (DeviceTransaction deviceTx : deviceTransactions) {
 162             Assert.assertTrue(deviceTx.wasSubmittedOrCancelled().get());
 163         }
 164     }
 165
 166     @Test
 167     public void bigAmountOfTransactionsOnDifferentDevicesTest() {
 168         int numberOfTxs = 1000;
 169         List<DeviceTransaction> deviceTransactions = new LinkedList<>();
 170
 171         try {
 172             for (int i = 0; i < numberOfTxs; i++) {
 173                 deviceTransactions.add(transactionManager.getDeviceTransaction(defaultDeviceId + " " + i).get().get());
 174             }
 175         } catch (InterruptedException | ExecutionException e) {
 176             Assert.fail("Exception catched! " + e);
 177         }
 178
 179         deviceTransactions.parallelStream()
 180                 .forEach(deviceTransaction -> deviceTransaction.submit(defaultTimeout, defaultTimeUnit));
 181
 182         deviceTransactions.parallelStream()
 183                 .forEach(deviceTransaction -> Assert.assertTrue(deviceTransaction.wasSubmittedOrCancelled().get()));
 184     }
 185
 186     @Test
 187     public void bigAmountOfTransactionsOnDifferentDevicesWithoutSubmitTest() {
 188         int numberOfTxs = 1000;
 189         List<DeviceTransaction> deviceTransactions = new LinkedList<>();
 190
 191         try {
 192             for (int i = 0; i < numberOfTxs; i++) {
 193                 deviceTransactions.add(transactionManager.getDeviceTransaction(defaultDeviceId + " " + i).get().get());
 194             }
 195         } catch (InterruptedException | ExecutionException e) {
 196             Assert.fail("Exception catched! " + e);
 197         }
 198
 199         try {
 200             Thread.sleep(transactionManager.getMaxDurationToSubmitTransaction() + 1000);
 201         } catch (InterruptedException e) {
 202             Assert.fail("Exception catched! " + e);
 203         }
 204         deviceTransactions.parallelStream()
 205                 .forEach(deviceTransaction -> Assert.assertTrue(deviceTransaction.wasSubmittedOrCancelled().get()));
 206     }
 207
 208     @Test
 209     public void notSubmittedTransactionTest() {
 210         Future<java.util.Optional<DeviceTransaction>> deviceTxFuture =
 211                 transactionManager.getDeviceTransaction(defaultDeviceId);
 212         try {
 213             deviceTxFuture.get();
 214             Thread.sleep(transactionManager.getMaxDurationToSubmitTransaction() + 1000);
 215         } catch (InterruptedException | ExecutionException e) {
 216             Assert.fail("Exception catched! " + e);
 217         }
 218         Mockito.verify(rwTransactionMock, Mockito.times(1)).cancel();
 219
 220         try {
 221             putAndSubmit(transactionManager, defaultDeviceId, defaultDatastore, defaultIid, defaultData);
 222         } catch (InterruptedException | ExecutionException e) {
 223             Assert.fail("Exception catched! " + e);
 224             return;
 225         }
 226
 227         Mockito.verify(rwTransactionMock, Mockito.times(1)).cancel();
 228         Mockito.verify(rwTransactionMock, Mockito.times(1)).put(defaultDatastore, defaultIid, defaultData);
 229         Mockito.verify(rwTransactionMock, Mockito.times(1)).submit();
 230     }
 231
 232     @Test
 233     public void dataBrokerTimeoutTransactionTest() {
 234         Mockito.when(dataBrokerMock.newReadWriteTransaction()).then(invocation -> {
 235             Thread.sleep(transactionManager.getMaxDurationToSubmitTransaction() + 1000);
 236             return rwTransactionMock;
 237         });
 238
 239         try {
 240             putAndSubmit(transactionManager, defaultDeviceId, defaultDatastore, defaultIid, defaultData);
 241         } catch (InterruptedException | ExecutionException e) {
 242             Assert.fail("Exception catched! " + e);
 243         }
 244
 245         Mockito.verify(rwTransactionMock, Mockito.times(1)).submit();
 246
 247         Mockito.when(dataBrokerMock.newReadWriteTransaction()).thenReturn(rwTransactionMock); // remove sleep
 248
 249         try {
 250             putAndSubmit(transactionManager, defaultDeviceId, defaultDatastore, defaultIid, defaultData);
 251         } catch (InterruptedException | ExecutionException e) {
 252             Assert.fail("Exception catched! " + e);
 253             return;
 254         }
 255
 256         Mockito.verify(rwTransactionMock, Mockito.times(2)).put(defaultDatastore, defaultIid, defaultData);
 257         Mockito.verify(rwTransactionMock, Mockito.times(2)).submit();
 258     }
 259
 260     @Test
 261     public void getFutureTimeoutTransactionTest() {
 262         Mockito.when(dataBrokerMock.newReadWriteTransaction()).then(invocation -> {
 263             Thread.sleep(3000);
 264             return rwTransactionMock;
 265         });
 266
 267         Exception throwedException = null;
 268
 269         Future<java.util.Optional<DeviceTransaction>> deviceTxFuture =
 270                 transactionManager.getDeviceTransaction(defaultDeviceId);
 271         try {
 272             deviceTxFuture.get(1000, TimeUnit.MILLISECONDS);
 273         } catch (InterruptedException | ExecutionException e) {
 274             Assert.fail("Exception catched! " + e);
 275         } catch (TimeoutException e) {
 276             throwedException = e;
 277         }
 278
 279         if (throwedException == null) {
 280             Assert.fail("TimeoutException should be thrown!");
 281             return;
 282         }
 283
 284         Mockito.when(dataBrokerMock.newReadWriteTransaction()).thenReturn(rwTransactionMock); // remove sleep
 285
 286         try {
 287             putAndSubmit(transactionManager, defaultDeviceId, defaultDatastore, defaultIid, defaultData);
 288         } catch (InterruptedException | ExecutionException e) {
 289             Assert.fail("Exception catched! " + e);
 290             return;
 291         }
 292
 293         Mockito.verify(rwTransactionMock, Mockito.times(1)).put(defaultDatastore, defaultIid, defaultData);
 294         Mockito.verify(rwTransactionMock, Mockito.times(1)).submit();
 295     }
 296
 297     @Test
 298     public void submitTxTimeoutTransactionTest() {
 299         ListeningExecutorService executor = MoreExecutors.listeningDecorator(Executors.newSingleThreadExecutor());
 300         Mockito.when(rwTransactionMock.submit()).then(invocation -> Futures.makeChecked(executor.submit(() -> {
 301             try {
 302                 Thread.sleep(3000);
 303             } catch (InterruptedException e) {
 304                 Assert.fail("Exception catched in future! " + e);
 305             }
 306             return null;
 307         }), input -> input));
 308
 309         Future<java.util.Optional<DeviceTransaction>> deviceTxFuture =
 310                 transactionManager.getDeviceTransaction(defaultDeviceId);
 311         DeviceTransaction deviceTx;
 312         try {
 313             deviceTx = deviceTxFuture.get().get();
 314         } catch (InterruptedException | ExecutionException e) {
 315             Assert.fail("Exception catched! " + e);
 316             return;
 317         }
 318
 319         deviceTx.put(defaultDatastore, defaultIid, defaultData);
 320
 321         Exception throwedException = null;
 322
 323         ListenableFuture<Void> submitFuture = deviceTx.submit(200, defaultTimeUnit);
 324         try {
 325             submitFuture.get();
 326         } catch (InterruptedException e) {
 327             Assert.fail("Exception catched! " + e);
 328         } catch (ExecutionException e) {
 329             throwedException = e;
 330         }
 331
 332         if (throwedException == null
 333                 || !throwedException.getMessage().contains(TimeoutException.class.getName())) {
 334             Assert.fail("TimeoutException inside of should be thrown!");
 335             return;
 336         }
 337
 338
 339         Mockito.when(rwTransactionMock.submit()).thenReturn(Futures.immediateCheckedFuture(null));
 340
 341         try {
 342             putAndSubmit(transactionManager, defaultDeviceId, defaultDatastore, defaultIid, defaultData);
 343         } catch (InterruptedException | ExecutionException e) {
 344             Assert.fail("Exception catched! " + e);
 345             return;
 346         }
 347
 348         Mockito.verify(rwTransactionMock, Mockito.times(2)).put(defaultDatastore, defaultIid, defaultData);
 349         Mockito.verify(rwTransactionMock, Mockito.times(2)).submit();
 350
 351         executor.shutdown();
 352     }
 353
 354     private <T extends DataObject> void putAndSubmit(DeviceTransactionManagerImpl deviceTxManager, String deviceId,
 355             LogicalDatastoreType store, InstanceIdentifier<T> path, T data)
 356             throws ExecutionException, InterruptedException {
 357         Future<java.util.Optional<DeviceTransaction>> deviceTxFuture = deviceTxManager.getDeviceTransaction(deviceId);
 358         DeviceTransaction deviceTx = deviceTxFuture.get().get();
 359         deviceTx.put(store, path, data);
 360         deviceTx.submit(defaultTimeout, defaultTimeUnit);
 361     }
 362 }