Python Pact消费者测试搭建及契约发布问题排查

核心问题汇总 #

你当前的代码存在4个直接导致测试无法走到契约发布阶段的硬伤：

• 业务代码里所有第三方服务地址硬编码写死为localhost:3001/localhost:3000，但Pact mock服务默认跑在1234端口，业务请求根本打不到mock服务上
• consumer fixture写法完全错误：Order.cancel是实例方法不是类方法，不能直接传参调用；初始化Order时传入的是UUID类、datetime.now函数引用而非实际值，实例初始化阶段就会报错
• Pact交互定义错误：不能把Like()这类匹配器直接拼在URL字符串里，且定义的全路径请求和mock服务监听地址不匹配，Pact无法识别路由规则
• Order类的status属性存在基础bug：__init__里直接给self.status赋值会触发property getter查找不存在的self._status，初始化直接抛AttributeError

第一步：修复业务代码的基础问题 #

首先把硬编码的服务地址改成可配置形式，同时修复status属性的bug，否则测试永远无法正常运行：

import os
import requests
from orders_service.exceptions import APIIntegrationError, InvalidActionError

# 服务地址做成可配置，测试时替换为Pact mock地址
KITCHEN_SERVICE_BASE_URL = os.getenv("KITCHEN_SERVICE_URL", "http://localhost:3001")
PAYMENT_SERVICE_BASE_URL = os.getenv("PAYMENT_SERVICE_URL", "http://localhost:3001")

class OrderItem:
    def __init__(self, id, product, quantity, size):
        self.id = id
        self.product = product
        self.quantity = quantity
        self.size = size

    def dict(self):
        return {
            'product': self.product,
            'size': self.size,
            'quantity': self.quantity
        }

class Order:
    def __init__(self, id, created, items, status, schedule_id=None,
                 delivery_id=None, order_=None):
        self._order = order_
        self._id = id
        self._created = created
        self.items = [OrderItem(**item) for item in items]
        self._status = status # 直接给私有属性赋值，避免触发property报错
        self.schedule_id = schedule_id
        self.delivery_id = delivery_id

    @property
    def id(self):
        return self._id or self._order.id

    @property
    def created(self):
        return self._created or self._order.created

    @property
    def status(self):
        return self._status or self._order.status
    
    @status.setter
    def status(self, value):
        self._status = value # 补全setter，否则赋值会报错

    def cancel(self):
        if self.status == 'progress':
            response = requests.get(
                # 替换硬编码地址为配置值
                f'{KITCHEN_SERVICE_BASE_URL}/kitchen/schedule/{self.schedule_id}/cancel',
                data={'order': [item.dict() for item in self.items]}
            )
            if response.status_code == 200:
                # 注意你原来的逻辑这里直接返回None，不返回响应体
                return
            raise APIIntegrationError(
                f'Could not cancel order with id {self.id}'
            )
        if self.status == 'delivery':
            raise InvalidActionError(f'Cannot cancel order with id {self.id}')

    # pay、schedule方法同理，把硬编码地址替换为配置的基地址即可

你之前注释的官方示例代码Order.cancel("http://{host}:{port}")，核心作用就是把Pact mock服务的地址传入客户端，替换掉硬编码的真实服务地址，让业务请求能打到Pact启动的mock服务上，这是Pact测试能生效的前提。

第二步：修正pytest fixture写法 #

fixture的作用是提前初始化测试依赖的通用对象，避免每个用例重复写初始化逻辑，不需要在fixture里直接调用业务方法。

import atexit
from datetime import datetime
import logging
import os
import uuid
import pytest
from pact import Consumer, Like, Provider, Term
from orders_service.orders import Order, OrderItem

log = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO)

PACT_BROKER_URL = "https://xxx.pactflow.io/"
PACT_BROKER_USERNAME = "xxx"
PACT_BROKER_PASSWORD = "xxx"
PACT_MOCK_HOST = "localhost"
PACT_MOCK_PORT = 1234
PACT_DIR = os.path.dirname(os.path.realpath(__file__))

# 测试前把服务基地址替换为Pact mock地址
os.environ["KITCHEN_SERVICE_URL"] = f"http://{PACT_MOCK_HOST}:{PACT_MOCK_PORT}"

@pytest.fixture(scope="session")
def pact(request):
    pact = Consumer("OrderService", version="1.0.0").has_pact_with(
        Provider("KitchenService"), # 名称要和你实际调用的厨房服务对应，不要随便写UserService
        host_name=PACT_MOCK_HOST,
        port=PACT_MOCK_PORT,
        pact_dir=PACT_DIR,
        publish_to_broker=True,
        broker_base_url=PACT_BROKER_URL,
        broker_username=PACT_BROKER_USERNAME,
        broker_password=PACT_BROKER_PASSWORD,
    )
    pact.start_service()
    yield pact
    pact.stop_service()
    pact.publish_to_broker = False
    # 删掉原来的atexit注册逻辑，重复停止服务会抛异常导致流程中断

@pytest.fixture
def scheduled_order():
    # 直接返回一个符合测试场景的Order实例即可，不要在这里调用cancel方法
    test_items = [{"id":1, "product":"coffee", "size":"big", "quantity":2}]
    return Order(
        id=uuid.uuid4(),
        created=datetime.now(),
        items=test_items,
        status="progress",
        schedule_id=12343
    )

第三步：修正测试用例逻辑 #

注意几个规则：Pact的with_request只需要写路径部分，不需要写全域名；匹配器不能直接f-string拼接到路径里，要用Term这类规则匹配器定义路径格式；with pact上下文退出时会自动做校验，不需要手动调用pact.verify()。

def test_cancel_scheduled_order(pact, scheduled_order):
    expected = {
        "id": "1e54e244-d0ab-46ed-a88a-b9e6037655ef",
        "order": [
            {"product": "coffee", "quantity": 1, "size": "small"}
        ],
        "scheduled": "Wed, 22 Jun 2022 09:21:26 GMT",
        "status": "cancelled"
    }

    (pact
        .given('A scheduled order exists and it is not cancelled already')
        .upon_receiving('a request for cancellation')
        .with_request(
            method='GET',
            # 用Term定义路径匹配规则，不要把匹配器直接拼字符串
            path=Term(
                matcher=r'/kitchen/schedule/\d+/cancel',
                generate=f'/kitchen/schedule/{scheduled_order.schedule_id}/cancel'
            )
        )
        # 你的cancel方法只判断200状态码，不需要额外校验响应体
        .will_respond_with(200))

    with pact:
        # 直接调用实例的cancel方法即可，不需要额外传参
        scheduled_order.cancel()
    # 走到这里说明请求符合Pact定义的交互规则，测试通过后会自动生成契约并发布

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