一、session鑒權的處理
1. requests的會話物件
就像一個瀏覽器一樣,它會在同一個會話中自動處理cookie資訊,不需要寫任何額外的代碼,
import requests
session = requests.Session() # 理解為就是一個瀏覽器
type(session)
requests.sessions.Session
session.post()# 登錄
session.get() # 獲取某個資料,會自動攜帶上一步收到的cookie
# 課堂派案例
headers = {'cookie': 'FZ_STROAGE.ketangpai.com=eyJTRUVTSU9OSUQiOiIzMTI5MjRiNTU2MzNmMDUxIiwiU0VFU0lPTkRBVEUiOjE2Mzk1NzA0NDQ3Njd9; ARK_ID=undefined; ketangpai_home_slb=3fbda3fc94d5d1be63720d9c156288d0; PHPSESSID=kmugv5id4lcecie33asikt3p96; ketangpai_home_remember=think%3A%7B%22username%22%3A%22MDAwMDAwMDAwMLV2x5eHz7dthN523LWtftmC0IDak4NubQ%22%2C%22expire%22%3A%22MDAwMDAwMDAwMLOGvd6IubtrhKiGl7G2dZ4%22%2C%22token%22%3A%22MDAwMDAwMDAwMMurrpWavLehhs1-lbO5hZWEzYfcepuomcWmmqaMiHtnr5ypzYWosKKZq6HQxtOK0ZCme5p-q6iZu2yrn4uNhJ3KedDYk7ivboS4jt6zuY2Ug6eDl36KYW0%22%2C%22sign%22%3A%2207f1bd0c97817e6d7ebe92bfe8e30fe9%22%7D',
'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/94.0.4606.61 Safari/537.36'}
res = requests.get(url='https://v4.ketangpai.com/UserApi/getUserInfo')
res.status_code
200
res.cookies
<RequestsCookieJar[Cookie(version=0, name='PHPSESSID', value='https://www.cnblogs.com/Nephalem-262667641/archive/2023/04/14/krm5vua2f6f07m5rjipa0uti16', port=None, port_specified=False, domain='v4.ketangpai.com', domain_specified=False, domain_initial_dot=False, path='/', path_specified=True, secure=True, expires=None, discard=True, comment=None, comment_url=None, rest={'HttpOnly': None}, rfc2109=False), Cookie(version=0, name='ketangpai_home_slb', value='https://www.cnblogs.com/Nephalem-262667641/archive/2023/04/14/3fbda3fc94d5d1be63720d9c156288d0', port=None, port_specified=False, domain='v4.ketangpai.com', domain_specified=False, domain_initial_dot=False, path='/', path_specified=True, secure=True, expires=None, discard=True, comment=None, comment_url=None, rest={'httponly': None}, rfc2109=False)]>
res.json()
{'status': 0, 'info': '您還未登陸!'}
session = requests.Session() # 1. 創建一個session物件
# 2. 登錄
login_url = 'https://v4.ketangpai.com/UserApi/login'
data = https://www.cnblogs.com/Nephalem-262667641/archive/2023/04/14/{'email': '877***[email protected]',
'password': 'Pyt***inlan',
'remember': 0}
# json data params
response = session.post(url=login_url, data=https://www.cnblogs.com/Nephalem-262667641/archive/2023/04/14/data)
session.cookies
<RequestsCookieJar[Cookie(version=0, name='PHPSESSID', value='https://www.cnblogs.com/Nephalem-262667641/archive/2023/04/14/5if12vo96vtulhfhr9bvu1nnr2', port=None, port_specified=False, domain='v4.ketangpai.com', domain_specified=False, domain_initial_dot=False, path='/', path_specified=True, secure=True, expires=None, discard=True, comment=None, comment_url=None, rest={'HttpOnly': None}, rfc2109=False), Cookie(version=0, name='ketangpai_home_remember', value='https://www.cnblogs.com/Nephalem-262667641/archive/2023/04/14/think%3A%7B%22username%22%3A%22MDAwMDAwMDAwMLV2x5eHz7dthN523LWtftmC0IDak4NubQ%22%2C%22expire%22%3A%22MDAwMDAwMDAwMLOGvd6IubtrhKigl7O2dZ4%22%2C%22token%22%3A%22MDAwMDAwMDAwMMurrpWavLehhs1-lbO5hZWEzYfcepuomcWmmqaMiHtnr5ypzYWosKKZq6HQxtOK0ZCme5p-q6iZu2yrn4uNhJ3KedDYk7ivboS4jt6zuY2Ug6edl4CKYW0%22%2C%22sign%22%3A%2298880a4b0ee67193316c6c40dd40441f%22%7D', port=None, port_specified=False, domain='v4.ketangpai.com', domain_specified=False, domain_initial_dot=False, path='/', path_specified=True, secure=True, expires=1639581779, discard=False, comment=None, comment_url=None, rest={'httponly': None}, rfc2109=False), Cookie(version=0, name='ketangpai_home_slb', value='https://www.cnblogs.com/Nephalem-262667641/archive/2023/04/14/3fbda3fc94d5d1be63720d9c156288d0', port=None, port_specified=False, domain='v4.ketangpai.com', domain_specified=False, domain_initial_dot=False, path='/', path_specified=True, secure=True, expires=None, discard=True, comment=None, comment_url=None, rest={'httponly': None}, rfc2109=False)]>
res = session.get(url='https://v4.ketangpai.com/UserApi/getUserInfo')
res.json()
{'status': 1,
'data': {'id': 'MDAwMDAwMDAwMLSGz96Iqb9phLVyoQ',
'username': '****',
'avatar': 'http://v4.ketangpai.com/Public/Common/img/40/26.png',
'usertype': '1',
'email': '877***[email protected]',
'stno': '',
'atteststate': 0,
'attestInfo': []}}
# 如果不用session物件,每一步都需要自己處理cookie
login_url = 'https://v4.ketangpai.com/UserApi/login'
data = https://www.cnblogs.com/Nephalem-262667641/archive/2023/04/14/{'email': 877***[email protected]',
'password': 'Pyt***inlan',
'remember': 0}
# 1.登錄
response = requests.post(url=login_url, data=https://www.cnblogs.com/Nephalem-262667641/archive/2023/04/14/data)
response.status_code
200
response.json()
{'status': 1,
'url': '/Main/index.html',
'token': 'MDAwMDAwMDAwMMurrpWavLehhs1-lbO5hZWEzYfcepuomcWmmqaMiHtnr5ypzYWosKKZq6HQxtOK0ZCme5p-haiZu2yrn4uNhJ3KedDYk7ivboS4jt6zuY2Ug7d33n96YW0',
'isenterprise': 0,
'uid': 'MDAwMDAwMDAwMLSGz96Iqb9phLVyoQ'}
# 2.獲取資料
res = requests.get(url='https://v4.ketangpai.com/UserApi/getUserInfo', cookies=response.cookies)
res.json()
{'status': 1,
'data': {'id': 'MDAwMDAwMDAwMLSGz96Iqb9phLVyoQ',
'username': '****',
'avatar': 'http://v4.ketangpai.com/Public/Common/img/40/26.png',
'usertype': '1',
'email': '877***[email protected]',
'stno': '',
'atteststate': 0,
'attestInfo': []}}
requests庫的session物件僅僅只是自動幫我們處理了cookie的攜帶問題,
2. 封裝處理session鑒權的http請求函式
思路:
- 保證在一個會話中使用同一個會話物件即可
- 給每一個用例類創建一個會話物件即可,
import json
import unittest
from jsonpath import jsonpath
import setting
from common import logger, db
from common.data_handler import (
replace_args_by_re,
generate_no_usr_phone)
from common.encrypt_handler import generate_sign
import requests
class BaseCase(unittest.TestCase):
"""
用例基類
"""
db = db
logger = logger
setting = setting
name = 'base_case'
session = requests.session() # 創建一個session物件用來處理session鑒權
@classmethod
def setUpClass(cls) -> None:
cls.logger.info('==========={}介面開始測驗==========='.format(cls.name))
@classmethod
def tearDownClass(cls) -> None:
cls.logger.info('==========={}介面結束測驗==========='.format(cls.name))
def flow(self, item):
"""
測驗流程
"""
self.logger.info('>>>>>>>用例{}開始執行>>>>>>>>'.format(item['title']))
# 把測驗資料系結到方法屬性case上,其他也要把一些變數系結到物件的屬性上
self._case = item
# 1. 處理測驗資料
self.process_test()
# 2. 發送請求
self.send_request()
# 3. 斷言
self.assert_all()
def process_test(self):
"""
測驗資料的處理
"""
# 1.1 生成測驗資料
self.generate_test_data()
# 1.2 替換依賴引數
self.replace_args()
# 1.3 處理url
self.process_url()
# 1.4 鑒權處理
self.auth_process()
def auth_process(self):
"""
v3版本鑒權處理
:return:
"""
request_data = https://www.cnblogs.com/Nephalem-262667641/archive/2023/04/14/self._case.get('request_data')
if request_data:
headers = request_data.get('headers')
if headers:
if headers.get('X-Lemonban-Media-Type') == 'lemonban.v3':
# 獲取token
token = self._case['request_data']['headers']['Authorization'].split(' ')[-1]
# 生成簽名
sign, timestamp = generate_sign(token, self.setting.SERVER_RSA_PUB_KEY)
# 添加到請求資料中
self._case['request_data']['json']['sign'] = sign
self._case['request_data']['json']['timestamp'] = timestamp
# if self._case['request_data']['headers']['X-Lemonban-Media-Type'] == 'lemonban.v3':
# # 獲取token
# token = self._case['request_data']['headers']['Authorization'].split(' ')[-1]
# # 生成簽名
# sign, timestamp = generate_sign(token, self.setting.SERVER_RSA_PUB_KEY)
# # 添加到請求資料中
# self._case['request_data']['json']['sign'] = sign
# self._case['request_data']['json']['timestamp'] = timestamp
def generate_test_data(self):
"""
生成測驗資料
"""
"""
生成測驗資料,不是固定流程,有不同可以復寫
:return:
"""
self._case = json.dumps(self._case)
if '$phone_number$' in self._case:
phone = generate_no_usr_phone()
self._case = self._case.replace('$phone_number$', phone)
self._case = json.loads(self._case)
def replace_args(self):
"""
替換引數
"""
self._case = json.dumps(self._case) # 把用例資料dumps成字串,一次替換
self._case = replace_args_by_re(self._case, self)
self._case = json.loads(self._case)
# 再將request_data, expect_data loads為字典
try:
self._case['request_data'] = json.loads(self._case['request_data'])
self._case['expect_data'] = json.loads(self._case['expect_data'])
except Exception as e:
self.logger.error('{}用例的測驗資料格式不正確'.format(self._case['title']))
raise e
def process_url(self):
"""
處理url
"""
if self._case['url'].startswith('http'):
# 是否是全地址
pass
elif self._case['url'].startswith('/'):
# 是否是短地址
self._case['url'] = self.setting.PROJECT_HOST + self._case['url']
else:
# 介面名稱
try:
self._case['url'] = self.setting.INTERFACES[self._case['url']]
except Exception as e:
self.logger.error('介面名稱:{}不存在'.format(self._case['url']))
raise e
def send_request(self):
"""
發送請求
@return:
"""
self._response = self.session.request(
url=self._case['url'], method=self._case['method'], **self._case['request_data'])
# self._response = send_http_request(url=self._case['url'], method=self._case['method'],
# **self._case['request_data'])
def assert_all(self):
try:
# 3.1 斷言回應狀態碼
self.assert_status_code()
# 3.2 斷言回應資料
self.assert_response()
# 回應結果斷言成功后就提取依賴資料
self.extract_data()
# 3.3 資料庫斷言后面的任務
self.assert_sql()
except Exception as e:
self.logger.error('++++++用例{}測驗失敗'.format(self._case['title']))
raise e
else:
self.logger.info('<<<<<<<<<用例{}測驗成功<<<<<<<'.format(self._case['title']))
def assert_status_code(self):
"""
斷言回應狀態碼
"""
try:
self.assertEqual(self._case['status_code'], self._response.status_code)
except AssertionError as e:
self.logger.warning('用例【{}】回應狀態碼斷言例外'.format(self._case['title']))
raise e
else:
self.logger.info('用例【{}】回應狀態碼斷言成功'.format(self._case['title']))
def assert_response(self):
"""
斷言回應資料
"""
if self._case['res_type'].lower() == 'json':
res = self._response.json()
elif self._case['res_type'].lower() == 'html':
# 擴展思路
res = self._response.text
try:
self.assertEqual(self._case['expect_data'], {'code': res['code'], 'msg': res['msg']})
except AssertionError as e:
self.logger.warning('用例【{}】回應資料斷言例外'.format(self._case['title']))
self.logger.warning('用例【{}】期望結果為:{}'.format(self._case['title'], self._case['expect_data']))
self.logger.warning('用例【{}】的回應結果:{}'.format(self._case['title'], res))
raise e
else:
self.logger.info('用例【{}】回應資料斷言成功'.format(self._case['title']))
def assert_sql(self):
"""
斷言資料庫
"""
if self._case.get('sql'): # 回傳指定鍵的值,如果鍵不在字典中回傳默認值 None 或者設定的默認值,
# 只有sql欄位有sql的才需要校驗資料庫
# 只有sql欄位有sql的才需要校驗資料庫
sqls = self._case['sql'].split(',')
for sql in sqls:
try:
self.assertTrue(self.db.exist(sql))
except AssertionError as e:
self.logger.warning('用例【{}】資料庫斷言例外,執行的sql為:{}'.format(self._case['title'], sql))
raise e
except Exception as e:
self.logger.warning('用例【{}】資料庫斷言例外,執行的sql為:{}'.format(self._case['title'], sql))
raise e
def extract_data(self):
"""
根據提取運算式提取對應的資料
:return:
"""
if self._case.get('extract'):
if self._case['res_type'].lower() == 'json':
self.extract_data_from_json()
elif self._case['res_type'].lower() == 'html':
self.extract_data_from_html()
elif self._case['res_type'].lower() == 'xml':
self.extract_data_from_xml()
else:
raise ValueError('res_type型別不正確,只支持json,html,xml')
def extract_data_from_json(self):
"""
從json資料中提取資料并系結到類屬性中
:return:
"""
try:
rules = json.loads(self._case.get('extract'))
except Exception as e:
raise ValueError('用例【{}】的extract欄位資料:{}格式不正確'.format(self._case['title'], self._case['extract']))
for rule in rules:
# 類屬性名
name = rule[0]
# 提取運算式
exp = rule[1]
# 根據jsonpath去回應中提取值
value = https://www.cnblogs.com/Nephalem-262667641/archive/2023/04/14/jsonpath(self._response.json(), exp)
# 如果能提取到值
if value:
# 把值系結到對應的類屬性上
setattr(self.__class__, name, value[0]) # 注意value是個串列
else:
# 提取不到值,說明jsonpath寫錯了,或者是回應又問題
raise ValueError('用例【{}】的提取運算式{}提取不到資料'.format(self._case['title'], self._case['extract']))
def extract_data_from_html(self):
"""
從html資料中提取資料并系結到類屬性中
:return:
"""
raise ValueError('請實作此方法')
def extract_data_from_xml(self):
"""
從xml資料中提取資料并系結到類屬性中
:return:
"""
raise ValueError('請實作此方法')
from unittestreport import ddt, list_data
from common.base_case import BaseCase
cases = [
{'title': '課堂派登錄',
'method': 'post',
'url': 'https://v4.ketangpai.com/UserApi/login',
'request_data': '{"data": {"email": "877***[email protected]", "password": "Pyt***inlan", "remember": 0}}',
'status_code': 200,
'res_type': 'json',
'expect_data': '{"status": 1}'
},
{
'title': '獲取用戶資訊',
'method': 'get',
'url': 'https://v4.ketangpai.com/UserApi/getUserInfo',
'request_data': '{}',
'status_code': 200,
'res_type': 'json',
'expect_data': '{"status": 1}'
}
]
@ddt
class TestCourseFlow(BaseCase):
name = "課堂派測驗"
@list_data(cases)
def test_course(self, item):
self.flow(item)
def assert_response(self):
"""
復寫assert_response方法實作課堂派的斷言
:return:
"""
if self._case['res_type'].lower() == 'json':
res = self._response.json()
elif self._case['res_type'].lower() == 'html':
# 擴展思路
res = self._response.text
try:
self.assertEqual(self._case['expect_data'], {'status': res['status']})
except AssertionError as e:
self.logger.warning('用例【{}】回應資料斷言例外'.format(self._case['title']))
self.logger.warning('用例【{}】期望結果為:{}'.format(self._case['title'], self._case['expect_data']))
self.logger.warning('用例【{}】的回應結果:{}'.format(self._case['title'], res))
raise e
else:
self.logger.info('用例【{}】回應資料斷言成功'.format(self._case['title']))
if __name__ == '__main__':
BaseCase.unittest.main()
本文來自博客園,作者:奈非天,轉載請注明原文鏈接:https://www.cnblogs.com/Nephalem-262667641/p/17318278.html
轉載請註明出處,本文鏈接:https://www.uj5u.com/qita/550146.html
標籤:其他
上一篇:Huffman實作
