This code snippet is sample test to explain the samza tests in detail

IntegrationTest.java

import com.linkedin.events.share.ShareEvent;
import com.linkedin.mirror.common.AvroUtils;
import com.linkedin.mirror.common.mock.event.ShareEventMock;
import com.linkedin.mirror.common.monitor.SamzaMetrics;
import com.linkedin.restli.client.RestClient;
import java.util.ArrayList;
import kafka.common.Config;
import org.apache.avro.generic.GenericRecord;
import org.junit.Test;
import org.mockito.Mock;
import org.mockito.MockitoAnnotations;
import org.testng.annotations.BeforeMethod;

import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;

public class IntegrationTest {
 @Mock
 private RestClient _mockRestClient;
 @Mock
 private SamzaMetrics _mockMetrics;

 private ShareInfoTestTask _shareInfoTestTask;
 private List<OutgoingMessageEnvelope> _shareInfoMessages = new ArrayList<>();


 // This lets us create a analytics task with a mocked rest client
 private class ShareInfoTestTask extends ShareInfoTask {
   RestClient _restClient;
   SamzaMetrics _metrics;
   public ShareInfoTestTask(RestClient restClient, SamzaMetrics metrics) {
     _restClient = restClient;
     _metrics = metrics;
   }
 }

 // Method for creating the config, can be overridden to add more values
 private Config createBaseConfig(String outputTopic) {
   Config mockConfig = mock(Config.class);
   when(mockConfig.get("task.inputs")).thenReturn("InputTopics");
   ...
   return mockConfig;
 }

 private void consumeEvent(ShareEvent event, String topic) {
   GenericRecord record = AvroUtils.specificRecordToGenericRecord(event, event.getSchema());
   IncomingMessageEnvelope message =
       new IncomingMessageEnvelope(new SystemStreamPartition("system", topic, null), null, null, record);
   _shareInfoTestTask.process(message, _mockTaskCoordinator);
 }

 private List<OutgoingMessageEnvelope> getShareInfoMessages(){
   return _shareInfoMessages;
 }

 @BeforeMethod
 public void setup(){
   //Mockito provided initialization method for mocks
   MockitoAnnotations.initMocks(this);
   //Create tasks with mock objects
   _shareInfoTestTask = new ShareInfoTestTask(_mockRestClient, _mockMetrics);
 }

 @Test
 public void testViewWithDecoration(){
   // Create our initial share
   ShareEvent shareEvent = mockShare(SHARE_ID_1, TIMESTAMP_1);
   consumeEvent(shareEvent,"test");

   assertEquals(getShareInfoMessages().size(), 1);

   // Verify we have one social gesture event that is a view with decorated actor data
   ShareSocialGestureEvent outputEvent = getShareInfoMessages().getMessage();
   assertEquals(outputEvent.viewCount, 1L);
 }
 
 //Mocked input event
 public ShareEvent mockShare(int shareId, long timeStamp){
   return new ShareEvent().setId(shareId).setTimeStamp(timeStamp); 
 }
}