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