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