Add tests for stream samples in dev guide

dataflow-website/stream-developer-guides/streams/standalone-stream-kafka/usage-cost-logger/src/test/java/io/spring/dataflow/sample/usagecostlogger/UsageCostLoggerApplicationTests.java

@@ -1,16 +1,53 @@
 package io.spring.dataflow.sample.usagecostlogger;
 
+import io.spring.dataflow.sample.UsageCostDetail;
 import org.junit.Test;
 import org.junit.runner.RunWith;
+import org.mockito.ArgumentCaptor;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
 import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.cloud.stream.annotation.EnableBinding;
+import org.springframework.cloud.stream.messaging.Sink;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Primary;
+import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.test.context.junit4.SpringRunner;
 
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+
 @RunWith(SpringRunner.class)
-@SpringBootTest
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
 public class UsageCostLoggerApplicationTests {
 
+	@Autowired
+	protected Sink sink;
+
+	@Autowired
+	protected UsageCostLogger usageCostLogger;
+
 	@Test
 	public void contextLoads() {
 	}
 
+	@Test
+	public void testUsageCostLogger() throws Exception {
+		ArgumentCaptor<UsageCostDetail> captor = ArgumentCaptor.forClass(UsageCostDetail.class);
+		this.sink.input().send(MessageBuilder.withPayload("{\"userId\":\"user3\",\"callCost\":10.100000000000001,\"dataCost\":25.1}").build());
+		verify(this.usageCostLogger).process(captor.capture());
+	}
+
+	@EnableAutoConfiguration
+	@EnableBinding(Sink.class)
+	static class TestConfig {
+
+		// Override `UsageCostLogger` bean for spying.
+		@Bean
+		@Primary
+		public UsageCostLogger usageCostLogger() {
+			return spy(new UsageCostLogger());
+		}
+	}
 }

dataflow-website/stream-developer-guides/streams/standalone-stream-kafka/usage-detail-sender/src/main/java/io/spring/dataflow/sample/usagedetailsender/UsageDetailSender.java

@@ -18,7 +18,7 @@ public class UsageDetailSender {
 	@Autowired
 	private Source source;
 
-	private String[] users = {"Glenn", "Sabby", "Mark", "Janne", "Ilaya"};
+	private String[] users = {"user1", "user2", "user3", "user4", "user5"};
 
 	@Scheduled(fixedDelay = 1000)
 	public void sendEvents() {

dataflow-website/stream-developer-guides/streams/standalone-stream-kafka/usage-detail-sender/src/test/java/io/spring/dataflow/sample/usagedetailsender/UsageDetailSenderApplicationTests.java

@@ -1,16 +1,44 @@
 package io.spring.dataflow.sample.usagedetailsender;
 
+import java.util.concurrent.TimeUnit;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.spring.dataflow.sample.UsageDetail;
+import org.json.JSONObject;
 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.cloud.stream.messaging.Source;
+import org.springframework.cloud.stream.test.binder.MessageCollector;
+import org.springframework.messaging.Message;
 import org.springframework.test.context.junit4.SpringRunner;
+import org.springframework.util.Assert;
+
+import static org.junit.Assert.assertTrue;
 
 @RunWith(SpringRunner.class)
-@SpringBootTest
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
 public class UsageDetailSenderApplicationTests {
 
+	@Autowired
+	private MessageCollector messageCollector;
+
+	@Autowired
+	private Source source;
+
 	@Test
 	public void contextLoads() {
 	}
 
+	@Test
+	public void testUsageDetailSender() throws Exception {
+		Message message = this.messageCollector.forChannel(this.source.output()).poll(1, TimeUnit.SECONDS);
+		String usageDetailJSON = message.getPayload().toString();
+		assertTrue(usageDetailJSON.contains("userId"));
+		assertTrue(usageDetailJSON.contains("duration"));
+		assertTrue(usageDetailJSON.contains("data"));
+	}
+
 }

dataflow-website/stream-developer-guides/streams/standalone-stream-rabbitmq/usage-cost-logger/src/test/java/io/spring/dataflow/sample/usagecostlogger/UsageCostLoggerApplicationTests.java

@@ -1,16 +1,53 @@
 package io.spring.dataflow.sample.usagecostlogger;
 
+import io.spring.dataflow.sample.UsageCostDetail;
 import org.junit.Test;
 import org.junit.runner.RunWith;
+import org.mockito.ArgumentCaptor;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
 import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.cloud.stream.annotation.EnableBinding;
+import org.springframework.cloud.stream.messaging.Sink;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Primary;
+import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.test.context.junit4.SpringRunner;
 
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+
 @RunWith(SpringRunner.class)
-@SpringBootTest
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
 public class UsageCostLoggerApplicationTests {
 
+	@Autowired
+	protected Sink sink;
+
+	@Autowired
+	protected UsageCostLogger usageCostLogger;
+
 	@Test
 	public void contextLoads() {
 	}
 
+	@Test
+	public void testUsageCostLogger() throws Exception {
+		ArgumentCaptor<UsageCostDetail> captor = ArgumentCaptor.forClass(UsageCostDetail.class);
+		this.sink.input().send(MessageBuilder.withPayload("{\"userId\":\"user3\",\"callCost\":10.100000000000001,\"dataCost\":25.1}").build());
+		verify(this.usageCostLogger).process(captor.capture());
+	}
+
+	@EnableAutoConfiguration
+	@EnableBinding(Sink.class)
+	static class TestConfig {
+
+		// Override `UsageCostLogger` bean for spying.
+		@Bean
+		@Primary
+		public UsageCostLogger usageCostLogger() {
+			return spy(new UsageCostLogger());
+		}
+	}
 }

dataflow-website/stream-developer-guides/streams/standalone-stream-rabbitmq/usage-cost-processor/src/test/java/io/spring/dataflow/sample/usagecostprocessor/UsageCostProcessorApplicationTests.java

@@ -1,16 +1,39 @@
 package io.spring.dataflow.sample.usagecostprocessor;
 
+import java.util.concurrent.TimeUnit;
+
 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.cloud.stream.messaging.Processor;
+import org.springframework.cloud.stream.test.binder.MessageCollector;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.test.context.junit4.SpringRunner;
 
+import static org.junit.Assert.assertTrue;
+
 @RunWith(SpringRunner.class)
 @SpringBootTest
 public class UsageCostProcessorApplicationTests {
 
+	@Autowired
+	private Processor processor;
+
+	@Autowired
+	private MessageCollector messageCollector;
+
 	@Test
 	public void contextLoads() {
 	}
 
+	@Test
+	public void testUsageCostProcessor() throws Exception {
+		this.processor.input().send(MessageBuilder.withPayload("{\"userId\":\"user3\",\"duration\":101,\"data\":502}").build());
+		Message message = this.messageCollector.forChannel(this.processor.output()).poll(1, TimeUnit.SECONDS);
+		assertTrue(message.getPayload().toString().equals("{\"userId\":\"user3\",\"callCost\":10.100000000000001,\"dataCost\":25.1}"));
+	}
+
 }

dataflow-website/stream-developer-guides/streams/standalone-stream-rabbitmq/usage-detail-sender/src/main/java/io/spring/dataflow/sample/usagedetailsender/UsageDetailSender.java

@@ -18,7 +18,7 @@ public class UsageDetailSender {
 	@Autowired
 	private Source source;
 
-	private String[] users = {"Glenn", "Sabby", "Mark", "Janne", "Ilaya"};
+	private String[] users = {"user1", "user2", "user3", "user4", "user5"};
 
 	@Scheduled(fixedDelay = 1000)
 	public void sendEvents() {

dataflow-website/stream-developer-guides/streams/standalone-stream-rabbitmq/usage-detail-sender/src/test/java/io/spring/dataflow/sample/usagedetailsender/UsageDetailSenderApplicationTests.java

@@ -1,16 +1,44 @@
 package io.spring.dataflow.sample.usagedetailsender;
 
+import java.util.concurrent.TimeUnit;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.spring.dataflow.sample.UsageDetail;
+import org.json.JSONObject;
 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.cloud.stream.messaging.Source;
+import org.springframework.cloud.stream.test.binder.MessageCollector;
+import org.springframework.messaging.Message;
 import org.springframework.test.context.junit4.SpringRunner;
+import org.springframework.util.Assert;
+
+import static org.junit.Assert.assertTrue;
 
 @RunWith(SpringRunner.class)
-@SpringBootTest
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
 public class UsageDetailSenderApplicationTests {
 
+	@Autowired
+	private MessageCollector messageCollector;
+
+	@Autowired
+	private Source source;
+
 	@Test
 	public void contextLoads() {
 	}
 
+	@Test
+	public void testUsageDetailSender() throws Exception {
+		Message message = this.messageCollector.forChannel(this.source.output()).poll(1, TimeUnit.SECONDS);
+		String usageDetailJSON = message.getPayload().toString();
+		assertTrue(usageDetailJSON.contains("userId"));
+		assertTrue(usageDetailJSON.contains("duration"));
+		assertTrue(usageDetailJSON.contains("data"));
+	}
+
 }