requests详解与封装
一、 基本概念
requests 模块是 python 基于 urllib,采用 Apache2 Licensed 开源协议的 HTTP 库。它比 urllib 更加方便,可以节约我们大量的工作,完全满足 HTTP 测试需求。
二、 安装
通过 pip install requests
安装 requests 库
三、快速入门
(一)导包
import requests
(二)常用HTTP请求
方法 | 说明 |
---|---|
GET | 请求获取URL位置的资源 |
HEAD | 请求获取URL位置资源的响应消息报告,即获得资源的头部信息 |
POST | 请求向URL位置的资源后附加新的消息 |
PUT | 请求向URL位置存储一个资源,覆盖原URL位置的资源 |
PATCH | 请求局部更新URL位置的资源,即改变该处资源的部分内容 |
DELETE | 请求删除URL位置存储的资源 |
GET
,HEAD
是从服务器获取信息到本地,一般用于获取资源等场景
PUT
,POST
,PATCH
,DELETE
是从本地向服务器提交信息。一般用于操作数据等场景其中
GET
,POST
为常用方法,本期重点讲解
代码演示:
import requests
requsts.requst()
requsts.get()
requsts.post()
requsts.head()
requsts.put()
requsts.patch()
requsts.delete()
(三)Get详解
常用参数
参数 | 类型 | 作用 |
---|---|---|
params | 字典 | url为基准的url地址,不包含查询参数;该方法会自动对params字典编码,然后和url拼接 |
url | 字符串 | requests 发起请求的地址 |
headers | 字典 | 请求头,发送请求的过程中请求的附加内容携带着一些必要的参数 |
cookies | 字典 | 携带登录状态 |
proxies | 字典 | 用来设置代理 ip 服务器 |
timeout | 整型 | 用于设定超时时间, 单位为秒 |
代码演示:
import requests
resp = requests.get(url="https://www.baidu.com")
print(resp)
(四)Post详解
常用参数
参数 | 类型 | 作用 |
---|---|---|
data | 字典 | 作为向服务器提供或提交资源时提交,主要用于 post 请求 |
json | 字典 | json格式的数据, json合适在相关的html |
files | 文件 | 向服务器接口提交文件数据 |
files演示:
import requests
file = {'file': open('C://Users//hello.txt', 'rb')}
res = requests.post(url="http://localhost:8081/uploadfile",files=file)
print(r.json())
data和json主要区别是请求数据的不同,data一般是表单数据,json是字典格式。
(五)response详解
属性 | 说明 |
---|---|
resp.status_code | http请求的返回状态,若为200则表示请求成功。 |
resp.raise_for_status() | 该语句在方法内部判断resp.status_code 是否等于200,如果不等于,则抛出异常 |
resp.text | http响应内容的字符串形式,即返回的页面内容 |
resp.encoding | 从http header 中猜测的相应内容编码方式 |
resp.apparent_encoding | 从内容中分析出的响应内容编码方式(备选编码方式) |
resp.content | http响应内容的二进制形式 |
resp.json() | 得到对应的 json 格式的数据,类似于字典 |
源码分析:requests请求调用的是session请求,session和requests请求的区别在于,Session可以自动管理cookie,而requests在需要cookie认证时,请求需要携带cookies参数。
四、请求、基础路径封装
SendUtil:
class SendUtil:
session = requests.Session()
# 初始化时,就获得项目名称,和环境名称
def __init__(self, team, workspace):
self.url = YamlUtil.read_yaml(team, workspace)
# 根据数据文件的绝对路径,在调用时,拼接上具体的模块名,即可完成接口拼接
def send(self, method, url, **kwargs):
url = self.url + url
# 将请求大小写统一设置为小写
method = str(method).lower()
# 多参数可以传入data,json,cookie等
res = self.session.request(method=method, url=url, **kwargs)
print(f"当前环境是:{YamlUtil.now_workspace}:{url}")
return res
1、将请求方法method、请求url,以及**kwargs封装成形参,在外部调用请求时,必须传入参数,内部使用session管理请求,达到cookie管理效果2、session为类属性,方便不同的方法去调用该session,防止产生资源浪费
3、在SendUtil类初始化时传入参数,传入的team为项目名称,workspace为环境名称,通过封装read_yaml去读取Yaml文件,随后将读取的数据传递给send请求方法的url,完成基础路径的拼接,这里这样做的意义是每个项目都有固定的基础路径,后面拼接的为具体的模块名,通过封装基础路径在调用时再拼接。
YamlUtil:
class YamlUtil:
# 获得数据文件的绝对路径
now_workspace = None
@classmethod
def get_path(cls):
return os.path.dirname(os.path.dirname(__file__))
# 根据数据文件的绝对路径获取数据,第一个参数表示是哪个项目,第二个表示是什么环境
@classmethod
def read_yaml(cls, team, workspace):
path = cls.get_path()+"/testcases/API/config.yaml"
with open(path, mode="r", encoding="utf-8") as file:
data = yaml.load(file, Loader=yaml.FullLoader)
cls.now_workspace = workspace
return data["BASE"][team][workspace]
1、get_path为获取数据文件绝对路径方法,在实际项目中,尽量使用绝对路径去拼接,避免资源找不到错误(FileNotFoundError),read_yaml方法需要传入两个参数,第一个是项目名,第二个是环境,通过绝对路径去打开yaml文件,然后返回数据的时候根据我们的项目名和环境名去匹配对应的基础路径
config.yaml:
BASE:
RuoYi:
DEV: http://8.129.162.221:login
TEST: http://8.129.162.221:getInfo
Shop:
DEV: http://8.129.162.220:upload
TEST: http://8.129.162.220:download
使用yaml格式去定义项目的基础路径,在read_yaml读取到项目路径后,返回项目和操作环境即可在跑自动化时一目了然。
五、数据读取封装
ReadYamlUtil:
def read_yaml(path):
with open(path, mode="r", encoding="utf-8") as file:
data = yaml.load(file, Loader=yaml.FullLoader)
return data
通过传入path读取Yaml文件内容,随后返回内容,注意要使用encoding
data.yaml:
-
name: 登录接口
description: 验证登录模块
request:
method: POST
url: http://8.129.162.221:xxx/login
data:
code: 1234
password: ****
username: cola
uuid: 63b2234b8*******3cded6dbedd739
validate: None
-
test_login:
@allure.epic("项目名称:若依管理系统")
@allure.feature("模块名称:登录模块")
class Test:
@allure.story("登录模块")
@allure.severity(allure.severity_level.BLOCKER)
@pytest.mark.run(order=2)
@pytest.mark.smoke
@pytest.mark.parametrize("data", read_yaml("data.yaml"))
def test_baidu01(self, data):
allure.dynamic.title(data["name"])
allure.dynamic.description(data["description"])
# 未封装基础路径之前,使用我们yaml读取url
# res = SendUtil.send(method=data["request"]["method"],
# url=data["request"]["url"], json=data["request"]["data"])
# 封装了基础路径之后的用法,通过具体模块名拼接基础路径
res = SendUtil("RuoYi", "DEV").send(method=data["request"]["method"], url="/login", json=data["request"]["data"])
print(res.json()["msg"])
assert res.json()["msg"] == "操作成功"
六、接口关联参数封装
ReadYaml:
接口关联的参数,如果每一个都放到全局变量中代码是非常冗余的,于是我们将其封装起来,常见办法有两种,一种是通过封装方法存入数据库,另一种方法是通过封装方法写入Yaml文件。
# 读取接口关联数据
@classmethod
def read_extract(cls, param="Authorization"):
path = cls.get_path()+"/testcases/API/extract.yaml"
with open(path, encoding="utf-8") as file:
data = yaml.load(stream=file, Loader=yaml.FullLoader)
return data[param]
# 写入接口关联数据
@classmethod
def write_extract(cls, data):
path = cls.get_path()+"/testcases/API/extract.yaml"
with open(path, mode="a", encoding="utf-8") as file:
yaml.dump(data=data, stream=file, allow_unicode=True)
# 清空接口关联数据
@classmethod
def clear_extract(cls):
path = cls.get_path() + "/testcases/API/extract.yaml"
with open(path, mode="w", encoding="utf-8") as file:
file.truncate()
test_login:
在登录模块成功登录之后,通过write_extract方法以字典的格式写入Yaml文件保存,注意token格式,例如我的项目token需要在前置加上"Bearer "
@allure.story("登录模块")
@allure.severity(allure.severity_level.BLOCKER)
@pytest.mark.run(order=1)
@pytest.mark.smoke
@pytest.mark.parametrize("data", YamlUtil.read_yaml("./testcases/API/data.yaml"))
def test_login(self, data):
allure.dynamic.title(data["name"])
allure.dynamic.description(data["description"])
# 封装了基础路径之后的用法,通过具体模块名拼接基础路径
res = SendUtil("RuoYi", "DEV").send(method=data["request"]["method"], url=data["request"]["url"], json=data["request"]["data"])
print(res.json())
# 登录成功后,写入关联数据token
YamlUtil.write_extract({"Authorization": "Bearer "+res.json()["token"]})
assert res.json()["msg"] == "操作成功"
在写入Yaml文件后,我们的token得到了保存,在其他接口调用时,即可通过读取yaml方法,获得token的值
@allure.story("获取个人信息模块")
@allure.severity(allure.severity_level.BLOCKER)
@pytest.mark.run(order=2)
@pytest.mark.smoke
@pytest.mark.parametrize("data", YamlUtil.read_yaml("./testcases/API/getInfo.yaml"))
def test_getinfo(self, data):
allure.dynamic.title(data["name"])
allure.dynamic.description(data["description"])
# 封装了基础路径之后的用法,通过具体模块名拼接基础路径
res = SendUtil("RuoYi", "DEV").send(method=data["request"]["method"],
url=data["request"]["url"],
headers=data["request"]["headers"])
print(res.json()["msg"])
assert res.json()["msg"] == "操作成功"
问题来了,当我们在进行参数化的时候,Yaml文件会产生多条Token决此办法需要通过conftest.py文件,每次调用前清空内容即可
@pytest.fixture(scope="session", autouse=True, name="fixture")
def execute_sql():
# 每次运行之前都清空接口关联的数据,否则每次执行都会产生相同的数据,久而久之数据会变得非常大
YamlUtil.clear_extract()
由于我们项目中不止一个接口会需要用到接口参数关联,可能还有需要参数需要关联,因此我对read_yaml方法进行了优化,可以根据关联参数的键进行配对,找到需要关联的参数
ReadYamlUtil:
param参数默认是token登录鉴权,如果添加了其他关联参数,传递的参数会覆盖此参数,如果不传参数会报错,因此选择了默认参数。
# 读取接口关联数据
@classmethod
def read_extract(cls, param="Authorization"):
path = cls.get_path()+"/testcases/API/extract.yaml"
with open(path, mode="r", encoding="utf-8") as file:
data = yaml.load(file, Loader=yaml.FullLoader)
data_format = {param: data[param]}
return data_format
为了实现方便管理,将来会把用例全部写入Yaml文件中,让不懂代码的人也可以跑接口自动化,因此需要将接口关联参数读取进行优化,通过
{{params}}
对关联参数进行修饰,在读取时,通过正则表达式去除两边括弧,然后根据中间关键词去读取接口关联参数。
getInfo.yaml
-
name: 获取个人信息接口
description: 个人信息模块
request:
method: GET
url: /getInfo
headers: '{{Authorization}}'
validate: None
SendUtil.py(重点)
class SendUtil:
session = requests.Session()
# 初始化时,就获得项目名称,和环境名称
def __init__(self, team, workspace):
self.base_url = YamlUtil.read_configyaml(team, workspace)
self.headers_dict = {}
# 先生成字符串格式,替换URL中的{{}},然后返回原有格式的URL
@classmethod
def replace_value(cls, data):
if data and isinstance(data, dict):
value = json.dumps(data)
else:
value = data
for item in range(0, value.count("{{")):
if "{{" in value and "}}" in value:
start_index = value.index("{{")
end_index = value.index("}}")
old_value = value[start_index:end_index+2]
new_value = YamlUtil.read_extract(old_value[2:-2])
# replace方法只能传字符串,如果是数值,需要转换
if type(new_value) == int:
new_value = str(new_value)
value = value.replace(old_value, new_value)
# 如果不先生成字符串格式,无法进行替换,替换了之后,需要把数据还原成字典格式
if value and isinstance(data, dict):
new_data = json.loads(value)
else:
new_data = value
return new_data
# 根据数据文件的绝对路径,在调用时,拼接上具体的模块名,即可完成接口拼接
def send(self, method, url, headers=None, **kwargs):
# 获取{{param}}参数,替换为关联参数
url = self.base_url + SendUtil.replace_value(url)
# 替换请求头
if headers:
# 取token的value
headers_value = SendUtil.replace_value(headers)
# 取key,拼接成字典格式
headers_key = headers[2: -2]
self.headers_dict = {headers_key: headers_value}
# 替换请求数据
for key, value in kwargs.items():
if key in ['params', 'data', 'json']:
kwargs[key] = self.replace_value(value)
method = str(method).lower()
# 多参数可以传入data,json,cookie等
res = self.session.request(method=method, url=url, headers=self.headers_dict, **kwargs)
print(f"当前环境是:{YamlUtil.now_workspace}:{url}")
return res
七、Yaml用例封装
1.通过自定义Yaml必填规则,让业务人员填写时遵守Yaml编写规则2.判断用例是否需要请求头,有的情况下,需要将请求头读取后添加
3.判断用例是否需要断言
analysis_yaml(分析Yaml):
# yaml测试用例规则约束
def analysis_yaml(self, case):
# 获取yaml用例的所有键
resp = None
case_key = dict(case).keys()
# 判断必填的键是否存在
if 'name' in case_key and 'base_url' in case_key and 'request' in case_key and 'validate' in case_key:
request_key = dict(case['request']).keys()
# 判断request中的method和url是否存在
if 'method' in request_key and 'url' in request_key:
# 获取method和url
method = case['request']['method']
url = case['request']['url']
# 从列表中移除method和url和headers,因为要把可变长度的参数传给send方法的**kwargs
del case['request']['method']
del case['request']['url']
headers = None
# 通过jsonpath判断是否存在请求头
if jsonpath.jsonpath(case, '$.request.headers'):
headers_value = case['request']['headers']
headers = self.replace_load(headers_value)
# 从列表中移除method和url和headers,因为要把可变长度的参数传给send方法的**kwargs
del case['request']['headers']
print("发送请求头为:", headers)
resp = self.send(method=method, url=url, headers=headers, **case['request'])
res_data = resp.json()
res_text = resp.text
if not case['validate'] is None:
self.validate_result(res_data, case['validate'])
elif case['validate'] is None:
print("无需断言")
if jsonpath.jsonpath(case, '$.extract'):
for key, value in dict(case['extract']).items():
# 通过正则表达式提取token
if '(.+?)' in value or '(*.?)' in value:
re_value = re.search(value, res_text)
if re_value:
extract_data = {key: "Bearer "+re_value.group(1)}
YamlUtil.write_extract(extract_data)
# 通过json表达式提取token
else:
extract_data = {key: "Bearer "+res_data[value]}
YamlUtil.write_extract(extract_data)
else:
print("request必填项不能为空")
else:
print("yaml用例必填项不能为空")
return resp
Yaml用例:
-
name: 岗位管理新增接口
description: 系统管理模块
base_url: http://8.129.162.225:8080
request:
method: POST
url: /system/post
json:
postCode: ${get_random_number(50,10000)}
postName: ${get_random_name(1,1000)}
postSort: 0
status: 0
headers: ${get_extract_data(Authorization)}
validate:
code: 200
equals: 操作成功
测试用例(先进行yaml规则校验,然后再发送请求):
@allure.story("岗位关联模块")
@allure.severity(allure.severity_level.BLOCKER)
@pytest.mark.run(order=3)
@pytest.mark.smoke
@pytest.mark.parametrize("data", YamlUtil.read_yaml("setPost.yaml"))
def test_set_post(self, data):
allure.dynamic.title(data["name"])
allure.dynamic.description(data["description"])
# 封装了基础路径之后的用法,通过具体模块名拼接基础路径
res = SendUtil("RuoYi", "DEV").analysis_yaml(data)
print(res.json())
八、Yaml优化—热加载
通过反射方法,将yaml用例中的方法映射到Debug_talk.py文件的方法中,形成参数替换
replace_load:
# 热加载替换方式
@classmethod
def replace_load(cls, data):
if data and isinstance(data, dict):
value = json.dumps(data, ensure_ascii=False)
else:
value = data
for item in range(0, value.count("${")):
if "${" in value and "}" in value:
start_index = value.index("${")
end_index = value.index("}")
old_value = value[start_index:end_index + 1]
# 获取yaml文件中的方法名
function_name = old_value[2: old_value.index('(')]
# 获取yaml文件中的参数
args_value = old_value[old_value.index('(')+1: old_value.index(')')]
# 将参数分割,变成单个个体
args_value_list = args_value.split(',')
# 通过反射方法,去获取新的值
new_value = getattr(DebugTalk(), function_name)(*args_value_list)
# 得到新的值,replace只能传字符串,需要强转
value = value.replace(old_value, str(new_value))
# 如果不先生成字符串格式,无法进行替换,替换了之后,需要把数据还原成字典格式
if data and isinstance(data, dict):
data = json.loads(value)
else:
data = value
return data
Yaml用例:
-
name: 岗位管理新增接口
description: 系统管理模块
base_url: http://8.129.162.225:8080
request:
method: POST
url: /system/post
json:
postCode: ${get_random_number(50,10000)}
postName: ${get_random_name(1,1000)}
postSort: 0
status: 0
headers: ${get_extract_data(Authorization)}
validate:
code: 200
equals: 操作成功
DebugTalk.py:
import random
from common.yaml_util import YamlUtil
class DebugTalk:
# 热加载获取随机数方法
@classmethod
def get_random_number(cls, min_num, max_num):
number = random.randint(int(min_num), int(max_num))
return number
# 热加载获取随机名称
@classmethod
def get_random_name(cls, min_num, max_num):
num = random.randint(int(in_num), int(max_num))
name = str(num) + "测试岗位"
return name
# 热加载获取token或者其他关联参数
@classmethod
def get_extract_data(cls, param):
return YamlUtil.read_extract(param=param)
九、断言封装
# 断言封装
def validate_result(self, result, expect):
if expect and isinstance(expect, dict):
for key, value in dict(expect).items():
if key == "code" and type(value) == int:
assert value == result[key]
if type(key) == str and key == "equals":
result_str = str(result)
assert value in result_str
十、其他的一些配置文件
pytest.ini(添加运行参数):
[pytest]
addopts = -vs -m 'smoke or newModel' --alluredir=reports/temps --clean-alluredir
testpaths = testcases/
python_files = "test_*.py"
python_classes = "TestApi*"
python_functions = "test_*"
markers =
smoke: All_test
newModel: some_test
conftest.py(全局夹具):
import pytest
from common.yaml_util import YamlUtil
@pytest.fixture(scope="session", autouse=True, name="fixture")
def execute_sql():
print("········夹具前置")
# 每次运行之前都清空接口关联的数据,否则每次执行都会产生相同的数据,久而久之数据会变得非常大
YamlUtil.clear_extract()
yield
print("夹具后置········")
run.py(用例执行):
import os
import pytest
if __name__ == '__main__':
# # 运行所有
# pytest.main()
# # 指定模块
# pytest.main(['-vs', 'testcases/test_api.py'])
# # 指定目录
# pytest.main(['-vs', 'testcases'])
# # 通过node id指定用例运行
# pytest.main(['-vs', 'testcases/test_api.py::TestApi::test_baidu'])
# # 失败重跑
# pytest.main(['-vs', '--reruns=2', 'testcases'])
# # 发现失败即可停止运行
# pytest.main(['-vs', '-x', 'testcases/test_api.py'])
# # 发现N个失败即可停止运行
# pytest.main(['-vs', '--maxfail=2', 'testcases/test_api.py'])
# # 生成测试报告
# pytest.main(['-vs', '--maxfail=2', '--html=report/report.html', 'testcases/test_api.py'])
# # 多线程运行
# pytest.main(['-vs', '-n 3', 'testcases/test_api.py'])
# 根据测试用例的部分字符串指定测试用例
# pytest.main(['-vs', '-k', 'test_baidu03 or test_baidu04', 'testcases/test_api.py'])
pytest.main()
os.system("allure generate reports/temps -o reports/allures --clean")
requirements.txt(项目依赖自动安装, pip install -r requirements):
pytest~=7.1.2
pytest-html
pytest-xdist
pytest-ordering
pytest-rerunfailures
allure-pytest
requests~=2.27.1
PyYAML~=6.0
jsonpath~=0.82
总结:只是一个基础框架,归根结底很多代码需要根据公司实际业务去进行优化,代码只是编程的实现方式,更重要的是思维方式,ok,以下一张图结束本节。