Python自动化 接口自动化:一种基于 Excel 合并单元格的设想与实践

Flymoon · 2020年05月06日 · 125 次阅读

前言

在实际工作中,使用Excel存储自动化测试数据是很多公司的首选,一方面由于Excel比较直观,方便操作;另一方面Python读取Excel也比较方便,不需要写多少代码就可以读出所有的测试数据。但是,随着项目的业务日趋复杂,使用Excel中的一行作为一个测试用例,对于复杂的测试场景来说,涉及到几个甚至十几个接口的时候,分散的接口用例关联起来难度较大,比如注册接口在第二行,登录接口在第五行,选择商品接口在第二十行,这种散落在各个地方的接口找起来就特别费时间。因此,需要一种方式,将这些关联的接口合并一起,一个合并单元格作为一条用例,一条用例可能包含多个关联接口,这样整齐划一的方式极大的方便了后期的维护

扩展读取Excel类

对合并单元格的读取,要求对现有的读取Excel的类做一扩展。具体的思路是:通过self.sh.merged_cells拿到合并单元格的行和列的范围,将这些行和列的范围添加到集合set中,分别命名为m_rowsm_cols,然后遍历m_rows,拿到其中的每一个合并行范围,再对这个合并行范围做遍历,将合并行每一行对应的单元格的值组合成一个列表,通过列表生成式获取列表title,再将合并单元格的列表和列表title通过zip,dict转化成一个字典,添加到all_case_data 通过对最大行范围和合并行范围取差集,得到的就是非合并行,然后和前面类似,非合并行每一行的值都是列表,和列表title转化成一个字典,添加到all_case_data 最后,你看到得到的all_case_data是一个顺序混乱的列表字典的形式,将sorted中的key指定为列表中的字典的其中一个key:case_data["case_id"][0]来升序排列(为了更好的排序,建议Excel中的case_id列采用001,002,003...这样的方式)

from openpyxl import load_workbook
from common.replace_variable import *
import json, re


class DoExcel:

    def __init__(self, file_path):
        self.file_path = file_path
        self.wb = load_workbook(self.file_path)
        self.sh = self.wb["case_datas"]


    #读取所有测试数据
    def read_all_caseData(self):
        #得到最大行、最大列
        max_row = self.sh.max_row
        max_column = self.sh.max_column
        #得到合并单元格的范围
        m_rows = []
        m_cols = []
        for m_area in self.sh.merged_cells:
            r1, r2, c1, c2 = m_area.min_row, m_area.max_row, m_area.min_col, m_area.max_col
            m_rows.append((r1, r2))
            m_cols.extend([c1, c2])
        m_rows = set(m_rows)
        m_cols = set(m_cols)

        #获取所有的测试数据,将其以字典的形式存储在列表中
        all_case_data = []
        title = [self.sh.cell(1, column).value for column in range(1, max_column + 1)]
        #获取合并行数据
        for m_row in m_rows:
            case_data = []
            for column in range(1, max_column + 1):
                for row in range(m_row[0], m_row[1] + 1):
                    if row == m_row[0]:
                        value = [[self.sh.cell(row, column).value]]
                    else:
                        for item in value:
                            item.append(self.sh.cell(row, column).value)
                case_data.extend(value)
            all_case_data.append(dict(zip(title, case_data)))

        #最大行范围和合并范围求差集,就是非合并行
        max_row_range = set(range(2, max_row + 1))
        for m_row in m_rows:
            m_row_range = set(range(m_row[0], m_row[1] + 1))
            max_row_range = max_row_range.difference(m_row_range)

        #获取非合并行数据
        for row in max_row_range:
            case_data = []
            for column in range(1, max_column + 1):
                value = [self.sh.cell(row, column).value]
                case_data.append(value)
            all_case_data.append(dict(zip(title, case_data)))

        #排序:根据case_id的数字从大到小
        all_case_data = sorted(all_case_data, key=lambda case_data: case_data["case_id"][0])

        #对检查的字符串做字典的转化,最终以字典的形式存储在列表中
        for case_data in all_case_data:
            self._check_convert_dict(case_data)
        print(all_case_data)
        return all_case_data


    #将检查项转化为字典
    def _check_convert_dict(self, case_data):
        for i in range(len(case_data["case_id"])):
            check_list = case_data["check"]
            new_check_list = []
            for check in check_list:
                new_check = {}
                items = re.findall("(\$.*?):.*?\"(.*?)\"", check)
                for item in items:
                    key, value = item
                    new_check[key] = value
                new_check_list.append(new_check)
        case_data["check"] = new_check_list




if __name__ == "__main__":
    all_case_data = DoExcel(r"E:\virtual_workshop\Python_API\test_datas\api_info.xlsx").read_all_caseData()

扩展测试类

拿到的all_case_data是列表字典的形式,其中的每个元素都是字典,每个字典的value都是列表,其中列表中的每一项元素都是合并单元格的每个单元格对应的值,就是这种形式 这就要求测试类test_my_request,在ddt的基础上,增加for循环。前面说过for循环遍历的N条用例都会默认是一条用例,正好将每一个合并单元格当做了一条用例对待

import unittest
import ddt
import json
from jsonpath import jsonpath
from common.do_excel import DoExcel
from common.replace_variable import ReplaceVariable
from common.context import Context
from common.my_request import *
from common.my_logging import *
from common.read_cfg import *
import logging


@ddt.ddt
class TestMyRequest(unittest.TestCase):
    do_excel = DoExcel(r"{0}\api_info.xlsx".format(testdatas_dir))
    all_case_data = do_excel.read_all_caseData()
    #读取配置文件
    test_host, sql_host = get_host_section()

    @ddt.data(*all_case_data)
    def test_my_request(self, case_data):


        logging.info("用例『{0}』测试开始了".format(case_data["case_desc"][0]))
        for i in range(len(case_data["case_id"])):

            #拼接url
            if case_data["host"][i] == "test":
                case_data["url"][i] = self.test_host + case_data["url"][i]
            elif case_data["host"][i] == "sql":
                case_data["url"][i] = self.sql_host + case_data["url"][i]

            #替换请求参数
            request_data = json.loads(ReplaceVariable.replace_variable(case_data["request_data"][i]))
            logging.info("请求数据: {0}".format(request_data))

            #请求数据反射
            request_extract = case_data["request_extract"][i]
            if request_extract:
                key, express = request_extract.split("=")
                setattr(Context, key, jsonpath(request_data, express)[0])


            if case_data["method"][i].lower() == "get":
                result = send_request(case_data["method"][i], case_data["url"][i], params=request_data)
            elif case_data["method"][i].lower() == "post":
                result = send_request(case_data["method"][i], case_data["url"][i], data=request_data)
            logging.info("响应结果: {0}".format(result))

            #响应数据反射
            response_extract = case_data["response_extract"][i]
            if response_extract:
                key, express = response_extract.split("=")
                setattr(Context, key, str(jsonpath(result, express)[0]))


            #断言
            for key, value in case_data["check"][i].items():
                #替换期望结果
                actual = str(jsonpath(result, key)[0])
                expect = ReplaceVariable.replace_variable(value)
                logging.info("实际结果: {0}".format(actual))
                logging.info("期望结果: {0}".format(expect))
                self.assertEqual(actual, expect)

        logging.info("\n\n")

数据库断言

常规的做法是,在Excel中写一条用例,通过PyMSQL读取,拿到数据后,再去响应结果里查找。这种方法很繁琐,因为你得解析响应结果。之前的接口框架也对请求数据和响应结果做了反射,为什么不利用这些数据呢?一个设想是,我把数据库的查询作为一个接口来测,使用Flask封装成接口,再通过Excel传入对应的sql语句,这个sql语句里需要的条件,我可能从上个接口里提取,提取到作为反射数据来用。举个例子,常规做法是我投资了200元,通过select语句查询这个人的投资记录,拿到投资记录的结果筛选出金额,再去响应结果里找,现在的做法是,我传入select * from t_user where uid=xxx and invest_money=200,将这个作为请求参数,传给sql对应的查询接口,只要查询有结果就行,不需要再做额外的解析和比对

from flask import Flask, request
from common.read_mysql import ReadMysql
import json

app = Flask(__name__)



@app.route("/query/one", methods=["post"])
def query_one():
    read_mysql = ReadMysql()
    query_sql = request.form.get("query_sql")
    one_data = read_mysql.select_one_data(query_sql)
    if one_data:
        return json.dumps({"code": "200", "result": str(one_data)})
    else:
        return json.dumps({"code": "404"})


app.run()
暂无回复。
需要 登录 后方可回复, 如果你还没有账号请点击这里 注册