[大数据]接口自动化测试中的数据处理

配置文件

采用 yaml 作为配置文件，将一些重要的配置数据，如 数据库配置、host配置、相应权限的账号数据 放到 yaml文件中

conf.yaml

db_info:
  dbname: swiper
  host: 192.168.0.103
  port: 3306
  user: bobo
  pwd: hh123456


ApiHost:
  mockhost: http://127.0.0.1:4523/mock/894992
  testhost: http://10.1.134.98:8787
  prehost: http://10.1.100.10:8080
  
user:
  phonemun: 15911112223
  pwd: bobo123456
	

封装读取配置文件

YamlHandler.py

import yaml

# 读取yaml文件
class ReadYaml:
    def __init__(self, path, param=None):
        self.path = path  # 文件路径
        self.param = param  # 不传默认获取所有数据

    # 获取yaml文件中的数据
    def get_data(self, encoding='utf-8'):
        with open(self.path, encoding=encoding) as f:
            data = yaml.load(f.read(), Loader=yaml.FullLoader)

            if self.param == None:
                return data  # 返回所有数据
            else:
                return data.get(self.param)  # 获取键为param的值

数据的准备和记录

与数据库交互进行数据替换

接口自动化测试项目中，有些数据需要从数据库中获取

以最简单的注册接口为例，手机号不能在系统数据库中已存在，那么在excel文档中，可以用一些特殊字符先标注出来，然后再用代码进行替换

与数据库进行交互，需要用到PyMySQL模块，详细介绍可查看该篇文章：PyMySQL

DBHandler.py

import pymysql
from pymysql.cursors import DictCursor



class MysqlUtil:

    def __init__(self,dbconf):
        self.dbconf = dbconf
        self.conn = self.get_conn()  # 连接对象
        self.cursor = self.get_cursor()  # 游标对象

    def get_conn(self):
        """ 获取连接对象 """
        conn = pymysql.connect(host=self.dbconf['host'],
                               port=self.dbconf['port'],
                               user=self.dbconf['user'],
                               passwd=self.dbconf['pwd'],
                               db=self.dbconf['dbname'],
                               charset='utf8')
        return conn

    def get_cursor(self):
        """获取游标对象"""
        # cursor = None
        cursor = self.conn.cursor(cursor=pymysql.cursors.DictCursor)
        return cursor

    def query(self, sql, args=None, one=True):
        '''查询语句，默认只查询一条'''
        try:
            # 执行sql语句
            self.cursor.execute(sql, args)
            # 提交事务（使得每次游标都在初始位置）
            self.conn.commit()
            # 获取结果
            if one:
                return self.cursor.fetchone()
            else:
                return self.cursor.fetchall()

        except:
            # 若出现错误，则回滚
            self.conn.rollback()


    def commit_data(self, sql):
        """
        提交数据(更新、插入、删除操作)
        """
        try:
            # 执行sql语句
            self.cursor.execute(sql)
            # 提交事务
            self.conn.commit()
        except:
            # 若出现错误，则回滚
            self.conn.rollback()


    def close(self):
        self.cursor.close()
        self.conn.close()

编写一个生成手机号码的函数

helper.py

import random


def gen_phonenun():
    '''自动生成手机号码'''
    phone = '1' + random.choice(['3','4','5,','7','8','9'])
    for i in range(9):
        num = random.randint(0,9)
        phone += str(num)
    return phone

测试用例文件

test_register.py

import json
import unittest
import os
from common import helper
from common import ddt
from common.RequestHandler import HTTPHandler
from common.excel_handler import ExcelHandler
from common import dir_config
from common.DBhandler import MysqlUtil
from common.YamlHandler import ReadYaml


@ddt.ddt
class Test_Register(unittest.TestCase):
    # 读取测试数据
    excel_handle = ExcelHandler(os.path.join(dir_config.testdatas_dir, "data.xlsx"))
    test_data = excel_handle.read_key_value("register")


    def setUp(self) -> None:
        self.req = HTTPHandler()
        self.db_info = ReadYaml(dir_config.yaml_dir,'db_info').get_data()
        self.db = MysqlUtil(self.db_info)
        self.host = ReadYaml(dir_config.yaml_dir,'ApiHost').get_data()['mockhost']


    def tearDown(self) -> None:
        self.req.close_session()
        self.db.close()



    @ddt.data(*test_data)
    def test_register(self,test_data):

        # 判断excel中是否 出现了 #new_phone# ，如果出现，随机生成一个，进行替换
        if '#new_phone#' in str(test_data["json"]):
            while True:
                mobilephone = helper.gen_phonenun()
                sql1 = '''SELECT *  FROM users WHERE phonenum = %s ;'''
                dbphone = self.db.query(sql=sql1,args=mobilephone)
                if not dbphone:
                    break
            test_data["json"] = test_data["json"].replace('$new_phone$', mobilephone)

        # 判断excel中是否 出现了 $exist_phone$ ，如果出现，随机生成一个，进行替换
        if '#exist_phone#' in str(test_data["json"]):
            sql2 = '''select phonenum from users limit 1;'''
            mobilephone = self.db.query(sql=sql2)["phonenum"]
            test_data["json"] = test_data["json"].replace('$exist_phone$', mobilephone)

        

        res =  self.req.visit(
                              url= self.host+test_data["url"], 
                              method=test_data["method"],
                              headers=eval(test_data["headers"]),
                              json=json.loads(test_data["json"])
        )

        self.assertEqual(res["code"],test_data["excepted"])

如果excel中出现了#new_phone#，则通过gen_phonenun方法生成一个手机号mobilephone，在数据库中查询手机号mobilephone是否存在。若不存在，则使用该号码，通过replace函数，将代码读取的test_data["json"]数据中的#new_phone#替换为手机号；若存在，则继续 while True 循环，直到生成的手机号在数据库中不出存在，跳出循环

如果excel中出现了#exist_phone#，则在数据库中查询一条已注册成功的手机号，通过replace函数，将代码读取的test_data["json"]数据中的#exist_phone#替换为手机号

Context环境管理方式

在接口测试中，大多数接口，需要获取登录成功后的一些权限校验或信息校验的内容，比如 token，或者一些接口需要用到key，而这些内容，在多个接口中会经常用到。这些内容，称为用例关联或用例依赖，在面试中，有时会问到，这种类型的数据怎么处理

例如，登录接口响应成功，返回一些用户信息

{
    "code": "0000",
    "msg": {
        "result": "success",
        "info": "登录成功",
        "errorinfo": null,
        "userinfo": {
            "memberid": "20012345644",
            "nickname": "沉觞",
            "type": 1,
            "permission": true,
            "tokeninfo": {
                "tokenType": "VIP",
                "tokenmsg": "21232f297a57a5a743894a0e4a801fc3"
            }
        }
    }
}

在其他接口的测试中，需要封装一个方法对登录接口返回内容的 memberid 和 token信息进行处理，这些内容，同样放到helper.py文件中

helper.py

def login():
    url = ReadYaml(dir_config.yaml_dir, 'ApiHost').get_data()['mockhost']+'/user/login'
    json = ReadYaml(dir_config.yaml_dir, 'user').get_data()
    headers = {"Content-Type":"application/json","apikey":"21232f297a57a5a743894a0e4a801fc3"}

    res = HTTPHandler().visit(url=url,method='post',headers=headers,json=json)
    return res


def save_token():
    data = login()

    tokenType = jsonpath(data, '$...tokenType')[0]
    tokenmsg = jsonpath(data, '$...tokenmsg')[0]
    memberid = jsonpath(data, '$...memberid')[0]
    token = tokenType + tokenmsg

如果用到memberid 和 token 信息的测试用例文件数量比较少，那么只需要测试用例文件的前置条件 Setup中调用save_token函数，即可获取memberid 和 token 信息；

但如果测试用例文件很多，那这些内容可以放到类变量里。新增一个Context类，Context中文意思叫上下文，在自动化测试中，对于一些临时数据处理、数据的准备和记录，经常用这种方式

上述测试用例py文件里， Setup中的db_info、host等内容，这些在其他测试用例文件中，也会经常用到，可以放到Context类使其变为类变量

与db_info、host等内容不同的是，memberid 和 token 信息，在Context类中定义memberid 和 token 函数，通过添加 @property装饰器方式将其变为实例属性，这样，就可以通过 Context().memberid的方式获取memberid 、Context().token的方式获取token

helper.py

import random
from jsonpath import jsonpath
from common import dir_config
from common.YamlHandler import ReadYaml
from common.RequestHandler import HTTPHandler

class Context:
    
    @property
    def token(self):
        '''
        token属性，且会动态变化
        通过Context().token可以获取token，自动调用这个方法
        '''
        data = login()
        tokenType = jsonpath(data, '$...tokenType')[0]
        tokenmsg = jsonpath(data, '$...tokenmsg')[0]
        t = tokenType + tokenmsg
        return t

    @property
    def memberid(self):
        data = login()
        m_id = jsonpath(data, '$...memberid')[0]
        return m_id
    
    
    host = ReadYaml(dir_config.yaml_dir,'ApiHost').get_data()['mockhost']
    db_info = ReadYaml(dir_config.yaml_dir,'db_info').get_data()
    user_info = ReadYaml(dir_config.yaml_dir, 'user').get_data()
    

        
def login():
    url = Context.host+'/user/login'
    json = ReadYaml(dir_config.yaml_dir, 'user').get_data()
    headers = {"Content-Type":"application/json","apikey":"21232f297a57a5a743894a0e4a801fc3"}

    res = HTTPHandler().visit(url=url,method='post',headers=headers,json=json)
    return res

# if __name__ == '__main__':
    # print(Context().memberid)
    # 输出结果 20012345644
    # print(Context().token)
    # 输出结果 VIP263A68169E5CCCEAE5A9739E28109AC1

注意： 这里login函数并不是 Context类里的函数

test_charge.py

import json
import time
import unittest
import os
from middleware import helper
from common import ddt
from common.RequestHandler import HTTPHandler
from common.excel_handler import ExcelHandler
from common import dir_config
from common.logger import Logger
from common.DBhandler import MysqlUtil
from common.YamlHandler import ReadYaml


@ddt.ddt
class Test_Register(unittest.TestCase):
    # 读取测试数据
    excel_handle = ExcelHandler(helper.Context.excelpath)
    test_data = excel_handle.read_key_value("charge")

    def setUp(self) -> None:
        self.req = HTTPHandler()
        self.db = MysqlUtil(helper.Context.db_info)
        self.host = helper.Context.host
        self.token = helper.Context().token
        self.memberid = helper.Context().memberid


    def tearDown(self) -> None:
        self.req.close_session()
        self.db.close()



    @ddt.data(*test_data)
    def test_charge(self,test_data):

        
        sql = '''SELECT money from memberinfo where memberid = %s;'''
        # 查询用户账户初始金额
        before_money = self.db.query(sql,args=[self.memberid])

        # 判断excel中是否 出现了 #memberid# ，如果出现，进行替换
        if '#memberid#' in str(test_data["json"]):
            test_data["json"] = test_data["json"].replace('#memberid#', str(self.memberid))

        # 读取excel中的headers数据（字典类型）
        headers = json.loads(test_data["headers"])
        # 添加token信息
        if headers['token'] == '#token#':
            headers['token'] = self.token

        res =  self.req.visit(
                              url= self.host+test_data["url"],
                              method=test_data["method"],
                              headers=headers,
                              json=json.loads(test_data["json"])
        )
        
		# 充值成功，断言 充值前金额+充值金额 是否等于 充值后金额
        if res['code'] == "0000":
            charge_money = json.loads(test_data["json"])["chargeinfo"]["chargeMoney"]
            # 查询用户账户充值后的金额
            after_money = self.db.query(sql, args=[self.memberid])
            
            self.assertEqual(before_money+charge_money,after_money)

正则

在excel文件中，会出现许多类似 $memberid$ 这种需要进行替换的数据，而在代码中，需要用 if 进行判断，如果if 这种判断增加，代码就会看起来很繁重，且执行效率会变慢

如果用上正则的方式进行替换，就会减少很多繁重的内容

re.search() 函数会在字符串中查找模式匹配，只要找到第一个匹配然后返回，若没有找到匹配则返回None

import re
data = '''{"memberid": "#memberid#","phonenum": "#phonenum#"}'''

pattern = r"#(.*?)#"
a = re.search(pattern,data).group(0)
b = re.search(pattern,data).group(1)

print(a)
# 输出结果 #memberid#
print(b)
# 输出结果memberid

re.sub()替换string中子串后返回替换后的新串

import re
data = '''{"memberid": "#memberid#","phonenum": "#phonenum#"}'''

pattern = r"#(.*?)#"
replace_data = re.sub(pattern,'这是替换内容',data,1)	# 替换一次
replace_data2 = re.sub(pattern,'这是替换内容2',data,2)	# 替换两次
print(replace_data) 
# 输出结果	{"memberid": "这是替换内容","phonenum": "#phonenum#"}
print(replace_data2)
# 输出结果	{"memberid": "这是替换内容2","phonenum": "这是替换内容2"}

helper.py

这一功能，同样可以放到helper.py里

import re

def replace_label(target):
    '''
    :param target: 需要匹配的字符串
    :return:返回替换成功的内容
    '''
    # 匹配的格式
    pattern = r"#(.*?)#"

    # 如果匹配的字符串中存在匹配项
    while re.search(pattern,target):
        # 获取字符串中需要替换的内容
        key = re.search(pattern,target).group(1)
        # 获取替换的数据
        value = str(getattr(Context(),key))
        # 将 target 中的值进行1次替换
        target = re.sub(pattern,value,target,1)
    return target

这个时候，Context环境管理方式中的Context类，就起到作用了

读取excel表格里的json数据后，若数据中包含需要替换的内容，如：memberid，将其赋予变量key；

通过 getattr()函数获取Context类的memberid属性的值，再通过 re.sub()函数进行替换

这块代码，可以直接使用replace_label函数进行简化

# 判断excel中是否 出现了 #memberid# ，如果出现，进行替换
        if '#memberid#' in str(test_data["json"]):
            test_data["json"] = test_data["json"].replace('#memberid#', str(self.memberid))
            
            

变为

test_data["json"] = helper.replace_label(test_data["json"])

添加token信息
        if headers['token'] == '#token#':
            headers['token'] = self.token

变为

headers = json.loads(helper.replace_label(test_data["headers"]))

日志处理

封装日志处理功能

logger.py

import logging

class Logger(logging.Logger):

    # 初始化 Logger
    def __init__(self,
                 name='root',
                 logger_level= 'DEBUG',
                 file=None,
                 logger_format = " [%(asctime)s]  %(levelname)s %(filename)s [ line:%(lineno)d ] %(message)s"
                 ):

        # 1、设置logger收集器，继承logging.Logger
        super().__init__(name)


        # 2、设置日志收集器level级别
        self.setLevel(logger_level)

        # 5、设置 handler 格式
        fmt = logging.Formatter(logger_format)

        # 3、设置日志处理器

        # 如果传递了文件，就会输出到file文件中
        if file:
            file_handler = logging.FileHandler(file)
            # 4、设置 file_handler 级别
            file_handler.setLevel(logger_level)
            # 6、设置handler格式
            file_handler.setFormatter(fmt)
            # 7、添加handler
            self.addHandler(file_handler)

        # 默认都输出到控制台
        stream_handler = logging.StreamHandler()
        # 4、设置 stream_handler 级别
        stream_handler.setLevel(logger_level)
        # 6、设置handler格式
        stream_handler.setFormatter(fmt)
        # 7、添加handler
        self.addHandler(stream_handler)

在执行用例文件中引入日志信息

run.py

import os
import time
import unittest
from common import dir_config
from common.HTMLTestRunnerNew import HTMLTestRunner
from common.logger import Logger



testloader = unittest.TestLoader()

# 全用例
suit_total = testloader.discover(dir_config.testcases_dir)
curTime = time.strftime("%Y-%m-%d %H_%M", time.localtime())

html_path = os.path.join(dir_config.htmlreport_dir,'{}_test.html'.format(curTime))
log_path = os.path.join(dir_config.logs_dir, '{}_test.log'.format(curTime))

logger = Logger(name="APItest",logger_level='DEBUG',file=log_path)

with open(html_path,"wb") as f:
    runner = HTMLTestRunner(f,title='测试报告',description='测试报告内容为：',tester='bobo')

    runner.run(suit_total)

在测试用例文件中引入日志记录功能

test_register.py

import json
import unittest
from middleware import helper
from common import ddt
from common.RequestHandler import HTTPHandler
from common.excel_handler import ExcelHandler
from run_case import logger
from common.DBhandler import MysqlUtil



@ddt.ddt
class Test_Register(unittest.TestCase):
    # 读取测试数据
    excel_handle = ExcelHandler(helper.Context.excelpath)
    test_data = excel_handle.read_key_value("charge")


    def setUp(self) -> None:
        self.req = HTTPHandler()
        self.db = MysqlUtil(helper.Context.db_info)
        self.host = helper.Context.host
        self.token = helper.Context().token
        self.memberid = helper.Context().memberid
        self.logger = logger


    def tearDown(self) -> None:
        self.req.close_session()
        self.db.close()


    @ddt.data(*test_data)
    def test_charge(self,test_data):

        # 查询用户账户初始金额
        sql = '''SELECT money from Member where memberid = %s;'''
        before_money = self.db.query(sql,args=[self.memberid])

        # 添加memberid信息 判断excel中是否 出现了 #memberid# ，如果出现，进行替换
        test_data["json"] = helper.replace_label(test_data["json"])

        # 添加token信息 判断excel中是否 出现了 #token# ，如果出现，进行替换
        headers = helper.replace_label(test_data["headers"])

        self.logger.info("用例名称：{};接口信息：url={}；method={}；headers={}；json={}".format(test_data["case_name"],
                                                                                   self.host+test_data["url"],
                                                                                   test_data["method"],
                                                                                   json.loads(headers),
                                                                                   json.loads(test_data["json"])
                                                                                   )
                         )

        res =  self.req.visit(
                              url= self.host+test_data["url"],
                              method=test_data["method"],
                              headers=json.loads(headers),
                              json=json.loads(test_data["json"])
        )
        self.logger.info("接口响应内容：{}".format(res))

        # 错误处理，抛出异常
        try:
            self.assertEqual(res["code"], test_data["excepted"])
            self.logger.info("接口响应code:{}，期望响应code:{}".format(res["code"], test_data["excepted"]))
            # 写入实际结果到excel表格
            self.excel_handle.write_change(helper.Context.excelpath,
                                           "register",
                                           test_data["caseid"] + 1, 9, "passed")

        except AssertionError as e:
            self.logger.error("测试用例执行失败：{}".format(e))
            self.excel_handle.write_change(helper.Context.excelpath,
                                           "register",
                                           test_data["caseid"] + 1, 9, "failed")
            raise e

        if res['code'] == "0000":
            charge_money = json.loads(test_data["json"])["chargeinfo"]["chargeMoney"]
            # 查询用户账户充值后的金额
            after_money = self.db.query(sql, args=[self.memberid])
            self.logger.info("用户账户初始金额:{}，充值金额:{}，用户账户充值后的金额:{}".format(before_money,charge_money,after_money))
            self.assertEqual(before_money+charge_money,after_money)
       

对于关键内容，加上日志功能，如接口的请求信息，接口的响应内容，对于失败用例，使用try...except进行错误处理，将错误内容抛出

log文件示例内容：

 [2022-05-06 00:05:01,157]  INFO test_charge.py [ line:50 ] 用例名称：充值成功;接口信息：url=http://127.0.0.1:4523/mock/894992/user/charge；method=post；headers={'Content-Type': 'application/json', 'apikey': '21232f297a57a5a743894a0e4a801fc3', 'token': 'VIP263A68169E5CCCEAE5A9739E28109AC1'}；json={'userinfo': {'memberid': '20012345644', 'type': 88, 'permission': True}, 'chargeinfo': {'chargeType': 'WX', 'chargeMoney': 43}}
 [2022-05-06 00:05:01,160]  INFO test_charge.py [ line:60 ] 接口响应内容：{'code': '0000', 'msg': {'result': 'success', 'info': '充值成功'}, 'errorinfo': None, 'chargeinfo': {'money': 10550, 'memberlevel': 'SVIP', 'integraladd': 43}}
 [2022-05-06 00:05:01,161]  INFO test_charge.py [ line:65 ] 接口响应code:0000，期望响应code:0000
 [2022-05-06 00:05:01,199]  INFO test_charge.py [ line:82 ] 用户账户初始金额:10507，充值金额:43，用户账户充值后的金额:10550
 [2022-05-06 00:05:01,222]  INFO test_charge.py [ line:50 ] 用例名称：鉴权失败;接口信息：url=http://127.0.0.1:4523/mock/894992/user/charge；method=post；headers={'Content-Type': 'application/json', 'apikey': '21232f297a57a5a743894a0e4a801fc3', 'token': ''}；json={'userinfo': {'memberid': '20012345644', 'type': 88, 'permission': True}, 'chargeinfo': {'chargeType': 'WX', 'chargeMoney': 43}}
 [2022-05-06 00:05:01,226]  INFO test_charge.py [ line:60 ] 接口响应内容：{'code': '0001', 'msg': {'result': 'false', 'errorinfo': '权限校验失败'}}

创作打卡挑战赛

赢取流量/现金/CSDN周边激励大奖