289 lines
13 KiB
Python
289 lines
13 KiB
Python
#!/usr/bin/python3
|
||
# -*- coding: UTF-8 -*-
|
||
# 设置utf-8 显示中文
|
||
"""
|
||
@Create Time: 2021/3/12 23:48
|
||
@Author: guo
|
||
@File：generate_yaml.py
|
||
"""
|
||
import json
|
||
|
||
import mitmproxy
|
||
from mitmproxy import http
|
||
import yaml
|
||
import copy
|
||
import re
|
||
import sys
|
||
import os
|
||
|
||
sys.path.append("../")
|
||
from config.get_conf_data import GetConfData
|
||
|
||
|
||
class GenYamlTemplate:
|
||
|
||
def __init__(self):
|
||
self.__yaml_case_template_path = "case_yml_template.yml"
|
||
self.__yaml_page_template_path = "page_yml_template.yml"
|
||
|
||
# 设置目标host
|
||
self.__host_target = ["192.168.9.105", "105.xxx.xxxx", "xxx.xxx.xxx"]
|
||
# 静态资源
|
||
self.__static_ext = ["js", "css", "ico", "jpg", "png", "gif", "jpeg", "bmp", "ttf"]
|
||
# 静态文件,在家校pc端查看,其中 "application/javascript" js 文件也是无效的。
|
||
self.__static_files = ["text/css", "image/jpeg", "image/png", "image/gif", "text/html",
|
||
"application/octet-stream",
|
||
"application/x-protobuf"]
|
||
# 图片
|
||
self.__media_types = ["image", "video", "audio"]
|
||
# 上面的几种静态资源，需要通过resonse来进行校验。存在于 response.headers("Content-Type")中
|
||
# 其中response为 flow.response
|
||
|
||
# 本次由于只需要生成操作的页面的接口对应的yaml文件。所以需要拿 request.headers("Content-Type") 是否为None进行判断即可。
|
||
self.__type = "Content-Type"
|
||
# 环境
|
||
self.__env_web = "web"
|
||
self.__env_app = "app"
|
||
self.__env = self.__env_web
|
||
# 读取yaml模板
|
||
self.__case_yaml_temp = yaml.safe_load(open(self.__yaml_case_template_path, encoding="utf8"))
|
||
self.__page_yaml_temp = yaml.safe_load(open(self.__yaml_page_template_path, encoding="utf-8"))
|
||
# remove 字符串。因为公司的有些remove接口 的request 的headers里没有Content-Type。 有些是有值。
|
||
self.__remove_str = "remove"
|
||
self.__post_str = "post"
|
||
self.__get_str = "get"
|
||
self.__json_str = "json"
|
||
self.__content_json = "application/json"
|
||
self.__data_str = "data"
|
||
self.__params_str = "params"
|
||
self.__case_tmp_str = "case_template"
|
||
self.__page_tmp_str = "page_template"
|
||
self.__web_str = "web"
|
||
self.__zhjx_gate = "zhjxgate"
|
||
self.__web_flag = f"/{self.__web_str}/"
|
||
self.__zhjx_flag= f"/{self.__zhjx_gate}/"
|
||
self.__env_web_flag = "web"
|
||
self.__env_app_flag = "app"
|
||
# 环境切换
|
||
self.__env_flag = self.__env_web_flag
|
||
self.__conf = GetConfData()
|
||
self.__yaml_dir = self.__conf.get_ymlfile_abspath()
|
||
self.__init_key()
|
||
self.__count=0
|
||
def __init_key(self):
|
||
"""需要用到的一些key,先不考虑把这些key写入到配置文件或独立出去."""
|
||
self.__app_login_key="app_login"
|
||
self.__web_login_key = "web_login"
|
||
self.__web_str_key = "web_str"
|
||
self.__app_str_key = "app_str"
|
||
self.__test_init_key = "test_init"
|
||
self.__login_type_key = "login_type"
|
||
self.__page_template_key = "page_template"
|
||
self.__case_template_key = "case_template"
|
||
self.__login_data_key = "login_data"
|
||
self.__api_data_key = "api_data"
|
||
self.__path_key = "path"
|
||
self.__data_type_key = "data_type"
|
||
self.__request_data_key = "request_data"
|
||
self.__method_key = "method"
|
||
self.__timestamp_key = "_"
|
||
|
||
|
||
def write_data_to_yaml(self, yaml_path: str, data: dict) -> None:
|
||
yml_path = yaml_path
|
||
# 先检查是否以.yaml 结尾，如果是统一修改为 .yml
|
||
yaml_suffix=".yaml"
|
||
yml_suffix=".yml"
|
||
exp=f"({yaml_suffix})$"
|
||
yml_path=re.sub(exp,yml_suffix,yml_path)
|
||
|
||
# 如果不是以.yml结尾，就更改为 .yml
|
||
if yml_path:
|
||
yml_path = yml_path if yml_path.endswith(".yml") else yml_path + ".yml"
|
||
|
||
# 拼接路径
|
||
target_path = os.path.join(self.__yaml_dir, yml_path)
|
||
# 获取目录
|
||
target_dir = os.path.dirname(target_path)
|
||
# 当目录不存在时，则进行创建操作
|
||
if not os.path.exists(target_dir):
|
||
# 当已经存在的目录，则不创建
|
||
os.makedirs(target_dir,exist_ok=True)
|
||
|
||
|
||
datas={}
|
||
# 检测文件是否存在
|
||
if os.path.exists(target_path):
|
||
data_tmp = yaml.safe_load(open(target_path,encoding="utf-8"))
|
||
# 当不为None时，赋值给变量
|
||
if data_tmp !=None:
|
||
datas= data_tmp
|
||
|
||
# 写入文件
|
||
with open(target_path,"w",encoding="utf-8") as fp :
|
||
datas.update(data)
|
||
# 以原来的顺序写入到文件中，保存原有顺序不变，同时中文不能乱码。
|
||
yaml.safe_dump(datas,fp,allow_unicode="utf-8",default_flow_style=False,sort_keys=False)
|
||
|
||
|
||
def request(self, flow: mitmproxy.http.HTTPFlow):
|
||
self.__count+=1
|
||
request = flow.request
|
||
host = request.host
|
||
# 获取不带参数的path，返回的tuple类型
|
||
path = request.path_components
|
||
# 获取url path路径
|
||
url_path = "/" + "/".join(path)
|
||
# 用于获取，并校验是否为删除接口
|
||
latest_path = path[-1]
|
||
|
||
# 获取请求方法类型
|
||
method = str.lower(request.method)
|
||
# 获取提交的参数,可按字典的操作方式进行操作。
|
||
# 在获取数据之前,要先判断是get请求还是post请求,两者存的位置不一样
|
||
if method == self.__get_str:
|
||
# 目前知道的,当为get请求时,请求的参数是存放于 request.query 中
|
||
params = request.query
|
||
# post 请求(非get请求)
|
||
else:
|
||
params = request.urlencoded_form
|
||
|
||
# 获取各项值
|
||
data={}
|
||
for key ,item in params.items():
|
||
# 有些字段是时间戳,可以不用录入,这个要根据各个公司的接口规范要求
|
||
# 目前时间戳是非必填字段
|
||
if self.__timestamp_key != key :
|
||
data[key] = item
|
||
|
||
|
||
# 获取请求的 Content-Type
|
||
content_type = request.headers.get(self.__type)
|
||
# 过滤请求，只保留目标数据
|
||
if (host in self.__host_target) and \
|
||
((content_type != None) or ((content_type == None) and latest_path.startswith(self.__remove_str))):
|
||
# 先拷贝yaml模板的数据,深度拷贝
|
||
case_yaml_datas = copy.deepcopy(self.__case_yaml_temp)
|
||
page_yaml_datas = copy.deepcopy(self.__page_yaml_temp)
|
||
|
||
|
||
# 至于用什么标识进行切割,要根据实际的项目进行确定,找出共同点
|
||
# 检测 "/web/" 字符串是否存在于path路径中
|
||
if self.__web_flag in url_path:
|
||
# 获取切割后的功能模块的路径.
|
||
tmp_path = url_path.split(self.__web_flag)[-1]
|
||
|
||
# 检测 "/zhjxgate/" 字符串是否存在于 path路径中
|
||
elif self.__zhjx_flag in url_path:
|
||
tmp_path = url_path.split(self.__zhjx_flag)[-1]
|
||
# 暂时不考虑其他情况.
|
||
else:
|
||
pass
|
||
|
||
# 获取当前的接口名称(如:add,remove,query,audit),在写入page 和 case层yml文件时,需要用到
|
||
key = os.path.basename(tmp_path)
|
||
page_key = key
|
||
case_key = "test_"+key
|
||
|
||
# 生成各自的key的数据,要使用deepcopy,防止出现引用现象.因为在使用浅拷贝时,只要是容器类型(字典,列表,元组,集合)时,
|
||
# 使用copy.copy时,是直接引用的对象的id,也就是内存中的地址.所以一改,就全部都改了.并且最后在赋值的时候,也是进行引用赋值
|
||
# 使用deepcopy 是相当于重新新建了一份数据,是完全独立的.
|
||
case_templ_data = copy.deepcopy(case_yaml_datas[self.__case_template_key])
|
||
page_templ_data = copy.deepcopy(page_yaml_datas[self.__page_template_key])
|
||
|
||
# 先生成各自层的数据
|
||
case_yaml_datas[case_key] = case_templ_data
|
||
page_yaml_datas[page_key] = page_templ_data
|
||
|
||
|
||
|
||
# 判断是web端 还是app端
|
||
if self.__env_flag == self.__env_web_flag:
|
||
# 下面进行web yaml模板的配置
|
||
case_login_key = self.__web_login_key
|
||
page_login_key = self.__web_login_key
|
||
# 终端标识
|
||
type_flag_key = self.__web_str_key
|
||
|
||
elif self.__env_flag == self.__env_app_flag:
|
||
case_login_key = self.__app_login_key
|
||
page_login_key = self.__app_login_key
|
||
type_flag_key = self.__app_str_key
|
||
|
||
case_login = copy.deepcopy(case_yaml_datas[case_login_key])
|
||
page_login = copy.deepcopy(page_yaml_datas[page_login_key])
|
||
# 获取登录类型
|
||
login_type = page_yaml_datas[type_flag_key]
|
||
|
||
# 将case模板中的 test_init 下的login_type 和 login_data 的值全部替换掉
|
||
case_yaml_datas[self.__test_init_key][self.__login_type_key] = login_type
|
||
# 必须再次进行deepcopy,否则会以引用的方式出来.
|
||
# 也就是说 deepcopy后的值,必须只能赋值一次,否则
|
||
case_login_data = copy.deepcopy(case_yaml_datas[case_login_key])
|
||
case_yaml_datas[self.__test_init_key][self.__login_data_key] = case_login_data
|
||
|
||
# 获取各自api_data数据
|
||
case_apidata = case_yaml_datas[case_key][self.__api_data_key]
|
||
page_apidata = page_yaml_datas[page_key][self.__api_data_key]
|
||
|
||
#替换page层的path
|
||
page_apidata[self.__path_key] = url_path
|
||
data_type=""
|
||
# 下面开始进行method判断
|
||
if method == self.__get_str :
|
||
# 当为get请求时，data_type 为：params
|
||
data_type = self.__params_str
|
||
|
||
elif method == self.__post_str:
|
||
# 当为post请求时，要判断content-type 是否为application/json，若为，则表示为 data_type=json
|
||
if self.__content_json in content_type:
|
||
data_type = self.__json_str
|
||
else:
|
||
data_type = self.__data_str
|
||
# 当为其他类型的请求方法时，暂时不考虑，或将设置为data
|
||
else:
|
||
data_type = self.__data_str
|
||
|
||
# 对各自的data_type 进行赋值
|
||
case_apidata[self.__data_type_key] = data_type
|
||
page_apidata[self.__data_type_key] = data_type
|
||
# 对page层的request_data下的method 进行赋值
|
||
page_apidata[self.__request_data_key][self.__method_key] = method
|
||
# 再对各自的request_data中的data 进行赋值 ,注意顺序
|
||
request_data = {}
|
||
request_data[data_type] = data
|
||
case_apidata[self.__request_data_key] = request_data
|
||
page_apidata[self.__request_data_key][data_type] = data
|
||
# 对各login_data 下的 login_data 进行赋值
|
||
case_apidata[self.__login_data_key][self.__login_data_key] = case_login
|
||
page_apidata[self.__login_data_key][self.__login_data_key] = page_login
|
||
# 对page层下的 login_data下的login_type 进行赋值
|
||
page_apidata[self.__login_data_key][self.__login_type_key] = login_type
|
||
|
||
#******************** 下面 开始进行 yml 文件的生成了 ××××××××××××××××××××××
|
||
'''
|
||
在进行yaml文件生成时，需要注意的是： 一个功能模块下的接口，应该在同个yaml文件中，且各自接口的名称(如add，query）作为各自的key
|
||
目前在进行分割时，是优先拿web进行侵害，当不存在web时，再拿zhjxgate进行分割。
|
||
需要注意的是，app的接口，ymlfile目录下的 app目录下，web接口在ymlfile目录下的 web目录下。
|
||
'''
|
||
|
||
# 获取路径
|
||
# 这块的写法是不严谨的,有可能会出现 key为 空的情况.即 /web/action,这种情况,此时的 tmp_path = "action"
|
||
# 当这情况时,此时的dir为空,也就是""
|
||
dir = os.path.dirname(tmp_path)
|
||
if dir == "":
|
||
dir = os.path.basename(tmp_path)
|
||
# 获取文件名称
|
||
filename = os.path.basename(dir)+".yml"
|
||
|
||
# 设置case层和 page层各自的yml文件
|
||
page_file = self.__web_str+"/"+dir+"/"+filename
|
||
case_file = self.__web_str+"/"+dir+"/"+"test_"+filename
|
||
|
||
|
||
self.write_data_to_yaml(case_file,case_yaml_datas)
|
||
self.write_data_to_yaml(page_file,page_yaml_datas)
|
||
|
||
|
||
addons = [GenYamlTemplate()]
|