diff --git a/docs/extend/architecture.mdx b/docs/extend/architecture.mdx index c44a675..9d34ef8 100644 --- a/docs/extend/architecture.mdx +++ b/docs/extend/architecture.mdx @@ -8,6 +8,8 @@ description: KTestify's three-layer architecture. Transport, Orchestration, and KTestify is built on a strict **three-layer separation of concerns**. Each layer has one job and knows nothing about the other layers' implementation details. +Since version `1.1.1`, the transport layer has **two sibling contracts** instead of one. `RecordFetcher` models a background stream you poll until a record appears (Kafka, Azure Blob). `RequestResponseClient` models a synchronous, caller-initiated call where you send a request right now and get an answer immediately (HTTP, gRPC, SOAP). Both return `List>`, so the orchestration and assertion layers below stay identical no matter which contract a given transport implements. + ```mermaid --- config: @@ -16,12 +18,14 @@ config: flowchart LR subgraph TRANSPORT A[KafkaRecordFetcher] + A2[RequestResponseClient<Req,V>] end subgraph ORCHESTRATION B[AbstractKafkaConsumer] C[RawKafkaConsumer] D[AvroKafkaConsumer] + B2[AbstractSynchronousConsumer] end subgraph ASSERTION @@ -29,27 +33,33 @@ flowchart LR F[FileRecordMatcher] G[XmlRecordMatcher] H[AvroFileRecordMatcher] + I[AttributeRecordMatcher] end A -->|List<ConsumedRecord>| B B --> C B --> D B -->|List<ConsumedRecord>| E + A2 -->|List<ConsumedRecord>| B2 + B2 -->|List<ConsumedRecord>| E E --> F E --> G E --> H + E --> I classDef transport fill:#E1F5EE,stroke:#0F6E56,color:#085041 classDef orchestration fill:#EEEDFE,stroke:#534AB7,color:#3C3489 classDef assertion fill:#FAC775,stroke:#854F0B,color:#412402 - class A transport - class B,C,D orchestration - class E,F,G,H assertion + class A,A2 transport + class B,C,D,B2 orchestration + class E,F,G,H,I assertion ``` **`ConsumedRecord` is the only data type that crosses layer boundaries.** +Synchronous transports additionally populate a new `attributes` map on `ConsumedRecord` (an HTTP status code, a future gRPC status code, an MQ reason code) for metadata that is not a protocol header. Asynchronous transports leave it empty. See [Core Concepts →](core-concepts) for the full field list. + --- ## Layer responsibilities @@ -72,6 +82,22 @@ Swapping Kafka for IBM MQ means writing a new `IbmMqRecordFetcher`, nothing e --- +### Transport, synchronous - `RequestResponseClient` + +Knows: how to send one request and turn the answer into `ConsumedRecord`. +Does NOT know: matchers, files, test frameworks, and does not block waiting for something to "appear". + +```java +public interface RequestResponseClient extends AutoCloseable { + List> execute(Req request) throws FetchException; + void close(); +} +``` + +This is the contract an HTTP client plugin implements. See [Synchronous Transports →](transports/synchronous-transports) for the full guide. + +--- + ### Orchestration - `AbstractKafkaConsumer` Knows: fetch → match → result wiring. @@ -94,6 +120,27 @@ try { --- +### Orchestration, synchronous - `AbstractSynchronousConsumer` + +Knows: buildRequest → execute → match → result wiring, for transports where the caller supplies the request explicitly instead of the fetcher blocking on a subscription. +Does NOT know: HTTP, gRPC, or any transport internals. + +```java +// AbstractSynchronousConsumer.call(), simplified +try { + Req request = buildRequest(); + List> records = client.execute(request); // transport + MatchResult result = matcher.match(records, buildMatchContext()); // assertion + return result.isPassed(); +} catch (FetchException e) { + throw new ConsumerException(e.getMessage()); +} +``` + +Unlike `AbstractKafkaConsumer`, the client is not closed in a `finally` block here. A `RequestResponseClient` is expected to be a longer lived, connection pooled client (like `java.net.http.HttpClient`), owned and closed by the plugin's shared scenario resources, not created and discarded per call. + +--- + ### Assertion - `RecordMatcher` Knows: `ConsumedRecord`, expected values, comparison algorithm. @@ -114,9 +161,12 @@ public interface RecordMatcher { ktestify-core ktestify-cucumber ─────────────────────────────── ────────────────────────────────── RecordFetcher BackgroundStepDefinition +RequestResponseClient ValidationStepDefinition KafkaRecordFetcher ◄──────── ConsumerContext (config only) -AbstractKafkaConsumer ValidationStepDefinition -RecordMatcher ConsumerValidationService +AbstractKafkaConsumer ConsumerValidationService +AbstractSynchronousConsumer +PollingRequestResponseClient +RecordMatcher MatchContext / MatchResult ConsumedRecord ``` @@ -138,21 +188,27 @@ flowchart TD C["AbstractKafkaConsumer<K, V>"] D["RawKafkaConsumer\nK=String, V=String"] E["AvroKafkaConsumer\nK=String, V=GenericRecord"] + C2["AbstractSynchronousConsumer<Req, V>"] A --> B --> C --> D C --> E + B --> C2 F(["RecordFetcher<V> · interface"]) G["KafkaRecordFetcher<K, V>"] F --> G + H(["RequestResponseClient<Req,V> · interface"]) + I["PollingRequestResponseClient<Req,V>"] + H --> I + classDef abstract fill:#EEEDFE,stroke:#534AB7,color:#3C3489 classDef concrete fill:#E1F5EE,stroke:#0F6E56,color:#085041 classDef iface fill:#FAC775,stroke:#854F0B,color:#412402 - class A,B,C abstract - class D,E,G concrete - class F iface + class A,B,C,C2 abstract + class D,E,G,I concrete + class F,H iface ``` @@ -167,6 +223,7 @@ RecordMatcher (@FunctionalInterface) ├── FieldsRecordMatcher ├── FileKeyRecordMatcher ├── KeyRecordMatcher +├── AttributeRecordMatcher ├── AvroFileRecordMatcher ├── AvroFileKeyRecordMatcher ├── AvroFieldsRecordMatcher @@ -179,5 +236,6 @@ RecordMatcher (@FunctionalInterface) - [Core concepts →](core-concepts) - [Transports, Kafka →](transports/kafka) - [Adding a transport →](transports/adding-a-transport) +- [Synchronous transports →](transports/synchronous-transports) - [Built-in matchers →](matchers/built-in-matchers) diff --git a/docs/extend/core-concepts.mdx b/docs/extend/core-concepts.mdx index 77fe153..de679e8 100644 --- a/docs/extend/core-concepts.mdx +++ b/docs/extend/core-concepts.mdx @@ -8,24 +8,29 @@ description: Key domain types in ktestify-core, ConsumedRecord, MatchContext, Ma ## `ConsumedRecord` -The **only** data type that crosses layer boundaries. It is the universal output of the transport layer and the universal input of the assertion layer. +The **only** data type that crosses layer boundaries. It is the universal output of the transport layer (`RecordFetcher` for asynchronous transports, `RequestResponseClient` for synchronous ones since `1.1.1`) and the universal input of the assertion layer. ```java @Value public class ConsumedRecord { - String source; // topic name + String source; // topic name, container name, request URL, ... int partition; long offset; String key; V value; // String for raw, GenericRecord for Avro - long timestamp; - Headers headers; + Instant timestamp; + Map headers; // protocol headers (Kafka headers, HTTP response headers, ...) + Map attributes; // NEW in 1.1.1, transport metadata, never null, defaults to emptyMap() static ConsumedRecord fromKafkaRecord(ConsumerRecord record) { ... } MatchedRecord toMatchedRecord() { ... } } ``` +`attributes` holds structured transport metadata that does not belong under `headers`, for example an HTTP status code and elapsed time, a future gRPC status code, or an MQ reason code. Kafka and Azure Blob leave it empty. A synchronous transport plugin populates it, and `AttributeRecordMatcher` (see [Built-in matchers →](matchers/built-in-matchers)) asserts against it. + +`ConsumedRecord` ships both a full constructor (accepting `attributes`) and a backward-compatible overload without it (defaults to `Collections.emptyMap()`), plus a `@Builder`, so existing transports keep compiling unchanged. + --- ## `MatchedRecord` @@ -38,7 +43,7 @@ Deduplication token, represents a record that has already been claimed by a cons ## `MatchContext` -Immutable context object passed to a `RecordMatcher`. Built by `AbstractKafkaConsumer.buildMatchContext()` from the `ConsumerContext`. +Immutable context object passed to a `RecordMatcher`. Built by `AbstractKafkaConsumer.buildMatchContext()` (or `AbstractSynchronousConsumer.buildMatchContext()` for synchronous transports) from the `ConsumerContext`. ```java @Value @Builder @@ -49,6 +54,7 @@ public class MatchContext { boolean strictMatching; String matchKey; String matchValue; + Map expectedAttributes; // NEW in 1.1.1, defaults to emptyMap(), used by AttributeRecordMatcher // Convenience, for single-record matchers public String getMatchFilePath() { @@ -58,6 +64,8 @@ public class MatchContext { } ``` +`expectedAttributes` holds the key/value pairs to assert against a record's `attributes` map, for example `{"statusCode": "200"}`. It follows the same convention as `excludedFields`, always non-null, defaults to an empty map, and matchers check `isEmpty()` rather than `null`. + --- ## `MatchResult` @@ -125,12 +133,30 @@ RecordMatcherFactory.forRaw("matchFile") → FileRecordMatcher RecordMatcherFactory.forAvro("matchFile") → AvroFileRecordMatcher RecordMatcherFactory.forRaw("matchXML") → XmlRecordMatcher RecordMatcherFactory.forAvro("matchXML") → throws ConsumerException ← not supported +RecordMatcherFactory.forRaw("methodMatchAttributes") → AttributeRecordMatcher<>() // NEW in 1.1.1, raw only ``` See [Built-in matchers →](matchers/built-in-matchers) for the full mapping table. --- +## Synchronous transports (`1.1.1`) + +Alongside `RecordFetcher` (async, poll until a record appears), `ktestify-core` now ships a sibling contract for synchronous, caller-initiated transports: + +```java +public interface RequestResponseClient extends AutoCloseable { + List> execute(Req request) throws FetchException; + void close(); +} +``` + +Its orchestration counterpart, `AbstractSynchronousConsumer`, mirrors `AbstractKafkaConsumer` (build a request, call `execute`, hand the result to a `RecordMatcher`, return `MatchResult.isPassed()`), and a generic decorator, `PollingRequestResponseClient`, adds retry-until-predicate-or-timeout semantics that any synchronous transport plugin can reuse instead of writing its own poll loop. + +Full details, including the client lifecycle and how to implement a new synchronous transport, live on the [Synchronous transports →](transports/synchronous-transports) page. + +--- + ## Dynamic variable system All file reads go through `FileUtils.getFileContent(path)`, which transparently calls `DynamicVariableProcessor.process(content)` before returning. To add a new variable type: diff --git a/docs/extend/matchers/built-in-matchers.mdx b/docs/extend/matchers/built-in-matchers.mdx index 8f032fa..caef821 100644 --- a/docs/extend/matchers/built-in-matchers.mdx +++ b/docs/extend/matchers/built-in-matchers.mdx @@ -1,12 +1,12 @@ --- sidebar_position: 1 title: Built-in Matchers -description: Reference for all 11 built-in RecordMatcher implementations and when to use each. +description: Reference for all built-in RecordMatcher implementations and when to use each. --- # Built-in Matchers -KTestify ships 11 `RecordMatcher` implementations, split into raw (String) and Avro (GenericRecord) variants. They are selected automatically by `RecordMatcherFactory` based on the `matchMethod` in `MatchContext`. +KTestify ships 12 `RecordMatcher` implementations, split into raw (String) and Avro (GenericRecord) variants. They are selected automatically by `RecordMatcherFactory` based on the `matchMethod` in `MatchContext`. --- @@ -20,6 +20,7 @@ KTestify ships 11 `RecordMatcher` implementations, split into raw (String) an | `methodMatchXML` | `XmlRecordMatcher` | ❌ `ConsumerException` | | `methodMatchXPath` | `XPathRecordMatcher` | ❌ `ConsumerException` | | `methodRecordKeyMatch` | `KeyRecordMatcher` | `AvroKeyRecordMatcher` | +| `methodMatchAttributes` | `AttributeRecordMatcher` | ❌ `ConsumerException` | | `null` / blank | `NoOpRecordMatcher` | `NoOpRecordMatcher` | ```java @@ -89,6 +90,18 @@ Always returns `MatchResult.pass()`. Used when `matchMethod` is `null` or blank, --- +### `AttributeRecordMatcher` (`1.1.1`) + +Generic, transport agnostic matcher introduced alongside `RequestResponseClient` for synchronous transports. Asserts one or more entries of `ConsumedRecord.getAttributes()` against `MatchContext.getExpectedAttributes()`. Never inspects `getValue()`, so it works for any `V`, not only `String`. + +- **MatchContext fields used:** `expectedAttributes` (map of key to expected value) +- **Matching rule:** every key in `expectedAttributes` must be present in the actual record's `attributes` with an exactly equal String value, only the first record in the list is used +- **On empty `expectedAttributes`:** returns `MatchResult.pass()` immediately, nothing to assert +- **Typical use:** HTTP status code assertions (`{"statusCode": "200"}`), reusable later by any future transport that populates `attributes` (gRPC status, MQ reason code, script exit code) +- **Not available for Avro**, `attributes` is populated by synchronous request/response clients, not Avro consumers + +--- + ## Avro matchers (V = GenericRecord) All Avro matchers first deserialise the `GenericRecord` to a JSON `Map` via `AvroUtils.toJsonMap()`, then delegate to comparison logic equivalent to their raw counterparts. @@ -125,4 +138,5 @@ Avro equivalent of `FieldsRecordMatcher`. Extracts a character range from the JS 2. `matchFilePaths` is always a `List`, single-record matchers use `get(0)`, batch matchers iterate by index. 3. XML/XPath matchers throw `ConsumerException` when called via `RecordMatcherFactory.forAvro()`. 4. `NoOpRecordMatcher` is the default when no match method is configured. +5. `expectedAttributes` defaults to `Collections.emptyMap()`, `AttributeRecordMatcher` checks `isEmpty()`, never `null`, same convention as `excludedFields`. diff --git a/docs/extend/transports/adding-a-transport.mdx b/docs/extend/transports/adding-a-transport.mdx index 9a61d33..899ad0b 100644 --- a/docs/extend/transports/adding-a-transport.mdx +++ b/docs/extend/transports/adding-a-transport.mdx @@ -184,4 +184,5 @@ No rebuild of `ktestify-cucumber` required. The plugin is discovered and its ste - [Plugin System →](../plugins/plugin-system), `KtestifyPlugin`, `PluginRegistry`, `PluginContext` - [Azure Blob plugin →](../plugins/azureblob), a real example of the plugin pattern +- [Synchronous transports →](synchronous-transports), the `RequestResponseClient` contract for HTTP, gRPC, and other call/response transports - [Architecture →](../architecture), how `RecordFetcher` and `ConsumedRecord` keep transports isolated \ No newline at end of file diff --git a/docs/extend/transports/synchronous-transports.mdx b/docs/extend/transports/synchronous-transports.mdx new file mode 100644 index 0000000..8d532a5 --- /dev/null +++ b/docs/extend/transports/synchronous-transports.mdx @@ -0,0 +1,149 @@ +--- +sidebar_position: 3 +title: Synchronous Transports +description: RequestResponseClient, AbstractSynchronousConsumer, and PollingRequestResponseClient, the sibling contracts for HTTP, gRPC, and other call/response transports. +--- + +# Synchronous Transports (`1.1.1`) + +`RecordFetcher` (see [Adding a Transport →](adding-a-transport)) models a background stream you poll until a record appears, a great fit for Kafka or Azure Blob polling, but the wrong shape for a transport where the caller sends a request and gets an answer immediately. + +Since `1.1.1`, `ktestify-core` ships a second, sibling transport contract for exactly that case: HTTP, gRPC, SOAP, or any other call/response protocol. + +--- + +## Why a second contract instead of forcing `RecordFetcher` + +Forcing an HTTP call through `RecordFetcher.fetch()` would be dishonest, that method takes no argument and is documented to block until something appears on a subscription. An HTTP request is caller-initiated and synchronous, it does not "wait for a record to appear", it sends a request right now and gets a response. + +The alternative, writing a plugin with its own ad hoc action/validation services and skipping the core transport contracts entirely, would mean that plugin stops following the shared architecture. It would also not help the next synchronous transport that comes along. + +So instead, `RequestResponseClient` sits next to `RecordFetcher` as an equally first class contract, returning the same `List>` shape, so the entire assertion layer (`RecordMatcher`, `MatchContext`, `MatchResult`, `RecordMatcherFactory`) is reused unchanged regardless of which contract a transport implements. + +--- + +## The contract + +```java +public interface RequestResponseClient extends AutoCloseable { + + List> execute(Req request) throws FetchException; + + @Override + void close(); +} +``` + +- `Req` is whatever request shape your transport needs (for example an HTTP request spec with method, URL, headers, and body). +- `execute(...)` returns a non-null, non-empty list, normally exactly one `ConsumedRecord`, wrapping the response. +- `FetchException` is thrown on connection errors, timeouts, or any non-recoverable transport error, exactly like `RecordFetcher.fetch()`. +- `close()` releases connection pools or other resources, and must be idempotent. + +--- + +## Mapping a response to `ConsumedRecord` + +There is no new response type. A synchronous transport builds a plain `ConsumedRecord`, using the new `attributes` map for status/metadata that does not belong under `headers`: + +| `ConsumedRecord` field | Typical HTTP mapping | +|---|---| +| `source` | request URL | +| `partition` | `0`, no concept | +| `offset` | `-1`, no concept | +| `key` | HTTP method (`GET`, `POST`, ...) | +| `value` | response body as a `String` | +| `timestamp` | the instant the response was received | +| `headers` | response HTTP headers | +| `attributes` | `{"statusCode": "200", "elapsedMs": "42"}` | + +Because the body lands in `value` and status/metadata lands in `attributes`, every existing raw matcher (`FileRecordMatcher`, `XmlRecordMatcher`, `XPathRecordMatcher`, `FieldsRecordMatcher`) keeps working unchanged for body assertions, and the new `AttributeRecordMatcher` (see [Built-in matchers →](../matchers/built-in-matchers)) handles status/metadata assertions. No new matcher code is needed per transport. + +--- + +## Orchestration - `AbstractSynchronousConsumer` + +Mirrors `AbstractKafkaConsumer`, but wired to a `RequestResponseClient` instead of a per-call `KafkaRecordFetcher`: + +```java +public abstract class AbstractSynchronousConsumer extends AbstractConsumer { + + protected final RequestResponseClient client; + protected final RecordMatcher matcher; + + protected abstract Req buildRequest(); + protected abstract MatchContext buildMatchContext(); + + @Override + public Boolean call() throws ConsumerException { + try { + Req request = buildRequest(); + List> records = client.execute(request); + MatchResult result = matcher.match(records, buildMatchContext()); + return result.isPassed(); + } catch (FetchException e) { + throw new ConsumerException(e.getMessage()); + } + } +} +``` + +A concrete consumer only needs to implement `buildRequest()` and `buildMatchContext()`, everything else is inherited. + +### Client lifecycle, note the difference from Kafka + +`AbstractKafkaConsumer` creates a fresh `KafkaRecordFetcher` per call and closes it in a `finally` block. `AbstractSynchronousConsumer` does **not** do that. A `RequestResponseClient` is expected to be a longer lived, connection pooled client, for example `java.net.http.HttpClient`, owned and closed once by the plugin's shared scenario resources, not created and discarded per request. + +--- + +## Retrying, `PollingRequestResponseClient` + +Every synchronous transport eventually needs "keep calling until the answer looks right", for example asserting an endpoint eventually returns `200` once an asynchronous side effect completes. Instead of every plugin writing its own sleep loop, `ktestify-core` ships a generic decorator: + +```java +public class PollingRequestResponseClient implements RequestResponseClient { + + public PollingRequestResponseClient( + RequestResponseClient delegate, + Predicate>> untilPredicate, + long timeoutMs, + long pollIntervalMs) { ... } + + @Override + public List> execute(Req request) throws FetchException { ... } +} +``` + +It wraps any other `RequestResponseClient`, retries `execute(...)` until `untilPredicate` passes or `timeoutMs` elapses, and on timeout returns the **last** result obtained instead of throwing, so the following `RecordMatcher` failure message shows the real final state rather than a generic timeout string. A `FetchException` from the delegate is only propagated when no successful attempt has ever produced a result. + +--- + +## Implementing a new synchronous transport + +``` +ktestify-plugin-http/ +├── pom.xml +└── src/main/java/io/github/ktestify/http/ + ├── HttpPlugin.java ← implements KtestifyPlugin + ├── io/ + │ ├── HttpRequestSpec.java ← the Req type: method, url, headers, query params, body + │ └── HttpRequestResponseClient.java ← implements RequestResponseClient + ├── HttpConsumer.java ← extends AbstractSynchronousConsumer + └── steps/ + └── ... ← Cucumber step definitions +``` + +`HttpRequestResponseClient` wraps `java.net.http.HttpClient` (already used elsewhere in the project, see the notifications plugin's webhook channel), builds a `HttpRequest`, sends it, and maps the `HttpResponse` to a `ConsumedRecord` using the table above. + +`HttpConsumer` only needs to supply `buildRequest()` (read the endpoint, method, path, and body from its own context object) and `buildMatchContext()` (map `matchMethod`, `matchFilePaths`, `excludedFields`, and `expectedAttributes`, exactly like `AbstractKafkaConsumer.buildMatchContext()` does for Kafka). + +No `RecordFetcher` implementation is needed for a purely synchronous transport, and no new matcher code is needed for body assertions. + +--- + +## See also + +- [Adding a Transport →](adding-a-transport), the asynchronous, `RecordFetcher` based path +- [Architecture →](../architecture), how both transport contracts feed the same orchestration and assertion layers +- [Core Concepts →](../core-concepts), `ConsumedRecord.attributes` and `MatchContext.expectedAttributes` +- [Built-in matchers →](../matchers/built-in-matchers), `AttributeRecordMatcher` + diff --git a/docs/write-tests/advanced/multi-row-datatables.mdx b/docs/write-tests/advanced/multi-row-datatables.mdx index afc1603..8e6f463 100644 --- a/docs/write-tests/advanced/multi-row-datatables.mdx +++ b/docs/write-tests/advanced/multi-row-datatables.mdx @@ -1,4 +1,4 @@ -q--- +--- sidebar_position: 3 title: Multi-Row DataTables description: How KTestify handles DataTables with more than one data row across producer, assertion, and batch steps. @@ -42,7 +42,7 @@ Every single-record assertion step (`Then expected record from file`, XML/XPath ```gherkin Then expected record from file - | topicAlias | file | expectedRecordKey | consumerReadTimeout | consumerDeltaTime | + | topicAlias | file | expectedRecordKey | consumerReadTimeout | consumerDeltaTime | | orders-out | expected-1.json | order-001 | 30 | 60 | | orders-out | expected-2.json | order-002 | 30 | 60 | ``` diff --git a/sidebars.ts b/sidebars.ts index 253818b..e7518c7 100644 --- a/sidebars.ts +++ b/sidebars.ts @@ -88,6 +88,7 @@ const sidebars: SidebarsConfig = { items: [ 'extend/transports/kafka', 'extend/transports/adding-a-transport', + 'extend/transports/synchronous-transports', ], }, {