AsyncAPI

For an AsyncAPI 3.x document the mock plays the application the document describes. Servers with the ws, http and mqtt protocols are served (and wss, https and mqtts over TLS with --tls-cert, cleartext without). Kafka and AMQP channels aren’t served, but are read from a real cluster or broker with --proxy-to kafka:// (below) and --proxy-to amqp:// (below); the others are listed as not served.

• WebSocket. A client connects to a channel’s address under its server’s pathname (ws://127.0.0.1:4010/v1/rooms/42 for rooms/{roomId}). It gets the messages of the channel’s send operations, one after another, every second by default: the message’s examples in turn, or values generated from its payload schema (always with --dynamic). What it sends is checked against the receive operations: a match is answered with the operation’s reply, anything else with a problem details message.
• HTTP. A receive operation takes messages with POST (or its HTTP binding’s method) and answers 202, or 200 with its reply; an invalid message gets 422. A send operation answers GET with a message.
• MQTT. A broker on its own port takes subscriptions to the channels’ topics (wildcards included: a parameter of an address takes its default, first enum value or example) and publishes the send operations’ messages with their MQTT binding’s QoS. Messages clients publish go to the other subscribers, as with any broker, once they match a receive operation; its reply is published to the reply channel’s topic. MQTT 3.1.1 and 5 clients are served; MQTT 5 subscribers also get the MQTT 5 properties of a message: its headers as user properties (the first example’s, else generated from the header schemas), and the payload format indicator, content type, correlation data and response topic of its MQTT binding (generated from their schemas when they are schemas) and the operation binding’s message expiry. An MQTT 5 client’s properties and user properties are checked as the MQTT proxy checks them; its request is answered on the topic of its reply address user property, or on its Response Topic when the reply channel has address: null, with its correlation id and Correlation Data. Retained messages expire with their Message Expiry Interval.

x-apigenie-mock on a send operation sets interval (milliseconds), count (messages per client, 0 for no limit), jitter, messages (the names to send) and publish: false.

operations:
  sendNotices:
    action: send
    channel: {$ref: '#/channels/room'}
    x-apigenie-mock: {interval: 500, count: 10}

The log has ws-open, ws-in, ws-out and ws-close events for WebSocket clients and mqtt-subscribe, mqtt-in, mqtt-out and mqtt-unsubscribe ones for MQTT, with the channel’s address, the operation and the message. When the server stops, WebSocket clients get the close code 1001 and MQTT 5 clients a DISCONNECT; after a reload they reconnect to the new version.

Contract testing of a WebSocket API

With --proxy-to ws://host:port an AsyncAPI document’s WebSocket channels aren’t mocked: each client of a channel gets a connection to the real server, and every message of either side is checked against the document. Point a frontend or a test suite at the mock instead of the server.

apigenie-server chat.yaml --proxy-to ws://localhost:9000 --report contract.json --fail-on errors

The mock connects to the server first, with the client’s path under the URL’s prefix, its query, Origin, cookies, Authorization and Sec-WebSocket-Protocol, plus Via and X-Forwarded-*; the client gets the subprotocol the server chose. A server that can’t be reached answers the client’s handshake with 502 (504 when it doesn’t answer in time), and a server that refuses it (401, 404, …) has its answer passed on. What clients send is checked against the channel’s receive operations, what the server sends against its send operations. A client’s message of an operation with a reply on the same channel waits for its reply: the first message the server sends that is one of the reply’s messages is checked as that reply (logged as name (reply) with the request’s operation), not against the send operations. Messages are relayed whole (fragments are put together first); pings are answered on each side; a close frame goes on with its code. No WebSocket extension is negotiated (permessage-deflate is left out of the handshake). HTTP channels go to the same server (see below); the MQTT channels to the broker of a second --proxy-to mqtt://..., or they keep being mocked.

Check	Severity	When
message.incompatible	error	The message matches none of the messages of the channel’s operations in its direction.
message.malformed	error	The message isn’t JSON where JSON is expected.
message.frame.unexpected	warn	A binary frame for a text or JSON message, or a text frame for a binary one (image/png, application/octet-stream, Protobuf, Avro, …).
message.qos.unexpected	warn	MQTT: a client published at a QoS other than the one of the operation’s MQTT binding.
channel.parameter.invalid	error	A parameter of the channel’s address (a path segment, a topic level) has a value outside the parameter’s enum.
channel.receive.undeclared	warn	A client sent a message on a channel the application has no receive operation for.
channel.send.undeclared	warn	The server sent a message on a channel it has no send operation for.
message.reply.uncorrelated	error	The reply’s correlationId (in the payload) isn’t the request’s, or is missing.
operation.query.invalid	error	HTTP channels: the request’s query doesn’t match the query of the operation’s HTTP binding.
message.headers.invalid	error	HTTP channels: the request’s or the reply’s header fields don’t match the message’s headers or its HTTP binding’s headers.

validation.request sets what happens to the clients’ messages and validation.response to the server’s: log by default for a proxy (reported, and forwarded), enforce with --strict. In enforce mode a message with an error isn’t forwarded, and the client gets a problem message instead (asyncapi.validate_incoming stays the mock’s setting):

{"type": "https://apigenie.pl/mock/errors#invalid-upstream-message", "title": "The upstream's message doesn't match the specification",
 "status": 422, "channel": "rooms/1", "direction": "upstream", "validation": [{"location": "message/text", "code": "type",
 "message": "expected type string, found integer", "check": "message.incompatible", "severity": "error"}]}

A client’s message gets #invalid-message with "direction": "client". The log has the ws-open, ws-in, ws-out and ws-close events with the operation, the message’s name, the violations and source: proxy (mock for what the proxy answered itself). contract.severity changes the checks’ severities as for the other proxies.

Coverage and reports. Each message of each operation is a branch: it is covered once a valid one went by. An operation is covered, partial or not-tested; a message that matches none is counted for the operation it was checked against first. GET /__apigenie/contract returns the report ("specification": "asyncapi", connections, messages in and out, channels, operations, violations), --report and --report-junit write it when the server stops (a test case per operation), and --fail-on and --min-coverage (the share of messages) make the exit status 5.

Contract testing of MQTT channels

With --proxy-to mqtt://host:port the mock’s broker relays to a real broker: each client that connects to the mock gets a connection of its own to the broker, and every message published either way is checked against the document.

apigenie-server devices.yaml --proxy-to mqtt://localhost:1884 --mqtt-port 1883 --report contract.json
apigenie-server events.yaml --proxy-to ws://localhost:9000 --proxy-to mqtt://localhost:1884

The upstream connection has the client’s MQTT version, client identifier (the one the mock assigns when the client gives none), clean start, keep-alive, will, user name and password. The client’s CONNECT gets the broker’s CONNACK, a refusal included; a broker that can’t be reached is answered with “server unavailable” (return code 3, MQTT 5 0x88). Subscriptions are the broker’s: the SUBACK is its answer, and the broker decides who gets which message. What clients publish is checked against the topic’s channel’s receive operations, with the QoS of the operation’s MQTT binding when it gives one (message.qos.unexpected); what the broker delivers against the send operations. A subscriber also gets other clients’ messages, so on a channel without send operations a delivery is checked against the receive operations. Topics of no channel are relayed unchecked (debug level in the log).

In enforce mode a client’s message with an error isn’t forwarded (MQTT 5 clients get reason code 0x99 in their acknowledgement), and a broker’s message with an error isn’t delivered. A client that disconnects is disconnected from the broker; one whose connection is lost has its upstream connection dropped too, so the broker publishes its will; a client the broker disconnects is disconnected. A reload keeps the clients connected. The log has mqtt-connect, mqtt-subscribe, mqtt-in, mqtt-out, mqtt-unsubscribe and mqtt-disconnect events with source: proxy (mock for what the proxy decided itself), and coverage and reports work as for WebSocket.

MQTT 5 properties are relayed both ways: user properties, payload format indicator, content type, response topic, correlation data and message expiry (not the topic alias and subscription identifiers, which belong to one connection). An MQTT 5 message’s are checked against the MQTT bindings (0.2.0):

Binding	Checked (message.properties.invalid, error)
message payloadFormatIndicator	The Payload Format Indicator is the same (absent counts as 0)
message contentType	The Content Type is the same, ignoring case
message correlationData	Correlation Data is present and matches the schema
message responseTopic	A Response Topic is present, the string given or matching the schema
operation messageExpiryInterval	A Message Expiry Interval is present and no longer

Its user properties are its headers, checked against the message’s headers (message.headers.invalid). MQTT 3.1.1 messages have no properties: nothing is checked for them. The report counts the messages whose bindings state properties, checked or not: summary.properties: {"checked": 3, "unchecked": 1}, properties of each operation, and a summary line MQTT 5 properties: checked on 3 messages, not on 1 (MQTT 3.1.1). A client’s QoS 1 and 2 messages are acknowledged by the mock before the broker acknowledges them, and the mock delivers at most QoS 1. An mqtts:// broker (port 8883 by default) is reached over TLS and verified as https:// upstreams are (see TLS).

Contract testing of Kafka channels

There is no Kafka broker in the mock. With --proxy-to kafka://host:port (port 9092 by default; kafkas://, port 9093, over TLS) the server is a client of the real cluster, a validating tap: it reads the topics of the document’s Kafka channels and checks every record on them, whoever produced it.

apigenie-server shop.yaml --proxy-to kafka://localhost:9092 --report contract.json
apigenie-server events.yaml --proxy-to mqtt://localhost:1883 --proxy-to kafka://localhost:9092

A channel with a kafka or kafka-secure server is a Kafka channel; its topic is the channel’s Kafka binding topic, else its address. Parameters of the address match the topic’s segments, so stock.{region} reads stock.eu and stock.us; the cluster’s topics are looked at again every 10 seconds for new ones. The tap reads every partition from the offset it had when the server started, outside any consumer group, so it never takes partitions from the application’s consumers and commits no offsets. A topic the cluster doesn’t have yet is reported once and read once it appears; a lost connection is reported and tried again.

A record doesn’t say who produced it, so the channel’s operations tell. On a channel with only send operations it is the application’s (a kafka-out event, checked against them); with only receive operations it is a client’s (kafka-in); with both it is checked as the application’s, and as a client’s when that doesn’t match. What is checked:

What	Checked against	Check
The value	The operations’ messages	message.incompatible, message.malformed
The headers	The message’s headers (values read as the type the schema says)	message.headers.invalid
The key	The message’s Kafka binding key (an error too when the record has none)	message.key.invalid (error)
The partition	The channel’s Kafka binding partitions	message.partition.invalid (warn)
The topic	The address’s parameters	channel.parameter.invalid

Records are on the cluster before the tap sees them, so enforce mode reports them like log mode. Kafka operations count in the coverage, GET /__apigenie/contract, --report and --report-junit like any other, and the report’s upstream lists the kafka:// URL. The log has kafka-in and kafka-out events with the topic as channel, the record’s partition and offset in the message, and its key (as kafka-key) and headers as the event’s headers; what the tap found or failed at are server events with source: kafka.

The SASL PLAIN user and password come from APIGENIE_KAFKA_USERNAME and APIGENIE_KAFKA_PASSWORD (apigenie test reads them too), never from the URL. kafkas:// brokers are verified as https:// upstreams are (see TLS). The cluster needs Kafka 2.3 or later. Compressed record batches aren’t read: they are reported (3 record(s) at offset 40 compressed with zstd) and skipped. A kafka:// upstream goes with one document, not with several services. See ADR-170 for what the client does and doesn’t do.

Contract testing of AMQP channels

There is no AMQP broker in the mock either. With --proxy-to amqp://host:port/vhost (port 5672 by default; amqps://, port 5671, over TLS; the path is the virtual host, /%2Fcarpet-express for /carpet-express, none for /) the server is an AMQP 0-9-1 client of the real broker, a validating tap.

APIGENIE_AMQP_USERNAME=conformance APIGENIE_AMQP_PASSWORD=conformance \
  apigenie-server customs.yaml --proxy-to amqp://localhost:5672/%2Fcarpet-express --report contract.json

A channel with an amqp or amqps server is an AMQP channel. Its AMQP channel binding says what it is:

• is: routingKey (the default): the address is a routing key of exchange.name (of exchange.type, topic by default). The tap binds a queue of its own (exclusive, deleted when it disconnects) to the exchange: for a topic exchange with the address’s parameters as * words (customs.{country}.filing is customs.*.filing), for a fanout or headers exchange with any key. Direct exchanges bind the address itself, so their addresses can’t have parameters. It reads that queue, so it gets a copy of every message the exchange routes there, and the application’s queues are left alone. An exchange the broker doesn’t have is reported, and looked for again every 10 seconds.
• is: queue, and routing keys of the default exchange: a queue the application consumes. Reading it would take the application’s messages, so the tap doesn’t, and says so when it starts. What apigenie test and the admin API send there is checked.

Messages are checked as the Kafka tap checks records, in the same direction: their body and headers against the operations’ messages, and their properties against the AMQP bindings (0.3.0) (message.properties.invalid, error):

Binding	Checked
operation deliveryMode, priority, userId, replyTo	The property is present and the same
operation expiration	An expiration is present and no longer
operation timestamp: true	A timestamp is present
message contentEncoding	The content encoding is the same
message messageType	The type property is the same

The log has amqp-in and amqp-out events with the routing key as channel and the exchange in the message; coverage and reports work as for Kafka, and enforce mode reports too. The PLAIN user and password come from APIGENIE_AMQP_USERNAME and APIGENIE_AMQP_PASSWORD (guest without them), never from the URL. An amqp:// upstream goes with one document. See ADR-171.

Contract testing of HTTP channels

With --proxy-to http://host:port (or the ws:// server of the WebSocket channels) an AsyncAPI document’s HTTP channels go to that server through the same transport as an OpenAPI document’s proxy (Via, X-Forwarded-*, 502 and 504 problems), checked against the document instead of an OpenAPI one:

• A receive operation takes requests with its HTTP binding’s method (POST by default). The request’s body is checked against the operation’s messages, its query against the binding’s query schema (operation.query.invalid) and its header fields against the message’s headers and HTTP binding headers, its traits’ included (message.headers.invalid). A 2XX response’s body and header fields are checked against its reply.
• A send operation is asked with GET, as the mock serves it; a 2XX response with a body is checked against its messages (not its header fields: they are the webhook’s).
• Other paths get 404 and other methods 405, without the server.

Query parameters and header fields are read as the type their schema says (integer, number, boolean, an array from repeated parameters); header fields the schemas don’t name are left out. apigenie test checks the requests of HTTP channels with the same code and the same checks (see Requests of HTTP channels).

In enforce mode an invalid request gets 422 (#invalid-message) without the server, and an invalid response is replaced by 502 (#invalid-upstream-message). Responses carry X-Apigenie-Mock-Source, X-Apigenie-Mock-Operation and X-Apigenie-Mock-Violations, and each exchange counts as a message of its operation in the coverage. Responses other than 2XX and the reply’s statusCode binding aren’t checked; fallback mode, --record-dir and --record-har are for OpenAPI documents.