Yaml版接口自动化详细讲解（Python + pytest + allure + log + yaml + mysql + redis + 钉钉通知 + Jenkins）

                other_tools  # 其他工具类

                        allure_data  # allure封装

                                allure_report_data.py  # allure报告数据清洗

                                详细讲解

"""描述: 收集 allure 报告"""import jsonfrom typing import List, Textfrom common.setting import ensure_path_sepfrom utils.read_files_tools.get_all_files_path import get_all_filesfrom utils.other_tools.models import TestMetrics# 定义了一个名为 AllureFileClean 的类class AllureFileClean:    """allure 报告数据清洗，提取业务需要得数据"""    # @classmethod 是 Python 中的一个装饰器，用来指示一个类方法。类方法与实例方法的不同之处在于，类方法第一个参数必须是类本身，Python 会自动传入该参数，通常命名为 cls，而不是 self，这也是 @classmethod 装饰器的作用之一。通过类名可以直接调用类方法，而不需要实例化对象。    @classmethod    # 使用了一个类装饰器 cls，其返回值是一个列表，该列表中的元素是一个测试用例数据。    def get_testcases(cls) -> List:        """ 获取所有 allure 报告中执行用例的情况"""        # 创建一个空列表 files 用于存放读取到的测试用例数据。        files = []        # 使用 get_all_files 函数来获取所有测试用例文件的路径，并通过迭代器 for 循环遍历每个路径。ensure_path_sep 函数将文件路径格式化为当前操作系统的标准格式（加上 \ 或 / 等分隔符），以确保路径可用。        for i in get_all_files(ensure_path_sep("\\report\\html\\data\\test-cases")):            # 使用 Python 的 with 语句打开测试用例文件，并赋值给变量 file，指定使用 utf-8 编码方式打开文件。            with open(i, 'r', encoding='utf-8') as file:                # 从打开的文件中读取 JSON 数据，并使用 json.load 函数将其加载到 date 变量中。                date = json.load(file)                # 将从文件中获取的数据 date 添加到 files 列表中。                files.append(date)        # 当获取所有测试用例文件的数据后，返回 files 列表作为结果，结束 get_testcases 方法的执行。        return files    # 总的来说，这段代码的作用是遍历指定路径下的所有测试用例文件，将其中的 JSON 数据提取出来，并将所有数据放入列表 files 中，最后返回该列表。这个过程可以以简单而灵活的方式提取并汇总数据，方便后续的数据处理和分析。    # 定义了一个名为 get_failed_case 的方法，返回一个 List 类型对象。    def get_failed_case(self) -> List:        """ 获取到所有失败的用例标题和用例代码路径"""        # 创建一个名为 error_case 的空列表，用于存储所有失败和中断的测试用例的标题和代码路径。        error_case = []        # 使用 for 循环遍历所有测试用例数据（该方法调用了 get_testcases 函数来获取测试用例数据列表，这个函数我们前面已经讲过）。        for i in self.get_testcases():            # 对于每个测试用例数据，如果它的执行状态为 'failed' 或 'broken'，则将其标题和代码路径作为元组 (title, path) 添加到 error_case 列表中。            if i['status'] == 'failed' or i['status'] == 'broken':                error_case.append((i['name'], i['fullName']))        # 当循环遍历完所有测试用例数据，返回 error_case 列表作为结果。        return error_case    # 定义了一个名为 get_failed_cases_detail 的方法，返回一个 Text 类型（即 str 类型）对象。    def get_failed_cases_detail(self) -> Text:        """ 返回所有失败的测试用例相关内容 """        # 调用 get_failed_case 方法获取所有失败和中断的测试用例的标题和代码路径。        date = self.get_failed_case()        # 创建一个空字符串变量 values，用于拼接所有失败用例的详细信息。        values = ""        # 如果存在失败用例，则将 values 设置为 失败用例:\n，表示下面的字符串是所有失败用例的详细信息。        if len(date) >= 1:            values = "失败用例:\n"            # 添加一行格式化的分隔符，用于区分每个测试用例的详细信息            values += "        **********************************\n"            # 遍历所有失败和中断的测试用例数据，对于每个测试用例数据，将其标题和代码路径拼接成字符串，并添加到 values 变量中。            for i in date:                values += "        " + i[0] + ":" + i[1] + "\n"        # 当循环遍历完所有失败和中断的测试用例数据后，返回 values，即所有失败用例的详细信息字符串。        return values    @classmethod    # get_case_count 的类方法，用于统计用例的数量和成功率。    def get_case_count(cls) -> "TestMetrics":        """ 统计用例数量 """        # 用 try 语句块来处理可能抛出的异常情况。        try:            # 定义变量 file_name 来保存所有测试用例执行结果的统计数据路径，确保文件路径的结尾符为 / 。            file_name = ensure_path_sep("/report/html/widgets/summary.json")            # 使用 with 语句打开文件并读取 JSON 数据，确保文件以 utf-8 编码打开。            with open(file_name, 'r', encoding='utf-8') as file:                # 使用 json.load 加载文件数据为 Python 对象，并保存到变量 data 中。                data = json.load(file)            # 获取统计数据中的用例总数。            _case_count = data['statistic']            # # 获取统计数据中的测试执行时间。            _time = data['time']            # 创建集合 keep_keys 来存储要保留的统计数据键名。            keep_keys = {"passed", "failed", "broken", "skipped", "total"}            # 使用字典推导式，将 _case_count 子字典中存在于 keep_keys 集合中的键名和对应的值，在新的字典 run_case_data 中保存下来。            run_case_data = {k: v for k, v in data['statistic'].items() if k in keep_keys}            # 判断运行用例总数大于0则执行下面代码块            if _case_count["total"] > 0:                # 计算测试用例的通过率，并将其存储在run_case_data["pass_rate"]键中，保留两位小数。                run_case_data["pass_rate"] = round(                    (_case_count["passed"] + _case_count["skipped"]) / _case_count["total"] * 100, 2                )            else:                # 如果没有运行的用例，则将通过率设置为0.0。                run_case_data["pass_rate"] = 0.0            # 如果测试用例总数为0，则将时间数据存储在run_case_data['time']键中，否则将运行时间除以1000，保留两位小数，并将其存储在run_case_data['time']键中。            run_case_data['time'] = _time if run_case_data['total'] == 0 else round(_time['duration'] / 1000, 2)            # 使用run_case_data创建一个TestMetrics对象，并将其作为函数的返回值。            return TestMetrics(**run_case_data)        # 如果读取文件时出现FileNotFoundError异常，则捕获该异常并将其存储在exc中。        except FileNotFoundError as exc:            # 从exc中重新抛出一个FileNotFoundError异常，并将新异常的说明文本存储在异常对象实例中。            raise FileNotFoundError(                "程序中检查到您未生成allure报告，"                "通常可能导致的原因是allure环境未配置正确"            ) from exc

                                allure_tools.py  # allure 方法封装

                                详细讲解

import jsonimport allurefrom utils.other_tools.models import AllureAttachmentType# 定义了一个名为allure_step的函数，它接受两个字符串参数step和var，并不返回任何内容（返回值为None）。def allure_step(step: str, var: str) -> None:    """    :param step: 步骤及附件名称    :param var: 附件内容    """    # 定义一个名为step的allure步骤，包含了需要记录的步骤信息。    with allure.step(step):        # 把一个JSON字符串作为附件附加到allure报告的当前步骤中。其中，json.dumps()函数将var对象转换为JSON字符串，然后使用allure.attach()函数将它作为JSON附件附加到报告中。该函数还指定了所附加的附件类型（JSON）。        allure.attach(            json.dumps(                str(var),                ensure_ascii=False,                indent=4),            step,            allure.attachment_type.JSON)# 定义了一个名为allure_attach的函数，它接受三个参数source、name和extension，不返回任何内容（返回值为None）。def allure_attach(source: str, name: str, extension: str):    """    allure报告上传附件、图片、excel等    :param source: 文件路径，相当于传一个文件    :param name: 附件名称    :param extension: 附件的拓展名称    :return:    """    # 获取name参数的拓展名，并转换为大写字母。    _name = name.split('.')[-1].upper()    # 从AllureAttachmentType枚举中获取与拓展名匹配的枚举值，如果找到了，则将它存储在变量_attachment_type中，否则将_attachment_type设置为None。    _attachment_type = getattr(AllureAttachmentType, _name, None)    # 附加一个文件到allure报告中。使用source参数指定文件路径，使用name参数指定文件名，使用attachment_type参数指定文件类型，使用extension参数指定文件的扩展名。    allure.attach.file(        source=source,        name=name,        attachment_type=_attachment_type if _attachment_type is None else _attachment_type.value,        extension=extension    )# 定义了一个名为allure_step_no的函数，它接受一个字符串参数step，不返回任何内容（返回值为None）。def allure_step_no(step: str):    """    无附件的操作步骤    :param step: 步骤名称    :return:    """    # 定义一个名为step的allure步骤，包含了需要记录的步骤信息。    with allure.step(step):        # 这个代码块为空，因此不执行任何操作。        pass

                                error_case_excel.p  # 收集allure异常用例，生成excel测试报告

                                详细讲解

import jsonimport shutilimport astimport xlwingsfrom common.setting import ensure_path_sepfrom utils.read_files_tools.get_all_files_path import get_all_filesfrom utils.notify.wechat_send import WeChatSendfrom utils.other_tools.allure_data.allure_report_data import AllureFileClean# TODO 还需要处理动态值# 定义一个名为ErrorTestCase的类。class ErrorTestCase:    """ 收集错误的excel """    # 定义一个名为__init__的方法，用于初始化对象，并接受一个默认参数self。    def __init__(self):        # 设置test_case_path属性，将其初始化为\report\html\data\test-cases\字符串。ensure_path_sep函数可能会在字符串结尾添加路径分隔符("/")或者""，以保证路径的完整性。        self.test_case_path = ensure_path_sep("\\report\\html\\data\\test-cases\\")    # 定义一个名为get_error_case_data的方法，用于获取失败用例的数据，其中self参数是该类的实例本身。    def get_error_case_data(self):        """        收集所有失败用例的数据        @return:        """        # 调用名为get_all_files的函数，传入test_case_path属性作为参数，获取该路径下的所有文件路径并返回path。        path = get_all_files(self.test_case_path)        # 创建一个空的列表files，用于存储所有执行失败的用例数据。        files = []        # 遍历path列表。        for i in path:            # 用"r"模式打开文件路径i，命名为file。encoding='utf-8'指定打开文件的字符编码，这里使用utf-8格式打开文件。            with open(i, 'r', encoding='utf-8') as file:                # 用json.load()方法从文件file中读取内容，并将读取的内容存储在date变量中。                date = json.load(file)                # 使用if语句判断读取的内容中的状态是否为failed或者broken，如果是则进入if语句块。                if date['status'] == 'failed' or date['status'] == 'broken':                    # 将符合条件的date添加到files列表中。                    files.append(date)        # 打印files列表中收集到的所有执行失败的用例数据。        print(files)        # 返回files列表。        return files    @classmethod    # 定义一个名为get_case_name的方法，并且该方法接受两个参数，一个是cls表示该方法属于该类，第二个参数是test_case，表示要处理的测试用例。    def get_case_name(cls, test_case):        """        收集测试用例名称        @return:        """        # 获取测试用例字典中的name属性，通过字符串的split()方法，按照'['进行分割，并且将分割后的结果存储在name列表中。        name = test_case['name'].split('[')        # 获取列表中的第二个元素，并使用切片操作去掉该元素的最后一个字符']'，将获取到的值存储在case_name变量中。        case_name = name[1][:-1]        # 返回得到的测试用例名称。        return case_name    @classmethod    # 定义了一个名为get_parameters的方法，用于获取allure报告中parameters参数中的内容，即请求前的数据。如果测试用例未发送请求导致异常情况，则该函数用于依然能够处理该用例。    def get_parameters(cls, test_case):        """        获取allure报告中的 parameters 参数内容, 请求前的数据        用于兼容用例执行异常，未发送请求导致的情况        @return:        """        # 获取测试用例字典中的parameters参数的第一个字典中的value属性。        parameters = test_case['parameters'][0]['value']        # 通过ast.literal_eval()方法对parameters进行字面上的评估，将字符串转换为相应的Python数据类型并进行返回。这里的parameters是一个字符串形式的dict类型，使用该方法能够将其转化为Python的dict类型，以方便处理。        return ast.literal_eval(parameters)    @classmethod    # 定义了一个名为get_test_stage的方法，用于获取allure报告中请求后的数据。    def get_test_stage(cls, test_case):        """        获取allure报告中请求后的数据        @return:        """        # 获取测试用例字典中的testStage字典中的steps属性，并将其存储在test_stage变量中。        test_stage = test_case['testStage']['steps']        # 将获取到的测试步骤数据进行返回。        return test_stage    # 定义了一个名为get_case_url的方法，用于获取测试用例的url    def get_case_url(self, test_case):        """        获取测试用例的 url        @param test_case:        @return:        """        # 如果测试用例的状态为broken，即测试用例的步骤数据异常。        if test_case['testStage']['status'] == 'broken':            # 调用get_parameters方法获取测试用例请求前的数据，再从中取出url属性的值，然后将其存储到_url变量中。            _url = self.get_parameters(test_case)['url']        # 如果测试用例的状态不为broken。        else:            # 调用get_test_stage方法获取测试用例请求后的数据，然后从中取出倒数第七个步骤的name属性的值的子串，存储在_url变量中。            _url = self.get_test_stage(test_case)[-7]['name'][7:]        # 将获取到的url数据进行返回。        return _url    # 定义了一个名为get_method的方法，用于获取测试用例的请求方式。    def get_method(self, test_case):        """        获取用例中的请求方式        @param test_case:        @return:        """        # 如果测试用例的状态为broken，即测试用例的步骤数据异常。        if test_case['testStage']['status'] == 'broken':            # 调用get_parameters方法获取测试用例请求前的数据，再从中取出method属性的值，然后将其存储到_method变量中。            _method = self.get_parameters(test_case)['method']        # 如果测试用例的状态不为broken。        else:            # 调用get_test_stage方法获取测试用例请求后的数据，然后从中取出倒数第六个步骤的name属性的值的子串，存储在_method变量中。            _method = self.get_test_stage(test_case)[-6]['name'][6:]        # 将获取到的请求方式进行返回。        return _method    # 定义了一个名为get_headers的方法，用于获取测试用例的请求头信息。    def get_headers(self, test_case):        """        获取用例中的请求头        @return:        """        # 如果测试用例的状态为broken，即测试用例的步骤数据异常。        if test_case['testStage']['status'] == 'broken':            # 调用get_parameters方法获取测试用例请求前的数据，再从中取出headers属性的值，然后将其存储到_headers变量中。            _headers = self.get_parameters(test_case)['headers']        # 如果测试用例的状态不为broken。        else:            # 从测试用例请求后的数据中取出倒数第五个步骤的attachments属性的第一个元素，从中取出source属性的值，即请求头附件的文件名。            _headers_attachment = self.get_test_stage(test_case)[-5]['attachments'][0]['source']            # 拼接请求头附件文件的完整路径，并将其存储到path变量中。            path = ensure_path_sep("\\report\\html\\data\\attachments\\" + _headers_attachment)            # 打开该文件，并将其存储到file变量中。            with open(path, 'r', encoding='utf-8') as file:                # 读取file变量中的JSON格式内容，并将其存储到_headers变量中。                _headers = json.load(file)        # 将获取到的请求头内容进行返回。        return _headers    # 定义了一个名为get_request_type的方法，用于获取测试用例的请求类型。    def get_request_type(self, test_case):        """        获取用例的请求类型        @param test_case:        @return:        """        # 调用get_parameters方法获取测试用例请求前的数据，再从中取出requestType属性的值，并将其存储在request_type变量中。        request_type = self.get_parameters(test_case)['requestType']        # 将获取到的请求类型进行返回。        return request_type    # 定义了一个名为get_case_data的方法，用于获取测试用例的请求体内容。    def get_case_data(self, test_case):        """        获取用例内容        @return:        """        # 如果测试用例的状态为broken，即测试用例的步骤数据异常。        if test_case['testStage']['status'] == 'broken':            # 调用get_parameters方法获取测试用例请求前的数据，再从中取出data属性的值，然后将其存储到_case_data变量中。            _case_data = self.get_parameters(test_case)['data']        # 如果测试用例的状态不为broken。        else:            # 从测试用例请求后的数据中取出倒数第四个步骤的attachments属性的第一个元素，从中取出source属性的值，即请求体附件的文件名。            _case_data_attachments = self.get_test_stage(test_case)[-4]['attachments'][0]['source']            # 拼接请求体附件文件的完整路径，并将其存储到path变量中。            path = ensure_path_sep("\\report\\html\\data\\attachments\\" + _case_data_attachments)            # 打开该文件，并将其存储到file变量中。            with open(path, 'r', encoding='utf-8') as file:                # 读取file变量中的JSON格式内容，并将其存储到_case_data变量中。                _case_data = json.load(file)        # 将获取到的请求体内容进行返回。        return _case_data    # 定义了一个名为get_dependence_case的方法，用于获取测试用例的依赖用例信息。    def get_dependence_case(self, test_case):        """        获取依赖用例        @param test_case:        @return:        """        # 调用get_parameters方法获取测试用例请求前的数据，再从中取出dependence_case_data属性的值，并将其存储在_dependence_case_data变量中。        _dependence_case_data = self.get_parameters(test_case)['dependence_case_data']        # 将获取到的依赖用例信息进行返回。        return _dependence_case_data    # 定义了一个名为get_sql的方法，用于获取测试用例的SQL语句。    def get_sql(self, test_case):        """        获取 sql 数据        @param test_case:        @return:        """        # 调用get_parameters方法获取测试用例请求前的数据，再从中取出sql属性的值，并将其存储在sql变量中。        sql = self.get_parameters(test_case)['sql']        # 将获取到的SQL语句进行返回。        return sql    # 定义了一个名为get_assert的方法，用于获取测试用例的断言数据。    def get_assert(self, test_case):        """        获取断言数据        @param test_case:        @return:        """        # 调用get_parameters方法获取测试用例请求前的数据，再从中取出assert_data属性的值，并将其存储在assert_data变量中。        assert_data = self.get_parameters(test_case)['assert_data']        # 将获取到的断言数据进行返回。        return assert_data    @classmethod    # 定义一个名为get_response的方法，并且该方法接受两个参数，一个是cls，表示该方法是一个类方法，第二个参数是test_case，表示要处理的测试用例。    def get_response(cls, test_case):        """        获取响应内容的数据        @param test_case:        @return:        """        # 如果测试用例的状态为broken，代表该测试用例没有响应数据，将其状态信息作为响应内容。        if test_case['testStage']['status'] == 'broken':            # 将测试用例状态信息作为响应内容，并将其存储在_res_date变量中。            _res_date = test_case['testStage']['statusMessage']        # 如果测试用例状态不是broken，则尝试获取响应数据。        else:            try:                # 获取最后一步操作的附件中的响应数据路径。                res_data_attachments = \                    test_case['testStage']['steps'][-1]['attachments'][0]['source']                # 拼接出响应数据的完整路径。                path = ensure_path_sep("\\report\\html\\data\\attachments\\" + res_data_attachments)                # 打开响应数据所在文件，读取其中的JSON数据，并将其赋值给_res_date变量。                with open(path, 'r', encoding='utf-8') as file:                    _res_date = json.load(file)            # 如果找不到响应数据所在文件，则代表程序中没有提取到响应数据，将None赋值给_res_date变量。            except FileNotFoundError:                _res_date = None        # 将获取到的响应数据进行返回。        return _res_date    @classmethod    # 定义了一个名为get_case_time的类方法，用于获取测试用例的运行时长。    def get_case_time(cls, test_case):        """        获取用例运行时长        @param test_case:        @return:        """        # 获取测试用例运行时长，并将其单位转换为毫秒，并将其存储在case_time变量中。        case_time = str(test_case['time']['duration']) + "ms"        # 将获取到的运行时长进行返回。        return case_time    @classmethod    # 定义了一个名为get_uid的类方法，用于获取测试用例在Allure报告中的唯一标识符（uid）。    def get_uid(cls, test_case):        """        获取 allure 报告中的 uid        @param test_case:        @return:        """        # 获取测试用例的唯一标识符（uid），并将其存储在uid变量中。        uid = test_case['uid']        # 将获取到的uid进行返回。        return uid# 定义了用于整理运行失败的测试用例成excel报告的方法class ErrorCaseExcel:    """ 收集运行失败的用例，整理成excel报告 """    # 定义类的构造方法。    def __init__(self):        # 定义一个名为_excel_template的私有变量，表示excel模板文件的路径。        _excel_template = ensure_path_sep("\\utils\\other_tools\\allure_data\\自动化异常测试用例.xlsx")        # 定义一个名为self._file_path的变量，表示生成excel报告的文件路径。        self._file_path = ensure_path_sep("\\Files\\" + "自动化异常测试用例.xlsx")        # 将_excel_template模板文件复制到self._file_path路径下。        shutil.copyfile(class="lazy" data-src=_excel_template, dst=self._file_path)        # 创建一个Excel应用程序对象。        self.app = xlwings.App(visible=False, add_book=False)        # 打开excel报告文件，返回一个xlwings.Book对象。        self.w_book = self.app.books.open(self._file_path, read_only=False)        # 选取指定名称的工作表异常用例。        self.sheet = self.w_book.sheets['异常用例']  # 或通过索引选取        # 创建一个ErrorTestCase对象，用于收集运行失败的测试用例。        self.case_data = ErrorTestCase()    # 定义一个名为background_color的方法，参数是self，表示类的实例本身，position表示要设置背景颜色的单元格位置，rgb表示要设置的背景色的RGB值。    def background_color(self, position: str, rgb: tuple):        """        excel 单元格设置背景色        @param rgb: rgb 颜色 rgb=(0，255，0)        @param position: 位置，如 A1, B1...        @return:        """        # 获取工作表中指定位置的单元格范围对象。        rng = self.sheet.range(position)        # 设置单元格范围对象的颜色属性为RGB值rgb，并将值赋给变量excel_rgb。        excel_rgb = rng.color = rgb        # 将设置的颜色RGB值返回。        return excel_rgb    # 定义一个名为column_width的方法，参数是self，表示类的实例本身，position表示要设置列宽的列的位置（如"A"，"B"等），width表示要设置的列宽度。    def column_width(self, position: str, width: int):        """        设置列宽        @return:        """        # 获取工作表中指定列的范围对象。        rng = self.sheet.range(position)        # 将指定列的列宽设置为width，并将值赋给变量excel_column_width。        excel_column_width = rng.column_width = width        # 将设置的列宽度返回。        return excel_column_width    # 定义一个名为row_height的方法，参数是self，表示类的实例本身，position表示要设置行高的行的位置，height表示要设置的行高度。    def row_height(self, position, height):        """        设置行高        @param position:        @param height:        @return:        """        # 获取工作表中指定行的范围对象。        rng = self.sheet.range(position)        # 将指定行的行高设置为height，并将值赋给变量excel_row_height。        excel_row_height = rng.row_height = height        # 将设置的行高度返回。        return excel_row_height    # 定义一个名为column_width_adaptation的方法，参数是self，表示类的实例本身，position表示要进行宽度自适应的列的范围    def column_width_adaptation(self, position):        """        excel 所有列宽度自适应        @return:        """        # 获取工作表中指定范围的对象。        rng = self.sheet.range(position)        # 将该范围内所有列的宽度自适应调整。        auto_fit = rng.columns.autofit()        返回自适应后的结果。        return auto_fit    # 定义一个名为row_width_adaptation的方法，参数是self，表示类的实例本身，position表示要进行宽度自适应的行的范围。    def row_width_adaptation(self, position):        """        excel 设置所有行宽自适应        @return:        """        # 获取工作表中指定范围的对象。        rng = self.sheet.range(position)        # 将该范围内所有行的宽度自适应调整。        row_adaptation = rng.rows.autofit()        # 返回自适应后的结果。        return row_adaptation    # 定义一个名为write_excel_content的方法。参数是self，表示类的实例本身，position表示要写入的位置，value表示要写入的内容，均为字符串类型。    def write_excel_content(self, position: str, value: str):        """        excel 中写入内容        @param value:        @param position:        @return:        """        # 将给定位置的单元格赋值为给定的值。由于sheet是当前实例的一个属性，因此可以使用self.sheet来获取该工作表对象。range(position)表示获取给定位置的单元格对象，然后对其value属性进行赋值操作。        self.sheet.range(position).value = value    # 定义一个名为write_case的方法，参数是self，表示类的实例本身。    def write_case(self):        """        用例中写入失败用例数据        @return:        """                # 从case_data对象获取失败用例数据。        _data = self.case_data.get_error_case_data()        # 如果存在失败用例，则继续执行。        if len(_data) > 0:            # 设置行号开始的编号，默认从第二行开始。            num = 2            # 遍历失败用例数据。            for data in _data:                # 在第num行的“A”列中写入用例的唯一标识。                self.write_excel_content(position="A" + str(num), value=str(self.case_data.get_uid(data)))                # 在第num行的“B”列中写入用例的名称。                self.write_excel_content(position='B' + str(num), value=str(self.case_data.get_case_name(data)))                # 在第num行的“C”列中写入用例的 URL。                self.write_excel_content(position="C" + str(num), value=str(self.case_data.get_case_url(data)))                # 在第num行的“D”列中写入用例的请求方法名。                self.write_excel_content(position="D" + str(num), value=str(self.case_data.get_method(data)))                # 在第num行的“E”列中写入用例的请求类型。                self.write_excel_content(position="E" + str(num), value=str(self.case_data.get_request_type(data)))                # 在第num行的“F”列中写入用例的头部信息。                self.write_excel_content(position="F" + str(num), value=str(self.case_data.get_headers(data)))                # 在第num行的“G”列中写入用例的请求数据。                self.write_excel_content(position="G" + str(num), value=str(self.case_data.get_case_data(data)))                # 在第num行的“H”列中写入用例依赖的其他用例。                self.write_excel_content(position="H" + str(num), value=str(self.case_data.get_dependence_case(data)))                # 在第num行的“I”列中写入用例的断言信息。                self.write_excel_content(position="I" + str(num), value=str(self.case_data.get_assert(data)))                # 在第num行的“J”列中写入用例的 SQL 语句。                self.write_excel_content(position="J" + str(num), value=str(self.case_data.get_sql(data)))                # 在第num行的“K”列中写入用例的执行时间。                self.write_excel_content(position="K" + str(num), value=str(self.case_data.get_case_time(data)))                # 在第num行的“L”列中写入用例的响应信息。                self.write_excel_content(position="L" + str(num), value=str(self.case_data.get_response(data)))                # 将行号加一，以便写入下一个用例。                num += 1            # 保存文件修改。            self.w_book.save()            # 关闭工作簿。            self.w_book.close()            # 结束 Excel 进程。            self.app.quit()            # 通知企业微信发送文件消息，包括更新的用例数量和文件路径。            WeChatSend(AllureFileClean().get_case_count()).send_file_msg(self._file_path)if __name__ == '__main__':    ErrorCaseExcel().write_case()

                        install_tool  # 安装工具

                                install_requirements.py  # 自动识别安装最新的依赖库

                                详细讲解

"""# @describe: 判断程序是否每次会更新依赖库，如有更新，则自动安装"""import osimport chardetfrom common.setting import ensure_path_sepfrom utils.logging_tool.log_control import INFOfrom utils import config# 这是一个使用 os.system 命令来执行系统命令 pip3 install chardet 的语句。该语句用于在 Python 环境中通过 PyPI 安装 chardet 库。 os.system 函数将给定的字符串作为一个命令运行，并返回其执行结果。具体来说，pip3 是 Python3 中的包管理器工具，install 是 它的一个子命令，用于安装第三方库，而 chardet 则是要安装的库名。执行该语句将自动下载并安装 chardet 库，以便在 Python 中使用它的功能。os.system("pip3 install chardet")# 定义一个名为 InstallRequirements 的类。class InstallRequirements:    """ 自动识别安装最新的依赖库 """    # 定义该类的构造函数，它将在创建类的新实例时自动调用。    def __init__(self):        # 设置 version_library_comparisons.txt 文件的路径。函数ensure_path_sep用于确保路径分隔符为正斜杠。        self.version_library_comparisons_path = ensure_path_sep("\\utils\\other_tools\\install_tool\\") \                  # 设置 requirements.txt 文件的路径。该路径使用os模块函数os.path.dirname获取当前文件所在的目录两级上的目录（即项目的根目录），并拼接 os.sep 和文件名requirements.txt。          + "version_library_comparisons.txt"        self.requirements_path = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) \     + os.sep + "requirements.txt"        # 从 config 中获取镜像 URL，该 URL 作为从镜像源下载软件包的基础 URL。        self.mirror_url = config.mirror_source        # 定义一个名为 read_version_library_comparisons_txt 的方法，它将从文件中读取字符串，并返回读取的内容。    def read_version_library_comparisons_txt(self):        """        获取版本比对默认的文件        @return:        """        # 使用 with 语句来安全地打开指定路径下的文件 version_library_comparisons.txt 并读取其内容。open() 函数的第一个参数是文件路径，第二个参数是打开文件的模式，r 代表只读模式，第三个参数是文件编码方式，这里是 utf-8。        with open(self.version_library_comparisons_path, 'r', encoding="utf-8") as file:            # 返回打开文件后读取的内容，并使用 strip 函数去除字符串开头和结尾的空格。            return file.read().strip(' ')    @classmethod    # 定义一个名为 check_charset 的类方法，它的第一个参数是 cls，指向当前类对象（而不是实例对象）。第二个参数是 file_path，表示要检查字符集的文件路径。    def check_charset(cls, file_path):        """获取文件的字符集"""        # 使用 with 语句打开指定文件，并使用二进制模式(rb)打开文件。        with open(file_path, "rb") as file:            # 读取文件前四个字节的内容。这足以确定文件的编码方式，因为对于大多数文件来说，其编码方式信息包含在开头的几个字节中。            data = file.read(4)            # 使用 chardet 模块检测文件的编码方式，并将检测到的编码方式存储在 charset 变量中。            charset = chardet.detect(data)['encoding']        # 返回检测到的文件编码方式。        return charset    # 定义一个名为 read_requirements 的方法，它将读取指定文件的内容，并返回文件的内容。    def read_requirements(self):        """获取安装文件"""        # 创建一个空字符串，用于存储文件内容。        file_data = ""        # 打开指定文件以读取其内容，并使用 with 语句确保安全关闭文件。 self.requirements_path 是文件的路径， r 表示以只读模式打开该文件， encoding 参数使用 check_charset 函数来确定文件的编码方式。        with open(                self.requirements_path,                'r',                encoding=self.check_charset(self.requirements_path)        ) as file:            # 读取打开文件的每一行，将其作为 line 变量的值。            for line in file:                # 检查读取的行是否包含某个字符串“[0m”，并用空字符串替换该字符串。于是获得的新字符串添加到 file_data 字符串中。                if "[0m" in line:                    line = line.replace("[0m", "")                file_data += line        # 重新打开指定文件以覆盖原始文件，然后将 file_data 中的内容写入文件，实现删除了 “[0m” 字符串的目的。        with open(                self.requirements_path,                "w",                encoding=self.check_charset(self.requirements_path)        ) as file:            file.write(file_data)        # 返回读取的内容字符串。        return file_data    # 定义名为 text_comparison 的方法，其目的是检查库的依赖项是否有更新，并自动安装更新的依赖项。    def text_comparison(self):        """        版本库比对        @return:        """        # 使用 self.read_version_library_comparisons_txt() 方法读取存储了库依赖项版本信息的文件内容，并将其赋值给 read_version_library_comparisons_txt 变量（可能是一个字符串）。        read_version_library_comparisons_txt = self.read_version_library_comparisons_txt()        # 使用 self.read_requirements() 方法读取当前库依赖项的版本信息，并将其存储在 read_requirements 变量中。        read_requirements = self.read_requirements()        # 检查 read_version_library_comparisons_txt 变量和 read_requirements 变量是否相等。如果相等，则没有更新的库依赖项，并输出相关信息；否则，将进行安装更新。        if read_version_library_comparisons_txt == read_requirements:            INFO.logger.info("程序中未检查到更新版本库，已为您跳过自动安装库")        # 程序中如出现不同的文件，则安装        else:            INFO.logger.info("程序中检测到您更新了依赖库，已为您自动安装")            # 如果应用程序中存在版本差异，则使用 os.system 方法调用 pip3 安装库依赖项。            os.system(f"pip3 install -r {self.requirements_path}")            # 更新依赖项库版本信息文件，将当前库依赖项写入文件以供比较使用。            with open(self.version_library_comparisons_path, "w",                      encoding=self.check_charset(self.requirements_path)) as file:                file.write(read_requirements)

                                version_library_comparisons.txt  # 依赖

aiofiles==0.8.0allure-pytest==2.9.45allure-python-commons==2.9.45asgiref==3.5.1atomicwrites==1.4.0attrs==21.2.0blinker==1.4Brotli==1.0.9certifi==2021.10.8cffi==1.15.0chardet==4.0.0charset-normalizer==2.0.7click==8.1.3colorama==0.4.4colorlog==6.6.0cryptography==36.0.0DingtalkChatbot==1.5.3et-xmlfile==1.1.0execnet==1.9.0Faker==9.8.3Flask==2.0.3h11==0.13.0h2==4.1.0hpack==4.0.0httptools==0.4.0hyperframe==6.0.1idna==3.3iniconfig==1.1.0itchat==1.3.10itsdangerous==2.1.2Jinja2==3.1.2jsonpath==0.82kaitaistruct==0.9ldap3==2.9.1MarkupSafe==2.1.1mitmproxy==8.0.0msgpack==1.0.3multidict==6.0.2openpyxl==3.0.9packaging==21.3passlib==1.7.4pluggy==1.0.0protobuf==3.19.4publicsuffix2==2.20191221py==1.11.0pyasn1==0.4.8pycparser==2.21pydivert==2.1.0PyMySQL==1.0.2pyOpenSSL==21.0.0pyparsing==3.0.6pyperclip==1.8.2pypng==0.0.21PyQRCode==1.2.1pytest==6.2.5pytest-forked==1.3.0pytest-xdist==2.4.0python-dateutil==2.8.2pywin32==304PyYAML==6.0requests==2.26.0requests-toolbelt==0.9.1ruamel.yaml==0.17.21ruamel.yaml.clib==0.2.6sanic==22.3.1sanic-routing==22.3.0six==1.16.0sortedcontainers==2.4.0text-unidecode==1.3toml==0.10.2tornado==6.1urllib3==1.26.7urwid==2.1.2websockets==10.3Werkzeug==2.1.2wsproto==1.1.0xlrd==2.0.1xlutils==2.0.0xlwings==0.27.7xlwt==1.3.0zstandard==0.17.0

                        exceptions.py  # 自定义异常类

                        详细讲解

# MyBaseFailure 类是一个基本的异常类，其他的异常都是它的子类，继承了它的特性和方法，如异常处理和错误信息的输出等。class MyBaseFailure(Exception):    pass# 这个异常类是用来描述 Jsonpath 提取失败的情况。class JsonpathExtractionFailed(MyBaseFailure):    pass# 这个异常类是用来表示由于某些原因找不到其所需的值而导致的“未找到”错误。class NotFoundError(MyBaseFailure):    pass# 这个异常类是 FileNotFoundError 和 NotFoundError 两个异常类的子类。它是用来描述文件未找到时的情况。class FileNotFound(FileNotFoundError, NotFoundError):    pass# 这个异常类是 NotFoundError 类的子类。它是用来描述在处理 SQL 数据时未找到数据表、字段或行的情况。class SqlNotFound(NotFoundError):    pass# 这个异常类是用来描述断言失败的情况，即函数或方法的返回值与预期值不一致的情况。class AssertTypeError(MyBaseFailure):    pass# 这个异常类是用来描述数据获取失败的情况。class DataAcquisitionFailed(MyBaseFailure):    pass# 这个异常类是用来描述变量或函数参数类型错误的情况。class ValueTypeError(MyBaseFailure):    pass# 这个异常类是用来描述发送消息失败的情况，例如在发送邮件或短信时出错。class SendMessageError(MyBaseFailure):    pass# 这个异常类是用来描述在处理请求或操作时未找到所需值的情况。class ValueNotFoundError(MyBaseFailure):    pass

​                          get_local_ip.py  # 获取本地IP

                        详细讲解

import socket# 定义一个名为 get_host_ip 的函数。def get_host_ip():    """    查询本机ip地址    :return:    """    # 初始化一个变量 '_s' ，并将其值置为 None。    _s = None    try:        # # 使用 Python 标准库中的 socket 模块创建一个 socket 对象 _s。其中 socket.AF_INET 表示建立面向 IPv4 的网络套接字， socket.SOCK_DGRAM 表示该套接字是数据报类型。接着，使用 connect 方法将 socket 对象 _s 连接到 ('8.8.8.8', 80) 这个公共 DNS 服务器。然后，使用 getsockname() 方法获取本机的 IP 地址，用变量 l_host 保存。        _s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)        _s.connect(('8.8.8.8', 80))        l_host = _s.getsockname()[0]    # finally 语句块中关闭 socket 连接，确保资源释放。    finally:        _s.close()    # 返回 l_host，本机 IP 地址。    return l_host

                        jsonpath_date_replace.py  # 处理jsonpath数据

                        详细讲解

# 定义一个名为 jsonpath_replace 的函数，该函数接受三个参数，分别是 change_data、 key_name 和 data_switch，用于处理 JSONPath 数据。def jsonpath_replace(change_data, key_name, data_switch=None):    """处理jsonpath数据"""    # 定义 _new_data 变量，初始值为 key_name 与一个空字符串拼接。    _new_data = key_name + ''    # 使用 for 循环遍历 change_data 变量中的每一个元素，i 是遍历时用于找到元素的迭代变量。    for i in change_data:        # 如果检测到 JSONPath 路径中存在 $，则什么都不做，直接跳过。        if i == '$':            pass        # 如果 data_switch 为 None 并且元素为 "data"，则在 _new_data 的后面追加一个字符串 '.data' 。        elif data_switch is None and i == "data":            _new_data += '.data'        # 如果检测到元素是索引（即 i 表示一个列表元素），则在 _new_data 的后面追加一对方括号和 /i[1:-1]/。        elif i[0] == '[' and i[-1] == ']':            _new_data += "[" + i[1:-1] + "]"        # 如果是「非索引形式」的元素，则在 _new_data 的后面追加一对方括号和引号组成的字符串 /"/i/"/（即元素 i 用引号括起来）。        else:            _new_data += '[' + '"' + i + '"' + "]"    # 返回最终的 _new_data 变量，其中包含了处理后的字符串形式的 JSONPath。    return _new_data

                        models.py  # 定义类和枚举变量

                        详细讲解

import typesfrom enum import Enum, uniquefrom typing import Text, Dict, Callable, Union, Optional, List, Anyfrom dataclasses import dataclassfrom pydantic import BaseModel# 定义一个 NotificationType 枚举类。class NotificationType(Enum):    """ 自动化通知方式 """    # 定义了5个枚举常量，分别对应自动化通知的5种方式，每个枚举常量都由名称和值组成。这里的值表示通知方式的代号，名称则表述了该通知方式的语义。    DEFAULT = '0'    DING_TALK = '1'    WECHAT = '2'    EMAIL = '3'    FEI_SHU = '4'# 使用装饰器 @dataclass 定义一个数据类 TestMetrics，该数据类包含了多个测试数据指标。@dataclassclass TestMetrics:    """ 用例执行数据 """    # 使用 docstring 注释对数据类进行描述，表明该数据类用于表示用例执行数据。    passed: int  # 表示通过的测试用例数    failed: int  # 表示失败的测试用例数    broken: int  # 表示中断的测试用例数    skipped: int  # 表示跳过的测试用例数    total: int  # 表示总共的测试用例数    pass_rate: float  # 表示测试通过率    time: Text  # 表示测试执行的时间# 定义一个 RequestType 枚举类。class RequestType(Enum):    """    request请求发送，请求参数的数据类型    """    # 使用 docstring 注释对枚举类进行描述，表明该枚举类用于定义请求参数的数据类型。定义了6个枚举常量，分别对应请求参数的6种数据类型。每个枚举常量都由名称和值组成。这里的值是字符串类型，表示不同的数据类型。    JSON = "JSON"    PARAMS = "PARAMS"    DATA = "DATA"    FILE = 'FILE'    EXPORT = "EXPORT"    NONE = "NONE"# 定义一个 TestCaseEnum 枚举类。class TestCaseEnum(Enum):    # 定义了多个枚举常量，每个枚举常量都是一个元组类型，由两个元素组成。第一个元素表示测试用例中对应的字段名称，第二个元素表示该字段是否必需。    URL = ("url", True)    HOST = ("host", True)    METHOD = ("method", True)    DETAIL = ("detail", True)    IS_RUN = ("is_run", True)    HEADERS = ("headers", True)    REQUEST_TYPE = ("requestType", True)    DATA = ("data", True)    DE_CASE = ("dependence_case", True)    DE_CASE_DATA = ("dependence_case_data", False)    CURRENT_RE_SET_CACHE = ("current_request_set_cache", False)    SQL = ("sql", False)    ASSERT_DATA = ("assert", True)    SETUP_SQL = ("setup_sql", False)    TEARDOWN = ("teardown", False)    TEARDOWN_SQL = ("teardown_sql", False)    SLEEP = ("sleep", False)# 定义一个 Method 枚举类。class Method(Enum):    # 定义了7个枚举常量，每个常量都代表一个 HTTP 请求方法。常量名称即为请求方法名，常量值为对应的字符串类型的请求方法。    GET = "GET"    POST = "POST"    PUT = "PUT"    PATCH = "PATCH"    DELETE = "DELETE"    HEAD = "HEAD"    OPTION = "OPTION"# 定义了一个名为 load_module_functions 的函数，输入参数 module 为要处理的 Python 模块，函数返回值为一个字典类型，其中键为函数名称，值为函数所在的内存地址。def load_module_functions(module) -> Dict[Text, Callable]:    """ 获取 module中方法的名称和所在的内存地址 """    # 先初始化一个名为 module_functions 的空字典。    module_functions = {}    # 使用内置函数 vars() 取出 module 模块中的所有属性，将其转换为字典类型，并使用 for 循环遍历其中的每一项。若遍历到的属性类型为函数类型，则将其加入到 module_functions 字典中，键为函数名称，值为函数所在的内存地址。    for name, item in vars(module).items():        if isinstance(item, types.FunctionType):            module_functions[name] = item    # 将 module_functions 字典作为函数返回值返回。    return module_functions# 定义了一个 DependentType 枚举类，使用装饰器 @unique 来保证枚举常量的唯一性。@uniqueclass DependentType(Enum):    """    数据依赖相关枚举    """    # 定义了4个枚举常量，常量名称即为相关数据的依赖类型，常量值为对应的字符串类型的依赖类型。    RESPONSE = 'response'    REQUEST = 'request'    SQL_DATA = 'sqlData'    CACHE = "cache"# 定义了一个名为 Assert 的数据类，它继承了 BaseModel 类，即 Assert 类具有了 BaseModel 类的所有属性和方法。class Assert(BaseModel):    # 定义了 Assert 类的4个属性。其中，jsonpath 的类型为 Text，表示 JSON 路径；type 的类型为 Text，表示数据的类型；value 的类型为 Any，表示数据的值；AssertType 的类型为 Union[None, Text]，表示断言类型。    jsonpath: Text    type: Text    value: Any    AssertType: Union[None, Text] = None# 此数据类 Assert 的作用是定义一个测试断言模型，其中 jsonpath 表示需要测试的数据在 JSON 文件中的路径，value 表示需要测试的数据的具体值，type 表示需要测试的数据的类型，例如字符串、数字等，AssertType 表示测试断言的类型，例如等于断言、不等断言等。使用数据类可以更加方便地处理这些数据，并在使用时有更好的类型提示和验证。# 定义了一个名为 DependentData 的数据类，它同样继承了 BaseModel 类，即 DependentData 类也具有了 BaseModel 类的所有属性和方法。class DependentData(BaseModel):    # 定义了 DependentData 类的4个属性。其中，dependent_type 的类型为 Text，表示依赖的数据类型；jsonpath 的类型为 Text，表示依赖数据在 JSON 文件中的路径；set_cache 的类型为 Optional[Text]，表示需要将依赖数据缓存，可选参数；replace_key 的类型为 Optional[Text]，表示需要替换的依赖数据在 JSON 文件中的路径，可选参数。    dependent_type: Text    jsonpath: Text    set_cache: Optional[Text]    replace_key: Optional[Text]# 此数据类 DependentData 的作用是定义测试用例中的数据依赖模型，其中 dependent_type 表示数据依赖类型，例如响应数据、请求数据等；jsonpath 表示需要依赖的数据在 JSON 文件中的路径；set_cache 表示依赖数据是否需要进行缓存；replace_key 表示需要替换的依赖数据在 JSON 文件中的路径。使用数据类可以更加方便地处理这些数据，并在使用时有更好的类型提示和验证。# 定义了一个名为 DependentCaseData 的数据类，它继承了 BaseModel 类，即 DependentCaseData 类也具有了 BaseModel 类的所有属性和方法。class DependentCaseData(BaseModel):    # 定义了 DependentCaseData 类的2个属性。其中，case_id 的类型为 Text，表示测试用例的 ID；dependent_data 的类型为 Union[None, List[DependentData]]，表示测试用例所依赖的数据，可选参数，数据类型为 DependentData 列表。    case_id: Text    dependent_data: Union[None, List[DependentData]] = None# 此数据类 DependentCaseData 的作用是定义测试用例所依赖的数据模型，其中 case_id 表示测试用例的 ID，dependent_data 表示测试用例所依赖的数据，可能有多个依赖数据，所以使用列表存储。DependentCaseData 数据类的属性 dependent_data 是一个可选参数，因为可能存在某些测试用例不需要依赖数据的情况。# 定义了一个名为 ParamPrepare 的数据类，它继承了 BaseModel 类，即 ParamPrepare 类也具有了 BaseModel 类的所有属性和方法。class ParamPrepare(BaseModel):    # 定义了 ParamPrepare 类的3个属性。其中，dependent_type 的类型为 Text，表示依赖的数据类型；jsonpath 的类型为 Text，表示需要依赖数据在 JSON 文件中的路径；set_cache 的类型为 Text，表示需要将依赖数据进行缓存。    dependent_type: Text    jsonpath: Text    set_cache: Text# 此数据类 ParamPrepare 的作用是定义 API 的参数准备数据模型，其中 dependent_type 表示参数依赖的数据类型，例如响应数据、请求数据等；jsonpath 表示需要依赖的数据在 JSON 文件中的路径；set_cache 表示依赖数据是否需要进行缓存。使用数据类可以更加方便地处理这些数据，并在使用时有更好的类型提示和验证。# 定义了一个名为 SendRequest 的数据类，它继承了 BaseModel 类，即 SendRequest 类也具有了 BaseModel 类的所有属性和方法。class SendRequest(BaseModel):    # 定义了 SendRequest 类的5个属性。其中，dependent_type 的类型为 Text，表示依赖的数据类型；jsonpath, cache_data, set_cache, 和 replace_key 的类型均为可选的 Optional[Text]。    dependent_type: Text    jsonpath: Optional[Text]    cache_data: Optional[Text]    set_cache: Optional[Text]    replace_key: Optional[Text]# 此数据类 SendRequest 的作用是定义 API 请求数据模型，其中 dependent_type 表示参数依赖的数据类型，例如响应数据、请求数据等；jsonpath 表示需要从依赖数据中提取的数据在 JSON 文件中的路径，由于该属性为可选参数，所以不一定所有的请求都需要依赖数据；cache_data 表示需要缓存的数据；set_cache 表示缓存的数据是否需要进行缓存，并可以为空；replace_key 表示需要替换的数据键名。# 定义了一个名为 TearDown 的数据类，它继承了 BaseModel 类，即 TearDown 类也具有了 BaseModel 类的所有属性和方法。class TearDown(BaseModel):    # 定义了 TearDown 类的3个属性。其中，case_id 的类型为 Text，表示测试用例的 ID；param_prepare 和 send_request 的类型均为可选参数的列表类型，其中 param_prepare 表示参数准备数据模型的列表，send_request 表示 API 请求数据模型的列表。    case_id: Text    param_prepare: Optional[List["ParamPrepare"]]    send_request: Optional[List["SendRequest"]]# 此数据类 TearDown 的作用是定义 API 测试用例的后置条件数据模型，其中 case_id 表示测试用例的 ID；param_prepare 表示测试用例后置条件中所需的参数准备数据模型列表，它是可选参数，因为不是所有的测试用例都需要参数准备；send_request 则表示测试用例后置条件中 API 请求数据模型的列表，同样是可选参数。使用数据类可以更加方便地处理这些数据，并在使用时有更好的类型提示和验证，从而减少代码出错的可能。# 定义了一个名为 CurrentRequestSetCache 的数据类，它继承了 BaseModel 类，即 CurrentRequestSetCache 类也具有了 BaseModel 类的所有属性和方法。class CurrentRequestSetCache(BaseModel):    # 定义了 CurrentRequestSetCache 类的3个属性。其中，type、jsonpath 和 name 的类型均为 Text，其中 type 表示当前缓存的类型，具体包括 response 和 request；jsonpath 表示需要从返回结果中提取的数据在 JSON 文件中的路径；name 表示需要缓存的数据的名称。    type: Text    jsonpath: Text    name: Text# 此数据类 CurrentRequestSetCache 的作用是定义 API 请求数据模型中当前请求需要缓存的数据模型，其中 type 表示缓存的类型，具体包括响应和请求；jsonpath 表示需要从响应结果或请求参数中提取的数据的路径，它是需要自己定义的；name 则表示缓存数据的名称。使用数据类可以更加方便地处理这些数据，并在使用时有更好的类型提示和验证，从而减少代码出错的可能。# 定义了一个名为 TestCase 的数据类，它同样是继承自 BaseModel 类，即 TestCase 类也具有了 BaseModel 类的所有属性和方法。class TestCase(BaseModel):    url: Text  # 表示 API 的请求地址    method: Text  # 表示请求方式，例如 GET、POST 等    detail: Text  # 表示测试用例的详细信息    assert_data: Union[Dict, Text]  # 表示需要进行断言的数据，可能为字典类型，也可能为文本类型    headers: Union[None, Dict, Text] = {}  # 表示请求头，它是可选参数，默认值为空字典    requestType: Text  # 表示请求的数据类型，例如 json、form data 等    is_run: Union[None, bool, Text] = None  # 表示是否需要运行该测试用例，它是可选参数，默认值为 None    data: Any = None  # 表示请求体数据，它是可选参数    dependence_case: Union[None, bool] = False  # 表示该测试用例是否依赖其他用例，默认为 False    dependence_case_data: Optional[Union[None, List["DependentCaseData"], Text]] = None  # 表示该测试用例依赖的其他用例的数据列表，默认为 None    sql: List = None  # 表示该测试用例执行前需要执行的 SQL 语句列表，它是可选参数    setup_sql: List = None  # 表示该测试用例执行完后需要执行的 SQL 语句列表，它是可选参数    status_code: Optional[int] = None  # 表示请求返回的状态码，它是可选参数    teardown_sql: Optional[List] = None  # 表示该测试用例执行完后需要执行的 SQL 语句列表，它是可选参数    teardown: Union[List["TearDown"], None] = None  # 表示模拟测试数据的后置条件    current_request_set_cache: Optional[List["CurrentRequestSetCache"]]  # 表示缓存当前请求需要的数据    sleep: Optional[Union[int, float]]  # 表示请求完后需要等待的时间，它是可选参数# 定义了一个名为 ResponseData 的数据类，它同样是继承自 BaseModel 类，即 ResponseData 类也具有了 BaseModel 类的所有属性和方法。class ResponseData(BaseModel):    url: Text  # 表示 API 的请求地址    is_run: Union[None, bool, Text]  # 表示是否需要运行该测试用例，它是可选参数，默认值为 None    detail: Text  # 表示测试用例的详细信息    response_data: Text  # 表示响应结果的文本内容    request_body: Any  # 表示请求体数据，它是可选参数    method: Text  # 表示请求方式，例如 GET、POST 等    sql_data: Dict  # 表示数据库执行后的结果数据，它是一个字典    yaml_data: "TestCase"  # 表示该测试用例的详细信息，是一个 TestCase 实例    headers: Dict  # 表示响应结果的请求头    cookie: Dict  # 表示响应结果的 cookie 信息    assert_data: Dict  # 表示需要进行断言的数据，是一个字典    res_time: Union[int, float]  # 表示响应时间    status_code: int  # 表示请求返回的状态码    teardown: List["TearDown"] = None  # 表示模拟测试数据的后置条件    teardown_sql: Union[None, List]  # 表示该测试用例执行完后需要执行的 SQL 语句列表，它是可选参数    body: Any  # 表示响应结果的请求体数据# 定义了一个名为 DingTalk 的数据类，它同样是继承自 BaseModel 类，即 DingTalk 类也具有了 BaseModel 类的所有属性和方法。class DingTalk(BaseModel):    # 定义了 DingTalk 类的两个属性    webhook: Union[Text, None]  # 表示钉钉机器人的 Webhook 地址，是一个文本类型的可选参数，默认值为 None    secret: Union[Text, None]  # 表示加签的密钥，是一个文本类型的可选参数，默认值为 None# 这个数据类可以用来存储钉钉机器人的相关信息，方便程序中的调用和使用。由于这两个属性都是可选参数，所以在使用时需要注意判断是否为空。# 定义了一个名为 MySqlDB 的数据类，它同样是继承自 BaseModel 类，即 MySqlDB 类也具有了 BaseModel 类的所有属性和方法。class MySqlDB(BaseModel):    switch: bool = False  # 表示该数据库是否需要连接，是一个布尔类型的可选参数，默认值为 False    host: Union[Text, None] = None  # 表示数据库的地址，是一个文本类型的可选参数，默认值为 None    user: Union[Text, None] = None  # 表示连接数据库的用户名，是一个文本类型的可选参数，默认值为 None    password: Union[Text, None] = None  # 表示连接数据库的密码，是一个文本类型的可选参数，默认值为 None    port: Union[int, None] = 3306  # 表示连接数据库的端口，是一个整型的可选参数，默认值为 3306# 这个数据类可以用来存储 MySQL 数据库的相关信息，方便程序中的调用和使用。由于所有属性都是可选参数，所以在使用时需要注意判断是否为空。此外，通过在属性定义中设置默认值，可以方便地进行参数配置。# 定义了一个名为 Webhook 的数据类，它同样是继承自 BaseModel 类，即 Webhook 类也具有了 BaseModel 类的所有属性和方法。class Webhook(BaseModel):    webhook: Union[Text, None]  # 表示 Webhook 地址，是一个文本类型的可选参数，默认值为 None# 这个数据类可以用来存储 Webhook 的相关信息，方便程序中的调用和使用。由于 webhook 属性是一个可选参数，所以在使用时需要注意判断是否为空。# 定义了一个名为 Email 的数据类，它同样是继承自 BaseModel 类，即 Email 类也具有了 BaseModel 类的所有属性和方法。class Email(BaseModel):    send_user: Union[Text, None]  # 表示发件人邮箱的用户名，是一个文本类型的可选参数，默认值为 None    email_host: Union[Text, None]  # 表示邮件服务器的地址，是一个文本类型的可选参数，默认值为 None    stamp_key: Union[Text, None]  # 表示发件人的密码或授权码，是一个文本类型的可选参数，默认值为 None    send_list: Union[Text, None]  # 表示邮件的接收人列表，是一个文本类型的可选参数，默认值为 None# 这个数据类可以用来存储发送邮件的相关信息，方便程序中的调用和使用。由于所有属性都是可选参数，所以在使用时需要注意判断是否为空。此外，通过在属性定义中设置默认值，可以方便地进行参数配置。# 定义了一个名为 Config 的数据类，它同样是继承自 BaseModel 类，即 Config 类也具有了 BaseModel 类的所有属性和方法。class Config(BaseModel):    project_name: Text  # 表示项目的名称，是一个文本类型的必选参数    env: Text  # 表示运行的环境，是一个文本类型的必选参数    tester_name: Text  # 表示测试人员的姓名，是一个文本类型的必选参数    notification_type: Text = '0'  # 表示通知的方式，是一个文本类型的可选参数，它的默认值为 '0'    excel_report: bool  # 表示是否生成 Excel 报告，是一个布尔类型的必选参数    ding_talk: "DingTalk"  # 表示钉钉机器人的相关信息，是一个 DingTalk 类的实例，是一个必选参数    mysql_db: "MySqlDB"  # 表示 MySQL 数据库的相关信息，是一个 MySqlDB 类的实例，是一个必选参数    mirror_source: Text  # 表示测试用例所在的镜像源，是一个文本类型的必选参数    wechat: "Webhook"  # 表示微信机器人的相关信息，是一个 Webhook 类的实例，是一个必选参数    email: "Email"  # 表示邮件相关信息，是一个 Email 类的实例，是一个必选参数    lark: "Webhook"  # 表示飞书机器人的相关信息，是一个 Webhook 类的实例，是一个必选参数    real_time_update_test_cases: bool = False  # 表示是否实时更新测试用例，是一个布尔类型的可选参数，它的默认值为 False    host: Text  # 表示测试服务器的地址，是一个文本类型的必选参数    app_host: Union[Text, None]  # 表示被测应用程序的地址，是一个文本类型的可选参数，默认值为 None# 这个数据类封装了一个测试框架的配置信息，包括项目名称、测试环境、测试人员信息、通知方式、测试用例相关信息、被测应用程序相关信息等。可以通过实例化 Config 类来快速配置测试框架的相关信息。由于部分属性有默认值，因此在实例化时可以只传递必选参数。# 定义了一个名为 AllureAttachmentType 的枚举类，它继承自 Python 内置的 Enum 类，表示这是一个枚举类。同时使用 @unique 修饰符确保枚举类成员的值各不相同。@uniqueclass AllureAttachmentType(Enum):    """    allure 报告的文件类型枚举    """    # 定义了一个包含 19 个成员的枚举，每个成员表示一种 Allure 报告中的文件类型。其中，每个成员都由一个名称和一个值构成，名称即为成员名称（如 TEXT、CSV），而值则为成员的实际取值（如 "text"、"csv"）。枚举中的所有成员都是不可变的，例如不能重新赋值或删除成员。    TEXT = "txt"    CSV = "csv"    TSV = "tsv"    URI_LIST = "uri"    HTML = "html"    XML = "xml"    JSON = "json"    YAML = "yaml"    PCAP = "pcap"    PNG = "png"    JPG = "jpg"    SVG = "svg"    GIF = "gif"    BMP = "bmp"    TIFF = "tiff"    MP4 = "mp4"    OGG = "ogg"    WEBM = "webm"    PDF = "pdf"# 这个枚举类可以用于定义 Allure 报告中的文件类型，所以在测试框架中被广泛使用。它可以帮助开发人员、测试人员和其他相关人员快速识别和解析 Allure 报告中包含的文件类型。@uniqueclass 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'    # 去重无序判断是否相等    setlist = 'setlist'    # list转成str 去重无序判断是否相等    str_set = 'str_set'    # 可散列判断    counter = 'counter'

                        thread_tool.py  # 定时器类

                        详细讲解

import timeimport threadingclass PyTimer:    """定时器类"""        # 定义了一个 __init__ 方法作为该类的构造函数，接收一个函数 func 和一些参数和关键字参数 *args、**kwargs。并且初始化了 4 个属性    def __init__(self, func, *args, **kwargs):        """构造函数"""        self.func = func  # 表示需要定时执行的函数        self.args = args  # 表示传递给需要定时执行的函数的位置参数或参数元组        self.kwargs = kwargs  # 表示传递给需要定时执行的函数的关键字参数或参数字典        self.running = False  # 表示定时器是否正在运行的标志，初始化为 False    # 定义了一个 _run_func 实例方法。    def _run_func(self):        """运行定时事件函数"""        # 使用 threading.Thread 方法创建一个新的线程对象 _thread，并将 self.func 函数作为该线程的目标函数，位置参数和关键字参数分别为 self.args 和 self.kwargs。        _thread = threading.Thread(target=self.func, args=self.args, kwargs=self.kwargs)        # 将 _thread 线程设置为守护线程，即子线程在主线程结束时会自动退出。        _thread.setDaemon(True)        # 启动 _thread 线程，开始执行定时事件函数。        _thread.start()    # 这个方法实现了在一个新线程中运行定时事件函数。使用多线程可以在程序启动后，让程序在后台运行不被阻塞，提高程序的响应能力。    # 定义了一个名为 _start 的定时器线程函数，包含参数 interval 和 once。interval 是任务执行的时间间隔， once 是一个布尔值，表示任务在定时完成后是否仅执行一次。    def _start(self, interval, once):        """启动定时器的线程函数"""        # 首先使用 max 函数将 interval 设定为 0.01 或更大的值。这是因为当 interval 的值太小可能会导致执行过程中的误差。        interval = max(interval, 0.01)        # 根据 interval 的大小， _dt 的值被设定为最小的执行时间。如果 interval 小于 0.05 秒，则 _dt 值为1/10倍的 interval。        if interval < 0.050:            _dt = interval / 10        # 否则， _dt 值为0.005秒。        else:            _dt = 0.005        # 如果任务只要单次执行，该代码块会先计算到达期限时间 deadline，等待直到时间到达 deadline（通过不断的 time.sleep(dt)），然后执行 self._run_func() 方法，以执行任务。        if once:            deadline = time.time() + interval            while time.time() < deadline:                time.sleep(_dt)            # 定时时间到，调用定时事件函数            self._run_func()        # 如果任务需要重复执行，该代码块会设置 self.running = True，然后在一个循环中重复执行任务。在循环中，该代码块先计算下一个到期时间 deadline，然后等待时间直到到期（通过不断的time.sleep(dt)）。然后，为下一个定时执行更新 deadline 的值。最后，该代码块调用 _run_func() 方法来执行任务。        else:            # 当 self.running 被设置为 False 时，定时器将再次暂停，任务将不会被重复执行。            self.running = True            deadline = time.time() + interval            while self.running:                while time.time() < deadline:                    time.sleep(_dt)                # 更新下一次定时时间                deadline += interval                # 定时时间到，调用定时事件函数                if self.running:                    self._run_func()    # 这是启动定时器的外部函数，在类中设定了 interval、once 其默认值为 False。    def start(self, interval, once=False):        """启动定时器        interval    - 定时间隔，浮点型，以秒为单位，最高精度10毫秒        once        - 是否仅启动一次，默认是连续的        """        # thread_ 是定义的一个线程对象，这里使用 threading.Thread 函数从 _start 中创建一个新线程。        thread_ = threading.Thread(target=self._start, args=(interval, once))        # 将线程设置为守护线程。在 Python 中，当一个线程为守护线程时，在主线程退出时，守护线程会自动退出。这可以确保定时器线程在主线程结束时自动结束。        thread_.setDaemon(True)        # 启动新线程，让定时器开始工作。线程开始执行 _start 方法，在该方法中，根据设定好的 interval 和 once参数，对定时器的执行进行相应设定和控制。        thread_.start()    def stop(self):        """停止定时器"""        # 是 running 属性的赋值操作，将类属性 self.running 设为 False，用于控制定时器的运行。        self.running = False    # 这个方法非常简单，它的作用就是让定时器主线程停止运行，因为在定时器的主线程中，使用了 while self.running: 循环，当 self.running 为 False 时，循环就会退出，从而实现停止定时器的目的。# 定义了一个函数 do_something，函数接收两个参数，其中 name 是必须传递的参数，而 gender 是可选参数，如果不传递则默认为 'male'。def do_something(name, gender='male'):    """执行"""    # 使用 time.time() 函数获取当前时间戳，并输出一个提示信息，表示在定时任务到达时，需要执行特定的任务。    print(time.time(), '定时时间到，执行特定任务')    # 使用字符串的 format 方法对字符串进行格式化输出，输出传入的 name 和 gender，其中 %s 是占位符。    print('name:%s, gender:%s', name, gender)    # 使用 time 模块的 sleep 函数使得程序暂停 5 秒，用于模拟执行一些需要耗费时间的操作。    time.sleep(5)    # 输出当前时间戳和一个提示信息，表示特定任务已经完成。    print(time.time(), '完成特定任务')# 定义了一个 PyTimer 对象，并传入了三个参数，分别是 do_something 函数、字符串 'Alice' 和字典 {'gender': 'female'}，其中 'Alice' 和 'female' 分别对应了 do_something 中的 name 和 gender 两个参数。timer = PyTimer(do_something, 'Alice', gender='female')# 调用了定时器对象 timer 的 start() 方法，开始了定时器操作。其中，第一个参数 0.5 表示每隔 0.5 秒执行一次任务，第二个参数 once=False 表示定时器会一直执行，而不是只执行一次。timer.start(0.5, once=False)# 使用 input 函数实现的阻塞操作，等待用户按下回车键，以便之后停止定时器操作。input('按回车键结束\n')  # 此处阻塞住进程# 调用了定时器对象 timer 的 stop() 方法，停止了定时器的操作。timer.stop()

                read_files_tools  # 读取文件工具

                        case_automatic_control.py  # 自动生成测试代码

                        详细讲解

import osfrom typing import Textfrom common.setting import ensure_path_sepfrom utils.read_files_tools.testcase_template import write_testcase_filefrom utils.read_files_tools.yaml_control import GetYamlDatafrom utils.read_files_tools.get_all_files_path import get_all_files# 定义了一个名为 TestCaseAutomaticGeneration 的类，用于自动生成测试用例代码。class TestCaseAutomaticGeneration:    # 类的 __init__() 方法中初始化了 self.yaml_case_data 和 self.file_path，它们分别用于存储读取的测试用例数据和测试用例文件路径。    def __init__(self):        # 它是一个成员变量，用于存储 YAML 格式的测试用例数据。        self.yaml_case_data = None        # 它是一个成员变量，用于存储测试用例数据文件的路径。        self.file_path = None    # @property 可以将一个方法转换成一个属性，可以像访问普通属性一样，使用 obj.case_date_path 的方式访问该属性。    @property    # 定义了一个名为 case_date_path 的属性，并使用了 ensure_path_sep("\\data") 方法返回一个格式化后的路径字符串。    def case_date_path(self) -> Text:        """返回 yaml 用例文件路径"""        return ensure_path_sep("\\data")    @property    # 定义了一个名为 case_path 的属性，它指定了测试用例代码存放的路径，并使用了 ensure_path_sep("\\test_case") 方法对路径进行了格式化。    def case_path(self) -> Text:        """ 存放用例代码路径"""        return ensure_path_sep("\\test_case")    @property    # 定义了一个名为 allure_epic 的属性，它用于获取指定的 YAML 文件中的测试数据中的 allureEpic 字段。如果该字段不存在或为空，则代码会抛出一个 AssertionError，其中包含了有关错误的详细信息，例如文件路径和字段名称，并提示用户进行检查。    def allure_epic(self):        # 从当前实例的 yaml_case_data 属性中获取 case_common 键所对应的值中的 allureEpic 键所对应的值，并将其赋值给 _allure_epic 变量。如果没有找到或者为空，那么 _allure_epic 将会是 None。        _allure_epic = self.yaml_case_data.get("case_common").get("allureEpic")        # 检查 _allure_epic 是否为空。如果为空，则使用 assert 关键字抛出一个断言错误，其中包含了错误信息。该错误信息包含了文件路径和缺失的字段名称。        assert _allure_epic is not None, (                "用例中 allureEpic 为必填项，请检查用例内容, 用例路径：'%s'" % self.file_path        )        # 返回 _allure_epic 的值作为这个属性的值，也就是当前 YAML 文件中的 allureEpic 字段的值。        return _allure_epic    @property    def allure_feature(self):        # 从 yaml_case_data 属性中获取当前 YAML 文件中的 case_common 键所对应的值中的 allureFeature 键所对应的值，并将其赋值给 _allure_feature 变量。如果没有找到或者为空，那么 _allure_feature 将会是 None。        _allure_feature = self.yaml_case_data.get("case_common").get("allureFeature")        # 检查 _allure_feature 是否为空。如果为空，则使用 assert 关键字抛出一个断言错误，其中包含了错误信息。该错误信息包含了文件路径和缺失的字段名称。        assert _allure_feature is not None, (                "用例中 allureFeature 为必填项，请检查用例内容, 用例路径：'%s'" % self.file_path        )        # 返回 _allure_feature 的值作为这个属性的值，也就是当前 YAML 文件中的 allureFeature 字段的值。        return _allure_feature    @property    def allure_story(self):        # 从 yaml_case_data 属性中获取当前 YAML 文件中的 case_common 键所对应的值中的 allureStory 键所对应的值，并将其赋值给 _allure_story 变量。如果没有找到或者为空，那么 _allure_story 将会是 None。        _allure_story = self.yaml_case_data.get("case_common").get("allureStory")        # 检查 _allure_story 是否为空。如果为空，则使用 assert 关键字抛出一个断言错误，其中包含了错误信息。该错误信息包含了文件路径和缺失的字段名称。        assert _allure_story is not None, (                "用例中 allureStory 为必填项，请检查用例内容, 用例路径：'%s'" % self.file_path        )        # 返回 _allure_story 的值作为这个属性的值，也就是当前 YAML 文件中的 allureStory 字段的值。        return _allure_story    @property    def file_name(self) -> Text:        """        通过 yaml文件的命名，将名称转换成 py文件的名称        :return:  示例： DateDemo.py        """        # 获取 self.file_path 属性中除去 self.case_date_path 前缀的部分，并将其赋值给 yaml_path 变量。        i = len(self.case_date_path)        yaml_path = self.file_path[i:]        # 这里初始化了一个 file_name 变量为 None。        file_name = None        # if-elif 循环检查 yaml_path 变量中是否包含 .yaml 或 .yml 字符串，如果包含，则将其替换成 .py 字符串，并将结果赋值给 file_name 变量。路径转换        if '.yaml' in yaml_path:            file_name = yaml_path.replace('.yaml', '.py')        elif '.yml' in yaml_path:            file_name = yaml_path.replace('.yml', '.py')        # 返回 file_name 变量的值作为这个方法的返回值。        return file_name    @property    def get_test_class_title(self):        """        自动生成类名称        :return: sup_apply_list --> SupApplyList        """        # 获取 self.file_name 中的文件名，去除了 .py 后缀，并将结果赋值给 _file_name 变量。        _file_name = os.path.split(self.file_name)[1][:-3]        # 使用 _file_name 的值初始化了一个 _name 列表，并且获取了 _name 的长度。        _name = _file_name.split("_")        _name_len = len(_name)        # for 循环将 _name 中的每个单词首字母转换成大写字母。将文件名称格式，转换成类名称: sup_apply_list --> SupApplyList        for i in range(_name_len):            _name[i] = _name[i].capitalize()        # 将 _name 中的所有单词合并成一个字符串，赋值给 _class_name 变量。        _class_name = "".join(_name)                # 返回 _class_name 变量的值作为这个方法的返回值，也就是测试用例所对应的类的名称。        return _class_name    @property    def func_title(self) -> Text:        """        函数名称        :return:        """        # 返回了 self.file_name 中的文件名，去除了 .py 后缀，作为函数的返回值。因为 self.file_name 中已经是 .py 后缀的文件名了，所以返回值就是一个函数名。        return os.path.split(self.file_name)[1][:-3]    @property    def spilt_path(self):        # 使用 os.sep 分隔符将路径字符串拆分成列表，把结果赋值给新的 path 变量。        path = self.file_name.split(os.sep)        # 将列表中最后一个元素修改为 "test_" + path[-1]，也就是在该文件名前面添加 "test_"。        path[-1] = path[-1].replace(path[-1], "test_" + path[-1])        # spilt_path 方法返回修改后的 path 列表，也即是测试用例文件的所在路径。        return path    @property    def get_case_path(self):        """        根据 yaml 中的用例，生成对应 testCase 层代码的路径        :return: D:\\Project\\test_case\\test_case_demo.py        """        # 将 spilt_path 方法返回的路径列表加上分隔符，生成新路径字符串。比如，如果原路径是 ['home', 'user', 'Documents', 'test.py']，那么这一行代码返回的结果就是 'home/user/Documents/test.py'。        new_name = os.sep.join(self.spilt_path)        # 调用了 ensure_path_sep 方法，确保路径字符串最后一个字符是路径分隔符。然后，将字符串 "\test_case" 与新生成的路径使用 + 拼接，并返回该字符串。比如，如果原路径是 D:\Project\test.py，那么这一行代码返回的结果就是 D:\Project\test_case\test.py。        return ensure_path_sep("\\test_case" + new_name)    @property    def case_ids(self):        # 使用列表推导式，遍历 yaml_case_data （即测试数据）中所有的键，如果该键不等于 "case_common"，那么返回该键。也就是说，这里会返回一个列表，表示测试数据中所有测试用例的 id 字段。        return [k for k in self.yaml_case_data.keys() if k != "case_common"]    @property    def get_file_name(self):        # 判断生成的 testcase 文件名称，需要以test_ 开头        # 根据 spilt_path 方法返回的文件路径，提取出文件名称。然后，使用 replace 方法将文件名替换为以 "test_" 开头的新文件名。比如，如果原文件名是 sample.py，那么这一行代码返回的结果就是 test_sample.py。        case_name = self.spilt_path[-1].replace(            self.spilt_path[-1], "test_" + self.spilt_path[-1]        )        # 返回新的测试用例 Python 文件名。        return case_name    def mk_dir(self) -> None:        """ 判断生成自动化代码的文件夹路径是否存在，如果不存在，则自动创建 """        # 调用 os.path.split 方法，将 get_case_path (测试数据文件路径) 分解成目录路径和文件名，这样就可以提取出测试数据文件所在的目录路径了。        _case_dir_path = os.path.split(self.get_case_path)[0]        # 通过 os.path.exists 方法判断目录路径是否存在，如果不存在就使用 os.makedirs 方法创建目录。也就是说，这个函数的作用是创建测试数据文件所在目录的子目录，用于存储生成的测试用例代码文件。        if not os.path.exists(_case_dir_path):            os.makedirs(_case_dir_path)    def get_case_automatic(self) -> None:        """ 自动生成 测试代码"""        # 使用 get_all_files 方法获取测试数据文件夹下所有的 YAML 文件路径，并将结果存储到 file_path 变量中。这些 YAML 文件将作为测试用例的源数据。        file_path = get_all_files(file_path=ensure_path_sep("\\data"), yaml_data_switch=True)                # 使用 for 循环遍历所有的测试数据文件。        for file in file_path:            # if 语句判断当前文件不是代理拦截的 YAML 文件。如果是，则不生成测试用例代码。            if 'proxy_data.yaml' not in file:                # 这里调用了 GetYamlData 类的 get_yaml_data 方法，获取 YAML 文件的数据内容，并将其保存到 yaml_case_data 变量中。                self.yaml_case_data = GetYamlData(file).get_yaml_data()                # 将当前文件路径保存到 file_path 变量中                self.file_path = file                # 调用 mk_dir 方法，在测试用例代码所需的目录路径不存在时创建目录。                self.mk_dir()                # 这里调用 write_testcase_file 函数，生成测试用例 Python 文件。                write_testcase_file(                    allure_epic=self.allure_epic,                    allure_feature=self.allure_feature,                    class_title=self.get_test_class_title,                    func_title=self.func_title,                    case_path=self.get_case_path,                    case_ids=self.case_ids,                    file_name=self.get_file_name,                    allure_story=self.allure_story                    )if __name__ == '__main__':    # 创建了 TestCaseAutomaticGeneration 类的一个实例，并调用 get_case_automatic 方法，来自动生成测试用例代码。也就是说，当你直接运行这个脚本的时候，就会自动生成测试用例代码。    TestCaseAutomaticGeneration().get_case_automatic()

                        clean_files.py  # 清理文件

                        详细讲解

import os# 函数定义，输入参数为要删除的路径名。def del_file(path):    """删除目录下的文件"""    # 获取指定路径下的所有子文件和子目录，并存储到 list_path 变量中。    list_path = os.listdir(path)    # 依次遍历 list_path 中的每个目录和文件，将它们的绝对路径存储到 c_path 变量中。    for i in list_path:        c_path = os.path.join(path, i)        # 如果当前路径 c_path 是一个目录，则递归调用 del_file 函数，继续删除目录中的文件和子目录。        if os.path.isdir(c_path):            del_file(c_path)        # 如果当前路径 c_path 是一个文件，则直接删除它。        else:            os.remove(c_path)# 这个函数会递归地删除指定的路径下所有的文件和子目录，直到删除完毕并返回。

                        excel_control.py  # 读写excel​​​​​​

                        详细讲解

import jsonimport xlrdfrom xlutils.copy import copyfrom common.setting import ensure_path_sep# 函数名为get_excel_data，参数为sheet_name和case_name，返回值为列表类型list。def get_excel_data(sheet_name: str, case_name: any) -> list:    """    读取 Excel 中的数据    :param sheet_name: excel 中的 sheet 页的名称    :param case_name: 测试用例名称    :return:    """    # 创建一个空列表res_list，用于存储读取Excel中的数据。    res_list = []    # 用于将路径中的斜杠转换为对应系统的分隔符。这里将Excel文件所在的路径作为参数传入，并将结果赋值给excel_dire变量。    excel_dire = ensure_path_sep("\\data\\TestLogin.xls")    # 使用xlrd库打开Excel文件，并将结果赋值给work_book变量。formatting_info=True表示保留Excel中的格式信息。    work_book = xlrd.open_workbook(excel_dire, formatting_info=True)    # 获取指定名称的子表（sheet）并将结果赋值给work_sheet变量。    work_sheet = work_book.sheet_by_name(sheet_name)    # 初始化变量idx为0，用于记录行数。    idx = 0    # 使用col_values方法获取第一列的所有值，并遍历每一个值。    for one in work_sheet.col_values(0):        # 判断当前行是否包含指定的测试用例名称，如果是，则继续执行下一步。        if case_name in one:             #获取当前行第10列（索引从0开始）的值，并将结果赋值给req_body_data变量，用于存储请求体数据。            req_body_data = work_sheet.cell(idx, 9).value            # 获取当前行第12列（索引从0开始）的值，并将结果赋值给resp_data变量，用于存储响应数据。            resp_data = work_sheet.cell(idx, 11).value            # 将请求体数据和响应数据组成一个元组，并添加到res_list列表中。            res_list.append((req_body_data, json.loads(resp_data)))        # 行数加1，继续读取下一行的数据。        idx += 1    # 返回读取到的Excel数据列表。    return res_list# 函数名为set_excel_data，参数为sheet_index，返回值为元组类型tuple。def set_excel_data(sheet_index: int) -> tuple:    """    excel 写入    :return:    """    # 用于指定Excel文件的路径。这里使用相对路径来指定文件所在的位置。    excel_dire = '../data/TestLogin.xls'    # 使用xlrd库打开Excel文件，并将结果赋值给work_book变量。formatting_info=True表示保留Excel中的格式信息。    work_book = xlrd.open_workbook(excel_dire, formatting_info=True)    # 使用xlutils.copy库的copy函数复制一份Excel文件，并将结果赋值给work_book_new变量。这里使用xlutils库是因为xlrd库只能读取Excel文件，不能进行写入操作。    work_book_new = copy(work_book)    # 获取指定索引的子表（sheet）并将结果赋值给work_sheet_new变量，用于写入数据    work_sheet_new = work_book_new.get_sheet(sheet_index)    # 返回复制后的Excel文件和要写入数据的子表（sheet）。    return work_book_new, work_sheet_new

                        get_all_files_path.py  # 获取所有文件路径

                        详细讲解

import os# 函数名为get_all_files，参数为file_path和yaml_data_switch，返回值为列表类型list。def get_all_files(file_path, yaml_data_switch=False) -> list:    """    获取文件路径    :param file_path: 目录路径    :param yaml_data_switch: 是否过滤文件为 yaml格式， True则过滤    :return:    """    # 创建一个空列表filename，用于存储文件路径。    filename = []    # 使用os.walk函数遍历指定路径下的所有文件和文件夹，并获取每一个子目录的路径、子目录列表和文件列表。    for root, dirs, files in os.walk(file_path):        # 遍历文件列表中的每一个文件路径。        for _file_path in files:            # 使用os.path.join函数将当前文件所在目录的路径和文件名拼接成完整的文件路径，并赋值给path变量。            path = os.path.join(root, _file_path)            # 判断yaml_data_switch参数是否为True，如果是，则继续执行下一步。            if yaml_data_switch:                # 判断文件路径中是否包含yaml或.yml，如果是，则将文件路径添加到filename列表中。                if 'yaml' in path or '.yml' in path:                    filename.append(path)            # 如果yaml_data_switch参数为False，则将文件路径添加到filename列表中。            else:                filename.append(path)    # 返回所有文件的路径列表。    return filename

                        get_yaml_data_analysis.py  # yaml用例数据清洗

                        详细讲解

from typing import Union, Text, Listfrom utils.read_files_tools.yaml_control import GetYamlDatafrom utils.other_tools.models import TestCasefrom utils.cache_process.cache_control import CacheHandlerfrom utils import configfrom utils.other_tools.models import RequestType, Method, TestCaseEnumimport osclass CaseDataCheck:    """    yaml 数据解析, 判断数据填写是否符合规范    """        # 类的构造方法，接收一个参数file_path，用于指定yaml文件的路径。    def __init__(self, file_path):        # 将传入的file_path赋值给实例变量self.file_path，以便在类的其他方法中使用。        self.file_path = file_path        # 判断指定的文件路径是否存在，如果不存在则抛出FileNotFoundError异常。        if os.path.exists(self.file_path) is False:            raise FileNotFoundError("用例地址未找到")        # 初始化一个实例变量self.case_data，用于存储解析后的yaml数据。        self.case_data = None        # 初始化一个实例变量self.case_id，用于存储当前解析的用例ID。        self.case_id = None    # 定义一个名为_assert的方法，接收一个参数attr，用于指定需要判断的参数是否存在。    def _assert(self, attr: Text):        # 判断指定的参数是否在yaml数据的keys中存在，如果不存在则抛出异常，提示缺少该参数。        assert attr in self.case_data.keys(), (            f"用例ID为 {self.case_id} 的用例中缺少 {attr} 参数，请确认用例内容是否编写规范."            f"当前用例文件路径：{self.file_path}"        )    # 定义一个名为check_params_exit的方法，用于检查yaml数据中是否缺少必要的参数。    def check_params_exit(self):        # 遍历TestCaseEnum枚举中所有的成员。        for enum in list(TestCaseEnum._value2member_map_.keys()):            # 判断枚举成员是否是必要的参数。            if enum[1]:                # 调用_assert方法，判断当前枚举成员对应的参数是否存在。                self._assert(enum[0])    # 定义一个名为check_params_right的方法，接收两个参数enum_name和attr，用于检查参数是否填写正确。    def check_params_right(self, enum_name, attr):        # 获取枚举成员的所有名称。        _member_names_ = enum_name._member_names_        # 判断枚举成员是否包含指定的参数，如果不包含则抛出异常，提示参数填写错误。        assert attr.upper() in _member_names_, (            f"用例ID为 {self.case_id} 的用例中 {attr} 填写不正确，"            f"当前框架中只支持 {_member_names_} 类型."            f"如需新增 method 类型，请联系管理员."            f"当前用例文件路径：{self.file_path}"        )        # 返回大写的参数名称。        return attr.upper()    # 装饰器，将方法转换成属性，以便在调用时不需要使用括号。    @property    # 定义一个名为get_method的属性方法，用于获取yaml数据中的请求方法。    def get_method(self) -> Text:        # 调用check_params_right方法，检查请求方法是否填写正确。        return self.check_params_right(            # 枚举类型，指定参数类型为请求方法。            Method,            # 从yaml数据中获取请求方法。            self.case_data.get(TestCaseEnum.METHOD.value[0])        )    @property    # 定义一个名为get_host的属性方法，用于获取完整的请求地址。    def get_host(self) -> Text:        # 拼接请求地址，包含主机地址和请求路径。        host = (                # 从yaml数据中获取主机地址。                self.case_data.get(TestCaseEnum.HOST.value[0]) +                # 从yaml数据中获取请求路径。                self.case_data.get(TestCaseEnum.URL.value[0])        )        # 返回完整的请求地址。        return host    @property    # 定义一个名为get_request_type的属性方法，用于获取yaml数据中的请求类型。    def get_request_type(self):        # 调用check_params_right方法，检查请求类型是否填写正确。        return self.check_params_right(            # 枚举类型，指定参数类型为请求类型。            RequestType,            # 从yaml数据中获取请求类型。            self.case_data.get(TestCaseEnum.REQUEST_TYPE.value[0])        )    @property    # 定义了一个名为get_dependence_case_data的方法，该方法不需要传入参数，属于类方法。    def get_dependence_case_data(self):        # 从self.case_data中获取DE_CASE的值，赋值给变量_dep_data。        _dep_data = self.case_data.get(TestCaseEnum.DE_CASE.value[0])        # 如果_dep_data存在。        if _dep_data:            # 断言self.case_data中是否存在DE_CASE_DATA的值，如果不存在，抛出异常。            assert self.case_data.get(TestCaseEnum.DE_CASE_DATA.value[0]) is not None, (                f"程序中检测到您的 case_id 为 {self.case_id} 的用例存在依赖，但是 {_dep_data} 缺少依赖数据."                f"如已填写，请检查缩进是否正确， 用例路径: {self.file_path}"            )        @ 返回self.case_data中的DE_CASE_DATA的值。        return self.case_data.get(TestCaseEnum.DE_CASE_DATA.value[0])    @property    # 定义了一个名为get_assert的方法，该方法不需要传入参数，属于类方法。    def get_assert(self):        # 从self.case_data中获取ASSERT_DATA的值，赋值给变量_assert_data。        _assert_data = self.case_data.get(TestCaseEnum.ASSERT_DATA.value[0])        # 断言_assert_data是否存在，如果不存在，抛出异常。        assert _assert_data is not None, (            f"用例ID 为 {self.case_id} 未添加断言，用例路径: {self.file_path}"        )        # 返回_assert_data的值。        return _assert_data    @property    # 定义了一个名为get_sql的方法，该方法不需要传入参数，属于类方法。    def get_sql(self):        # 从self.case_data中获取SQL的值，赋值给变量_sql。        _sql = self.case_data.get(TestCaseEnum.SQL.value[0])        # 判断数据库开关为开启状态，并且sql不为空        if config.mysql_db.switch and _sql is None:            # 返回None。            return None        # 返回_sql的值。        return _sql# 定义了一个名为case_process的方法，需要传入一个case_id_switch参数，属于类方法。def case_process(self, case_id_switch: Union[None, bool] = None):    # 从self.file_path中获取yaml文件中的数据，赋值给变量data。    data = GetYamlData(self.file_path).get_yaml_data()    case_list = []    # 循环遍历data中的每一个元素，赋值给变量key和values。    for key, values in data.items():        # 如果key不等于case_common。        if key != 'case_common':            # 将values赋值给self.case_data。            self.case_data = values            # 将key赋值给self.case_id。            self.case_id = key            # 调用CaseDataCheck类中的check_params_exit方法，检查yaml文件中的参数是否正确。            super().check_params_exit()            # 定义一个字典case_date，用于存储每一个测试用例的数据。            case_date = {                'method': self.get_method, # 获取请求方法。                'is_run': self.case_data.get(TestCaseEnum.IS_RUN.value[0]), # 获取是否执行该测试用例。                'url': self.get_host, # 获取请求url。                '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, # 获取执行的sql语句。                "assert_data": self.get_assert, # 获取断言数据。                "setup_sql": self.case_data.get(TestCaseEnum.SETUP_SQL.value[0]), # 获取测试用例执行前需要执行的sql语句。                "teardown": self.case_data.get(TestCaseEnum.TEARDOWN.value[0]), # 获取测试用例执行后需要执行的方法。                "teardown_sql": self.case_data.get(TestCaseEnum.TEARDOWN_SQL.value[0]), # 获取测试用例执行后需要执行的sql语句。                "sleep": self.case_data.get(TestCaseEnum.SLEEP.value[0]), # 获取测试用例执行前需要等待的时间。            }            # 如果case_id_switch为True，则将测试用例的id和数据存储为字典的形式            if case_id_switch is True:                case_list.append({key: TestCase(**case_date).dict()})            # 否则直接存储测试用例的数据。            else:                case_list.append(TestCase(**case_date).dict())    # 返回case_list    return case_listclass GetTestCase:    # 该标记表示该方法是一个静态方法，可以通过类名直接调用，不需要实例化类。    @staticmethod    # 定义了一个名为case_data的方法，该方法需要传入一个参数case_id_lists，该参数是一个列表类型，用于存储测试用例的id。    def case_data(case_id_lists: List):        # 初始化一个空列表，用于存储测试用例数据。        case_lists = []        # 循环遍历case_id_lists中的每一个元素，赋值给变量i。        for i in case_id_lists:            # 通过调用CacheHandler类的get_cache方法，获取指定测试用例的数据，并将其赋值给变量_data。            _data = CacheHandler.get_cache(i)            # 将获取到的测试用例数据存储到case_lists列表中。            case_lists.append(_data)        # 返回存储测试用例数据的列表case_lists。        return case_lists

                        regular_control.py  # 正则

                        详细讲解

import reimport datetimeimport randomfrom datetime import date, timedelta, datetimefrom jsonpath import jsonpathfrom faker import Fakerfrom utils.logging_tool.log_control import ERRORclass Context:    """ 正则替换 """    # 用于初始化对象的属性。    def __init__(self):        # self.faker 是对象的一个属性，用于存储一个Faker对象，Faker 是一个Python库，用于生成随机数据，如姓名、地址、电话号码等。使用了 locale='zh_CN' 参数来指定生成的数据的语言环境为中文。        self.faker = Faker(locale='zh_CN')    @classmethod    def generate_email(cls) -> str:        """        :return: 随机生成邮箱        """        Mailbox_number = 'register' + str(random.randint(0, 9999999)) + '@123.com'        return Mailbox_number    def generate_google_email(self) -> str:        """        :return: 随机生成Google邮箱        """        Google_email = 'google' + str(random.randint(0, 9999999)) + '@test.com'        return Google_email    def random_int(self) -> int:        """        :return: 随机数        """        _data = random.randint(0, 5000)        return _data    def get_phone(self) -> int:        """        :return: 随机生成手机号码        """        phone = self.faker.phone_number()        return phone    @classmethod    def get_time(cls) -> str:        """        计算当前时间        :return:        """        # 使用 datetime.now() 方法获取当前时间，然后使用 strftime() 方法将时间格式化为 '%Y-%m-%d %H:%M:%S' 的字符串格式。%Y 表示年份（4位数字），%m 表示月份（2位数字），%d 表示日期（2位数字），%H 表示小时（24小时制，2位数字），%M 表示分钟（2位数字），%S 表示秒（2位数字）。因此，now_time 变量将包含当前时间的字符串表示，格式为 '年-月-日 时:分:秒'。        now_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')        return now_time    # 装饰器，用于定义类方法。类方法与实例方法不同，类方法是针对整个类而不是实例的方法。    @classmethod    # 这是一个类方法，用于获取接口域名。-> str表示返回类型为字符串。    def host(cls) -> str:        from utils import config        """ 获取接口域名 """        # 返回config模块中的host变量。        return config.host    @classmethod    # 这是一个类方法，用于获取app的host。-> str表示返回类型为字符串。    def app_host(cls) -> str:        from utils import config        """获取app的host"""        # 返回config模块中的app_host变量。        return config.app_host# 这是一个函数，用于提取sql中的json数据。def sql_json(js_path, res):    """ 提取 sql中的 json 数据 """    # 这是一个jsonpath语法，用于从res变量中提取js_path指定的json数据。[0]表示取第一个元素。    _json_data = jsonpath(res, js_path)[0]    # 如果_json_data为False，即提取失败，则抛出ValueError异常。    if _json_data is False:        # 抛出ValueError异常，提示sql中的jsonpath获取失败。f""表示格式化字符串。        raise ValueError(f"sql中的jsonpath获取失败 {res}, {js_path}")    # 返回提取的json数据。    return jsonpath(res, js_path)[0]# 这是一个函数，用于处理sql中的依赖数据。def sql_regular(value, res=None):    """    这里处理sql中的依赖数据，通过获取接口响应的jsonpath的值进行替换    :param res: jsonpath使用的返回结果    :param value:    :return:    """    # 使用正则表达式查找value中以$json()格式包含的内容，并将其存储在sql_json_list列表中。    sql_json_list = re.findall(r"\$json\((.*?)\)\$", value)    # 循环遍历sql_json_list中的元素。    for i in sql_json_list:        # 使用正则表达式构造一个模式，用于匹配以$json()格式包含的内容。        pattern = re.compile(r'\$json\(' + i.replace('$', "\$").replace('[', '\[') + r'\)\$')        # 调用sql_json()函数获取json数据，并将其转换为字符串类型。        key = str(sql_json(i, res))        # 使用re.sub()函数将匹配到的内容替换成json数据中对应的值，替换次数为1。        value = re.sub(pattern, key, value, count=1)    # 返回替换后的value值。    return value# 定义了一个名为 cache_regular 的函数，用于通过正则表达式的方式读取缓存中的内容，参数为 value。def cache_regular(value):    from utils.cache_process.cache_control import CacheHandler    """    通过正则的方式，读取缓存中的内容    例：$cache{login_init}    :param value:    :return:    """    # 正则表达式\$cache\{(.*?)\}用来匹配字符串中以$cache{}包含的内容，.*?表示匹配任意字符，()表示括号内的内容为需要匹配的内容。re.findall()方法返回一个列表，包含所有匹配的结果。    # 正则获取 $cache{login_init}中的值 --> login_init    regular_dates = re.findall(r"\$cache\{(.*?)\}", value)    # for循环遍历regular_dates列表中的所有元素，即匹配到的所有缓存数据。    for regular_data in regular_dates:        # 定义了一个缓存数据类型的列表        value_types = ['int:', 'bool:', 'list:', 'dict:', 'tuple:', 'float:']        # 如果缓存数据的类型在value_types列表中，则将缓存数据类型和缓存数据名称分别提取出来        if any(i in regular_data for i in value_types) is True:            value_types = regular_data.split(":")[0]            regular_data = regular_data.split(":")[1]            # 使用re.compile()方法创建一个正则表达式对象pattern，用来匹配需要替换的字符串。            pattern = re.compile(r'\'\$cache\{' + value_types.split(":")[0] + ":" + regular_data + r'\}\'')        # 如果缓存数据的类型不在value_types列表中，则直接使用re.compile()方法创建一个正则表达式对象pattern，用来匹配需要替换的字符串。        else:            pattern = re.compile(                r'\$cache\{' + regular_data.replace('$', "\$").replace('[', '\[') + r'\}'            )        try:            # 使用CacheHandler.get_cache()方法读取缓存数据            cache_data = CacheHandler.get_cache(regular_data)            # 使用re.sub()方法将匹配到的字符串替换为缓存数据            value = re.sub(pattern, str(cache_data), value)        # 如果读取缓存数据或替换字符串过程中出现异常，则跳过并不处理。        except Exception:            pass    # 将处理好的字符串作为函数的返回值。    return value# 定义了一个名为 regular 的函数，该函数接受一个参数 target。def regular(target):    """    新版本    使用正则替换请求数据    :return:    """    try:        # 定义了一个正则表达式，用于匹配字符串中的 "${{}}" 格式的内容。        regular_pattern = r'\${{(.*?)}}'        # 使用 while 循环，当字符串中还存在 "${{}}" 格式的内容时，继续循环。        while re.findall(regular_pattern, target):            # 使用 re.search() 函数在字符串中找到第一个匹配的 "${{}}" 格式的内容，并获取其中的关键字。            key = re.search(regular_pattern, target).group(1)            # 定义了一个列表，包含了可能出现的数据类型。            value_types = ['int:', 'bool:', 'list:', 'dict:', 'tuple:', 'float:']            # 如果关键字中包含了可能出现的数据类型，则执行下面的代码。            if any(i in key for i in value_types) is True:                # 从关键字中获取函数名。                func_name = key.split(":")[1].split("(")[0]                # 从关键字中获取参数列表字符串，并去除字符串末尾的 ")"。                value_name = key.split(":")[1].split("(")[1][:-1]                # 如果参数列表字符串为空，则执行下面的代码。                if value_name == "":                    # 从 Context 类中获取函数名为 func_name 的函数，并执行该函数，获取返回值。                    value_data = getattr(Context(), func_name)()                # 如果参数列表字符串不为空，则执行下面的代码。                else:                    # 从 Context 类中获取函数名为 func_name 的函数，并将参数列表字符串转换为参数列表，执行该函数，获取返回值。                    value_data = getattr(Context(), func_name)(*value_name.split(","))                # 定义正则表达式模式，用于匹配字符串中的 "'${{}}'" 格式的内容。                regular_int_pattern = r'\'\${{(.*?)}}\''                # 将字符串中第一个匹配的 "'${{}}'" 格式的内容替换为 value_data。                target = re.sub(regular_int_pattern, str(value_data), target, 1)            # 如果关键字中不包含可能出现的数据类型，则执行下面的代码。            else:                # 从关键字中获取函数名。                func_name = key.split("(")[0]                # 从关键字中获取参数列表字符串，并去除字符串末尾的 ")"。                value_name = key.split("(")[1][:-1]                # 如果参数列表字符串为空，则执行下面的代码。                if value_name == "":                    # 从 Context 类中获取函数名为 func_name 的函数，并执行该函数，获取返回值。                    value_data = getattr(Context(), func_name)()                # 如果参数列表字符串不为空，则执行下面的代码。                else:                    # 从 Context 类中获取函数名为 func_name 的函数，并将参数列表字符串转换为参数列表，执行该函数，获取返回值。                    value_data = getattr(Context(), func_name)(*value_name.split(","))                # 将字符串中第一个匹配的 "${{}}" 格式的内容替换为 value_data。                target = re.sub(regular_pattern, str(value_data), target, 1)        # 返回处理后的字符串。        return target    # 捕获 AttributeError 异常。    except AttributeError:        # 打印错误日志，提示未找到对应的替换的数据。        ERROR.logger.error("未找到对应的替换的数据, 请检查数据是否正确 %s", target)        # 抛出异常。        raise    # 捕获 IndexError 异常。    except IndexError:        # 打印错误日志，提示 yaml 中的 ${{}} 函数方法不正确。        ERROR.logger.error("yaml中的 ${{}} 函数方法不正确，正确语法实例：${{get_time()}}")        # 抛出异常。        raise

                        swagger_for_yaml.py  # Swagger文档转换，生成YAML用例

import jsonfrom jsonpath import jsonpathfrom common.setting import ensure_path_sepfrom typing import Dictfrom ruamel import yamlimport osclass SwaggerForYaml:    def __init__(self):        self._data = self.get_swagger_json()    @classmethod    def get_swagger_json(cls):        """        获取 swagger 中的 json 数据        :return:        """        try:            with open('./file/test_OpenAPI.json', "r", encoding='utf-8') as f:                row_data = json.load(f)                return row_data        except FileNotFoundError:            raise FileNotFoundError("文件路径不存在，请重新输入")    def get_allure_epic(self):        """ 获取 yaml 用例中的 allure_epic """        _allure_epic = self._data['info']['title']        return _allure_epic    @classmethod    def get_allure_feature(cls, value):        """ 获取 yaml 用例中的 allure_feature """        _allure_feature = value['tags']        return str(_allure_feature)    @classmethod    def get_allure_story(cls, value):        """ 获取 yaml 用例中的 allure_story """        _allure_story = value['summary']        return _allure_story    @classmethod    def get_case_id(cls, value):        """ 获取 case_id """        _case_id = value.replace("/", "_")        return "01" + _case_id    @classmethod    def get_detail(cls, value):        _get_detail = value['summary']        return "测试" + _get_detail    @classmethod    def get_request_type(cls, value, headers):        """ 处理 request_type """        if jsonpath(obj=value, expr="$.parameters") is not False:            _parameters = value['parameters']            if _parameters[0]['in'] == 'query':                return "params"            else:                if 'application/x-www-form-urlencoded' or 'multipart/form-data' in headers:                    return "data"                elif 'application/json' in headers:                    return "json"                elif 'application/octet-stream' in headers:                    return "file"                else:                    return "data"    @classmethod    def get_case_data(cls, value):        """ 处理 data 数据 """        _dict = {}        if jsonpath(obj=value, expr="$.parameters") is not False:            _parameters = value['parameters']            for i in _parameters:                if i['in'] == 'header':                    ...                else:                    _dict[i['name']] = None        else:            return None        return _dict    @classmethod    def yaml_cases(cls, data: Dict, file_path: str) -> None:        """        写入 yaml 数据        :param file_path:        :param data: 测试用例数据        :return:        """        _file_path = ensure_path_sep("\\data\\" + file_path[1:].replace("/", os.sep) + '.yaml')        _file = _file_path.split(os.sep)[:-1]        _dir_path = ''        for i in _file:            _dir_path += i + os.sep        try:            os.makedirs(_dir_path)        except FileExistsError:            ...        with open(_file_path, "a", encoding="utf-8") as file:            yaml.dump(data, file, Dumper=yaml.RoundTripDumper, allow_unicode=True)            file.write('\n')    @classmethod    def get_headers(cls, value):        """ 获取请求头 """        _headers = {}        if jsonpath(obj=value, expr="$.consumes") is not False:            _headers = {"Content-Type": value['consumes'][0]}        if jsonpath(obj=value, expr="$.parameters") is not False:            for i in value['parameters']:                if i['in'] == 'header':                    _headers[i['name']] = None        else:            _headers = None        return _headers    def write_yaml_handler(self):        _api_data = self._data['paths']        for key, value in _api_data.items():            for k, v in value.items():                yaml_data = {                    "case_common": {"allureEpic": self.get_allure_epic(), "allureFeature": self.get_allure_feature(v),        "allureStory": self.get_allure_story(v)},                    self.get_case_id(key): {                        "host": "${{host()}}", "url": key, "method": k, "detail": self.get_detail(v),                        "headers": self.get_headers(v), "requestType": self.get_request_type(v, self.get_headers(v)),                        "is_run": None, "data": self.get_case_data(v), "dependence_case": False,                        "assert": {"status_code": 200}, "sql": None}}                self.yaml_cases(yaml_data, file_path=key)

                        testcase_template.py  # 测试用例模板

"""# @describe: 用例模板"""import datetimeimport osfrom utils.read_files_tools.yaml_control import GetYamlDatafrom common.setting import ensure_path_sepfrom utils.other_tools.exceptions import ValueNotFoundErrordef write_case(case_path, page):    """ 写入用例数据 """    with open(case_path, 'w', encoding="utf-8") as file:        file.write(page)def write_testcase_file(*, allure_epic, allure_feature, class_title,                        func_title, case_path, case_ids, file_name, allure_story):    """        :param allure_story:        :param file_name: 文件名称        :param allure_epic: 项目名称        :param allure_feature: 模块名称        :param class_title: 类名称        :param func_title: 函数名称        :param case_path: case 路径        :param case_ids: 用例ID        :return:        """    conf_data = GetYamlData(ensure_path_sep("\\common\\config.yaml")).get_yaml_data()    author = conf_data['tester_name']    now = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')    real_time_update_test_cases = conf_data['real_time_update_test_cases']    page = f'''#!/usr/bin/env python# -*- coding: utf-8 -*-# @Time   : {now}# @Author : {author}import allureimport pytestfrom utils.read_files_tools.get_yaml_data_analysis import GetTestCasefrom utils.assertion.assert_control import Assertfrom utils.requests_tool.request_control import RequestControlfrom utils.read_files_tools.regular_control import regularfrom utils.requests_tool.teardown_control import TearDownHandlercase_id = {case_ids}TestData = GetTestCase.case_data(case_id)re_data = regular(str(TestData))@allure.epic("{allure_epic}")@allure.feature("{allure_feature}")class Test{class_title}:    @allure.story("{allure_story}")    @pytest.mark.parametrize('in_data', eval(re_data), ids=[i['detail'] for i in TestData])    def test_{func_title}(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()

                        yaml_control.py  # yaml文件读写

import osimport astimport yaml.scannerfrom utils.read_files_tools.regular_control import regularclass GetYamlData:    """ 获取 yaml 文件中的数据 """    def __init__(self, file_dir):        self.file_dir = str(file_dir)    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    def write_yaml_data(self, key: str, value) -> int:        """        更改 yaml 文件中的值, 并且保留注释内容        :param key: 字典的key        :param value: 写入的值        :return:        """        with open(self.file_dir, 'r', encoding='utf-8') as file:            # 创建了一个空列表，里面没有元素            lines = []            for line in file.readlines():                if line != '\n':                    lines.append(line)            file.close()        with open(self.file_dir, 'w', encoding='utf-8') as file:            flag = 0            for line in lines:                left_str = line.split(":")[0]                if key == left_str.lstrip() and '#' not in line:                    newline = f"{left_str}: {value}"                    line = newline                    file.write(f'{line}\n')                    flag = 1                else:                    file.write(f'{line}')            file.close()            return flagclass GetCaseData(GetYamlData):    """ 获取测试用例中的数据 """    def get_different_formats_yaml_data(self) -> list:        """        获取兼容不同格式的yaml数据        :return:        """        res_list = []        for i in self.get_yaml_data():            res_list.append(i)        return res_list    def get_yaml_case_data(self):        """        获取测试用例数据, 转换成指定数据格式        :return:        """        _yaml_data = self.get_yaml_data()        # 正则处理yaml文件中的数据        re_data = regular(str(_yaml_data))        return ast.literal_eval(re_data)

                recording  # 代理录制

                        mitmproxy_control.py  # mitmproxy库拦截获取网络请求

from urllib.parse import parse_qs, urlparsefrom typing import Any, Union, Text, List, Dict, Tupleimport astimport osimport mitmproxy.httpfrom mitmproxy import ctxfrom ruamel import yamlclass Counter:    """    代理录制，基于 mitmproxy 库拦截获取网络请求    将接口请求数据转换成 yaml 测试用例    参考资料: https://blog.wolfogre.com/posts/usage-of-mitmproxy/    """    def __init__(self, filter_url: List, filename: Text = './data/proxy_data.yaml'):        self.num = 0        self.file = filename        self.counter = 1        # 需要过滤的 url        self.url = filter_url    def response(self, flow: mitmproxy.http.HTTPFlow) -> None:        """        mitmproxy抓包处理响应，在这里汇总需要数据, 过滤 包含指定url，并且响应格式是 json的        :param flow:        :return:        """        # 存放需要过滤的接口        filter_url_type = ['.css', '.js', '.map', '.ico', '.png', '.woff', '.map3', '.jpeg', '.jpg']        url = flow.request.url        ctx.log.info("=" * 100)        # 判断过滤掉含 filter_url_type 中后缀的 url        if any(i in url for i in filter_url_type) is False:            # 存放测试用例            if self.filter_url(url):                data = self.data_handle(flow.request.text)                method = flow.request.method                header = self.token_handle(flow.request.headers)                response = flow.response.text                case_id = self.get_case_id(url) + str(self.counter)                cases = {                    case_id: {                        "host": self.host_handle(url),                        "url": self.url_path_handle(url),                        "method": method,                        "detail": None,                        "headers": header,                        'requestType': self.request_type_handler(method),                        "is_run": True,                        "data": data,                        "dependence_case": None,                        "dependence_case_data": None,                        "assert": self.response_code_handler(response),                        "sql": None                    }                }                # 判断如果请求参数时拼接在url中，提取url中参数，转换成字典                if "?" in url:                    cases[case_id]['url'] = self.get_url_handler(url)[1]                    cases[case_id]['data'] = self.get_url_handler(url)[0]                ctx.log.info("=" * 100)                ctx.log.info(cases)                # 判断文件不存在则创建文件                try:                    self.yaml_cases(cases)                except FileNotFoundError:                    os.makedirs(self.file)                self.counter += 1    @classmethod    def get_case_id(cls, url: Text) -> Text:        """        通过url，提取对应的user_id        :param url:        :return:        """        _url_path = str(url).split('?')[0]        # 通过url中的接口地址，最后一个参数，作为case_id的名称        _url = _url_path.split('/')        return _url[-1]    def filter_url(self, url: Text) -> bool:        """过滤url"""        for i in self.url:            # 判断当前拦截的url地址，是否是addons中配置的host            if i in url:                # 如果是，则返回True                return True        # 否则返回 False        return False    @classmethod    def response_code_handler(cls, response) -> Union[Dict, None]:        """        处理接口响应，默认断言数据为code码，如果接口没有code码，则返回None        @param response:        @return:        """        try:            data = cls.data_handle(response)            return {"code": {"jsonpath": "$.code", "type": "==", "value": data['code'], "AssertType": None}}        except KeyError:            return None        except NameError:            return None    @classmethod    def request_type_handler(cls, method: Text) -> Text:        """ 处理请求类型，有params、json、file,需要根据公司的业务情况自己调整 """        if method == 'GET':            # 如我们公司只有get请求是prams，其他都是json的            return 'params'        return 'json'    @classmethod    def data_handle(cls, dict_str) -> Any:        """处理接口请求、响应的数据，如null、true格式问题"""        try:            if dict_str != "":                if 'null' in dict_str:                    dict_str = dict_str.replace('null', 'None')                if 'true' in dict_str:                    dict_str = dict_str.replace('true', 'True')                if 'false' in dict_str:                    dict_str = dict_str.replace('false', 'False')                dict_str = ast.literal_eval(dict_str)            if dict_str == "":                dict_str = None            return dict_str        except Exception as exc:            raise exc    @classmethod    def token_handle(cls, header) -> Dict:        """        提取请求头参数        :param header:        :return:        """        # 这里是将所有请求头的数据，全部都拦截出来了        # 如果公司只需要部分参数，可以在这里加判断过滤        headers = {}        for key, value in header.items():            headers[key] = value        return headers    def host_handle(self, url: Text) -> Tuple:        """        解析 url        :param url: https://xxxx.test.xxxx.com/#/goods/listShop        :return: https://xxxx.test.xxxx.com/        """        host = None        # 循环遍历需要过滤的hosts数据        for i in self.url:            # 这里主要是判断，如果我们conf.py中有配置这个域名，则用例中展示 ”${{host}}“，动态获取用例host            # 大家可以在这里改成自己公司的host地址            if 'https://www.wanandroid.com' in url:                host = '${{host}}'            elif i in url:                host = i        return host    def url_path_handle(self, url: Text):        """        解析 url_path        :param url: https://xxxx.test.xxxx.com/shopList/json        :return: /shopList/json        """        url_path = None        # 循环需要拦截的域名        for path in self.url:            if path in url:                url_path = url.split(path)[-1]        return url_path    def yaml_cases(self, data: Dict) -> None:        """        写入 yaml 数据        :param data: 测试用例数据        :return:        """        with open(self.file, "a", encoding="utf-8") as file:            yaml.dump(data, file, Dumper=yaml.RoundTripDumper, allow_unicode=True)            file.write('\n')    def get_url_handler(self, url: Text) -> Tuple:        """        将 url 中的参数 转换成字典        :param url: /trade?tradeNo=&outTradeId=11        :return: {“outTradeId”: 11}        """        result = None        url_path = None        for i in self.url:            if i in url:                query = urlparse(url).query                # 将字符串转换为字典                params = parse_qs(query)                # 所得的字典的value都是以列表的形式存在，如请求url中的参数值为空，则字典中不会有该参数                result = {key: params[key][0] for key in params}                url = url[0:url.rfind('?')]                url_path = url.split(i)[-1]        return result, url_path# 1、本机需要设置代理，默认端口为: 8080# 2、控制台输入 mitmweb -s .\utils\recording\mitmproxy_control.py - p 8888命令开启代理模式进行录制addons = [    Counter(["https://www.wanandroid.com"])    ]

                requests_tool  # 请求数据模块

                        dependent_case.py  # 数据依赖处理

import astimport jsonfrom typing import Text, Dict, Union, Listfrom jsonpath import jsonpathfrom utils.requests_tool.request_control import RequestControlfrom utils.mysql_tool.mysql_control import SetUpMySQLfrom utils.read_files_tools.regular_control import regular, cache_regularfrom utils.other_tools.jsonpath_date_replace import jsonpath_replacefrom utils.logging_tool.log_control import WARNINGfrom utils.other_tools.models import DependentTypefrom utils.other_tools.models import TestCase, DependentCaseData, DependentDatafrom utils.other_tools.exceptions import ValueNotFoundErrorfrom utils.cache_process.cache_control import CacheHandlerfrom utils import configclass DependentCase:    """ 处理依赖相关的业务 """    def __init__(self, dependent_yaml_case: TestCase):        self.__yaml_case = dependent_yaml_case    @classmethod    def get_cache(cls, case_id: Text) -> Dict:        """        获取缓存用例池中的数据，通过 case_id 提取        :param case_id:        :return: case_id_01        """        _case_data = CacheHandler.get_cache(case_id)        return _case_data    @classmethod    def jsonpath_data(            cls,            obj: Dict,            expr: Text) -> list:        """        通过jsonpath提取依赖的数据        :param obj: 对象信息        :param expr: jsonpath 方法        :return: 提取到的内容值,返回是个数组        对象: {"data": applyID} --> jsonpath提取方法: $.data.data.[0].applyId        """        _jsonpath_data = jsonpath(obj, expr)        # 判断是否正常提取到数据，如未提取到，则抛异常        if _jsonpath_data is False:            raise ValueNotFoundError(                f"jsonpath提取失败！\n 提取的数据: {obj} \n jsonpath规则: {expr}"            )        return _jsonpath_data    @classmethod    def set_cache_value(cls, dependent_data: "DependentData") -> Union[Text, None]:        """        获取依赖中是否需要将数据存入缓存中        """        try:            return dependent_data.set_cache        except KeyError:            return None    @classmethod    def replace_key(cls, dependent_data: "DependentData"):        """ 获取需要替换的内容 """        try:            _replace_key = dependent_data.replace_key            return _replace_key        except KeyError:            return None    def url_replace(            self,            replace_key: Text,            jsonpath_dates: Dict,            jsonpath_data: list) -> None:        """        url中的动态参数替换        # 如: 一般有些接口的参数在url中,并且没有参数名称, /api/v1/work/spu/approval/spuApplyDetails/{id}        # 那么可以使用如下方式编写用例, 可以使用 $url_params{}替换,        # 如/api/v1/work/spu/approval/spuApplyDetails/$url_params{id}        :param jsonpath_data: jsonpath 解析出来的数据值        :param replace_key: 用例中需要替换数据的 replace_key        :param jsonpath_dates: jsonpath 存放的数据值        :return:        """        if "$url_param" in replace_key:            _url = self.__yaml_case.url.replace(replace_key, str(jsonpath_data[0]))            jsonpath_dates['$.url'] = _url        else:            jsonpath_dates[replace_key] = jsonpath_data[0]    def _dependent_type_for_sql(            self,            setup_sql: List,            dependence_case_data: "DependentCaseData",            jsonpath_dates: Dict) -> None:        """        判断依赖类型为 sql，程序中的依赖参数从 数据库中提取数据        @param setup_sql: 前置sql语句        @param dependence_case_data: 依赖的数据        @param jsonpath_dates: 依赖相关的用例数据        @return:        """        # 判断依赖数据类型，依赖 sql中的数据        if setup_sql is not None:            if config.mysql_db.switch:                setup_sql = ast.literal_eval(cache_regular(str(setup_sql)))                sql_data = SetUpMySQL().setup_sql_data(sql=setup_sql)                dependent_data = dependence_case_data.dependent_data                for i in dependent_data:                    _jsonpath = i.jsonpath                    jsonpath_data = self.jsonpath_data(obj=sql_data, expr=_jsonpath)                    _set_value = self.set_cache_value(i)                    _replace_key = self.replace_key(i)                    if _set_value is not None:                        CacheHandler.update_cache(cache_name=_set_value, value=jsonpath_data[0])                        # cache(_set_value).set_caches(jsonpath_data[0])                    if _replace_key is not None:                        jsonpath_dates[_replace_key] = jsonpath_data[0]                        self.url_replace(replace_key=_replace_key,jsonpath_dates=jsonpath_dates,jsonpath_data=jsonpath_data,                        )            else:                WARNING.logger.warning("检查到数据库开关为关闭状态，请确认配置")    def dependent_handler(            self,            _jsonpath: Text,            set_value: Text,            replace_key: Text,            jsonpath_dates: Dict,            data: Dict,            dependent_type: int    ) -> None:        """ 处理数据替换 """        jsonpath_data = self.jsonpath_data(            data,            _jsonpath        )        if set_value is not None:            if len(jsonpath_data) > 1:                CacheHandler.update_cache(cache_name=set_value, value=jsonpath_data)            else:                CacheHandler.update_cache(cache_name=set_value, value=jsonpath_data[0])        if replace_key is not None:            if dependent_type == 0:                jsonpath_dates[replace_key] = jsonpath_data[0]            self.url_replace(replace_key=replace_key, jsonpath_dates=jsonpath_dates, jsonpath_data=jsonpath_data)    def is_dependent(self) -> Union[Dict, bool]:        """        判断是否有数据依赖        :return:        """        # 获取用例中的dependent_type值，判断该用例是否需要执行依赖        _dependent_type = self.__yaml_case.dependence_case        # 获取依赖用例数据        _dependence_case_dates = self.__yaml_case.dependence_case_data        _setup_sql = self.__yaml_case.setup_sql        # 判断是否有依赖        if _dependent_type is True:            # 读取依赖相关的用例数据            jsonpath_dates = {}            # 循环所有需要依赖的数据            try:                for dependence_case_data in _dependence_case_dates:                    _case_id = dependence_case_data.case_id                    # 判断依赖数据为sql，case_id需要写成self，否则程序中无法获取case_id                    if _case_id == 'self':                        self._dependent_type_for_sql(setup_sql=_setup_sql,dependence_case_data=dependence_case_data,jsonpath_dates=jsonpath_dates)                    else:                        re_data = regular(str(self.get_cache(_case_id)))                        re_data = ast.literal_eval(cache_regular(str(re_data)))                        res = RequestControl(re_data).http_request()                        if dependence_case_data.dependent_data is not None:dependent_data = dependence_case_data.dependent_datafor i in dependent_data:    _case_id = dependence_case_data.case_id    _jsonpath = i.jsonpath    _request_data = self.__yaml_case.data    _replace_key = self.replace_key(i)    _set_value = self.set_cache_value(i)    # 判断依赖数据类型, 依赖 response 中的数据    if i.dependent_type == DependentType.RESPONSE.value:        self.dependent_handler(            data=json.loads(res.response_data),            _jsonpath=_jsonpath,            set_value=_set_value,            replace_key=_replace_key,            jsonpath_dates=jsonpath_dates,            dependent_type=0        )    # 判断依赖数据类型, 依赖 request 中的数据    elif i.dependent_type == DependentType.REQUEST.value:        self.dependent_handler(            data=res.body,            _jsonpath=_jsonpath,            set_value=_set_value,            replace_key=_replace_key,            jsonpath_dates=jsonpath_dates,            dependent_type=1        )    else:        raise ValueError(            "依赖的dependent_type不正确，只支持request、response、sql依赖\n"            f"当前填写内容: {i.dependent_type}"        )                return jsonpath_dates            except KeyError as exc:                # pass                raise ValueNotFoundError(                    f"dependence_case_data依赖用例中，未找到 {exc} 参数，请检查是否填写"                    f"如已填写，请检查是否存在yaml缩进问题"                ) from exc            except TypeError as exc:                raise ValueNotFoundError(                    "dependence_case_data下的所有内容均不能为空！"                    "请检查相关数据是否填写，如已填写，请检查缩进问题"                ) from exc        else:            return False    def get_dependent_data(self) -> None:        """        jsonpath 和 依赖的数据,进行替换        :return:        """        _dependent_data = DependentCase(self.__yaml_case).is_dependent()        _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)

                        encryption_algorithm_control.py  # 加密算法

import hashlibfrom hashlib import sha256import hmacfrom typing import Textimport binasciifrom pyDes import des, ECB, PAD_PKCS5def hmac_sha256_encrypt(key, data):    """hmac sha 256算法"""    _key = key.encode('utf8')    _data = data.encode('utf8')    encrypt_data = hmac.new(_key, _data, digestmod=sha256).hexdigest()    return encrypt_datadef md5_encryption(value):    """ md5 加密"""    str_md5 = hashlib.md5(str(value).encode(encoding='utf-8')).hexdigest()    return str_md5def sha1_secret_str(_str: Text):    """    使用sha1加密算法，返回str加密后的字符串    """    encrypts = hashlib.sha1(_str.encode('utf-8')).hexdigest()    return encryptsdef des_encrypt(_str):    """    DES 加密    :return: 加密后字符串，16进制    """    # 密钥，自行修改    _key = 'PASSWORD'    secret_key = _key    _iv = secret_key    key = des(secret_key, ECB, _iv, pad=None, padmode=PAD_PKCS5)    _encrypt = key.encrypt(_str, padmode=PAD_PKCS5)    return binascii.b2a_hex(_encrypt)def encryption(ency_type):    """    :param ency_type: 加密类型    :return:    """    def decorator(func):        def swapper(*args, **kwargs):            res = func(*args, **kwargs)            _data = res['body']            if ency_type == "md5":                def ency_value(data):                    if data is not None:                        for key, value in data.items():if isinstance(value, dict):    ency_value(data=value)else:    data[key] = md5_encryption(value)            else:                raise ValueError("暂不支持该加密规则，如有需要，请联系管理员")            ency_value(_data)            return res        return swapper    return decorator

                        request_control.py  # 请求封装

import astimport osimport randomimport timeimport urllibfrom typing import Tuple, Dict, Union, Textimport requestsimport urllib3from requests_toolbelt import MultipartEncoderfrom common.setting import ensure_path_sepfrom utils.other_tools.models import RequestTypefrom utils.logging_tool.log_decorator import log_decoratorfrom utils.mysql_tool.mysql_control import AssertExecutionfrom utils.logging_tool.run_time_decorator import execution_durationfrom utils.other_tools.allure_data.allure_tools import allure_step, allure_step_no, allure_attachfrom utils.read_files_tools.regular_control import cache_regularfrom utils.requests_tool.set_current_request_cache import SetCurrentRequestCachefrom utils.other_tools.models import TestCase, ResponseDatafrom utils import config# from utils.requests_tool.encryption_algorithm_control import encryptionurllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)class RequestControl:    """ 封装请求 """    def __init__(self, yaml_case):        self.__yaml_case = TestCase(**yaml_case)    def file_data_exit(            self,            file_data) -> None:        """判断上传文件时，data参数是否存在"""        # 兼容又要上传文件，又要上传其他类型参数        try:            _data = self.__yaml_case.data            for key, value in ast.literal_eval(cache_regular(str(_data)))['data'].items():                if "multipart/form-data" in str(self.__yaml_case.headers.values()):                    file_data[key] = str(value)                else:                    file_data[key] = value        except KeyError:            ...    @classmethod    def multipart_data(            cls,            file_data: Dict):        """ 处理上传文件数据 """        multipart = MultipartEncoder(            fields=file_data,  # 字典格式            boundary='-----------------------------' + str(random.randint(int(1e28), int(1e29 - 1)))        )        return multipart    @classmethod    def check_headers_str_null(            cls,            headers: Dict) -> Dict:        """        兼容用户未填写headers或者header值为int        @return:        """        headers = ast.literal_eval(cache_regular(str(headers)))        if headers is None:            headers = {"headers": None}        else:            for key, value in headers.items():                if not isinstance(value, str):                    headers[key] = str(value)        return headers    @classmethod    def multipart_in_headers(            cls,            request_data: Dict,            header: Dict):        """ 判断处理header为 Content-Type: multipart/form-data"""        header = ast.literal_eval(cache_regular(str(header)))        request_data = ast.literal_eval(cache_regular(str(request_data)))        if header is None:            header = {"headers": None}        else:            # 将header中的int转换成str            for key, value in header.items():                if not isinstance(value, str):                    header[key] = str(value)            if "multipart/form-data" in str(header.values()):                # 判断请求参数不为空, 并且参数是字典类型                if request_data and isinstance(request_data, dict):                    # 当 Content-Type 为 "multipart/form-data"时，需要将数据类型转换成 str                    for key, value in request_data.items():                        if not isinstance(value, str):request_data[key] = str(value)                    request_data = MultipartEncoder(request_data)                    header['Content-Type'] = request_data.content_type        return request_data, header    def file_prams_exit(self) -> Dict:        """判断上传文件接口，文件参数是否存在"""        try:            params = self.__yaml_case.data['params']        except KeyError:            params = None        return params    @classmethod    def text_encode(            cls,            text: Text) -> Text:        """unicode 解码"""        return text.encode("utf-8").decode("utf-8")    @classmethod    def response_elapsed_total_seconds(            cls,            res) -> float:        """获取接口响应时长"""        try:            return round(res.elapsed.total_seconds() * 1000, 2)        except AttributeError:            return 0.00    def upload_file(            self) -> Tuple:        """        判断处理上传文件        :return:        """        # 处理上传多个文件的情况        _files = []        file_data = {}        # 兼容又要上传文件，又要上传其他类型参数        self.file_data_exit(file_data)        _data = self.__yaml_case.data        for key, value in ast.literal_eval(cache_regular(str(_data)))['file'].items():            file_path = ensure_path_sep("\\Files\\" + value)            file_data[key] = (value, open(file_path, 'rb'), 'application/octet-stream')            _files.append(file_data)            # allure中展示该附件            allure_attach(source=file_path, name=value, extension=value)        multipart = self.multipart_data(file_data)        # ast.literal_eval(cache_regular(str(_headers)))['Content-Type'] = multipart.content_type        self.__yaml_case.headers['Content-Type'] = multipart.content_type        params_data = ast.literal_eval(cache_regular(str(self.file_prams_exit())))        return multipart, params_data, self.__yaml_case    def request_type_for_json(            self,            headers: Dict,            method: Text,            **kwargs):        """ 判断请求类型为json格式 """        _headers = self.check_headers_str_null(headers)        _data = self.__yaml_case.data        _url = self.__yaml_case.url        res = requests.request(            method=method,            url=cache_regular(str(_url)),            json=ast.literal_eval(cache_regular(str(_data))),            data={},            headers=_headers,            verify=False,            params=None,            **kwargs        )        return res    def request_type_for_none(            self,            headers: Dict,            method: Text,            **kwargs) -> object:        """判断 requestType 为 None"""        _headers = self.check_headers_str_null(headers)        _url = self.__yaml_case.url        res = requests.request(            method=method,            url=cache_regular(_url),            data=None,            headers=_headers,            verify=False,            params=None,            **kwargs        )        return res    def request_type_for_params(            self,            headers: Dict,            method: Text,            **kwargs):        """处理 requestType 为 params """        _data = self.__yaml_case.data        url = self.__yaml_case.url        if _data is not None:            # url 拼接的方式传参            params_data = "?"            for key, value in _data.items():                if value is None or value == '':                    params_data += (key + "&")                else:                    params_data += (key + "=" + str(value) + "&")            url = self.__yaml_case.url + params_data[:-1]        _headers = self.check_headers_str_null(headers)        res = requests.request(            method=method,            url=cache_regular(url),            headers=_headers,            verify=False,            data={},            params=None,            **kwargs)        return res    def request_type_for_file(            self,            method: Text,            headers,            **kwargs):        """处理 requestType 为 file 类型"""        multipart = self.upload_file()        yaml_data = multipart[2]        _headers = multipart[2].headers        _headers = self.check_headers_str_null(_headers)        res = requests.request(            method=method,            url=cache_regular(yaml_data.url),            data=multipart[0],            params=multipart[1],            headers=ast.literal_eval(cache_regular(str(_headers))),            verify=False,            **kwargs        )        return res    def request_type_for_data(            self,            headers: Dict,            method: Text,            **kwargs):        """判断 requestType 为 data 类型"""        data = self.__yaml_case.data        _data, _headers = self.multipart_in_headers(            ast.literal_eval(cache_regular(str(data))),            headers        )        _url = self.__yaml_case.url        res = requests.request(            method=method,            url=cache_regular(_url),            data=_data,            headers=_headers,            verify=False,            **kwargs)        return res    @classmethod    def get_export_api_filename(cls, res):        """ 处理导出文件 """        content_disposition = res.headers.get('content-disposition')        filename_code = content_disposition.split("=")[-1]  # 分隔字符串，提取文件名        filename = urllib.parse.unquote(filename_code)  # url解码        return filename    def request_type_for_export(            self,            headers: Dict,            method: Text,            **kwargs):        """判断 requestType 为 export 导出类型"""        _headers = self.check_headers_str_null(headers)        _data = self.__yaml_case.data        _url = self.__yaml_case.url        res = requests.request(            method=method,            url=cache_regular(_url),            json=ast.literal_eval(cache_regular(str(_data))),            headers=_headers,            verify=False,            stream=False,            data={},            **kwargs)        filepath = os.path.join(ensure_path_sep("\\Files\\"), self.get_export_api_filename(res))  # 拼接路径        if res.status_code == 200:            if res.text:  # 判断文件内容是否为空                with open(filepath, 'wb') as file:                    # iter_content循环读取信息写入，chunk_size设置文件大小                    for chunk in res.iter_content(chunk_size=1):                        file.write(chunk)            else:                print("文件为空")        return res    @classmethod    def _request_body_handler(cls, data: Dict, request_type: Text) -> Union[None, Dict]:        """处理请求参数 """        if request_type.upper() == 'PARAMS':            return None        else:            return data    @classmethod    def _sql_data_handler(cls, sql_data, res):        """处理 sql 参数 """        # 判断数据库开关，开启状态，则返回对应的数据        if config.mysql_db.switch and sql_data is not None:            sql_data = AssertExecution().assert_execution(                sql=sql_data,                resp=res.json()            )        else:            sql_data = {"sql": None}        return sql_data    def _check_params(            self,            res,            yaml_data: "TestCase",    ) -> "ResponseData":        data = ast.literal_eval(cache_regular(str(yaml_data.data)))        _data = {            "url": res.url,            "is_run": yaml_data.is_run,            "detail": yaml_data.detail,            "response_data": res.text,            # 这个用于日志专用，判断如果是get请求，直接打印url            "request_body": self._request_body_handler(                data, yaml_data.requestType            ),            "method": res.request.method,            "sql_data": self._sql_data_handler(sql_data=ast.literal_eval(cache_regular(str(yaml_data.sql))), res=res),            "yaml_data": yaml_data,            "headers": res.request.headers,            "cookie": res.cookies,            "assert_data": yaml_data.assert_data,            "res_time": self.response_elapsed_total_seconds(res),            "status_code": res.status_code,            "teardown": yaml_data.teardown,            "teardown_sql": yaml_data.teardown_sql,            "body": data        }        # 抽离出通用模块，判断 http_request 方法中的一些数据校验        return ResponseData(**_data)    @classmethod    def api_allure_step(            cls,            *,            url: Text,            headers: Text,            method: Text,            data: Text,            assert_data: Text,            res_time: Text,            res: Text    ) -> None:        """ 在allure中记录请求数据 """        allure_step_no(f"请求URL: {url}")        allure_step_no(f"请求方式: {method}")        allure_step("请求头: ", headers)        allure_step("请求数据: ", data)        allure_step("预期数据: ", assert_data)        _res_time = res_time        allure_step_no(f"响应耗时(ms): {str(_res_time)}")        allure_step("响应结果: ", res)    @log_decorator(True)    @execution_duration(3000)    # @encryption("md5")    def http_request(            self,            dependent_switch=True,            **kwargs    ):        """        请求封装        :param dependent_switch:        :param kwargs:        :return:        """        from utils.requests_tool.dependent_case import DependentCase        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        }        is_run = ast.literal_eval(cache_regular(str(self.__yaml_case.is_run)))        # 判断用例是否执行        if is_run is True or is_run is None:            # 处理多业务逻辑            if dependent_switch is True:                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

                        set_current_request_cache.py  # 缓存设置

import jsonfrom typing import Textfrom jsonpath import jsonpathfrom utils.other_tools.exceptions import ValueNotFoundErrorfrom utils.cache_process.cache_control import CacheHandlerclass SetCurrentRequestCache:    """将用例中的请求或者响应内容存入缓存"""    def __init__(            self,            current_request_set_cache,            request_data,            response_data    ):        self.current_request_set_cache = current_request_set_cache        self.request_data = {"data": request_data}        self.response_data = response_data.text    def set_request_cache(            self,            jsonpath_value: Text,            cache_name: Text) -> None:        """将接口的请求参数存入缓存"""        _request_data = jsonpath(            self.request_data,            jsonpath_value        )        if _request_data is not False:            CacheHandler.update_cache(cache_name=cache_name, value=_request_data[0])            # cache(cache_name).set_caches(_request_data[0])        else:            raise ValueNotFoundError(                "缓存设置失败，程序中未检测到需要缓存的数据。"                f"请求参数: {self.request_data}"                f"提取的 jsonpath 内容: {jsonpath_value}"            )    def set_response_cache(            self,            jsonpath_value: Text,            cache_name    ):        """将响应结果存入缓存"""        _response_data = jsonpath(json.loads(self.response_data), jsonpath_value)        if _response_data is not False:            CacheHandler.update_cache(cache_name=cache_name, value=_response_data[0])            # cache(cache_name).set_caches(_response_data[0])        else:            raise ValueNotFoundError("缓存设置失败，程序中未检测到需要缓存的数据。"         f"请求参数: {self.response_data}"         f"提取的 jsonpath 内容: {jsonpath_value}")    def set_caches_main(self):        """设置缓存"""        if self.current_request_set_cache is not None:            for i in self.current_request_set_cache:                _jsonpath = i.jsonpath                _cache_name = i.name                if i.type == 'request':                    self.set_request_cache(jsonpath_value=_jsonpath, cache_name=_cache_name)                elif i.type == 'response':                    self.set_response_cache(jsonpath_value=_jsonpath, cache_name=_cache_name)

                        teardown_control.py  # 请求处理

"""# @describe: 请求后置处理"""import astimport jsonfrom typing import Dict, Textfrom jsonpath import jsonpathfrom utils.requests_tool.request_control import RequestControlfrom utils.read_files_tools.regular_control import cache_regular, sql_regular, regularfrom utils.other_tools.jsonpath_date_replace import jsonpath_replacefrom utils.mysql_tool.mysql_control import MysqlDBfrom utils.logging_tool.log_control import WARNINGfrom utils.other_tools.models import ResponseData, TearDown, SendRequest, ParamPreparefrom utils.other_tools.exceptions import JsonpathExtractionFailed, ValueNotFoundErrorfrom utils.cache_process.cache_control import CacheHandlerfrom utils import configclass TearDownHandler:    """ 处理yaml格式后置请求 """    def __init__(self, res: "ResponseData"):        self._res = res    @classmethod    def jsonpath_replace_data(            cls,            replace_key: Text,            replace_value: Dict) -> Text:        """ 通过jsonpath判断出需要替换数据的位置 """        _change_data = replace_key.split(".")        # jsonpath 数据解析        _new_data = jsonpath_replace(            change_data=_change_data,            key_name='_teardown_case',            data_switch=False        )        if not isinstance(replace_value, str):            _new_data += f" = {replace_value}"        # 最终提取到的数据,转换成 _teardown_case[xxx][xxx]        else:            _new_data += f" = '{replace_value}'"        return _new_data    @classmethod    def get_cache_name(            cls,            replace_key: Text,            resp_case_data: Dict) -> None:        """        获取缓存名称，并且讲提取到的数据写入缓存        """        if "$set_cache{" in replace_key and "}" in replace_key:            start_index = replace_key.index("$set_cache{")            end_index = replace_key.index("}", start_index)            old_value = replace_key[start_index:end_index + 2]            cache_name = old_value[11:old_value.index("}")]            CacheHandler.update_cache(cache_name=cache_name, value=resp_case_data)            # cache(cache_name).set_caches(resp_case_data)    @classmethod    def regular_testcase(cls, teardown_case: Dict) -> Dict:        """处理测试用例中的动态数据"""        test_case = regular(str(teardown_case))        test_case = ast.literal_eval(cache_regular(str(test_case)))        return test_case    @classmethod    def teardown_http_requests(cls, teardown_case: Dict) -> "ResponseData":        """        发送后置请求        @param teardown_case: 后置用例        @return:        """        test_case = cls.regular_testcase(teardown_case)        res = RequestControl(test_case).http_request(            dependent_switch=False        )        return res    def dependent_type_response(            self,            teardown_case_data: "SendRequest",            resp_data: Dict) -> Text:        """        判断依赖类型为当前执行用例响应内容        :param : teardown_case_data: teardown中的用例内容        :param : resp_data: 需要替换的内容        :return:        """        _replace_key = teardown_case_data.replace_key        _response_dependent = jsonpath(            obj=resp_data,            expr=teardown_case_data.jsonpath        )        # 如果提取到数据，则进行下一步        if _response_dependent is not False:            _resp_case_data = _response_dependent[0]            data = self.jsonpath_replace_data(                replace_key=_replace_key,                replace_value=_resp_case_data            )        else:            raise JsonpathExtractionFailed(                f"jsonpath提取失败，替换内容: {resp_data} \n"                f"jsonpath: {teardown_case_data.jsonpath}"            )        return data    def dependent_type_request(            self,            teardown_case_data: Dict,            request_data: Dict) -> None:        """        判断依赖类型为请求内容        :param : teardown_case_data: teardown中的用例内容        :param : request_data: 需要替换的内容        :return:        """        try:            _request_set_value = teardown_case_data['set_value']            _request_dependent = jsonpath(                obj=request_data,                expr=teardown_case_data['jsonpath']            )            if _request_dependent is not False:                _request_case_data = _request_dependent[0]                self.get_cache_name(                    replace_key=_request_set_value,                    resp_case_data=_request_case_data                )            else:                raise JsonpathExtractionFailed(                    f"jsonpath提取失败，替换内容: {request_data} \n"                    f"jsonpath: {teardown_case_data['jsonpath']}"                )        except KeyError as exc:            raise ValueNotFoundError("teardown中缺少set_value参数，请检查用例是否正确") from exc    def dependent_self_response(            self,            teardown_case_data: "ParamPrepare",            res: Dict,            resp_data: Dict) -> None:        """        判断依赖类型为依赖用例ID自己响应的内容        :param : teardown_case_data: teardown中的用例内容        :param : resp_data: 需要替换的内容        :param : res: 接口响应的内容        :return:        """        try:            _set_value = teardown_case_data.set_cache            _response_dependent = jsonpath(                obj=res,                expr=teardown_case_data.jsonpath            )            # 如果提取到数据，则进行下一步            if _response_dependent is not False:                _resp_case_data = _response_dependent[0]                # 拿到 set_cache 然后将数据写入缓存                # cache(_set_value).set_caches(_resp_case_data)                CacheHandler.update_cache(cache_name=_set_value, value=_resp_case_data)                self.get_cache_name(                    replace_key=_set_value,                    resp_case_data=_resp_case_data                )            else:                raise JsonpathExtractionFailed(                    f"jsonpath提取失败，替换内容: {resp_data} \n"                    f"jsonpath: {teardown_case_data.jsonpath}")        except KeyError as exc:            raise ValueNotFoundError("teardown中缺少set_cache参数，请检查用例是否正确") from exc    @classmethod    def dependent_type_cache(cls, teardown_case: "SendRequest") -> Text:        """        判断依赖类型为从缓存中处理        :param : teardown_case_data: teardown中的用例内容        :return:        """        if teardown_case.dependent_type == 'cache':            _cache_name = teardown_case.cache_data            _replace_key = teardown_case.replace_key            # 通过jsonpath判断出需要替换数据的位置            _change_data = _replace_key.split(".")            _new_data = jsonpath_replace(                change_data=_change_data,                key_name='_teardown_case',                data_switch=False            )            # jsonpath 数据解析            value_types = ['int:', 'bool:', 'list:', 'dict:', 'tuple:', 'float:']            if any(i in _cache_name for i in value_types) is True:                # _cache_data = cache(_cache_name.split(':')[1]).get_cache()                _cache_data = CacheHandler.get_cache(_cache_name.split(':')[1])                _new_data += f" = {_cache_data}"            # 最终提取到的数据,转换成 _teardown_case[xxx][xxx]            else:                # _cache_data = cache(_cache_name).get_cache()                _cache_data = CacheHandler.get_cache(_cache_name)                _new_data += f" = '{_cache_data}'"            return _new_data    def send_request_handler(            self, data: "TearDown",            resp_data: Dict,            request_data: Dict    ) -> None:        """        后置请求处理        @return:        """        _send_request = data.send_request        _case_id = data.case_id        # _teardown_case = ast.literal_eval(cache('case_process').get_cache())[_case_id]        _teardown_case = CacheHandler.get_cache(_case_id)        for i in _send_request:            if i.dependent_type == 'cache':                exec(self.dependent_type_cache(teardown_case=i))            # 判断从响应内容提取数据            if i.dependent_type == 'response':                exec(                    self.dependent_type_response(                        teardown_case_data=i,                        resp_data=resp_data)                )            # 判断请求中的数据            elif i.dependent_type == 'request':                self.dependent_type_request(                    teardown_case_data=i,                    request_data=request_data                )        test_case = self.regular_testcase(_teardown_case)        self.teardown_http_requests(test_case)    def param_prepare_request_handler(            self,            data: "TearDown",            resp_data: Dict) -> None:        """        前置请求处理        @param data:        @param resp_data:        @return:        """        _case_id = data.case_id        # _teardown_case = ast.literal_eval(cache('case_process').get_cache())[_case_id]        _teardown_case = CacheHandler.get_cache(_case_id)        _param_prepare = data.param_prepare        res = self.teardown_http_requests(_teardown_case)        for i in _param_prepare:            # 判断请求类型为自己,拿到当前case_id自己的响应            if i.dependent_type == 'self_response':                self.dependent_self_response(                    teardown_case_data=i,                    resp_data=resp_data,                    res=json.loads(res.response_data)                )    def teardown_handle(self) -> None:        """        为什么在这里需要单独区分 param_prepare 和 send_request        假设此时我们有用例A，teardown中我们需要执行用例B        那么考虑用户可能需要获取获取teardown的用例B的响应内容，也有可能需要获取用例A的响应内容，        因此我们这里需要通过关键词去做区分。这里需要考虑到，假设我们需要拿到B用例的响应，那么就需要先发送请求然后在拿到响应数据        那如果我们需要拿到A接口的响应，此时我们就不需要在额外发送请求了，因此我们需要区分一个是前置准备param_prepare，        一个是发送请求send_request        @return:        """        # 拿到用例信息        _teardown_data = self._res.teardown        # 获取接口的响应内容        _resp_data = self._res.response_data        # 获取接口的请求参数        _request_data = self._res.yaml_data.data        # 判断如果没有 teardown        if _teardown_data is not None:            # 循环 teardown中的接口            for _data in _teardown_data:                if _data.param_prepare is not None:                    self.param_prepare_request_handler(                        data=_data,                        resp_data=json.loads(_resp_data)                    )                elif _data.send_request is not None:                    self.send_request_handler(                        data=_data,                        request_data=_request_data,                        resp_data=json.loads(_resp_data)                    )        self.teardown_sql()    def teardown_sql(self) -> None:        """处理后置sql"""        sql_data = self._res.teardown_sql        _response_data = self._res.response_data        if sql_data is not None:            for i in sql_data:                if config.mysql_db.switch:                    _sql_data = sql_regular(value=i, res=json.loads(_response_data))                    MysqlDB().execute(cache_regular(_sql_data))                else:                    WARNING.logger.warning("程序中检查到您数据库开关为关闭状态，已为您跳过删除sql: %s", i)

                times_tool  # 时间模块

                        time_control.py   # 时间设置

import timefrom typing import Textfrom datetime import datetimedef count_milliseconds():    """    计算时间    :return:    """    access_start = datetime.now()    access_end = datetime.now()    access_delta = (access_end - access_start).seconds * 1000    return access_deltadef timestamp_conversion(time_str: Text) -> int:    """    时间戳转换，将日期格式转换成时间戳    :param time_str: 时间    :return:    """    try:        datetime_format = datetime.strptime(str(time_str), "%Y-%m-%d %H:%M:%S")        timestamp = int(            time.mktime(datetime_format.timetuple()) * 1000.0            + datetime_format.microsecond / 1000.0        )        return timestamp    except ValueError as exc:        raise ValueError('日期格式错误, 需要传入得格式为 "%Y-%m-%d %H:%M:%S" ') from excdef time_conversion(time_num: int):    """    时间戳转换成日期    :param time_num:    :return:    """    if isinstance(time_num, int):        time_stamp = float(time_num / 1000)        time_array = time.localtime(time_stamp)        other_style_time = time.strftime("%Y-%m-%d %H:%M:%S", time_array)        return other_style_timedef now_time():    """    获取当前时间, 日期格式: 2021-12-11 12:39:25    :return:    """    localtime = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())    return localtimedef now_time_day():    """    获取当前时间, 日期格式: 2021-12-11    :return:    """    localtime = time.strftime("%Y-%m-%d", time.localtime())    return localtimedef get_time_for_min(minute: int) -> int:    """    获取几分钟后的时间戳    @param minute: 分钟    @return: N分钟后的时间戳    """    return int(time.time() + 60 * minute) * 1000def get_now_time() -> int:    """    获取当前时间戳, 整形    @return: 当前时间戳    """    return int(time.time()) * 1000

pytest.ini  # Pytest 的配置文件

[pytest]addopts = -p no:warningstestpaths = test_case/python_files = test_*.pypython_classes = Test*python_function = test_*markers =    smoke: 冒烟测试

run.py  # 运行入口

import osimport tracebackimport pytestfrom utils.other_tools.models import NotificationTypefrom utils.other_tools.allure_data.allure_report_data import AllureFileCleanfrom utils.logging_tool.log_control import INFOfrom utils.notify.wechat_send import WeChatSendfrom utils.notify.ding_talk import DingTalkSendMsgfrom utils.notify.send_mail import SendEmailfrom utils.notify.lark import FeiShuTalkChatBotfrom utils.other_tools.allure_data.error_case_excel import ErrorCaseExcelfrom utils import configdef run():    # 从配置文件中获取项目名称    try:        INFO.logger.info(            """                 _      /\         (_)    /  \   _ __  _    / /\ \ | '_ \| |  / ____ \| |_) | | /_/    \_\ .__/|_|          | |                |_|                        开始执行{}项目...                """.format(config.project_name)        )        # 判断现有的测试用例，如果未生成测试代码，则自动生成        # TestCaseAutomaticGeneration().get_case_automatic()        pytest.main(['-s', '-W', 'ignore:Module already imported:pytest.PytestWarning',                     '--alluredir', './report/tmp', "--clean-alluredir"])        """                   --reruns: 失败重跑次数                   --count: 重复执行次数                   -v: 显示错误位置以及错误的详细信息                   -s: 等价于 pytest --capture=no 可以捕获print函数的输出                   -q: 简化输出信息                   -m: 运行指定标签的测试用例                   -x: 一旦错误，则停止运行                   --maxfail: 设置最大失败次数，当超出这个阈值时，则不会在执行测试用例                    "--reruns=3", "--reruns-delay=2"                   """        os.system(r"allure generate ./report/tmp -o ./report/html --clean")        allure_data = AllureFileClean().get_case_count()        notification_mapping = {            NotificationType.DING_TALK.value: DingTalkSendMsg(allure_data).send_ding_notification,            NotificationType.WECHAT.value: WeChatSend(allure_data).send_wechat_notification,            NotificationType.EMAIL.value: SendEmail(allure_data).send_main,            NotificationType.FEI_SHU.value: FeiShuTalkChatBot(allure_data).post        }        if config.notification_type != NotificationType.DEFAULT.value:            notify_type = config.notification_type.split(",")            for i in notify_type:                notification_mapping.get(i.lstrip(""))()        if config.excel_report:            ErrorCaseExcel().write_case()        # 程序运行之后，自动启动报告，如果不想启动报告，可注释这段代码        os.system(f"allure serve ./report/tmp -h 127.0.0.1 -p 8080")    except Exception:        # 如有异常，相关异常发送邮件        e = traceback.format_exc()        send_email = SendEmail(AllureFileClean.get_case_count())        send_email.error_mail(e)        raiseif __name__ == '__main__':    run()

持续更新中...

来源地址：https://blog.csdn.net/weixin_51084345/article/details/130552022