【python自动化】七月PytestAutoApi开源框架学习笔记（二）

Python
215
0
0
2024-01-27
【python自动化】playwright长截图&切换标签页&JS注入实战
上一篇
【python自动化】七月PytestAutoApi开源框架学习笔记（一）
下一篇

执行流程

「注：请先阅读作者的README.md文档https://gitee.com/yu_xiao_qi/pytest-auto-api2/blob/master/README.md」

本文很长：建议对PytestAutoApi开源框架确实有在学习的小伙伴阅读。

框架主要是基于 Python + pytest + allure + log + yaml + mysql + redis + 钉钉通知 + Jenkins 实现的接口自动化框架。

此文成文于2023.04.13。

目录结构

├── common                         // 配置
│   ├── conf.yaml                  // 公共配置
│   ├── setting.py                 // 环境路径存放区域
├── data                           // 测试用例数据
├── File                           // 上传文件接口所需的文件存放区域
├── logs                           // 日志层
├── report                         // 测试报告层
├── test_case                      // 测试用例代码
├── utils                          // 工具类
│   └── assertion                
│       └── assert_control.py      // 断言
│       └── assert_type.py         // 断言类型
│   └── cache_process              // 缓存处理模块
│       └── cacheControl.py
│       └── redisControl.py  
│   └── logging_tool                   // 日志处理模块
│       └── log_control.py
│       └── log_decorator.py       // 日志装饰器
│       └── run_time_decorator.py   // 统计用例执行时长装饰器
│   └── mysql_tool                 // 数据库模块      
│       └── mysql_control.py   
│   └── notify                 // 通知模块
│       └── ding_talk.py     // 钉钉通知 
│       └── lark.py       // 飞书通知
│       └── send_mail.py     // 邮箱通知
│       └── wechat_send.py   // 企业微信通知
│   └── other_tools                 // 其他工具类
│       └── allure_data             // allure封装
│           └── allure_report_data.py // allure报告数据清洗
│           └── allure_tools.py   // allure 方法封装
│           └── error_case_excel.py   // 收集allure异常用例，生成excel测试报告
│       └── install_tool            
│           └── install_requirements.py
│           └── version_library_comparisons.txt
│       └── get_local_ip.py      // 获取本地IP
|		└── address_detection.py
|		└── exceptions.py
|		└── jsonpath_date_replace.py
|		└── models.py
|		└── thread_tool.py
│   └── read_files_tools             // 文件操作
│       └── case_automatic_control.py // 自动生成测试代码 
│       └── clean_files.py          // 清理文件
│       └── excel_control.py         // 读写excel
│       └── get_all_files_path.py   // 获取所有文件路径
│       └── get_yaml_data_analysis.py // yaml用例数据清洗
│       └── regular_control.py        // 正则
│       └── swagger_for_yaml.py         
│       └── yaml_control.py          
│       └── testcase_template.py          // yaml文件读写
│   └── recording             // 代理录制
│       └── mitmproxy_control.py
│   └── requests_tool 
│       └── dependentCase.py        // 数据依赖处理
│       └── encryption_algorithm_control.py      
│       └── request_control.py      // 请求封装
│       └── set_current_request_cache.py      
│       └── teardown_control.py      
│   └── times_tool
│       └── time_control.py      
├── Readme.md                       // help
├── pytest.ini                  
├── run.py                           // 运行入口

参数配置

入口文件-run.py

1、通过pytest收集相关用例

2、通过allure_data收集测试数据

3、将数据发送到对应的钉钉、飞书、微信、邮件

4、如果程序报错则抛出异常，并发送异常邮件

pytest.ini

指定pytest收集用例和执行用例的规则

test_case初始化数据缓存

在test_case文件夹下的__init__.py文件中，进行了case解析，将yaml文件解析到缓存文件中。

此段代码由于放在__init__.py文件中，所以当pytest收集用例并执行的第一时间，会先执行该文件，也就是先把所有的用例文件内容参数加载到缓存中，关于缓存知识点可以参考上节内容。

from common.setting import ensure_path_sep
from utils.read_files_tools.get_yaml_data_analysis import CaseData
from utils.read_files_tools.get_all_files_path import get_all_files
from utils.cache_process.cache_control import CacheHandler, _cache_config


def write_case_process():
    """
    获取所有用例，写入用例池中
    :return:
    """

    # 循环拿到所有存放用例的文件路径
    for i in get_all_files(file_path=ensure_path_sep("\\data"), yaml_data_switch=True):
        # 循环读取文件中的数据
        case_process = CaseData(i).case_process(case_id_switch=True)
        if case_process is not None:
            # 转换数据类型
            for case in case_process:
                for k, v in case.items():
                    # 判断 case_id 是否已存在
                    case_id_exit = k in _cache_config.keys()
                    # 如果case_id 不存在，则将用例写入缓存池中
                    if case_id_exit is False:
                        CacheHandler.update_cache(cache_name=k, value=v)
                        # case_data[k] = v
                    # 当 case_id 为 True 存在时，则抛出异常
                    elif case_id_exit is True:
                        raise ValueError(f"case_id: {k} 存在重复项, 请修改case_id\n"
                                         f"文件路径: {i}")


write_case_process()

1、如上代码所示get_all_files会遍历获取data文件夹下所有的yaml和.yml文件，这两种文件里面写的是测试用例具体步骤。

2、CaseData(i).case_process(case_id_switch=True),解析测试数据，对应的代码在utils->read_files_tools_analysis.py文件中。最后返回的是一个列表套字典格式的数据，这个具体细节在下面一小节讲。

3、在循环中的每个用例中，通过遍历键值对for k, v in case.items()来检查用例是否已存在于缓存池中。

4、如果该用例的case_id不存在于缓存_cache_config字典中，则将该用例数据通过CacheHandler.update_cache(cache_name=k, value=v)写入缓存池中，如果存在该case_id则会抛出异常，「所以在写用例时，case_id必须要是唯一的」

5、存入到了缓存池格式为{"case_id_1":{"数据_1"},"case_id_2":{"数据_2"}}

解析yaml测试数据

接上文中的第二点，CaseData(i).case_process(case_id_switch=True)，对应主代码如下：

class CaseData(CaseDataCheck):

    def case_process(self, case_id_switch: Union[None, bool] = None):
        data = GetYamlData(self.file_path).get_yaml_data()
        case_list = []
        for key, values in data.items():
            # 公共配置中的数据，与用例数据不同，需要单独处理
            if key != 'case_common':
                self.case_data = values
                self.case_id = key
                super().check_params_exit()
                case_date = {
                    'method': self.get_method,
                    'is_run': self.case_data.get(TestCaseEnum.IS_RUN.value[0]),
                    'url': self.get_host,
                    'detail': self.case_data.get(TestCaseEnum.DETAIL.value[0]),
                    'headers': self.case_data.get(TestCaseEnum.HEADERS.value[0]),
                    'requestType': super().get_request_type,
                    'data': self.case_data.get(TestCaseEnum.DATA.value[0]),
                    'dependence_case': self.case_data.get(TestCaseEnum.DE_CASE.value[0]),
                    'dependence_case_data': self.get_dependence_case_data,
                    "current_request_set_cache": self.case_data.get(TestCaseEnum.CURRENT_RE_SET_CACHE.value[0]),
                    "sql": self.get_sql,
                    "assert_data": self.get_assert,
                    "setup_sql": self.case_data.get(TestCaseEnum.SETUP_SQL.value[0]),
                    "teardown": self.case_data.get(TestCaseEnum.TEARDOWN.value[0]),
                    "teardown_sql": self.case_data.get(TestCaseEnum.TEARDOWN_SQL.value[0]),
                    "sleep": self.case_data.get(TestCaseEnum.SLEEP.value[0]),
                }
                if case_id_switch is True:
                    case_list.append({key: TestCase(**case_date).dict()})
                else:
                    case_list.append(TestCase(**case_date).dict())

        return case_list

1、CaseData这个类继承了CaseDataCheck类，用于用例的规范检查，都很容易看懂，这个就直接看CaseDataCheck源码就好了，源码在同文件下。

2、case_id_switch，默认传入要都为True，后续都是转化为键值对的形式，case_id作为key，它下面的测试数据作为value，value也是一个字典。对应代码case_list.append({key: TestCase(**case_date).dict()})

3、获取yaml中的具体数据，GetYamlData(self.file_path).get_yaml_data()

def get_yaml_data(self) -> dict:
    """
        获取 yaml 中的数据
        :param: fileDir:
        :return:
        """
    # 判断文件是否存在
    if os.path.exists(self.file_dir):
        data = open(self.file_dir, 'r', encoding='utf-8')
        res = yaml.load(data, Loader=yaml.FullLoader)
    else:
        raise FileNotFoundError("文件路径不存在")
        return res

示列yaml文件取自源码中的，get_user_info.yaml

  • # 公共参数case_common: allureEpic: 开发平台接口 allureFeature: 个人信息模块 allureStory: 个人信息接口get_user_info_01: host: {{host()}} url: /user/lg/userinfo/json method: GET detail: 正常获取个人身份信息 headers: Content-Type: multipart/form-data; # 这里cookie的值，写的是存入缓存的名称 cookie: cache{login_cookie} # 请求的数据，是 params 还是 json、或者file、data requestType: data # 是否执行，空或者 true 都会执行 is_run: data: # 是否有依赖业务，为空或者false则表示没有 dependence_case: False # 依赖的数据 dependence_case_data: assert: # 断言接口状态码 errorCode: jsonpath: .errorCode type: == value: 0 AssertType: # 断言接口返回的username username: jsonpath: .data.userInfo.username type: == value: '18800000001' AssertType: sql:

返回的格式如下

4、公共配置中的数据（case_common字段），与用例数据不同，需要单独处理，这里处理的是用例数据

5、super().check_params_exit():

def _assert(self, attr: Text):
    # 断言如果为真，程序继续运行，如果为假，则程序会抛出AssertionError异常，并附带后面括号的提示内容
    assert attr in self.case_data.keys(), (
        f"用例ID为 {self.case_id} 的用例中缺少 {attr} 参数，请确认用例内容是否编写规范."
        f"当前用例文件路径：{self.file_path}"
    )

def check_params_exit(self):
    for enum in list(TestCaseEnum._value2member_map_.keys()):
        if enum[1]:
            self._assert(enum[0])

通过判断TestCaseEnum自定义的枚举类参数的第二个值，如果是True则是必填项，当必填项未填则提醒用户填写。

6、每个字段都需要经过校验，校验完成后case_data是一个字典，使用TestCase映射字典存储后，再存在case_list里面。最后返回case_list。

以源码中的get_user_info.yaml为列，此步骤返回的case_list输出的结果如下：

测试用例执行

pytest所要收集的测试用例都在test_case文件夹下

conftest.py

位于test_case的根目录下

session级别操作

1、先将report文件夹下的文件全部删除，autouse=False，只执行一次

2、获取登录的cookie，在这里是已经写死了的，autouse=True，并且将这个cookie值存到缓存当中。

缓存文件utils/cache_process/cache_control.py

在该缓存文件中有一个空字典_cache_config = {}，每次加入则以键值对的形式放在这个字典中，需要用的话就从里面取出。

_cache_config = {}


class CacheHandler:
    @staticmethod
    def get_cache(cache_data):
        try:
            return _cache_config[cache_data]
        except KeyError:
            raise ValueNotFoundError(f"{cache_data}的缓存数据未找到，请检查是否将该数据存入缓存中")

    @staticmethod
    def update_cache(*, cache_name, value):
        _cache_config[cache_name] = value

3、跳过用例

@pytest.fixture(scope="function", autouse=True)
def case_skip(in_data):
    """处理跳过用例"""
    in_data = TestCase(**in_data) # 这里直接是TestCase的对象，可以直接点出它下面的所有属性
    # 如果yaml文件中,is_run字段为False，则会执行以下代码，从而达到跳过该条用例的目的
    if ast.literal_eval(cache_regular(str(in_data.is_run))) is False:
        allure.dynamic.title(in_data.detail)
        allure_step_no(f"请求URL: {in_data.is_run}")
        allure_step_no(f"请求方式: {in_data.method}")
        allure_step("请求头: ", in_data.headers)
        allure_step("请求数据: ", in_data.data)
        allure_step("依赖数据: ", in_data.dependence_case_data)
        allure_step("预期数据: ", in_data.assert_data)
        pytest.skip()

用例demo分析

这里以源码中的test_case -> UserInfo -> test_get_user_info.py示列文件作为我们用例分析的demo文件。

代码如下：

import allure
import pytest
from utils.read_files_tools.get_yaml_data_analysis import GetTestCase
from utils.assertion.assert_control import Assert
from utils.requests_tool.request_control import RequestControl
from utils.read_files_tools.regular_control import regular
from utils.requests_tool.teardown_control import TearDownHandler


case_id = ['get_user_info_01']
TestData = GetTestCase.case_data(case_id)
re_data = regular(str(TestData))


@allure.epic("开发平台接口")
@allure.feature("个人信息模块")
class TestGetUserInfo:

    @allure.story("个人信息接口")
    @pytest.mark.parametrize('in_data', eval(re_data), ids=[i['detail'] for i in TestData])
    def test_get_user_info(self, in_data, case_skip):
        """
        :param :
        :return:
        """
        res = RequestControl(in_data).http_request()
        TearDownHandler(res).teardown_handle()
        Assert(assert_data=in_data['assert_data'],
               sql_data=res.sql_data,
               request_data=res.body,
               response_data=res.response_data,
               status_code=res.status_code).assert_type_handle()


if __name__ == '__main__':
    pytest.main(['test_test_get_user_info.py', '-s', '-W', 'ignore:Module already imported:pytest.PytestWarning'])

对应的yaml文件如下：

# 公共参数
case_common:
  allureEpic: 开发平台接口
  allureFeature: 个人信息模块
  allureStory: 个人信息接口

get_user_info_01:
    host: ${{host()}}
    url: /user/lg/userinfo/json
    method: GET
    detail: 正常获取个人身份信息
    headers:
      Content-Type: multipart/form-data;
      # 这里cookie的值，写的是存入缓存的名称
      cookie: $cache{login_cookie}
    # 请求的数据，是 params 还是 json、或者file、data
    requestType: data
    # 是否执行，空或者 true 都会执行
    is_run:
    data:
      # 是否有依赖业务，为空或者false则表示没有
    dependence_case: False
        # 依赖的数据
    dependence_case_data:
    assert:
      # 断言接口状态码
      errorCode:
        jsonpath: $.errorCode
        type: ==
        value: 0
        AssertType:
      # 断言接口返回的username
      username:
        jsonpath: $.data.userInfo.username
        type: ==
        value: '18800000001'
        AssertType:
    sql:

加载yaml中的数据

「1、case_id = ['get_user_info_01']」
  • 一个yaml文件下的case_id，用列表存储，可以有多个。
「2、TestData = GetTestCase.case_data(case_id)」
  • TestData 的格式为 [{case_data_1},{case_data_2}]
class GetTestCase:
    @staticmethod
    def case_data(case_id_lists: List):
        case_lists = []
        # 一个yaml文件中，有一个或多个case_id，把每一个case_id下的数据取出放到一个列表里
        for i in case_id_lists:
            _data = CacheHandler.get_cache(i)
            case_lists.append(_data)
        return case_lists
  • **CacheHandler.get_cache(i)**：将数据从缓存中取出数据，得到的TestData 数据格式为[[数据1],[数据2]]，所取的数据从哪来——在上文中的解析yaml测试数据一节中有详细分析。
「3、re_data = regular(str(TestData))」
  • 把TestData强制转化成了str类型
  • 使用正则替换请求数据，比如{{ host() }}数据，需要进行处理，utils -> read_files_tools -> regular_control.py -> regular

为了更客观的看到正则的数据，我把除了get_user_info.yaml的用例都设置成了skip。上方的代码也是我增加了print数据。

第一个输出的target内容如下: （要方便看的话需要结合get_user_info.yaml文件）

("[{'url': '${{host()}}/user/lg/userinfo/json', 'method': 'GET', 'detail': "
 "'正常获取个人身份信息', 'assert_data': {'errorCode': {'jsonpath': '$.errorCode', "
 "'type': '==', 'value': 0, 'AssertType': None}, 'username': {'jsonpath': "
 "'$.data.userInfo.username', 'type': '==', 'value': '18800000001', "
 "'AssertType': None}}, 'headers': {'Content-Type': 'multipart/form-data;', "
 "'cookie': '$cache{login_cookie}'}, 'requestType': 'DATA', 'is_run': None, "
 "'data': None, 'dependence_case': False, 'dependence_case_data': None, 'sql': "
 "None, 'setup_sql': None, 'status_code': None, 'teardown_sql': None, "
 "'teardown': None, 'current_request_set_cache': None, 'sleep': None}]")

第二个输出的key内容如下：

host()

由此推断regular_pattern = r'\{{(.*?)}}'这个规则是为了过滤提取形如{{host()}}格式的参数。

比如 key 是 int:host(1,2,3) 那就走if分支，提取方法名和参数
如果 key 是 host(1,2,3) 那就走else分支，提取方法名和参数

第三个输出的是else分支下的target，就是进行了替换后的target

欧耶耶耶耶耶else里面的target:[{'url': 'https://www.wanandroid.com/user/lg/userinfo/json', 'method': 'GET', 'detail': '正常获取个人身份信息', 'assert_data': {'errorCode': {'jsonpath': '$.errorCode', 'type': '==', 'value': 0, 'AssertType': None}, 'username': {'jsonpath': '$.data.userInfo.username', 'type': '==', 'value': '18800000001', 'AssertType': None}}, 'headers': {'Content-Type': 'multipart/form-data;', 'cookie': '$cache{login_cookie}'}, 'requestType': 'DATA', 'is_run': None, 'data': None, 'dependence_case': False, 'dependence_case_data': None, 'sql': None, 'setup_sql': None, 'status_code': None, 'teardown_sql': None, 'teardown': None, 'current_request_set_cache': None, 'sleep': None}]

替换使用的是getattr，进行获取Context()类下面对应的方法。后面加个括号表示直接调用获取其返回值。

  • 无参数走if，有参数走else
  • if value_name == "": value_data = getattr(Context(), func_name)() else: value_data = getattr(Context(), func_name)(*value_name.split(","))

每一次执行后都会进行target的新旧替换，直至这个字符串里面没有形如{{host()}}格式的参数。

target = re.sub(regular_pattern, str(value_data), target, 1) # 1表示替换的最大次数为1

同理 $cache{login_02_v_code}获取缓存中的数据是有专门的一个cache_regular方法，只是规则不一样。处理步骤和上面的如出一辙。

数据驱动执行用例

@pytest.mark.parametrize('in_data', eval(re_data), ids=[i['detail'] for i in TestData])

这里的re_data的数据类型是<class 'str'>，我们使用eval方法把他转化成列表。

ids取detail描述作为allure报告中的标识，如下图所示:


循环读取执行用例，列表有多长，就有多少个测试用例。

用例前置

def test_get_user_info(self, in_data, case_skip):

in_data：不做过多的说明，就是yaml文件中经过处理的所有数据。

case_skip：这个是在conftest.py设置的操作，用来判断用例是否要跳过。

request请求配置

res = RequestControl(in_data).http_request()

对应方法所在的位置：\utils\requests_tool\request_control.py --> http_request()

in_data就是我们前面运行的那一大串：

[{'url': 'https://www.wanandroid.com/user/lg/userinfo/json', 'method': 'GET', 'detail': '正常获取个人身份信息', 'assert_data': {'errorCode': {'jsonpath': '$.errorCode', 'type': '==', 'value': 0, 'AssertType': None}, 'username': {'jsonpath': '$.data.userInfo.username', 'type': '==', 'value': '18800000001', 'AssertType': None}}, 'headers': {'Content-Type': 'multipart/form-data;', 'cookie': '$cache{login_cookie}'}, 'requestType': 'DATA', 'is_run': None, 'data': None, 'dependence_case': False, 'dependence_case_data': None, 'sql': None, 'setup_sql': None, 'status_code': None, 'teardown_sql': None, 'teardown': None, 'current_request_set_cache': None, 'sleep': None}]

对request请求类型，如json、params、file、data等类型进行相关处理，具体可以看源码，这里我做一个流程化分析，封装的很全面。

# 这里只是获取到了一个对象，当没有真的执行 ，这些对象方法都在当前py文件中
# 比如 self.request_type_for_json，只是将这个方法地址存在了 requests_type_mapping 字典中
# 在self.request_type_for_json后面加了 () 才代表执行了这个函数对象
requests_type_mapping = {
            RequestType.JSON.value: self.request_type_for_json,
            RequestType.NONE.value: self.request_type_for_none,
            RequestType.PARAMS.value: self.request_type_for_params,
            RequestType.FILE.value: self.request_type_for_file,
            RequestType.DATA.value: self.request_type_for_data,
            RequestType.EXPORT.value: self.request_type_for_export
        }

以下这一部分是对多业务逻辑进行处理：

# 处理多业务逻辑
if dependent_switch is True:
    # 对jsonpath和依赖的数据进行替换
    DependentCase(self.__yaml_case).get_dependent_data()

res = requests_type_mapping.get(self.__yaml_case.requestType)(
    headers=self.__yaml_case.headers,
    method=self.__yaml_case.method,
    **kwargs
)

if self.__yaml_case.sleep is not None:
    time.sleep(self.__yaml_case.sleep)

_res_data = self._check_params(
    res=res,
    yaml_data=self.__yaml_case)

self.api_allure_step(
    url=_res_data.url,
    headers=str(_res_data.headers),
    method=_res_data.method,
    data=str(_res_data.body),
    assert_data=str(_res_data.assert_data),
    res_time=str(_res_data.res_time),
    res=_res_data.response_data
)
# 将当前请求数据存入缓存中
SetCurrentRequestCache(
    current_request_set_cache=self.__yaml_case.current_request_set_cache,
    request_data=self.__yaml_case.data,
    response_data=res
).set_caches_main()

return _res_data
多业务逻辑代码分析
「片段1」
# 处理多业务逻辑
if dependent_switch is True:
    DependentCase(self.__yaml_case).get_dependent_data()

utils -> requests_tool -> dependent_case.py

def get_dependent_data(self) -> None:
    """
    jsonpath 和 依赖的数据,进行替换
    :return:
    """
    _dependent_data = DependentCase(self.__yaml_case).is_dependent()
    print("数据依赖ssssssssssssss：",_dependent_data)
    _new_data = None
    # 判断有依赖
    if _dependent_data is not None and _dependent_data is not False:
        # if _dependent_data is not False:
        for key, value in _dependent_data.items():
            # 通过jsonpath判断出需要替换数据的位置
            _change_data = key.split(".")
            # jsonpath 数据解析
            # 不要删 这个yaml_case
            yaml_case = self.__yaml_case
            _new_data = jsonpath_replace(change_data=_change_data, key_name='yaml_case')
            # 最终提取到的数据,转换成 __yaml_case.data
            _new_data += ' = ' + str(value)
            exec(_new_data)

_dependent_data = DependentCase(self.__yaml_case).is_dependent()，源码自己点开。

  • 如果 _dependent_type 是 True ，则表示当前用例需要运行数据依赖。
  • 否则返回 False，表示当前用例不需要运行数据依赖。
  • 如果依赖的数据是 SQL，则根据 _dependence_case_dates 中的相关内容，在数据库中执行 SQL 语句，并获取查询结果。
  • 如果依赖的数据是请求响应体中的数据，则依据 jsonpath 表达式获取响应数据，并且根据依赖关系设置的值进行处理（如替换，存储等）。
  • 如果依赖数据是请求数据，则从请求数据中根据 jsonpath 表达式获取数据，并进行处理。
  • 如果依赖数据类型不符合要求，则抛出异常。
  • 这段代码是 Python 中的一个类方法，具体实现了在运行测试用例时，判断测试用例是否有数据依赖关系，并且根据依赖关系获取数据。 具体解析如下： 综上所述，以上代码实现了数据依赖处理的功能，并且根据依赖关系获取依赖数据。
  • __yaml_case 是当前类的一个私有属性，用来获取测试用例相关的数据，包括依赖的类型、依赖用例的数据、依赖数据的设置等等。
  • _dependent_type = self.__yaml_case.dependence_case 获取当前测试用例是否需要运行数据依赖关系，只有是True的时候才会执行依赖。
  • _dependence_case_dates = self.__yaml_case.dependence_case_data 获取当前测试用例的依赖用例数据。
  • _setup_sql = self.__yaml_case.setup_sql 获取设置setup_sql数据库语句，属于前置sql操作。
  • 判断是否有依赖：
  • 如果需要数据依赖，则根据依赖关系获取依赖数据，并返回依赖数据。
  • 最后是异常处理的内容，如果在获取依赖数据的过程中出现了异常（如 KeyError 或 TypeError），则抛出异常并给出具体的原因。
「片段2」
 # 假如请求的requestType是data
# 那下面代码相当于是 self.request_type_for_data( headers=self.__yaml_case.headers,method=self.__yaml_case.method,**kwargs)
res = requests_type_mapping.get(self.__yaml_case.requestType)(
    headers=self.__yaml_case.headers,
    method=self.__yaml_case.method,
    **kwargs
)

这段代码是一个 Python 中的函数调用语句，使用了字典对象的 get() 方法来实现根据 key 值来获取 value 的功能。作用是根据请求类型来调用对应的方法，然后将 self.__yaml_case.headers 和 self.__yaml_case.method 传递给该方法并返回结果给变量 res，同时也把 kwargs 中的其他参数传递给该方法。

具体解析如下：

requests_type_mapping 是一个字典对象。

self.__yaml_case.requestType 获取了一个字符串类型的请求类型（例如 GET、POST、DELETE 等）。

requests_type_mapping.get(self.__yaml_case.requestType)

根据self.__yaml_case.requestType作为 key 值，从requests_type_mapping字典对象获取相应的 value 值。

  • 如果能找到对应的 value 值，则根据获取到的函数来执行对应的请求。
  • 如果requests_type_mapping 中不存在对应的 key 值，则返回默认值 None。

如果能在 requests_type_mapping 字典对象中找到对应的 value 值，则通过调用该函数来执行相应的请求，将 self.__yaml_case.headers 和 self.__yaml_case.method 作为参数传递给该函数。

**kwargs 表示可变长参数，将其中的参数作为命名参数传递给被调用的函数，并执行该函数。

调用结果由 res 变量接收，通常是一个请求响应对象。

数据依赖

写成如下形式：

关键字：current_request_set_cache

示列：

current_request_set_cache:
    # 1、response 从响应中提取内容 2、request从请求中提取内容
    - type: response
        jsonpath: $.data.data.[0].id
        # 自定义的缓存名称
        name: test_query_shop_brand_list_02_id

「作者建议：」也可以在collect_addtool_01中提取出相关数据，放在缓存里面，后续别的接口直接从缓存里面读

后置操作(数据清洗)

utils -> requests_tool -> teardown_control.py -> teardown_handle()

TearDownHandler(res).teardown_handle()

这里的res是在上一步返回出来的，res的数据格式为：

class ResponseData(BaseModel):
    url: Text
    is_run: Union[None, bool, Text]
    detail: Text
    response_data: Text
    request_body: Any
    method: Text
    sql_data: Dict
    yaml_data: "TestCase"
    headers: Dict
    cookie: Dict
    assert_data: Dict
    res_time: Union[int, float]
    status_code: int
    teardown: List["TearDown"] = None
    teardown_sql: Union[None, List]
    body: Any

  1. self._res 是当前类的私有属性，代表当前执行测试用例后的结果信息。
  2. _teardown_data = self._res.teardown 获取当前测试用例中 teardown 的相关数据。
  3. _resp_data = self._res.response_data 获取当前测试用例的接口响应数据。
  4. _request_data = self._res.yaml_data.data 获取接口的请求参数。
  5. 如果存在 teardown 数据，则根据关键字进行区分处理。
  6. 如果 teardown 数据中包含关键字 param_prepare，则需要处理前置请求参数等数据。具体实现是调用 self.param_prepare_request_handler() 方法，处理相关数据并返回结果。
  7. 如果 teardown 数据中包含关键字 send_request，则需要发送请求数据。具体实现是调用 self.send_request_handler() 方法，发送相关请求并处理结果。
  8. 最后调用 self.teardown_sql() 方法处理数据操作。
  9. 在 teardown 数据处理的过程中，为了判断当前处理的 teardown 数据是否需要发送请求，需要对两种情况进行区分，一个是 param_prepare，一个是 send_request。

综上所述，该方法主要是根据不同的 teardown 数据操作，进行不同的数据处理，包括发送请求、处理请求参数等。由于 teardown 数据操作可能需要获取当前及之前测试用例的请求响应等信息，因此需要区分处理方式。

断言操作

Assert(assert_data=in_data['assert_data'],
               sql_data=res.sql_data,
               request_data=res.body,
               response_data=res.response_data,
               status_code=res.status_code).assert_type_handle()

\utils\assertion\assert_control.py -> assert_type_handle()

主要代码段：（该文件其他代码从上往下看就可以，整齐划一通俗易懂）

def assert_type_handle(self):
    # 判断请求参数数据库断言
    if self.get_assert_type == "R_SQL":
        self._assert(self._assert_request_data, self.get_sql_data, self.get_message)

        # 判断请求参数为响应数据库断言
    elif self.get_assert_type == "SQL" or self.get_assert_type == "D_SQL":
        self._assert(self._assert_resp_data, self.get_sql_data, self.get_message)

        # 判断非数据库断言类型
    elif self.get_assert_type is None:
        self._assert(self._assert_resp_data, self.get_value, self.get_message)
    else:
        raise AssertTypeError("断言失败，目前只支持数据库断言和响应断言")

该段代码的具体解析如下：

  1. 获取当前实例的断言类型 (get_assert_type)。
  2. 如果断言类型为 "R_SQL"，则调用 _assert() 方法进行请求参数和数据库数据的断言。
  3. 如果断言类型为 "SQL" 或 "D_SQL"，则调用 _assert() 方法进行响应数据和数据库数据的断言。
  4. 如果断言类型为 None，则调用 _assert() 方法进行响应数据和请求参数中的 value 值的断言。
  5. 如果断言类型不为以上三种，抛出自定义异常 AssertTypeError，提示该方法目前只支持数据库断言和响应断言。

注：_assert() 方法为类中的另一个私有方法，用于进行具体的数据断言。

综上所述，该方法主要根据断言类型分别调用不同的数据断言方法。如果类型不支持，则抛出异常。

「断言包含的类型」

class AssertMethod(Enum):
    """断言类型"""
    equals = "=="
    less_than = "lt"
    less_than_or_equals = "le"
    greater_than = "gt"
    greater_than_or_equals = "ge"
    not_equals = "not_eq"
    string_equals = "str_eq"
    length_equals = "len_eq"
    length_greater_than = "len_gt"
    length_greater_than_or_equals = 'len_ge'
    length_less_than = "len_lt"
    length_less_than_or_equals = 'len_le'
    contains = "contains"
    contained_by = 'contained_by'
    startswith = 'startswith'
    endswith = 'endswith'

与作者的问答

大佬，你的框架里面，dependence_case，最新的代码和gitee上的教程内容有点不一样。1、教程里面是set_cache，然后再使用这个缓存数据 2、你最新的代码是直接写的replace_key的一个提取表达式 我两种方式都写了一下，请问这两种方式都是可以的是嘛？还有个问题就是，当前A用例如果依赖了B用例，那么运行A用例的时候，B用例时还会再执行一遍嘛？
收到，第一个问题推荐使用setcache关键字，replacekey为旧版本研发的功能，目前在特定场景下使用会存在一定的问题
第二个问题，是的，dependentcase类似源码调用函数的功能,需要使用的时候去调用它，可以减少部分冗余的用例字段
如果不去调用，是不是可以用current_request_set_cache这个关键字，在所要依赖的用例里面提取出相关数据放在缓存里面，后续别的接口直接从缓存里面读
是的

结束

allure报告生成

数据统计汇总

结果发送

部分代码分析：

「片段1」

pytest.main(['-s', '-W', 'ignore:Module already imported:pytest.PytestWarning',
             '--alluredir', './report/tmp', "--clean-alluredir"])

该代码段是运行 Python 中的 Pytest 模块，并传入了一些参数来设置测试的选项。具体来说，该代码执行以下操作：

  1. pytest.main()：调用 Pytest 模块中的 main() 函数开始运行测试。
  2. ['-s', '-W', 'ignore:Module already imported:pytest.PytestWarning', '--alluredir', './report/tmp', "--clean-alluredir"]：设置 Pytest 运行测试的选项参数。
  • -s：表示向标准输出流输出所有测试结果的详细信息，包括 log 信息等。
  • -W：表示忽略特定类型的警告。
  • ignore:Module already imported:pytest.PytestWarning：表示忽略警告 "Module already imported"。
  • --alluredir：表示指定 Allure 测试报告的输出目录。
  • ./report/tmp：表示将测试结果数据写入到指定的 Allure 测试报告数据文件夹中。
  • --clean-alluredir：表示在生成 Allure 测试报告前，先清除指定目录中的旧数据，以免对测试报告的生成产生影响。

以上就是该代码的执行方式和每个参数的作用。在执行 Pytest 测试时，可以使用这些参数来自定义测试运行方式和测试报告的生成。

「片段2」

os.system(r"allure generate ./report/tmp -o ./report/html --clean")

这行代码是在 Python 中调用系统命令，使用 os.system() 函数来执行系统命令。具体命令为：

allure generate ./report/tmp -o ./report/html --clean

该命令使用 Allure 工具生成测试报告，从 ./report/tmp 路径下读取测试结果数据，然后将报告输出到 ./report/html 路径下。

其中，--clean 参数表示在生成测试报告前，清空输出目录，以防止旧的测试报告文件对新测试报告的干扰。

因为是通过 os.system() 函数调用的系统命令，所以该代码的执行结果和命令行的执行结果是一致的。执行成功会生成新的 Allure 报告，执行失败会抛出异常。

【python自动化】playwright长截图&切换标签页&JS注入实战
上一篇
【python自动化】七月PytestAutoApi开源框架学习笔记（一）
下一篇