package com.rackspacecloud.blueflood.service;

import com.rackspacecloud.blueflood.rollup.Granularity;
import com.rackspacecloud.blueflood.rollup.SlotKey;
import org.junit.Before;
import org.junit.Test;
import org.mockito.Matchers;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;

import java.util.concurrent.ThreadPoolExecutor;

import static org.junit.Assert.assertSame;
import static org.mockito.Matchers.anyLong;
import static org.mockito.Mockito.*;

public class RollupServiceTest {

    ScheduleContext context;
    ShardStateManager shardStateManager;
    ThreadPoolExecutor locatorFetchExecutors;
    ThreadPoolExecutor rollupReadExecutors;
    ThreadPoolExecutor rollupWriteExecutors;
    ThreadPoolExecutor enumValidatorExecutor;
    long rollupDelayMillis;
    long delayedMetricRollupDelayMillis;
    long pollerPeriod;
    long configRefreshInterval;

    RollupService service;

    @Before
    public void setUp() {

        context = mock(ScheduleContext.class);
        shardStateManager = mock(ShardStateManager.class);
        locatorFetchExecutors = mock(ThreadPoolExecutor.class);
        rollupReadExecutors = mock(ThreadPoolExecutor.class);
        rollupWriteExecutors = mock(ThreadPoolExecutor.class);
        enumValidatorExecutor = mock(ThreadPoolExecutor.class);
        rollupDelayMillis = 300000;
        delayedMetricRollupDelayMillis = 300000;
        pollerPeriod = 0;
        configRefreshInterval = 10000;

        service = new RollupService(context, shardStateManager,
                locatorFetchExecutors, rollupReadExecutors,
                rollupWriteExecutors, enumValidatorExecutor, rollupDelayMillis,
                delayedMetricRollupDelayMillis, pollerPeriod,
                configRefreshInterval);
    }

    @Test
    public void pollSchedulesEligibleSlots() {

        // when
        service.poll();

        // then
        verify(context).scheduleEligibleSlots(anyLong(), anyLong());
        verifyNoMoreInteractions(context);

        verifyZeroInteractions(shardStateManager);
        verifyZeroInteractions(locatorFetchExecutors);
        verifyZeroInteractions(rollupReadExecutors);
        verifyZeroInteractions(rollupWriteExecutors);
        verifyZeroInteractions(enumValidatorExecutor);
    }

    @Test
    public void runSingleSlotDequeuesAndExecutes() {

        // given
        when(context.hasScheduled()).thenReturn(true).thenReturn(false);

        SlotKey slotkey = SlotKey.of(Granularity.MIN_20, 2, 0);
        doReturn(slotkey).when(context).getNextScheduled();

        final Runnable[] capture = new Runnable[1];
        doAnswer(new Answer() {
            @Override
            public Object answer(InvocationOnMock invocationOnMock) throws Throwable {
                capture[0] = (Runnable)invocationOnMock.getArguments()[0];

                // now that the runnable has been queue, stop the outer loop
                service.setShouldKeepRunning(false);

                return null;
            }
        }).when(locatorFetchExecutors).execute(Matchers.<Runnable>any());


        // when
        service.run();

        // then
        verify(context).scheduleEligibleSlots(anyLong(), anyLong());  // from poll
        verify(context, times(2)).hasScheduled();
        verify(context).getNextScheduled();
        verify(context, times(2)).getCurrentTimeMillis();  // one from LocatorFetchRunnable ctor, one from run
        verifyNoMoreInteractions(context);

        verifyZeroInteractions(shardStateManager);

        verify(locatorFetchExecutors).execute(Matchers.<Runnable>any());
        assertSame(LocatorFetchRunnable.class, capture[0].getClass());
        verifyNoMoreInteractions(locatorFetchExecutors);

        verifyZeroInteractions(rollupReadExecutors);
        verifyZeroInteractions(rollupWriteExecutors);
        verifyZeroInteractions(enumValidatorExecutor);
    }

}

@@ -306,24 +306,80 @@
 ng.Answer;%0A%0A
+import java.util.concurrent.RejectedExecutionException;%0A
 import java.
@@ -414,16 +414,16 @@
 ecutor;%0A
-
 %0Aimport 
@@ -2947,16 +2947,17 @@
 en queue
+d
 , stop t
@@ -3954,10 +3954,1631 @@
 %0A    %7D%0A%0A
+    @Test%0A    public void IfTheExecutionIsRejectedThenTheSlotKeyIsPushedBack() %7B%0A%0A        // given%0A        when(context.hasScheduled()).thenReturn(true).thenReturn(false);%0A%0A        SlotKey slotkey = SlotKey.of(Granularity.MIN_20, 2, 0);%0A        doReturn(slotkey).when(context).getNextScheduled();%0A%0A        final RejectedExecutionException cause = new RejectedExecutionException(%22exception for testing purposes%22);%0A        doAnswer(new Answer() %7B%0A             @Override%0A             public Object answer(InvocationOnMock invocationOnMock) throws Throwable %7B%0A                 // now we're inside the loop, stop the outer loop%0A                 service.setShouldKeepRunning(false);%0A%0A                 throw cause;%0A             %7D%0A        %7D).when(locatorFetchExecutors).execute(Matchers.%3CRunnable%3Eany());%0A%0A        // when%0A        service.run();%0A%0A        // then%0A        verify(context).scheduleEligibleSlots(anyLong(), anyLong());  // from poll%0A        verify(context, times(2)).hasScheduled();%0A        verify(context).getNextScheduled();%0A        verify(context, times(2)).getCurrentTimeMillis();  // one from LocatorFetchRunnable ctor, one from run%0A        verify(context).pushBackToScheduled(Matchers.%3CSlotKey%3Eany(), anyBoolean());%0A        verifyNoMoreInteractions(context);%0A%0A        verifyZeroInteractions(shardStateManager);%0A%0A        verify(locatorFetchExecutors).execute(Matchers.%3CRunnable%3Eany());%0A        verifyNoMoreInteractions(locatorFetchExecutors);%0A%0A        verifyZeroInteractions(rollupReadExecutors);%0A        verifyZeroInteractions(rollupWriteExecutors);%0A        verifyZeroInteractions(enumValidatorExecutor);%0A    %7D%0A
 %7D%0A

Test failure mode.