debezium/dbz#1810 Add automated E2E tests for the auditlog example.

Signed-off-by: Mohnish <kmohnishm@gmail.com>

Author: KMohnishM

.github/workflows/auditlog-workflow.yml

@@ -5,27 +5,36 @@ on:
     paths:
       - 'auditlog/**'
       - '.github/workflows/auditlog-workflow.yml'
+      - 'scripts/run-example-test.py'
   pull_request:
     paths:
       - 'auditlog/**'
       - '.github/workflows/auditlog-workflow.yml'
+      - 'scripts/run-example-test.py'
 
 jobs:
-  build:
+  test:
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v6
-      - name: Set up JDK 8
+      - uses: actions/checkout@v4
+
+      - name: Set up Java
         uses: actions/setup-java@v4
         with:
           java-version: '8'
           distribution: 'temurin'
-      - name: Cache local Maven repository
-        uses: actions/cache@v4
+          cache: 'maven'
+
+      - name: Build auditlog services
+        run: mvn clean package -DskipTests -f auditlog/pom.xml
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
         with:
-          path: ~/.m2/repository
-          key: ${{ runner.os }}-maven-${{ hashFiles('auditlog/**/pom.xml') }}
-          restore-keys: |
-            ${{ runner.os }}-maven-
-      - name: Check changes in [auditlog] example
-        run: cd auditlog && mvn clean package -B -Dorg.slf4j.simpleLogger.log.org.apache.maven.cli.transfer.Slf4jMavenTransferListener=warn -Dmaven.wagon.http.pool=false -Dmaven.wagon.httpconnectionManager.ttlSeconds=120
+          python-version: '3.11'
+
+      - name: Install Python dependencies
+        run: pip install pyyaml requests
+
+      - name: Run auditlog test
+        run: python scripts/run-example-test.py auditlog

auditlog/test.yaml

@@ -0,0 +1,63 @@
+name: auditlog
+description: Tests Debezium PostgreSQL connector with audit log enrichment via Kafka Streams
+env_file: ../.env
+compose_file: docker-compose.yaml
+
+cleanup:
+  type: docker_compose_down
+  volumes: true
+
+steps:
+  - name: Start all services
+    type: docker_compose_up
+    detach: true
+
+  - name: Wait for Kafka Connect REST API to be ready
+    type: http_wait
+    url: http://localhost:8083/connectors
+    timeout_seconds: 120
+    interval_seconds: 5
+
+  - name: Register Debezium PostgreSQL connector
+    type: http_put
+    url: http://localhost:8083/connectors/inventory-connector/config
+    body_file: register-postgres.json
+    expected_status: [200, 201]
+
+  - name: Wait for connector to be RUNNING
+    type: http_wait
+    url: http://localhost:8083/connectors/inventory-connector/status
+    timeout_seconds: 60
+    interval_seconds: 5
+    expected_json_path: connector.state
+    expected_value: RUNNING
+
+  - name: Wait for vegetables-service to be ready
+    type: http_wait
+    url: http://localhost:8080/vegetables
+    timeout_seconds: 120
+    interval_seconds: 5
+    expected_status: [200, 405]
+
+  - name: Create a new vegetable record (farmerbob)
+    type: http_post
+    url: http://localhost:8080/vegetables
+    headers:
+      Authorization: "Bearer eyJraWQiOiJqd3Qua2V5IiwidHlwIjoiSldUIiwiYWxnIjoiUlMyNTYifQ.eyJzdWIiOiJmYXJtZXJib2IiLCJ1cG4iOiJmYXJtZXJib2IiLCJhdXRoX3RpbWUiOjE1NjY0NTgxMTMsImlzcyI6ImZhcm1zaG9wIiwiZ3JvdXBzIjpbImZhcm1lcnMiLCJjdXN0b21lcnMiXSwiZXhwIjo0MTAyNDQ0Nzk5LCJpYXQiOjE1NjY0NTgxMTMsImp0aSI6IjQyIn0.CscbJN8amqKryYvnVO1184J8F67HN2iTEjVN2VOPodcnoeOd7_iQVKUjC3h-ye5apkJjvAsQKrjzlrGCHRfl-n6jC9F7IkOtjoWnJ4wQ9BBo1SAtPw_Czt1I_Ujm-Kb1p5-BWACCBCVVFgYZTWP_laz5JZS7dIvs6VqoNnw7A4VpA6iPfTVfYlNY3u86-k1FvEg_hW-N9Y9RuihMsPuTdpHK5xdjCrJiD0VJ7-0eRQ8RXpycHuHN4xfmV8MqXBYjYSYDOhbnYbdQVbf0YJoFFqfb75my5olN-97ITsi2MS62W_y-RNT0qZrbytqINA3fF3VQsSY6VcaqRAeygrKm_Q"
+      Date: "Mon, 20 Apr 2026 12:00:00 GMT"
+    body:
+      name: "Cucumber"
+      description: "Fresh and crunchy"
+    expected_status: [200, 201, 204]
+
+  - name: Verify enriched message appears in Kafka
+    type: kafka_consume
+    topic: dbserver1.inventory.vegetable.enriched
+    timeout_seconds: 60
+    expected_content: "Cucumber"
+
+  - name: Verify metadata enrichment (user farmerbob)
+    type: kafka_consume
+    topic: dbserver1.inventory.vegetable.enriched
+    timeout_seconds: 30
+    expected_content: "farmerbob"

scripts/run-example-test.py

@@ -90,41 +90,64 @@ def step_docker_compose_exec(step, config):
     run_cmd(cmd)
 
 
-def step_http_put(step, config):
+def _http_request(method, step, config):
     url = step["url"]
-    body_file = step["body_file"]
+    body_file = step.get("body_file")
+    body = step.get("body")
+    headers = step.get("headers", {})
     expected_statuses = step.get("expected_status", [200, 201])
+    
     if isinstance(expected_statuses, int):
         expected_statuses = [expected_statuses]
 
-    with open(body_file) as f:
-        body = json.load(f)
+    if body_file:
+        with open(body_file) as f:
+            body = json.load(f)
 
-    print(f"  PUT {url}")
-    resp = requests.put(url, json=body, timeout=30)
+    print(f"  {method} {url}")
+    resp = requests.request(method, url, json=body, headers=headers, timeout=30)
+    
     if resp.status_code not in expected_statuses:
         raise RuntimeError(
-            f"PUT {url} returned {resp.status_code}, expected one of {expected_statuses}.\n"
+            f"{method} {url} returned {resp.status_code}, expected one of {expected_statuses}.\n"
             f"Response: {resp.text}"
         )
     print(f"  -> {resp.status_code}")
 
 
+def step_http_put(step, config):
+    _http_request("PUT", step, config)
+
+
+def step_http_post(step, config):
+    _http_request("POST", step, config)
+
+
 def step_http_wait(step, config):
     url = step["url"]
     timeout = step.get("timeout_seconds", 60)
     interval = step.get("interval_seconds", 5)
     json_path = step.get("expected_json_path")
     expected_value = step.get("expected_value")
+    headers = step.get("headers", {})
+    expected_statuses = step.get("expected_status")
+    if isinstance(expected_statuses, int):
+        expected_statuses = [expected_statuses]
 
     deadline = time.time() + timeout
     print(f"  Polling {url} (timeout={timeout}s)")
 
     while True:
         try:
-            resp = requests.get(url, timeout=10)
-            # Only accept 2xx status codes as success
-            if 200 <= resp.status_code < 300:
+            resp = requests.get(url, headers=headers, timeout=10)
+            
+            is_success = False
+            if expected_statuses is not None:
+                is_success = resp.status_code in expected_statuses
+            else:
+                is_success = 200 <= resp.status_code < 300
+
+            if is_success:
                 if json_path and expected_value:
                     data = resp.json()
                     # resolve simple dot-separated path
@@ -199,13 +222,15 @@ def step_wait(step, config):
     "docker_compose_stop": step_docker_compose_stop,
     "docker_compose_exec": step_docker_compose_exec,
     "http_put": step_http_put,
+    "http_post": step_http_post,
     "http_wait": step_http_wait,
     "kafka_consume": step_kafka_consume,
     "env_override": step_env_override,
     "wait": step_wait,
 }
 
 
+
 # ---------------------------------------------------------------------------
 # Main
 # ---------------------------------------------------------------------------