Pytest-集成Requests与Allure搭建接口自动化框架

1. 项目目录结构

在正式开始之前，先了解一下整个项目的目录结构，这样更容易理解代码的组织方式。完整的接口自动化测试框架目录结构如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
./pytest_api_testing_framework
├── allure-results                  # allure测试结果
├── config                          # 存放各环境配置文件
│   ├── __init__.py
│   ├── common_config.py
│   ├── config.dev.ini
│   ├── config.prod.ini
│   └── config.sit.ini
├── conftest.py                     # 全局共享的fixture
├── data                            # 参数化
├── logs                            # 日志
│   └── 2024-11-25_TEST.log
├── pytest.ini                      # pytest配置文件
├── scripts                         # 存放测试脚本
│   ├── __init__.py     
│   └── test_login.py
├── uri                             # URI管理模块，存放各接口路径
│   ├── __init__.py
│   └── endpoints.py
└── utils                           # 工具类/方法
    ├── __init__.py
    ├── base_request.py
    ├── common_assert.py
    ├── logger.py
    └── tools.py

2. 准备工作

在开始编写代码之前，首先需要安装一些必要的依赖：

1
pip install requests pytest allure-pytest configparser loguru curlify
  • requests：用于发送HTTP请求。
  • pytest：用于编写和运行测试用例。
  • allure-pytest：用于生成Allure报告。
  • configparser：用于读取配置文件，支持多环境。
  • loguru：用于增强日志记录。
  • curlify：用于将请求转换为cURL命令，方便调试。

3. 配置文件管理（多环境支持）

为了实现多环境支持，我们在config/目录下创建不同的配置文件，例如config.dev.ini、config.prod.ini，存储不同环境的配置信息。

1
2
3
4
# config.dev.ini
[env]
host = http://localhost:1248

1
2
3
4
# config.prod.ini
[env]
host = https://arkrunner.cn

接着，在config目录下编写一个读取配置文件的工具模块。通过common_config.py来读取配置

1
2
3
4
5
6
7
8
9
10
11
12
13
# config/common_config.py.py
import configparser
import os

# 获取当前运行文件的绝对路径
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
# 读取配置文件
conf = configparser.ConfigParser()

env = os.environ.get('ENV', 'dev')

conf.read(os.path.join(BASE_DIR, f'config.{env}.ini'), encoding='utf-8')

这样就可以在运行测试时指定环境，比如：

1
ENV=dev pytest

4. 日志模块（基于Loguru）

在自动化测试中，日志记录是非常重要的一部分，可以帮助我们在测试失败时快速定位问题。我们在utils/
目录下创建一个日志模块，使用loguru来简化日志管理。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
# utils/logger.py
import logging
import os
import sys
import time

from loguru import logger as uru_logger

BASEDIR = os.getcwd()

log_path = os.path.join(os.path.join(BASEDIR, "logs"), time.strftime("%Y-%m-%d_HIS_TEST.log"))
flag = 0
handler_id = 1
file_log_handler_flag = 0
allure_log_handler_flag = 0


class AllureHandler(logging.Handler):
    def emit(self, record):
        logging.getLogger(record.name).handle(record)


class MyLogger:
    logger = uru_logger

    # log level: TRACE < DEBUG < INFO < SUCCESS < WARNING < ERROR < CRITICAL
    def __init__(self, level: str = "debug", log_file_path=log_path):
        self.stdout_handler(level="warning")
        self.file_handler(level="debug", log_file_path=log_file_path)
        # 多线程不开启allure日志，日志会被打乱
        # self.allure_handler(level=level)

    def stdout_handler(self, level):
        """配置控制台输出日志"""
        global flag
        # 添加控制台输出的格式,sys.stdout为输出到屏幕;
        if flag != 0:
            return
            # 清空所有设置
        self.logger.remove()
        h_id = self.logger.add(
            sys.stdout,
            level=level.upper(),
            format="<green>{time:YYYY-MM-DD HH:mm:ss}</green> "  # 颜色>时间
                   "<m>[{thread.name}]</m>-"  # 进程名
                   "<cyan>[{module}</cyan>.<cyan>{function}</cyan>"  # 模块名.方法名
                   ":<cyan>{line}]</cyan>-"  # 行号
                   "<level>[{level}]</level>: "  # 等级
                   "<level>{message}</level>",  # 日志内容
        )
        flag += 1
        global handler_id
        handler_id = h_id

    def file_handler(self, level, log_file_path):
        """配置日志文件"""
        global file_log_handler_flag
        # 控制只添加一个file_handler
        if file_log_handler_flag == 0:
            self.logger.add(
                log_file_path,
                level=level.upper(),
                format="{time:YYYY-MM-DD HH:mm:ss} "
                       "[{thread.name}]-"
                       "[{module}.{function}:{line}]-[{level}]:{message}",
                rotation="10 MB",
                encoding="utf-8",
            )
            file_log_handler_flag += 1

    def allure_handler(self, level):
        """日志输出到allure报告中"""
        _format = "{time:YYYY-MM-DD HH:mm:ss} [{module}.{function}:{line}]-[{level}]:{message}"
        self.logger.add(AllureHandler(), level=level.upper(), format=_format)

    @classmethod
    def change_level(cls, level):
        """更改stdout_handler级别"""
        # 清除stdout_handler配置
        logger.remove(handler_id=handler_id)
        # 重新载入配置
        cls.logger.add(
            sys.stdout,
            level=level.upper(),
            format="<green>{time:YYYY-MM-DD HH:mm:ss}</green> "  # 颜色>时间
                   "<m>[{process.name}]</m>-"  # 进程名
                   "<m>[{thread.name}]</m>-"  # 进程名
                   "<cyan>[{module}</cyan>.<cyan>{function}</cyan>"  # 模块名.方法名
                   ":<cyan>{line}]</cyan>-"  # 行号
                   "<level>[{level}]</level>: "  # 等级
                   "<level>{message}</level>",  # 日志内容
        )


_logger = MyLogger()
logger = _logger.logger

if __name__ == '__main__':
    logger.debug("debug...")
    logger.info("info...")
    logger.warning("warning...")
    logger.error("error...")
    print(BASEDIR)

Loguru提供了更强大的日志功能，包括日志文件的自动分割和保留策略，使日志管理更加灵活和方便。

5. URI统一管理

为了便于管理接口路径，我们在uri/目录下创建endpoints.py，用于存放所有API的路径：

1
2
3
4
# uri/endpoints.py
class Endpoints:
    LOGIN = "/api/login"
    USER_PROFILE = "/user/profile"

这样可以保证各个接口的路径统一管理，方便修改和维护。

6. 封装HTTP请求及请求日志记录

接下来，我们封装HTTP请求操作，在utils/目录下创建一个统一的请求类BaseRequest，并编写一个装饰器来记录每个请求的cURL命令。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
# Created on 2024/7/2.
# @author loki.zuo
from functools import wraps

import allure
import curlify
import requests

from config.common_config import conf
from utils.logger import logger
from utils.common_assert import CommonAssert


def record_curl_command(func):
    """
    用于记录请求Curl命令的装饰器
    :param func:
    :return:
    """

    @wraps(func)
    def wrapper(*args, **kwargs):
        result = func(*args, **kwargs)
        with allure.step('记录当前请求curl_command及完整返回'):
            curl_command = curlify.to_curl(result.request, ).encode('utf-8').decode('unicode_escape')
            logger.info(f'请求Curl:  {curl_command}')
            # logger.info(f'返回数据: {result.text}')
            allure.attach(curl_command, 'Curl', allure.attachment_type.TEXT)
            allure.attach(result.text, '返回文本', allure.attachment_type.JSON)
        if result.status_code != 200:
            logger.warning('请求curl命令' + curlify.to_curl(result.request))
            logger.warning('返回数据: ' + str(result.text))
        # 加入通用断言

        if 'common_assert' not in kwargs or kwargs['common_assert']:
            CommonAssert.assert_equal(200, result.status_code)
            CommonAssert.assert_equal(200, int(result.json()['resultCode']),
                                      'resultMessage' + (result.json()['resultMessage']))

        return result

    return wrapper


class BaseRequest(object):
    def __init__(self):
        self.host = conf.get('env', 'host')
        self.common_headers = {"Content-Type": "application/json",
                               'User-Agent': 'Mozilla/5.0 (Linux; Android 6.0; Nexus 5 Build/MRA58N) '
                                             'AppleWebKit/537.36 (KHTML,'
                                             'like Gecko) Chrome/126.0.0.0 Mobile Safari/537.36'}

    @allure.step('发送GET请求: uri: {uri}')
    @record_curl_command
    def get(self, uri, headers=None, params=None, common_assert=True, timeout=15):
        if headers is not None:
            headers = headers | self.common_headers
        return requests.get(url=self.host + uri, headers=headers, params=params, timeout=timeout)

    @allure.step('发送POST请求: uri: {uri}')
    @record_curl_command
    def post(self, uri, data=None, headers=None, common_assert=True, timeout=15):

        if headers is not None:
            headers = headers | self.common_headers
        return requests.post(url=self.host + uri, json=data, headers=headers, timeout=timeout)

    @allure.step('发送PUT请求: uri: {uri}')
    @record_curl_command
    def put(self, uri, data=None, headers=None, params=None, common_assert=True, timeout=15):

        if headers is not None:
            headers = headers | self.common_headers
        return requests.put(url=self.host + uri, json=data, headers=headers, params=params, timeout=timeout)

    @allure.step('发送Patch请求: uri: {uri}')
    @record_curl_command
    def patch(self, uri, data=None, headers=None, params=None, common_assert=True, timeout=15):

        if headers is not None:
            headers = headers | self.common_headers
        return requests.patch(url=self.host + uri, json=data, headers=headers, params=params, timeout=timeout)

    @allure.step('发送DELETE请求: uri: {uri}')
    @record_curl_command
    def delete(self, uri, headers=None, params=None, common_assert=True, timeout=15):

        if headers is not None:
            headers = headers | self.common_headers
        return requests.delete(url=self.host + uri, headers=headers, params=params, timeout=timeout)

通过record_curl_command装饰器，每次请求都会记录一个相应的cURL命令，便于调试和重现请求。如果请求失败，日志中也会记录下详细信息。

7. 编写测试用例

在scripts/目录下，我们编写一个简单的登录测试用例。在这里，我们会结合Allure的注解，使报告更详细：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
# scripts/test_login.py

import allure

from uri.endpoints import Endpoints
from utils.base_request import base_request
from utils.logger import logger


@allure.feature("用户模块")
@allure.story("用户登录")
@allure.severity(allure.severity_level.CRITICAL)
def test_login():
    with allure.step("发送登录请求"):
        data = {"username": "sky", "password": "123456"}
        response = base_request.post(uri=Endpoints.LOGIN, data=data)

    with allure.step("验证响应数据"):
        assert "token" in response.json()['data'], "Response does not contain token"

8. Pytest配置及运行测试

在pytest.ini中进行pytest全局配置

1
2
3
4
5
6
7
8
9
10
11
12
13
14
[pytest]
# pytest命令行参数
addopts = -sv --alluredir allure-results --clean-alluredir --capture=sys --reruns 0 --junitxml=./static/report.xml --log-format="%(message)s"


# Allure 中文支持
disable_test_id_escaping_and_forfeit_all_rights_to_community_support = True

# 定义测试用例root目录及用例匹配规则
testpaths = ./scripts/
python_files = test*.py
python_classes = Test*
python_functions = test_*

在命令行中执行pytest：

1
pytest

然后用Allure命令行工具查看报告：

1
allure serve ./allure-results

这样，你就可以看到一个美观且详细的测试报告，里面包含了所有步骤的信息。

9. 项目目录结构更新

在实现上述功能后，项目的目录结构如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
./pytest_api_testing_framework
├── allure-results                  # allure测试结果
├── config                          # 存放各环境配置文件
│   ├── __init__.py
│   ├── common_config.py
│   ├── config.dev.ini
│   ├── config.prod.ini
│   └── config.sit.ini
├── conftest.py                     # 全局共享的fixture
├── data                            # 参数化
├── logs                            # 日志
│   └── 2024-11-25_TEST.log
├── pytest.ini                      # pytest配置文件
├── scripts                         # 存放测试脚本
│   ├── __init__.py     
│   └── test_login.py
├── uri                             # URI管理模块，存放各接口路径
│   ├── __init__.py
│   └── endpoints.py
└── utils                           # 工具类/方法
    ├── __init__.py
    ├── base_request.py
    ├── common_assert.py
    ├── logger.py
    └── tools.py

10.总结

本文介绍了一个最基本的接口自动化测试框架搭建流程，旨在为初学者提供入门演示。需要注意的是，这只是一个简单的 Demo，并未考虑企业级项目中的各种复杂场景和细节优化。如果您计划将类似框架应用于实际项目，建议结合具体需求进行更深入的优化和扩展，例如：

  • 多系统/多角色的鉴权管理
  • 更加完善的多环境配置文件管理
  • 与Jenkins结合实现持续集成
  • 企业微信/钉钉/飞书webhook推送
  • 与项目管理工具结合，在结果异常时创建工单

当前，如果你愿意花点元子，我将提供真正的企业级接口自动化测试框架及演示项目，手把手进行教学