基于Python+Requests+Allure搭建接口自动化框架 | 极客日志

Python

基于Python+Requests+Allure搭建接口自动化框架

基于Python结合Requests库发送HTTP请求，利用Pytest作为测试执行器，Allure生成报告，以及YAML进行数据驱动的接口自动化测试框架搭建过程。涵盖关键字封装、全局配置管理、用例数据关联、断言逻辑（含数据库）、日志集成、Session复用及文件上传下载等功能，并提供了项目打包部署方案。

雪落无声发布于 2026/2/9更新于 2026/4/183 浏览

Python 中的*args 与**kwargs

接收可变参数

*args：会将传入的额外位置参数打包成一个元组（tuple）
**kwargs：会将传入的额外关键字参数打包成一个字典（dict）

# 调用时，参数只需传递一个一个的值，自动会打包为元组 args
def my_tool(*args):
    print(args)


 ():
    (kwargs)

 __name__ == :
    my_tool(, , )  
    my_tools(name=, age=)

data = {'username': 'zs', 'password': '123'}
names = ['zs', 'ls', 'wu']

def func1(username, password):
    print(username + password)

def func2(name1, name2, name3):
    print(name1 + name2 + name3)

if __name__ == '__main__':
    func1(**data)  # zs123
    func2(*names)  # zslswu

import requests
url = 'http://127.0.0.1:5000/hello'
data = {'username': 'zs', 'password': 123}
# 发送 get 请求
# 请求方式一：requests.request(method, url, **kwargs)
ret = requests.request('get', url=url)
print(ret.json())

# 请求方式二：requests.get(url, params=None, **kwargs)
ret = requests.get(url=url, params=data)
print(ret.json())

ret = requests.request('post', url=url, json=data)
ret = requests.request('post', url=url, data=data)
print(ret.json())

ret = requests.post(url=url, json=data)
ret = requests.post(url=url, data=data)
print(ret.json())
# data 与 json 参数区别：
# json 要求传字典，json.loads 将字符串转为字典，content-type 为 application/json
# data 要求字符串，json.dumps 将字典转为字符串，content-type 不是 json 一般为 application/x-www-form-urlencoded
# data 可以传字典等格式数据，只是会自动编码为字符串，如果传的本身是字符串，则不进行编码

params = {'page': 2, 'len': 10}
requests.post(url=url, json=data, params=params)

# 上传文件
ret = requests.post(url=url, files={'image': open('a.jpg', 'rb')})

session = requests.session()
# 登录
ret = session.post(url=url, json=data)
# 登录后操作
ret = session.post(url=url, json=data)

import jsonpath
jsonpath.jsonpath(response, '$.name')
jsonpath.jsonpath(response, '$..title')
jsonpath.jsonpath(response, '$.names[0]')
jsonpath.jsonpath(response, '$.names[*]')
jsonpath.jsonpath(response, '$.books[0,3]')  # 返回 book 数组中 category 为 fiction 的元素
jsonpath.jsonpath(response, "$.store.book[?(@.category == 'fiction')]")

base_config:
  case_type: 'ApiCase'
  case_name: 'login'
  case_module: '教务系统'
  author: 'hlk'
test_steps:
  - 登录接口:
      request_type: send_post
      url: 'http://127.0.0.1'
      params:
        s: /hello
        pageIndex: 2
      data_type: json
      request_data:
        username: hlk
        password: 123

import yaml

def read_yaml(file_path):
    data = []
    with open(file_path, 'r', encoding='utf-8') as file:
        data.append(yaml.safe_load(file))
    return data

class MyRequest:
    def __init__(self, request):
        self.request = request

    def send_post(self, **kwargs):
        url = kwargs.get('url', None)
        params = kwargs.get('params', None)
        headers = kwargs.get('headers', None)
        files = kwargs.get('files', None)
        data = kwargs.get('data', None)
        request_data = {'url': url, 'params': params, 'headers': headers, 'files': files}
        if kwargs.get('data_type') == 'json':
            request_data['json'] = data
        if kwargs.get('data_type') == 'data':
            request_data['data'] = data
        try:
            ret = self.request.request('post', **request_data)
        except Exception as e:
            print('请求异常：', e)
        return ret

import pytest, requests
from api_frame.HAT.parse.read_yml import read_yaml
from api_frame.HAT.utils.my_request import MyRequest

class TestRunner:
    case_info = read_yaml('../../test_cases/yml_cases/login.yml')

    @pytest.mark.parametrize('case_info', case_info)
    # case_info 要求是列表
    def test_excute_case(self, case_info):
        my_request = MyRequest(requests)
        base_config = case_info.get('base_config', {})  # 基础配置，用于报告
        test_steps = case_info.get('test_steps', [])  # 测试步骤列表
        for test_step in test_steps:
            # step_name = list(test_step.keys())[0]
            # step_value = list(test_step.values())[0]
            (step_name, step_value), = test_step.items()  # 注意逗号，解包一个包含一个元素的可迭代对象
            request_type = step_value.get('request_type')  # 发送请求
            try:
                request_func = my_request.__getattribute__(request_type)  # 获取请求方法
            except Exception as e:
                print('my_request 中没有找到对应的方法', e)
            ret = request_func(**step_value).json()
            print(ret)

if __name__ == '__main__':
    pytest.main(['-vs', __file__])

import pytest, os
from allure_combine import combine_allure

# -vs 详细内容
# --capture=sys 系统配置 生成 stdout 附件
# --clean-alluredir 清空报告报告数据，保证每次生成报告都使用最新的数据
# --alluredir=allure-results 结果数据目录，生成的 原始测试结果文件（JSON 格式）
# ./core/test_runner.py 用例文件夹
pytest_args = ['-vs', '--capture=sys', '--clean-alluredir', '--alluredir=allure-results', './core/test_runner.py']
pytest.main(pytest_args)

# 生成测试报告
# allure-report 最终测试报告目录，根据 allure-results 中数据创建报告
os.system('allure generate allure-results -o allure-report -c')
# -o 指定目录 -c 先清空在生成
# 生成的报告只能在 pycharm 打开，需要安装 allure-combine
# npm install -g allure-combine
combine_allure('./allure-report')

base_config = case_info.get('base_config', {})
allure.dynamic.parameter('case_info', '')  # 记录参数
allure.dynamic.feature(base_config.get('case_module'))  # 用例所属模块
allure.dynamic.story(base_config.get('case_module_sec'))  # 模块子功能
allure.dynamic.title(base_config.get('case_name'))  # 用例标题

with allure.step(step_name):  # 划分步骤，必须在函数中使用
    request_type = step_value.get('request_type')  # 发送请求
    try:
        request_func = my_request.__getattribute__(request_type)  # 获取请求方法
    except Exception as e:
        print('my_request 中没有找到对应的方法', e)
    ret = request_func(**step_value).json()
    print(ret)

# 调用该函数时，自动创建 step
@allure.step('发送 post 请求')
def send_post(self, **kwargs):
    url = kwargs.get('url', None)
    params = kwargs.get('params', None)
    headers = kwargs.get('headers', None)
    files = kwargs.get('files', None)
    data = kwargs.get('data', None)

test_steps = case_info.get('test_steps', [])  # 测试步骤列表
# total=len(test_steps)：告诉 tqdm 总共有多少步（用于计算百分比）
# desc='开始执行'：初始描述文字（显示在进度条左边）
# with ... as pbar：使用上下文管理器，确保进度条正确关闭
# pbar：进度条对象，后续可通过它动态更新内容
with tqdm(total=len(test_steps), desc='开始执行') as pbar:
    for test_step in test_steps:
        # step_name = list(test_step.keys())[0]
        # step_value = list(test_step.values())[0]
        (step_name, step_value), = test_step.items()  # 注意逗号，解包一个包含一个元素的可迭代对象
        pbar.set_description(f'{base_config.get("case_name")}-当前步骤:{step_name}')  # 进度条描述，进度条左侧的文字
        pbar.update(1)  # 更新进度，进度条前进一格

login_url: 'http://127.0.0.1:8080/login'
register_url: 'http://127.0.0.1:8080/register'
file_upload_url: 'http://127.0.0.1:8080/upload'
file_download_url: 'http://127.0.0.1:8080/download/a.png'
databases:
  hlk_test_dev:
    user: 'root'
    database: 'hlk_test_dev'
    password: '12345678'
    host: 'localhost'
    port: 3306
session_reuse: True

test_steps:
  - 添加接口:
      request_type: send_post
      url: '{{login_url}}'
      data_type: json
      request_data:
        username: hlk
        password: 123
        token: '{{token}}'

class GlobalVar:
    _dict = {}

    def set_dict(self, dict):
        # 先查找实例属性_dict，找不到再查找类属性_dict
        # 如果使用赋值操作如 self._dict = {}, 则会创建一个实例属性_dict，不会修改到类属性
        # 如果使用赋值操作 (self.attr = value)，会优先在实例级别创建一个新的属性 attr，而不会修改类属性
        self._dict.update(dict)
        # GlobalVar._dict.update(dict)
        # 添加@classmethod

    def set_dict_by_kv(self, key, value):
        self._dict[key] = value

    def get_dict(self):
        return self._dict

    def get_dict_by_key(self, key):
        return self._dict.get(key)

global_var = read_context('./test_cases/yaml_cases/context.yml')
# def read_context(file_path):
#     with open(file_path, 'r', encoding='utf-8') as file:
#         data = yaml.load(file, Loader=yaml.SafeLoader)
#     return data
GlobalVar().set_dict(global_var)  # 设置到全局变量中

from jinja2 import Template
# jinja2 的模版必须是字符串，且渲染后返回字符串
# 渲染 yml 文件，eval 将字符串转为原本字典
step_value = eval(Template(str(step_value)).render(GlobalVar().get_dict()))

base_config:
  case_type: 'ApiCase'
  case_name: 'login'
  case_story: '登录模块'
  case_module: '教务系统'
  author: 'hlk'
setup_script:
  - "print('前置执行 1')"
  - "print('前置执行 2')"
teardown_script:
  - "print('后置执行 1')"
  - "print('后置执行 2')"
test_steps:
  - 登录接口:
      request_type: send_post
      url: '{{login_url}}'
      data_type: json
      request_data:
        username: hlk
        password: 123

# 前置脚本执行
setup_scripts = case_info.get('setup_script')
if setup_scripts:
    for setup_script in setup_scripts:
        # 执行任意代码
        exec(setup_script, {'context': GlobalVar().get_dict()})

# 后置脚本执行
teardown_scripts = case_info.get('teardown_script')
if teardown_scripts:
    for teardown_script in teardown_scripts:
        # 执行任意代码
        exec(teardown_script, {'context': GlobalVar().get_dict()})

# 读取文件夹下符合规则的 yaml 文件
def read_rule_yaml(file_dir):
    case_infos = []
    files = os.listdir(file_dir)  # 列表推导式，yml 文件且是 test 结尾的
    rule_files = [file for file in files if file.endswith('yml') and file.split('.')[0].endswith('test')]
    rule_files.sort()
    for rule_file in rule_files:
        file = os.path.join(file_dir, rule_file)  # 文件路径
        with open(file, 'r', encoding='utf-8') as f:
            case_info = yaml.safe_load(f)
            case_infos.append(case_info)
    return case_infos

case_info = read_rule_yaml('../test_cases/yaml_cases/')  # 读取文件夹下满足规则的所有用例

response = self.request.request('post', **request_data)  # 将响应保存到全局变量中
GlobalVar().set_dict({'response_data': response.json()})

test_steps:
  - 登录接口:
      request_type: send_post
      url: '{{login_url}}'
      data_type: json
      request_data:
        username: hlk
        password: 123
  - 提取保存数据:
      request_type: extract_data
      extract_expression: '$..token'
      extract_var: 'token'

@allure.step('提取保存数据')
def extract_data(self, **kwargs):
    index = kwargs.get('extract_index', None)
    if index == None:
        # 提取下标默认为 0
        index = 0
    response = GlobalVar().get_dict_by_key('response_data')
    extract_expression = kwargs.get('extract_expression', None)
    extract_var = kwargs.get('extract_var', None)
    extract_value = jsonpath.jsonpath(response, extract_expression)[index]  # 从响应中提取数据
    if not extract_value:
        raise Exception(f'没有从响应中取到对应值{extract_expression}')
    GlobalVar().set_dict_by_kv(extract_var, extract_value)
    print(f'提取{extract_var}的值{extract_value}')

step_value = eval(Template(str(step_value)).render(GlobalVar().get_dict()))

- 删除接口:
    request_type: send_post
    url: '{{register_url}}'
    data_type: json
    request_data:
      username: hlk
      password: 123
      token: '{{token}}'

- 断言数据:
    request_type: validate_data
    validate:
      - ==:
          actual: '{{msg}}'  # 实际值已经在上述步骤中添加到全局变量中了
          expect: good
          assert_fail_reason: '实际与预期不相等'
      - in:
          actual: '{{msg}}'
          expect: 'goods'

@allure.step('断言数据')
def validate_data(self, **kwargs):
    comparators = {
        '==': lambda x, y: x == y,
        '!=': lambda x, y: x != y,
        '>=': lambda x, y: x >= y,
        '<=': lambda x, y: x <= y,
        'in': lambda x, y: x in y,
        'not in': lambda x, y: x not in y,
    }
    validates = kwargs.get('validate')
    for validate in validates:
        (comparator, validate_data), = validate.items()
        if comparator not in comparators:
            raise Exception(f'{comparator}比较符不在比较器中')
        expect = validate_data.get('expect')
        actual = validate_data.get('actual')
        assert_fail_reason = validate_data.get('assert_fail_reason', None)
        if not comparators[comparator](expect, actual):
            if assert_fail_reason:
                raise AssertionError(assert_fail_reason)
            else:
                raise AssertionError(f'{actual} 与 {expect} 不相等')

- 断言全部数据:
    request_type: validate_all_data
    expect:
      code: 20
      msg: 'goosd'
      data:
        token: 'esjdksaos'
        name: 'zs'
        uid: '44'
        auths: ['prod', 'dev']
    filter_fields: ["root['data']['uid']", "root['data']['token']"]

@allure.step('断言全部响应数据')
def validate_all_data(self, **kwargs):
    try:
        actual_response = GlobalVar.get_dict_by_key('response_data')  # 上次接口请求已经将响应放到全局变量里
        expect_response = kwargs.get('expect')
        filter_fields = kwargs.get('filter_fields', [])
        ignore_order = kwargs.get('ignore_order', True)  # 默认忽略排序
        ignore_case = kwargs.get('ignore_case', True)  # 默认忽略大小写
        data = {
            'exclude_paths': filter_fields,
            'ignore_order': ignore_order,
            'ignore_string_case': ignore_case
        }
        diff = DeepDiff(actual_response, expect_response, **data)
    except Exception as e:
        assertFalse, f'批量断言失败:{e}'
    # 如果 diff 成功则返回空列表，如果 diff 存在差异则返回有差异的信息
    # 所以希望 diff 为空时校验通过，diff 不为空时校验失败，所以需要加 not
    assert not diff, f'批量断言失败{diff}'

import pymysql
from pymysql import cursors

# 数据库链接
connect = pymysql.connect(
    user='root',
    database='hlk_test_dev',
    password='12345678',
    host='localhost',
    port=3306,
    cursorclass=cursors.DictCursor,  # 字典形式展示数据
    charset='utf8'
)

# 生成游标
cursor = connect.cursor()

# 游标执行 sql
sql = 'select * from user'
cursor.execute(sql)

# 游标获取结果
ret = cursor.fetchall()
print(ret)

# 关闭游标，关闭数据库
cursor.close()
connect.close()

- 提取数据库数据:
    request_type: extract_database
    database_name: hlk_test_dev
    sql: select * from user
    extract_var: [username, password]
- 断言数据库:
    request_type: validate_data
    validate:
      - in:
          actual: '''{{username}}'''
          expect: 'zs'
          assert_fail_reason: '实际与预期不相等'

databases:
  hlk_test_dev:
    user: 'root'
    database: 'hlk_test_dev'
    password: '12345678'
    host: 'localhost'
    port: 3306

@allure.step('提取数据库数据')
def extract_database(self, **kwargs):
    database_name = kwargs.get('database_name')
    db_config = GlobalVar().get_dict_by_key('databases').get(database_name)
    config = {'cursorclass': cursors.DictCursor, 'charset': 'utf8'}
    config.update(db_config)
    connect = pymysql.connect(**config)
    # 生成游标
    cursor = connect.cursor()
    # 游标执行 sql
    cursor.execute(kwargs.get('sql'))
    # 游标获取结果
    ret = cursor.fetchall()
    # [{'id': 1, 'username': 'zs', 'password': '123'}]
    # 关闭游标，关闭数据库
    cursor.close()
    connect.close()
    # 提取保存数据
    result = {}
    vars = kwargs.get('extract_var')  # 保存的变量名 [username, password]
    # 查询如果是多个结果则放在数组中
    for var in vars:
        if len(ret) <= 1:
            result[var] = ''
        else:
            result[var] = []
    for i in ret:  # [{'id': 1, 'username': 'zs', 'password': '123'}]
        for key, value in i.items():
            if key == var:
                if len(ret) <= 1:
                    result[var] = value
                else:
                    result[var].append(value)
    # result = {'username':[], 'password':[]}
    GlobalVar().set_dict(result)  # 保存到全局变量

test_steps:
  - 登录接口:
      request_type: send_post
      url: '{{login_url}}'
      data_type: json
      request_data:
        username: '{{username}}'
        password: '{{password}}'
  - 提取响应数据 msg:
      request_type: extract_data
      extract_expression: '$..msg'
      extract_var: 'msg'
  - 断言数据:
      request_type: validate_data
      validate:
        - ==:
            actual: '{{msg}}'
            expect: '{{expect}}'
            assert_fail_reason: '实际与预期不相等'

ddt:
  - title: '正确用户名'
    username: hlk
    password: 123
    expect: 'login success'
  - title: '错误用户名'
    username: zs
    password: 456
    expect: 'login fail'

def ddt_yaml_parse(file_dir):
    case_infos = read_rule_yaml(file_dir)
    new_case_infos = []
    for case_info in case_infos:
        ddts = case_info.get('ddt')  # 数据驱动
        if ddts:
            case_info.pop('ddt')  # 存在数据驱动时，遍历每个数据驱动，组装每一条新的 case
            for ddt in ddts:
                # 注意需要深拷贝，否则对同一变量修改，之前 new_case_infos 添加的也就同步被修改了
                case_info = copy.deepcopy(case_info)
                case_info.update({'ddt': ddt})
                new_case_infos.append((case_info))
        else:
            new_case_infos.append(case_info)  # 无数据驱动时直接添加到 case 数组中
    return new_case_infos

# 在模版渲染前，将 ddt 保存到全局变量中
GlobalVar().set_dict(case_info.get('ddt'))

from loguru import logger

# 保存到文件中
logger.add('log.log',
           rotation='100 MB',  # 单个日志文件大小
           retention='10 days')  # 保存时间

# 日志级别：debug<info<warning<error
logger.debug('调试级别')
logger.info('程序正常运行')
logger.warning('警告信息')
logger.error('错误信息')
try:
    a = 1/0
    logger.info('run...')
except Exception as e:
    logger.error(e)

time_str = datetime.now().strftime("%Y-%m-%d_%H:%M:%S")
logger.configure(handlers=[
    {"sink": sys.stdout, "level": 'DEBUG'},
    {"sink": sys.stderr, "level": 'DEBUG'},
    {"sink": f'./log/log_{time_str}.log', "level": 'DEBUG'}
])

# 配置日志
if not os.path.exists('./log'):
    os.mkdir('log')
time_str = datetime.now().strftime("%Y-%m-%d_%H:%M:%S")
logger.configure(handlers=[
    {"sink": sys.stdout, "level": 'DEBUG'},
    {"sink": sys.stderr, "level": 'DEBUG'},
    {"sink": f'./log/log_{time_str}.log', "level": 'DEBUG'}
])

import io
import allure
from loguru import logger

# 日志管理器
class AllureStepLogger():
    def __init__(self):
        # io 流从内存中读取数据，区别 open 是从文件读取数据
        # StringIO 读取文本，BytesIO 读取二进制
        self.log_buffer = io.StringIO()
        # log_buffer 存储读取到内存的日志信息
        self.sink_id = None

    # 上下文管理器，使用到 with 语句，会调__enter__方法
    def __enter__(self):
        self.sink_id = logger.add(self.log_buffer, level='DEBUG')

    # 退出 with 语句时，会调__exit__方法
    def __exit__(self, exc_type, exc_val, exc_tb):
        logger.remove(self.sink_id)
        log = self.log_buffer.getvalue()  # 获取 io 流读取到内存的日志信息
        # 添加到 allure 中
        if log.strip():  # strip 去除首和尾指定字符，默认空格
            allure.attach(
                body=log,  # 附件内容
                name='步骤日志',  # 附件名称
                attachment_type=allure.attachment_type.TEXT  # 附件类型，JSON，ZIP，TEXT
                # , # 附件扩展名，不提供则根据附件类型推测
            )
            # 向 allure 报告中添加附件
        self.log_buffer.close()

# 关联测试步骤与报告
from contextlib import contextmanager

@contextmanager  # 装饰器，该函数可以使用 with 语句
def allure_step_with_log(step_name):
    with allure.step(step_name):
        with AllureStepLogger() as collector:
            # yield 之前代码相当于__enter__，yield 之后相当于__exit__，yield 后的值返回给调用方
            # @contextmanager 必须和 yield 同时使用
            # yield 暂停函数的执行，并返回一个值给调用方，将控制权交回调用方
            # 这里每次使用 with 调用时都会创建新对象，与之前调用无关
            yield collector

with allure_step_with_log(step_name):  # with allure.step(step_name): # 报告右侧显示的步骤

session_reuse: True

# 用于判断是否 session 复用，决定使用何种方式进行实例化关键字类
global_session = None

class ApiCaseContext:
    def __init__(self):
        self.request = None
        self.my_request = None

    def init_kwargs_func(self):
        global global_session
        session_reuse = GlobalContext().get_dict_by_key('session_reuse')  # 从全局变量拿配置
        if session_reuse is not None and session_reuse == True:
            if global_session == None:
                global_session = requests.session()
            self.request = global_session
        else:
            self.request = requests
        self.my_request = MyRequest(self.request)
        return self.my_request

# my_request = MyRequest(requests)
my_request = ApiCaseContext().init_kwargs_func()

test_steps:
  - 文件上传接口:
      request_type: send_post
      url: '{{file_upload_url}}'
      data_type: files
      request_data:
        img: 'a.png'  # 文件路径

if data_type == 'files':
    if isinstance(data, dict):
        files = {}
        for key, file_path in data.items():
            files[key] = open(file_path, 'rb')
        request_data['files'] = files
    else:
        file = open(data, 'rb')
        request_data['files'] = {'files': file}

- 文件下载接口:
    request_type: send_get
    url: '{{file_download_url}}'
    save_path: '/Users/user/Desktop/hlk_test_dev/requests_test/HAT/utils/download/b.png'  # 保存文件路径
    stream: True  # 流式下载 (可不设置)
    chunk_size: 1024  # 块大小 (可不设置)

@allure.step('发送 get 请求')
def send_get(self, **kwargs):
    url = kwargs.get('url')
    params = kwargs.get('params')
    data_type = kwargs.get('data_type')
    data = kwargs.get('request_data')
    save_path = kwargs.get('save_path')
    stream = kwargs.get('stream', False)  # 流式下载
    chunk_size = kwargs.get('chunk_size', 1024)  # 块大小
    request_data = {'url': url, 'params': params, 'stream': stream if save_path else False,}
    if data_type == 'json':
        request_data['json'] = data
    if data_type == 'data':
        request_data['data'] = data
    try:
        response = self.request.request('get', **request_data)
        logger.debug('发送请求 success')
        if save_path:
            # 确保目录已存在，exist_ok=True：如果已存在不会抛异常
            os.makedirs(os.path.dirname(save_path), exist_ok=True)
            if stream:  # 大文件保存
                with open(save_path, 'wb') as file:
                    # response.iter_content 以固定大小（chunk_size）分块读取响应内容
                    for chunk in response.iter_content(chunk_size=chunk_size):
                        file.write(chunk)
            else:  # 正常文件保存
                with open(save_path, 'wb') as file:
                    file.write(response.content)  # response.content 获取二进制
        except Exception as e:
            print('请求错误：', e)
        return response

pytest_args = ['-vs', '--capture=sys', '--clean-alluredir', '--alluredir=allure-results', './core/core.py',  # 核心执行器的目录
               '--case=../test_cases/yaml_cases_01',  # 自定义执行 case 目录参数
               '--type=yml'  # 自定义执行用例格式 yml，excel 参数]
# plugins，插件列表，通常传入类实例或模块
# pytest 运行时，会自动扫描所有加载的插件（包括 conftest.py 和通过 plugins 参数传入的插件）
pytest.main(pytest_args, plugins=[CasePlugin()])

import pytest
from requests_test.HAT.parse.yaml_parse import parse_pytest_parameter  # 钩子函数可以放在 conftest 文件，pytest.ini 文件，插件
# 将钩子函数定义在自定义插件中
class CasePlugin:
    # 固定名称 pytest_addoption，pytest 钩子函数，用于向 pytest 的命令行添加自定义参数
    def pytest_addoption(self, parser):
        parser.addoption('--case',  # 参数名
                         action='store',  # 表示需要存储一个值
                         default='',  # 默认值
                         help='用例目录')  # 参数说明
        parser.addoption('--type',  # 参数名
                         action='store',  # 表示需要存储一个值
                         default='',  # 默认值
                         help='用例格式类型，yml 还是 excel')  # 参数说明

    # 固定名称 pytest_generate_tests，钩子函数，用于动态生成测试参数
    # metafunc: pytest.Metafunc 对象，它包含了测试函数的元信息，例如参数化信息、测试函数名称等。
    # metafunc.config 对象来访问 pytest 的命令行选项和配
    # @pytest.mark.parametrize 和 pytest_generate_test 钩子为同一个参数提供值，那么 pytest_generate_tests 会覆盖掉
    # 为不同值赋值，会同时生效，笛卡尔积
    def pytest_generate_tests(self, metafunc):
        case = metafunc.config.getoption('case')  # 获取执行参数
        type = metafunc.config.getoption('type')
        case_infos, case_names = parse_pytest_parameter(case, type)  # 获取指定用例目录下的用例
        if 'case_info' in metafunc.fixturenames:  # 测试函数所有参数名
            # ids 为参数化测试用例重命名
            metafunc.parametrize('case_info', case_infos, ids=case_names)  # 动态生成参数

    # 固定名称 pytest_generate_tests，钩子函数
    # 用于在收集测试用例（还未开始执行）进行修改或操作如如更改名称，跳过某些用例，重新排序
    def pytest_collection_modifyitems(self, items):
        for item in items:
            # 用例名称一般是测试函数名，如果提供 ids 则为 id 名称
            # 解决 Unicode 转义字符的问题，正确显示中文
            item.name = item.name.encode('utf-8').decode('unicode_escape')
            # 节点 id，类似模块路径::测试函数名 的字符串，如 tests/test_example.py::test_example
            item._nodeid = item._nodeid.encode('utf-8').decode('unicode_escape')
            item.add_marker(pytest.mark.slow)  # 添加标记
            if 'skip' in item.name:
                item.add_marker(pytest.mark.skip(reason='skip this'))  # 跳过某条用例

import sys
# 方式一：sys.argv
# 是一个列表，第一个元素为脚本名称，后续元素为字符串参数
# sys.argv 不会对参数进行解析，所有参数都是字符串形式
# python main.py a=1 b=2 args=['main.py','a=1', 'b=2']
args = sys.argv

import argparse
# 方式二：使用 argparse 模块
# argparse 可以定义与解析参数
parser = argparse.ArgumentParser(description='入口执行参数')
# 1、创建解析器
# 2、添加参数，存在-（短格式），--（长格式）是可选参数，否则是必填参数
# 长短格式参数可以同时定义
parser.add_argument('-a', help='参数 a', type=int, default=0)  # 可选参数
parser.add_argument('b', help='参数 b', type=str)  # 必填参数，不能设置默认值
# 长短格式同时定义，执行时任选一种就行
parser.add_argument('-c', '--chlp', help='参数 b', type=str, default='name')
# 3、解析命令行参数，要解析必须定义上述添参数规则
args = parser.parse_args()
# python main1.py -a 1 hhh -c sss
print(args.a)  # 1
print(args.b)  # hhh
print(args.chlp)  # sss，长短格式同时定义时，使用长格式参数获取

from loguru import logger
import sys
from datetime import datetime
import pytest, os, argparse
from HAT.core import core
from HAT.core.case_plugins import CasePlugin
from _pytest.config import ExitCode
import subprocess

# 配置日志
if not os.path.exists('HAT/log'):
    os.mkdir('HAT/log')
time_str = datetime.now().strftime("%Y-%m-%d_%H:%M:%S")
logger.configure(handlers=[
    {"sink": sys.stdout, "level": 'DEBUG'},
    {"sink": sys.stderr, "level": 'DEBUG'},
    {"sink": f'./log/log_{time_str}.log', "level": 'DEBUG'}
])

def args_parser():
    parser = argparse.ArgumentParser(description='接口测试工具')
    # 定义参数规则
    parser.add_argument('--type', type=str, default='yml', help='用例文件类型')
    parser.add_argument('--case', type=str, default='./test_cases/yaml_cases_01', help='指定测试用例的文件夹路径.')
    parser.add_argument('--alluredir', type=str, default=os.path.join(os.getcwd(), "allure_results"), help='运行结果数据文件路径')
    parser.add_argument('--allure_report', type=str, default=os.path.join(os.getcwd(), "allure_report"), help='测试报告保存路径')
    args = parser.parse_args()
    # 根据规则解析
    return args

cmd_args = args_parser()

def run():
    pytest_args = ['-vs', '--capture=sys', '--clean-alluredir']
    if cmd_args.type:
        pytest_args.append(f'--type={cmd_args.type}')
    if cmd_args.case:
        pytest_args.append(f'--case={cmd_args.case}')
    if cmd_args.alluredir:
        pytest_args.append(f'--alluredir={cmd_args.alluredir}')
    pytest_args.append(core.__file__)
    logger.info('pytest 开始执行用例')
    # 返回值：0 都通过，1 存在失败，2 存在 error，5 没找到路径
    exit_code = pytest.main(pytest_args, plugins=[CasePlugin()])
    # OK,0 TESTS_FAILED,1
    if exit_code == ExitCode.OK or exit_code == ExitCode.TESTS_FAILED:
        try:
            # shell 为 True，系统的 shell 执行命令，False 是接调用操作系统的 exec 系列系统调用
            # check 是否以抛出异常方式运行
            # universal_newlines，标准输出会被解码为字符串，否则是字节流
            subprocess.run(f'allure generate {cmd_args.alluredir} -o {cmd_args.allure_report} --clean', shell=True, check=True, universal_newlines=True)
        except subprocess.CalledProcessError as e:
            logger.error(f"测试报告出现问题！{e}")
    else:
        logger.error('pytest 执行异常')

if __name__ == '__main__':
    run()

import setuptools

""" 打包成一个 可执行模块 """
# with open("README.md", "r", encoding="utf-8") as fh:
#     long_description = fh.read()
long_description = '自动化测试工具'  # 详细描述，一般从 README.md 读取

setuptools.setup(
    # 关于项目的介绍 - 随便写都可以
    name="apiPlatform",
    version="1.0.1",
    author="author",
    author_email="[email protected]",
    description="自动化测试工具",
    license="GPLv3",
    long_description=long_description,
    long_description_content_type="text/markdown",
    classifiers=[
        "Programming Language :: Python :: 3",
        "License :: OSI Approved :: GNU General Public License (GPL)",
        "Operating System :: OS Independent",
    ],
    # 需要自动安装的依赖，pin install 报名时，会自动安装这个依赖
    install_requires=[
        "pytest==8.4.2",
        "Jinja2==3.1.6",
        "jsonpath==0.82.2",
    ],
    py_modules=["main"],  # 指定单个文件，setuptools 会在跟目录下自动找到 main.py 并将其作为模块打包
    packages=setuptools.find_packages(),  # 扫描项目结构，自动发现项目包作为安装的一部分
    python_requires=">=3.8",
    entry_points={
        'console_scripts': [
            # tc-api-test 命令名称
            # 调用 main 文件的 run 方法
            'tc-api-test=main:run'
        ]
    },
    zip_safe=False  # 强制以解压形式安装包
)
# 示例执行命令
# tc-api-test --type=yml --case=./test_cases/yaml_cases_01

基于Python+Requests+Allure搭建接口自动化框架

Python 中的*args 与**kwargs

使用 requests 发送请求

发送 get 请求

发送 post 请求

使用 session 发送请求

jsonpath 使用

框架基本封装过程

目录介绍

yaml 用例

基本格式

读取 yaml

封装请求关键字

基本核心执行器

main 执行入口

核心执行器优化

优化报告

增加全局配置

yaml 用例文件改造

全局变量类

增加前置后置脚本执行

多用例执行

用例之间进行数据关联

断言操作

单个字段进行断言

整个响应结果进行断言

数据库断言

yml 数据驱动

allure 与 loguru 日志

基本使用

创建日志管理器关联步骤日志

session 复用机制

文件上传与下载

文件上传

文件下载

pytest 自定义插件

打包

main 文件内容替换

python 获取命令执行参数#

main 文件内容

导出框架依赖包

增加 setup 文件

微信扫一扫，关注极客日志

更多推荐文章

相关免费在线工具