package org.zalando.nakadi.controller;

import com.codahale.metrics.MetricRegistry;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.json.JSONArray;
import org.junit.Test;
import org.mockito.Mockito;
import org.springframework.http.converter.StringHttpMessageConverter;
import org.springframework.http.converter.json.MappingJackson2HttpMessageConverter;
import org.springframework.test.web.servlet.MockMvc;
import org.springframework.test.web.servlet.ResultActions;
import org.springframework.test.web.servlet.request.MockHttpServletRequestBuilder;
import org.zalando.nakadi.config.JsonConfig;
import org.zalando.nakadi.domain.BatchItemResponse;
import org.zalando.nakadi.domain.EventPublishResult;
import org.zalando.nakadi.domain.EventPublishingStep;
import org.zalando.nakadi.exceptions.InternalNakadiException;
import org.zalando.nakadi.exceptions.NakadiException;
import org.zalando.nakadi.exceptions.NoSuchEventTypeException;
import org.zalando.nakadi.metrics.EventTypeMetricRegistry;
import org.zalando.nakadi.metrics.EventTypeMetrics;
import org.zalando.nakadi.service.EventPublisher;
import org.zalando.nakadi.utils.JsonTestHelper;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutionException;

import static org.hamcrest.CoreMatchers.equalTo;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.eq;
import static org.mockito.Mockito.mock;
import static org.springframework.http.MediaType.APPLICATION_JSON;
import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post;
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.content;
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
import static org.springframework.test.web.servlet.setup.MockMvcBuilders.standaloneSetup;
import static org.zalando.nakadi.domain.EventPublishingStatus.ABORTED;
import static org.zalando.nakadi.domain.EventPublishingStatus.FAILED;
import static org.zalando.nakadi.domain.EventPublishingStatus.SUBMITTED;

public class EventPublishingControllerTest {

    public static final String TOPIC = "my-topic";
    private static final String EVENT_BATCH = "[{\"payload\": \"My Event Payload\"}]";

    private final ObjectMapper objectMapper = new JsonConfig().jacksonObjectMapper();
    private final MetricRegistry metricRegistry;
    private final JsonTestHelper jsonHelper;
    private final EventPublisher publisher;

    private final MockMvc mockMvc;
    private final EventTypeMetricRegistry eventTypeMetricRegistry;

    public EventPublishingControllerTest() throws NakadiException, ExecutionException {
        jsonHelper = new JsonTestHelper(objectMapper);
        metricRegistry = new MetricRegistry();
        publisher = mock(EventPublisher.class);
        eventTypeMetricRegistry = new EventTypeMetricRegistry(metricRegistry);

        final EventPublishingController controller = new EventPublishingController(publisher, eventTypeMetricRegistry);

        final MappingJackson2HttpMessageConverter jackson2HttpMessageConverter = new MappingJackson2HttpMessageConverter(objectMapper);
        mockMvc = standaloneSetup(controller)
                .setMessageConverters(new StringHttpMessageConverter(), jackson2HttpMessageConverter)
                .build();
    }

    @Test
    public void whenResultIsSubmittedThen200() throws Exception {
        final EventPublishResult result = new EventPublishResult(SUBMITTED, null, null);

        Mockito
                .doReturn(result)
                .when(publisher)
                .publish(any(JSONArray.class), eq(TOPIC));

        postBatch(TOPIC, EVENT_BATCH)
                .andExpect(status().isOk())
                .andExpect(content().string(""));
    }

    @Test
    public void whenInvalidPostBodyThen400() throws Exception {
        final String expectedPayload = "{\"type\":\"http://httpstatus.es/400\",\"title\":\"Bad Request\",\"status\":400," +
                "\"detail\":\"A JSONArray text must start with '[' at 1 [character 2 line 1]\"}";
        postBatch(TOPIC, "invalid json array").andExpect(status().isBadRequest()).andExpect(content().string(expectedPayload));
    }

    @Test
    public void whenResultIsAbortedThen422() throws Exception {
        final EventPublishResult result = new EventPublishResult(ABORTED, EventPublishingStep.PARTITIONING, responses());

        Mockito
                .doReturn(result)
                .when(publisher)
                .publish(any(JSONArray.class), eq(TOPIC));

        postBatch(TOPIC, EVENT_BATCH)
                .andExpect(status().isUnprocessableEntity())
                .andExpect(content().string(jsonHelper.matchesObject(responses())));
    }

    @Test
    public void whenResultIsAbortedThen207() throws Exception {
        final EventPublishResult result = new EventPublishResult(FAILED, EventPublishingStep.PUBLISHING, responses());

        Mockito
                .doReturn(result)
                .when(publisher)
                .publish(any(JSONArray.class), eq(TOPIC));

        postBatch(TOPIC, EVENT_BATCH)
                .andExpect(status().isMultiStatus())
                .andExpect(content().string(jsonHelper.matchesObject(responses())));
    }

    @Test
    public void whenEventTypeNotFoundThen404() throws Exception {
        Mockito
                .doThrow(NoSuchEventTypeException.class)
                .when(publisher)
                .publish(any(JSONArray.class), eq(TOPIC));

        postBatch(TOPIC, EVENT_BATCH)
                .andExpect(content().contentType("application/problem+json"))
                .andExpect(status().isNotFound());
    }

    @Test
    public void publishedEventsAreReportedPerEventType() throws Exception {
        final EventPublishResult success = new EventPublishResult(SUBMITTED, null, null);
        Mockito
                .doReturn(success)
                .doReturn(success)
                .doThrow(InternalNakadiException.class)
                .when(publisher)
                .publish(any(), any());

        postBatch(TOPIC, EVENT_BATCH);
        postBatch(TOPIC, EVENT_BATCH);
        postBatch(TOPIC, EVENT_BATCH);

        final EventTypeMetrics eventTypeMetrics = eventTypeMetricRegistry.metricsFor(TOPIC);

        assertThat(eventTypeMetrics.getResponseCount(200), equalTo(2L));
        assertThat(eventTypeMetrics.getResponseCount(500), equalTo(1L));
    }

    private List<BatchItemResponse> responses() {
        final BatchItemResponse response = new BatchItemResponse();
        response.setPublishingStatus(ABORTED);
        response.setStep(EventPublishingStep.VALIDATING);

        final List<BatchItemResponse> responses = new ArrayList<>();
        responses.add(response);

        return responses;
    }

    private ResultActions postBatch(final String eventType, final String batch) throws Exception {
        final String url = "/event-types/" + eventType + "/events";
        final MockHttpServletRequestBuilder requestBuilder = post(url)
                .contentType(APPLICATION_JSON)
                .content(batch);

        return mockMvc.perform(requestBuilder);
    }
}
@@ -710,62 +710,8 @@
 lt;%0A
-import org.zalando.nakadi.domain.EventPublishingStep;%0A
 impo
@@ -760,32 +760,32 @@
 akadiException;%0A
+
 import org.zalan
@@ -2051,16 +2051,234 @@
 BMITTED;
+%0Aimport static org.zalando.nakadi.domain.EventPublishingStep.PARTITIONING;%0Aimport static org.zalando.nakadi.domain.EventPublishingStep.PUBLISHING;%0Aimport static org.zalando.nakadi.domain.EventPublishingStep.VALIDATING;
 %0A%0Apublic
@@ -4567,36 +4567,16 @@
 BORTED, 
-EventPublishingStep.
 PARTITIO
@@ -5077,36 +5077,16 @@
 FAILED, 
-EventPublishingStep.
 PUBLISHI
@@ -6768,16 +6768,16 @@
 ORTED);%0A
+
         
@@ -6793,36 +6793,16 @@
 setStep(
-EventPublishingStep.
 VALIDATI

add static import