# Stream Applications Test Support

This module contains common components to support testing of stream applications.


## Stream Application Integration Testing

The `integration` package contains components supporting integration testing stream apps using https://www.testcontainers.org/[TestContainers].

### StreamAppContainer

An extension of https://www.testcontainers.org/features/creating_container/[GenericContainer].
This is the base class for testing Spring Cloud Stream Application docker images.

Currently, this supports apps built for Kafka or RabbitMQ message brokers using `KafkaStreamAppContainer` or `RabbitMQStreamAppContainer`.
This extension sets the binding properties for the standard `input` and `output` destinations.
If your app uses function binding, you must configure the https://cloud.spring.io/spring-cloud-static/spring-cloud-stream/current/reference/html/spring-cloud-stream.html#_functional_binding_names[function destinations to map to input and output].
Note that the pre-packaged stream applications include the required mappings already.

NOTE: Multiple input and output ports are currently not supported.

These containers depend on an already running message broker TestContainer.
Typically, they are created in static initializers, so all tests running within a JVM use the same message broker instance.

For example:

```java
final static Network network = Network.SHARED;

	protected final static KafkaContainer kafka = new KafkaContainer(
			DockerImageName.parse("confluentinc/cp-kafka:5.5.1"))
					.withExposedPorts(9092, 9093)
					.withNetwork(network);

	static {
		kafka.start();
	}

@Container
static StreamAppContainer timeSource = new KafkaStreamAppContainer("springcloudstream/time-source-kafka:3.0.0-SNAPSHOT", kafka);
```

The StreamAppContainer uses the shared network and connects to the message broker using `kafka.getNetworkAliases().get(0) + ":9092"`, in this case.

Alternately,

```java
InetAddress.getLocalHost().getHostAddress() + ":" + kafka.getMappedPort(9092)
```

will work for both the containers, and the host machine.

The host machine can connect to Kafka using the random mapped port:

```java
"localhost:" + kafka.getMappedPort(9092);
```

These containers are intended to work with `OutputMatchers`, described below.

### Stream Application Container Test Support

This `integration` package provides test annotations for Junit 5 `@KafkaStreamAppTest` and `RabbitMQStreamAppTest`.
These meta-annotations include required Spring configuration, and `@TestContainers`.

The test strategy is to run a single container, and use the message broker directly to publish messages to the input and
verify the expected output message.  Simply use the provided `KafkaTemplate` or `RabbitTemplate` to publish input messages to
a `source` or `processor`.

To send messages to the container's input destination(s) autowire `TestTopicSender`.
This wraps `RabbitTemplate` or `KafkaTempate`, respectively.
You may also use the templates directly.

``` java
@ Autowired
private TestTopicSender
```

### Verifying output messages.

This framework creates a message listener on a known output topic.
The StreamAppContainer are statically configured to output to that topic, so the listener receives all output messages.
Configure your test using one of the following methods;

```java
protected <P> Callable<Boolean> payloadMatches(Predicate<P>... payloadMatchers);
Callable<Boolean> messageMatches(Predicate<Message<?>>... messageMatchers);
```
These are `Callable<Boolean>` which happens to be the type accepted by the https://github.com/awaitility/awaitility[awaitility]
```java
await().until()
```
method, which is a convenient way to wait for the output message.
When the message arrives, the link:src/main/java/org/springframework/cloud/stream/app/test/integration/TestTopicListener.java[TestTopicListener] implementation for the message broker will test all registered predicates.

For convenience, you can autowire `OutputMatcher` which wraps the TestTopicListener, and exposes `Callable<Boolean>` methods.

let's look at simple example test for the famous `time-source`:

```java
@KafkaStreamAppTest
public class KafkaTimeSourceTests {

	// "MM/dd/yy HH:mm:ss";
	private final static Pattern pattern = Pattern.compile(".*\\d{2}/\\d{2}/\\d{2}\\s+\\d{2}:\\d{2}:\\d{2}");

	static LogMatcher logMatcher = LogMatcher.contains("Started TimeSource");

    @Autowired
    private OutputMatcher outputMatcher;

	@Container
	static StreamAppContainer source = new KafkaStreamAppContainer(StreamAppContainerTestUtils
			.imageName(StreamAppContainerTestUtils.SPRINGCLOUDSTREAM_REPOSITOTRY, "time-source-kafka", VERSION));
	@Test
	void test() {
		await().atMost(DEFAULT_DURATION).until(logMatcher.matches());
		await().atMost(DEFAULT_DURATION).until(outputMatcher.payloadMatches((String s) -> pattern.matcher(s).matches()));
	}
}
```

We start an ApplicationContext `KafkaStreamAppContainerTestConfiguration` which starts the `KafkaConfig.kafka` TestContainer in a static initializer.
The Time Source emits the time every second. In this case, it's hard to know what the expected output payload is, but it should at least match the date pattern.
Then we wait for a message on the output topic that matches the pattern.


NOTE: Timing concerns: The `payloadMatches` is called repeatedly by awaitility. The first time, it registers the MessageMacher with the message listener.
Subsequently, The TopicTestListener detects that it has already been registered, so it just checks if the predicate is satisfied.
Potential, there can be a race condition if the message is consumed before the MessageMacher is invoked the first time.
To address this, the `KafkaTestListener` rewinds the topic to offset `0` each time a new MessageMacher is registered.
RabbitMQ doesn't have this replay capability. Rabbit is no longer responsible once the consumer acknowledges the message.
To work around this, the  `RabbitMQTestListener` maintains a cache of any unverified messages for a few minutes.
If a MessageMacher has not been satisfied, the test listener checks the cache to see if any of those messages match.
The following test case verifies the expected behavior.

```java
@Test
void verifierOnTheFlyOutOfOrder() {
    testTopicSender.send(STREAM_APPLICATIONS_TEST_TOPIC, "hello test1");
    testTopicSender.send(STREAM_APPLICATIONS_TEST_TOPIC, "hello test2");
    await().atMost(Duration.ofSeconds(30))
        .until(outputMatcher.payloadMatches(s -> s.equals("hello test2"), s -> s.equals("hello test1")));
	}
```
The `hello test1` MessageMatcher did not exist when `hello test1` was consumed, and is rejected by the first MessageMatcher,
so it is cached and tested when the second MessageMatcher is created.

If you need to, you can register MessageMatchers in advance, in a `@BeforeEach` method if you `@Autowire` the OutputMatcher.
But this doesn't work for statically declared containers which are more efficient and common with TestContainers.

### Testing Stream Applications

The @link:src/main/java/org/springframework/cloud/stream/app/test/integration/StreamApps.java[StreamApps] component
is convenient for testing an entire stream.
This realizes the concepts of `source`, `processor`, and `sink` , and similar Spring Cloud Data Flow, wires them up behind the scenes.

Here is a test for the canonical `TikTok` stream:

```java
@RabbitMQStreamAppTest
public class RabbitTikTokTests {

	private static LogMatcher logMatcher = LogMatcher.matchesRegex(".*\\d{2}/\\d{2}/\\d{2}\\s+\\d{2}:\\d{2}:\\d{2}")
			.times(3);

	@Container
	private static final StreamApps streamApp = kafkaStreamApps(KafkaTikTokTests.class.getSimpleName(),
			KafkaConfig.kafka)
					.withSourceContainer(
							new RabbitMQStreamAppContainer(StreamAppContainerTestUtils.imageName(
									"time-source-rabbit",
									VERSION)))
					.withSinkContainer(
							new RabbitMQStreamAppContainer(StreamAppContainerTestUtils.imageName(
									"log-sink-rabbit",
									VERSION)).withLogConsumer(logMatcher)
											.log())
					.build();

	@Test
	void test() {
		await().atMost(DEFAULT_DURATION).until(logMatcher.matches());
	}
}
```

Here, the link:src/main/java/org/springframework/cloud/stream/app/test/integration/LogMatcher.java[LogMatcher].
This is an extension of TestContainer's `LogConsumer`. Here, we verify the LogSink logs at least 3 messages that match the pattern.

link:src/main/java/org/springframework/cloud/stream/app/test/integration/AppLog.java[AppLog] is also another useful LogConsumer
to enable container logging.

You can find many sample tests in https://github.com/spring-cloud/spring-cloud-stream-acceptance-tests/tree/master/stream-applications-integration-tests[].



























