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

Posted 七月的小尾巴

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了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()

以上是关于python实现处理swagger接口文档,转换为yaml格式的自动化用例的主要内容,如果未能解决你的问题,请参考以下文章

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

swagger文档转换为WebApiClient声明式代码

将Swagger中所有接口地址用Python写入Excel中

利用swagger模块开发flask的api接口帮助文档

自定义枚举 --- Swagger文档展示

Http通信协议接口接入swagger步骤实现自动生成接口文档