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

1
2
3
4
5
6
7
8
9
10
11
12
13

第二步：确认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
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

第四步： 处理用例的 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
1
2
3
4
5
6

第五步： 处理 请求头

这里从网上找了两个接口文档，文档中找到的和请求头相关的参数只有 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
1
2
3
4
5
6
7
8
9

第六步：处理框架中的 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"
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

第七步：处理 请求参数

    @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
1
2
3
4
5
6
7
8
9
10
11

第八步：封装写入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')
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

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

#!/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()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139

下面我们运行一下代码，看看生成yaml测试用例之后的效果吧~