目录

接口自动化框架目录

基础类封装概览

基础类封装方法详解

Log.py文件

Request.py文件

Python是目前主流的接口自动化脚本语言之一。

本专栏将联系业务介绍使用python接口自动化脚本范例。

适合对象：已有一定python基础，对自动化感兴趣的小伙伴。

接口自动化框架目录

目录说明：

在Project目录下新建你的项目，里面再次拆分出本项目的相关配置公共模块、接口层、用例层。

公共模块功能：存放配置、封装公共调用方法

接口层功能：封装调用接口

用例层功能：调用封装的接口，传入不同参数校验、场景模式校验，设置断言

基础类封装概览

将公共的模块封装为基础类，比如发送请求的request、log日志封装等等。

这些模块是每个业务线或者接口都需要使用到的，故将其放在Common下面进行编写封装。

基础类封装方法详解

Log.py文件

目的：用例执行过程中打印日志，定位问题。

小tips：

导入模块，如果有标红显示，说明你的lib下面没有这个第三方的资源了，可以在Terminal下面使用命令直接下载pip install colorama

下面为封装Log日志的方法：

# -*- coding: utf-8 -*-"""封装log方法"""import loggingimport osimport timeimport statimport datetimefrom hashlib import md5from colorama import Fore, StyleTRACE_ID_LIST = []LEVELS = {'debug': logging.DEBUG,'info': logging.INFO,'warning': logging.WARNING,'error': logging.ERROR,'critical': logging.CRITICAL}logger = logging.getLogger()level = 'default'def set_file_permission(filename):# 设置日志目录文件及权限path = filename[0:filename.rfind('/')]if not os.path.isdir(path):os.makedirs(path)if not os.path.isfile(filename):return Nonetry:# 赋予文件777的权限,防止不同用户访问时报错os.chmod(filename, stat.S_IRWXU | stat.S_IRWXG | stat.S_IRWXO)return Trueexcept Exception as e:print('设置日志文件权限错误，error：{}'.format(e))return Falsedef set_handler(levels):# 日志文件内写入内容if levels == 'error':set_file_permission(MyLog.err_file)logger.addHandler(MyLog.err_handler)set_file_permission(MyLog.log_file)logger.addHandler(MyLog.handler)def remove_handler(levels):if levels == 'error':logger.removeHandler(MyLog.err_handler)logger.removeHandler(MyLog.handler)def get_current_time():return datetime.datetime.now().strftime(MyLog.date)class MyLog:backup_count = 5path = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))local_time = time.strftime('%Y-%m-%d', time.localtime(time.time()))log_file = path + '/Log/info.log'err_file = path + '/Log/error.log'logger.setLevel(LEVELS.get(level, logging.NOTSET))date = '%Y-%m-%d %H:%M:%S.%f'from logging.handlers import TimedRotatingFileHandler#分割日志记录handler = TimedRotatingFileHandler(filename=log_file, when="MIDNIGHT", interval=1, backupCount=backup_count,encoding='utf-8')err_handler = TimedRotatingFileHandler(filename=err_file, when="MIDNIGHT", interval=1, backupCount=backup_count,encoding='utf-8')@staticmethoddef debug(log_meg):set_handler('debug')logger.debug("[DEBUG " + get_current_time() + "]" + str(log_meg))print(Fore.LIGHTGREEN_EX + "[DEBUG]" + str(log_meg) + Style.RESET_ALL)remove_handler('debug')@staticmethoddef info(log_meg):set_handler('info')logger.info("[INFO " + get_current_time() + "]" + str(log_meg))print("[INFO]" + str(log_meg))remove_handler('info')@staticmethoddef warning(log_meg):set_handler('warning')logger.warning("[WARNING " + get_current_time() + "]" + str(log_meg))print(Fore.LIGHTYELLOW_EX + "[WARNING]" + str(log_meg) + Style.RESET_ALL)remove_handler('warning')@staticmethoddef error(log_meg):try:raise RuntimeErrorexcept RuntimeError:import sysf = sys.exc_info()[2].tb_frame.f_backdatas = 'The error source: ' + f.f_code.co_filename + ' ' + f.f_code.co_name + ' ' + str(f.f_lineno)set_handler('error')logger.error("[ERROR " + get_current_time() + "]" + str(log_meg), exc_info=True)logger.error(datas)print(Fore.LIGHTRED_EX + "[ERROR]" + str(log_meg) + Style.RESET_ALL)print(datas)remove_handler('error')@staticmethoddef critical(log_meg):set_handler('critical')logger.critical("[CRITICAL " + get_current_time() + "]" + str(log_meg))print(Fore.RED + "[CRITICAL]" + str(log_meg) + Style.RESET_ALL)remove_handler('critical')

Request.py文件

# -*- coding: utf-8 -*-"""封装request"""import requestsfrom retry import retryfrom requests.adapters import HTTPAdapterfrom Common.Log import MyLogimport urllib3import jsonclass Request:def __init__(self):self.log = MyLog@retry((Exception, requests.RequestException), tries=3, delay=1, backoff=2, max_delay=4)def base_request(self, url, data=None, header=None, method=None, time_out=None, body_json=None, params=None):"""客户端请求基础方法:param url: 完整的请求url，包括url和path:param data: 请求的数据:param header: 请求头:param method 请求方式，支持POST和GET方式请求:param time_out 请求超时时间:param body_json : 以json进行传参:param params: GET请求的时直接传参:return:"""header = {} if header is None else headermethod = 'POST' if method is None else methodtime_out = 60 if time_out is None else time_out# 定义user_agentuser_agent = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/74.0.3729.169 Safari/537.36'user_agent_header = {'User-Agent': user_agent}if header.get('User-Agent') is None:header.update(user_agent_header)if header.get('Content-Type') is None and type(data) is str:header.update({'Content-Type': 'application/json;charset=utf-8'})# 将请求body字典转化为json格式，并去除其中的空格if (type(data) is dict or type(data) is list) and 'application/json' in str(header.get('Content-Type')):data = json.dumps(data, separators=(',', ':'))self.log.info('请求的url: %s' % str(url))if data:self.log.info('请求的data: %s' % str(data))if params:self.log.info(f'请求的params: {params}')if body_json:self.log.info(f'请求的body_json: {body_json}')self.log.info('请求的header: %s' % str(header))self.log.info('请求的method: %s' % str(method))# 关闭警告urllib3.disable_warnings()# 增加失败重试session = requests.Session()session.mount('http://', HTTPAdapter(max_retries=1))session.mount('https://', HTTPAdapter(max_retries=1))try:res = session.request(method=method, url=url, params=params, data=data, json=body_json, headers=header, timeout=time_out, verify=False, proxies=None, stream=False)except requests.RequestException as e:self.log.info('%s%s' % ('RequestException url: ', url))self.log.info(e)raise RuntimeError(e)except Exception as e:self.log.info('%s%s' % ('Exception url: ', url))self.log.info(e)raise RuntimeError(e)# time_consuming为响应时间，单位为毫秒time_consuming = res.elapsed.microseconds / 1000# time_total为响应时间，单位为秒time_total = res.elapsed.total_seconds()response_dicts = dict()response_dicts['code'] = res.status_codetry:response_dicts['body'] = res.json()except Exception as e:self.log.info('解析响应json异常：{}'.format(e))response_dicts['body'] = ''self.log.info('响应text:{}'.format(res.text))response_dicts['time_consuming'] = time_consumingresponse_dicts['time_total'] = time_totalresponse_dicts['text'] = res.textresponse_dicts['headers'] = res.headersresponse_dicts['cookies'] = dict(res.cookies)self.log.info('响应码：%s' % response_dicts['code'])self.log.info('响应体：%s' % response_dicts['body'])self.log.info('请求总耗时：%s' % response_dicts['time_total'])return response_dicts

如果写的不错，来个三连行～～～