Python环境下基于Kafka的契约测试实现方案咨询

核心问题排查 #

你当前使用的是pact的HTTP同步契约测试流程，完全不符合异步Kafka消息的测试场景，两个问题的根因都来自这里：

1. 强制要求HTTP响应码是因为你调用了will_respond_with这类专为HTTP场景设计的API，消息契约不需要定义HTTP相关属性
2. 时序错乱是因为你没有先完成Kafka消费者的初始化监听就触发了上游请求，加上你设置了auto.offset.reset='latest'，发送请求时还没开始监听的话，自然会漏掉这条消息

优化实现步骤 #

1. 依赖准备 #

确保你安装的是最新版pact-python，已内置消息契约支持，不需要额外依赖。

2. 修正后的测试代码 #

import atexit
import unittest
from pact import MessageConsumer, MessageProvider, MessagePact
from confluent_kafka import Consumer, Producer
import json
import threading
import time
import requests

# 定义契约双方
pact = MessagePact(
    consumer=MessageConsumer('KafkaConsumer'),
    provider=MessageProvider('KafkaProvider')
)

# Kafka配置
KAFKA_CONFIG = {
    'bootstrap.servers': 'your_kafka_server',
    'group.id': 'contract_test_group',
    'auto.offset.reset': 'earliest' # 改成earliest避免时序问题漏消费
}
TOPIC = 'usertopic'

# 消费者监听逻辑
def consume_user_message():
    consumer = Consumer(KAFKA_CONFIG)
    consumer.subscribe([TOPIC])
    try:
        # 最多等待10秒获取消息
        for _ in range(10):
            msg = consumer.poll(timeout=1.0)
            if msg is None:
                continue
            if msg.error():
                raise Exception(f"Consumer error: {msg.error()}")
            return json.loads(msg.value().decode('utf-8'))
    finally:
        consumer.close()

# 触发上游请求的逻辑
def trigger_user_request():
    requests.get('http://your-upstream-service/user/')

class KafkaContractTest(unittest.TestCase):
    expected_msg = {
        'username': 'UserA',
        'id': 123,
        'groups': ['Editors']
    }
    result = None

    def test_user_message_contract(self):
        # 定义消息契约，不需要任何HTTP相关配置
        (pact
         .given('UserA exists and is not an administrator')
         .expects_to_receive('a user info event for UserA')
         .with_content(self.expected_msg))

        with pact:
            # 先启动消费者监听线程
            consumer_thread = threading.Thread(target=lambda: setattr(self, 'result', consume_user_message()))
            consumer_thread.start()
            # 等待1秒确保消费者完成订阅
            time.sleep(1)
            # 再触发上游请求生产消息
            trigger_user_request()
            consumer_thread.join()

        # 校验消息匹配契约
        self.assertEqual(self.result, self.expected_msg)

if __name__ == '__main__':
    unittest.main()

3. 流程简化点 #

• 消费者侧执行测试后会自动生成契约文件，你只需要把契约文件同步给提供者部门即可
• 提供者侧不需要搭测试环境对接你的服务，只需要读取契约文件，校验自己生产的Kafka消息是否符合契约定义即可
• 后续迭代如果消息schema有变更，只需要更新契约重新走两边的校验流程，不需要跨部门联调就能发现兼容性问题

内容的提问来源于stack exchange，提问作者Zeruno