package com.github.cosmo0920.fluentd.flume.plugins;

import java.nio.charset.StandardCharsets;

import org.apache.flume.Sink.Status;
import org.apache.flume.Channel;
import org.apache.flume.Event;
import org.apache.flume.Transaction;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.event.EventBuilder;

import org.komamitsu.fluency.Fluency;

import static org.junit.Assert.*;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.ArgumentCaptor;
import org.mockito.Matchers;
import static org.mockito.Mockito.*;

import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Map;
import java.util.HashMap;

public class TestFluentdSink {
	private static final Logger logger = LoggerFactory.getLogger(FluentdSink.class);

	@Test
	public void testShouldCommitASuccessfulTransaction() throws EventDeliveryException {
		Channel mockChannel = mock(Channel.class);
		Transaction mockTransaction = mock(Transaction.class);
		Event mockEvent = mock(Event.class);
		FluencyPublisher mockFluencyPublisher = mock(FluencyPublisher.class);

		when(mockChannel.getTransaction()).thenReturn(mockTransaction);
		when(mockChannel.take()).thenReturn(mockEvent);
		when(mockEvent.getBody()).thenReturn("test body".getBytes(StandardCharsets.UTF_8));

		FluentdSink sink = new FluentdSink();
		sink.setChannel(mockChannel);
		sink.publisher = mockFluencyPublisher;
		assertNotNull(sink.publisher);

		Status status = sink.process();
		assertEquals(status, Status.READY);

		verify(mockTransaction, times(1)).begin();
		verify(mockTransaction, times(1)).commit();
		verify(mockTransaction, times(1)).close();
	}

	@Test
	public void testShouldRollbackAnUnsuccessfulTransaction() throws EventDeliveryException, IOException {
		Channel mockChannel = mock(Channel.class);
		Transaction mockTransaction = mock(Transaction.class);
		Event mockEvent = mock(Event.class);
		FluencyPublisher mockFluencyPublisher = mock(FluencyPublisher.class);

		when(mockChannel.getTransaction()).thenReturn(mockTransaction);
		when(mockChannel.take()).thenReturn(mockEvent);
		when(mockEvent.getBody()).thenReturn("test body".getBytes(StandardCharsets.UTF_8));

		FluentdSink sink = new FluentdSink();
		sink.setChannel(mockChannel);
		sink.publisher = mockFluencyPublisher;
		assertNotNull(sink.publisher);
		doThrow(new IOException()).when(mockFluencyPublisher).send(mockEvent);

		Status status = sink.process();
		assertEquals(status, Status.BACKOFF);

		verify(mockTransaction, times(1)).begin();
		verify(mockTransaction, times(1)).rollback();
		verify(mockTransaction, times(1)).close();
	}

	@Test
	public void testShouldRollbackWhenThereAreNoEventsInChannel() throws EventDeliveryException, IOException {
		Channel mockChannel = mock(Channel.class);
		Transaction mockTransaction = mock(Transaction.class);
		FluencyPublisher mockFluencyPublisher = mock(FluencyPublisher.class);

		when(mockChannel.getTransaction()).thenReturn(mockTransaction);
		when(mockChannel.take()).thenReturn(null);

		FluentdSink sink = new FluentdSink();
		sink.setChannel(mockChannel);
		sink.publisher = mockFluencyPublisher;
		assertNotNull(sink.publisher);

		Status status = sink.process();
		assertEquals(status, Status.BACKOFF);

		verify(mockTransaction, times(1)).begin();
		verify(mockTransaction, times(1)).rollback();
		verify(mockTransaction, times(1)).close();
	}
}

@@ -159,16 +159,58 @@
 hannel;%0A
+import org.apache.flume.ChannelException;%0A
 import o
@@ -3428,32 +3428,832 @@
 atus.BACKOFF);%0A%0A
+%09%09verify(mockTransaction, times(1)).begin();%0A%09%09verify(mockTransaction, times(1)).rollback();%0A%09%09verify(mockTransaction, times(1)).close();%0A%09%7D%0A%0A%09@Test%0A%09public void testShouldRollbackWhenUnableToGetEventsFromChannel() throws EventDeliveryException, IOException %7B%0A%09%09Channel mockChannel = mock(Channel.class);%0A%09%09Transaction mockTransaction = mock(Transaction.class);%0A%09%09FluencyPublisher mockFluencyPublisher = mock(FluencyPublisher.class);%0A%0A%09%09when(mockChannel.getTransaction()).thenReturn(mockTransaction);%0A%09%09doThrow(new ChannelException(%22Mock Exception%22)).when(mockChannel).take();%0A%0A%09%09FluentdSink sink = new FluentdSink();%0A%09%09sink.setChannel(mockChannel);%0A%09%09sink.publisher = mockFluencyPublisher;%0A%09%09assertNotNull(sink.publisher);%0A%0A%09%09Status status = sink.process();%0A%09%09assertEquals(status, Status.BACKOFF);%0A%0A
 %09%09verify(mockTra

Add test case for unable to get Channel circumstance