Pact是一个契约测试框架，有多种语言实现，本文主要介绍模块pact-python进行契约测。
官网：https://docs.pact.io/implementation_guides/python/readme
安装命令：pip install pact-python
安装过程中如果报错，安装失败，需要进如pact包下载官网，找到对应的ZIP包自行下载，
https://github.com/pact-foundation/pact-ruby-standalone/releases
放到pact\bin目录下，再重新用命令pip install pact-python 安装
此外需要安装另外两个依赖包：pip install requests 和pip install Flask
Step 1: 写一个 customer：customer_success.py
import requests
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 6.1; WOW64; rv:23.0) Gecko/20100101 Firefox/23.0',
"Content-Type": "application/json",
}
data={
"data": {
"transactionType": "C", #
"brnNo": "9998199930", #
"taxInvoiceNumber": "202106161221150000000001",
"supplierBrnNo":"9998199954",
"creationDate": "20210616", #
"taxInvoiceType": "11", #
"supplyPrice": 75000000000, #
"taxAmount": 320000000, #
"buyerBrnNo": "9998199916",
# 非必填
"mainItems": "전산장비",
"remark1": "하드웨어 10종",
"remark2": "소프트웨어 30종",
"supplierCmpName": "서울데이터시스템",
"buyerCmpName": "종각은행"
},
"header": {
"chanlId": "IDC",
"grpId": "GROUPID001",
"messageTimeStamp": '2021-05-17T20:00:20.123Z',
"trackingId": "ABC123DEF456"
}
}
url = 'http://127.0.0.1:8080'
def OLTP_inquiry():
api = '/api/v2/coperator/taxInvoice'
f_url = url + api
print("请求接口：",f_url)
json_data = data
print("请求参数：",json_data)
resp = requests.post(f_url, json=json_data,headers=headers)
print("响应参数：",resp.json())
if __name__ == '__main__':
OLTP_inquiry()
Step 2: 写一个 customer：customer_error.py
import requests
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 6.1; WOW64; rv:23.0) Gecko/20100101 Firefox/23.0',
"Content-Type": "application/json",
}
data={
"data": {
"transactionType": "C", # C-create,目前只支持C
"brnNo": "9998199930", # BRN
"taxInvoiceNumber": "202106161221150000000000", # BRN
"supplierBrnNo":"9998199954",
"creationDate": "20210616", #
"taxInvoiceType": "11", # 11-Electronic Tax Invoice，21-Electronic Tax Invoice(Exempt from taxation)，31-Electronic Tax Invoice(Customs)
"supplyPrice": 75000000000, # Supply Price in KRW
"taxAmount": 320000000, # Tax Amount in KRW
"buyerBrnNo": "9998199916",
# 非必填
"mainItems": "전산장비",
"remark1": "하드웨어 10종",
"remark2": "소프트웨어 30종",
"supplierCmpName": "서울데이터시스템",
"buyerCmpName": "종각은행"
},
"header": {
"chanlId": "IDC",
"grpId": "GROUPID001",
"messageTimeStamp": '2021-05-17T20:00:20.123Z',
"trackingId": "ABC123DEF456"
}
}
url = 'http://127.0.0.1:8080'
def OLTP_inquiry_error():
api = '/api/v2/coperator/taxInvoice'
f_url = url + api
print("请求接口：",f_url)
json_data = data
print("请求参数：",json_data)
resp = requests.post(f_url, json=json_data,headers=headers)
print("响应参数：",resp.json())
if __name__ == '__main__':
OLTP_inquiry_error()
Step 3: 写一个 provider：oltp_service.py
from flask import Flask,request
app = Flask(__name__)
@app.route('/api/v2/coperator/taxInvoice',methods=["POST"])
def create_tax_invoice():
resp_success ={
"result": {
"trackingId": "test",
"messageTimeStamp": "2021-06-16T15:45:59.123Z",
"chanlId": "IDC",
"grpId": "group id"
},
"data": {
"ackSignal": "0"
}
}
resp_error ={
"result": {
"trackingId": "test",
"messageTimeStamp": "2021-06-16T15:45:59.123Z",
"chanlId": "IDC",
"grpId": "group id"
},
"data": {
},
"error": {
"code": "1234modulenm",
"description": "사업자번호가 존재 하지 않습니다.",
"descriptionKr": "Not Exist Business Registration Number"
}
}
data = request.get_json()
print(data)
resp_success['result'] = data['header']
resp_error['result'] = data['header']
tax_no = data['data']['taxInvoiceNumber']
if tax_no[-1:]=='0':
return resp_error
else:
return resp_success
if __name__ == '__main__':
app.run(port=8080)
Step 4: 获取预期结果
先运行oltp_service.py文件启动provider服务，然后分别运行customer_success.py和customer_error.py
最终得到正常类和异常类两种预期结果。
Step 5: 写一个正常类和一个错误类案例
根据上面两个预期结果编写两个对应的测试用例
正常类测试用例：invoice_tax_pact.py
import atexit
import requests
import unittest
from pact.consumer import Consumer
from pact.provider import Provider
from pact import Like
# 定义一个pact，消费者是ModuleB，生产者是ModuleA，契约文件存放在pacts文件夹下
pact = Consumer('customer_success').has_pact_with(Provider('oltp_service'), pact_dir='../pacts')
# 启动服务
pact.start_service()
atexit.register(pact.stop_service)
# 测试用例
class UserTesting(unittest.TestCase):
def test_service(self):
post_data = {
"data": {
"transactionType": "C",
"brnNo": "9998199930",
"taxInvoiceNumber": "202106161221150000000001",
"supplierBrnNo": "9998199954",
"creationDate": "20210616", #
"taxInvoiceType": "11",
"supplyPrice": 75000000000,
"taxAmount": 320000000,
"buyerBrnNo": "9998199916",
"mainItems": "전산장비",
"remark1": "하드웨어 10종",
"remark2": "소프트웨어 30종",
"supplierCmpName": "서울데이터시스템",
"buyerCmpName": "종각은행"
},
"header": {
"chanlId": "IDC",
"grpId": "GROUPID001",
"messageTimeStamp": '2021-05-17T20:00:20.123Z',
"trackingId": "ABC123DEF456"
}
}
# 消费者定义的期望结果
expected = {
"result": {
"trackingId": "test",
"messageTimeStamp": "2021-06-16T15:45:59.123Z",
"chanlId": "IDC",
"grpId": "group id"
},
"data": {
"ackSignal": "0"
}
}
# 消费者定义的契约的实际内容。包括请求参数、请求方法、请求头、响应值等
(pact
.given('test service.')
.upon_receiving('post request for oltp_service')
.with_request(method='post', path='/api/v2/coperator/taxInvoice',body=post_data,
headers={'Content-Type': 'application/json'})
# 具体校验内容
.will_respond_with(200, body={
'data': {
"ackSignal": "0"
},
"result": {
"trackingId": Like('hello world'), # string类型
"chanlId": "IDC",
"grpId": "GROUPID001"
},
}))
# pact自带一个mock服务，端口 1234,用requests向mock接口发送请求，验证mock的结果是否正确
with pact:
res = requests.post("http://localhost:1234/api/v2/coperator/taxInvoice",json=post_data).json()
self.assertEqual(res['data'], expected['data'])
if __name__ == "__main__":
ut = UserTesting()
ut.test_service()
Step 5: 运行测试案例获取契约文件
运行上面的invoice_tax_pact.py文件后，会在对应pacts目录生成契约文件
Step 6: 运行服务和测试案例
先启动第三步创建的服务oltp_service.py
然后打开终端，进入契约文件目录，运行命令
pact-verifier --provider-base-url=http://127.0.0.1:8080 --pact-url=customer_success-oltp_service.json
