Request+Unittest接口测试框架（Python）

一、为何要写代码实现接口自动化

你们知道不少接口测试工具能够实现对接口的测试，如postman、jmeter、fiddler等等，并且使用方便，那么为何还要写代码实现接口自动化呢？工具虽然方便，但也不足之处：

测试数据不可控制

接口测试本质是对数据的测试，调用接口，输入一些数据，随后，接口返回一些数据。验证接口返回数据的正确性。在用工具运行测试用例以前不得不手动向数据库中插入测试数据。这样咱们的接口测试是否是就没有那么“自动化了”。

没法测试加密接口

这是接口测试工具的一大硬伤，如咱们前面开发的接口用工具测试彻底没有问题，但遇到须要对接口参 数进行加密/解密的接口，例如 md五、base6四、AES 等常见加密方式。本书第十一章会对加密接口进行介绍。 又或者接口的参数须要使用时间戳，也是工具很难模拟的。

扩展能力不足

当咱们在享受工具所带来的便利的同时，每每也会受制于工具所带来的局限。例如，我想将测试结果生 成 HMTL 格式测试报告，我想将测试报告发送到指定邮箱。我想对接口测试作定时任务。我想对接口测试作持续集成。这些需求都是工具难以实现的。

二、接口自动化测试设计

接口测试调用过程能够用下图归纳，增长了测试数据库

通常的接口工具测试过程：

一、接口工具调用被测系统的接口（传参 username="zhangsan"）。

二、系统接口根据传参（username="zhangsan"）向正式数据库中查询数据。

三、将查询结果组装成必定格式的数据，并返回给被调用者。

四、人工或经过工具的断言功能检查接口测试的正确性。

接口自动化测试项目，为了使接口测试对数据变得可控，测试过程以下：

一、接口测试项目先向测试数据库中插入测试数据（zhangsan 的我的信息）。

二、调用被测系统接口（传参 username="zhangsan"）。

三、系统接口根据传参（username="zhangsan"）向测试数据库中进行查询并获得 zhangsan 我的信息。

四、将查询结果组装成必定格式的数据，并返回给被调用者。

五、经过单元测试框架断言接口返回的数据（zhangsan 的我的信息），并生成测试报告。

为了使正式数据库的数据不被污染，建议使用独立的测试数据库。

二、requests库

Requests 使用的是 urllib3，所以继承了它的全部特性。Requests 支持 HTTP 链接保持和链接池，支持使用cookie保持会话，支持文件上传，支持自动肯定响应内容的编码。对request库的更详细的介绍能够看我以前接口测试基础的博客：

http://www.cnblogs.com/ailiailan/p/7388784.html

http://www.cnblogs.com/ailiailan/p/7412945.html

三、接口测试代码示例

下面以以前用python+django开发的用户签到系统为背景，展现接口测试的代码。

为何开发接口？开发的接口主要给谁来用？

前端和后端分离是近年来 Web 应用开发的一个发展趋势。这种模式将带来如下优点：

一、后端能够不用必须精通前端技术（HTML/JavaScript/CSS），只专一于数据的处理，对外提供 API 接口。

二、前端的专业性愈来愈高，经过 API 接口获取数据，从而专一于页面的设计。

三、先后端分离增长接口的应用范围，开发的接口能够应用到 Web 页面上，也能够应用到移动 APP 上。

在这种开发模式下，接口测试工做就会变得尤其重要了。

开发实现的接口代码示例：

 1 # 添加发布会接口实现
 2 def add_event(request):
 3     eid = request.POST.get('eid','')                 # 发布会id
 4     name = request.POST.get('name','')               # 发布会标题
 5     limit = request.POST.get('limit','')             # 限制人数
 6     status = request.POST.get('status','')           # 状态
 7     address = request.POST.get('address','')         # 地址
 8     start_time = request.POST.get('start_time','')   # 发布会时间
 9 
10     if eid =='' or name == '' or limit == '' or address == '' or start_time == '':
11         return JsonResponse({'status':10021,'message':'parameter error'})
12 
13     result = Event.objects.filter(id=eid)
14     if result:
15         return JsonResponse({'status':10022,'message':'event id already exists'})
16 
17     result = Event.objects.filter(name=name)
18     if result:
19         return JsonResponse({'status':10023,'message':'event name already exists'})
20 
21     if status == '':
22         status = 1
23 
24     try:
25         Event.objects.create(id=eid,name=name,limit=limit,address=address,status=int(status),start_time=start_time)
26     except ValidationError:
27         error = 'start_time format error. It must be in YYYY-MM-DD HH:MM:SS format.'
28         return JsonResponse({'status':10024,'message':error})
29 
30     return JsonResponse({'status':200,'message':'add event success'})

经过POST请求接收发布会参数：发布会id、标题、人数、状态、地址和时间等参数。

首先，判断eid、name、limit、address、start_time等字段均不能为空，不然JsonResponse()返回相应的状态码和提示。JsonResponse()是一个很是有用的方法，它能够直接将字典转化成Json格式返回到客户端。

接下来，判断发布会id是否存在，以及发布会名称（name）是否存在；若是存在将返回相应的状态码和 提示信息。

再接下来，判断发布会状态是否为空，若是为空，将状态设置为1（True）。

最后，将数据插入到 Event 表，在插入的过程当中若是日期格式错误，将抛出 ValidationError 异常，接收 该异常并返回相应的状态和提示，不然，插入成功，返回状态码200和“add event success”的提示。

 1 # 发布会查询接口实现
 2 def get_event_list(request):
 3 
 4     eid = request.GET.get("eid", "")      # 发布会id
 5     name = request.GET.get("name", "")    # 发布会名称
 6 
 7     if eid == '' and name == '':
 8         return JsonResponse({'status':10021,'message':'parameter error'})
 9 
10     if eid != '':
11         event = {}
12         try:
13             result = Event.objects.get(id=eid)
14         except ObjectDoesNotExist:
15             return JsonResponse({'status':10022, 'message':'query result is empty'})
16         else:
17             event['eid'] = result.id
18             event['name'] = result.name
19             event['limit'] = result.limit
20             event['status'] = result.status
21             event['address'] = result.address
22             event['start_time'] = result.start_time
23             return JsonResponse({'status':200, 'message':'success', 'data':event})
24 
25     if name != '':
26         datas = []
27         results = Event.objects.filter(name__contains=name)
28         if results:
29             for r in results:
30                 event = {}
31                 event['eid'] = r.id
32                 event['name'] = r.name
33                 event['limit'] = r.limit
34                 event['status'] = r.status
35                 event['address'] = r.address
36                 event['start_time'] = r.start_time
37                 datas.append(event)
38             return JsonResponse({'status':200, 'message':'success', 'data':datas})
39         else:
40             return JsonResponse({'status':10022, 'message':'query result is empty'})

经过GET请求接收发布会id和name 参数。两个参数都是可选的。首先，判断当两个参数同时为空，接口返回状态码10021，参数错误。

若是发布会id不为空，优先经过id查询，由于id的惟一性，因此，查询结果只会有一条，将查询结果 以 key:value 对的方式存放到定义的event字典中，并将数据字典做为整个返回字典中data对应的值返回。

name查询为模糊查询，查询数据可能会有多条，返回的数据稍显复杂；首先将查询的每一条数据放到一 个字典event中，再把每个字典再放到数组datas中，最后再将整个数组作为返回字典中data对应的值返回。

接口测试代码示例：

 1 #查询发布会接口测试代码
 2 import requests
 3 
 4 url = "http://127.0.0.1:8000/api/get_event_list/"
 5 r = requests.get(url, params={'eid':'1'})
 6 result = r.json()
 7 print(result)
 8 assert result['status'] == 200
 9 assert result['message'] == "success"
10 assert result['data']['name'] == "xx 产品发布会"
11 assert result['data']['address'] == "北京林匹克公园水立方"
12 assert result['data']['start_time'] == "2016-10-15T18:00:00"

由于“发布会查询接口”是GET类型，因此，经过requests库的get()方法调用，第一个参数为调用接口的URL地址，params设置接口的参数，参数以字典形式组织。

json()方法能够将接口返回的json格式的数据转化为字典。

接下来就是经过 assert 语句对接字典中的数据进行断言。分别断言status、message 和data的相关数据等。

使用unittest单元测试框架开发接口测试用例 

 1 #发布会查询接口测试代码 
 2 import unittest
 3 import requests  4  5 class GetEventListTest(unittest.TestCase):  6  7 def setUp(self):  8 self.base_url = "http://127.0.0.1:8000/api/get_event_list/"  9 10 def test_get_event_list_eid_null(self): 11 ''' eid 参数为空 ''' 12 r = requests.get(self.base_url, params={'eid':''}) 13 result = r.json() 14 self.assertEqual(result['status'], 10021) 15 self.assertEqual(result['message'], 'parameter error') 16 17 def test_get_event_list_eid_error(self): 18 ''' eid=901 查询结果为空 ''' 19 r = requests.get(self.base_url, params={'eid':901}) 20 result = r.json() 21 self.assertEqual(result['status'], 10022) 22 self.assertEqual(result['message'], 'query result is empty') 23 24 def test_get_event_list_eid_success(self): 25 ''' 根据 eid 查询结果成功 ''' 26 r = requests.get(self.base_url, params={'eid':1}) 27 result = r.json() 28 self.assertEqual(result['status'], 200) 29 self.assertEqual(result['message'], 'success') 30 self.assertEqual(result['data']['name'],u'mx6发布会') 31 self.assertEqual(result['data']['address'],u'北京国家会议中心') 32 33 def test_get_event_list_nam_result_null(self): 34 ''' 关键字‘abc’查询 ''' 35 r = requests.get(self.base_url, params={'name':'abc'}) 36 result = r.json() 37 self.assertEqual(result['status'], 10022) 38 self.assertEqual(result['message'], 'query result is empty') 39 40 def test_get_event_list_name_find(self): 41 ''' 关键字‘发布会’模糊查询 ''' 42 r = requests.get(self.base_url, params={'name':'发布会'}) 43 result = r.json() 44 self.assertEqual(result['status'], 200) 45 self.assertEqual(result['message'], 'success') 46 self.assertEqual(result['data'][0]['name'],u'mx6发布会') 47 self.assertEqual(result['data'][0]['address'],u'北京国家会议中心')
48
49if __name__ == '__main__':
50      unittest.main() 

unittest单元测试框架能够帮助组织和运行接口测试用例。

四、接口自动化测试框架实现

关于接口自动化测试，unittest 已经帮咱们作了大部分工做，接下来只须要集成数据库操做，以及HTMLTestRunner测试报告生成扩展便可。

框架结构以下图：

pyrequests 框架：

db_fixture/： 初始化接口测试数据。

interface/： 用于编写接口自动化测试用例。

report/： 生成接口自动化测试报告。

db_config.ini ： 数据库配置文件。

HTMLTestRunner.py unittest 单元测试框架扩展，生成 HTML 格式的测试报告。

run_tests.py ： 执行全部接口测试用例。

4.一、数据库配置

首先，须要修改被测系统将数据库指向测试数据库。以 MySQL数据库为例，针对django项目而言，修改.../guest/settings.py 文件。能够在系统测试环境单首创建一个测试库。这样作的目的是让接口测试的数据不会清空或污染到功能测试库的数据。其余框架开发的项目与django项目相似，这个工做通常由开发同窗完成，咱们测试同窗更多关注的是测试框架的代码。

4.二、框架代码实现

4.2.1、首先，创建数据库配置文件.../db_config.ini

4.2.2、接下来，简单封装数据库操做，数据库表数据的插入和清除，.../db_fixture/mysql_db.py

 1 import pymysql.cursors
 2 import os
 3 import configparser as cparser
 4 
 5 
 6 # ======== Reading db_config.ini setting ===========
 7 base_dir = str(os.path.dirname(os.path.dirname(__file__)))
 8 base_dir = base_dir.replace('\\', '/')
 9 file_path = base_dir + "/db_config.ini"
10 
11 cf = cparser.ConfigParser()
12 
13 cf.read(file_path)
14 host = cf.get("mysqlconf", "host")
15 port = cf.get("mysqlconf", "port")
16 db   = cf.get("mysqlconf", "db_name")
17 user = cf.get("mysqlconf", "user")
18 password = cf.get("mysqlconf", "password")
19 
20 
21 # ======== MySql base operating ===================
22 class DB:
23 
24     def __init__(self):
25         try:
26             # Connect to the database
27             self.connection = pymysql.connect(host=host,
28                                               port=int(port),
29                                               user=user,
30                                               password=password,
31                                               db=db,
32                                               charset='utf8mb4',
33                                               cursorclass=pymysql.cursors.DictCursor)
34         except pymysql.err.OperationalError as e:
35             print("Mysql Error %d: %s" % (e.args[0], e.args[1]))
36 
37     # clear table data
38     def clear(self, table_name):
39         # real_sql = "truncate table " + table_name + ";"
40         real_sql = "delete from " + table_name + ";"
41         with self.connection.cursor() as cursor:
42             cursor.execute("SET FOREIGN_KEY_CHECKS=0;")
43             cursor.execute(real_sql)
44         self.connection.commit()
45 
46     # insert sql statement
47     def insert(self, table_name, table_data):
48         for key in table_data:
49             table_data[key] = "'"+str(table_data[key])+"'"
50         key   = ','.join(table_data.keys())
51         value = ','.join(table_data.values())
52         real_sql = "INSERT INTO " + table_name + " (" + key + ") VALUES (" + value + ")"
53         #print(real_sql)
54 
55         with self.connection.cursor() as cursor:
56             cursor.execute(real_sql)
57 
58         self.connection.commit()
59 
60     # close database
61     def close(self):
62         self.connection.close()
63 
64     # init data
65     def init_data(self, datas):
66         for table, data in datas.items():
67             self.clear(table)
68             for d in data:
69                 self.insert(table, d)
70         self.close()
71 
72 
73 if __name__ == '__main__':
74 
75     db = DB()
76     table_name = "sign_event"
77     data = {'id':1,'name':'红米','`limit`':2000,'status':1,'address':'北京会展中心','start_time':'2016-08-20 00:25:42'}
78     table_name2 = "sign_guest"
79     data2 = {'realname':'alen','phone':12312341234,'email':'alen@mail.com','sign':0,'event_id':1}
80 
81     db.clear(table_name)
82     db.insert(table_name, data)
83     db.close()

首先，读取 db_config.ini 配置文件。 建立 DB 类，__init__()方法初始化，经过 pymysql.connect()链接数据库。

由于这里只用到数据库表的清除和插入，因此只建立 clear()和 insert()两个方法。其中，insert()方法对数 据的插入作了简单的格式转化，可将字典转化成 SQL 插入语句，这样格式转化了方便了数据库表数据的建立。

最后，经过 close()方法用于关闭数据库链接。

4.2.3、接下来接下来建立测试数据，.../db_fixture/test_data.py

 1 import sys
 2 sys.path.append('../db_fixture')
 3 try:
 4     from mysql_db import DB
 5 except ImportError:
 6     from .mysql_db import DB
 7 
 8 # create data
 9 datas = {
10     'sign_event':[
11         {'id':1,'name':'红米Pro发布会','`limit`':2000,'status':1,'address':'北京会展中心','start_time':'2017-08-20 14:00:00'},
12         {'id':2,'name':'可参加人数为0','`limit`':0,'status':1,'address':'北京会展中心','start_time':'2017-08-20 14:00:00'},
13         {'id':3,'name':'当前状态为0关闭','`limit`':2000,'status':0,'address':'北京会展中心','start_time':'2017-08-20 14:00:00'},
14         {'id':4,'name':'发布会已结束','`limit`':2000,'status':1,'address':'北京会展中心','start_time':'2001-08-20 14:00:00'},
15         {'id':5,'name':'小米5发布会','`limit`':2000,'status':1,'address':'北京国家会议中心','start_time':'2017-08-20 14:00:00'},
16     ],
17     'sign_guest':[
18         {'id':1,'realname':'alen','phone':13511001100,'email':'alen@mail.com','sign':0,'event_id':1},
19         {'id':2,'realname':'has sign','phone':13511001101,'email':'sign@mail.com','sign':1,'event_id':1},
20         {'id':3,'realname':'tom','phone':13511001102,'email':'tom@mail.com','sign':0,'event_id':5},
21     ],
22 }
23 
24 # Inster table datas
25 def init_data():
26     DB().init_data(datas)
27 
28 
29 if __name__ == '__main__':
30     init_data()

init_data()函数用于读取 datas 字典中的数据，调用 DB 类中的 clear()方法清除数据库，而后，调用 insert() 方法插入表数据。

4.2.4、编写接口测试用例。建立添加发布会接口测试文件.../interface/add_event_test.py

 1 import unittest
 2 import requests
 3 import os, sys
 4 parentdir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
 5 sys.path.insert(0, parentdir)
 6 from db_fixture import test_data
 7 
 8 
 9 class AddEventTest(unittest.TestCase):
10     ''' 添加发布会 '''
11 
12     def setUp(self):
13         self.base_url = "http://127.0.0.1:8000/api/add_event/"
14 
15     def tearDown(self):
16         print(self.result)
17 
18     def test_add_event_all_null(self):
19         ''' 全部参数为空 '''
20         payload = {'eid':'','':'','limit':'','address':"",'start_time':''}
21         r = requests.post(self.base_url, data=payload)
22         self.result = r.json()
23         self.assertEqual(self.result['status'], 10021)
24         self.assertEqual(self.result['message'], 'parameter error')
25 
26     def test_add_event_eid_exist(self):
27         ''' id已经存在 '''
28         payload = {'eid':1,'name':'一加4发布会','limit':2000,'address':"深圳宝体",'start_time':'2017'}
29         r = requests.post(self.base_url, data=payload)
30         self.result = r.json()
31         self.assertEqual(self.result['status'], 10022)
32         self.assertEqual(self.result['message'], 'event id already exists')
33 
34     def test_add_event_name_exist(self):
35         ''' 名称已经存在 '''
36         payload = {'eid':11,'name':'红米Pro发布会','limit':2000,'address':"深圳宝体",'start_time':'2017'}
37         r = requests.post(self.base_url,data=payload)
38         self.result = r.json()
39         self.assertEqual(self.result['status'], 10023)
40         self.assertEqual(self.result['message'], 'event name already exists')
41 
42     def test_add_event_data_type_error(self):
43         ''' 日期格式错误 '''
44         payload = {'eid':11,'name':'一加4手机发布会','limit':2000,'address':"深圳宝体",'start_time':'2017'}
45         r = requests.post(self.base_url,data=payload)
46         self.result = r.json()
47         self.assertEqual(self.result['status'], 10024)
48         self.assertIn('start_time format error.', self.result['message'])
49 
50     def test_add_event_success(self):
51         ''' 添加成功 '''
52         payload = {'eid':11,'name':'一加4手机发布会','limit':2000,'address':"深圳宝体",'start_time':'2017-05-10 12:00:00'}
53         r = requests.post(self.base_url,data=payload)
54         self.result = r.json()
55         self.assertEqual(self.result['status'], 200)
56         self.assertEqual(self.result['message'], 'add event success')
57 
58 
59 if __name__ == '__main__':
60     test_data.init_data() # 初始化接口测试数据
61     unittest.main()

在测试接口以前，调用test_data.py文件中的init_data()方法初始化数据库中的测试数据。

建立AddEventTest测试类继承 unittest.TestCase 类，经过建立测试用例，调用相关接口，并验证接口返回 的数据。

4.2.5、建立run_tests.py文件

当开发的接口达到必定数量后，就须要考虑分文件分目录的来划分接口测试用例，如何批量的执行不一样文件目录下的用例呢？unittest单元测试框架提供的discover()方法能够帮助咱们作到这一点。并使用 HTMLTestRunner 扩展生成 HTML 格式的测试报告。

 1 import time, sys
 2 sys.path.append('./interface')
 3 sys.path.append('./db_fixture')
 4 from HTMLTestRunner import HTMLTestRunner
 5 import unittest
 6 from db_fixture import test_data
 7 
 8 
 9 # 指定测试用例为当前文件夹下的 interface 目录
10 test_dir = './interface'
11 discover = unittest.defaultTestLoader.discover(test_dir, pattern='*_test.py')
12 
13 
14 if __name__ == "__main__":
15     test_data.init_data() # 初始化接口测试数据
16 
17     now = time.strftime("%Y-%m-%d %H_%M_%S")
18     filename = './report/' + now + '_result.html'
19     fp = open(filename, 'wb')
20     runner = HTMLTestRunner(stream=fp,
21                             title='Guest Manage System Interface Test Report',
22                             description='Implementation Example with: ')
23     runner.run(discover)
24     fp.close()

首先，经过调用test_data.py文件中的init_data()函数来初始化接口测试数据。

使用unittest框架所提供的discover()方法，查找 interface/ 目录下，全部匹配*_test.py 的测试文件（*星 号匹配任意字符）。

HTMLTestRunner 为unittest单元测试框架的扩展，利用它所提供的HTMLTestRunner()类来替换unittest单元测试框架的TextTestRunner()类，从而生成HTML格式的测试报告。

遗憾的是HTMLTestRunner并不支持Python3.x，你们能够在网上找到适用于Python3.x的HTMLTestRunner.py文件，使用在本身的接口自动化工程中。

经过 time 的 strftime()方法获取当前时间，而且转化成必定的时间格式。做为测试报告的名称。这样作目的是是为了不由于生成的报告的名称重名而形成报告的覆盖。最终，将测试报告存放于report/目录下面。以下图，一张完整的接口自动化测试报告。