floraachy
248 lines
9.4 KiB
Python
248 lines
9.4 KiB
Python
# -*- coding: utf-8 -*-
|
||
# @Time : 2023/8/22 14:13
|
||
# @Author : chenyinhua
|
||
# @File : postman_for_yaml.py
|
||
# @Software: PyCharm
|
||
# @Desc:
|
||
|
||
# 标准库导入
|
||
import os
|
||
import re
|
||
import json
|
||
from typing import Dict
|
||
# 第三方库导入
|
||
from ruamel import yaml
|
||
|
||
"""
|
||
相比较于pyyaml, Ruamel可以保持YAML文件的结构和顺序不变。
|
||
安装：pip install ruamel.yaml
|
||
"""
|
||
|
||
|
||
class PostmanForYaml:
|
||
"""
|
||
将postman接口文档转为YAML格式用例
|
||
"""
|
||
|
||
def __init__(self, case_dir, postman_path):
|
||
"""
|
||
:param case_dir: 用例需要保存的目录
|
||
:param postman_path: 需要读取的swagger文件的路径
|
||
"""
|
||
self._data = self.get_postman_json(postman_path)
|
||
self.case_dir = case_dir
|
||
|
||
def get_postman_json(self, postman_path):
|
||
"""
|
||
获取 postman 中的 json 数据
|
||
:param postman_path: 需要读取的swagger文件的路径
|
||
:return:
|
||
"""
|
||
result = []
|
||
try:
|
||
with open(postman_path, "r", encoding='utf-8') as f:
|
||
row_data = json.load(f)
|
||
|
||
def _parse_api(content):
|
||
nonlocal row_data
|
||
nonlocal result
|
||
if isinstance(content, list):
|
||
for item in content:
|
||
_parse_api(content=item)
|
||
elif isinstance(content, dict):
|
||
if 'item' in content.keys():
|
||
_parse_api(content=content['item'])
|
||
elif 'request' in content.keys():
|
||
# 获取所有接口的相关数据
|
||
yaml_data = {
|
||
"case_common": {
|
||
"allure_epic": self.get_allure_epic(row_data),
|
||
"allure_feature": self.get_allure_feature(content),
|
||
"allure_story": self.get_allure_story(content)
|
||
},
|
||
"case_info": [
|
||
{
|
||
"id": f"case_{self.get_case_id(self.get_url(content)).lower()}_01",
|
||
"title": self.get_title(content),
|
||
"run": False,
|
||
"url": self.get_url(content),
|
||
"severity": None,
|
||
"method": self.get_method(content),
|
||
"headers": self.get_headers(content),
|
||
"cookies": None,
|
||
"request_type": self.get_request_type_payload(content).get("request_type"),
|
||
"payload": self.get_request_type_payload(content).get("payload"),
|
||
"files": self.get_request_type_payload(content).get("files"),
|
||
"extract": None,
|
||
"assert_response": {'eq': {'http_code': 200}},
|
||
"assert_sql": None
|
||
|
||
}
|
||
]
|
||
}
|
||
result.append({
|
||
self.get_case_id(self.get_url(content)): yaml_data
|
||
})
|
||
_parse_api(content=row_data)
|
||
return result
|
||
except FileNotFoundError:
|
||
raise FileNotFoundError("文件路径不存在，请重新输入")
|
||
|
||
def get_allure_epic(self, content):
|
||
"""
|
||
获取 yaml 用例中的 allure_epic
|
||
"""
|
||
_allure_epic = content['info']['name']
|
||
return _allure_epic
|
||
|
||
@classmethod
|
||
def get_allure_feature(cls, content):
|
||
"""
|
||
获取 yaml 用例中的 allure_feature
|
||
这里直接获取最下级的item.name，因为可以有多级item，不好判断
|
||
"""
|
||
|
||
_allure_feature = content['name']
|
||
return str(_allure_feature)
|
||
|
||
@classmethod
|
||
def get_allure_story(cls, content):
|
||
"""
|
||
获取 yaml 用例中的 allure_story
|
||
这里直接获取最下级的item.name，因为可以有多级item，不好判断
|
||
"""
|
||
_allure_story = content['name']
|
||
return _allure_story
|
||
|
||
@classmethod
|
||
def get_case_id(cls, content):
|
||
"""
|
||
获取 case_id， 是根据接口路径生成的
|
||
"""
|
||
# 这里接收到的参数content其实是url
|
||
# 移除？后面拼接的值
|
||
url_path = content.split("?")[0]
|
||
# 去除"http://"或"https://"部分
|
||
if url_path.startswith(("http://", "https://")):
|
||
url_path = url_path.split("//", 1)[1]
|
||
# 使用正则表达式匹配并删除第一个斜线以及斜线前的内容
|
||
new_url_path = re.sub(r'^.*?/', '', url_path)
|
||
# 将剩余的斜线替换成_
|
||
_case_id = new_url_path.replace("/", "_")
|
||
return _case_id
|
||
|
||
@classmethod
|
||
def get_title(cls, content):
|
||
"""
|
||
获取接口的标题
|
||
"""
|
||
_get_detail = content.get('name')
|
||
return "测试 " + _get_detail
|
||
|
||
@classmethod
|
||
def get_url(cls, content):
|
||
"""
|
||
获取接口的url
|
||
"""
|
||
request = content.get('request')
|
||
url = request.get('url')
|
||
url_raw = url.get('raw') if url else url
|
||
_url = url_raw.replace('{{', '${').replace('}}', '}')
|
||
# 使用正则表达式匹配":dept"内的内容并替换为${dept}
|
||
_get_url = re.sub(r':(\w+)', r'${\1}', _url)
|
||
return _get_url
|
||
|
||
@classmethod
|
||
def get_method(cls, content):
|
||
"""
|
||
获取接口的method
|
||
"""
|
||
request = content.get('request')
|
||
_get_method = request.get('method', 'GET').upper()
|
||
return _get_method
|
||
|
||
@classmethod
|
||
def get_headers(cls, content):
|
||
"""
|
||
获取请求头
|
||
"""
|
||
_headers = {}
|
||
request = content.get('request')
|
||
if request:
|
||
_headers = request.get('header')
|
||
_headers = {item.get('key'): item.get('value') for item in _headers} if _headers else {}
|
||
auth = request.get('auth')
|
||
if auth:
|
||
auth_type = auth.get('type')
|
||
if auth.get(auth_type):
|
||
auth_value = {item.get('key'): item.get('value') for item in auth.get(auth_type) if
|
||
(item and item.get('key'))}
|
||
_headers.update(auth_value)
|
||
# 如果_headers是{}就返回None
|
||
return None if not _headers else _headers
|
||
|
||
@classmethod
|
||
def get_request_type_payload(cls, content):
|
||
"""
|
||
获取request_type， 并响应处理payload及file参数
|
||
"""
|
||
api = {
|
||
"request_type": 'json',
|
||
"payload": {},
|
||
"files": {"file": []}
|
||
}
|
||
request = content.get('request')
|
||
if request:
|
||
body = request.get('body')
|
||
if body:
|
||
# api接口请求参数类型
|
||
request_mode = body.get('mode')
|
||
if request_mode in ['raw', 'formdata', 'urlencoded']:
|
||
api["request_type"] = 'json' if request_mode == 'raw' else 'data'
|
||
request_data = body.get(request_mode)
|
||
if request_data:
|
||
if request_mode == 'raw':
|
||
api["payload"].update(
|
||
json.loads(request_data.replace('\t', '').replace('\n', '').replace('\r', '')))
|
||
elif request_mode in ['formdata', 'urlencoded']:
|
||
for item in request_data:
|
||
if item['type'] == "text":
|
||
api["payload"][item['key']] = item.get('value', '')
|
||
elif item['type'] == "file":
|
||
api["files"]["file"].append(item.get('src', ''))
|
||
api["request_type"] = "file"
|
||
else:
|
||
raise ValueError("不支持的请求参数类型")
|
||
api["payload"] = None if not api["payload"] else api["payload"]
|
||
api["files"] = None if not api["files"]["file"] else api["files"]
|
||
return api
|
||
|
||
def yaml_cases(self, data: Dict, file_path: str) -> None:
|
||
"""
|
||
写入 yaml 数据
|
||
:param file_path:
|
||
:param data: 测试用例数据
|
||
:return:
|
||
"""
|
||
# 检查目录不存在则创建， 存在则不创建
|
||
os.makedirs(self.case_dir, exist_ok=True)
|
||
_file_name = file_path + '.yaml'
|
||
_file_path = os.path.join(self.case_dir, _file_name)
|
||
if _file_name in os.listdir(self.case_dir):
|
||
data.pop("case_common")
|
||
data = data["case_info"]
|
||
with open(_file_path, "a", encoding="utf-8") as file:
|
||
yaml.dump(data, file, Dumper=yaml.RoundTripDumper, allow_unicode=True)
|
||
file.write('\n')
|
||
|
||
def write_yaml_handler(self):
|
||
# 获取所有接口的相关数据
|
||
for case in self._data:
|
||
for k, v in case.items():
|
||
self.yaml_cases(data=v, file_path=k)
|
||
|
||
|
||
if __name__ == '__main__':
|
||
PostmanForYaml(case_dir=r"C:\1_xinjinyuan_chy\1project\apiautotest\files\postman",
|
||
postman_path=r"C:\1_xinjinyuan_chy\1project\apiautotest\files\Gitlink.postman_collection.json").write_yaml_handler()
|