安集商城接口自动化项目架构介绍

一、项目介绍该项目是一个在线购物的商城网站包括用户生命周期管理商品信息查询与展示下单支付等相关功能。PythonpytestsqlalchemyrequestsallurejsonpathyamlJenkinsLinux二、项目架构pythonproject/ ├── run.py # 【运行入口】启动测试、选择报告类型 ├── pytest.ini # pytest配置文件 ├── environment.xml # Allure环境配置 ├── extract.yaml # 【数据共享】接口提取数据存储 ├── conftest.py # 【测试配置】pytest fixtures │ ├── conf/ # 【配置层】 │ ├── setting.py # 全局配置常量文件路径、超时时间等 │ ├── config.ini # 环境配置API地址、数据库连接 │ └── operationConfig.py # 配置文件读取工具读取*.ini文件 │ ├── common/ # 【公共工具层】 │ ├── sendrequest.py # HTTP请求封装 │ ├── readyaml.py # YAML数据读写 │ ├── assertions.py # 断言验证工具 │ ├── recordlog.py # 日志记录 │ ├── debugtalk.py # 自定义函数库加密、时间、获取yaml数据 │ ├── connection.py # 数据库连接 │ └── ... # 其他工具 │ ├── base/ # 【核心业务层】 │ ├── apiutil.py # 接口请求处理核心类 │ ├── apiutil_business.py # 业务场景封装 │ ├── generateId.py # 测试用例ID生成 │ └── removefile.py # 文件清理模块 │ ├── testcase/ # 【测试用例层】 │ ├── conftest.py # 测试用例级fixtures │ │ │ ├── Bussiness interface/ # 业务场景测试用例 │ │ └── BussinessScenario.yml │ │ │ ├── Single interface/ # 用户管理测试用例 │ │ ├── addUser.yaml │ │ ├── updateUser.yaml │ │ ├── deleteUser.yaml │ │ └── queryUser.yaml │ │ │ └── ProductManager/ # 商品管理用例 │ ├── login_dw.yaml │ ├── productDetail.yaml │ ├── orderPay.yaml │ └── commitOrder.yaml │ ├── data/ # 【测试数据层】 │ ├── loginName.yaml # 登录测试数据 │ ├── login_data.csv # CSV测试数据 │ ├── vehicleNo.csv # 车辆编号数据 │ ├── 测试数据.xls # Excel测试数据 │ └── sql/ # SQL脚本 │ ├── logs/ # 【日志输出】 │ └── test.日期.log # 运行日志 │ ├── report/ # 【报告输出】 │ ├── temp/ # Allure原始数据 │ ├── tmreport/ # TM报告 │ ├── allureReport/ # Allure报告 │ └── results.xml # JUnit XML结果 │ └── venv/ # Python虚拟环境三、项目运行第一阶段项目启动1. 入口文件执行运行命令python run.py2. 配置读取(conf/setting.py)配置项值说明REPORT_TYPEallure报告类型allure或tmDIR_BASED:\PythonMall\pythonproject\pythonproject项目根目录API_TIMEOUT60接口超时时间秒FILE_PATH[EXTRACT]extract.yaml路径接口提取数据存储文件第二阶段登录认证session级别执行时机conftest.py中的system_loginfixture自动执行数据流向loginName.yaml → 读取登录测试数据 → 发送登录请求 → 提取token → 存入extract.yaml具体数据示例loginName.yaml 内容- baseInfo: api_name: 用户登录 url: /dar/user/login method: post testCase: - case_name: 用户名和密码正确登录验证 data: user_name: test01 passwd: admin123 extract: token: $.token # 从响应中提取token转换成JSON格式# 返回值是一个 list每个元素对应 yaml 中的一个用例组 case_info [ { baseInfo: # api_info[0][0] → baseInfo 部分 { api_name: 用户登录, url: /dar/user/login, method: post, header: {Content-Type: application/x-www-form-urlencoded;charsetUTF-8} }, testCase: # api_info[0][1] → testCase 中的第一个测试用例 { case_name: 用户名和密码正确登录验证, data: {user_name: test01, passwd: admin123}, validation: [{contains: {error_code: None}}, {eq: {msg: 登录成功}}], extract: {token: $.token} } } ]调用函数replace_load(self, data)def replace_load(self, data): yaml数据替换解析 str_data data if not isinstance(data, str): str_data json.dumps(data, ensure_asciiFalse) # print(从yaml文件获取的原始数据, str_data) for i in range(str_data.count(${)): if ${ in str_data and } in str_data: start_index str_data.index($) end_index str_data.index(}, start_index) ref_all_params str_data[start_index:end_index 1] # 取出yaml文件的函数名 func_name ref_all_params[2:ref_all_params.index(()] # 取出函数里面的参数 func_params ref_all_params[ref_all_params.index(() 1:ref_all_params.index())] # 传入替换的参数获取对应的值,类的反射----getattr,setattr,del.... extract_data getattr(DebugTalk(), func_name)(*func_params.split(,) if func_params else ) if extract_data and isinstance(extract_data, list): extract_data ,.join(e for e in extract_data) str_data str_data.replace(ref_all_params, str(extract_data)) # print(通过解析后替换的数据, str_data) # 还原数据 if data and isinstance(data, dict): data json.loads(str_data) else: data str_data return data实际发送的请求数据登录后 extract.yaml 变化# 登录前文件为空或只有之前的数据 Cookie: access_token_cookie: ... token: 6D2Bf021DEDfe45e9D37EAfDB99c6 # ← 新增登录后提取的token第三阶段测试用例执行执行入口测试类如TestUserManager→ 测试方法如test_add_user数据替换过程addUser.yaml 原始数据- baseInfo: api_name: 新增用户 url: /dar/user/addUser method: POST testCase: - case_name: 正常新增用户 data: username: testadduser token: ${get_extract_data(token)} # ← 引用提取的数据 validation: - contains: { msg: 新增成功 }变量替换过程${get_extract_data(token)} ↓ DebugTalk().get_extract_data(token) ↓ 读取 extract.yaml 获取 token 值 ↓ 替换为6D2Bf021DEDfe45e9D37EAfDB99c6实际发送的请求数据{ username: testadduser, token: 6D2Bf021DEDfe45e9D37EAfDB99c6 # ← 替换后的值 }调用函数specification_yaml(self, case_info)def specification_yaml(self, case_info): 规范yaml测试用例的写法 :param case_info: list类型,调试取case_info[0]--dict :return: params_type [params, data, json] cookie None try: base_url self.conf.get_section_for_data(api_envi, host) # base_url self.replace_load(case_info[baseInfo][url]) url base_url case_info[baseInfo][url] allure.attach(url, f接口地址{url}) api_name case_info[baseInfo][api_name] allure.attach(api_name, f接口名{api_name}) method case_info[baseInfo][method] allure.attach(method, f请求方法{method}) header self.replace_load(case_info[baseInfo][header]) allure.attach(str(header), 请求头信息, allure.attachment_type.TEXT) try: cookie self.replace_load(case_info[baseInfo][cookies]) allure.attach(str(cookie), Cookie, allure.attachment_type.TEXT) except: pass for tc in case_info[testCase]: case_name tc.pop(case_name) allure.attach(case_name, f测试用例名称{case_name}, allure.attachment_type.TEXT) # 断言结果解析替换 val self.replace_load(tc.get(validation)) tc[validation] val # 字符串形式的列表转换为list类型 validation eval(tc.pop(validation)) allure_validation str([str(list(i.values())) for i in validation]) allure.attach(allure_validation, 预期结果, allure.attachment_type.TEXT) extract tc.pop(extract, None) extract_lst tc.pop(extract_list, None) for key, value in tc.items(): if key in params_type: tc[key] self.replace_load(value) file, files tc.pop(files, None), None if file is not None: for fk, fv in file.items(): allure.attach(json.dumps(file), 导入文件) files {fk: open(fv, rb)} res self.run.run_main(nameapi_name, urlurl, case_namecase_name, headerheader, cookiescookie, methodmethod, filefiles, **tc) res_text res.text allure.attach(res_text, 接口响应信息, allure.attachment_type.TEXT) status_code res.status_code allure.attach(self.allure_attach_response(res.json()), 接口响应信息, allure.attachment_type.TEXT) try: res_json json.loads(res_text) if extract is not None: self.extract_data(extract, res_text) if extract_lst is not None: self.extract_data_list(extract_lst, res_text) # 处理断言 assert_res.assert_result(validation, res_json, status_code) except JSONDecodeError as js: logs.error(系统异常或接口未请求) raise js except Exception as e: logs.error(str(traceback.format_exc())) raise e except Exception as e: logs.error(e) raise e步骤解析1、读取配置拼接URLbase_url self.conf.get_section_for_data(api_envi, host) url base_url case_info[baseInfo][url]2、提取接口基础信息并添加到Allure报告api_name case_info[baseInfo][api_name] # api_name 用户登录 allure.attach(api_name, f接口名{api_name}) method case_info[baseInfo][method] # method post allure.attach(method, f请求方法{method}) header self.replace_load(case_info[baseInfo][header]) # header {Content-Type: application/x-www-form-urlencoded;charsetUTF-8} allure.attach(str(header), 请求头信息, allure.attachment_type.TEXT)3、遍历测试用例tc { case_name: 用户名和密码正确登录验证, data: {user_name: test01, passwd: admin123}, extract: {token: $.token} }4、提取用例名称case_name tc.pop(case_name) # case_name 用户名和密码正确登录验证5、处理断言validation eval(tc.pop(validation)) assert_res.assert_result(validation, res_json, status_code)6、提取数据配置extract tc.pop(extract, None) # extract {token: $.token} extract_lst tc.pop(extract_list, None) # extract_lst None7、处理请求参数params_type [params, data, json] for key, value in tc.items(): if key in params_type: tc[key] self.replace_load(value) # tc {data: {user_name: test01, passwd: admin123}} # 因为data中没有${}变量所以replace_load后保持不变8、处理文件上传file, files tc.pop(files, None), None if file is not None: for fk, fv in file.items(): allure.attach(json.dumps(file), 导入文件) files {fk: open(fv, rb)}open(fv, moderb) 以 二进制只读模式 打开文件files {fk: 文件对象} 构建用于 requests 库上传的文件字典9、发送HTTP请求res self.run.run_main( nameapi_name, # 用户登录 urlurl, # http://127.0.0.1:8787/dar/user/login case_namecase_name, # 用户名和密码正确登录验证 headerheader, # {Content-Type: application/x-www-form-urlencoded} cookiescookie, # None methodmethod, # post **tc # {data: {user_name: test01, passwd: admin123}} ) # 实际发送的HTTP请求 # POST http://127.0.0.1:8787/dar/user/login # Content-Type: application/x-www-form-urlencoded # user_nametest01passwdadmin12310、获取响应res_text res.text # res_text {status:200,msg:登录成功,token:6D2Bf021DEDfe45e9D37EAfDB99c6} status_code res.status_code # status_code 20011、解析响应并提取数据res_json json.loads(res_text) # res_json {status:200, msg:登录成功, token:6D2Bf021DEDfe45e9D37EAfDB99c6} # 如果有extract配置提取数据 if extract is not None: self.extract_data(extract, res_text) # extract_data({token: $.token}, {status:200,...})extract_lst tc.pop(extract_list, None)value dictionary.pop(key, default) #安全地移除并返回字典中的值key要移除的键名default如果键不存在时返回的默认值可选第四阶段接口响应与数据提取1. 发送HTTP请求- baseInfo: api_name: 新增用户 url: /dar/user/addUser method: POST header: Content-Type: application/x-www-form-urlencoded;charsetUTF-8 testCase: - case_name: 正常新增用户 data: username: testadduser password: tset6789890 role_id: 123456789 dates: 2023-12-31 phone: 13800000000 token: ${get_extract_data(token)} validation: - contains: { msg: 新增成功 } extract: userId: $.data.userId orderNumber: $.data.orderNumberSendRequest 发送请求POST http://127.0.0.1:8787/dar/user/addUser Content-Type: application/x-www-form-urlencoded ​ usernametestaddusertoken6D2Bf021DEDfe45e9D37EAfDB99c62. 获取响应HTTP/1.1 200 OK { status: 200, msg: 新增成功, data: { userId: 7972512823030970797, orderNumber: 981712696053255936860 } }调用函数从响应中读取数据写入到extract.yamlextract_data(self, testcase_extarct, response)def extract_data(self, testcase_extarct, response): 提取接口的返回值支持正则表达式和json提取器 :param testcase_extarct: testcase文件yaml中的extract值 :param response: 接口的实际返回值 :return: try: pattern_lst [(.*?), (.?), r(\d), r(\d*)] for key, value in testcase_extarct.items(): # 处理正则表达式提取 for pat in pattern_lst: if pat in value: ext_lst re.search(value, response) if pat in [r(\d), r(\d*)]: extract_data {key: int(ext_lst.group(1))} else: extract_data {key: ext_lst.group(1)} self.read.write_yaml_data(extract_data) # 处理json提取参数 if $ in value: ext_json jsonpath.jsonpath(json.loads(response), value)[0] if ext_json: extarct_data {key: ext_json} logs.info(提取接口的返回值, extarct_data) else: extarct_data {key: 未提取到数据请检查接口返回值是否为空} self.read.write_yaml_data(extarct_data) except Exception as e: logs.error(e)步骤解析正则表达式提取模式含义(.*?)匹配任意字符非贪婪(.?)匹配至少一个任意字符非贪婪(\d)匹配一个或多个数字(\d*)匹配零个或多个数字pattern_lst [(.*?), (.?), r(\d), r(\d*)] ... ext_lst re.search(value, response) # ext_lst.group(1) 13800000000 extract_data {key: ext_lst.group(1)} # extract_data {phone: 13800000000}re.search(pattern, string, flags0)参数pattern: 正则表达式模式string: 要搜索的字符串flags: 可选标志如re.IGNORECASE、re.MULTILINE等返回值匹配成功返回 Match 对象匹配失败返回 NoneJSONPath表达式含义$.data.keydata对象下的key$.list[*].key数组中所有元素的keyext_json jsonpath.jsonpath(json.loads(response), value)[0] #[0] 是取第一个匹配的值# 如果找到返回 [实际值]# 如果没找到返回 False 或 []3. 数据提取并存入 extract.yamlextract.yaml 更新# 新增提取的数据 userId: 7972512823030970797 orderNumber: 981712696053255936860第五阶段断言验证Assertions 断言处理addUser.yaml 中的断言配置validation: - contains: { status_code: 200 } # 响应状态码包含200 - contains: { msg: 新增成功 } # 响应消息包含新增成功调用函数contains_assert(self, value, response, status_code)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_typeallure.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_typeallure.attachment_type.TEXT) logs.error(响应文本断言失败预期结果为【%s】,实际结果为【%s】 % (assert_value, resp_list)) return flag# 接口实际响应JSON 对象 response { code: 200, msg: 操作成功, data: { id: 123, name: test } } # 来自 YAML 测试用例的 validation 配置 value { status_code: 200, msg: 操作成功, data: xxx }1、状态码断言for assert_key, assert_value in value.items(): if assert_key status_code: if assert_value ! status_code: flag 1 # 记录断言失败信息到 Allure 报告2、响应体字段提取JSONPathresp_list jsonpath.jsonpath(response, $..%s % assert_key)从响应数据中递归查找所有匹配指定键的值3、字符串拼接if isinstance(resp_list[0], str): resp_list .join(resp_list) # [操作, 成功] → 操作成功4、包含断言判断assert_value None if assert_value.upper() NONE else assert_value if assert_value in resp_list: # 断言成功 else: # 断言失败flag 1断言过程# 伪代码演示 response {status: 200, msg: 新增成功} validation [ {contains: {status_code: 200}}, {contains: {msg: 新增成功}} ] ​ # 验证1response中是否包含status_code: 200 assert status_code in response and response[status_code] 200 # ✅ 通过 ​ # 验证2response中是否包含msg: 新增成功 assert msg in response and 新增成功 in response[msg] # ✅ 通过第六阶段日志记录logs/ 目录下的日志文件logs/test.20260514.log日志内容示例2026-05-14 10:30:15 - INFO - -------------接口测试开始-------------- 2026-05-14 10:30:15 - INFO - 用户登录接口: 开始发送请求... 2026-05-14 10:30:16 - INFO - 用户登录接口: 请求成功响应状态码200 2026-05-14 10:30:16 - INFO - 提取接口的返回值{token: 6D2Bf021DEDfe45e9D37EAfDB99c6} 2026-05-14 10:30:20 - INFO - 新增用户接口: 开始发送请求... 2026-05-14 10:30:21 - INFO - 新增用户接口: 请求成功响应状态码200 2026-05-14 10:30:21 - INFO - 提取接口的返回值{userId: 7972512823030970797} 2026-05-14 10:30:21 - INFO - -------------接口测试结束--------------核心函数pytest.fixture(autouseTrue) def start_test_and_end(): logs.info(-------------接口测试开始--------------) yield logs.info(-------------接口测试结束--------------)pytest.fixture(autouseTrue)在每个测试函数执行前后自动运行无需显式调用def get_testcase_yaml(file): testcase_list [] try: with open(file, r, encodingutf-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)})读取和解析 YAML 测试用例文件 将 YAML 数据转换为测试框架可执行的参数格式。- baseInfo: api_name: 用户登录 url: /dar/user/login method: post testCase: - case_name: 正常登录 data: {user_name: test01, passwd: admin123} - case_name: 密码错误 data: {user_name: test01, passwd: wrong}转换为[ [{api_name: 用户登录, url: /dar/user/login, method: post}, {case_name: 正常登录, data: {user_name: test01, passwd: admin123}}], [{api_name: 用户登录, url: /dar/user/login, method: post}, {case_name: 密码错误, data: {user_name: test01, passwd: wrong}}] ]调用顺序1. conftest.py:start_test_and_end() ↓ logs.info(-------------接口测试开始--------------) ↓ 2. apiutil.py:specification_yaml() ↓ 3. sendrequest.py:run_main() ↓ logs.info(接口名称用户登录) logs.info(请求地址/dar/user/login) logs.info(请求方式POST) ↓ 4. apiutil.py:extract_data() ↓ logs.info(提取接口的返回值{token: xxx}) ↓ 5. conftest.py:start_test_and_end() ↓ logs.info(-------------接口测试结束--------------)第七阶段测试报告生成根据 REPORT_TYPE 配置生成不同报告模式1allure 报告REPORT_TYPE allure执行命令pytest.main([ -s, -v, --alluredir./report/temp, # allure原始数据目录 ./testcase, # 测试用例目录 --clean-alluredir, # 清理旧报告 --junitxml./report/results.xml # JUnit XML格式结果 ]) ​ # 复制环境信息 shutil.copy(./environment.xml, ./report/temp) ​ # 启动allure服务 os.system(allure serve ./report/temp)生成的报告结构report/ ├── temp/ # allure原始数据 │ ├── environment.xml │ ├── suites/ # 测试套件数据 │ ├── widgets/ # 报告组件数据 │ └── index.html # 报告入口 └── results.xml # JUnit XML结果模式2tm 报告REPORT_TYPE tm执行命令pytest.main([ -vs, --pytest-tmreport-nametestReport.html, # 报告文件名 --pytest-tmreport-path./report/tmreport # 报告路径 ]) ​ # 自动打开浏览器 webbrowser.open_new_tab(os.getcwd() /report/tmreport/testReport.html)生成的报告结构report/ └── tmreport/ └── testReport.html # HTML测试报告关键数据变化汇总阶段数据来源处理过程输出/结果登录loginName.yaml发送POST请求提取token到extract.yaml用例执行addUser.yaml等${get_extract_data(token)}替换发送带token的请求响应处理HTTP响应JSON解析提取userId、orderNumber等数据存储响应数据write_yaml_data()更新extract.yaml断言验证响应validation字符串/数值比较通过/失败日志记录全过程logs.info/error()logs/test.日期.log报告生成测试结果pytest插件HTML报告四、项目总结基于 PythonpytestAllure 构建的电商平台接口测试解决方案通过 YAML 数据驱动实现测试用例与代码解耦支持接口关联JSONPath/正则提取、灵活断言包含/相等验证和可视化报告生成覆盖用户登录、商品管理、订单支付等核心业务流程。