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
import json
import random
import string
import requests
import yaml
from jsonpath import jsonpath
from login_session_function import LoginFunction
# from pymysql import OperationalError
import path_setting
class BaseRequest:
params = {}
@classmethod
def format(cls, r):
cls.r = r
print(json.dumps(json.loads(r.text), indent=2, ensure_ascii=False))
# 封装yaml文件读取
@classmethod
def yaml_load(cls, path) -> list:
with open(path, encoding='utf-8') as f:
return yaml.safe_load(f)
# 调用yaml加载文件API加载
def api_load(self, path: object) -> object:
return self.yaml_load(path)
def jsonpath(self, path, r=None, **kwargs):
if r is None:
r = self.r.json()
return jsonpath(r, path)
def api_send(self, req: object) -> object:
host = self.api_load(path_setting.HOSTYAML_CONFIG)
# 获取调用该方法的路径
import inspect
ins_file = inspect.stack()[1].filename
ins_dir = ins_file.split('/')[-2]
host_service = ins_dir.split('_')[1]
# default: backend
if host_service not in host['develop_host']:
host_service = 'backend'
url_host = host['develop_host'][host_service]
# if host["develop_host"].get("doctor") is not None:
# url_host = host['develop_host']['doctor']
# elif host["develop_host"].get("backend") is not None:
# url_host = host['develop_host']['backend']
raw = yaml.dump(req) # 将一个python对象生成为yaml文档
for key, value in self.params.items():
raw = raw.replace(f"${{{key}}}", repr(value))
req = yaml.safe_load(raw)
print('req:', req)
s = requests
if req.get('isLogin'):
s = LoginFunction().get_session(host_service)
# 调用具体case的url
r = s.request(
req['method'],
url=url_host + req['url'],
params=req.get('params'),
# headers=user_headers,
data=req.get('data'),
json=req.get('json'),
# verify=False
# proxies={"http":"172.30.9.226:8888"}
)
print("0000000",r)
# return r.json()
try:
content_type = r.headers.get("content-type").split(";")[0]
except:
content_type = "application/json"
if content_type == 'application/json':
return r.json()
return r.text
# 随机生成trace_id
def trace_id(self):
return ''.join(random.sample(string.ascii_lowercase + string.digits, 32))
'''
def get_cookie(self, req: dict):
host = self.api_load(path_setting.HOSTYAML_CONFIG)
print(host)
r = requests.request(
req['method'],
url=host['develop_host']['url'] + req['url'],
params=req['params'],
headers=req['headers'],
data=req['data'],
json=req['json']
)
dict = {}
for i in r.cookies:
dict[i.name] = i.value
headers = '_gtid={};sessionid={}'.format(dict["_gtid"], dict["sessionid"])
return headers
def read_header(self):
with open("../in_common/get_cookie.txt", 'r', encoding='utf-8') as f:
cookies = f.read()
headers = {"cookie": cookies}
return headers
'''
if __name__ == '__main__':
# BaseRequest().api_load("../api/api.yaml")
print(BaseRequest().trace_id())