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.ListeningExecutorService;
  15 import com.google.common.util.concurrent.MoreExecutors;
  16 import java.util.LinkedList;
  17 import java.util.List;
  18 import java.util.Optional;
  19 import java.util.concurrent.ExecutionException;
  20 import java.util.concurrent.Executors;
  21 import java.util.concurrent.Future;
  22 import java.util.concurrent.TimeUnit;
  23 import java.util.concurrent.TimeoutException;
  24 import org.eclipse.jdt.annotation.NonNull;
  25 import org.junit.After;
  26 import org.junit.Assert;
  27 import org.junit.Before;
  28 import org.junit.Ignore;
  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.when(rwTransactionMock.commit()).thenReturn(FluentFutures.immediateNullFluentFuture());
  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     @Ignore
  93     public void advancedPositiveTransactionTest() {
  94         try {
  95             Future<java.util.Optional<DeviceTransaction>> firstDeviceTxFuture =
  96                     transactionManager.getDeviceTransaction(defaultDeviceId);
  97             DeviceTransaction firstDeviceTx = firstDeviceTxFuture.get().get();
  98
  99             Future<java.util.Optional<DeviceTransaction>> secondDeviceTxFuture =
 100                     transactionManager.getDeviceTransaction(defaultDeviceId);
 101             Assert.assertFalse(secondDeviceTxFuture.isDone());
 102
 103             Future<java.util.Optional<DeviceTransaction>> thirdDeviceTxFuture =
 104                     transactionManager.getDeviceTransaction(defaultDeviceId);
 105             Assert.assertFalse(thirdDeviceTxFuture.isDone());
 106
 107             firstDeviceTx.put(defaultDatastore, defaultIid, defaultData);
 108             Assert.assertFalse(secondDeviceTxFuture.isDone());
 109             Assert.assertFalse(thirdDeviceTxFuture.isDone());
 110             Thread.sleep(200);
 111             Assert.assertFalse(secondDeviceTxFuture.isDone());
 112             Assert.assertFalse(thirdDeviceTxFuture.isDone());
 113
 114             Future<java.util.Optional<DeviceTransaction>> anotherDeviceTxFuture =
 115                     transactionManager.getDeviceTransaction("another-id");
 116             Assert.assertTrue(anotherDeviceTxFuture.isDone());
 117             anotherDeviceTxFuture.get().get().commit(defaultTimeout, defaultTimeUnit);
 118
 119             firstDeviceTx.commit(defaultTimeout, defaultTimeUnit);
 120             Thread.sleep(200);
 121             Assert.assertTrue(secondDeviceTxFuture.isDone());
 122             Assert.assertFalse(thirdDeviceTxFuture.isDone());
 123
 124             DeviceTransaction secondDeviceTx = secondDeviceTxFuture.get().get();
 125             secondDeviceTx.put(defaultDatastore, defaultIid, defaultData);
 126             Assert.assertFalse(thirdDeviceTxFuture.isDone());
 127
 128             secondDeviceTx.commit(defaultTimeout, defaultTimeUnit);
 129             Thread.sleep(200);
 130             Assert.assertTrue(thirdDeviceTxFuture.isDone());
 131
 132             DeviceTransaction thirdDeviceTx = thirdDeviceTxFuture.get().get();
 133             thirdDeviceTx.put(defaultDatastore, defaultIid, defaultData);
 134             thirdDeviceTx.commit(defaultTimeout, defaultTimeUnit);
 135
 136             Mockito.verify(rwTransactionMock, Mockito.times(3)).put(defaultDatastore, defaultIid, defaultData);
 137             Mockito.verify(rwTransactionMock, Mockito.times(4)).commit();
 138         } catch (InterruptedException | ExecutionException e) {
 139             Assert.fail("Exception catched! " + e);
 140         }
 141     }
 142
 143     @Test
 144     public void bigAmountOfTransactionsOnSameDeviceTest() {
 145         int numberOfTxs = 100;
 146         List<Future<java.util.Optional<DeviceTransaction>>> deviceTransactionFutures = new LinkedList<>();
 147         List<DeviceTransaction> deviceTransactions = new LinkedList<>();
 148
 149         for (int i = 0; i < numberOfTxs; i++) {
 150             deviceTransactionFutures.add(transactionManager.getDeviceTransaction(defaultDeviceId));
 151         }
 152
 153         try {
 154             for (Future<java.util.Optional<DeviceTransaction>> futureTx : deviceTransactionFutures) {
 155                 DeviceTransaction deviceTx = futureTx.get().get();
 156                 deviceTx.commit(defaultTimeout, defaultTimeUnit);
 157                 deviceTransactions.add(deviceTx);
 158             }
 159         } catch (InterruptedException | ExecutionException e) {
 160             Assert.fail("Exception catched! " + e);
 161         }
 162
 163         for (DeviceTransaction deviceTx : deviceTransactions) {
 164             Assert.assertTrue(deviceTx.wasSubmittedOrCancelled().get());
 165         }
 166     }
 167
 168     @Test
 169     public void bigAmountOfTransactionsOnDifferentDevicesTest() {
 170         int numberOfTxs = 1000;
 171         List<DeviceTransaction> deviceTransactions = new LinkedList<>();
 172
 173         try {
 174             for (int i = 0; i < numberOfTxs; i++) {
 175                 deviceTransactions.add(transactionManager.getDeviceTransaction(defaultDeviceId + " " + i).get().get());
 176             }
 177         } catch (InterruptedException | ExecutionException e) {
 178             Assert.fail("Exception catched! " + e);
 179         }
 180
 181         deviceTransactions.parallelStream()
 182                 .forEach(deviceTransaction -> deviceTransaction.commit(defaultTimeout, defaultTimeUnit));
 183
 184         deviceTransactions.parallelStream()
 185                 .forEach(deviceTransaction -> Assert.assertTrue(deviceTransaction.wasSubmittedOrCancelled().get()));
 186     }
 187
 188     @Test
 189     public void bigAmountOfTransactionsOnDifferentDevicesWithoutSubmitTest() {
 190         int numberOfTxs = 1000;
 191         List<DeviceTransaction> deviceTransactions = new LinkedList<>();
 192
 193         try {
 194             for (int i = 0; i < numberOfTxs; i++) {
 195                 deviceTransactions.add(transactionManager.getDeviceTransaction(defaultDeviceId + " " + i).get().get());
 196             }
 197         } catch (InterruptedException | ExecutionException e) {
 198             Assert.fail("Exception catched! " + e);
 199         }
 200
 201         try {
 202             Thread.sleep(transactionManager.getMaxDurationToSubmitTransaction() + 1000);
 203         } catch (InterruptedException e) {
 204             Assert.fail("Exception catched! " + e);
 205         }
 206         deviceTransactions.parallelStream()
 207                 .forEach(deviceTransaction -> Assert.assertTrue(deviceTransaction.wasSubmittedOrCancelled().get()));
 208     }
 209
 210     @Test
 211     public void notSubmittedTransactionTest() {
 212         Future<java.util.Optional<DeviceTransaction>> deviceTxFuture =
 213                 transactionManager.getDeviceTransaction(defaultDeviceId);
 214         try {
 215             deviceTxFuture.get();
 216             Thread.sleep(transactionManager.getMaxDurationToSubmitTransaction() + 1000);
 217         } catch (InterruptedException | ExecutionException e) {
 218             Assert.fail("Exception catched! " + e);
 219         }
 220         Mockito.verify(rwTransactionMock, Mockito.times(1)).cancel();
 221
 222         try {
 223             putAndSubmit(transactionManager, defaultDeviceId, defaultDatastore, defaultIid, defaultData);
 224         } catch (InterruptedException | ExecutionException e) {
 225             Assert.fail("Exception catched! " + e);
 226             return;
 227         }
 228
 229         Mockito.verify(rwTransactionMock, Mockito.times(1)).cancel();
 230         Mockito.verify(rwTransactionMock, Mockito.times(1)).put(defaultDatastore, defaultIid, defaultData);
 231         Mockito.verify(rwTransactionMock, Mockito.times(1)).commit();
 232     }
 233
 234     @Test
 235     public void dataBrokerTimeoutTransactionTest() {
 236         Mockito.when(dataBrokerMock.newReadWriteTransaction()).then(invocation -> {
 237             Thread.sleep(transactionManager.getMaxDurationToSubmitTransaction() + 1000);
 238             return rwTransactionMock;
 239         });
 240
 241         try {
 242             putAndSubmit(transactionManager, defaultDeviceId, defaultDatastore, defaultIid, defaultData);
 243         } catch (InterruptedException | ExecutionException e) {
 244             Assert.fail("Exception catched! " + e);
 245         }
 246
 247         Mockito.verify(rwTransactionMock, Mockito.times(1)).commit();
 248
 249         Mockito.when(dataBrokerMock.newReadWriteTransaction()).thenReturn(rwTransactionMock); // remove sleep
 250
 251         try {
 252             putAndSubmit(transactionManager, defaultDeviceId, defaultDatastore, defaultIid, defaultData);
 253         } catch (InterruptedException | ExecutionException e) {
 254             Assert.fail("Exception catched! " + e);
 255             return;
 256         }
 257
 258         Mockito.verify(rwTransactionMock, Mockito.times(2)).put(defaultDatastore, defaultIid, defaultData);
 259         Mockito.verify(rwTransactionMock, Mockito.times(2)).commit();
 260     }
 261
 262     @Test
 263     public void getFutureTimeoutTransactionTest() {
 264         Mockito.when(dataBrokerMock.newReadWriteTransaction()).then(invocation -> {
 265             Thread.sleep(3000);
 266             return rwTransactionMock;
 267         });
 268
 269         Exception throwedException = null;
 270
 271         Future<java.util.Optional<DeviceTransaction>> deviceTxFuture =
 272                 transactionManager.getDeviceTransaction(defaultDeviceId);
 273         try {
 274             deviceTxFuture.get(1000, TimeUnit.MILLISECONDS);
 275         } catch (InterruptedException | ExecutionException e) {
 276             Assert.fail("Exception catched! " + e);
 277         } catch (TimeoutException e) {
 278             throwedException = e;
 279         }
 280
 281         if (throwedException == null) {
 282             Assert.fail("TimeoutException should be thrown!");
 283             return;
 284         }
 285
 286         Mockito.when(dataBrokerMock.newReadWriteTransaction()).thenReturn(rwTransactionMock); // remove sleep
 287
 288         try {
 289             putAndSubmit(transactionManager, defaultDeviceId, defaultDatastore, defaultIid, defaultData);
 290         } catch (InterruptedException | ExecutionException e) {
 291             Assert.fail("Exception catched! " + e);
 292             return;
 293         }
 294
 295         Mockito.verify(rwTransactionMock, Mockito.times(1)).put(defaultDatastore, defaultIid, defaultData);
 296         Mockito.verify(rwTransactionMock, Mockito.times(1)).commit();
 297     }
 298
 299     @Test
 300     @Ignore
 301     public void submitTxTimeoutTransactionTest() {
 302         Future<java.util.Optional<DeviceTransaction>> deviceTxFuture =
 303                 transactionManager.getDeviceTransaction(defaultDeviceId);
 304         DeviceTransaction deviceTx;
 305         try {
 306             deviceTx = deviceTxFuture.get().get();
 307         } catch (InterruptedException | ExecutionException e) {
 308             Assert.fail("Exception catched! " + e);
 309             return;
 310         }
 311
 312         deviceTx.put(defaultDatastore, defaultIid, defaultData);
 313
 314         Exception throwedException = null;
 315
 316         FluentFuture<? extends @NonNull CommitInfo> submitFuture = deviceTx.commit(200, defaultTimeUnit);
 317         try {
 318             submitFuture.get();
 319         } catch (InterruptedException e) {
 320             Assert.fail("Exception catched! " + e);
 321         } catch (ExecutionException e) {
 322             throwedException = e;
 323         }
 324
 325         if (throwedException == null
 326                 || !throwedException.getMessage().contains(TimeoutException.class.getName())) {
 327             Assert.fail("TimeoutException inside of should be thrown!");
 328             return;
 329         }
 330
 331
 332         Mockito.doReturn(FluentFutures.immediateNullFluentFuture()).when(rwTransactionMock.commit());
 333
 334         try {
 335             putAndSubmit(transactionManager, defaultDeviceId, defaultDatastore, defaultIid, defaultData);
 336         } catch (InterruptedException | ExecutionException e) {
 337             Assert.fail("Exception catched! " + e);
 338             return;
 339         }
 340
 341         Mockito.verify(rwTransactionMock, Mockito.times(2)).put(defaultDatastore, defaultIid, defaultData);
 342         Mockito.verify(rwTransactionMock, Mockito.times(2)).commit();
 343         ListeningExecutorService executor = MoreExecutors
 344                 .listeningDecorator(Executors.newSingleThreadExecutor());
 345         executor.shutdown();
 346     }
 347
 348     private <T extends DataObject> void putAndSubmit(DeviceTransactionManagerImpl deviceTxManager, String deviceId,
 349             LogicalDatastoreType store, InstanceIdentifier<T> path, T data)
 350             throws ExecutionException, InterruptedException {
 351         Future<java.util.Optional<DeviceTransaction>> deviceTxFuture = deviceTxManager.getDeviceTransaction(deviceId);
 352         DeviceTransaction deviceTx = deviceTxFuture.get().get();
 353         deviceTx.put(store, path, data);
 354         deviceTx.commit(defaultTimeout, defaultTimeUnit);
 355     }
 356 }