Add Kafka integration tests and deploy-kafka EaaS task

Agent · Agent · commit 71ca4b6f0a18 · 2026-06-26T13:05:01.000Z
- Add three integration tests in tests/test_integration_api.py: test_kafka_compose_created, test_kafka_compose_tagged, test_kafka_compose_untagged. Each test verifies that the corresponding Kafka topic receives a message after a CTS compose operation. All three tests are skipped when KAFKA_URL is not set. - Add a kafka_url module-scoped fixture that skips when KAFKA_URL is unset. - Add a _consume_kafka_message helper that uses kafka-python KafkaConsumer with auto_offset_reset=earliest and a unique group_id per call so that messages published before the consumer subscribes are still visible. - Add kafka-python to test-requirements.txt. - Add deploy-kafka task to .tekton/integration-test-eaas.yaml deploying a Bitnami Kafka 3.7 single-node KRaft broker accessible at kafka:9092. - Add deploy-kafka to deploy-cts runAfter dependency list. - Inject MESSAGING_BACKEND and related Kafka settings into the cts-config ConfigMap inside deploy-cts so CTS publishes to the ephemeral Kafka broker. - Update run-tests kubectl exec script to install kafka-python alongside pytest and to pass KAFKA_URL=kafka:9092 to the pytest invocation. Closes #84 Generated-By: OpenCode (google-vertex-anthropic/claude-sonnet-4-6@default)
diff --git a/.tekton/integration-test-eaas.yaml b/.tekton/integration-test-eaas.yaml
@@ -457,6 +457,118 @@ spec:
     - name: kubeconfig-secret
       value: $(tasks.provision-environment.results.secretRef)
 
+  - name: deploy-kafka
+    runAfter:
+    - provision-environment
+    taskSpec:
+      params:
+      - name: kubeconfig-secret
+        type: string
+      steps:
+      - name: create-kafka
+        image: quay.io/konflux-ci/appstudio-utils:latest
+        script: |
+          #!/usr/bin/env bash
+          set -euo pipefail
+
+          KUBECONFIG=/tmp/kubeconfig
+          kubectl get secret $(params.kubeconfig-secret) -o jsonpath='{.data.kubeconfig}' | base64 -d > $KUBECONFIG
+          export KUBECONFIG
+
+          echo "=========================================="
+          echo "Deploying Kafka (Bitnami KRaft single-node)"
+          echo "=========================================="
+
+          kubectl apply -f - <<'EOFYAML'
+          apiVersion: apps/v1
+          kind: Deployment
+          metadata:
+            name: kafka
+            labels:
+              app: kafka
+          spec:
+            replicas: 1
+            selector:
+              matchLabels:
+                app: kafka
+            template:
+              metadata:
+                labels:
+                  app: kafka
+              spec:
+                containers:
+                - name: kafka
+                  image: docker.io/bitnami/kafka:3.7
+                  ports:
+                  - containerPort: 9092
+                    name: client
+                  - containerPort: 9093
+                    name: controller
+                  env:
+                  - name: KAFKA_CFG_NODE_ID
+                    value: "0"
+                  - name: KAFKA_CFG_PROCESS_ROLES
+                    value: "controller,broker"
+                  - name: KAFKA_CFG_CONTROLLER_QUORUM_VOTERS
+                    value: "0@kafka:9093"
+                  - name: KAFKA_CFG_LISTENERS
+                    value: "PLAINTEXT://:9092,CONTROLLER://:9093"
+                  - name: KAFKA_CFG_ADVERTISED_LISTENERS
+                    value: "PLAINTEXT://kafka:9092"
+                  - name: KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP
+                    value: "CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT"
+                  - name: KAFKA_CFG_CONTROLLER_LISTENER_NAMES
+                    value: "CONTROLLER"
+                  - name: KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE
+                    value: "true"
+                  - name: ALLOW_PLAINTEXT_LISTENER
+                    value: "yes"
+                  resources:
+                    requests:
+                      memory: "256Mi"
+                      cpu: "100m"
+                    limits:
+                      memory: "512Mi"
+                      cpu: "500m"
+                  readinessProbe:
+                    tcpSocket:
+                      port: 9092
+                    initialDelaySeconds: 20
+                    periodSeconds: 5
+                    timeoutSeconds: 3
+                    failureThreshold: 24
+          ---
+          apiVersion: v1
+          kind: Service
+          metadata:
+            name: kafka
+            labels:
+              app: kafka
+          spec:
+            ports:
+            - port: 9092
+              targetPort: 9092
+              name: client
+            - port: 9093
+              targetPort: 9093
+              name: controller
+            selector:
+              app: kafka
+          EOFYAML
+
+          echo "Waiting for Kafka to be ready..."
+          if ! kubectl wait --for=condition=available --timeout=300s deployment/kafka; then
+            echo "Kafka deployment failed! Debug info:"
+            kubectl describe deployment kafka
+            kubectl describe pod -l app=kafka
+            kubectl logs -l app=kafka --tail=50 || echo "No logs available"
+            exit 1
+          fi
+          echo "✓ Kafka is ready"
+    params:
+    - name: kubeconfig-secret
+      value: $(tasks.provision-environment.results.secretRef)
+
   - name: deploy-database
     runAfter:
     - provision-environment
@@ -597,6 +709,7 @@ spec:
     - deploy-database
     - deploy-openldap
     - deploy-dex
+    - deploy-kafka
     taskSpec:
       params:
       - name: kubeconfig-secret
@@ -643,6 +756,14 @@ spec:
                   ]
                   ADMINS = {"groups": [], "users": ["builder@example.com"]}
                   ALLOWED_BUILDERS = {"groups": [], "users": ["builder@example.com"]}
+                  MESSAGING_BACKEND = "kafka"
+                  MESSAGING_BROKER_URLS = ["kafka:9092"]
+                  MESSAGING_KAFKA_SECURITY_PROTOCOL = "PLAINTEXT"
+                  MESSAGING_KAFKA_SASL_MECHANISM = ""
+                  MESSAGING_KAFKA_USERNAME = ""
+                  MESSAGING_KAFKA_PASSWORD = ""
+                  MESSAGING_KAFKA_COMPRESSION_TYPE = "none"
+                  MESSAGING_TOPIC_PREFIX = "cts."
             httpd.conf: |
               ServerRoot "/etc/httpd"
               PidFile /tmp/httpd.pid
@@ -916,9 +1037,9 @@ spec:
             echo 'Installing Dex CA certificate...'
             echo '$DEX_CA_B64' | base64 -d > /tmp/dex-ca.crt
 
-            echo 'Installing pytest and requests...'
+            echo 'Installing pytest, requests, and kafka-python...'
             python3 -m ensurepip
-            python3 -m pip install --target /tmp/test-deps --quiet pytest requests
+            python3 -m pip install --target /tmp/test-deps --quiet pytest requests kafka-python
 
             echo ''
             echo 'Cloning repository...'
@@ -930,7 +1051,7 @@ spec:
 
             echo ''
             echo 'Running pytest...'
-            PYTHONPATH=/tmp/test-deps REQUESTS_CA_BUNDLE=/tmp/dex-ca.crt CTS_URL=http://cts:8080 AUTH_BACKEND=oidc_or_kerberos DEX_URL=https://dex:5556 python3 -m pytest tests/test_integration_api.py -v -s -o addopts=
+            PYTHONPATH=/tmp/test-deps REQUESTS_CA_BUNDLE=/tmp/dex-ca.crt CTS_URL=http://cts:8080 AUTH_BACKEND=oidc_or_kerberos DEX_URL=https://dex:5556 KAFKA_URL=kafka:9092 python3 -m pytest tests/test_integration_api.py -v -s -o addopts=
           "
           TEST_RESULT=$?
           set -e
diff --git a/test-requirements.txt b/test-requirements.txt
@@ -9,3 +9,5 @@ tox
 # Let's update this package to avoid this problem.
 itsdangerous>=1.1.0
 freezegun
+# Required for Kafka integration tests (test_integration_api.py)
+kafka-python
diff --git a/tests/test_integration_api.py b/tests/test_integration_api.py
@@ -15,6 +15,8 @@
 import json
 import os
 import ssl
+import time
+import uuid
 import urllib.parse
 from urllib.request import urlopen, Request
 from urllib.error import HTTPError, URLError
@@ -645,3 +647,142 @@ def test_auth_get_endpoints_accessible_without_token(http_client):
         status == 200
     ), f"Expected 200 for unauthenticated GET /api/1/tags/, got {status}"
     assert isinstance(data, dict), "GET /api/1/tags/ must return a dict"
+
+
+# Kafka messaging integration tests
+# These tests require a live Kafka broker and are skipped when KAFKA_URL is not set.
+
+
+@pytest.fixture(scope="module")
+def kafka_url():
+    """Return the Kafka broker URL from the KAFKA_URL environment variable.
+
+    Skips all tests that depend on this fixture when the variable is not set,
+    preserving backward compatibility with environments that have no Kafka broker.
+    """
+    url = os.environ.get("KAFKA_URL")
+    if not url:
+        pytest.skip("requires KAFKA_URL")
+    return url
+
+
+def _consume_kafka_message(kafka_url, topic, timeout_ms=10000):
+    """Poll *topic* for a single message and return its value (decoded JSON dict).
+
+    Uses ``auto_offset_reset='earliest'`` and a per-call UUID group_id so that
+    messages published before the consumer subscribes are still visible.
+
+    Returns the first message value received within *timeout_ms*, or raises
+    ``AssertionError`` if no message arrives in time.
+    """
+    from kafka import KafkaConsumer
+
+    group_id = f"cts-test-{uuid.uuid4()}"
+    consumer = KafkaConsumer(
+        topic,
+        bootstrap_servers=kafka_url,
+        auto_offset_reset="earliest",
+        group_id=group_id,
+        value_deserializer=lambda b: json.loads(b.decode("utf-8")),
+        consumer_timeout_ms=timeout_ms,
+    )
+    try:
+        for record in consumer:
+            return record.value
+        raise AssertionError(
+            f"No message received on Kafka topic '{topic}' within {timeout_ms} ms"
+        )
+    finally:
+        consumer.close()
+
+
+def test_kafka_compose_created(write_http_client, kafka_url):
+    """Importing a compose must publish a message on the cts.compose-created topic."""
+    data = import_compose(write_http_client, "KafkaCreatedTest", "1.0", "20260601")
+    compose_id = data["payload"]["compose"]["id"]
+    print(f"  Imported compose: {compose_id}")
+
+    # Allow a short grace period for the async publisher to send the message.
+    time.sleep(2)
+
+    msg = _consume_kafka_message(kafka_url, "cts.compose-created")
+
+    assert msg is not None, "Expected a message on cts.compose-created, got None"
+    assert msg.get("event") == "compose-created", (
+        f"Expected event='compose-created', got event={msg.get('event')!r}"
+    )
+    # Verify the message is not empty / a silent zero-payload stub
+    assert msg.get("compose_id") or msg.get("id") or "compose" in str(msg), (
+        f"Message does not reference any compose: {msg}"
+    )
+    print(f"  ✓ Received compose-created message: {msg}")
+
+
+def test_kafka_compose_tagged(write_http_client, kafka_url):
+    """Tagging a compose must publish a message on the cts.compose-tagged topic."""
+    # Create a tag and import a compose
+    tag_data = create_tag(
+        write_http_client,
+        "kafka-tag-test",
+        "Tag for Kafka tagged test",
+        "https://example.com/docs/kafka-tag-test",
+    )
+    tag_name = tag_data["name"]
+
+    compose_data = import_compose(write_http_client, "KafkaTaggedTest", "1.0", "20260602")
+    compose_id = compose_data["payload"]["compose"]["id"]
+    print(f"  Imported compose: {compose_id}, tag: {tag_name}")
+
+    # Tag the compose — this should trigger a cts.compose-tagged message.
+    tag_compose(write_http_client, compose_id, tag_name)
+    print(f"  Tagged compose with '{tag_name}'")
+
+    # Allow a short grace period for the async publisher.
+    time.sleep(2)
+
+    msg = _consume_kafka_message(kafka_url, "cts.compose-tagged")
+
+    assert msg is not None, "Expected a message on cts.compose-tagged, got None"
+    assert msg.get("event") == "compose-tagged", (
+        f"Expected event='compose-tagged', got event={msg.get('event')!r}"
+    )
+    assert msg.get("compose_id") or msg.get("id") or "compose" in str(msg), (
+        f"Message does not reference any compose: {msg}"
+    )
+    print(f"  ✓ Received compose-tagged message: {msg}")
+
+
+def test_kafka_compose_untagged(write_http_client, kafka_url):
+    """Untagging a compose must publish a message on the cts.compose-untagged topic."""
+    # Create a tag, import a compose, tag it, then untag it.
+    tag_data = create_tag(
+        write_http_client,
+        "kafka-untag-test",
+        "Tag for Kafka untagged test",
+        "https://example.com/docs/kafka-untag-test",
+    )
+    tag_name = tag_data["name"]
+
+    compose_data = import_compose(write_http_client, "KafkaUntaggedTest", "1.0", "20260603")
+    compose_id = compose_data["payload"]["compose"]["id"]
+    print(f"  Imported compose: {compose_id}, tag: {tag_name}")
+
+    tag_compose(write_http_client, compose_id, tag_name)
+    print(f"  Tagged compose with '{tag_name}'")
+
+    untag_compose(write_http_client, compose_id, tag_name)
+    print(f"  Untagged compose")
+
+    # Allow a short grace period for the async publisher.
+    time.sleep(2)
+
+    msg = _consume_kafka_message(kafka_url, "cts.compose-untagged")
+
+    assert msg is not None, "Expected a message on cts.compose-untagged, got None"
+    assert msg.get("event") == "compose-untagged", (
+        f"Expected event='compose-untagged', got event={msg.get('event')!r}"
+    )
+    assert msg.get("compose_id") or msg.get("id") or "compose" in str(msg), (
+        f"Message does not reference any compose: {msg}"
+    )
+    print(f"  ✓ Received compose-untagged message: {msg}")
