Python 自动化处理 Yaml 文件-

Posted 2023-03-28

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Python 自动化处理 Yaml 文件-相关的知识，希望对你有一定的参考价值。

参考技术A

Yaml文件内容—示例1：

Yaml文件内容—示例2：

从上述示例文件内容可以看到 Yaml 数据结构：

1). 对象：键值对的集合（简称 "映射或字典"）

例如：family 和 address 这两个对象后面分别有对应的键值对集合。

2). 键值对用冒号 “:” 结构表示，冒号与值之间需用空格分隔

例如：

family 对象中的 key 为 name 与其对应的 value 值 Smile_Family 之间是使用空格分隔的。

address 对象中的 key 为 province 与其对应的 value 值 BeiJing 之间是使用空格分隔的。

3). 数组：一组按序排列的值（简称 "序列或列表"），数组前加有 “-” 符号，符号与值之间需用空格分隔

例如：

parents 中的 John 和 Jane

children 中的 Lily 和 Frank

4). 纯量(scalars)：单个的、不可再分的值。例如：字符串、bool值、整数、浮点数、时间、日期、null等

None值可用null也可用 ~ 表示；

yaml文件内容如下：

Python解析输出为：

这个例子输出一个字典，其中value包括所有基本类型

Yaml文件内容如下：

Python解析输出为：

Yaml文件内容如下：

Python输出为:

如果字符串没有空格或特殊字符，不需要加引号，但如果其中有空格或特殊字符，则需要加引号。

这里要注意单引号和双引号的区别：

单引号中的特殊字符转到 Python 会被转义，也就是到最后是原样输出；

双引号不会被 Python 转义，到最后是输出了特殊字符；

Yaml文件内容如下：

Python输出：

python实现处理swagger接口文档，转换为yaml格式的自动化用例

前言

之前有很多小伙伴反馈，希望我出一期将swagger文档转换为 yaml格式的自动化用例，那么本期福利来咯~~这一篇文档，将会带领你们实现如何通过 swagger文档转换为 yaml格式的用例，全程干货满满~

话不多说，直接开干

第一步：读取swagger接口文档

首先，大家既然看了这篇文章，那么想必公司是有swagger接口文档的，我们需要导出json格式的数据，数据是类似下方这样滴~

导出数据之后，首先第一步，我们需要封装一个读取 swagger 接口文档的数据

    @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("文件路径不存在，请重新输入")

第二步：确认yaml自动化用例的数据结构

第二个，我们就需要来确认一下自己yaml自动化用例的格式了，我的是下方这样的，这个是我封装的接口自动化框架的用例数据结构，基本上每个框架的数据结构都大同小异，可以根据自己的数据结构适当的调整代码，当然，我的接口自动化框架也是开源的，感兴趣的小伙伴可以直接看我框架中的源码，这一块的代码也在里面

框架地址：https://gitee.com/yu_xiao_qi/pytest-auto-api2

首先，映入眼帘的，我们可以看到有一部分公共数据，这里分别是这条用例需要用到的三个allure的装饰器，下方就是用例的内容，分别有用例ID、url、请求方式，请求体、测试步骤、断言。下面我们会依次处理这里的所有数据

第三步：获取用例中的allure 装饰器内容

这几个装饰器其实都非常简单，我们只需要提取文档中的指定内容即可。
如下图所示

allureEpic --> 对应文档中的 title
allureFeature --> 对应文档中的 tags
allureStory --> 对应文档中的 summary

那么我们了解了思路之后，首先封装一个获取对应的函数

class SwaggerForYaml:
    def __init__(self):
        self._data = self.get_swagger_json()
        
    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

第四步：处理用例的 case_id

当我们处理了 common 的数据，此时我们需要处理用例ID，在我的框架中，用例ID是需要唯一的，那么如何确保唯一性呢，这里我的处理是如接口为 /user_login，则id我们改成 01_user_login

毕竟我们接口肯定是唯一的，那么我么通过接口内容进行处理，转换成 case_id


    @classmethod
    def get_case_id(cls, value):
        """ 获取 case_id """
        _case_id = value.replace("/", "_")
        return "01" + _case_id

第五步：处理请求头

这里从网上找了两个接口文档，文档中找到的和请求头相关的参数只有 consumes 这个参数，但是这个是参数是个list，可是我看了一些文档，里面貌似都只有一个参数，不确定多个参数的情况下，会是什么样的，所以我这里只处理了第一个，将数据转成 "Content-Type": "multipart/form-data;"


    @classmethod
    def get_headers(cls, value):
        """ 获取请求头 """
        if jsonpath(obj=value, expr="$.consumes") is not False:
            _headers = "Content-Type": value['consumes'][0]
            return _headers
        else:
            return None

第六步：处理框架中的 requestType

首先我的框架中，有个参数规则为 requestType，参数值分别为 data、file、json、None、Params

那么如果请求参数我们是拼接到 url上面的话，则使用的是 params
如果我们请求的参数是json格式的，则用例中requestType是 json
如果我们请求的接口是上传文件类型的，则requestType为file
如果我们请求的参数是表单格式的，则参数为 data

因此我们了解了框架的规则之后，这里我们通过数据进行数据处理

    @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:
                _dict[i['name']] = None
        else:
            return None
        return _dict

第八步：封装写入yaml文件

    @classmethod
    def yaml_cases(cls, data: Dict, file_path: str) -> None:
        """
        写入 yaml 数据
        :param file_path:
        :param data: 测试用例数据
        :return:
        """

        _file_path = ConfigHandler.data_path + 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')

现在整体我们需要处理的数据都处理好了之后，我们来看看整体的代码

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
# @Time   : 2022/8/11 10:51
# @Author : 余少琪
"""
import json
from jsonpath import jsonpath
from common.setting import ConfigHandler
from typing import Dict
from ruamel import yaml
import os


class 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:
                _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 = ConfigHandler.data_path + 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):
        """ 获取请求头 """
        if jsonpath(obj=value, expr="$.consumes") is not False:
            _headers = "Content-Type": value['consumes'][0]
            return _headers
        else:
            return None

    def write_yaml_handler(self):

        _api_data = self._data['paths']
        # num = len(Counter(self._data['paths']).items())
        for key, value in _api_data.items():
            for k, v in value.items():
                yaml_data = 
                    "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)


if __name__ == '__main__':
    SwaggerForYaml().write_yaml_handler()