1、前言:某些接口需要引用上个接口返回的值,作为下个接口的入参,但笔者又不想在本地维护及创建此文件,此时引出fixture内置函数中的临时文件存储tmpdir
2、首先下面的源码是使用flask框架写的图书管理系统,使用的basic认证方式,调用查看数据接口时需登录返回token
from flask import Flask,make_response,jsonify,abort,request
from flask_restful import Api,Resource
from flask_httpauth import HTTPBasicAuth
from flask import Flask
from flask_jwt import JWT, jwt_required, current_identity
from werkzeug.security import safe_str_cmp
app=Flask(__name__)
app.debug = True
app.config['SECRET_KEY'] = 'super-secret'
api = Api(app=app)
auth = HTTPBasicAuth()
@auth.get_password
def get_password(name):
if name == 'admin':
return 'admin'
@auth.error_handler
def authoorized():
return make_response(jsonify({'msg':"请认证"}),401)
books=[
{'id':1,'author':'Teacher','name':'Python接口自动化测试实战','done':True},
{'id':2,'author':'Teacher','name':'Selenium3自动化测试实战','done':False},
{'id':3,'author':'Tao','name':'Jmeter接口测试','done':False}
]
class User(object):
def __init__(self, id, username, password):
self.id = id
self.username = username
self.password = password
def __str__(self):
return "User(id='%s')" % self.id
users = [
User(1, 'admin', 'Admin'),
User(2, 'admin', 'qwe123'),
]
username_table = {u.username: u for u in users}
userid_table = {u.id: u for u in users}
def authenticate(username, password):
user = username_table.get(username, None)
if user and safe_str_cmp(user.password.encode('utf-8'), password.encode('utf-8')):
return user
def identity(payload):
user_id = payload['identity']
return userid_table.get(user_id, None)
jwt = JWT(app, authenticate, identity)
class Books(Resource):
# decorators = [auth.login_required]
decorators=[jwt_required()]
def get(self):
return jsonify({'status':0,'msg':'ok','datas':books})
def post(self):
if not request.json:
return jsonify({'status':1001,'msg':'请求参数不是JSON的数据,请检查,谢谢!'})
else:
book = {
'id': books[-1]['id'] + 1,
'author': request.json.get('author'),
'name': request.json.get('name'),
'done': True
}
books.append(book)
return jsonify({'status':1002,'msg': '添加书籍成功','datas':book}, 201)
class Book(Resource):
# decorators = [auth.login_required]
decorators = [jwt_required()]
def get(self,book_id):
book = list(filter(lambda t: t['id'] == book_id, books))
if len(book) == 0:
return jsonify({'status': 1003, 'msg': '很抱歉,您查询的书的信息不存在'})
else:
return jsonify({'status': 0, 'msg': 'ok', 'datas': book})
def put(self,book_id):
book = list(filter(lambda t: t['id'] == book_id, books))
if len(book) == 0:
return jsonify({'status': 1003, 'msg': '很抱歉,您查询的书的信息不存在'})
elif not request.json:
return jsonify({'status': 1001, 'msg': '请求参数不是JSON的数据,请检查,谢谢!'})
elif 'author' not in request.json:
return jsonify({'status': 1004, 'msg': '请求参数author不能为空'})
elif 'name' not in request.json:
return jsonify({'status': 1005, 'msg': '请求参数name不能为空'})
elif 'done' not in request.json:
return jsonify({'status': 1006, 'msg': '请求参数done不能为空'})
elif type(request.json['done'])!=bool:
return jsonify({'status': 1007, 'msg': '请求参数done为bool类型'})
else:
book[0]['author'] = request.json.get('author', book[0]['author'])
book[0]['name'] = request.json.get('name', book[0]['name'])
book[0]['done'] = request.json.get('done', book[0]['done'])
return jsonify({'status': 1008, 'msg': '更新书籍成功', 'datas': book})
def delete(self,book_id):
book = list(filter(lambda t: t['id'] == book_id, books))
if len(book) == 0:
return jsonify({'status': 1003, 'msg': '很抱歉,您查询的书的信息不存在'})
else:
books.remove(book[0])
return jsonify({'status': 1009, 'msg': '删除书籍成功'})
api.add_resource(Books,'/v1/api/books')
api.add_resource(Book,'/v1/api/book/<int:book_id>')
if __name__ == '__main__':
app.run(debug=True,port=5550)
3、其次:把上述的登录接口分离到conftest.py 模块中,实现fixture的共享机制
import pytest
import requests
# 获取flask_api.py模块的access_token 认证登录
@pytest.fixture()
def getToken():
login_url = "http://127.0.0.1:5550/auth"
data = {"username": "admin","password": "Admin"}
response = requests.post(url=login_url,json=data)
# 返回token值
return response.json()['access_token']
4、最后:下面举个Api接口书籍管理系统的为例(test_fixture_books_temdir.py 模块):
import requests
import pytest
# 每一条测试用例必须有清理工作,就是原来数据是怎么的,用例执行完后还是怎么样的
# 此设置是添加书籍信息返回的ID,然后传ID更新数书籍信息,再删除此更新后的书籍信息
def addBooks(getToken):
# 新增书籍信息
data = {
"author": "Teacher",
"name": "安全&渗透测试",
"done": False
}
r = requests.post(
url='http://127.0.0.1:5550/v1/api/books',
json=data,
headers={'Authorization':'JWT {}'.format(getToken)}
)
print('添加书籍: ',r.json())
return r.json()[0]['datas']['id']
def queryBooks(getToken,BookID):
# 查询书籍信息
r = requests.get(
url='http://127.0.0.1:5550/v1/api/book/{0}'.format(BookID),
headers={'Authorization':'JWT {}'.format(getToken)}
)
print('查询书籍: ',r.json())
return r
def setBooks(getToken,BookID):
# 更新书籍信息
data = {
"author":"TAO",
"name":"Fiddler工具",
"done":False
}
r = requests.put(
url='http://127.0.0.1:5550/v1/api/book/{0}'.format(BookID),
json=data,
headers={'Authorization':'JWT {}'.format(getToken)}
)
print('更新书籍: ',r.json())
return r
def deleteBooks(getToken,BookID):
# 删除新增书籍信息
r = requests.delete(
url='http://127.0.0.1:5550/v1/api/book/{0}'.format(BookID),
headers={'Authorization':'JWT {}'.format(getToken)}
)
print('删除书籍: ',r.json())
def test_query_books(getToken,tmpdir):
fp = tmpdir.join('book.md')
fp.write(addBooks(getToken))
queryBooks(getToken,BookID=fp.read())
setBooks(getToken,BookID=fp.read())
# 验证:查询更新后书籍信息
assert 'Fiddler' in setBooks(getToken,BookID=fp.read()).json()['datas'][0]['name']
assert queryBooks(getToken,fp.read()).status_code == 200
deleteBooks(getToken,BookID=fp.read())
if __name__ == '__main__':
pytest.main()
5、下面是接口运行结果:
(接口自动化需做到,创建返回A,查询返回A,更新返回A,删除依然是A)