50 lines
1.9 KiB
Python
50 lines
1.9 KiB
Python
# -*- coding: utf-8 -*-
|
||
# @Time : 2023/11/8 16:53
|
||
# @Author : floraachy
|
||
# @File : api_workflow.py
|
||
# @Software: PyCharm
|
||
# @Desc:
|
||
|
||
|
||
# 标准库导入
|
||
# 第三方库导入
|
||
from loguru import logger
|
||
# 本地应用/模块导入
|
||
from utils.files_utils.yaml_handle import YamlHandle
|
||
from utils.requests_utils.request_control import RequestPreDataHandle, RequestHandle, after_request_extract
|
||
from utils.requests_utils.assert_handle import assert_response
|
||
|
||
|
||
class ApiWorkFlow:
|
||
def __init__(self, api_file_path: str):
|
||
"""
|
||
:param:api_file_path 接口yaml文件路径
|
||
:param:key 对应接口的id
|
||
:param:global_var 全局变量，保存接口相关变量的实际值， 例如接口中的${login}，会从GLOBAL_VAR中找到key=login的元素进行替换
|
||
"""
|
||
self.api_file_path = api_file_path
|
||
|
||
def get_api_mode(self, key):
|
||
"""
|
||
根据指定的yaml文件路径，以及key值，获取对应的接口
|
||
:param:key 对应接口的id
|
||
"""
|
||
api_modes = YamlHandle(filename=self.api_file_path).read_yaml
|
||
matching_api = next((item for item in api_modes["case_info"] if item["id"] == key), None)
|
||
if matching_api:
|
||
return matching_api
|
||
else:
|
||
logger.error(f"根据{key}, 未能找到对应的接口")
|
||
raise KeyError(f"根据{key}, 未能找到对应的接口")
|
||
|
||
def api_work_flow(self, key: str, global_var: dict = {}) -> dict:
|
||
api_mode = self.get_api_mode(key)
|
||
api_info = RequestPreDataHandle(request_data=api_mode, global_var=global_var).request_data_handle()
|
||
# 发送请求
|
||
response = RequestHandle(api_info, global_var).http_request()
|
||
# 进行响应断言
|
||
assert_response(response, api_info["assert_response"])
|
||
# 断言成功后进行参数提取
|
||
res = after_request_extract(response, api_info["extract"])
|
||
return res
|