#558 Fluent and type-safe testing DSL for kstreamplify-core-test

README.md

@@ -32,6 +32,7 @@ Kstreamplify adds extra features to Kafka Streams, simplifying development so yo
   * [Java](#java)
   * [Unit Test](#unit-test)
     * [Override Properties](#override-properties)
+    * [Fluent Testing DSL](#fluent-testing-dsl)
 * [Avro Serializer and Deserializer](#avro-serializer-and-deserializer)
 * [Error Handling](#error-handling)
   * [Set up DLQ Topic](#set-up-dlq-topic)
@@ -269,6 +270,143 @@ public class MyKafkaStreamsTest extends KafkaStreamsStarterTest {
 }
 ```
 
+#### Fluent Testing DSL
+
+`KafkaStreamsStarterTest` also provides a fluent, type-safe `Given → When → Then` DSL through the `test()` method. It
+is built on top of the same Topology Test Driver infrastructure and reuses your `TopicWithSerde` declarations, so you no
+longer need to create `TestInputTopic`/`TestOutputTopic` instances or read records manually.
+
+The following test pipes a record and asserts the output:
+
+```java
+@Test
+void shouldUpperCase() {
+    test()
+        .given(inputTopic)
+        .record("1", inputValue)
+        .when()
+        .then(outputTopic)
+        .hasExactly(1)
+        .containsKey("1")
+        .satisfies(record -> assertEquals(expectedOutputValue, record.value()));
+}
+```
+
+Feed multiple input topics (useful for joins and enrichment) with `and(...)`, and pipe several records or a bulk
+collection:
+
+```java
+test()
+    .given(inputTopic)
+    .record("1", firstInputValue)
+    .record("2", secondInputValue)
+    .records(List.of(KeyValue.pair("3", thirdInputValue)))
+    .and(secondInputTopic)
+    .record("1", otherInputValue)
+    .when()
+    .then(outputTopic)
+    .hasExactly(3);
+```
+
+The available output assertions are `hasExactly(...)`, `isEmpty()`, `containsKey(...)`, `doesNotContainKey(...)`,
+`containsRecord(...)`, `containsValue(...)`, `containsHeader(...)`, `containsExactly(...)`, `satisfies(...)`,
+`satisfies(index, ...)` and `allSatisfy(...)`:
+
+```java
+test()
+    .given(inputTopic)
+    .record("1", inputValue)
+    .when()
+    .then(outputTopic)
+    .containsExactly(Map.entry("1", expectedOutputValue))
+    .containsValue(expectedOutputValue::equals)
+    .containsHeader("correlation-id", "CORRELATION-1")
+    .allSatisfy(record -> assertNotNull(record.value()));
+```
+
+Assert dead letter queue (DLQ) records without reading the DLQ topic yourself. The internal `KafkaError`
+representation is hidden behind `DlqRecord`:
+
+```java
+test()
+    .given(inputTopic)
+    .record("1", invalidInputValue)
+    .when()
+    .thenDlq()
+    .hasExactly(1)
+    .containsKey("1")
+    .containsError(IllegalStateException.class, "Invalid value")
+    .containsHeader("correlation-id", "CORRELATION-1")
+    .satisfies(error -> {
+        assertEquals("1", error.key());
+        assertEquals("java.lang.IllegalStateException", error.exceptionTypeName());
+        assertEquals("Invalid value", error.errorMessage());
+    });
+```
+
+Assert the content of a state store:
+
+```java
+test()
+    .given(inputTopic)
+    .record("1", inputValue)
+    .when()
+    .thenStateStore("my-store")
+    .hasExactly(1)
+    .contains("1", inputValue)
+    .doesNotContainKey("999")
+    .containsValue(expectedStoreValue::equals);
+```
+
+A single chain can assert several output topics, the DLQ and the state stores, and can even feed additional records
+with `andGiven(...)`:
+
+```java
+test()
+    .given(inputTopic)
+    .record("1", inputValue)
+    .and(secondInputTopic)
+    .record("1", otherInputValue)
+    .when()
+    .then(outputTopic)
+    .containsRecord("1", expectedOutputValue)
+    .andDlq()
+    .isEmpty()
+    .andStateStore("my-store")
+    .containsKey("1");
+```
+
+For windowed and time-dependent topologies, control event time and wall clock time. As defined by Kafka Streams,
+`advanceTime(...)` only moves the event time used by the records piped afterwards, while `advanceWallClockTime(...)`
+only triggers the wall-clock-time punctuators:
+
+```java
+test()
+    .given(inputTopic)
+    .record("1", firstInputValue, Instant.parse("2026-08-17T10:00:00Z"))
+    .advanceTime(Duration.ofMinutes(5))
+    .record("2", secondInputValue)
+    .when()
+    .advanceWallClockTime(Duration.ofSeconds(10))
+    .then(outputTopic)
+    .containsKey("1");
+```
+
+A record piped without an explicit timestamp uses the current event time, which starts at `getInitialWallClockTime()`
+and moves with `advanceTime(...)` and with the last explicitly timestamped record.
+
+When an assertion fails, the DSL reports the topic, the expectation and the records actually produced:
+
+```
+Expected 2 record(s) on topic 'output_topic' but found 1.
+Actual records:
+  1=EXPECTED_OUTPUT_VALUE
+```
+
+The DSL is an additive layer: existing tests keep working and advanced scenarios can still access the underlying
+`TopologyTestDriver` through `driver()`, which is available on every stage. The same context is used for the whole test
+method, so the records read by `then(...)` and `thenDlq()` are accumulated and can be asserted several times.
+
 ## Avro Serializer and Deserializer
 
 When working with Avro schemas, you can use the `SerdesUtils` class to easily serialize or deserialize records:

kstreamplify-core-test/src/main/java/com/michelin/kstreamplify/KafkaStreamsStarterTest.java

@@ -28,6 +28,7 @@
 import com.michelin.kstreamplify.initializer.KafkaStreamsStarter;
 import com.michelin.kstreamplify.serde.SerdesUtils;
 import com.michelin.kstreamplify.serde.TopicWithSerde;
+import com.michelin.kstreamplify.test.KstreamplifyTestContext;
 import io.confluent.kafka.schemaregistry.testutil.MockSchemaRegistry;
 import java.io.IOException;
 import java.nio.file.Files;
@@ -57,6 +58,8 @@ public abstract class KafkaStreamsStarterTest {
     /** The DLQ topic. */
     protected TestOutputTopic<String, KafkaError> dlqTopic;
 
+    private KstreamplifyTestContext testContext;
+
     /** Constructor. */
     protected KafkaStreamsStarterTest() {}
 
@@ -84,6 +87,8 @@ void generalSetUp() {
                 KafkaStreamsExecutionContext.getDlqTopicName(),
                 new StringDeserializer(),
                 SerdesUtils.<KafkaError>getValueSerdes().deserializer());
+
+        testContext = new KstreamplifyTestContext(testDriver, dlqTopic, getInitialWallClockTime());
     }
 
     /**
@@ -133,13 +138,25 @@ protected Map<String, String> getSpecificProperties() {
         return Collections.emptyMap();
     }
 
+    /**
+     * Start a fluent {@code Given → When → Then} test scenario on the topology under test. The same context is returned
+     * for the whole test method, so the event time advancement and the records already read are preserved across the
+     * calls.
+     *
+     * @return The {@link KstreamplifyTestContext} bound to the current test driver
+     */
+    protected KstreamplifyTestContext test() {
+        return testContext;
+    }
+
     /**
      * Close everything after each test.
      *
      * @throws IOException If an I/O error occurs while deleting the state directory
      */
     @AfterEach
     protected void generalTearDown() throws IOException {
+        testContext = null;
         testDriver.close();
         Files.deleteIfExists(
                 Path.of(KafkaStreamsExecutionContext.getProperties().getProperty(STATE_DIR_CONFIG)));

kstreamplify-core-test/src/main/java/com/michelin/kstreamplify/test/AssertionStage.java

@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package com.michelin.kstreamplify.test;
+
+import com.michelin.kstreamplify.serde.TopicWithSerde;
+import org.apache.kafka.streams.TopologyTestDriver;
+
+/**
+ * Base class of all the {@code then} stages of the testing DSL. It keeps a reference to the parent
+ * {@link KstreamplifyTestContext} so that a single fluent chain can assert on several output topics, on the dead letter
+ * queue and on state stores, or feed additional records.
+ */
+public abstract class AssertionStage {
+    private final KstreamplifyTestContext context;
+
+    /**
+     * Constructor.
+     *
+     * @param context The parent test context
+     */
+    AssertionStage(KstreamplifyTestContext context) {
+        this.context = context;
+    }
+
+    /**
+     * Continue the chain with typed assertions on another output topic.
+     *
+     * @param topic The output topic to assert on
+     * @param <K> The type of the key
+     * @param <V> The type of the value
+     * @return An {@link OutputAssertion} holding the records produced on the provided topic
+     */
+    public <K, V> OutputAssertion<K, V> and(TopicWithSerde<K, V> topic) {
+        return context.then(topic);
+    }
+
+    /**
+     * Continue the chain with assertions on the records sent to the dead letter queue.
+     *
+     * @return A {@link DlqAssertion} holding the DLQ records
+     */
+    public DlqAssertion andDlq() {
+        return context.thenDlq();
+    }
+
+    /**
+     * Continue the chain with assertions on the content of the provided key-value state store.
+     *
+     * @param storeName The name of the state store
+     * @param <K> The type of the key
+     * @param <V> The type of the value
+     * @return A {@link StateStoreAssertion} bound to the state store
+     */
+    public <K, V> StateStoreAssertion<K, V> andStateStore(String storeName) {
+        return context.thenStateStore(storeName);
+    }
+
+    /**
+     * Continue the chain by feeding additional records to an input topic.
+     *
+     * @param topic The input topic to feed
+     * @param <K> The type of the key
+     * @param <V> The type of the value
+     * @return A {@link GivenStage} bound to the provided topic
+     */
+    public <K, V> GivenStage<K, V> andGiven(TopicWithSerde<K, V> topic) {
+        return context.given(topic);
+    }
+
+    /**
+     * Expose the underlying topology test driver as an escape hatch for advanced tests.
+     *
+     * @return The underlying topology test driver
+     */
+    public TopologyTestDriver driver() {
+        return context.driver();
+    }
+}