package com.adven.concordion.extensions.exam.kafka;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.internals.RecordHeader;
import org.apache.kafka.common.utils.Bytes;
import org.junit.Before;
import org.junit.Test;

import java.io.UnsupportedEncodingException;
import java.nio.charset.Charset;
import java.util.Properties;

import static com.adven.concordion.extensions.exam.RandomUtils.anyString;
import static org.assertj.core.api.Assertions.assertThat;


public class DefaultEventConsumerTest {

    private DefaultEventConsumer consumer;

    @Before
    public void setUp() {
        consumer = new DefaultEventConsumer(1000L, new Properties());
    }

    @Test
    public void toEvent() throws UnsupportedEncodingException {
        final String topic = anyString();
        final String key = anyString();
        final String message = anyString();
        final Bytes messageInBytes = new Bytes(message.getBytes(Charset.forName("UTF-8")));
        final ConsumerRecord<String, Bytes> record = new ConsumerRecord<>(topic, 1, 1L, key, messageInBytes);
        final Event<Bytes> event = consumer.toEvent(record);
        assertThat(event.getKey()).isEqualTo(key);
        assertThat(event.getTopicName()).isEqualTo(topic);
        assertThat(event.getMessage()).isEqualTo(Bytes.wrap(message.getBytes("UTF-8")));
    }

    @Test(expected = NullPointerException.class)
    public void testToEventWithNullRecord() {
        consumer.toEvent(null);
    }

    @Test(expected = NullPointerException.class)
    public void testConsumeWithNullTopic() {
        consumer.consume(null);
    }

    @Test(expected = NullPointerException.class)
    public void testCreateInstanceWithNullProperty() {
        new DefaultEventConsumer(10L, (Properties) null);
    }

    @Test(expected = NullPointerException.class)
    public void testEventHeaderWithNull() {
        consumer.eventHeader(null);
    }

    @Test
    public void testHeaderToString() throws UnsupportedEncodingException {
        final byte[] value = anyString().getBytes("UTF-8");
        final Header header = new RecordHeader(anyString(), value);
        final byte[] result = consumer.headerToString(header);
        assertThat(result).isEqualTo(value);
    }

    @Test
    public void testHeaderToStringWithNullHeader() {
        final byte[] result = consumer.headerToString(null);
        assertThat(result).isEqualTo(new byte[]{});
    }

}
@@ -717,15 +717,8 @@
 mer(
-1000L, 
 new 
@@ -729,16 +729,23 @@
 erties()
+, 1000L
 );%0A    %7D
@@ -1823,13 +1823,8 @@
 mer(
-10L, 
 (Pro
@@ -1832,24 +1832,29 @@
 erties) null
+, 10L
 );%0A    %7D%0A%0A  

fix build