/*
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.spotify.reaper.service;

import com.spotify.reaper.core.RepairRun;
import com.spotify.reaper.core.RepairSegment;
import com.spotify.reaper.storage.IStorage;
import com.spotify.reaper.storage.MemoryStorage;

import org.apache.cassandra.service.ActiveRepairService;
import org.joda.time.DateTime;
import org.joda.time.DateTimeUtils;
import org.junit.Test;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;

public class RepairRunnerTest {

  @Test
  public void noSegmentsTest() throws InterruptedException {
    final int RUN_ID = 1;
    final int CF_ID = 1;
    final double INTENSITY = 0.5f;
    final long TIME_CREATION = 41l;
    final long TIME_START = 42l;
    final long TIME_END = 43l;
    final String TEST_CLUSTER = "TestCluster";

    IStorage storage = new MemoryStorage();

    // place a dummy cluster into storage
    storage.addCluster(new Cluster(TEST_CLUSTER, null, Collections.<String>singleton(null)));

    // place a dummy repair run into the storage
    DateTimeUtils.setCurrentMillisFixed(TIME_CREATION);
    RepairRun.Builder runBuilder = new RepairRun.Builder("TestCluster", CF_ID,
        RepairRun.RunState.NOT_STARTED, DateTime.now(), INTENSITY);
    storage.addRepairRun(runBuilder);
    storage.addRepairSegments(Collections.<RepairSegment.Builder>emptySet(), RUN_ID);

    // start the repair
    DateTimeUtils.setCurrentMillisFixed(TIME_START);
    RepairRunner.initializeThreadPool(1, 180);
    RepairRunner.startNewRepairRun(storage, RUN_ID, new JmxConnectionFactory() {
      @Override
      public JmxProxy create(Optional<RepairStatusHandler> handler, String host)
          throws ReaperException {
        return null;
      }
    });
    Thread.sleep(200);

    // check if the start time was properly set
    DateTime startTime = storage.getRepairRun(RUN_ID).getStartTime();
    assertNotNull(startTime);
    assertEquals(TIME_START, startTime.getMillis());

    // end the repair
    DateTimeUtils.setCurrentMillisFixed(TIME_END);
    RepairRun run = storage.getRepairRun(RUN_ID);
    storage.updateRepairRun(run.with().runState(RepairRun.RunState.RUNNING).build(RUN_ID));
    RepairRunner.startNewRepairRun(storage, RUN_ID, new JmxConnectionFactory() {
      @Override
      public JmxProxy create(Optional<RepairStatusHandler> handler, String host)
          throws ReaperException {
        return null;
      }
    });
    Thread.sleep(200);
  }


  @Test
  public void testHangingRepair() throws ReaperException, InterruptedException {
    final String CLUSTER_NAME = "reaper";
    final String KS_NAME = "reaper";
    final String CF_NAME = "reaper";
    final long TIME_RUN = 41l;
    final long TIME_RERUN = 42l;
    final double INTENSITY = 0.5f;

    IStorage storage = new MemoryStorage();

    storage.addCluster(new Cluster(CLUSTER_NAME, null, Collections.<String>singleton(null)));

    ColumnFamily cf =
        storage.addColumnFamily(new ColumnFamily.Builder(CLUSTER_NAME, KS_NAME, CF_NAME, 1, false));

    DateTimeUtils.setCurrentMillisFixed(TIME_RUN);
    RepairRun repairRun = storage.addRepairRun(
        new RepairRun.Builder(CLUSTER_NAME, cf.getId(), RepairRun.RunState.NOT_STARTED,
            DateTime.now(), INTENSITY));

    storage.addRepairSegments(Collections.singleton(
        new RepairSegment.Builder(repairRun.getId(), new RingRange(BigInteger.ZERO, BigInteger.ONE),
            RepairSegment.State.NOT_STARTED)));

    final JmxProxy jmx = mock(JmxProxy.class);

    when(jmx.getClusterName()).thenReturn(CLUSTER_NAME);
    when(jmx.isConnectionAlive()).thenReturn(true);
    when(jmx.tokenRangeToEndpoint(anyString(), any(RingRange.class)))
        .thenReturn(Lists.newArrayList(""));

    final AtomicInteger repairAttempts = new AtomicInteger(0);
    when(jmx.triggerRepair(any(BigInteger.class), any(BigInteger.class), anyString(), anyString()))
        .then(new Answer<Integer>() {
          @Override
          public Integer answer(InvocationOnMock invocation) throws Throwable {
            return repairAttempts.incrementAndGet();
          }
        });

    RepairRunner.initializeThreadPool(1, 1);
    final RepairRunner repairRunner = new RepairRunner(storage, 1, new JmxConnectionFactory() {
      @Override
      public JmxProxy create(Optional<RepairStatusHandler> handler, String host)
          throws ReaperException {
        return jmx;
      }
    });

    assertEquals(storage.getRepairSegment(1).getState(), RepairSegment.State.NOT_STARTED);
    assertEquals(0, repairAttempts.get());
    repairRunner.run();
    assertEquals(1, repairAttempts.get());
    assertEquals(storage.getRepairSegment(1).getState(), RepairSegment.State.RUNNING);
    repairRunner.handle(repairAttempts.get(), ActiveRepairService.Status.STARTED,
        "Repair " + repairAttempts + " started");
    assertEquals(DateTime.now(), storage.getRepairSegment(1).getStartTime());
    assertEquals(RepairRun.RunState.RUNNING, storage.getRepairRun(1).getRunState());

    Thread.sleep(1500);
    assertEquals(2, repairAttempts.get());
    assertEquals(storage.getRepairSegment(1).getState(), RepairSegment.State.RUNNING);

    DateTimeUtils.setCurrentMillisFixed(TIME_RERUN);
    repairRunner.handle(repairAttempts.get(), ActiveRepairService.Status.STARTED,
        "Repair " + repairAttempts + " started");
    assertEquals(DateTime.now(), storage.getRepairSegment(1).getStartTime());
    assertEquals(RepairRun.RunState.RUNNING, storage.getRepairRun(1).getRunState());

    repairRunner.handle(repairAttempts.get(), ActiveRepairService.Status.FINISHED,
        "Repair " + repairAttempts + " finished");
    Thread.sleep(100);
    assertEquals(RepairRun.RunState.DONE, storage.getRepairRun(1).getRunState());
  }


  @Test
  public void testAlreadyStartedRepair() {
    // TODO
  }
}

@@ -1728,39 +1728,145 @@
     
-RepairRun.RunState.NOT_STARTED,
+                                                 RepairRun.RunState.NOT_STARTED,%0A                                                        
  Dat
@@ -4336,25 +4336,16 @@
 class)))
-%0A        
 .thenRet
@@ -4348,16 +4348,25 @@
 nReturn(
+%0A        
 Lists.ne
@@ -4386,69 +4386,116 @@
 ));%0A
-%0A
     
-final AtomicInteger repairAttempts = new AtomicInteger(0
+when(jmx.switchNode(Matchers.%3COptional%3CRepairStatusHandler%3E%3Eany(), anyString()))%0A        .thenReturn(jmx
 );%0A 
@@ -4584,32 +4584,38 @@
 ), anyString()))
+.then(
 %0A        .then(n
@@ -4607,22 +4607,16 @@
         
-.then(
 new Answ
@@ -4829,433 +4829,56 @@
 nner
-.initializeThreadPool(1, 1);%0A    final RepairRunner repairRunner = new RepairRunner(storage, 1, new JmxConnectionFactory() %7B%0A      @Override%0A      public JmxProxy create(Optional%3CRepairStatusHandler%3E handler, String host)%0A          throws ReaperException %7B%0A        return jmx;%0A      %7D%0A    %7D);%0A%0A    assertEquals(storage.getRepairSegment(1).getState(), RepairSegment.State.NOT_STARTED);%0A    assertEquals(0, repairAttempts.get()
+ repairRunner = new RepairRunner(storage, 1, jmx
 );%0A 

Complete hanging repair test case