把所有的请求方法集中到一个地方 目的:利于后期维护和加入日志 异常处理 等
目的:框架封装好之后,功能测试不需要写代码,也能执行自动化测试
config.yml
base:base_php_url: http://47.107.116.139base_wx_url: https://api.weixin.qq.com
yaml_util.py
import osimport yamldef get_path():return os.path.abspath(os.path.dirname(__file__))
def read_config_file(one_node,two_node):with open(f'{get_path()}/config.yml',encoding="utf-8") as f:value=yaml.load(f,yaml.FullLoader)# print(value[one_node][two_node])return value[one_node][two_node]# if __name__ == '__main__':
# print (read_config_file("base","base_php_url"))
requests_util.py
import requestsfrom pytestdemo.common.yaml_util import read_config_fileclass RequestsUtil:base_url=""def __init__(self,base,base_url):self.base_url=read_config_file(base,base_url)session=requests.Session()def send_request(self,method,url,**kwargs):method=str(method).lower()url=self.base_url+urlres=RequestsUtil.session.request(method,url,**kwargs);return res
test_demo5.py
import refrom pytestdemo.common.requests_util import RequestsUtilclass Test5Api:csrf_token=""def test_get_csrf_token(self):res =RequestsUtil("base","base_php_url").send_request("get",url=f'/phpwind')# 通过正则表达式取值obj= re.search('name="csrf_token" value="(.*?)"',res.text)print(obj)Test5Api.csrf_token=obj.group(1)print(Test5Api.csrf_token)
接口关联
新建extract.yml
yaml_util.py
import osimport yamldef get_path():return os.path.abspath(os.getcwd().split("common")[0])
def read_config_file(one_node,two_node):with open(f'{get_path()}/common/config.yml',encoding="utf-8") as f:value=yaml.load(f,yaml.FullLoader)# print(value[one_node][two_node])return value[one_node][two_node]
def read_extract_file(node_name):with open(f'{get_path()}/common/extract.yml',encoding="utf-8") as f:value=yaml.load(f,yaml.FullLoader)return value[node_name]
def write_extract_file(data):with open(f'{get_path()}/common/extract.yml',encoding="utf-8",mode="a") as f:yaml.dump(data,f,allow_unicode=True)def clean_extract_file():with open(f'{get_path()}/common/extract.yml',encoding="utf-8",mode="w") as f:f.truncate()
# if __name__ == '__main__':
# # print (read_config_file("base","base_php_url"))
# print(get_path())
test_demo5.py
import refrom pytestdemo.common.requests_util import RequestsUtil
from pytestdemo.common.yaml_util import write_extract_file, read_extract_fileclass Test5Api:def test_get_csrf_token(self):res =RequestsUtil("base","base_php_url").send_request("get",url=f'/phpwind')# 通过正则表达式取值obj= re.search('name="csrf_token" value="(.*?)"',res.text)print(obj)write_extract_file({"csrf_token":obj.group(1)})def test_login(self):data = {"username":"msxy","password":"msxy","csrf_token": read_extract_file("csrf_token"),"backurl": "http://47.107.116.139/phpwind/"}print(data)res = RequestsUtil("base","base_php_url").send_request("post","/phpwind/index.php?m=u&c=login",data=data)print(res.request.headers)print(res.text)
conftest.py
@pytest.fixture(scope="session",autouse=True)
def clean_extract():clean_extract_file()
request 封装&接口关联
yaml_util.py
import osimport yamldef get_path():return os.path.abspath(os.path.dirname(__file__))
def read_config_file(one_node,two_node):with open(f'{get_path()}/config.yml',encoding="utf-8") as f:value=yaml.load(f,yaml.FullLoader)# print(value[one_node][two_node])return value[one_node][two_node]
def read_extract_file(node_name):with open(f'{get_path()}/extract.yml',encoding="utf-8") as f:value=yaml.load(f,yaml.FullLoader)return value[node_name]
def write_extract_file(data):with open(f'{get_path()}/extract.yml',encoding="utf-8",mode="a") as f:yaml.dump(data,f,allow_unicode=True)def clean_extract_file():with open(f'{get_path()}/extract.yml',encoding="utf-8",mode="w") as f:f.truncate()
# if __name__ == '__main__':
# print (read_config_file("base","base_php_url"))
requests_util.py
import jsonimport requestsfrom pytestdemo.common.yaml_util import read_config_file, read_extract_fileclass RequestsUtil:base_url=""def __init__(self,base,base_url):self.base_url=read_config_file(base,base_url)session=requests.Session()def replace_value(self,data):if data and isinstance(data,dict):str=json.dumps(data)else:str=data# 将变量替换成值for a in range(1,str.count("{{")+1):if "{{" in str and "}}" in str:print(str)start_index=str.index("{{")end_index = str.index("}}")print(start_index,end_index)old_value=str[start_index:end_index+2]new_value=read_extract_file(old_value[2:-2])str=str.replace(old_value,new_value)if data and isinstance(data,dict):data=json.loads(str)else:data=strreturn datadef send_request(self,method,url,**kwargs):method=str(method).lower()url=self.base_url+self.replace_value(url)for key,value in kwargs.items():if key in ["params","data","json"]:kwargs[key]=self.replace_value(value)res=RequestsUtil.session.request(method,url,**kwargs);return res
test_demo4.py
import randomfrom pytestdemo.common.requests_util import RequestsUtil
from pytestdemo.common.yaml_util import write_extract_fileclass Test4Api:access_token=""def test_get_token(self):data = {"grant_type": "client_credential","appid": "wx6b11b3efd1cdc290","secret": "106a9c6157c4db5f6029918738f9529d"}res =RequestsUtil("base","base_wx_url").send_request("get",url="/cgi-bin/token", params=data)return_date= res.json()print(return_date["access_token"])write_extract_file({"access_token":return_date["access_token"]})def test_get_news(self):data={"tag":{"id":23475,"name":"ms"+str(random.randint(100000,999999)),"keyss":"{{access_token}}"}}res=RequestsUtil("base","base_wx_url").send_request("post","/cgi-bin/tags/update?access_token={{access_token}}",json=data)print(res.text )