/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you 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 storm.starter.tools;

import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;

import java.util.Map;

import static org.fest.assertions.api.Assertions.assertThat;

public class SlotBasedCounterTest {

  private static final int ANY_NUM_SLOTS = 1;
  private static final int ANY_SLOT = 0;
  private static final Object ANY_OBJECT = "ANY_OBJECT";

  @DataProvider
  public Object[][] illegalNumSlotsData() {
    return new Object[][]{ { -10 }, { -3 }, { -2 }, { -1 }, { 0 } };
  }

  @Test(expectedExceptions = IllegalArgumentException.class, dataProvider = "illegalNumSlotsData")
  public void negativeOrZeroNumSlotsShouldThrowIAE(int numSlots) {
    new SlotBasedCounter<Object>(numSlots);
  }

  @DataProvider
  public Object[][] legalNumSlotsData() {
    return new Object[][]{ { 1 }, { 2 }, { 3 }, { 20 } };
  }

  @Test(dataProvider = "legalNumSlotsData")
  public void positiveNumSlotsShouldBeOk(int numSlots) {
    new SlotBasedCounter<Object>(numSlots);
  }

  @Test
  public void newInstanceShouldHaveEmptyCounts() {
    // given
    SlotBasedCounter<Object> counter = new SlotBasedCounter<Object>(ANY_NUM_SLOTS);

    // when
    Map<Object, Long> counts = counter.getCounts();

    // then
    assertThat(counts).isEmpty();
  }

  @Test
  public void shouldReturnNonEmptyCountsWhenAtLeastOneObjectWasCounted() {
    // given
    SlotBasedCounter<Object> counter = new SlotBasedCounter<Object>(ANY_NUM_SLOTS);
    counter.incrementCount(ANY_OBJECT, ANY_SLOT);

    // when
    Map<Object, Long> counts = counter.getCounts();

    // then
    assertThat(counts).isNotEmpty();

    // additional tests that go beyond what this test is primarily about
    assertThat(counts.size()).isEqualTo(1);
    assertThat(counts.get(ANY_OBJECT)).isEqualTo(1);
  }

  @DataProvider
  public Object[][] incrementCountData() {
    return new Object[][]{ { new String[]{ "foo", "bar" }, new int[]{ 3, 2 } } };
  }

  @Test(dataProvider = "incrementCountData")
  public void shouldIncrementCount(Object[] objects, int[] expCounts) {
    // given
    SlotBasedCounter<Object> counter = new SlotBasedCounter<Object>(ANY_NUM_SLOTS);

    // when
    for (int i = 0; i < objects.length; i++) {
      Object obj = objects[i];
      int numIncrements = expCounts[i];
      for (int j = 0; j < numIncrements; j++) {
        counter.incrementCount(obj, ANY_SLOT);
      }
    }

    // then
    for (int i = 0; i < objects.length; i++) {
      assertThat(counter.getCount(objects[i], ANY_SLOT)).isEqualTo(expCounts[i]);
    }
    assertThat(counter.getCount("nonexistentObject", ANY_SLOT)).isEqualTo(0);
  }

  @Test
  public void shouldReturnZeroForNonexistentObject() {
    // given
    SlotBasedCounter<Object> counter = new SlotBasedCounter<Object>(ANY_NUM_SLOTS);

    // when
    counter.incrementCount("somethingElse", ANY_SLOT);

    // then
    assertThat(counter.getCount("nonexistentObject", ANY_SLOT)).isEqualTo(0);
  }

  @Test
  public void shouldIncrementCountOnlyOneSlotAtATime() {
    // given
    int numSlots = 3;
    Object obj = Long.valueOf(10);
    SlotBasedCounter<Object> counter = new SlotBasedCounter<Object>(numSlots);

    // when (empty)
    // then
    assertThat(counter.getCount(obj, 0)).isEqualTo(0);
    assertThat(counter.getCount(obj, 1)).isEqualTo(0);
    assertThat(counter.getCount(obj, 2)).isEqualTo(0);

    // when
    counter.incrementCount(obj, 1);

    // then
    assertThat(counter.getCount(obj, 0)).isEqualTo(0);
    assertThat(counter.getCount(obj, 1)).isEqualTo(1);
    assertThat(counter.getCount(obj, 2)).isEqualTo(0);
  }

  @Test
  public void wipeSlotShouldSetAllCountsInSlotToZero() {
    // given
    SlotBasedCounter<Object> counter = new SlotBasedCounter<Object>(ANY_NUM_SLOTS);
    Object countWasOne = "countWasOne";
    Object countWasThree = "countWasThree";
    counter.incrementCount(countWasOne, ANY_SLOT);
    counter.incrementCount(countWasThree, ANY_SLOT);
    counter.incrementCount(countWasThree, ANY_SLOT);
    counter.incrementCount(countWasThree, ANY_SLOT);

    // when
    counter.wipeSlot(ANY_SLOT);

    // then
    assertThat(counter.getCount(countWasOne, ANY_SLOT)).isEqualTo(0);
    assertThat(counter.getCount(countWasThree, ANY_SLOT)).isEqualTo(0);
  }

  @Test
  public void wipeZerosShouldRemoveAnyObjectsWithZeroTotalCount() {
    // given
    SlotBasedCounter<Object> counter = new SlotBasedCounter<Object>(2);
    int wipeSlot = 0;
    int otherSlot = 1;
    Object willBeRemoved = "willBeRemoved";
    Object willContinueToBeTracked = "willContinueToBeTracked";
    counter.incrementCount(willBeRemoved, wipeSlot);
    counter.incrementCount(willContinueToBeTracked, wipeSlot);
    counter.incrementCount(willContinueToBeTracked, otherSlot);

    // when
    counter.wipeSlot(wipeSlot);
    counter.wipeZeros();

    // then
    assertThat(counter.getCounts()).doesNotContainKey(willBeRemoved);
    assertThat(counter.getCounts()).containsKey(willContinueToBeTracked);
  }
}