MySession.py
9.5 KB
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
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
#!/usr/bin/python
# -*- coding: UTF-8 -*-
import requests
import json
from commons import common as com
from commons.Logging import Logger
from commons.clientSession import cliSession
log=Logger()
class mysession():
"封装了requests的基类，以供后期统一使用"
url = "http://test.uap.diligrp.com/login/login.action"
header = {
"Host": "test.uap.diligrp.com",
"Connection": "keep-alive",
"Content-Length": "33",
"Cache-Control": "max-age=0",
"Upgrade-Insecure-Requests": "1",
"Origin": "http://test.uap.diligrp.com",
"Content-Type": "application/x-www-form-urlencoded",
"User-Agent": "Mozilla/5.0(WindowsNT10.0;Win64;x64)AppleWebKit/537.36(KHTML,likeGecko)Chrome/90.0.4430.212Safari/537.36",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9",
"Referer": "http://test.uap.diligrp.com/login/index.html",
"Accept-Encoding": "gzip,deflate",
"Accept-Language": "zh-CN,zh-TW;q=0.9,zh;q=0.8,en;q=0.7",
"Cookie": "UAP_accessToken=;UAP_refreshToken=;UAP_loginPath="}
body = "userName=sg_wenze&password=111111"
def __init__(self):
"如下代码，可以通过配置文件来控制测试环境和灰度环境，http和https"
self.url = mysession.url.replace("http://test.", com.get_global_config("global_data", "environment", "en"))
self.header = mysession.header
self.body = mysession.body
self.timeout = (5, 5)
self.max_retries = 3
self.keep_alive = False
self.ssl_verify = False
self.proxies = None
# self.proxies={'http': 'http://localhost:8888', 'https': 'http://localhost:8888'}
self.allow_redirects = False
self.firmid={"group":"1","hd":"2","cd":"3","qqhe":"4","mdj":"5","gy":"6","cc":"7","sg":"8","sy":"9"}
def cliLogin(self):
self.webHeaders, self.clientHeaders, self.userInfo = cliSession().loginUser()
def get_session(self, account, **kwargs):
# print("get_session")
"如下代码，可以通过配置文件来控制登录的账户session"
self.body = self.body.replace("sg_wenze",
com.get_global_config("global_data", "account", account).split("&")[0])
self.body = self.body.replace("111111", com.get_global_config("global_data", "account", account).split("&")[1])
# requests.session()会话保持,比如使用session成功的登录了某个网站，
# 则在再次使用该session对象求求该网站的其他网页都会默认使用该session之前使用的cookie等参数
self.se = requests.session()
#更新cookie加入UAP_firmId
firm=account.split("_")[0]
co = requests.cookies.RequestsCookieJar()
co.set("UAP_firmId", self.firmid[firm])
self.se.cookies.update(co)
# 使用session对象的方法POST/GET等
re = self.se.post(url=self.url, headers=self.header, data=self.body, proxies=self.proxies, **kwargs)
#获取关键信息供其他接口header使用
self.UAP_accessToken=self.se.cookies["UAP_accessToken"]
self.UAP_refreshToken=self.se.cookies["UAP_refreshToken"]
# 返回session对象，供其他接口使用
return self.se
def close_session(self):
"关闭session"
self.se.close()
def check_login(self, account, **kwargs):
"验证登录接口"
self.body = self.body.replace("sg_wenze",
com.get_global_config("global_data", "account", account).split("&")[0])
self.body = self.body.replace("111111", com.get_global_config("global_data", "account", account).split("&")[1])
# POST请求，
re = self.se.post(url=self.url, headers=self.header, data=self.body, proxies=self.proxies,
allow_redirects=False, **kwargs)
# 判断请求
assert "UAP_accessToken" in re.headers["Set-Cookie"]
assert "UAP_refreshToken" in re.headers["Set-Cookie"]
assert "UAP_loginPath" in re.headers["Set-Cookie"]
print("登录接口验证成功")
return re
def useHeadersRequests(self, method=None, url=None, headers=None, data=None, **kwargs):
"""
模拟客户端接口操作，使用headers进行cookies传递,调用requests库进行接口访问
:param method:接口请求方式，POST，GET等
:param url:接口路径
:param data:接口数据
：kwargs：其他requests.request()支持参数可以直接传递
"""
print(url)
# print(headers)
print(data)
log.info("{0:=^86}".format(''))
log.info("{}\n{}\n{}\n".format(url, data, kwargs))
if "gateway" in url:
# 判断接口路径，通过接口路径钟是否包含gateway来判定接口是否是由客户端进行访问，来判定headers使用
self.clientHeaders = dict(self.clientHeaders, **headers)
res = requests.request(method=method, url=url, data=data, headers=self.clientHeaders, **kwargs)
else:
self.webHeaders = dict(self.webHeaders, **headers)
res = requests.request(method=method, url=url, data=data, headers=self.webHeaders, **kwargs)
return res
def get(self, url, **kwargs):
"""Sends a GET request. Returns :class:`Response` object.
:param url: URL for the new :class:`Request` object.
:param \*\*kwargs: Optional arguments that ``request`` takes.
:rtype: requests.Response
"""
# 记录日志
log.info("{0:=^86}".format(''))
log.info("{}\n{}\n".format(url, kwargs))
# 进行请求
re = self.se.get(url, **kwargs, proxies=self.proxies, timeout=self.timeout)
return re
def post(self, url, data=None, json=None, **kwargs):
"""Sends a POST request. Returns :class:`Response` object.
:param url: URL for the new :class:`Request` object.
:param data: (optional) Dictionary, list of tuples, bytes, or file-like
object to send in the body of the :class:`Request`.
:param json: (optional) json to send in the body of the :class:`Request`.
:param \*\*kwargs: Optional arguments that ``request`` takes.
:rtype: requests.Response
"""
# 记录日志
log.info("{0:=^86}".format(''))
log.info("{}\n{}\n{}\n{}".format(url, data, json, kwargs))
# 进行请求
re = self.se.post(url, data=data, json=json, proxies=self.proxies, **kwargs, timeout=self.timeout)
return re
def options(self, url, **kwargs):
"""Sends a OPTIONS request. Returns :class:`Response` object.
:param url: URL for the new :class:`Request` object.
:param \*\*kwargs: Optional arguments that ``request`` takes.
:rtype: requests.Response
"""
# 记录日志
log.info(url, kwargs)
# 进行请求
re = self.se.options(url, **kwargs)
return re
def head(self, url, **kwargs):
"""Sends a HEAD request. Returns :class:`Response` object.
:param url: URL for the new :class:`Request` object.
:param \*\*kwargs: Optional arguments that ``request`` takes.
:rtype: requests.Response
"""
# 记录日志
log.info(url, kwargs)
# 进行请求
re = self.se.head(url, **kwargs)
return re
def put(self, url, data=None, **kwargs):
"""Sends a PUT request. Returns :class:`Response` object.
:param url: URL for the new :class:`Request` object.
:param data: (optional) Dictionary, list of tuples, bytes, or file-like
object to send in the body of the :class:`Request`.
:param \*\*kwargs: Optional arguments that ``request`` takes.
:rtype: requests.Response
"""
# 记录日志
log.info(url, data, kwargs)
# 进行请求
re = self.se.put(url, data, **kwargs)
return re
def delete(self, url, **kwargs):
"""Sends a DELETE request. Returns :class:`Response` object.
:param url: URL for the new :class:`Request` object.
:param \*\*kwargs: Optional arguments that ``request`` takes.
:rtype: requests.Response
"""
# 记录日志
log.info(url, kwargs)
# 进行请求
re = self.se.delete(url, **kwargs)
return re
def set_mark(self):
"用户自定义优先级方法"
mark_list = eval(com.get_global_config("global_data", "mark", "list"))
print("预设运行标记：", mark_list)
global _global_mark
if type(mark_list) == type([]) and len(mark_list) != 0:
_global_mark = mark_list
return _global_mark
elif type(mark_list) == type([]) and len(mark_list) == 0:
_global_mark = False
return _global_mark
else:
raise Exception("error,pls check mark data")
def mark(self, m=None):
try:
# 没有输入任何预设值,默认跑全部
if _global_mark == False:
return True
# 输入预设值且未标记用例,默认跑全部
elif _global_mark != False and (m in _global_mark):
return True
else:
return False
except Exception as e:
return False
my = mysession()
my.set_mark()
my.cliLogin()
s1 = my.get_session("sg_user01")
s2 = my.get_session("sg_user02")
# sg1=my.get_session("sg_user01")
# sy1=my.get_session("sy_user01")