Verifying Your First AsyncAPI Operation

Drift talks to a real broker. Create compose.yaml:

services:
  kafka:
    image: confluentinc/cp-kafka:latest
    container_name: drift-tutorial-kafka
    ports:
      - "9092:9092"
    environment:
      KAFKA_ENABLE_KRAFT: "yes"
      KAFKA_NODE_ID: "1"
      KAFKA_PROCESS_ROLES: "broker,controller"
      KAFKA_CONTROLLER_LISTENER_NAMES: "CONTROLLER"
      KAFKA_LISTENERS: "PLAINTEXT://:9092,CONTROLLER://:9093"
      KAFKA_ADVERTISED_LISTENERS: "PLAINTEXT://localhost:9092"
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: "PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT"
      KAFKA_CONTROLLER_QUORUM_VOTERS: "1@kafka:9093"
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: "true"
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: "1"
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: "1"
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: "1"
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: "0"
      ALLOW_PLAINTEXT_LISTENER: "yes"

Start it:

docker compose up -d

Confirm the broker is up:

docker compose ps

Create the topic Drift will subscribe to:

docker exec drift-tutorial-kafka kafka-topics \
  --bootstrap-server localhost:9092 \
  --create --if-not-exists \
  --topic drift.examples.order-created \
  --partitions 1 --replication-factor 1

This describes a service that publishes an order.created event. Save it as asyncapi/send.asyncapi.yaml:

asyncapi: 3.1.0
info:
  title: Drift AsyncAPI tutorial
  version: 1.0.0
defaultContentType: application/json

servers:
  local:
    host: localhost:9092
    protocol: kafka
    description: Local Kafka broker

channels:
  orderCreated:
    address: drift.examples.order-created
    messages:
      orderCreated:
        $ref: "#/components/messages/orderCreated"

operations:
  sendOrderCreated:
    action: send
    title: Send order created
    summary: Publishes an order.created event.
    channel:
      $ref: "#/channels/orderCreated"
    messages:
      - $ref: "#/channels/orderCreated/messages/orderCreated"

components:
  messages:
    orderCreated:
      name: orderCreated
      title: Order created event
      correlationId:
        location: "$message.header#/correlation-id"
      headers:
        type: object
        properties:
          correlation-id:
            type: string
        required: [correlation-id]
      payload:
        type: object
        required: [eventType, orderId, customerId, status]
        properties:
          eventType:
            type: string
            const: order.created
          orderId:
            type: string
          customerId:
            type: string
          status:
            type: string
            const: created

Two details matter:

action: send describes what your application does. Your service is the publisher, so Drift's job is to subscribe and capture. Drift's role is always the opposite of your service's.

correlationId tells Drift where to find the identifier that marks the message as the one this test is waiting for.

Drift cannot make your service publish. A trigger hook does that. In a real project the trigger would call your service. For this tutorial it publishes directly. Save it as hooks/trigger-send.py:

#!/usr/bin/env python3
"""Publish one order.created event to Kafka."""
import argparse
import json
from kafka import KafkaProducer

parser = argparse.ArgumentParser()
parser.add_argument("--correlation-id", required=True)
parser.add_argument("--order-id", required=True)
parser.add_argument("--customer-id", required=True)
args = parser.parse_args()

producer = KafkaProducer(bootstrap_servers="localhost:9092")
producer.send(
    "drift.examples.order-created",
    value=json.dumps({
        "eventType": "order.created",
        "orderId": args.order_id,
        "customerId": args.customer_id,
        "status": "created",
    }).encode(),
    headers=[("correlation-id", args.correlation_id.encode())],
)
producer.flush()
print(f"published order.created for {args.correlation_id}")

Install the dependency:

python3 -m venv .venv
source .venv/bin/activate
pip install kafka-python

The trigger must publish the same correlation ID Drift is filtering on. Drift passes it in as an argument, so the two cannot drift apart.

Save this as drift/send.testcases.yaml:

# yaml-language-server: $schema=https://download.pactflow.io/drift/schemas/drift.testcases.v1.schema.json
drift-testcase-file: v1
title: "AsyncAPI Kafka send flow"
description: |
  Observe an outbound order.created event after a trigger hook fires.

sources:
  - name: async-send
    path: ../asyncapi/send.asyncapi.yaml

plugins:
  - name: asyncapi
  - name: kafka
  - name: json

operations:
  SendOrderCreated_Observe:
    target: async-send:sendOrderCreated:orderCreated
    description: "Observe the order.created event emitted after the trigger hook runs"
    parameters:
      timeout-ms: 5000
      correlation-id: send-order-created-001
      orderId: ORD-100
      customerId: CUST-9
    trigger:
      executable-type: command
      value: python3
      parameters:
        args:
          - ../hooks/trigger-send.py
          - --correlation-id
          - ${parameters.correlation-id}
          - --order-id
          - ${parameters.orderId}
          - --customer-id
          - ${parameters.customerId}
      timeout-ms: 2000
    expected:
      headers:
        correlation-id: ${parameters.correlation-id}
      payload:
        eventType: order.created
        orderId: ${parameters.orderId}
        customerId: ${parameters.customerId}
        status: created

What each part does:

Run the Verification

drift verify -f drift/send.testcases.yaml

Drift subscribes to drift.examples.order-created, runs the trigger, waits for a message carrying send-order-created-001, then validates it.

You should see the operation pass.

sequenceDiagram

    autonumber
    participant D as Drift
    participant B as Kafka
    participant T as trigger-send.py

    D->>B: Subscribe to drift.examples.order-created
    D->>T: Run the trigger
    T->>B: Publish order.created
    B-->>D: Deliver the matching message
    D->>D: Validate against the AsyncAPI schema<br/>and your expected block

Seeing a failure is as useful as seeing a pass.

In drift/send.testcases.yaml, change the expected status:

 expected:
      payload:
        status: shipped     # the service actually publishes "created"

Run it again:

drift verify -f drift/send.testcases.yaml

Drift captures the message and reports a mismatch on status, showing what it expected and what it received.

Change it back to created and confirm the test passes again.

  trigger:
      parameters:
        args:
          - ../hooks/trigger-send.py
          - --correlation-id
          - a-different-id          # was ${parameters.correlation-id}

docker compose down