package net.mh.kafkabrowser;

import net.mh.kafkabrowser.model.BrowserConsumer;
import net.mh.kafkabrowser.model.Topic;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.test.web.client.TestRestTemplate;
import org.springframework.hateoas.Link;
import org.springframework.http.*;
import org.springframework.kafka.test.utils.KafkaTestUtils;
import org.springframework.test.context.ActiveProfiles;
import org.springframework.test.context.junit4.SpringRunner;

import java.time.Duration;

import static com.github.grantwest.eventually.EventuallyLambdaMatcher.eventuallyEval;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.*;

/**
 * Created by markus on 08.04.17.
 */
@RunWith(SpringRunner.class)
@ActiveProfiles(profiles = {"test"})
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
public class SmallTopicIntegrationTest extends AbstractKafkaIntegrationTest {

    @Autowired
    private TestRestTemplate restTemplate;

    @Before
    public void setUp() throws Exception {
        KafkaProducer<Integer, String> kafkaProducer = new KafkaProducer<>(KafkaTestUtils.producerProps(embeddedKafka));
        for (Integer i = 0; i < 10; i++) {
            kafkaProducer.send(new ProducerRecord<>("test", i, "test" + i));
        }
    }

    @Test
    public void testTopics() {
        HttpHeaders headers = new HttpHeaders();
        headers.setContentType(MediaType.APPLICATION_JSON_UTF8);
        ResponseEntity<BrowserConsumer> browserConsumer = restTemplate.postForEntity("/consumer", new HttpEntity<>("{\"keyDeserializer\": \"org.apache.kafka.common.serialization.IntegerDeserializer\", \"valueDeserializer\": \"org.apache.kafka.common.serialization.StringDeserializer\"}", headers), BrowserConsumer.class);
        assertThat(browserConsumer.getStatusCode(), equalTo(HttpStatus.OK));

        BrowserConsumer body = browserConsumer.getBody();

        Link topics = body.getLink("topics");
        assertThat(topics, notNullValue());
        assertThat(topics.getHref(), endsWith("test"));

        assertThat(() -> restTemplate.getForEntity(String.format("/consumer/%s/topic/test", body.getConsumerId()), Topic.class).getBody().getCurrentRecords(), eventuallyEval(hasSize(10), Duration.ofSeconds(10)));

        ResponseEntity<Topic> topic = restTemplate.getForEntity(String.format("/consumer/%s/topic/test", body.getConsumerId()), Topic.class);
        Topic topicBody = topic.getBody();
        assertThat(topicBody.hasLink("next"), is(false));
        assertThat(topicBody.hasLink("previous"), is(false));
    }
}

@@ -230,33 +230,8 @@
 rd;%0A
-import org.junit.Before;%0A
 impo
@@ -1258,321 +1258,12 @@
 te;%0A
-%0A
     
-@Before%0A    public void setUp() throws Exception %7B%0A        KafkaProducer%3CInteger, String%3E kafkaProducer = new KafkaProducer%3C%3E(KafkaTestUtils.producerProps(embeddedKafka));%0A        for (Integer i = 0; i %3C 10; i++) %7B%0A            kafkaProducer.send(new ProducerRecord%3C%3E(%22test%22, i, %22test%22 + i));%0A        %7D%0A    %7D%0A
 %0A   
@@ -1295,16 +1295,26 @@
 stTopics
+_EmptyPage
 () %7B%0A   
@@ -2042,24 +2042,37 @@
     
-assertThat(() -%3E
+ResponseEntity%3CTopic%3E topic =
  res
@@ -2174,95 +2174,596 @@
 ass)
-.getBody().getCurrentRecords(), eventuallyEval(hasSize(10), Duration.ofSeconds(10))
+;%0A%0A        Topic topicBody = topic.getBody();%0A        assertThat(topicBody.hasLink(%22next%22), is(false));%0A        assertThat(topicBody.hasLink(%22previous%22), is(false));%0A    %7D%0A%0A    @Test%0A    public void testTopics_OnePage() %7B%0A        KafkaProducer%3CInteger, String%3E kafkaProducer = new KafkaProducer%3C%3E(KafkaTestUtils.producerProps(embeddedKafka));%0A        for (Integer i = 0; i %3C 10; i++) %7B%0A            kafkaProducer.send(new ProducerRecord%3C%3E(%22test%22, i, %22test%22 + i));%0A        %7D%0A%0A        HttpHeaders headers = new HttpHeaders();%0A        headers.setContentType(MediaType.APPLICATION_JSON_UTF8
 );%0A
-%0A
     
@@ -2777,36 +2777,56 @@
 ponseEntity%3C
-Topic%3E topic
+BrowserConsumer%3E browserConsumer
  = restTempl
@@ -2821,34 +2821,35 @@
  = restTemplate.
-ge
+pos
 tForEntity(Strin
@@ -2847,245 +2847,553 @@
 ity(
-String.format(%22/consumer/%25s/topic/test%22, body.getConsumerId()), Topic.class);%0A        Topic topicBody = topic.getBody();%0A        assertThat(topicBody.hasLink(%22nex
+%22/consumer%22, new HttpEntity%3C%3E(%22%7B%5C%22keyDeserializer%5C%22: %5C%22org.apache.kafka.common.serialization.IntegerDeserializer%5C%22, %5C%22valueDeserializer%5C%22: %5C%22org.apache.kafka.common.serialization.StringDeserializer%5C%22%7D%22, headers), BrowserConsumer.class);%0A        assertThat(browserConsumer.getStatusCode(), equalTo(HttpStatus.OK));%0A%0A        assertThat(() -%3E restTemplate.getForEntity(String.format(%22/consumer/%25s/topic/tes
 t%22
-)
 , 
-is(false));%0A        assertThat(topicBody.hasLink(%22previous%22), is(false
+browserConsumer.getBody().getConsumerId()), Topic.class).getBody().getCurrentRecords(), eventuallyEval(hasSize(10), Duration.ofSeconds(10)
 ));%0A

split up tests