robotframework-requests--中文註解版
最近工作原因在研究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--中文註解版