Unittest+接口测试

Unittest+接口测试

1、通过Unittest实现用例的组织和执行,requests库实现接口测试。 2、目的:以云平台为例(http://www.nlecloud.com)实现注册、登录、更新APIKey、新增项目、新增设备、新增传感器、新增执行器、删除项目的接口自动化测试，并断言每条用例是否执行成功。

一：环境的部署

(1)安装requests、json、ddt、xlrd包,通过pip install xxx的方式。

二：目录结构说明

三:创建dataconfig包

(1) 在dataconfig文件下创建NLE.xlsx，内容如下所示:

四:创建utils公共类

(1)新建http_method.py文件封装requests基本方法

#coding=utf8 import requests,json class Method: #post、put、delete封装 #注意:get如果不传data的话，注意写先定义好data＝None，然后注意参数顺序，data＝None放在最后的位置。 def send_request(self,url,method,headers,data=None): if (type(headers) == str): headers = json.loads(headers) if (type(data) == dict): data = json.dumps(data) res_str = None if method == "post": res_str = requests.post(url=url,headers=headers,data=data) elif method == "put": res_str = requests.put(url=url,headers=headers,data=data) else: res_str = requests.delete(url=url,headers=headers,data=data) res_dict = json.loads(res_str.content) return res_dict

(2)创建path_config.py来定义路径

import os BASE_PATH = os.path.split(os.path.dirname(os.path.abspath(__file__)))[0] print(BASE_PATH) FILE_PATH = os.path.join(BASE_PATH, 'dataconfig','NLE.xlsx') TEST_PATH = os.path.join(BASE_PATH, 'test') print(TEST_PATH)

(3)创建opera_excel_1.py来操作基本的excel

import xlrd from utils.path_config import FILE_PATH class OperationExcel: def __init__(self,sheet_num): self.sheet_num = sheet_num # 获取table数据 def get_data(self): data = xlrd.open_workbook(FILE_PATH) tables = data.sheets()[self.sheet_num] return tables # 获取单元格的行数 def get_lines(self): tables = self.get_data() return tables.nrows # 获取某一个单元格的内容 def get_cell_value(self,row,col): return self.get_data().cell_value(row,col) #获取第一行的列数 def cols_count(self): title = self.get_data().row_values(0) return len(title) """ if __name__ == '__main__': s = OperationExcel(0) s.cols_count() """

五:创建data包

(1)创建get_Columns_2.py来获取excel中的列号

#封装获取列号方法，以后维护方便一些(目的：获取id，url，请求方法，是否运行，预算结果等所在excel中的列号) class global_var: id = 0 case_name = 1 URL = 2 request_way = 3 header = 4 json_data = 5 exspect = 6 #获取id所在列号 def id_colnum(): return global_var.id #获取case_name所在列号 def case_name_colnum(): return global_var.case_name #获取URL所在列号 def URL_colnum(): return global_var.URL #获取request_way所在列号 def request_way_colnum(): return global_var.request_way #获取header所在列号 def header_colnum(): return global_var.header #获取json_data所在列号 def json_data_colnum(): return global_var.json_data #获取respect所在列号 def exspect_colnum(): return global_var.exspect

(2)创建get_data_3.py来获取excel中的每一行

#获取EXCEl中数据 from utils.opera_excel_1 import OperationExcel from data import get_Columns_2 class GetData: def __init__(self,sheet_num): self.opera_excel = OperationExcel(sheet_num) # 获取EXCEL中所有数据 def data(self): arr_data1 = [] rows_count = self.opera_excel.get_lines() #行数循环 for i in range(1,rows_count): #列数循环 arr_data0 = [] for j in range(0,self.opera_excel.cols_count()): row_col_data = self.opera_excel.get_cell_value(i,j) arr_data0.append(row_col_data) arr_data1.append(arr_data0) # print(arr_data1[6]) return arr_data1 if __name__ == '__main__': s = GetData(0) s.data()

(3)创建analyze_data_4.py来解析数据

#对数据进行解析处理组合得到有效数据，并执行熨平 from data import get_Columns_2,get_data_3 import json class Analyze: #对获取到的数据进行解析处理,并执行用例。 def analyze_data(self): raw_data = get_data_3.GetData(0).data() arr_list = [] for i in range(0,len(raw_data)): request_url = raw_data[i][get_Columns_2.URL_colnum()] request_way = raw_data[i][get_Columns_2.request_way_colnum()] request_header = raw_data[i][get_Columns_2.header_colnum()] request_data = raw_data[i][get_Columns_2.json_data_colnum()] #去除字符串中的换行符\n和多余的空格 request_data = request_data.replace('\n','').replace(' ','') request_expect = raw_data[i][get_Columns_2.exspect_colnum()] arr_list.append([request_url,request_way,request_header,request_data,request_expect]) # print(arr_list[2]) # print(type(arr_list)) return arr_list if __name__ == '__main__': s = Analyze() s.analyze_data()

六:创建test包

(1)在test下创建test_case.py文件

#coding=utf8 import unittest from utils.http_method import Method from data.analyze_data_4 import Analyze import json from ddt import ddt,data,unpack @ddt class TestMethod(unittest.TestCase): get_data = Analyze() data_all = get_data.analyze_data() def setUp(self): self.run = Method() @data(data_all[0]) @unpack def test_01_register(self,*args): #args[3]中含有中文，将字符串转为二进制 res = self.run.send_request(args[0],args[1],args[2],args[3].encode('utf-8')) # print(res) self.assertDictContainsSubset(json.loads(args[4]),res,msg="测试用例失败") @data(data_all[1]) @unpack def test_02_login(self,*args): res = self.run.send_request(args[0],args[1],args[2],args[3]) # print(res) global UserID UserID = res['ResultObj']['UserID'] self.assertDictContainsSubset(json.loads(args[4]),res['ResultObj'],msg="测试用例失败") return res @data(data_all[2]) @unpack def test_03_up_apikey(self,*args): res_dict3 = json.loads(args[3]) res_dict3['OperUserID'] = res_dict3['UserID'] = UserID res= self.run.send_request(args[0],args[1],args[2],res_dict3) # print(res) self.assertDictContainsSubset(json.loads(args[4]),res,msg="测试用例失败") @data(data_all[3]) @unpack def test_04_login_again(self,*args): res = self.run.send_request(args[0],args[1],args[2],args[3]) # print(res) global Token Token = res['ResultObj']['AccessToken'] self.assertDictContainsSubset(json.loads(args[4]),res['ResultObj'],msg="测试用例失败") @data(data_all[4]) @unpack def test_05_new_project(self,*args): global res_dict2 res_dict2 = json.loads(args[2]) res_dict2['AccessToken'] = Token res = self.run.send_request(args[0],args[1],res_dict2,args[3]) # print(res) global ProjectID ProjectID = res['ResultObj'] self.assertDictContainsSubset(json.loads(args[4]),res,msg="测试用例失败") @data(data_all[5]) @unpack def test_06_new_device(self,*args): res_dict3 = json.loads(args[3]) res_dict3['ProjectIdOrTag'] = str(ProjectID) res = self.run.send_request(args[0],args[1],res_dict2,res_dict3) # print(res) global deviceId1 deviceId1 = res['ResultObj'] self.assertDictContainsSubset(json.loads(args[4]),res,msg="测试用例失败") @data(data_all[6]) @unpack def test_07_new_sensor(self,*args): str0 = args[0].replace("{deviceId}",str(deviceId1)) res = self.run.send_request(str0,args[1],res_dict2,args[3]) # print(res) self.assertDictContainsSubset(json.loads(args[4]),res,msg="测试用例失败") @data(data_all[7]) @unpack def test_08_new_actor(self,*args): str0 = args[0].replace("{deviceId}",str(deviceId1)) res = self.run.send_request(str0,args[1],res_dict2,args[3]) # print(res) self.assertDictContainsSubset(json.loads(args[4]),res,msg="测试用例失败") @data(data_all[8]) @unpack def test_09_delete_project(self,*args): str3 = args[3].replace("{ProjectId}",str(ProjectID)) res = self.run.send_request(args[0],args[1],res_dict2,str3) # print(res) self.assertDictContainsSubset(json.loads(args[4]),res,msg="测试用例失败") if __name__ == "__main__": unittest.main(verbosity=2)

七：创建run_case.py文件

import unittest from utils.path_config import TEST_PATH if __name__ == '__main__': discover = unittest.defaultTestLoader.discover(TEST_PATH, pattern='*.py') runner = unittest.TextTestRunner(verbosity=2) runner.run(discover)

八:执行结果

我们把excel中的第一条注册用例预期结果中的"Status": 1改为"Status": 0，然后再运行一下，可以看到第一条用例是执行Fail的

源码下载

地址:https://github.com/songteng2012/Interface

九:后续

(1)模块划分不太合理。因为case之间关联性比较强，要把test_case.py中的流程拆分出来分到别的包里写。–>待优化