标签：%s main create 代理服务 utf-8 创建 get请求 debug utf8
最近工作原因在研究RobotFramework对REST测试的方案，找到几个相关类库。但使用requests感觉更方便，研究了一下requests类库的源码，并将注释换成中文为方便使用。关于Requests类库的源码找到两个版本，VCR版本，另一个下面这个算是普通版吧。关于VCR部分的内容还在研究，稍后发出来。
[lang=python]
#! /url/bin/python
# -*- coding: UTF-8 -*-
import requests import
import json
from urllib import urlencode
from robot.libraries.BuiltIn import BuiltIn
from robot import utils
try:
from requests_ntlm import HttpNtlmAuth
except ImportError:
pass
class RestKeywords(object):
ROBOT_LIBRARY_SCOPE = ‘Global‘
def __init__(self):
self._cache = utils.ConnectionCache(u‘未创建Sessions‘)
self.builtin = BuiltIn()
def _utf8_urlencode(self, data):
if isinstance(data, unicode):
return data.encode(‘utf-8‘)
if not isinstance(data, dict):
return data
utf8_data = {}
for k, v in data.iteritems():
utf8_data[k] = unicode(v).encode(‘utf-8‘)
return urlencode(utf8_data)
def _create_session(self, alias, url, headers, cookies,
auth, timeout, proxies, verify):
""" Create Session: 创建一个连接server的HTTP会话
`url` 服务器基本的URL地址
`alias` Robot Framework 对当前创建的HTTP会话指定的别名
`headers` 默认的headers字典
`auth` 对于HTTP基本身份验证词典的用户名和密码
`timeout` 连接超时
`proxies` 代理服务器URL
`verify` 如果请求验证证书，该值设置为True
"""
self.builtin.log(u‘创建Session：%s‘ % alias, ‘DEBUG‘)
session = requests.Session()
session.headers.update(headers)
session.auth = auth if auth else session.auth
session.proxies = proxies if proxies else session.proxies
session.verify = self.builtin.convert_to_boolean(verify)
# 通过传递进入说请求
# cant pass these into the Session anymore
self.timeout = timeout
self.cookies = cookies
self.verify = verify
# cant use hooks:(
session.url = url
self._cache.register(session, alias=alias)
return session
def create_session(self, alias, url, headers={}, cookies=None,
auth=None, timeout=None, proxies=None, verify=False):
""" Create Session: 创建一个连接server的HTTP会话
`url` 服务器基本的URL地址
`alias` Robot Framework 对当前创建的HTTP会话指定的别名
`headers` 默认的headers字典
`auth` 对于HTTP基本身份验证词典的[‘DOMAIN‘, ‘用户名‘, ‘密码‘]
`timeout` 连接超时
`proxies` 代理服务器URL
`verify` 如果请求验证证书，该值设置为True
"""
auth = request.auth.HTTPBasicAuth(*auth) if auth else None
return self._create_session(alias, url, headers, cookies, auth,
timeout, proxies, verify)
def create_ntln_session(self, alias, url, headers={}, cookies=None,
auth=None, timeout=None, proxies=None, verify=False):
""" Create Session: 创建一个连接server的HTTP会话
`url` 服务器基本的URL地址
`alias` Robot Framework 对当前创建的HTTP会话指定的别名
`headers` 默认的headers字典
`auth` 对于HTTP基本身份验证词典的[‘DOMAIN‘, ‘用户名‘, ‘密码‘]
`timeout` 连接超时
`proxies` 代理服务器URL
`verify` 如果请求验证证书，该值设置为True
"""
if not HttpNtlmAuth:
raise AssertionError(u‘未加载Reuqest HTLM模块‘)
elif len(auth) != 3:
raise AssertionError(u‘验证参数数量不正确 ‘
u‘- 预期3个参数, 实际{}‘.format(len(auth)))
else:
ntlm_auth = HttpNtlmAuth(‘{}\\{}‘.format(auth[0], auth[1]), auth[2])
return self._create_session(alias, url, headers, cookies,
ntlm_auth, timeout, proxies, verify)
def delete_all_sessions(self):
"""删除所有Session对象"""
self._cache.empty_cache()
def to_json(self, content, pretty_print=False):
"""将字符串转换为一个JSON对象
`content`字符串的内容转换成JSON
`pretty_print`如果被定义，将content输出打印JSON格式
"""
if pretty_print:
json_ = self._json_pretty_print(content)
else:
json_ = json.loads(content)
return json_
def get_request(self, alias, uri, headers=None, params={}, allow_redirects=None):
""" 使用指定别名中找到的会话对象发送GET请求
`uri` 向该uri发送GET 请求
`alias` 将用于识别在高速缓存中的Session 对象
`headers` 请求以字典格式的报头
"""
# 根据alias获取Session对象
session = self._cache.switch(alias)
# 对参数进行encode编码转换
params = self._utf8_urlencode(params)
# 判断是否允许重定向
redir = True if allow_redirects is None else allow_redirects
response = self._get_request(session, uri, headers, params, redir)
return response
def get(self, alias, uri, headers=None, params={}, allow_redirects=None):
""" * * * 警告 - 建议使用get request * * *
使用指定别名中找到的会话对象发送GET请求
`uri` 向该uri发送GET 请求
`alias` 将用于识别在高速缓存中的Session 对象
`headers` 请求以字典格式的报头
"""
print u"警告强烈反对当前使用GET方法提交请求"
session = self._cache.switch(alias)
params = self._utf8_urlencode(params)
redir = True if allow_redirects is None else allow_redirects
response = self._get_request(session, uri, headers, params, redir)
return response
def post_request(self, alias, uri, data={}, headers=None, files={}, allow_redirects=None):
""" 使用指定别名中找到的会话对象发送POST请求
`uri` 向该uri发送POST 请求
`alias` 将用于识别在高速缓存中的Session 对象
`headers` 请求以字典格式的报头
`data` 将原Body内容POST的数据或二进帛数据转换为键值字典，并以urlencoded编码形式发送
`files` 包含数据文件的文件名的字典才能POST到服务器
"""
# 根据alias获取Session对象
session = self._cache.switch(alias)
# 对data进行encode编码转换
data = self._utf8_urlencode(data)
# 判断是否允许重定向
redir = True if allow_redirects is None else allow_redirects
response = self._post_request(session, uri, data, headers, files, redir)
return response
def post(self, alias, uri, data={}, headers=None, files={}, allow_redirects=None):
""" * * * 警告 - 建议使用post request * * *
使用指定别名中找到的会话对象发送POST请求
`uri` 向该uri发送POST 请求
`alias` 将用于识别在高速缓存中的Session 对象
`headers` 请求以字典格式的报头
`data` 将原Body内容POST的数据或二进帛数据转换为键值字典，并以urlencoded编码形式发送
`files` 包含数据文件的文件名的字典才能POST到服务器
"""
print u"警告强烈反对当前使用POST方法提交请求"
session = self._cache.switch(alias)
data = self._utf8_urlencode(data)
redir = True if allow_redirects is None else allow_redirects
response = self._post_request(session, uri, data, headers, files, redir)
return response
def patch_request(self, alias, uri, data={}, headers=None, files={}, allow_redirects=None):
""" 使用指定别名中找到的会话对象发送PATCH请求
`uri` 向该uri发送PATCH请求
`alias` 将用于识别在高速缓存中的Session 对象
`headers` 请求以字典格式的报头
`data` 将原Body内容PATCH的数据或二进帛数据转换为键值字典，并以urlencoded编码形式发送
`files` 包含数据文件的文件名的字典才能PATCH到服务器
"""
session = self._cache.switch(alias)
data = self._utf8_urlencode(data)
redir = True if allow_redirects is None else allow_redirects
response = self._patch_request(session, uri, data, headers, files, redir)
return response
def patch(self, alias, uri, data={}, headers=None, files={}, allow_redirects=None):
""" * * * 警告 - 建议使用patch request * * *
使用指定别名中找到的会话对象发送PATCH请求
`uri` 向该uri发送PATCH请求
`alias` 将用于识别在高速缓存中的Session 对象
`headers` 请求以字典格式的报头
`data` 将原Body内容PATCH的数据或二进帛数据转换为键值字典，并以urlencoded编码形式发送
`files` 包含数据文件的文件名的字典才能PATCH到服务器
"""
print u"警告强烈反对当前使用PATCH方法提交请求"
session = self._cache.switch(alias)
data = self._utf8_urlencode(data)
redir = True if allow_redirects is None else allow_redirects
response = self._patch_request(session, uri, data, headers, files, redir)
return response
def put_request(self, alias, uri, data=None, headers=None, allow_redirects=None):
"""使用指定别名中找到的会话对象发送PUT请求
`uri` 向该uri发送PUT请求
`alias` 将用于识别在高速缓存中的Session 对象
`headers` 请求以字典格式的报头
`data` 将原Body内容PUT的数据或二进帛数据转换为键值字典，并以urlencoded编码形式发送
"""
session = self._cache.switch(alias)
data = self._utf8_urlencode(data)
redir = True if allow_redirects is None else allow_redirects
response = self._put_request(session, uri, data, headers, redir)
return response
def put(self, alias, uri, data=None, headers=None, allow_redirects=None):
"""* * * 警告 - 建议使用put request * * *
使用指定别名中找到的会话对象发送PUT请求
`uri` 向该uri发送PUT请求
`alias` 将用于识别在高速缓存中的Session 对象
`headers` 请求以字典格式的报头
`data` 将原Body内容PUT的数据或二进帛数据转换为键值字典，并以urlencoded编码形式发送
"""
print u"警告强烈反对当前使用put方法提交请求"
session = self._cache.switch(alias)
data = self._utf8_urlencode(data)
redir = True if allow_redirects is None else allow_redirects
response = self._put_request(session, uri, data, headers, redir)
return response
def delete_request(self, alias, uri, data=(), headers=None, allow_redirects=None):
""" 使用指定别名中找到的会话对象发送DELETE请求
`uri` 向该uri发送DELETE请求
`alias` 将用于识别在高速缓存中的Session 对象
`headers` 请求以字典格式的报头
`data` 将原Body内容DELETE的数据或二进帛数据转换为键值字典，并以urlencoded编码形式发送
"""
session = self._cache.switch(alias)
data = self._utf8_urlencode(data)
redir = True if allow_redirects is None else allow_redirects
response = self._delete_request(session, uri, data, headers, redir)
return response
def delete(self, alias, uri, data=(), headers=None, allow_redirects=None):
"""* * * 警告 - 建议使用delete request * * *
使用指定别名中找到的会话对象发送DELETE请求
`uri` 向该uri发送DELETE请求
`alias` 将用于识别在高速缓存中的Session 对象
`headers` 请求以字典格式的报头
`data` 将原Body内容DELETE的数据或二进帛数据转换为键值字典，并以urlencoded编码形式发送
"""
print u"警告强烈反对当前使用delete方法提交请求"
session = self._cache.switch(alias)
data = self._utf8_urlencode(data)
redir = True if allow_redirects is None else allow_redirects
response = self._delete_request(session, uri, data, headers, redir)
return response
def head_request(self, alias, uri, headers=None, allow_redirects=None):
""" 使用指定别名中找到的会话对象发送HEAD请求
`uri` 向该uri发送HEAD请求
`alias` 将用于识别在高速缓存中的Session 对象
`headers` 请求以字典格式的报头
"""
session = self._cache.switch(alias)
redir = False if allow_redirects is None else allow_redirects
response = self._head_request(session, uri, headers, redir)
return response
def head(self, alias, uri, headers=None, allow_redirects=None):
"""* * * 警告 - 建议使用head request * * *
使用指定别名中找到的会话对象发送HEAD请求
`uri` 向该uri发送HEAD请求
`alias` 将用于识别在高速缓存中的Session 对象
`headers` 请求以字典格式的报头
"""
print u"警告强烈反对当前使用head方法提交请求"
session = self._cache.switch(alias)
redir = False if allow_redirects is None else allow_redirects
response = self._head_request(session, uri, headers, redir)
return response
def options_request(self, alias, uri, headers=None, allow_redirects=None):
""" 使用指定别名中找到的会话对象发送OPTIONS请求
`uri` 向该uri发送OPTIONS请求
`alias` 将用于识别在高速缓存中的Session 对象
`headers` 请求以字典格式的报头
"""
session = self._cache.switch(alias)
redir = True if allow_redirects is None else allow_redirects
response = self._options_request(session, uri, headers, redir)
return response
def options(self, alias, uri, headers=None, allow_redirects=None):
"""* * * 警告 - 建议使用options request * * *
使用指定别名中找到的会话对象发送OPTIONS请求
`uri` 向该uri发送OPTIONS请求
`alias` 将用于识别在高速缓存中的Session 对象
`headers` 请求以字典格式的报头
"""
print u"警告强烈反对当前使用options方法提交请求"
session = self._cache.switch(alias)
redir = True if allow_redirects is None else allow_redirects
response = self._options_request(session, uri, headers, redir)
return response
def _get_request(self, session, uri, headers, params, allow_redirects):
resp = session.get(self._get_url(session, uri),
headers=headers,
params=params,
cookies=self.cookies, timeout=self.timeout,
allow_redirects=allow_redirects)
# 存储最后响应对象
session.last_resp = resp
self.builtin.log("GET 响应: %s DEBUG" % resp.content)
return resp
def _post_request(self, session, uri, data, headers, files, allow_redirects):
resp = session.post(self._get_url(session, uri),
data=data, headers=headers,
files=files,
cookies=self.cookies, timeout=self.timeout,
allow_redirects=allow_redirects)
# 存储最后响应对象
session.last_resp = resp
self.builtin.log("POST 响应: %s DEBUG" % resp.content)
return resp
def _patch_request(self, session, uri, data, headers, files, allow_redirects):
resp = session.patch(self._get_url(session, uri),
data=data, headers=headers,
files=files,
cookies=self.cookies, timeout=self.timeout,
allow_redirects=allow_redirects)
# 存储最后响应对象
session.last_resp = resp
self.builtin.log("PATCH 响应: %s DEBUG" % resp.content)
return resp
def _put_request(self, session, uri, data, headers, allow_redirects):
resp = session.put(self._get_url(session, uri),
data=data, heaers=headers,
cookies=self.cookies, timeout=self.timeout,
allow_redirects=allow_redirects)
self.builtin.log("PUT 响应: %s DEBUG" % resp.content)
# 存储最后响应对象
session.last_resp =resp
return resp
def _delete_request(self, session, uri, data, headers, allow_redirects):
resp = session.delete(self._get_url(session, uri), data=data,
headers=headers, cookies=self.cookies,
timeout=self.timeout,
allow_redirects=allow_redirects)
# 存储最后响应对象
session.last_resp = resp
self.builtin.log("DELETE 响应: %s DEBUG" % resp.content)
return resp
def _head_request(self, session, uri, headers, allow_redirects):
resp = session.head(self._get_url(session, uri), headers=headers,
cookies=self.cookies, timeout=self.timeout,
allow_redirects=allow_redirects)
# 存储最后响应对象
session.last_resp = resp
self.builtin.log("HEAD 响应: %s DEBUG" % resp.content)
return resp
def _options_request(self, session, uri, headers, allow_redirects):
resp = session.head(self._get_url(session, uri), headers=headers,
cookies=self.cookies, timeout=self.timeout,
allow_redirects=allow_redirects)
# 存储最后响应对象
session.last_resp = resp
self.builtin.log("OPTIONS 响应: %s DEBUG" % resp.content)
return resp
def _get_url(self, session, uri):
"""根据Helper 模块获取完整URL"""
url = session.url
if uri:
slash = ‘‘ if uri.startswith(‘/‘) else ‘/‘
url = "%s%s%s" % (session.url, slash, uri)
return url
def _json_pretty_print(self, content):
""" 转换为JSON 对象
‘content‘ 将content转换为JSON object
"""
temp = json.loads(content)
return json.dumps(temp, sort_keys=True, indent=4, separators=(‘,‘, ‘: ‘))
robotframework-requests--中文注解版
标签：%s main create 代理服务 utf-8 创建 get请求 debug utf8
原文地址：https://www.cnblogs.com/changyou615/p/8776496.html
