【发布时间】：2022-12-15 02:06:45
【问题描述】：
我正在尝试使用 Pact 进行消费者测试，但我很挣扎。如果有人可以在我出错的地方帮助我，我将不胜感激。
我要测试的文件如下：
import requests
from orders_service.exceptions import (
APIIntegrationError,
InvalidActionError
)
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
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
def cancel(self):
if self.status == 'progress':
response = requests.get(
f'http://localhost:3001/kitchen/schedule/{self.schedule_id}/cancel',
data={'order': self.items}
)
if response.status_code == 200:
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}')
def pay(self):
response = requests.post(
'http://localhost:3001/payments', data={'order_id': self.id}
)
if response.status_code == 200:
return
raise APIIntegrationError(
f'Could not process payment for order with id {self.id}'
)
def schedule(self):
response = requests.post(
'http://localhost:3000/kitchen/schedule',
data={'order': [item.dict() for item in self.items]}
)
if response.status_code == 201:
return response.json()['id']
raise APIIntegrationError(
f'Could not schedule order with id {self.id}'
)
def dict(self):
return {
'id': self.id,
'order': [item.dict() for item in self.items],
'status': self.status,
'created': self.created,
}
消费者测试我无法让它进入发布合同的阶段。有 2 个领域我不太熟悉，首先是 python fixture。我真的不确定这里需要做什么或如何做，最后是测试最底部的“consumer.cancel()”。
一些帮助让我设置和一个方式将不胜感激。这是我为测试写的：
import atexit
from datetime import datetime
import logging
import os
from uuid import UUID
import requests
import pytest
import subprocess
from pact import Consumer, Like, Provider, Term, Format
from orders_service.orders import Order, OrderItem
log = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO)
# If publishing the Pact(s), they will be submitted to the Pact Broker here.
# For the purposes of this example, the broker is started up as a fixture defined
# in conftest.py. For normal usage this would be self-hosted or using Pactflow.
PACT_BROKER_URL = "https://xxx.pactflow.io/"
PACT_BROKER_USERNAME = xxx
PACT_BROKER_PASSWORD = xxx
# Define where to run the mock server, for the consumer to connect to. These
# are the defaults so may be omitted
PACT_MOCK_HOST = "localhost"
PACT_MOCK_PORT = 1234
# Where to output the JSON Pact files created by any tests
PACT_DIR = os.path.dirname(os.path.realpath(__file__))
@pytest.fixture
def consumer() -> Order.cancel:
# return Order.cancel("http://{host}:{port}".format(host=PACT_MOCK_HOST, "port=PACT_MOCK_PORT))
order = [OrderItem(**{"id":1, "product":"coffee", "size":"big", "quantity":2})]
payload = Order(id=UUID, created=datetime.now, items=order, status="progress")
return Order.cancel(payload)
@pytest.fixture(scope="session")
def pact(request):
"""Setup a Pact Consumer, which provides the Provider mock service. This
will generate and optionally publish Pacts to the Pact Broker"""
# When publishing a Pact to the Pact Broker, a version number of the Consumer
# is required, to be able to construct the compatability matrix between the
# Consumer versions and Provider versions
# version = request.config.getoption("--publish-pact")
# publish = True if version else False
pact = Consumer("UserServiceClient", version=1).has_pact_with(
Provider("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()
# Make sure the Pact mocked provider is stopped when we finish, otherwise
# port 1234 may become blocked
atexit.register(pact.stop_service)
yield pact
# This will stop the Pact mock server, and if publish is True, submit Pacts
# to the Pact Broker
pact.stop_service()
# Given we have cleanly stopped the service, we do not want to re-submit the
# Pacts to the Pact Broker again atexit, since the Broker may no longer be
# available if it has been started using the --run-broker option, as it will
# have been torn down at that point
pact.publish_to_broker = False
def test_cancel_scheduled_order(pact, consumer):
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('get', f'http://localhost:3001/kitchen/schedule/{Like(12343)}/cancel')
.will_respond_with(200, body=Like(expected)))
with pact:
payload = Order(UUID, datetime.now, {"product":"coffee", "size":"large", "quantity":1}, "progress")
print(payload)
response = consumer.cancel(payload)
assert response['status'] == "cancelled"
pact.verify()
我最初也有（改编自契约中的例子）：
# return Order.cancel("http://{host}:{port}".format(host=PACT_MOCK_HOST, "port=PACT_MOCK_PORT))
但我不确定它是如何工作的
谢谢你帮助我
【问题讨论】：