164 lines
7.4 KiB
Python
164 lines
7.4 KiB
Python
#!/usr/bin/python3
|
||
# -*- coding: UTF-8 -*-
|
||
# 设置utf-8 显示中文
|
||
"""
|
||
@Author: guozg
|
||
@File：api_base.py
|
||
"""
|
||
|
||
import json
|
||
import sys
|
||
from string import Template
|
||
|
||
import pytest
|
||
import requests
|
||
import jsonpath
|
||
|
||
sys.path.append('../')
|
||
from common.yml.get_yml_data import GetYmlData
|
||
from login.login import Login
|
||
from config.get_conf_data import GetConfData
|
||
from common.yml.get_yml_keys import GetYmlKeys
|
||
from common.cookie.get_app_cookie_header import GetAppCookieHeader
|
||
from common.cookie.get_web_cookie import GetWebCookie
|
||
from common.yml.handle_api_yml_data import HandleYmlData
|
||
from common.yml.handle_dependence_data import HandleDependenceData
|
||
from common.util.handle_jsonfile import HandleJsonFile
|
||
from common.util.get_file_path import GetFilePath
|
||
from common.yml.handle_response import HandleResponse
|
||
from common.yml.handle_checkdata import HandleCheck
|
||
|
||
class ApiBase:
|
||
def __init__(self):
|
||
self._ymldata = GetYmlData()
|
||
self.__login = Login()
|
||
self.__conf = GetConfData()
|
||
self.__unify_host = self.__conf.get_unify_host()
|
||
self.__host = self.__conf.get_host()
|
||
self.__yml_keys = GetYmlKeys()
|
||
self.__hjson = HandleJsonFile()
|
||
self.__fpath = GetFilePath()
|
||
self.__hresponse = HandleResponse()
|
||
self.__hcheck = HandleCheck()
|
||
|
||
def steps(self, params: dict = None,page_yml_dir=None):
|
||
'''
|
||
步骤驱动 \n
|
||
:param params: 外部传入的请求参数值，可以理解为 case所对应的yml 文件里的值
|
||
:param page_yml_dir: page层对应的yml文件所在的目录(不包括data)
|
||
:return: Response(request的响应结果)
|
||
'''
|
||
# 定义返回结果
|
||
res = None
|
||
# 获取第二层调用者对应的yml数据
|
||
yml_data: dict = self._ymldata.get_yml_data(page_yml_dir)
|
||
print("\n")
|
||
print("-------------处理前的yml数据如下：----------------")
|
||
print(yml_data)
|
||
handle_yml_data = HandleYmlData(yml_data, params)
|
||
# 获取处理后的yml数据(此时的yml数据已经进行了参数的替换等操作)
|
||
yml_data = handle_yml_data.get_handle_yml_data()
|
||
print('\n--------------处理后的yml 数据如下：-------------')
|
||
print(yml_data)
|
||
print("\n")
|
||
|
||
type = yml_data[self.__yml_keys.get_type_key()]
|
||
|
||
# 由于现在 接口和 ui的 base文件是分开的,所以目前不需要进行判断当前的类型.
|
||
# if str.lower(type) == "api"
|
||
|
||
# 获取api_data 的数据
|
||
api_data: dict = yml_data[self.__yml_keys.get_api_data_key()]
|
||
run_whether = api_data[self.__yml_keys.get_run_whether_key()]
|
||
|
||
"""第一步:判断是否执行."""
|
||
# 非被依赖的case或function，在执行之前，要先判断是否需要运行，当不需要运行时，直接跳过
|
||
if run_whether == False:
|
||
pytest.skip("该用例不需要执行,直接跳过")
|
||
else:
|
||
|
||
# # 获取接口的请求参数
|
||
# request_data: dict = api_data[self.__yml_keys.get_request_data_key()]
|
||
|
||
# -------------------先检测是否有cookie依赖,也就是在请求接口之前,是否要进行登录-----------------------
|
||
# 判断是否需要进行登录，所就是所谓的cookie依赖
|
||
login_whether = api_data[self.__yml_keys.get_dependence_login_key()]
|
||
logindata = api_data[self.__yml_keys.get_login_data_key()]
|
||
|
||
# app 字符串
|
||
app_str = self.__conf.get_app_str()
|
||
# 获取token_filepath
|
||
token_path = logindata[self.__yml_keys.get_token_filepath_key()]
|
||
|
||
token = None
|
||
"""第二步:判断是否有登录依赖,主要是获取token"""
|
||
# 当为true时，表示需要进行登录，其他值均表示不需要进行登录。
|
||
if login_whether == True:
|
||
# 要获取登录的类型，是app还是web。因为两者的cookie存储方式不一致。
|
||
login_type = str.lower(logindata[self.__yml_keys.get_login_type_key()])
|
||
|
||
"""
|
||
不在判断是global 还是local,直接优先取token_filepath的值.判断逻辑
|
||
1:先获取token_filepath 的值,为None时,获取login_data的值,进行登录.当不为None时,进行如下判断
|
||
2:文件的大小是否0,为0时,获取login_data的值,进行登录;不为0时进行session时长限制的判断
|
||
3:获取文件的上次修改时间,然后与当前进行对比,获取时间差(单位是分钟).
|
||
4:获取时间时间差后,与config.yml文件里配置的时差进行比较
|
||
5:当小于时,则读取json文件里的token,否则获取login_data的值进行登录.
|
||
注:当不为None时,只在进行了login_data 登录,那么在登录的时候,就会将返回的token进行保存,以供下次使用(免登录)
|
||
保存登录的token,是在login的方法里完成的.
|
||
"""
|
||
# 由于在login的方法里,当在获取token时,只要token_path 不为None时,会保存token到指定的json文件里.
|
||
# 所以不需要再判断token_path == None的情况
|
||
flag = self.__fpath.check_tokenfile_available(login_type, token_path)
|
||
if flag == True:
|
||
token = self.__hjson.read_json_file_to_dict(token_path)
|
||
# 表示文件无效
|
||
else:
|
||
login_data = logindata[self.__yml_keys.get_login_data_key()]
|
||
if login_type == app_str:
|
||
token = self.__login.get_app_token(login_data,token_path)
|
||
else:
|
||
token = self.__login.get_web_token(login_data,token_path)
|
||
|
||
dependence_case = api_data[self.__yml_keys.get_dependence_case_key()]
|
||
"""第三步:判断是否有数据依赖"""
|
||
# 有数据依赖
|
||
if dependence_case == True:
|
||
print(f"----该接口有数据依赖，依赖信息如下：-----")
|
||
yml_data = HandleDependenceData(yml_data).run_all_dependence_case()
|
||
print(f"该case进行依赖数据替换后，ymldata数据如下：\n{yml_data}")
|
||
api_data = yml_data[self.__yml_keys.get_api_data_key()]
|
||
# 获取接口的请求参数
|
||
request_data: dict = api_data[self.__yml_keys.get_request_data_key()]
|
||
|
||
"""第四步:执行接口."""
|
||
if login_whether == True:
|
||
if login_type == app_str:
|
||
res = requests.request(**request_data, headers=token)
|
||
else:
|
||
res = requests.request(**request_data, cookies=token)
|
||
else:
|
||
res = requests.request(**request_data)
|
||
|
||
# 增加一层判断，判断text是否有值
|
||
text = res.text
|
||
if text == None or text == "":
|
||
pytest.fail(f"执行该接口后，未返回任何数据，故将case置为fail")
|
||
|
||
else:
|
||
print(f"\n----该case的接口执行情况如下：----")
|
||
print(f"接口的请求响应状态码为: {res.status_code}")
|
||
print(f"接口的请求响应结果如下:\n{res.json()}")
|
||
|
||
"""第五步：检测是否保存响应结果"""
|
||
self.__hresponse.handle_response(res,yml_data)
|
||
|
||
"""第六步：数据检查,同时按配置保存校验结果"""
|
||
self.__hcheck.handle_checkdata(res,yml_data)
|
||
|
||
|
||
return res
|
||
|
||
|
||
|