写点什么

Python 在接口测试中的应用

作者:向阳逐梦
  • 2022-10-25
    四川
  • 本文字数:7760 字

    阅读完需:约 25 分钟

1. 介绍

接口测试的方式有很多,可以使用的工具有 jmeter,postman,soapUI 等,也可以自己写代码进行接口测试(Python,java,go 等等),工具的使用相对来说都比较简单,开箱即用。但如果接口中定义了一些需要加解密、数字签名等一些动态计算时,工具就很无助,唯有写代码进行实现测试。所以要根据接口协议,有针对性的进行选择工具,当工具不适合项目时需要进行工具开发。对于工具开发选择语言,语言本身没有优劣之分,选择自己擅长的和合适的。相对于 java,go 语言,Python 不需要编译,而且支持的库很多,是其特点;其支持 http 协议接口的有 urllib,urllib2 以及 requests 库,支持 websocket 协议的库有 websockets、websocket-client,支持 grpc 的库有 grpcio、grpcio-tools,以及支持图形界面的 PyQt5。本文中使用的 Python 版本为 3.9.0,下面简单介绍下 http、websocket、grpc 接口示例,最后介绍使用 PyQt5 打造自己的测试工具。

2. http 协议接口

对于 http 协议接口,requests 库用起来最方便,因此采用 requests 库来做 http 协议的接口测试。安装 requests:

pip install requests
复制代码

2.1 GET 请求示例

天气查询接口:http://wthrcdn.etouch.cn/weather_mini

import requestsr = requests.get('http://wthrcdn.etouch.cn/weather_mini')print(r.text)
复制代码

输出结果:'{"status":1002,"desc":"invilad-citykey"}'

2.2 带参数的 GET 请求示例

参数:city 城市名称

import requestsdata ={}data['city']='北京'r = requests.get('http://wthrcdn.etouch.cn/weather_mini',params=data)print(r.text)
复制代码

输出结果:'{"data":{"yesterday":{"date":"31 日星期二","high":"高温 27℃","fx":"南风","low":"低温 17℃","fl":"","type":"小雨"},"city":"北京","forecast":[{"date":"1 日星期三","high":"高温 29℃","fengli":"","low":"低温 18℃","fengxiang":"南风","type":"晴"},{"date":"2 日星期四","high":"高温 30℃","fengli":"","low":"低温 20℃","fengxiang":"南风","type":"晴"},{"date":"3 日星期五","high":"高温 30℃","fengli":"","low":"低温 22℃","fengxiang":"南风","type":"阴"},{"date":"4 日星期六","high":"高温 26℃","fengli":"","low":"低温 21℃","fengxiang":"南风","type":"中雨"},{"date":"5 日星期天","high":"高温 26℃","fengli":"","low":"低温 20℃","fengxiang":" 东风","type":"小雨"}],"ganmao":"感冒易发期,外出请适当调整衣物,注意补充水分。","wendu":"23"},"status":1000,"desc":"OK"}'

假如服务对请求中 city 的一致性进行校验,header 中需要传入 Authorization,为 city 值的 md5

import hashlibdata ={}data['city']='合肥'auth=hashlib.md5(data['city'].encode(encoding='utf-8')).hexdigest()headers = {'Authorization': auth}r = requests.get('http://wthrcdn.etouch.cn/weather_mini',params=data,headers=headers)print(r.text)
复制代码

输出结果:'{"data":{"yesterday":{"date":"31 日星期二","high":"高温 35℃","fx":"南风","low":"低温 25℃","fl":"","type":"晴"},"city":"合肥","forecast":[{"date":"1 日星期三","high":"高温 30℃","fengli":"","low":"低温 24℃","fengxiang":"北风","type":"小雨"},{"date":"2 日星期四","high":"高温 26℃","fengli":"","low":"低温 23℃","fengxiang":"西北风","type":"阴"},{"date":"3 日星期五","high":"高温 29℃","fengli":"","low":"低温 23℃","fengxiang":"东风","type":"多云"},{"date":"4 日星期六","high":"高温 29℃","fengli":"","low":"低温 24℃","fengxiang":"东风","type":"小雨"},{"date":"5 日星期天","high":"高温 31℃","fengli":"","low":"低温 24℃","fengxiang":"东北风","type":"小雨"}],"ganmao":"感冒低发期,天气舒适,请注意多吃蔬菜水果,多喝水哦。","wendu":"26"},"status":1000,"desc":"OK"}'

2.3 POST 请求示例

本示例使用 requests 官网示例,POST 请求不再多介绍,可参考官网。Github API v3 接受编码为 JSON 的 POST 数据

import jsonurl = 'https://api.github.com/some/endpoint'payload = {'some': 'data'}r = requests.post(url, data=json.dumps(payload))
复制代码

3. Websocket 协议接口

安装 websockets 库

pip install websockets
复制代码

下面是一个 NLP 接口的例子,客户端连接服务端,发送文本,接收返回结果,其中文本需要进行 base64 加密。

import uuidimport websockets
async def send_msg(websocket): text = '我要查余额' bs = str(base64.b64encode(text.encode('utf-8')), 'utf-8') sid = uuid.uuid4() timestamp = str(int(time.time()) * 1000) msg = '{"app":"test","sn":"123","timeStamp":"' + timestamp + '","version":"1.0.0","data":"' + bs + '",' \ '"sid":"' + str(sid) + '"} ' print(msg) await websocket.send(msg) rev_text = await websocket.recv() print(rev_text)
async def client(): async with websockets.connect('ws://xxx.xxx.com/nlp') as websocket: await send_msg(websocket)
if __name__ == '__main__': asyncio.get_event_loop().run_until_complete(client())
复制代码

执行结果示例:{"appid":"test","sn":"123","timeStamp":"1630586234000","version":"1.0.0","data":"5oiR6KaB5p+l6K+i5L2Z6aKd",sid":"4c6f5a64-2374-4616-9385-376c0d25bc1e"}{"code":"0000","msg":"succ","nlp":"{"sceneKey":"","sid":"d28c7631-54d9-493e-8db8-f67a5ecc2a25","uid":"d28c7631-54d9-493e-8db8-f67a5ecc2a25","text":"我要查询余额",,"semantic":[{"intent":"querybalance","intentName":"查询余额","intentCnxt":"","template":"{Query}余额","score":0.865519,"slots":[]}],"dialog":{"status":0,"uid":"","reaction":{"text":""},"slots":null},"answer":{"answerType":"","text":"","emotion":"","topic":"","topicID":"","type":"","question":{"question":"","question_ws":""}}}","extendData":"","sid":"d28c7631-54d9-493e-8db8-f67a5ecc2a25","last":true}

4. grpc 协议接口

grpc 协议需要 Python 3.5 版本以上,安装所需的库:

pip install grpciopip install grpcio-tools
复制代码

下面以虚拟主播服务接口为例,服务使用了 grpc 双向流进行交互,与服务端建立连接后,会返回一个 rtmp 地址,推送虚拟主播视频流,向服务端发送文本,会驱动虚拟主播播报文本。

4.1 编写 avatar.proto 文件

syntax = "proto3";
package protocol;
message AvatarRequest { string app = 1; string timestamp = 2; string version = 3; string anchorId = 4; string vcn = 5; bytes data = 6; int32 width = 7; int32 height = 8; string type=9;
}
message AvatarResponse { string code = 1; string msg = 2; string type = 3; bytes data = 4; string status = 5; string sessionId = 6;}
service AvatarInteraction { rpc Interact (stream AvatarRequest) returns (stream AvatarResponse) { }}
复制代码

4.2 使用工具生成 grpc 代码

python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. ./avatar.proto 最终生成 avatar_pb2.py 和 avatar_pb2_grpc.py 两个文件

4.3 grpc 请求示例

将上述两个文件,引入到项目代码中

import sysimport threadingimport timeimport grpcimport avatar_pb2 as pbimport avatar_pb2_grpc as avatar

class avatarClient(threading.Thread): def __init__(self): threading.Thread.__init__(self) self.status = False self.serverUrl = "" self.anchorId = '1001' self.vcn = 'xiaoyuan' self.app = 'testapp' self.type = '0' self.sendTxt = False self.text = "" self.connectType = '1001' self.pingType = '-1000' self.width = 1080 self.height = 1920 self.sid = "" self.print_res = printRes() self.lastPingTime = int(time.time() * 1000)
# 连接服务 def connect_sever(self): return pb.AvatarRequest(app=self.app, timestamp=str(int(time.time() * 1000)), version='1.0.0', anchorId=self.anchorId, vcn=self.vcn, type=self.connectType, width=self.width, height=self.height)
# 发送文本 def send_text(self): return pb.AvatarRequest(app=self.app, timestamp=str(int(time.time() * 1000)), version='1.0.0', anchorId=self.anchorId, vcn=self.vcn, type='3', data=str(self.text).encode('utf-8'))
# 发送ping def send_ping(self): return pb.AvatarRequest(app=self.app, timestamp=str(int(time.time() * 1000)), version='1.0.0', anchorId=self.anchorId, vcn=self.vcn, type=self.pingType)
def send_message(self): yield self.connect_sever() #间隔10秒,发送ping,进行连接保活 while True: if self.sendTxt: yield self.send_text() self.sendTxt = False if (int(time.time() * 1000) - int(self.lastPingTime)) < 10000: continue else: yield self.send_ping() self.lastPingTime = int(time.time() * 1000)
def run(self): self.status = True if self.serverUrl == "": return with grpc.insecure_channel(self.serverUrl) as channel: stub = avatar.AvatarInteractionStub(channel) self.channel = channel responses = stub.Interact(self.send_message())
self.print_res.setClient(responses, self) self.print_res.start() time.sleep(1) while True: if self.print_res.runStatus == False: break client.text = input("input your text:\n") client.sendTxt = True time.sleep(1)
def stop(self): try: self.status = False self.print_res.runStatus = False time.sleep(1) except KeyboardInterrupt: pass

class printRes(threading.Thread): def __init__(self): threading.Thread.__init__(self) self.runStatus = True
def setClient(self, responses, client): self.responses = responses self.client = client
def run(self): try: for response in self.responses: print(response) if not self.runStatus: sys.exit(0) if response.code != "000000": self.runStatus = False sys.exit(0) except Exception: print("err")

if __name__ == '__main__': client = avatarClient() client.serverUrl = "xxx.xxx.xxx.xxx:xxxx" client.start()
复制代码

5. 打造自己的测试工具

在编辑器中执行测试,有时候交互性不太友好,因此我们可以打造一个带交互界面的测试工具。QT 是跨平台 C++库的集合,它实现高级 API 来访问现代桌面和移动系统的许多方面。这些服务包括定位和定位服务、多媒体、NFC 和蓝牙连接、基于铬的 web 浏览器以及传统的 UI 开发。PyQt5 是一套针对 QtV5 的全面 Python 绑定。它被实现为超过 35 个扩展模块,并使 Python 可以作为一个替代的应用程序开发语言 C++在所有支持的平台上包括 IOS 和 Android。

5.1 安装 PyQt5

pip install PyQt5
复制代码

如果要使用图像界面进行设计,可以安装 Qt Designer 图形界面开发工具,相关配置可以自行搜索。

pip install PyQt5-tools
复制代码

5.2 打造 websocket 测试工具

下面以 websocket 接口为例,实现一个 nlp 的测试工具

# -*- coding: utf-8 -*-import sysimport asyncioimport base64import timeimport uuidimport websocketsfrom PyQt5 import QtCore, QtWidgetsfrom PyQt5.QtWidgets import *

class Ui_mainWindow(object): def __init__(self, mainWindow): self.mainWindow = mainWindow self.setupUi()
def setupUi(self): self.mainWindow.setObjectName("mainWindow") self.mainWindow.setWindowModality(QtCore.Qt.WindowModal) self.mainWindow.resize(600, 300) self.mainWindow.setWindowTitle("NLP测试工具")
self.centralwidget = QtWidgets.QWidget(self.mainWindow) self.centralwidget.setObjectName("centralwidget")
self.widget1 = QtWidgets.QWidget(self.centralwidget) self.widget1.setGeometry(QtCore.QRect(10, 30, 400, 25)) self.widget1.setObjectName("widget1") self.horizontalLayout = QtWidgets.QHBoxLayout(self.widget1) self.horizontalLayout.setContentsMargins(0, 0, 0, 0) self.horizontalLayout.setObjectName("horizontalLayout")
self.serverUrlLabel = QtWidgets.QLabel(self.widget1) self.serverUrlLabel.setObjectName("label") self.serverUrlLabel.setText("服务地址:") self.horizontalLayout.addWidget(self.serverUrlLabel)
self.serverUrlLineEdit = QtWidgets.QLineEdit(self.widget1) self.serverUrlLineEdit.setEnabled(True) self.serverUrlLineEdit.setText("ws://xxx..xxx.xxx/nlp") self.serverUrlLineEdit.setMinimumWidth(320) self.serverUrlLineEdit.setMaximumWidth(320) self.serverUrlLineEdit.setObjectName("serverUrlLineEdit") self.horizontalLayout.addWidget(self.serverUrlLineEdit, stretch=500, alignment=QtCore.Qt.AlignLeft)
self.sendTextWidget = QtWidgets.QWidget(self.centralwidget) self.sendTextWidget.setGeometry(QtCore.QRect(0, 70, 500, 70)) self.sendTextWidget.setObjectName("sendTextWidget") self.sendTextLayout = QtWidgets.QHBoxLayout(self.sendTextWidget) self.sendTextLayout.setContentsMargins(0, 0, 0, 0) self.sendTextLayout.setObjectName("sendTextLayout")
self.sendTextLabel = QtWidgets.QLabel(self.sendTextWidget) self.sendTextLabel.setContentsMargins(20, 0, 0, 0) self.sendTextLabel.setText("文本:") self.sendTextLayout.addWidget(self.sendTextLabel)
self.sendTextTextEdit = QtWidgets.QTextEdit(self.sendTextWidget) self.sendTextTextEdit.setObjectName("sendTextTextEdit") self.sendTextLayout.addWidget(self.sendTextTextEdit)
self.sendTextButton = QtWidgets.QPushButton(self.sendTextWidget) self.sendTextButton.setObjectName("sendText") self.sendTextButton.setText("发送") self.sendTextButton.clicked.connect(self.send_text) self.sendTextLayout.addWidget(self.sendTextButton)
self.widget3 = QtWidgets.QWidget(self.centralwidget) self.widget3.setGeometry(QtCore.QRect(0, 160, 420, 90)) self.widget3.setObjectName("widget3")
self.logLayout = QtWidgets.QHBoxLayout(self.widget3) self.logLayout.setContentsMargins(0, 0, 0, 0) self.logLayout.setObjectName("logLayout")
self.logLabel = QtWidgets.QLabel(self.widget3) self.logLabel.setObjectName("logLabel") self.logLabel.setContentsMargins(20, 0, 0, 0) self.logLabel.setLineWidth(30) self.logLabel.setText("返回:") self.logLayout.addWidget(self.logLabel)
self.logTextEdit = QtWidgets.QTextEdit(self.widget3) self.logTextEdit.setContentsMargins(0, 0, 0, 0) self.logTextEdit.setObjectName("logTextEdit") self.logLayout.addWidget(self.logTextEdit)
self.mainWindow.setCentralWidget(self.centralwidget) self.statusbar = QtWidgets.QStatusBar(self.mainWindow) self.statusbar.setObjectName("statusbar") self.mainWindow.setStatusBar(self.statusbar)
QtCore.QMetaObject.connectSlotsByName(self.mainWindow)
def send_text(self): self.websocketClient = WebsocketClient() self.websocketClient.url = self.serverUrlLineEdit.text() self.websocketClient.text = self.sendTextTextEdit.toPlainText() result = self.websocketClient.run() self.logTextEdit.setText(result)
复制代码


class WebsocketClient:    def __init__(self):        self.url = ""
async def send_msg(self, websocket): if self.text == "": return "empty text" bs = str(base64.b64encode(self.text.encode('utf-8')), 'utf-8') sid = uuid.uuid4() timestamp = str(int(time.time()) * 1000) msg = '{"app":"test","sn":"123","timeStamp":"' + timestamp + '","version":"1.0.0","data":"' + bs + '",' \ '"sid":"' + str(sid) + '"} ' await websocket.send(msg) rev_text = await websocket.recv() self.result = rev_text self.text = "" return rev_text
async def connect(self): try: async with websockets.connect(self.url) as websocket: res = await self.send_msg(websocket) return res except Exception as e: return str(e)
def run(self): return asyncio.get_event_loop().run_until_complete(self.connect())

if __name__ == "__main__": app = QApplication(sys.argv) window = QMainWindow() box = Ui_mainWindow(window) window.show() sys.exit(app.exec_())
复制代码

执行结果:

5.3 虚拟主播工具介绍

上一节中,介绍了虚拟主播服务 grpc 接口示例,那么可以使用 PyQt5 打造虚拟主播的测试工具,此处不再提供示例代码,运行如下图:


发布于: 刚刚阅读数: 3
用户头像

向阳逐梦

关注

人生享受编程,编程造就人生! 2022-06-01 加入

某公司芯片测试工程师,嵌入式开发工程师,InfoQ签约作者,阿里云星级博主,华为云·云享专家。座右铭:向着太阳,追逐梦想!

评论

发布
暂无评论
Python在接口测试中的应用_学习方法_向阳逐梦_InfoQ写作社区