/**
 * 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 org.apache.metron.rest.service.impl;

import com.google.common.collect.ImmutableList;
import org.apache.metron.rest.model.*;
import org.apache.metron.rest.service.StormStatusService;
import org.hamcrest.CoreMatchers;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.Mock;
import org.mockito.MockitoAnnotations;

import java.util.List;

import static org.hamcrest.MatcherAssert.assertThat;
import static org.mockito.Mockito.*;

public class CachedStormStatusServiceImplTest {

  @Mock
  private StormStatusService stormService;
  private CachedStormStatusServiceImpl cachedStormStatusService;

  @BeforeEach
  public void setup() {
    MockitoAnnotations.initMocks(this);
    cachedStormStatusService = new CachedStormStatusServiceImpl(stormService, 150, 30);
  }

  @Test
  public void caches_supervisor_summary() {
    SupervisorStatus supervisorStatus1 = new SupervisorStatus();
    SupervisorStatus supervisorStatus2 = new SupervisorStatus();
    SupervisorSummary supervisorSummary = new SupervisorSummary(
        new SupervisorStatus[]{supervisorStatus1, supervisorStatus2});
    when(stormService.getSupervisorSummary()).thenReturn(supervisorSummary);
    // should hit the cache
    for (int i = 0; i < 100; i++) {
      cachedStormStatusService.getSupervisorSummary();
    }
    SupervisorSummary summary = cachedStormStatusService.getSupervisorSummary();
    assertThat("Number of supervisors did not match.", summary.getSupervisors().length,
        CoreMatchers.equalTo(2));
    verify(stormService, times(1)).getSupervisorSummary();
    cachedStormStatusService.reset();
    summary = cachedStormStatusService.getSupervisorSummary();
    assertThat("Number of supervisors did not match.", summary.getSupervisors().length,
        CoreMatchers.equalTo(2));
    verify(stormService, times(2)).getSupervisorSummary();
  }

  @Test
  public void caches_topology_summary() {
    TopologyStatus topologyStatus1 = new TopologyStatus();
    TopologyStatus topologyStatus2 = new TopologyStatus();
    TopologySummary topologySummary = new TopologySummary(
        new TopologyStatus[]{topologyStatus1, topologyStatus2});
    when(stormService.getTopologySummary()).thenReturn(topologySummary);
    // should hit the cache
    for (int i = 0; i < 100; i++) {
      cachedStormStatusService.getTopologySummary();
    }
    TopologySummary summary = cachedStormStatusService.getTopologySummary();
    assertThat("Number of topologies did not match.", summary.getTopologies().length,
        CoreMatchers.equalTo(2));
    verify(stormService, times(1)).getTopologySummary();
    cachedStormStatusService.reset();
    summary = cachedStormStatusService.getTopologySummary();
    assertThat("Number of topologies did not match.", summary.getTopologies().length,
        CoreMatchers.equalTo(2));
    verify(stormService, times(2)).getTopologySummary();
  }

  @Test
  public void caches_topology_status_by_name() {
    String topologyName1 = "topology-1";
    String topologyName2 = "topology-2";
    TopologyStatus topologyStatus1 = new TopologyStatus();
    topologyStatus1.setName(topologyName1);
    TopologyStatus topologyStatus2 = new TopologyStatus();
    topologyStatus2.setName(topologyName2);
    when(stormService.getTopologyStatus(topologyName1)).thenReturn(topologyStatus1);
    when(stormService.getTopologyStatus(topologyName2)).thenReturn(topologyStatus2);
    // should hit the cache
    for (int i = 0; i < 100; i++) {
      cachedStormStatusService.getTopologyStatus(topologyName1);
      cachedStormStatusService.getTopologyStatus(topologyName2);
    }
    TopologyStatus status1 = cachedStormStatusService.getTopologyStatus(topologyName1);
    TopologyStatus status2 = cachedStormStatusService.getTopologyStatus(topologyName2);
    assertThat("Name did not match for topology 1.", status1.getName(),
        CoreMatchers.equalTo(topologyName1));
    assertThat("Name did not match for topology 2.", status2.getName(),
        CoreMatchers.equalTo(topologyName2));
    verify(stormService, times(1)).getTopologyStatus(topologyName1);
    verify(stormService, times(1)).getTopologyStatus(topologyName2);
    cachedStormStatusService.reset();
    cachedStormStatusService.getTopologyStatus(topologyName1);
    cachedStormStatusService.getTopologyStatus(topologyName2);
    verify(stormService, times(2)).getTopologyStatus(topologyName1);
    verify(stormService, times(2)).getTopologyStatus(topologyName2);
  }

  @Test
  public void caches_all_topology_status() {
    TopologyStatus topologyStatus1 = new TopologyStatus();
    TopologyStatus topologyStatus2 = new TopologyStatus();
    List<TopologyStatus> allTopologyStatus = ImmutableList.of(topologyStatus1, topologyStatus2);
    when(stormService.getAllTopologyStatus()).thenReturn(allTopologyStatus);
    // should hit the cache
    for (int i = 0; i < 100; i++) {
      cachedStormStatusService.getAllTopologyStatus();
    }
    List<TopologyStatus> allStatus = cachedStormStatusService.getAllTopologyStatus();
    assertThat("Number of topologies returned by all topology status check did not match.",
        allStatus.size(), CoreMatchers.equalTo(2));
    verify(stormService, times(1)).getAllTopologyStatus();
    cachedStormStatusService.reset();
    cachedStormStatusService.getAllTopologyStatus();
    verify(stormService, times(2)).getAllTopologyStatus();
  }

  @Test
  public void admin_functions_act_as_simple_passthroughs_to_storm_service() {
    TopologyResponse topologyResponse = new TopologyResponse();
    when(stormService.activateTopology(anyString())).thenReturn(topologyResponse);
    when(stormService.deactivateTopology(anyString())).thenReturn(topologyResponse);
    for (int i = 0; i < 100; i++) {
      cachedStormStatusService.activateTopology("foo");
      cachedStormStatusService.deactivateTopology("foo");
    }
    verify(stormService, times(100)).activateTopology(anyString());
    verify(stormService, times(100)).deactivateTopology(anyString());
  }
}