一、项目介绍

技术栈(自动化)：Python+pytest+sqlalchemy+requests+allure+jsonpath+yaml+Jenkins+Linux**

简介：该项目是一个在线购物的商城网站，包括用户注册，登录，下单，上架/下架商品，下单支付等相关功能,主要针对该网站搭建了接口自动化框架,封装了相关请求,实现接口关联,结果断言,数据库断言等脚本；

功能覆盖

• 登录
• 用户管理（增删改查）——功能校验
• 商品管理 ——回归测试
• 业务集成 ——端到端场景

整体架构流程

采用 数据驱动模型 (Data-Driven)）思想，将业务逻辑（YAML）与底层驱动（Python）解耦。

pythonproject/
├── base/                       # 【基础类封装层】核心功能
│   ├── apiutil.py              # 核心执行器：解析YAML、处理变量替换、执行请求
│   ├── apiutil_business.py     # 业务场景执行器：专门处理多接口关联的复杂业务流
│   ├── generateId.py           # ID生成器：利用yield生成Allure报告所需的有序ID
│   └── removefile.py           # 文件管理工具：用于清理旧的日志或临时报告文件
├── common/                     # 【公共方法层】通用工具插件库
│   ├── assertions.py           # 断言引擎：支持JsonPath、正则表达式、数据库校验
│   ├── connection.py           # 数据库连接池：封装MySQL/MongoDB等操作
│   ├── debugtalk.py            # 热加载函数库：存放YAML中通过${}调用的自定义函数
│   ├── readyaml.py             # YAML处理器：负责extract.yaml的读写及用例加载
│   ├── recordlog.py            # 日志模块：基于logging封装，支持控制台与文件同步输出
│   ├── sendrequest.py          # 网络请求封装：基于requests.Session，处理Cookie持久化
│   ├── dingRobot.py            # 钉钉机器人：用于推送测试结果通知
│   └── semail.py               # 邮件通知：集成SMTP，支持发送HTML报告及附件
├── conf/                       # 【配置层】全局静态配置
│   ├── config.ini              # 核心配置文件：存放Host、数据库、邮箱等私密信息
│   ├── operationConfig.py      # 配置解析器：封装ConfigParser，提供简单的配置读取接口
│   └── setting.py              # 路径常量：定义项目根目录、日志/报告的绝对路径
├── data/                       # 【测试数据层】存放非YAML格式的外部数据
│   ├── sql/                    # 存放复杂的SQL模板文件 (.xml)
│   ├── login_data.csv          # CSV格式的参数化数据
│   └── 测试数据.xls             # Excel格式的原始数据
├── testcase/                   # 【测试用例层】
│   ├── ProductManager/         # 业务模块：商品管理相关的测试集 (YAML+Python)
│   ├── Single interface/       # 接口维度：单接口的原子测试用例
│   ├── Business interface/     # 场景维度：多接口串联的集成测试用例
│   └── conftest.py             # Pytest钩子：定义全局Fixture（如登录、初始化）
├── report/                     # 【报告层】
│   ├── allureReport/           # 最终生成的Allure HTML可视化报告
│   └── temp/                   # Allure运行时的原始JSON结果（临时存放）
├── logs/                       # 【日志层】存放运行过程中的.log文件
├── venv/                       # 本地开发使用的Python虚拟环境
├── environment.xml             # Allure报告环境信息配置（显示在报告首页）
├── extract.yaml                # 【核心关联文件】运行时自动存放接口间依赖的变量
├── pytest.ini                  # Pytest配置文件：定义运行参数、命名规则及标签
├── requirements.txt            # 项目依赖库清单
├── run.py                      # 【项目入口】一键执行测试、生成并打开报告
└── 使用前请阅读此文件.md          # 项目说明文档

三、项目流程

3.1 程序入口run.py

import shutil
import pytest
import os
import webbrowser
from conf.setting import REPORT_TYPE

if __name__ == '__main__':

    if REPORT_TYPE == 'allure':
        pytest.main(
            ['-s', '-v', '--alluredir=./report/temp', './testcase', '--clean-alluredir',
             '--junitxml=./report/results.xml'])

        shutil.copy('./environment.xml', './report/temp')
        os.system(f'allure serve ./report/temp')

    elif REPORT_TYPE == 'tm':
        pytest.main(['-vs', '--pytest-tmreport-name=testReport.html', '--pytest-tmreport-path=./report/tmreport'])
        webbrowser.open_new_tab(os.getcwd() + '/report/tmreport/testReport.html')

代码说明：

主程序入口

if __name__ == '__main__':

• 当脚本作为主程序运行时才执行以下代码块

if REPORT_TYPE == 'allure':
        pytest.main(
            ['-s', '-v', '--alluredir=./report/temp', './testcase', '--clean-alluredir',
             '--junitxml=./report/results.xml'])

• 判断当前要生成的报告是allure格式

-s：显示用例里的 print() 打印内容
-v：输出详细日志，方便看哪个用例成功 / 失败
–alluredir=./report/temp： 把 allure 报告需要的 json 临时文件存到 /report/temp
./testcase ：指定运行 testcase 文件夹下的所有测试用例
–clean-alluredir ：运行前先清空旧的临时报告文件，避免数据错乱
–junitxml=./report/results.xml ：生成 JUnit 格式的报告（ Jenkins 等平台可用）

shutil.copy('./environment.xml', './report/temp')

• 复制环境配置文件environment.xml到报告目录。复制过去后，allure 网页报告首页才会显示环境信息

os.system(f'allure serve ./report/temp')

• 启动 allure 服务，自动打开浏览器展示报告。
• 报告里能看到：用例数、成功 / 失败率、请求日志、响应数据、错误截图等

3.2 登录流程

3.2.1 启动与触发阶段

登录动作被封装在 conftest.py 的 system_login 固件（Fixture）中

@pytest.fixture(scope='session', autouse=True)
@allure.story("登录")
def system_login():
    try:
        api_info = get_testcase_yaml('./data/loginName.yaml')
        RequestBase().specification_yaml(api_info[0][0], api_info[0][1])
    except Exception as e:
        logs.error(f'登录接口出现异常，导致后续接口无法继续运行，请检查程序！，{e}')
        exit()

• 执行等级scope=‘session’ （整个测试过程只运行一次）， autouse=True （自动执行）
• 当运行 run.py 时，Pytest 引擎启动后的第一件事就是加载 system_login 。

3.2.2 数据加载阶段

数据源 ：从 loginName.yaml 读取登录信息

 - baseInfo:
    api_name: 用户登录
    url: /dar/user/login
    method: post
    header:
      Content-Type: application/x-www-form-urlencoded;charset=UTF-8
  testCase:
    - case_name: 用户名和密码正确登录验证
      data:
        user_name: test01
        passwd: admin123
      validation:
        - contains: { 'error_code': none }
        - eq: { 'msg': '登录成功' }
      extract:
        token: $.token

• 调用\common\readyaml.py 中get_testcase_yaml() 方法将 YAML 中的 URL、请求头、账号密码等结构化数据解析为 Python 字典。

def get_testcase_yaml(file):
    testcase_list = []
    try:
        with open(file, 'r', encoding='utf-8') as f:
            data = yaml.safe_load(f)
            if len(data) <= 1:
                yam_data = data[0]
                base_info = yam_data.get('baseInfo')
                for ts in yam_data.get('testCase'):
                    param = [base_info, ts]
                    testcase_list.append(param)
                return testcase_list
            else:
                return data
    except UnicodeDecodeError:
        logs.error(f"[{file}]文件编码格式错误，--尝试使用utf-8编码解码YAML文件时发生了错误，请确保你的yaml文件是UTF-8格式！")
    except FileNotFoundError:
        logs.error(f'[{file}]文件未找到，请检查路径是否正确')
    except Exception as e:
        logs.error(f'获取【{file}】文件数据时出现未知错误: {str(e)}')

3.2.3 请求构建与发送

• apiutil.py 的 specification_yaml 方法介入

def specification_yaml(self, base_info, test_case):
        """
        接口请求处理基本方法
        :param base_info: list类型，读取yaml文件里面的baseInfo
        :param test_case: yaml文件里面的testCase
        :return:
        """
        try:
            params_type = ['data', 'json', 'params']
            url_host = self.conf.get_section_for_data('api_envi', 'host')
            api_name = base_info['api_name']
            allure.attach(api_name, f'接口名称：{api_name}', allure.attachment_type.TEXT)
            url = url_host + base_info['url']
            allure.attach(api_name, f'接口地址：{url}', allure.attachment_type.TEXT)
            method = base_info['method']
            allure.attach(api_name, f'请求方法：{method}', allure.attachment_type.TEXT)
            header = self.replace_load(base_info['header'])
            allure.attach(api_name, f'请求头：{header}', allure.attachment_type.TEXT)
            # 处理cookie
            cookie = None
            if base_info.get('cookies') is not None:
                cookie = eval(self.replace_load(base_info['cookies']))
            case_name = test_case.pop('case_name')
            allure.attach(api_name, f'测试用例名称：{case_name}', allure.attachment_type.TEXT)
            # 处理断言
            val = self.replace_load(test_case.get('validation'))
            test_case['validation'] = val
            validation = eval(test_case.pop('validation'))
            # 处理参数提取
            extract = test_case.pop('extract', None)
            extract_list = test_case.pop('extract_list', None)
            # 处理接口的请求参数
            for key, value in test_case.items():
                if key in params_type:
                    test_case[key] = self.replace_load(value)

            # 处理文件上传接口
            file, files = test_case.pop('files', None), None
            if file is not None:
                for fk, fv in file.items():
                    allure.attach(json.dumps(file), '导入文件')
                    files = {fk: open(fv, mode='rb')}

            res = self.run.run_main(name=api_name, url=url, case_name=case_name, header=header, method=method,
                                    file=files, cookies=cookie, **test_case)
            status_code = res.status_code
            allure.attach(self.allure_attach_response(res.json()), '接口响应信息', allure.attachment_type.TEXT)

            try:
                res_json = json.loads(res.text)  # 把json格式转换成字典字典
                if extract is not None:
                    self.extract_data(extract, res.text)
                if extract_list is not None:
                    self.extract_data_list(extract_list, res.text)
                # 处理断言
                self.asserts.assert_result(validation, res_json, status_code)
            except JSONDecodeError as js:
                logs.error('系统异常或接口未请求！')
                raise js
            except Exception as e:
                logs.error(e)
                raise e

        except Exception as e:
            raise e

• 网络层：由 sendrequest.py 负责发送。它使用 requests.Session() ， Session 会自动保存服务器返回的 Set-Cookie 。

3.2.4 令牌提取与持久化

实现接口关联的核心步骤

• 提取 ：在登录接口的 YAML 配置中，定义了 extract: token: $.token 。
  extract: token: $.token

• 捕获 ： apiutil.py 执行完登录请求后，使用 JSONPath 从响应体中提取出 token 的真实值。

• 写入 ：调用 readyaml.py 将提取到的 token 写入到根目录的 extract.yaml 文件中。

3.2.5 校验与结果反馈

• 断言 ： assertions.py 会校验响应结果是否包含“登录成功”字样，且状态码是否符合预期

def contains_assert(self, value, response, status_code):
        """
        字符串包含断言模式，断言预期结果的字符串是否包含在接口的响应信息中
        :param value: 预期结果，yaml文件的预期结果值
        :param response: 接口实际响应结果
        :param status_code: 响应状态码
        :return: 返回结果的状态标识
        """
        # 断言状态标识，0成功，其他失败
        flag = 0
        for assert_key, assert_value in value.items():
            if assert_key == "status_code":
                if assert_value != status_code:
                    flag += 1
                    allure.attach(f"预期结果：{assert_value}\n实际结果：{status_code}", '响应代码断言结果:失败',
                                  attachment_type=allure.attachment_type.TEXT)
                    logs.error("contains断言失败：接口返回码【%s】不等于【%s】" % (status_code, assert_value))
            else:
                resp_list = jsonpath.jsonpath(response, "$..%s" % assert_key)
                if isinstance(resp_list[0], str):
                    resp_list = ''.join(resp_list)
                if resp_list:
                    assert_value = None if assert_value.upper() == 'NONE' else assert_value
                    if assert_value in resp_list:
                        logs.info("字符串包含断言成功：预期结果【%s】,实际结果【%s】" % (assert_value, resp_list))
                    else:
                        flag = flag + 1
                        allure.attach(f"预期结果：{assert_value}\n实际结果：{resp_list}", '响应文本断言结果：失败',
                                      attachment_type=allure.attachment_type.TEXT)
                        logs.error("响应文本断言失败：预期结果为【%s】,实际结果为【%s】" % (assert_value, resp_list))
        return flag

• 日志 ： recordlog.py 记录登录的详细入参和回参。

def output_logging(self):
        """获取logger对象"""
        logger = logging.getLogger(__name__)
        # 防止重复打印日志
        if not logger.handlers:
            logger.setLevel(setting.LOG_LEVEL)
            log_format = logging.Formatter(
                '%(levelname)s - %(asctime)s - %(filename)s:%(lineno)d -[%(module)s:%(funcName)s] - %(message)s')
            # 日志输出到指定文件，滚动备份日志
            fh = RotatingFileHandler(filename=logfile_name, mode='a', maxBytes=5242880,
                                     backupCount=7,
                                     encoding='utf-8')  # maxBytes:控制单个日志文件的大小，单位是字节,backupCount:用于控制日志文件的数量

            fh.setLevel(setting.LOG_LEVEL)
            fh.setFormatter(log_format)
            # 将相应的handler添加在logger对象中
            logger.addHandler(fh)

            # 输出到控制台
            sh = logging.StreamHandler()
            sh.setLevel(setting.STREAM_LOG_LEVEL)
            sh.setFormatter(log_format)
            logger.addHandler(sh)
        return logger

• 容错处理 ：如果登录失败， conftest.py 会直接执行 exit() 停止整个测试，防止因为没有权限导致后续成百上千个用例无效运行。

except Exception as e:
        logs.error(f'登录接口出现异常，导致后续接口无法继续运行，请检查程序！，{e}')
        exit()

3.3 测试用例执行流程（以用户管理模块为例）

3.3.1启动阶段

执行 pytest.main() ，Pytest 开始扫描 testcase/ 目录。

3.3.2前置拦截与鉴权

1. 自动登录 ： system_login 固件设置了 autouse=True 且 scope=‘session’ ，它会首先执行。
2. 数据流转 ：登录接口成功后，通过 RequestBase 将返回的 token 写入到 extract.yaml 中。
3. 日志开启 ： recordlog.py 初始化本次运行的日志文件。

3.3.3用例加载与解析

Pytest 进入了用户管理模块的脚本

import allure
import pytest

from common.readyaml import get_testcase_yaml
from base.apiutil import RequestBase
from base.generateId import m_id, c_id


@allure.feature(next(m_id) + '用户管理模块（单接口）')
class TestUserManager:

    # 场景，allure报告的目录结构
    @allure.story(next(c_id) + "新增用户")
    # 测试用例执行顺序设置
    @pytest.mark.run(order=1)
    # 参数化，yaml数据驱动
    @pytest.mark.parametrize('base_info,testcase', get_testcase_yaml("./testcase/Single interface/addUser.yaml"))
    def test_add_user(self, base_info, testcase):
        allure.dynamic.title(testcase['case_name'])
        RequestBase().specification_yaml(base_info, testcase)

    @allure.story(next(c_id) + "修改用户")
    @pytest.mark.run(order=2)
    @pytest.mark.parametrize('base_info,testcase', get_testcase_yaml("./testcase/Single interface/updateUser.yaml"))
    def test_update_user(self, base_info, testcase):
        allure.dynamic.title(testcase['case_name'])
        RequestBase().specification_yaml(base_info, testcase)

    @allure.story(next(c_id) + "删除用户")
    @pytest.mark.run(order=3)
    @pytest.mark.parametrize('base_info,testcase', get_testcase_yaml("./testcase/Single interface/deleteUser.yaml"))
    def test_delete_user(self, base_info, testcase):
        allure.dynamic.title(testcase['case_name'])
        RequestBase().specification_yaml(base_info, testcase)

    @allure.story(next(c_id) + "查询用户")
    @pytest.mark.run(order=4)
    @pytest.mark.parametrize('base_info,testcase', get_testcase_yaml("./testcase/Single interface/queryUser.yaml"))
    def test_query_user(self, base_info, testcase):
        allure.dynamic.title(testcase['case_name'])
        RequestBase().specification_yaml(base_info, testcase)

def get_testcase_yaml(file):
    testcase_list = []
    try:
        with open(file, 'r', encoding='utf-8') as f:
            data = yaml.safe_load(f)
            if len(data) <= 1:
                yam_data = data[0]
                base_info = yam_data.get('baseInfo')
                for ts in yam_data.get('testCase'):
                    param = [base_info, ts]
                    testcase_list.append(param)
                return testcase_list
            else:
                return data
    except UnicodeDecodeError:
        logs.error(f"[{file}]文件编码格式错误，--尝试使用utf-8编码解码YAML文件时发生了错误，请确保你的yaml文件是UTF-8格式！")
    except FileNotFoundError:
        logs.error(f'[{file}]文件未找到，请检查路径是否正确')
    except Exception as e:
        logs.error(f'获取【{file}】文件数据时出现未知错误: {str(e)}')

1. 参数化注入 ： @pytest.mark.parametrize 调用 readyaml.py 读取 addUser.yaml 。
2. 数据解析 ： get_testcase_yaml() 将 YAML 文件中的数据解析为 Python 列表。Pytest 根据数据的组数，动态生成相应数量的测试用例。

3.3.4核心执行循环

对于每一个生成的测试用例，都会经历以下循环：

1. 热加载替换 ：调用 apiutil.py::replace_load() 。它扫描请求参数，发现 ${get_extract_data(token)} ，于是去 extract.yaml 翻找刚才登录存好的 Token。
2. 发送请求 ：通过common::sendrequest.py 的 run_main() 方法发送真正的 HTTP 请求。

def run_main(self, name, url, case_name, header, method, cookies=None, file=None, **kwargs):
        """
        接口请求
        :param name: 接口名
        :param url: 接口地址
        :param case_name: 测试用例
        :param header:请求头
        :param method:请求方法
        :param cookies：默认为空
        :param file: 上传文件接口
        :param kwargs: 请求参数，根据yaml文件的参数类型
        :return:
        """

        try:
            # 收集报告日志
            logs.info('接口名称：%s' % name)
            logs.info('请求地址：%s' % url)
            logs.info('请求方式：%s' % method)
            logs.info('测试用例名称：%s' % case_name)
            logs.info('请求头：%s' % header)
            logs.info('Cookie：%s' % cookies)
            req_params = json.dumps(kwargs, ensure_ascii=False)
            if "data" in kwargs.keys():
                allure.attach(req_params, '请求参数', allure.attachment_type.TEXT)
                logs.info("请求参数：%s" % kwargs)
            elif "json" in kwargs.keys():
                allure.attach(req_params, '请求参数', allure.attachment_type.TEXT)
                logs.info("请求参数：%s" % kwargs)
            elif "params" in kwargs.keys():
                allure.attach(req_params, '请求参数', allure.attachment_type.TEXT)
                logs.info("请求参数：%s" % kwargs)
        except Exception as e:
            logs.error(e)
        # time.sleep(0.5)
        requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
        response = self.send_request(method=method,
                                     url=url,
                                     headers=header,
                                     cookies=cookies,
                                     files=file,
                                     timeout=setting.API_TIMEOUT,
                                     verify=False,
                                     **kwargs)
        return response

3. 断言校验 ：
   • assertions.py 介入，比对响应状态码、返回文本。
   • 如果配置了 db 断言，会通过 connection.py 去数据库里查数据。
4. 关联提取 ：如果 YAML 里有 extract 字段，再次将新产生的数据（如 user_id ）写回 extract.yaml 。

3.3.5生成Allure报告与反馈

1. 每个用例跑完，Pytest 都会在 report/temp 下生成一个 JSON 结果文件。
2. 报告转化 ：所有用例跑完， run.py 恢复执行，调用 os.system(‘allure serve …’) 。
3. 浏览器自动弹出，展示包含测试步骤、请求详情、响应结果、日志以及环境信息的 Allure 报告。

3.4 Allure报告

最终浏览器打开的报告页面如图（默认英文）

• 总览 (Overview) ：展示本次测试的成功率、用例总数、环境信息及执行趋势。
• 类别 (Categories) ：对失败原因进行分类（如：代码缺陷、环境不稳定）。
• 测试套件 (Suites) ：按代码文件的组织结构展示用例。
• 功能 (Behaviors) ： 按 @allure.feature 和 @allure.story 的业务维度展示用例。你可以清晰地看到整个“购物流程”的每一步执行情况。
• 图表 (Graphs) ：通过柱状图和圆饼图展示测试状态分布、优先级分布及耗时情况。

四.项目功能

4.1 用户管理模块 (Single interface) ——原子级功能校验

该模块主要负责系统基础数据的 CRUD（增删改查） 稳定性测试。

• 覆盖了管理员对用户的全生命周期管理，包括：
  • 新增/注册 ：验证新用户入库逻辑。
  • 修改/更新 ：验证用户信息变更的准确性。
  • 查询/检索 ：验证数据读取的实时性与准确性。
  • 删除/注销 ：验证数据清理逻辑。

4.2 商品管理模块 (ProductManager) —— 模块级回归测试

该模块专注于电商系统最核心的业务组件，确保商品流转与交易基础功能。

• 模拟后台及前台对商品的操作：
  • 商品列表 ：获取海量商品数据，验证分页、过滤等逻辑。
  • 商品详情 ：深度校验 SKU 规格、价格、库存状态。
  • 交易预操作 ：涵盖提交订单、模拟支付等独立动作。
• 特点 ：
  • 顺序控制 ：利用 @pytest.mark.run(order=n) 保证测试逻辑符合业务先后顺序（如：先查到商品 ID，再进行下单）。
  • 会话保持 ：利用 requests.Session() 自动维护登录态。

4.3 业务集成模块 (Business interface) —— 端到端场景建模

该模块模拟真实用户在电商平台上的完整行为链条。

• 实现了一个完整的 闭环业务流 ：
  • 全链路串联 ： 登录 -> 浏览商品 -> 查看详情 -> 确认规格 -> 提交订单 -> 完成支付 -> 校验订单状态 。
• 功能特点 ：
  • 通过 extract.yaml 实现了数据的传递。例如：支付接口引用的订单号，是由前一秒运行的下单接口实时生成的。
  • 冒烟/回归首选 ：该模块常用于 CI/CD 流程中的冒烟测试，只要这条线通了，就说明系统的核心交易功能是正常的。

五.项目总结

所有的测试用例流程思想是一样的（读YAML -> 换变量 -> 发请求 -> 验结果 -> 存关联）

1. 统一的驱动方式 (Pytest + YAML）

• 使用 @pytest.mark.parametrize 加载 YAML。
• 调用核心引擎方法 RequestBase().specification_yaml() 。

2. 统一的核心引擎 ( apiutil.py )
所有的测试用例最终都会汇聚到 specification_yaml 这个方法里。

• 所有的 ${变量替换} 都在这里发生。
• 所有的 extract 数据提取 都在这里执行。
• 所有的 allure.attach 报告记录 都在这里生成。

3.统一的生命周期 ( conftest.py )

• 所有的用例在运行前，都会检查并执行 system_login
• 所有的请求都会通过 recordlog.py 记录