找回密码
 立即注册
首页 业界区 业界 Pytest 测试用例自动生成:接口自动化进阶实践 ...

Pytest 测试用例自动生成:接口自动化进阶实践

韩素欣 4 小时前
引言:为什么我们要抛弃 “手写用例”?
在接口自动化实践中,当项目规模扩大、用例数量激增时,传统“手写Pytest用例”的模式往往会陷入瓶颈。
做接口自动化的同学,大概率都踩过这样的硬编码坑:写一条 “新增 - 查询 - 删除” 的流程用例,要重复写 3 个接口的请求、参数与断言代码;不同同事写的用例,有的把数据塞代码里,有的存 Excel,交接时看得头大;新手没代码基础,想加个用例还要先学 Python 语法。
一、遇到的3个核心痛点

我们公司在维护Pytest接口自动化项目时,深刻感受到手写用例带来的诸多困扰,随着项目规模扩大,问题愈发凸显:

  • 用例编写效率低,重复劳动多。一条流程用例要调用多个接口,每个接口的请求头、参数、断言都要手写,浪费时间。
  • 代码混乱无规范,维护成本高。测试同学各自为战,测试数据存储方式不一样(硬编码、data.py、Excel等);并且重复编写“发送请求”“数据库查询”等通用功能,导致项目冗余代码堆积,新人接手时难以梳理逻辑。
  • 门槛高,新手难上手。无Python基础的测试同学,需先学习requests库、Pytest语法、断言写法等技术内容,再结合混乱的项目结构,入门难度大,难以快速参与用例编写。
二、核心解决方案:数据与逻辑分离,自动生成测试用例

针对上述痛点,我们提出核心解决方案:测试人员仅负责“设计测试数据”(基于YAML),用例生成器自动完成“用例代码编写”,通过“数据与逻辑分离”的思路,从根源解决手写用例的弊端。
1. 核心设计思路


  • 把 “测试数据” 和 “用例逻辑” 彻底分开,使数据与逻辑解耦。将接口参数、断言规则、前置后置操作等测试数据,按约定格式存入YAML文件,测试人员无需关注代码逻辑,专注业务数据设计。
  • 自动生成 Pytest 测试用例文件。定义一个用例生成器模块,,读取YAML文件中的测试数据,自动校验格式并生成标准化的Pytest用例代码,完全替代手写用例。
2. 方案核心优势


  • 零代码门槛:测试人员无需编写Python代码,只需按模板填写YAML,降低技术要求。
  • 输出标准化:生成的用例命名、目录结构、日志格式、断言方式完全统一,告别代码混乱。
  • 批量高效生成:支持整个目录的 YAML 文件批量生成,一次生成上百条用例;
  • 零维护成本:接口变更时,只改 YAML 数据,生成器重新运行即可更新用例。
3. 完整实施流程

完整流程为:编写YAML测试数据 → 运行生成器自动生成测试用例 → 执行自动生成的Pytest用例
三、关键步骤:从 YAML 设计到自动生成用例

下面通过“实操步骤+代码示例”的方式,详细说明方案的落地过程,以“新增设备→查询设备→解绑设备”的完整流程用例为例。
第一步:设计标准化YAML测试数据格式

YAML文件是方案的核心,需兼顾“完整性”与“易用性”,既要覆盖接口测试的全场景需求,又要让测试人员容易理解和填写。
我们设计的YAML格式支持:基础信息配置、前置/后置操作、多接口步骤串联、多样化断言(常规断言+数据库断言)。
YAML示例如下(test_device_bind.yaml):
  1. # test_device_bind.yaml
  2. testcase:
  3.   name: bind_device  # 用例唯一标识,建议和文件名一致(去掉test_)
  4.   description: 新增设备→查询设备→解绑设备  # 用例说明,清晰易懂
  5.   allure:  # Allure报告配置,方便统计
  6.     epic: 商家端
  7.     feature: 设备管理
  8.     story: 新增设备
  9.    
  10.   setups:  # 前置操作:执行测试前的准备(如数据库查询、数据初始化)
  11.     - id: check_database
  12.       description: 检查设备是否已存在
  13.       operation_type: db  # 操作类型:db=数据库操作
  14.       query: SELECT id FROM device WHERE imei = '865403062000000'
  15.       expected: id  # 预期查询结果存在id字段
  16.   steps:  # 核心测试步骤:每个步骤对应一个接口请求
  17.     - id: device_bind  # 步骤唯一标识,用于跨步骤取值
  18.       description: 新增设备
  19.       project: merchant  # 所属项目(用于获取对应的host、token)
  20.       path: '/device/bind'  # 接口路径
  21.       method: POST  # 请求方法
  22.       headers:
  23.         Content-Type: 'application/json'
  24.         Authorization: '{{merchant.token}}'  # 从全局变量取merchant的token
  25.       data:  # 请求参数
  26.         code: deb45899-957-10972b35515
  27.         name: test_device_name
  28.         imei: '865403062000000'
  29.       assert:  # 断言配置,支持多种断言类型
  30.         - type: equal  # 等于断言
  31.           field: code  # 响应字段:code
  32.           expected: 0  # 预期值
  33.         - type: is not None  # 非空断言
  34.           field: data.id  # 响应字段:data.id
  35.         - type: equal
  36.           field: message
  37.           expected: success
  38.     - id: device_list  # 第二个步骤:查询新增的设备
  39.       description: 查询设备列表
  40.       project: merchant
  41.       path: '/device/list'
  42.       method: GET
  43.       headers:
  44.         Content-Type: 'application/json'
  45.         Authorization: '{{merchant.token}}'
  46.       data:
  47.         goodsId: '{{steps.device_bind.data.id}}'  # 跨步骤取值:从device_bind步骤的响应中取id
  48.       assert:
  49.         - type: equal
  50.           field: status_code  # 断言HTTP状态码
  51.           expected: 200
  52.         - type: equal
  53.           field: data.code
  54.           expected: '{{steps.device_bind.data.code}}'  # 跨步骤取参数
  55.         - type: mysql_query  # 数据库断言:查询设备是否存在
  56.           query: SELECT id FROM users WHERE name='test_device_name'
  57.           expected: id
  58.   teardowns:  # 后置操作:测试完成后清理数据(如解绑设备、删除数据库记录)
  59.     - id: device_unbind
  60.       description: 解绑设备
  61.       operation_type: api  # 操作类型:api=接口请求
  62.       project: plateform
  63.       path: '/device/unbind'
  64.       method: POST
  65.       headers:
  66.         Content-Type: 'application/json'
  67.         Authorization: '{{merchant.token}}'
  68.       data:
  69.         deviceId: '{{steps.device_bind.data.id}}'  # 跨步骤取新增设备的id
  70.       assert:
  71.         - type: equal
  72.           field: code
  73.           expected: 0
  74.     - id: clear_database
  75.       description: 清理数据库
  76.       operation_type: db  # 数据库操作
  77.       query: DELETE FROM device WHERE id = '{{steps.device_bind.data.id}}'
复制代码
第二步:编写用例生成器(自动生成的 “核心引擎”)

用例生成器的作用是:读取 YAML 文件→校验数据格式→生成标准的 Pytest 用例代码,支持单个文件或目录批量处理。
以下是生成器核心代码(case_generator.py),关键逻辑已添加详细注释:
  1. # case_generator.py
  2. # @author:  xiaoqq
  3. import os
  4. import yaml
  5. from utils.log_manager import log
  6. class CaseGenerator:
  7.         """
  8.         测试用例文件生成器
  9.         """
  10.         def generate_test_cases(self, project_yaml_list=None, output_dir=None):
  11.                 """
  12.                 根据YAML文件生成测试用例并保存到指定目录
  13.                 :param project_yaml_list: 列表形式,项目名称或YAML文件路径
  14.                 :param output_dir: 测试用例文件生成目录
  15.                 """
  16.                 # 如果没有传入project_yaml_list,默认遍历tests目录下所有project
  17.                 if not project_yaml_list:
  18.                         project_yaml_list = ["tests/"]
  19.                
  20.                 # 遍历传入的project_yaml_list
  21.                 for item in project_yaml_list:
  22.                         if os.path.isdir(item):  # 如果是项目目录,如tests/merchant
  23.                                 self._process_project_dir(item, output_dir)
  24.                         elif os.path.isfile(item) and item.endswith('.yaml'):  # 如果是单个YAML文件
  25.                                 self._process_single_yaml(item, output_dir)
  26.                         else:  # 如果是项目名称,如merchant
  27.                                 project_dir = os.path.join("tests", item)
  28.                                 self._process_project_dir(project_dir, output_dir)
  29.                
  30.                 log.info("测试用例生成完毕!")
  31.        
  32.         def _process_project_dir(self, project_dir, output_dir):
  33.                 """
  34.                 处理项目目录,遍历项目下所有YAML文件生成测试用例
  35.                 :param project_dir: 项目目录路径
  36.                 :param output_dir: 测试用例文件生成目录
  37.                 """
  38.                 for root, dirs, files in os.walk(project_dir):
  39.                         for file in files:
  40.                                 if file.endswith('.yaml'):
  41.                                         yaml_file = os.path.join(root, file)
  42.                                         self._process_single_yaml(yaml_file, output_dir)
  43.        
  44.         def _process_single_yaml(self, yaml_file, output_dir):
  45.                 """
  46.                 处理单个YAML文件,生成对应的测试用例文件
  47.                 :param yaml_file: YAML文件路径
  48.                 :param output_dir: 测试用例文件生成目录
  49.                 """
  50.                 # 读取YAML文件内容
  51.                 _test_data = self.load_test_data(yaml_file)
  52.                 validate_test_data = self.validate_test_data(_test_data)
  53.                 if not validate_test_data:
  54.                         log.warning(f"{yaml_file} 数据校验不通过,跳过生成测试用例。")
  55.                         return
  56.                 test_data = _test_data['testcase']
  57.                 teardowns = test_data.get('teardowns')
  58.                 validate_teardowns = self.validate_teardowns(teardowns)
  59.                
  60.                 # 生成测试用例文件的相对路径。yaml文件路径有多个层级时,获取项目名称,以及tests/后、yaml文件名前的路径
  61.                 relative_path = os.path.relpath(yaml_file, 'tests')
  62.                 path_components = relative_path.split(os.sep)
  63.                 project_name = path_components[0] if path_components[0] else path_components[1]
  64.                 # 移除最后一个组件(文件名)
  65.                 if path_components:
  66.                         path_components.pop()  # 移除最后一个元素
  67.                 directory_path = os.path.join(*path_components)        # 重新组合路径
  68.                 directory_path = directory_path.rstrip(os.sep)        # 确保路径不以斜杠结尾
  69.                
  70.                 module_name = test_data['name']
  71.                 description = test_data.get('description')
  72.                 # 日志记录中的测试用例名称
  73.                 case_name = f"test_{module_name} ({description})" if description is not None else f"test_{module_name}"
  74.                
  75.                 # 判断test_data中的name是否存在"_",存在则去掉将首字母大写组成一个新的字符串,否则首字母大写
  76.                 module_class_name = (''.join(s.capitalize() for s in module_name.split('_'))
  77.                                                          if '_' in module_name else module_name.capitalize())
  78.                 file_name = f'test_{module_name}.py'
  79.                
  80.                 # 生成文件路径
  81.                 if output_dir:
  82.                         file_path = os.path.join(output_dir, directory_path, file_name)
  83.                 else:
  84.                         file_path = os.path.join('test_cases', directory_path, file_name)
  85.                
  86.                 # 检查test_cases中对应的.py文件是否存在,存在则跳过生成
  87.                 if os.path.exists(file_path):
  88.                         log.info(f"测试用例文件已存在,跳过生成: {file_path}")
  89.                         return
  90.                
  91.                 # 创建目录
  92.                 os.makedirs(os.path.dirname(file_path), exist_ok=True)
  93.                
  94.         # 解析Allure配置
  95.                 allure_epic = test_data.get("allure", {}).get("epic", project_name)
  96.                 allure_feature = test_data.get("allure", {}).get("feature")
  97.                 allure_story = test_data.get("allure", {}).get("story", module_name)
  98.                
  99.         # 生成并写入用例代码
  100.                 with open(file_path, 'w', encoding='utf-8') as f:
  101.             # 写入导入语句
  102.                         f.write(f"# Auto-generated test module for {module_name}\n")
  103.                         f.write(f"from utils.log_manager import log\n")
  104.                         f.write(f"from utils.globals import Globals\n")
  105.                         f.write(f"from utils.variable_resolver import VariableResolver\n")
  106.                         f.write(f"from utils.request_handler import RequestHandler\n")
  107.                         f.write(f"from utils.assert_handler import AssertHandler\n")
  108.                         if validate_teardowns:
  109.                                 f.write(f"from utils.teardown_handler import TeardownHandler\n")
  110.                                 f.write(f"from utils.project_login_handler import ProjectLoginHandler\n")
  111.                         f.write(f"import allure\n")
  112.                         f.write(f"import yaml\n\n")
  113.                        
  114.             # 写入类装饰器(Allure配置)
  115.                         f.write(f"@allure.epic('{allure_epic}')\n")
  116.                         if allure_feature:
  117.                                 f.write(f"@allure.feature('{allure_feature}')\n")
  118.                         f.write(f"class Test{module_class_name}:\n")
  119.                        
  120.             # 写入setup_class(类级前置操作)
  121.                         f.write(f"    @classmethod\n")
  122.                         f.write(f"    def setup_class(cls):\n")
  123.                         f.write(f"        log.info('========== 开始执行测试用例:{case_name} ==========')\n")
  124.                         f.write(f"        cls.test_case_data = cls.load_test_case_data()\n")        # 获取测试数据
  125.                         # 如果存在teardowns,则将步骤列表转换为字典, 在下面的测试方法中通过 id 查找步骤的信息
  126.                         if validate_teardowns:
  127.                                 f.write(f"        cls.login_handler = ProjectLoginHandler()\n")
  128.                                 f.write(f"        cls.teardowns_dict = {{teardown['id']: teardown for teardown in cls.test_case_data['teardowns']}}\n")
  129.                                 f.write(f"        for teardown in cls.test_case_data.get('teardowns', []):\n")
  130.                                 f.write(f"            project = teardown.get('project')\n")
  131.                                 f.write(f"            if project:\n")
  132.                                 f.write(f"                cls.login_handler.check_and_login_project(project, Globals.get('env'))\n")
  133.                         # 将步骤列表转换为字典, 在下面的测试方法中通过 id 查找步骤的信息
  134.                         f.write(f"        cls.steps_dict = {{step['id']: step for step in cls.test_case_data['steps']}}\n")
  135.                         f.write(f"        cls.session_vars = {{}}\n")
  136.                         f.write(f"        cls.global_vars = Globals.get_data()\n")  # 获取全局变量
  137.                         # 创建VariableResolver实例并保存在类变量中
  138.                         f.write(f"        cls.VR = VariableResolver(global_vars=cls.global_vars, session_vars=cls.session_vars)\n")
  139.                         f.write(f"        log.info('Setup completed for Test{module_class_name}')\n\n")
  140.                        
  141.             # 写入加载测试数据的静态方法
  142.                         f.write(f"    @staticmethod\n")
  143.                         f.write(f"    def load_test_case_data():\n")
  144.                         f.write(f"        with open(r'{yaml_file}', 'r', encoding='utf-8') as file:\n")
  145.                         f.write(f"            test_case_data = yaml.safe_load(file)['testcase']\n")
  146.                         f.write(f"        return test_case_data\n\n")
  147.                        
  148.             # 写入核心测试方法
  149.                         f.write(f"    @allure.story('{allure_story}')\n")
  150.                         f.write(f"    def test_{module_name}(self):\n")
  151.                         f.write(f"        log.info('Starting test_{module_name}')\n")
  152.                         # 遍历步骤,生成接口请求和断言代码
  153.                         for step in test_data['steps']:
  154.                                 step_id = step['id']
  155.                                 step_project = step.get("project") # 场景测试用例可能会请求不同项目的接口,需要在每个step中指定对应的project
  156.                                 f.write(f"        # Step: {step_id}\n")
  157.                                 f.write(f"        log.info(f'开始执行 step: {step_id}')\n")
  158.                                 f.write(f"        {step_id} = self.steps_dict.get('{step_id}')\n")
  159.                                 if step_project:
  160.                                         f.write(f"        project_config = self.global_vars.get('{step_project}')\n")
  161.                                 else:
  162.                                         f.write(f"        project_config = self.global_vars.get('{project_name}')\n")
  163.                                 # 生成请求代码
  164.                                 f.write(f"        response = RequestHandler.send_request(\n")
  165.                                 f.write(f"            method={step_id}['method'],\n")
  166.                                 f.write(f"            url=project_config['host'] + self.VR.process_data({step_id}['path']),\n")
  167.                                 f.write(f"            headers=self.VR.process_data({step_id}.get('headers')),\n")
  168.                                 f.write(f"            data=self.VR.process_data({step_id}.get('data')),\n")
  169.                                 f.write(f"            params=self.VR.process_data({step_id}.get('params')),\n")
  170.                                 f.write(f"            files=self.VR.process_data({step_id}.get('files'))\n")
  171.                                 f.write(f"        )\n")
  172.                                 f.write(f"        log.info(f'{step_id} 响应:{{response}}')\n")
  173.                                 f.write(f"        self.session_vars['{step_id}'] = response\n")
  174.                                 # 生成断言代码
  175.                                 if 'assert' in step:
  176.                                         f.write(f"        db_config = project_config.get('mysql')\n")
  177.                                         f.write(f"        AssertHandler().handle_assertion(\n")
  178.                                         f.write(f"            asserts=self.VR.process_data({step_id}['assert']),\n")
  179.                                         f.write(f"            response=response,\n")
  180.                                         f.write(f"            db_config=db_config\n")
  181.                                         f.write(f"        )\n\n")
  182.                        
  183.                         # 写入teardown_class(类级后置操作)
  184.                         if validate_teardowns:
  185.                                 f.write(f"    @classmethod\n")
  186.                                 f.write(f"    def teardown_class(cls):\n")
  187.                                 f.write(f"        log.info('Starting teardown for the Test{module_class_name}')\n")
  188.                                 for teardown_step in teardowns:
  189.                                         teardown_step_id = teardown_step['id']
  190.                                         teardown_step_project = teardown_step.get("project")
  191.                                         f.write(f"        {teardown_step_id} = cls.teardowns_dict.get('{teardown_step_id}')\n")
  192.                                         if teardown_step_project:
  193.                                                 f.write(f"        project_config = cls.global_vars.get('{teardown_step_project}')\n")
  194.                                         else:
  195.                                                 f.write(f"        project_config = cls.global_vars.get('{project_name}')\n")
  196.                                         # 处理API类型的后置操作
  197.                                         if teardown_step['operation_type'] == 'api':
  198.                                                 f.write(f"        response = RequestHandler.send_request(\n")
  199.                                                 f.write(f"            method={teardown_step_id}['method'],\n")
  200.                                                 f.write(f"            url=project_config['host'] + cls.VR.process_data({teardown_step_id}['path']),\n")
  201.                                                 f.write(f"            headers=cls.VR.process_data({teardown_step_id}.get('headers')),\n")
  202.                                                 f.write(f"            data=cls.VR.process_data({teardown_step_id}.get('data')),\n")
  203.                                                 f.write(f"            params=cls.VR.process_data({teardown_step_id}.get('params')),\n")
  204.                                                 f.write(f"            files=cls.VR.process_data({teardown_step_id}.get('files'))\n")
  205.                                                 f.write(f"        )\n")
  206.                                                 f.write(f"        log.info(f'{teardown_step_id} 响应:{{response}}')\n")
  207.                                                 f.write(f"        cls.session_vars['{teardown_step_id}'] = response\n")
  208.                                                 if 'assert' in teardown_step:
  209.                                                         # if any(assertion['type'].startswith('mysql') for assertion in teardown_step['assert']):
  210.                                                         #         f.write(f"        db_config = project_config.get('mysql')\n")
  211.                                                         f.write(f"        db_config = project_config.get('mysql')\n")
  212.                                                         f.write(f"        AssertHandler().handle_assertion(\n")
  213.                                                         f.write(f"            asserts=cls.VR.process_data({teardown_step_id}['assert']),\n")
  214.                                                         f.write(f"            response=response,\n")
  215.                                                         f.write(f"            db_config=db_config\n")
  216.                                                         f.write(f"        )\n\n")
  217.                                         # 处理数据库类型的后置操作
  218.                                         elif teardown_step['operation_type'] == 'db':
  219.                                                 f.write(f"        db_config = project_config.get('mysql')\n")
  220.                                                 f.write(f"        TeardownHandler().handle_teardown(\n")
  221.                                                 f.write(f"            asserts=cls.VR.process_data({teardown_step_id}),\n")
  222.                                                 f.write(f"            db_config=db_config\n")
  223.                                                 f.write(f"        )\n\n")
  224.                                                 f.write(f"        pass\n")
  225.                                         else:
  226.                                                 log.info(f"未知的 operation_type: {teardown_step['operation_type']}")
  227.                                                 f.write(f"        pass\n")
  228.                                 f.write(f"        log.info('Teardown completed for Test{module_class_name}.')\n")
  229.                         f.write(f"\n        log.info(f"Test case test_{module_name} completed.")\n")
  230.                
  231.                 log.info(f"已生成测试用例文件: {file_path}")
  232.                
  233.         @staticmethod
  234.         def load_test_data(test_data_file):
  235.         """读取YAML文件,处理读取异常"""
  236.                 try:
  237.                         with open(test_data_file, 'r', encoding='utf-8') as file:
  238.                                 test_data = yaml.safe_load(file)
  239.                         return test_data
  240.                 except FileNotFoundError:
  241.                         log.error(f"未找到测试数据文件: {test_data_file}")
  242.                 except yaml.YAMLError as e:
  243.                         log.error(f"YAML配置文件解析错误: {e},{test_data_file} 跳过生成测试用例。")
  244.        
  245.         @staticmethod
  246.         def validate_test_data(test_data):
  247.                  """校验测试数据格式是否符合要求"""
  248.                 if not test_data:
  249.                         log.error("test_data 不能为空.")
  250.                         return False
  251.                 if not test_data.get('testcase'):
  252.                         log.error("test_data 必须包含 'testcase' 键.")
  253.                         return False
  254.                 if not test_data['testcase'].get('name'):
  255.                         log.error("'testcase' 下的 'name' 字段不能为空.")
  256.                         return False
  257.                 steps = test_data['testcase'].get('steps')
  258.                 if not steps:
  259.                         log.error("'testcase' 下的 'steps' 字段不能为空.")
  260.                         return False
  261.                
  262.                 for step in steps:
  263.                         if not all(key in step for key in ['id', 'path', 'method']):
  264.                                 log.error("每个步骤必须包含 'id', 'path', 和 'method' 字段.")
  265.                                 return False
  266.                         if not step['id']:
  267.                                 log.error("步骤中的 'id' 字段不能为空.")
  268.                                 return False
  269.                         if not step['path']:
  270.                                 log.error("步骤中的 'path' 字段不能为空.")
  271.                                 return False
  272.                         if not step['method']:
  273.                                 log.error("步骤中的 'method' 字段不能为空.")
  274.                                 return False
  275.                 return True
  276.        
  277.         @staticmethod
  278.         def validate_teardowns(teardowns):
  279.                 """
  280.                 验证 teardowns 数据是否符合要求
  281.                 :param teardowns: teardowns 列表
  282.                 :return: True 如果验证成功,否则 False
  283.                 """
  284.                 if not teardowns:
  285.                         # log.warning("testcase 下的 'teardowns' 字段为空.")
  286.                         return False
  287.                
  288.                 for teardown in teardowns:
  289.                         if not all(key in teardown for key in ['id', 'operation_type']):
  290.                                 log.warning("teardown 必须包含 'id' 和 'operation_type' 字段.")
  291.                                 return False
  292.                         if not teardown['id']:
  293.                                 log.warning("teardown 中的 'id' 字段为空.")
  294.                                 return False
  295.                         if not teardown['operation_type']:
  296.                                 log.warning("teardown 中的 'operation_type' 字段为空.")
  297.                                 return False
  298.                        
  299.                         if teardown['operation_type'] == 'api':
  300.                                 required_api_keys = ['path', 'method', 'headers', 'data']
  301.                                 if not all(key in teardown for key in required_api_keys):
  302.                                         log.warning("对于 API 类型的 teardown,必须包含 'path', 'method', 'headers', 'data' 字段.")
  303.                                         return False
  304.                                 if not teardown['path']:
  305.                                         log.warning("teardown 中的 'path' 字段为空.")
  306.                                         return False
  307.                                 if not teardown['method']:
  308.                                         log.warning("teardown 中的 'method' 字段为空.")
  309.                                         return False
  310.                        
  311.                         elif teardown['operation_type'] == 'db':
  312.                                 if 'query' not in teardown or not teardown['query']:
  313.                                         log.warning("对于数据库类型的 teardown,'query' 字段不能为空.")
  314.                                         return False
  315.                 return True
  316. if __name__ == '__main__':
  317.     # 运行生成器,生成指定YAML文件的用例
  318.         CG = CaseGenerator()
  319.         CG.generate_test_cases(project_yaml_list=["tests/merchant/test_device_bind.yaml"])
复制代码
第三步:运行生成器,自动生成Pytest用例

运行上述生成器代码后,会自动在指定目录(默认test_cases)生成标准化的Pytest用例文件(如test_device_bind.py),无需手动修改,可通过项目入口文件执行(入口文件详细代码可参考文末开源项目)。
生成的用例代码示例(关键部分):
  1. # Auto-generated test module for device_bind
  2. from utils.log_manager import log
  3. from utils.globals import Globals
  4. from utils.variable_resolver import VariableResolver
  5. from utils.request_handler import RequestHandler
  6. from utils.assert_handler import AssertHandler
  7. from utils.teardown_handler import TeardownHandler
  8. import allure
  9. import yaml
  10. @allure.epic('商家端')
  11. @allure.feature('设备管理')
  12. class TestDeviceBind:
  13.     @classmethod
  14.     def setup_class(cls):
  15.         log.info('========== 开始执行测试用例:test_device_bind (新增设备) ==========')
  16.         cls.test_case_data = cls.load_test_case_data()
  17.         cls.steps_dict = {step['id']: step for step in cls.test_case_data['steps']}
  18.         cls.session_vars = {}
  19.         cls.global_vars = Globals.get_data()
  20.         cls.VR = VariableResolver(global_vars=cls.global_vars, session_vars=cls.session_vars)
  21.         log.info('Setup 完成')
  22.     @staticmethod
  23.     def load_test_case_data():
  24.         with open(r'tests/merchant\device_management\test_device_bind.yaml', 'r', encoding='utf-8') as file:
  25.             test_case_data = yaml.safe_load(file)['testcase']
  26.         return test_case_data
  27.     @allure.story('新增设备')
  28.     def test_device_bind(self):
  29.         log.info('开始执行 test_device_bind')
  30.         
  31.         # Step: device_bind
  32.         log.info(f'开始执行 step: device_bind')
  33.         device_bind = self.steps_dict.get('device_bind')
  34.         project_config = self.global_vars.get('merchant')
  35.         response = RequestHandler.send_request(
  36.             method=spu_deviceType['method'],
  37.             url=project_config['host'] + self.VR.process_data(device_bind['path']),
  38.             headers=self.VR.process_data(device_bind.get('headers')),
  39.             data=self.VR.process_data(device_bind.get('data')),
  40.             params=self.VR.process_data(device_bind.get('params')),
  41.             files=self.VR.process_data(device_bind.get('files'))
  42.         )
  43.         log.info(f'device_bind 请求结果为:{response}')
  44.         self.session_vars['device_bind'] = response
  45.         db_config = project_config.get('mysql')
  46.         AssertHandler().handle_assertion(
  47.             asserts=self.VR.process_data(device_bind['assert']),
  48.             response=response,
  49.             db_config=db_config
  50.         )
  51.         # Step: device_list
  52.         log.info(f'开始执行 step: device_list')
  53.         device_list = self.steps_dict.get('device_list')
  54.         project_config = self.global_vars.get('merchant')
  55.         response = RequestHandler.send_request(
  56.             method=device_list['method'],
  57.             url=project_config['host'] + self.VR.process_data(device_list['path']),
  58.             headers=self.VR.process_data(device_list.get('headers')),
  59.             data=self.VR.process_data(device_list.get('data')),
  60.             params=self.VR.process_data(device_list.get('params')),
  61.             files=self.VR.process_data(device_list.get('files'))
  62.         )
  63.         log.info(f'device_list 请求结果为:{response}')
  64.         self.session_vars['device_list'] = response
  65.         db_config = project_config.get('mysql')
  66.         AssertHandler().handle_assertion(
  67.             asserts=self.VR.process_data(device_list['assert']),
  68.             response=response,
  69.             db_config=db_config
  70.         )
  71.              
  72.         log.info(f"Test case test_device_bind completed.")
  73.         @classmethod
  74.     def teardown_class(cls):
  75.             # 示例代码省略
  76.             ......
  77.         log.info(f'Teardown completed for TestDeviceBind.')
复制代码
四、其他核心工具类

生成的用例文件依赖多个自定义工具类,这些工具类封装了通用功能,确保用例可正常运行。以下是各工具类的核心作用(详细实现可参考文末开源项目):
工具类作用log_manager统一日志记录,输出用例执行过程Globals存储全局配置,如各项目的host、token、数据库连接信息、环境变量等。VariableResolver解析 YAML 中的变量(如{{steps.device_bind.data.id}}),支持全局变量、跨步骤变量取值。RequestHandler统一发送 HTTP 请求,处理超时、重试AssertHandler解析YAML中的断言配置,支持常规断言(等于、非空、包含等)和数据库断言。TeardownHandler处理后置操作,支持接口请求型和数据库操作型的后置清理逻辑。五、方案落地价值:重构后我们获得了什么?


  • 效率翻倍:用例编写时间减少 70%+。以前写一条 3 步流程用例要 15 分钟,现在写 YAML 只需要 5 分钟,生成用例秒级完成,还不用关心代码格式。
  • 维护成本大幅降低:接口变更时,仅需修改对应YAML文件的相关字段(如参数、断言),重新运行生成器即可更新用例,无需全局搜索和修改代码,避免引入新bug。
  • 入门门槛极低:无Python基础的测试人员,只需学习简单的YAML格式规则,按模板填写数据即可参与用例编写,团队协作效率大幅提升。
  • 项目规范统一:所有用例的命名、目录结构、日志格式、断言方式均由生成器统一控制,彻底告别“各自为战”的混乱局面,项目可维护性显著增强。
六、后续优化方向

目前方案已满足核心业务需求,但仍有优化空间,后续将重点推进以下方向:

  • 支持用例间依赖:实现用例级别的数据传递,比如用例A的输出作为用例B的输入,满足更复杂的业务场景。
  • 增强YAML灵活性:支持在YAML中调用自定义Python函数(如生成随机数、加密参数),提升数据设计的灵活性。
  • 简化YAML编写:增加通用配置默认值(如默认请求头、默认项目配置),减少重复填写工作。
  • 多数据源支持:新增Excel/CSV导入功能,满足不熟悉YAML格式的测试人员需求,进一步降低使用门槛。
七、参考项目

如果想直接落地,可以参考我的开源示例项目:api-auto-test,里面包含了完整的工具类实现、YAML 模板、生成器代码和执行脚本。
               
1.png
        
2.png
        本文作者:给你一页白纸        版权申明:本博客所有文章除特殊声明外,均采用BY-NC-SA         许可协议。转载请注明出处!        声援博主:如果觉得这篇文章对您有帮助,请点一下右下角的                “推荐”                图标哦,您的                “推荐”                是我写作的最大动力。您也可以点击下方的        【关注我】        按钮,关注博主不迷路。
来源:程序园用户自行投稿发布,如果侵权,请联系站长删除
免责声明:如果侵犯了您的权益,请联系站长,我们会及时删除侵权内容,谢谢合作!

相关推荐

您需要登录后才可以回帖 登录 | 立即注册