如何提升接口自动化脚本的编码效率？-Ally Halo Blog

在接口自动化测试实践中，编码效率直接影响项目交付质量和迭代速度。面对复杂的业务逻辑和频繁的需求变更，如何提升编码效率是一个关键问题。本文将从方法封装、日志生成、数据驱动三个方面，结合实际工作中的落地示例，分享如何高效编写接口自动化脚本。

一、通用方法封装

断言方法封装
接口请求方法封装
业务数据方法封装

1. 断言方法封装

在接口测试中，断言是验证接口返回结果是否符合预期的核心步骤。如果每个测试用例都手动编写断言逻辑，不仅效率低下，还容易出错。通过封装断言方法，可以大幅减少重复代码。

思路：主要就是进行 json 字符串的对齐

返回体需要校验的内容：

接口文档中所描述的返回字段是否存在
返回体中不存在除接口文档外的多余字段
返回体中字段的数据类型
数据源校验，前置构建的数据和查询到的数据需要一一对齐
“子对象”也需要校验

# common/assertions.py
import unittest
from common.caseOutput import error


class AssertCommon(unittest.TestCase):
    def code_assert(self, expect, actual):
        if expect != actual:
            text = f'res code different, expect code: {expect}, actual code: {actual}.'
            error('assert fail! ' + text)
            self.fail(text)

    def __assertEqual(self):
        pass

    def json_assert(self, expect, actual):
        """
        json通用断言方法
        :param expect: 定义预期返回体
        :param actual: 实际返回的json
        :return: 断言成功返回None，断言失败触发fail
        """
        # 字段是否存在的校验
        for key, value in expect.items():
            self.assertIn(key, actual.keys())
        # 校验是否存在多余的字段
        self.assertEqual(len(expect.keys()), len(actual.keys()),
                         msg=f'response keys len different, response keys have: {list(actual.keys())}')
        for key, value in expect.items():
            # 进行数据类型的校验
            if isinstance(value, type):
                self.assertEqual(value, type(actual[key]),
                                 msg=f'{key} type error! actual type is {str(type(actual[key]))}')
            elif isinstance(value, list):
                for i in range(len(value)):
                    if isinstance(value[i], type):
                        self.assertEqual(value[i], type(actual[key][i]),
                                         msg=f'list element {actual[key][i]} type different, actual response {actual[key]}')
                    elif isinstance(value[i], dict):
                    	# 遇到列表字典的嵌套结构，需要实现递归
                        self.json_assert(value[i], actual[key][i])
                    else:
                        self.assertEqual(value[i], actual[key][i],
                                         msg=f'list element {actual[key][i]} value different, actual response {actual[key]}')
            else:
                self.assertEqual(value, actual[key],
                                 msg=f'{key} value error! actual value is {str(actual[key])}')

2. 接口请求方法封装

不是直接封装通用的请求方法，而是为了减少编码成本提高代码可读性的封装，能够在当前业务场景的基础上做调整。

场景一：如果是http协议，可以根据当前业务不同的请求方式进行封装。

封装内容：

经常被复用的请求头内容
请求数据和返回数据的日志输出
接口请求超时的处理机制

import requests
from common.caseOutput import info, error, case


class BusinessRe:
    @staticmethod
    def post(url, sid, user_id, body, headers=None):
        if headers is None:
            headers = {
                'Cookie': f'wps_sid={sid}',
                'X-user-key': str(user_id),
                'Content-Type': 'application/json'
            }

        info(f'request url: {url}')
        info(f'request headers: {headers}')
        info(f'request body: {body}')
        try:
            res = requests.post(url, headers=headers, json=body, timeout=5)
        except TimeoutError:
            error(f'url: {url}, requests timeout！')
            return 'Requests Timeout!'
        info(f'response code: {res.status_code}')
        info(f'response body: {res.text}')
        return res

场景二：如果是其他协议，比方说长连接 socket 协议

需要封装的内容：

长连接的建立方法
长连接的断开方法
长连接的重连方法
长连接的消息接收方法
长连接的消息发送方法
长连接的认证方法
长连接的数据加密方法

场景三：如果是自研的协议

需要封装的内容：

建立
断开
消息接收
消息发送

3. 业务数据方法封装

在真实工作中，接口自动化要尽量避免直接对数据库的操作，如果不进行管控可能会导致线上真实数据被清理的风险。而这时我们只能依靠业务所提供的接口来完成相关数据的构建或清理，如果每条用例都单独去实现前置数据的脚本实现和数据清理，编码成本会很高，这是咱们就需要对业务核心数据的构建、清理、查询进行业务层的方法。

数据清理场景：

数据清理方法，需要基于用户来实现，尽量满足调用方的便利性，根据业务数据清理的流程来实现接口调用过程即可。实现好的数据清理方法一般都是放在 setup 用例初始化方法下或个别业务场景。

from businessCommon.bR import BusinessRe
from common.yamlLoader import YamlRead

env_config = YamlRead().env_config()
api_config = YamlRead().api_config()


def all_notes_clear(user_id, sid):
    """指定某一个用户进行全量的便签清理"""
    # 获取全量的便签数据
    host = env_config['host']
    url = host + f'/v3/notesvr/user/{user_id}/home/startindex/0/rows/999/notes'
    res = BusinessRe().get(url, sid)
    for note in res.json()['webNotes']:
        note_id = note['noteId']
        del_url = host + api_config['note_delete']['path']
        body = {'noteId': note_id}
        BusinessRe().post(del_url, user_id=user_id, sid=sid, body=body)
    clean_note_url = host + api_config['cleanRecyclebin']['path']
    body = {
        'noteIds': [-1]
    }
    BusinessRe().post(clean_note_url, user_id=user_id, sid=sid, body=body)


if __name__ == '__main__':
    all_notes_clear(env_config['user_id'], env_config['wps_sid'])

构建数据方法：

如果是构建方法，需要考虑数量的控制，需要基于用户对象实现数据的构建方法；结果返回需要把构建时的关键信息返回回去。实现好的数据构建方法一般是直接写在用例当中。

import time
from businessCommon.bR import BusinessRe
from common.yamlLoader import YamlRead

env_config = YamlRead().env_config()
api_config = YamlRead().api_config()


class DataCreate:
    """创建用例前置和后置数据"""
    host = env_config['host']

    def note_create(self, num, user_id, sid, group_id=None, remind_time=None):
        """通用的便签新建方法"""
        note_lists = []
        for i in range(num):
            note_id = str(int(time.time() * 1000))
            # 新建便签主体接口
            url_info = self.host + api_config['note_create_info']['path']

            if remind_time:  # 日历便签
                body = {
                    "noteId": note_id,
                    "remindTime": remind_time,
                    "remindType": 0,
                    'star': 0
                }

            elif group_id:  # 分组便签
                body = {
                    "noteId": note_id,
                    "groupId": group_id,
                    'star': 0
                }

            else:
                body = {
                    "noteId": note_id,
                    'star': 0
                }

            BusinessRe.post(url_info, sid=sid, user_id=user_id, body=body)

            url_content = self.host + api_config['note_create_content']['path']
            body = {
                "noteId": note_id,
                "title": '75u8dlZyTLqWCm/b2PLNlg==',
                "summary": 'pIDnRrCwq8sUW3gyWpo7iw==',
                "body": 'wlby4RxbJjQKcx7rwTpn/w==',
                "localContentVersion": 1,
                "BodyType": 0
            }

            BusinessRe.post(url_content, sid=sid, user_id=user_id, body=body)
            note_lists.append(body)
        return note_lists

二、日志自动生成

在自动化测试中，日志是定位问题的重要工具。通过自动输出关键日志（如请求参数、响应结果、断言结果等），可大幅提升调试效率。

# common/logsOutput.py
from colorama import Fore
import functools
import time
import inspect
from datetime import datetime
import os
from main import DIR

timeout = 10


def case(text):
    """
    打印用例信息并输出对应的日志
    :param text: str 控制台要输出的内容或要打印的日志文本数据
    :return:
    """
    formatted_time = datetime.now().strftime('%H:%M:%S:%f')[:-3]  # 日志的输出时间
    stack = inspect.stack()
    code_path = f"{os.path.basename(stack[1].filename)}:{stack[1].lineno}"  # 当前执行文件的绝对路径和执行代码行号
    content = f"[CASE]{formatted_time}-{code_path} >> {text}"
    print(Fore.LIGHTCYAN_EX + content)
    str_time = datetime.now().strftime("%Y%m%d")
    with open(file=DIR + '\\logs\\' + f'{str_time}_info.log', mode='a', encoding='utf-8') as f:
        f.write(content + '\n')


def info(text):
    """
    打印用例运行时数据并输出对应的日志
    :param text: str 控制台要输出的内容或要打印的日志文本数据
    :return:
    """
    formatted_time = datetime.now().strftime('%H:%M:%S:%f')[:-3]  # 定义了日志的输出时间
    stack = inspect.stack()
    code_path = f"{os.path.basename(stack[1].filename)}:{stack[1].lineno}"  # 当前执行文件的绝对路径和执行代码行号
    content = f"[INFO]{formatted_time}-{code_path} >> {text}"
    print(Fore.WHITE + content)
    str_time = datetime.now().strftime("%Y%m%d")
    with open(file=DIR + '\\logs\\' + f'{str_time}_info.log', mode='a', encoding='utf-8') as f:
        f.write(content + '\n')


def error(text):
    """
    打印用例断言失败信息或异常信息并输出对应的日志
    :param text: str 控制台要输出的内容或要打印的日志文本数据
    :return:
    """
    formatted_time = datetime.now().strftime('%H:%M:%S:%f')[:-3]  # 定义了日志的输出时间
    stack = inspect.stack()
    code_path = f"{os.path.basename(stack[1].filename)}:{stack[1].lineno}"  # 当前执行文件的绝对路径和执行代码行号
    content = f"[ERROR]{formatted_time}-{code_path} >> {text}"
    print(Fore.LIGHTRED_EX + content)
    str_time = datetime.now().strftime("%Y%m%d")
    with open(file=DIR + '\\logs\\' + f'{str_time}_info.log', mode='a', encoding='utf-8') as f:
        f.write(content + '\n')
    with open(file=DIR + '\\logs\\' + f'{str_time}_error.log', mode='a', encoding='utf-8') as f:
        f.write(content + '\n')


def case_decoration(func):
    @functools.wraps(func)  # 解决参数冲突问题
    def case_improve(*args, **kwargs):
        start = time.perf_counter()
        class_name = args[0].__class__.__name__  # 获取类名
        method_name = func.__name__  # 获取方法名
        docstring = inspect.getdoc(func)  # 获取方法注释
        case('-------------------------------------------------------------------------')
        case(f"Method Name:{method_name}, Class Name:{class_name}")
        case(f"Test Description:{docstring}")
        func(*args, **kwargs)  # 执行测试用例
        end = time.perf_counter()
        handle_time = end - start
        case('Case run time: %.2fs' % handle_time)
        if handle_time > timeout:
            error('case run timeout!')

    return case_improve


def class_case_decoration(cls):
    """用例的日志类级别装饰器"""
    for name, method in inspect.getmembers(cls, inspect.isfunction):
        if name.startswith('test'):
            setattr(cls, name, case_decoration(method))
    return cls

三、数据驱动的测试

核心思想‌：数据与测试代码分离，使测试代码在测试数据大量变化时保持不变。
实现方式‌：通常将测试数据存储在外部文件（如Excel、CSV、YAML等）中，通过特定的方法（如openpyxl、ddt、parametrize等）读取数据，并将其传入测试脚本中。
优点‌：提高了测试的复用性和执行效率，降低了维护成本，同时增强了测试的灵活性和可扩展性。

数据驱动的落地项：变量集中管理

在接口自动化测试中，变量管理是提升脚本可维护性的关键。通过集中管理变量，可以避免硬编码，提升脚本的灵活性。

可提取的变量数据：

环境维度：用户id、身份信息、host、加密秘钥、开关等
接口维度：path、必填参数、选填参数、请求demo、返回体校验的demo

实现步骤：

step1：用例中可复用的变量需要提炼在类属性上。
step2：将类类型按环境和接口进行分类。
step3：将环境相关的类属性存储在环境变量的 yaml 配置文件当中。
step4：实现 yaml 读取方法。
step5：将接口相关的类属性（接口路径、请求方式、必填参数、非必填参数）存在apiconfig 配置文件中。

类属性的管理：

# common/configLoder.py 
from main import DIR, ENVIRON
import yaml


class YamlRead:
    @staticmethod
    def env_config():
        """环境变量的读取方式"""
        with open(file=f'{DIR}/data/envConfig/{ENVIRON}/config.yml', mode='r', encoding='utf-8') as f:
            return yaml.load(f, Loader=yaml.FullLoader)

    @staticmethod
    def api_config():
        with open(file=f'{DIR}/data/apiConfig/config.yml', mode='r', encoding='utf-8') as f:
            return yaml.load(f, Loader=yaml.FullLoader)

接口 yaml 配置文件管理

note_create_info:
  path: /v3/notesvr/set/noteinfo
  method: post
  mustKeys:
    - noteId
  notMustKeys:
    - star
    - remindTime
    - remindType
    - groupId
group_create:
  path: /v3/notesvr/set/notegroup
  method: post
  mustKeys:
    - groupId
    - groupName
  notMustKeys:
    - order

环境变量 yaml 配置管理

host: 'http://note-api.xxx.cn'
user_id: 90000001
wps_sid: 'V02SdJIgPxxxxxxxxxxxxxxxxxxx58bfd'

yaml 配置读取方法封装

from main import DIR, ENVIRON
import yaml


class YamlRead:
    @staticmethod
    def env_config():
        """环境变量的读取方式"""
        with open(file=f'{DIR}/data/envConfig/{ENVIRON}/config.yml', mode='r', encoding='utf-8') as f:
            return yaml.load(f, Loader=yaml.FullLoader)

    @staticmethod
    def api_config():
        with open(file=f'{DIR}/data/apiConfig/config.yml', mode='r', encoding='utf-8') as f:
            return yaml.load(f, Loader=yaml.FullLoader)

目录CONTENT

如何提升接口自动化脚本的编码效率？