applications/forwardingrules-sync/src/test/java/org/opendaylight/openflowplugin/applications/frsync/impl/SyncReactorFutureZipDecoratorTest.java

   1 /**
   2  * Copyright (c) 2016 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.openflowplugin.applications.frsync.impl;
  10
  11 import com.google.common.util.concurrent.Futures;
  12 import com.google.common.util.concurrent.ListenableFuture;
  13 import com.google.common.util.concurrent.ListeningExecutorService;
  14 import com.google.common.util.concurrent.MoreExecutors;
  15 import com.google.common.util.concurrent.ThreadFactoryBuilder;
  16 import java.util.ArrayList;
  17 import java.util.List;
  18 import java.util.concurrent.CountDownLatch;
  19 import java.util.concurrent.ExecutorService;
  20 import java.util.concurrent.Executors;
  21 import java.util.concurrent.TimeUnit;
  22 import org.junit.After;
  23 import org.junit.Before;
  24 import org.junit.Test;
  25 import org.junit.runner.RunWith;
  26 import org.mockito.ArgumentMatchers;
  27 import org.mockito.InOrder;
  28 import org.mockito.Mock;
  29 import org.mockito.Mockito;
  30 import org.mockito.runners.MockitoJUnitRunner;
  31 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  32 import org.opendaylight.openflowplugin.applications.frsync.SyncReactor;
  33 import org.opendaylight.openflowplugin.applications.frsync.util.SyncupEntry;
  34 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
  35 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
  36 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.Nodes;
  37 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
  38 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.NodeKey;
  39 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  40 import org.slf4j.Logger;
  41 import org.slf4j.LoggerFactory;
  42
  43 /**
  44  * Test for {@link SyncReactorFutureZipDecorator}.
  45  */
  46 @RunWith(MockitoJUnitRunner.class)
  47 public class SyncReactorFutureZipDecoratorTest {
  48
  49     private static final Logger LOG = LoggerFactory.getLogger(SyncReactorFutureZipDecoratorTest.class);
  50     private static final NodeId NODE_ID = new NodeId("testNode");
  51     private SyncReactorFutureZipDecorator reactor;
  52     private InstanceIdentifier<FlowCapableNode> fcNodePath;
  53     private ListeningExecutorService syncThreadPool;
  54     private final LogicalDatastoreType configDS = LogicalDatastoreType.CONFIGURATION;
  55     private final LogicalDatastoreType operationalDS = LogicalDatastoreType.OPERATIONAL;
  56
  57     @Mock
  58     private SyncReactor delegate;
  59     @Mock
  60     private SyncupEntry syncupEntry;
  61
  62     @Before
  63     public void setUp() {
  64         final ExecutorService executorService = Executors.newSingleThreadExecutor(new ThreadFactoryBuilder()
  65                 .setDaemon(false)
  66                 .setNameFormat("frsync-test-%d")
  67                 .setUncaughtExceptionHandler((thread, ex) -> LOG.error("Uncaught exception {}", thread, ex))
  68                 .build());
  69         syncThreadPool = MoreExecutors.listeningDecorator(executorService);
  70         reactor = new SyncReactorFutureZipDecorator(delegate, syncThreadPool);
  71         fcNodePath = InstanceIdentifier.create(Nodes.class).child(Node.class, new NodeKey(NODE_ID))
  72                 .augmentation(FlowCapableNode.class);
  73     }
  74
  75     @Test
  76     public void testSyncupWithOptimizedConfigDeltaCompression() throws Exception {
  77         final FlowCapableNode dataBefore = Mockito.mock(FlowCapableNode.class);
  78         final FlowCapableNode dataAfter = Mockito.mock(FlowCapableNode.class);
  79         final FlowCapableNode dataAfter2 = Mockito.mock(FlowCapableNode.class);
  80         final CountDownLatch latchForFirst = new CountDownLatch(1);
  81         final CountDownLatch latchForNext = new CountDownLatch(1);
  82
  83         final SyncupEntry first = new SyncupEntry(dataBefore, configDS, null, operationalDS);
  84         final SyncupEntry second = new SyncupEntry(dataAfter, configDS, dataBefore, configDS);
  85         final SyncupEntry third = new SyncupEntry(null, configDS, dataAfter, configDS);
  86         final SyncupEntry fourth = new SyncupEntry(dataAfter2, configDS, null, configDS);
  87         final SyncupEntry zipped = new SyncupEntry(dataAfter2, configDS, dataBefore, configDS);
  88         final List<ListenableFuture<Boolean>> allResults = new ArrayList<>();
  89
  90         Mockito.when(delegate.syncup(ArgumentMatchers.<InstanceIdentifier<FlowCapableNode>>any(), Mockito.eq(first)))
  91                 .thenAnswer(invocationOnMock -> {
  92                     LOG.info("unlocking next configs");
  93                     latchForNext.countDown();
  94                     latchForFirst.await();
  95                     LOG.info("unlocking first delegate");
  96                     return Futures.immediateFuture(Boolean.TRUE);
  97                 });
  98
  99         allResults.add(reactor.syncup(fcNodePath, first));
 100         latchForNext.await();
 101
 102         mockSyncupWithEntry(second);
 103         allResults.add(reactor.syncup(fcNodePath, second));
 104         mockSyncupWithEntry(third);
 105         allResults.add(reactor.syncup(fcNodePath, third));
 106         mockSyncupWithEntry(fourth);
 107         allResults.add(reactor.syncup(fcNodePath, fourth));
 108         latchForFirst.countDown();
 109
 110         Futures.allAsList(allResults).get(1, TimeUnit.SECONDS);
 111         LOG.info("all configs done");
 112
 113         syncThreadPool.shutdown();
 114         boolean terminated = syncThreadPool.awaitTermination(1, TimeUnit.SECONDS);
 115         if (!terminated) {
 116             LOG.info("thread pool not terminated.");
 117             syncThreadPool.shutdownNow();
 118         }
 119         final InOrder inOrder = Mockito.inOrder(delegate);
 120         inOrder.verify(delegate).syncup(fcNodePath, first);
 121         inOrder.verify(delegate).syncup(fcNodePath, zipped);
 122         inOrder.verifyNoMoreInteractions();
 123     }
 124
 125     @Test
 126     public void testSyncupConfigEmptyQueue() throws Exception {
 127         final FlowCapableNode dataBefore = Mockito.mock(FlowCapableNode.class);
 128         final FlowCapableNode dataAfter = Mockito.mock(FlowCapableNode.class);
 129         final CountDownLatch latchForNext = new CountDownLatch(1);
 130
 131         final SyncupEntry first = new SyncupEntry(dataBefore, configDS, null, configDS);
 132         final SyncupEntry second = new SyncupEntry(dataAfter, configDS, dataBefore, configDS);
 133
 134         Mockito.when(delegate.syncup(ArgumentMatchers.<InstanceIdentifier<FlowCapableNode>>any(), Mockito.eq(first)))
 135                 .thenAnswer(invocationOnMock -> {
 136                     LOG.info("unlocking next config");
 137                     latchForNext.countDown();
 138                     return Futures.immediateFuture(Boolean.TRUE);
 139                 });
 140
 141         reactor.syncup(fcNodePath, first);
 142         latchForNext.await();
 143         mockSyncupWithEntry(second);
 144         reactor.syncup(fcNodePath, second);
 145
 146         boolean terminated = syncThreadPool.awaitTermination(1, TimeUnit.SECONDS);
 147         if (!terminated) {
 148             LOG.info("thread pool not terminated.");
 149             syncThreadPool.shutdownNow();
 150         }
 151         final InOrder inOrder = Mockito.inOrder(delegate);
 152         inOrder.verify(delegate).syncup(fcNodePath, first);
 153         inOrder.verify(delegate).syncup(fcNodePath, second);
 154         inOrder.verifyNoMoreInteractions();
 155     }
 156
 157     @Test
 158     public void testSyncupRewriteZipEntryWithOperationalDelta() throws Exception {
 159         final FlowCapableNode configBefore = Mockito.mock(FlowCapableNode.class);
 160         final FlowCapableNode configAfter = Mockito.mock(FlowCapableNode.class);
 161         final FlowCapableNode configActual = Mockito.mock(FlowCapableNode.class);
 162         final FlowCapableNode freshOperational = Mockito.mock(FlowCapableNode.class);
 163         final CountDownLatch latchForFirst = new CountDownLatch(1);
 164         final CountDownLatch latchForNext = new CountDownLatch(1);
 165
 166         final SyncupEntry first = new SyncupEntry(configAfter, configDS, configBefore, configDS);
 167         final SyncupEntry second = new SyncupEntry(configActual, configDS, freshOperational, operationalDS);
 168
 169         Mockito.when(delegate.syncup(ArgumentMatchers.<InstanceIdentifier<FlowCapableNode>>any(), Mockito.eq(first)))
 170                 .thenAnswer(invocationOnMock -> {
 171                     LOG.info("unlocking for fresh operational");
 172                     latchForNext.countDown();
 173                     latchForFirst.await();
 174                     LOG.info("unlocking first delegate");
 175                     return Futures.immediateFuture(Boolean.TRUE);
 176                 });
 177
 178         reactor.syncup(fcNodePath, first);
 179         latchForNext.await();
 180
 181         mockSyncupWithEntry(second);
 182         reactor.syncup(fcNodePath, second);
 183         latchForFirst.countDown();
 184
 185         syncThreadPool.shutdown();
 186         boolean terminated = syncThreadPool.awaitTermination(1, TimeUnit.SECONDS);
 187         if (!terminated) {
 188             LOG.info("thread pool not terminated.");
 189             syncThreadPool.shutdownNow();
 190         }
 191         Mockito.verify(delegate, Mockito.times(1)).syncup(fcNodePath, second);
 192     }
 193
 194     private void mockSyncupWithEntry(final SyncupEntry entry) {
 195         Mockito.when(delegate.syncup(ArgumentMatchers.any(), Mockito.eq(entry)))
 196                 .thenReturn(Futures.immediateFuture(Boolean.TRUE));
 197     }
 198
 199     @After
 200     public void tearDown() {
 201         syncThreadPool.shutdownNow();
 202     }
 203 }