/*
 * 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.beam.it.splunk;

import static com.google.common.truth.Truth.assertThat;
import static org.junit.Assert.assertThrows;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;

import com.splunk.Job;
import com.splunk.ServiceArgs;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.util.Collections;
import org.apache.beam.sdk.io.splunk.SplunkEvent;
import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.impl.client.CloseableHttpClient;
import org.awaitility.core.ConditionTimeoutException;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;
import org.mockito.Answers;
import org.mockito.Mock;
import org.mockito.junit.MockitoJUnit;
import org.mockito.junit.MockitoRule;
import org.testcontainers.images.builder.Transferable;

/** Unit tests for {@link SplunkResourceManager}. */
@RunWith(JUnit4.class)
public class SplunkResourceManagerTest {

  @Rule public final MockitoRule mockito = MockitoJUnit.rule();

  @Mock(answer = Answers.RETURNS_DEEP_STUBS)
  private SplunkClientFactory clientFactory;

  @Mock private SplunkContainer container;

  private static final String TEST_ID = "test-id";
  private static final String HOST = "localhost";
  private static final String HEC_SCHEMA = "http";
  private static final String HEC_TOKEN = "token";
  private static final String QUERY = "query";
  private static final String EVENT = "myEvent";
  private static final int DEFAULT_SPLUNK_HEC_INTERNAL_PORT = 8088;
  private static final int MAPPED_SPLUNK_HEC_INTERNAL_PORT = 50000;
  private static final int DEFAULT_SPLUNKD_INTERNAL_PORT = 8089;
  private static final int MAPPED_SPLUNKD_INTERNAL_PORT = 50001;

  private SplunkResourceManager testManager;

  @Before
  public void setUp() {
    when(container.withDefaultsFile(any(Transferable.class))).thenReturn(container);
    when(container.withPassword(anyString())).thenReturn(container);
    when(container.getMappedPort(DEFAULT_SPLUNKD_INTERNAL_PORT))
        .thenReturn(MAPPED_SPLUNKD_INTERNAL_PORT);
    doReturn(container).when(container).withLogConsumer(any());

    testManager =
        new SplunkResourceManager(clientFactory, container, SplunkResourceManager.builder(TEST_ID));
  }

  @Test
  public void testCreateResourceManagerBuilderReturnsSplunkResourceManager() {
    assertThat(
            SplunkResourceManager.builder(TEST_ID)
                .setHecPort(DEFAULT_SPLUNK_HEC_INTERNAL_PORT)
                .setSplunkdPort(DEFAULT_SPLUNKD_INTERNAL_PORT)
                .setHost(HOST)
                .useStaticContainer()
                .build())
        .isInstanceOf(SplunkResourceManager.class);
  }

  @Test
  public void testCreateResourceManagerThrowsCustomPortErrorWhenUsingStaticContainer() {
    assertThat(
            assertThrows(
                    SplunkResourceManagerException.class,
                    () ->
                        SplunkResourceManager.builder(TEST_ID)
                            .setHost(HOST)
                            .useStaticContainer()
                            .build())
                .getMessage())
        .containsMatch("the hecPort and splunkdPort were not properly set");
  }

  @Test
  public void testGetHttpEndpointReturnsCorrectValue() {
    when(container.getMappedPort(DEFAULT_SPLUNK_HEC_INTERNAL_PORT))
        .thenReturn(MAPPED_SPLUNK_HEC_INTERNAL_PORT);
    assertThat(
            new SplunkResourceManager(
                    clientFactory, container, SplunkResourceManager.builder(TEST_ID))
                .getHttpEndpoint())
        .isEqualTo(String.format("%s://%s:%d", HEC_SCHEMA, HOST, MAPPED_SPLUNK_HEC_INTERNAL_PORT));
  }

  @Test
  public void testGetHecEndpointReturnsCorrectValue() {
    when(container.getMappedPort(DEFAULT_SPLUNK_HEC_INTERNAL_PORT))
        .thenReturn(MAPPED_SPLUNK_HEC_INTERNAL_PORT);
    assertThat(
            new SplunkResourceManager(
                    clientFactory, container, SplunkResourceManager.builder(TEST_ID))
                .getHecEndpoint())
        .isEqualTo(
            String.format(
                "%s://%s:%d/services/collector/event",
                HEC_SCHEMA, HOST, MAPPED_SPLUNK_HEC_INTERNAL_PORT));
  }

  @Test
  public void testGetHecTokenReturnsCorrectValueWhenSet() {
    assertThat(
            new SplunkResourceManager(
                    clientFactory,
                    container,
                    SplunkResourceManager.builder(TEST_ID).setHecToken(HEC_TOKEN))
                .getHecToken())
        .isEqualTo(HEC_TOKEN);
  }

  @Test
  public void testSendHttpEventsShouldThrowErrorWhenHttpClientFailsToExecuteRequest()
      throws IOException {
    SplunkEvent event = SplunkEvent.newBuilder().withEvent(EVENT).create();

    CloseableHttpClient mockHttpClient = clientFactory.getHttpClient();
    doThrow(IOException.class).when(mockHttpClient).execute(any(HttpPost.class));

    assertThrows(SplunkResourceManagerException.class, () -> testManager.sendHttpEvent(event));
  }

  @Test
  public void testSendHttpEventsShouldThrowErrorWhenHttpClientReturnsErrorCode()
      throws IOException {
    SplunkEvent event = SplunkEvent.newBuilder().withEvent(EVENT).create();

    try (CloseableHttpResponse mockResponse =
        clientFactory.getHttpClient().execute(any(HttpPost.class))) {
      when(mockResponse.getStatusLine().getStatusCode()).thenReturn(404);
    }

    assertThrows(SplunkResourceManagerException.class, () -> testManager.sendHttpEvent(event));
  }

  @Test
  public void testSendHttpEventsShouldReturnTrueIfSplunkDoesNotThrowAnyError() throws IOException {
    SplunkEvent event = SplunkEvent.newBuilder().withEvent(EVENT).create();

    try (CloseableHttpResponse mockResponse =
        clientFactory.getHttpClient().execute(any(HttpPost.class))) {
      when(mockResponse.getStatusLine().getStatusCode()).thenReturn(200);
    }

    assertThat(testManager.sendHttpEvents(ImmutableList.of(event, event))).isTrue();
    verify(clientFactory.getHttpClient()).execute(any(HttpPost.class));
  }

  @Test
  public void testGetEventsShouldThrowErrorWhenServiceClientFailsToExecuteRequest() {
    Job mockJob =
        clientFactory.getServiceClient(any(ServiceArgs.class)).getJobs().create(anyString());
    doThrow(ConditionTimeoutException.class).when(mockJob).isDone();

    assertThrows(ConditionTimeoutException.class, () -> testManager.getEvents(QUERY));
  }

  @Test
  public void testGetEventsShouldThrowErrorWhenXmlReaderFailsToParseResponse() {
    Job mockJob =
        clientFactory.getServiceClient(any(ServiceArgs.class)).getJobs().create(anyString());

    when(mockJob.isDone()).thenReturn(true);
    when(mockJob.getEvents())
        .thenReturn(
            new InputStream() {
              @Override
              public int read() throws IOException {
                throw new IOException();
              }
            });

    assertThrows(SplunkResourceManagerException.class, () -> testManager.getEvents(QUERY));
  }

  @Test
  public void testGetEventsShouldReturnTrueIfSplunkDoesNotThrowAnyError() {
    Job mockJob =
        clientFactory.getServiceClient(any(ServiceArgs.class)).getJobs().create(anyString());
    String rawEvent =
        "<results preview='0'>"
            + "<meta><fieldOrder>"
            + "<field>_raw</field><field>_sourcetype</field><field>_time</field>"
            + "<field>host</field><field>index</field><field>source</field>"
            + "</fieldOrder></meta>"
            + "<result offset='0'>"
            + "<field k='_raw'><v xml:space='preserve' trunc='0'>myEvent</v></field>"
            + "<field k='_sourcetype'><value><text>mySourceType</text></value></field>"
            + "<field k='_time'><value><text>1970-01-01T00:00:00.123+00:00</text></value></field>"
            + "<field k='host'><value><text>myHost</text></value></field>"
            + "<field k='index'><value><text>myIndex</text></value></field>"
            + "<field k='source'><value><text>mySource</text></value></field>"
            + "</result></results>";
    InputStream inputStream = new ByteArrayInputStream(rawEvent.getBytes(StandardCharsets.UTF_8));
    SplunkEvent splunkEvent =
        SplunkEvent.newBuilder()
            .withEvent("myEvent")
            .withHost("myHost")
            .withSource("mySource")
            .withSourceType("mySourceType")
            .withIndex("myIndex")
            .withTime(123L)
            .create();

    when(mockJob.isDone()).thenReturn(true);
    when(mockJob.getEvents()).thenReturn(inputStream);

    assertThat(testManager.getEvents())
        .containsExactlyElementsIn(Collections.singletonList(splunkEvent));
  }
}
