python requests接口自动化测试（数据库断言）

前言

熟练掌握接口自动化测试体系背后的这些技能和处理问题的思路，实现时间、人力、收益的平衡，对于一个经验尚浅的初、中级测试开发人员来说绝对是一个艰巨的挑战。

五步教会你写接口自动化用例

需要安装三方包:requests pytest pytest-htmlpip install requests pytest pytest-html

导入requests模块
import requests
组装请求参数和数据
url = 'http://127.0.0.1:5000/add/'
params = {"a":3, "b":5} # get请求url参数, 字典格式
data = {"a":3, "b":5} # post请求请求数据, 字典格式
发送请求得到response对象
resp = requests.get(url=url, params=params)
resp = requests.post(url=url,data=data)
解析response对象resp.text # 获取响应文本
断言结果
assert resp.text == '8'
完整代码:

Copy# 1. 导入包
import requests
 
base_url = "http://127.0.0.1:5005"
 
# 2. 组装请求
def test_add_normal():
	# url  字符串格式
    url = base_url + "/add/"
    
	# data {} 字典格式
    data = {"a": "1", "b": "2"}
    # 3. 发送请求,获取响应对象
    response = requests.post(url=url, data=data)
# 4. 解析响应
# 5. 断言结果
    assert response.text == '3'

REST类接口自动测试方法#

请求格式为json

三处不同:

必须通过headers指定内容类型为application/json: ```headers={"Content-Type":"application/json"}
请求数据要转化为字符串: data=json.dumps(data) (使用json.dumps需要import json)
json格式的响应数据,在接口调试通过和稳定的情况下可以使用response.json()解析为字典格式,进行断言

完整代码:

Copy# 1. 导入包
import requests
import json
 
base_url = "http://127.0.0.1:5005"
 
def test_sub_normal():
    url = base_url + "/api/sub/"
    headers = {"Content-Type": "application/json"} # 1. 必须通过headers指定请求内容类型为json
    data = {"a": "4", "b": "2"}
    data = json.dumps(data) # 2. 序列化成字符串
    response = requests.post(url=url, headers=headers, data=data)
    # 3. 响应解析 # 响应格式为: {"code":"100000", "msg": "成功", "data": "2"}
    resp_code = response.json().get("code") 
    resp_msg = response.json().get("msg")
    resp_data = response.json().get("data")
    # 断言
    assert response.status_code == 200
    assert resp_code == "100000"
    assert resp_msg == "成功"
    assert resp_data == "2"

补充1: 感受Python黑科技之exec()动态生成用例:

数据文件: test_add_data.xls

TestCase	Url	Method	DataType	a	b	Excepted	Actual	Status
test_add_normal	/api/add/	GET	Url	3	5	8
test_add_zero	/api/add/	POST	FORM	0	0	0
test_add_negetive	/api/add/	POST	FORM	-3	5	2
test_add_float	/api/add/	POST	FORM	3.2	5.2	8.4
test_add_null	/api/add/	POST	FORM			0

Copyimport requests
import xlrd
 
base_url = 'http://127.0.0.1:5005'
 
# 1.打开excel
wb = xlrd.open_workbook("test_add_data.xls")
# 2. 获取sheet
sh = wb.sheet_by_index(0)  # wb.sheet_by_name("Sheet1")
# 行数 sh.nrows 列数 sh.ncols
# 获取单元格数据
# print(sh.cell(1,0).value)
# print(sh.nrows)
tpl = '''def {case_name}():
    url = base_url + '/add/'
    data = {{"a": "{a}", "b":"{b}"}}
    response = requests.get(url=url, data=data)
    assert response.text == '{expected}'
'''
 
for row in range(1, sh.nrows):
    case_name = sh.cell(row,0).value
    a = sh.cell(row, 4).value
    b = sh.cell(row, 5).value
    expected = sh.cell(row, 6).value
    case = tpl.format(case_name=case_name, a=a, b=b, expected=expected)
    exec(case)

动态生成的用例支持pytest发现和执行

自动化接口用例运行

将自动化测试用例保存为test*.py,一个文件里可以写多个用例, 如上面两个接口保存为test_user.py
在自动化用例脚本所在目录打开命令行, 运行pytest(运行所有test开头的.py用例)或pytest test_user.py(只运行该脚本中用例)
pytest一些参数-q: 安静模式(不显示环境信息) pytest -q test_user.py--html=report.html:执行完生成report.html报告(文件名可以自己指定)pytest test_user.py --html=test_user_report.html--resultlog=test.log: 执行完成生成执行结果log文件pytest test_user.py --resultlog=run.log

requests库详解

请求方法#

requests.get()
requests.post()
requests.delete()
.....
requests.session() # 用来保持session会话,如登录状态

请求参数#

url: 接口地址, str url="http://127.0.0.1:5000/add/"
headers: 请求头, dict headers={"Content-Type": "application/json"}
params: url参数, dict params={"a":"1":"b":"2"}
data: 请求数据, dict data={"a":"1":"b":"2"}
files: 文件句柄, dict files={"file": open("1.jpg")}
timeout: 超时时间,单位s, str, 超过时间会报超时错误```requests.get(url=url,params=params,timeout=10)

响应解析

resp: 响应对象
resp.status_code: 响应状态码
resp.text # 响应文本
resp.json() # 响应转化为json对象(字典)-慎用:如果接口出错或返回格式不是json格式,使用这个方法会报错
resp.content # 响应内容, 二进制类型

接口安全验证,参数化及断言

各种类型接口的测试

GET请求接口

Copyrequests.get(url=url, params=params)

表单类型

Copyrequests.post(url=url, data=data)

REST类型

Copyrequests.post(url=url, headers={"Content-Type": "application/json"}, data=json.dumps(data)

上传文件

Copyrequests.post(url=url, files={"file": open("1.jpg", "rb")})

Session依赖

Copysession=requests.session(); session.post(...); session.post()

接口依赖

接口依赖之动态中间值

Copyresp=requests.get(...);token=resp.split("=")[1];resp2=requests.post(....token...)

验签接口

Copyimport hashlib
 
def md5(str):
    m = hashlib.md5()
    m.update(str.encode('utf8'))
    return m.hexdigest()  #返回摘要，作为十六进制数据字符串值
 
def makeSign(params):
    if not isinstance(params, dict):
        print("参数格式不正确，必须为字典格式")
        return None
    if 'sign' in params:
        params.pop('sign')
    sign = ''
    for key in sorted(params.keys()):
        sign = sign + key + '=' + str(params[key]) + '&'
    sign = md5(sign + 'appsecret=' + appsecret)
    params['sign'] = sign
    return params

Copydata = makeSign(data);resp = requests.post(url=url, headers=headers, data=json.dumps(data))

接口依赖之Mock Server

Mock和单元测试的桩(Stub)类似, 是通过建立一个模拟对象来解决依赖问题的一种方法.

应用场景:
1. 依赖第三方系统接口, 无法调试
2. 所依赖接口尚未具备(前后端分离项目, 前端开发时,后端接口尚未开发完毕)
3. 所依赖接口需要修改或不稳定
4. 依赖接口较多或场景复杂, 所依赖接口不是主要验证目标的

解决方法:
1. 通过Mock.js/RAP/RAP2来动态生成, 模拟接口返回数据
2. 自己使用Flask大家简单的Mock接口
3. 使用Python自带的mock库

Copy...

SOAP接口

pip install suds

Copyfrom suds.client import Client
 
ip = '127.0.0.1'
port = '5001'
 
client = Client("http://%s:%s/?wsdl" % (ip, port))
result = client.service.addUser("张790", "123456")
print(result)

XML-RPC接口

Copyimport xmlrpc.client
 
user = xmlrpc.client.ServerProxy('http://127.0.0.1:5002')
print(user.getAll())

参数化

参数化是用来解决动态参数问题的

数据文件参数化

csv数据文件优点：以逗号分隔，轻量级缺点：参数过多不便于区分

Copyimport csv
 
csv_data = csv.reader(open('data/reg.csv'))

config数据文件优点：方便支持多组数据，支持备注

Copyimport configparser
	cf=configparser.ConfigParser()
	cf.read('data/reg.config', encoding='utf-8')
	cf.sections()
	cf.options(section)
	cf.get(section,option)

json数据文件优点：REST接口常用数据格式，格式清楚，适用于多参数及含有嵌套参数缺点：不支持备注，多组数据不清晰

Copyimport json
with open('data/reg.json', encoding='utf-8') as f:
    json_data = json.loads(f)  #json_data为列表或字典

Copyjson的序列化和反序列化
需求：python的字典/列表等数据格式为内存对象，需要做存储（持久化）和进行数据交换
 
序列化: 从内存对象到可存储数据, 方便存储和交换数据
	json.dumps: 列表/字典/json->字符串 ```str_data = json.dumps(dict_data)```
	json.dump: 列表/字典/json->数据文件 ```json.dump(dict_data, open(data_file, "w"))```
反序列化: 从存储数据到内存对象
	json.loads: 字符串->字典/列表```json.loads('{"a":1,"b":2}') #得到字典{"a":1,"b":2}```
	json.load: json数据文档->列表/字典```dict_data = json.load(open('data.json'))```

excel数据文件优点：直观，构造数据方便缺点：嵌套数据不方便格式化

pip install xlrd

Copyimport xlrd
 
wb=xlrd.open_workbook("data/reg.xlsx")
sh=wb.sheet_by_index(0)
sh=wb.sheet_by_name('Sheet1")
sh.nrows
sh.ncols
sh.cell(x,y).value

xml数据文件优点：方便自定义多层结构，SOAP,RPC通用格式

Copyfrom xml.dom.minidom import parse
dom=parse('data/reg.xml')
root=dom.documentElement
user_nodes=root.getElementsByTagName("user")
user_node.getAttribute('title')
user_node.hasAttribute('title')
name_node=user_node.getElementsByTagName('name')[0]
name=name_node.childNodes[0].data

案例1: 自动执行excel用例并将结果回写excel

数据文件: test_user.xlsx

TestCase	Url	Method	DataType	Data	Excepted	Resp.text	Status
test_user_reg_normal	/api/user/reg/	POST	JSON	{"name":"九小1","passwd": "123456"}	resp.json()["code"]=="100000"
test_user_login_normal	/api/user/login/	POST	FORM	{"name":"九小1","passwd": "123456"}	"成功" in resp.text

Copyimport xlrd
from xlutils.copy import copy
import json
import requests
import sys
 
base_url = "http://127.0.0.1:5000"
 
def run_excel(file_name, save_file="result.xls"):
    wb=xlrd.open_workbook(file_name)
    sh=wb.sheet_by_index(0)
 
    wb2 = copy(wb)
    sh2 = wb2.get_sheet(0)
 
 
    for i in range(1,sh.nrows):
        url = base_url + sh.cell(i,1).value
        data = json.loads(sh.cell(i,4).value)
        headers = {}
        method = sh.cell(i,2).value
        data_type = sh.cell(i,3).value
        excepted = sh.cell(i,5).value
        if data_type.lower() == 'json':
            data = json.dumps(data)
            headers = {"Content-Type":"application/json"}
 
        if method.lower() == "get":
            resp = requests.get(url=url,headers=headers)
        else:
            resp = requests.post(url=url,headers=headers,data=data)
        if eval(excepted):
            status = "PASS"
        else:
            status = "FAIL"
        sh2.write(i,6, resp.text)
        sh2.write(i,7, status)
 
    wb2.save(save_file)
    print("保存成功")
        
        
if __name__ == "__main__":
    if len(sys.argv)==2:
        run_excel(sys.argv[1])
    elif len(sys.argv)>2:
        run_excel(sys.argv[1],sys.argv[2])
    else:
        print("调用格式: python run_excel.py 用例文件 输出文件")

保存脚本为run_excel.py, (最好和数据文件test_user.xlsx放在同一目录下), 在脚本所在目录打开命令行,运行

Copypython run_excel.py test_user.xlsx

生成的result.xls预览

TestCase	Url	Method	DataType	Data	Excepted	Resp.text	Status
test_user_reg_normal	/api/user/reg/	POST	JSON	{"name":"九小1","passwd": "123456"}	resp.json()["code"]=="100000"	{"code":"100001","data":{"name":"\u4e5d\u5c0f1","passwod":"e10adc3949ba59abbe56e057f20f883e"},"msg":"\u5931\u8d25\uff0c\u7528\u6237\u5df2\u5b58\u5728"}	FAIL
test_user_login_normal	/api/user/login/	POST	FORM	{"name":"九小1","passwd": "123456"}	"成功" in resp.text	<h1>登录成功</h1>	PASS

随机数据参数化

import random

随机数random.random(): 随机0，1random.randint(0,100): 随机整数random.randrange(0,100,2): 随机偶数random.uniform(1,100): 随机浮点数
随机选择random.choice('abcdefg')random.choice(['赵','钱','孙','李','周'])

Copy随机姓名的实现:
#随机汉字: chr(random.randint(0x4e00, 0x9fbf))
name=random.choice(['赵','钱','孙','李','周'])+chr(random.randint(0x4e00, 0x9fbf)

随机样本random.sample('abcdefg', 2)

随机2个字符拼接: ''.join(random.sample('abcdefg',2)

洗牌random.shuffle([1, 2, 3, 4, 5, 6]): 随机改版列表数据

断言/检查点

断言/检查点是用来自动验证接口返回数据/业务操作的结果是否满足预期

响应断言

正则表达式

元字符. : 任意字符\d: 任意数字 - \D: 任意非数字\w: 任意字符或数字 - \W: 任意非字符及数字\s: 任意空白字符(包含空格或\n等) - \S: 任意非空白字符^: 匹配开头$: 匹配结尾[]: 匹配其中任何一个字符{n,m}: 匹配n-m个重复: 匹配重复任意次(包含0次): 匹配重复至少一次? : 匹配0或1次(): 分组,获取需要的部分数据| : 或, 匹配多个pattern\元字符: 取消元字符转义
贪婪匹配及非贪婪匹配
系统函数re.findall(): re.S,支持跨行re.match()re.search()re.sub()/re.subn()re.complie()

数据库断言

从数据库读取数据,验证数据库数据是否符合预期

MySQL断言

pip install pymysql

导入pymysql: import pymysql
建立数据库连接:

Copyconn = pymysql.connect(host='',port=3306,db='',user='',passwd='',charset='utf8')

从连接建立操作游标: cur=conn.cursor()
使用游标执行sql命令: cur.execute("select * from user")
获取执行结果:cur.fetchall(): 获取所有结果cur.fetchmany(3): 获取多条结果cur.fetchone(): 获取一条结果
关闭游标及连接(先关游标再关连接):cur.close();conn.close()

PostgreSQL

pip install pyscopg2

Copyimport pyscopg2
conn=pyscopg2.connect(host='',port='',dbname='',user='',password='') # 注意是dbname, password
cur=conn.curser()
cur.execute("...")
cur.fetchall()

Oracle

pip install cx_Oracle

Copy...

Mongodb

pip install pymongo

Copyfrom pymongo import MongoClient
 
conn = MongoClient('', 27017)
db = conn.mydb
my_set = db.test_set
 
for i in my_set.find({"name": "张三"}):
    print(i)
 
print(my_set.findone({"name": "张三"}))

Redis断言

pip install redis

Copyimport redis
 
r = redis.Redis(host='192.168.100.198', port=6379, password='!QE%^E2sdf23RGF@ml239', db=0)
print(r.dbsize())
print(r.get('package'))
print(r.hget('event_order_advance_008aea6a62115ec4923829ee09f76a9c18243f5d', 'user'))

服务器断言

pip install paramik

Copyimport paramiko
 
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())  
client.connect('192.168.100.241', 22, username='root', password='1234567', timeout=4)
stdin, stdout, stderr = client.exec_command('cat /proc/meminfo')
print(stdout.read())
client.close()

完整用例:

Copyimport requests
import pytest
import json
import hashlib
import re
 
 
def md5(string):
    m = hashlib.md5()
    m.update(string.encode('utf8'))
    return m.hexdigest()
 
def makeSign(data):
    sign=''
    for key in sorted(data.keys()):
        sign += key + '=' + str(data[key]) + '&'
    sign += 'appsecret=NTA3ZTU2ZWM5ZmVkYTVmMDBkMDM3YTBi'
    data['sign'] = md5(sign)
    return data
 
class DB():
    def __init__(self):
        # 建立连接
        self.conn = pymysql.connect(host='localhost',port=3307,user='root',passwd='',db='api',charset='utf8')
 
        # 建立一个游标
        self.cur = self.conn.cursor()
    
    def __del__(self):
        self.cur.close()
        self.conn.close()
 
    def getUserByName(self, name):
        self.cur.execute("select * from user where name='%s'" % name)
        return self.cur.fetchone()
 
    def checkUser(self, name, passwd):
        user = self.getUserByName(name)
        if user:
            if user[2] == md5(passwd):
                return True
            else:
                return False
        else:
            return None
            
class TestUser(): # pytest识别不能用__init__方法
    base_url = 'http://127.0.0.1:5000'
    db = DB()
 
    def test_login(self):
        url = self.base_url + '/api/user/login/'
        data = {"name": "张三", "passwd": "123456"}
        resp = requests.post(url=url, data=data)
 
        #断言
        assert resp.status_code == 200
        assert '登录成功' in resp.text
 
    def test_reg(self):
        url = self.base_url + '/api/user/reg/'
        headers = {"Content-Type": "application/json"}
        data = {'name': '张10', 'passwd': '123456'}
        resp = requests.post(url=url, headers=headers, data=json.dumps(data))
 
        #断言
        assert resp.json()['code'] == '100000'
        assert resp.json()['msg'] == '成功'
        assert self.db.getUserByName('张10')
 
 
    def test_uploadImage(self):
        url = self.base_url + '/api/user/uploadImage/'
        files = {'file': open("复习.txt")}
        resp = requests.post(url=url, files=files)
 
        #断言
        assert resp.status_code == 200
        assert '成功' in resp.text
        # todo 服务器断言
 
    def test_getUserList(self):
        session = requests.session()
        login_url = self.base_url + '/api/user/login/'
        login_data = {"name": "张三", "passwd": "123456"}
        session.post(url=login_url, data=login_data)
 
        url = self.base_url + '/api/user/getUserList/'
        resp = session.get(url=url)
 
        #断言
        assert resp.status_code == 200
        assert '用户列表' in resp.text
        assert re.findall('\w{32}',t2) != []
 
    def test_updateUser(self):
        session = requests.session()  # 接口依赖的接口需要用session
        get_token_url = self.base_url + '/api/user/getToken/'
        params = {"appid": '136425'}
        token_resp = session.get(url=get_token_url, params=params)
        assert re.findall('token=\w{32}$')
        token = token_resp.text.split('=')[1]
 
        url = self.base_url + '/api/user/updateUser/?token=' + token
        data = {'name': '张三', 'passwd': '234567'}
        headers = {"Content-Type": "application/json"}
        resp = session.post(url=url, headers=headers, data=json.dumps(data))
 
        #断言
        assert resp.status_code == 200
        assert resp.json()['code'] == '100000'
        assert resp.json()['msg'] == '成功'
        assert self.db.checkUser('张三', '234567')
 
    def test_delUser(self):
        url = self.base_url + '/api/user/delUser/'
        headers = {"Content-Type": "application/json"}
        data = {'name': '张10', 'passwd': '123456'}
        data = makeSign(data)
        resp = requests.post(url=url, headers=headers, data=json.dumps(data))
 
        #断言
        assert resp.status_code == 200
        assert resp.json()['code'] == '100000'
        assert resp.json()['msg'] == '成功' 
        assert not self.db.getUserByName('张10')
 
 
if __name__ == '__main__':
    t = TestUser()
    # t.test_updateUser()
    # t.test_updateUser()
    t.test_delUser()
    # pytest.main("-q test_user2.py")

结语

这篇贴子到这里就结束了，最后，希望看这篇帖子的朋友能够有所收获。欢迎留言，或是关注我的专栏和我交流。

作为一个软件测试的过来人，我想尽自己最大的努力，帮助每一个伙伴都能顺利找到工作。所以我整理了下面这份资源，现在免费分享给大家，有需要的小伙伴可以关注【公众号：开心螺蛳粉】自提！