实现类似httprunner的接口框架续

前言

之前写过一个简单的httprunner的实现：30行左右代码实现一个类似httprunner的接口框架
使用Python的string.Template()来替换$变量，使用Python表达式来处理变量提取和响应断言。功能上只实现了核心的接口的顺序请求及变量的提取和断言。
这里对其功能进行扩充以下功能：

增加配置，baseurl，请求默认配置，用户自定义变量
步骤增加，skip跳过控制，times循环控制
步骤中支持直接$变量名引用环境变量，及响应文本response_text，响应头，response_headers, 状态码status_code，响应时间response_time等。
使用Session会话维持，根据request字典是否包含data/json/files，设置默认请求方法

Yaml数据文件格式

config：配置
- baseurl：接口域名端口配置
- request：请求默认配置，如默认headers, timeout等
- variables：用户自定义变量
tests：测试步骤
- name: 步骤名称
- skip: 是否跳过
- times: 循环次数
- request: 请求数据，对应requests.request()方法的参数
- extact: 提取变量，Python表达式，使用的eval()计算，存储到上下文context字典变量中
- verify: 断言，Python表达式，使用eval()计算

示例数据data.yaml如下：

config:
  name: '测试用例'
  request:
    timeout: 30
    headers:
      x-test: abc123
  variables: 
    client_id: kPoFYw85FXsnojsy5bB9hu6x
    client_secret: &client_secret l7SuGBkDQHkjiTPU3m6NaNddD6SCvDMC

tests:
  - name: 步骤1-获取百度token接口 # 接口名称
    request:  # 请求报文
      url: https://aip.baidubce.com/oauth/2.0/token
      params:
        grant_type: client_credentials
        client_id: $client_id
        client_secret: *client_secret  # 使用锚点
    extract:  # 提取变量, 字典格式
      token:  response.json()['access_token']  # RESPONSE系统变量，代表响应对象
    verify:
      - status_code == 200
  - name: 步骤2-百度ORC接口  # 第二个接口
    request:
      url: https://aip.baidubce.com/rest/2.0/ocr/v1/general_basic?access_token=${token}  # 使用变量
      data:  # 请求体（表单格式）
        url: https://upload-images.jianshu.io/upload_images/7575721-40c847532432e852.png?imageMogr2/auto-orient/strip%7CimageView2/2/w/1240
    verify:  # 断言, 列表格式
      - response.json()['words_result_num'] == 6
  - name: 步骤3-跳过
    skip: True
  - name: 步骤4-重复执行
    times: 3
    request:
      url: https://httpbin.org/get

注：由于yaml语法中自带锚点功能，如配置variables中client_secret即设置了锚点，引用也非常方便。因此variables的也可以不使用模板替换，直接使用yaml的锚点引用。

实现步骤

关键字定义

不同的设计者对字段喜欢用不同的关键字，如Robot Framework中最外层使用settings/testcases/variables/keywords，httprunner中config/test，Jenkins Pipelines中使用options/stages等。
这里对使用的关键字进行了定义，读者也可以改为自己使用的关键字。

CONFIG = 'config'  # 配置关键字  settings
STEPS = 'tests'  # 步骤关键字  steps/teststeps/testcases

NAME = 'name'  # 名称
VAIABLES = 'variables'  # 用户自定义变量关键字
BASEURL = 'baseurl'
REQUEST = 'request'  # 请求配置,请求数据关键字
CHECK = 'verify'  # 验证关键字  check/validate/assert
EXTRACT = 'extract'   # 提取关键字 output/register
SKIP = 'skip'  # 跳过步骤关键字
TIMES = 'times'  # 循环步骤关键字  circle

配置解析

首先我们使用requests.session()建立一个会话，会话可以保持登录等请求状态，并可以对其设置默认请求参数。

session = requests.session()
config = data.get(CONFIG)
if config:
    name = config.get(NAME)
    variables = config.get(VAIABLES, {})
    baseurl = config.get(BASEURL)
    request = config.get(REQUEST)
    if request:
        for key, value in request.items():
            session.__setattr__(key, value)

如果存在request配置，则将字典格式的配置信息，添加为会话对象session的属性。

上下文变量

上下文变量是保存用户自定义变量，环境变量，用户提取的变量，和响应的一些变量的。之前直接使用的locals()即当前局部变量。这里新建一个专用的变量context。由于包含多个部分的内容，这里可以使用Python的ChainMap，导入方式为from collections import ChainMap，也可以直接使用字典格式，使用update更新值。

context = ChainMap(variables, os.environ)

vaiables是用户自定义变量，os.environ是环境变量，ChaInMap类似一种联合字典，逐个字典查找需要的键值，更新时变量更新到第一个字典中。

步骤解析

步骤对应data.yaml中的tests段，格式是一个列表。
使用循环，遍历执行每一个步骤，如果步骤中设置了skip则跳过。执行是times次数循环执行步骤。

context['steps'] = []  # 用于保存所有步骤的请求和响应，便于跨步骤引用
steps = data.get(STEPS)
for step in steps:
    step_name = step.get(NAME)
    skip = step.get(SKIP)
    times = step.get(TIMES, 1)
    request = step.get(REQUEST)
    if skip or not request:
        print(' 跳过步骤:', step_name)
        continue

    for i in range(times):
        print(' 执行步骤:', step_name, f'第{i+1}轮' if step.get(TIMES) else '')

打印步骤时，如果包含times字段则在步骤后输出第几轮。

请求变量解析

不同于httprunner的随处可用 $引用变量或函数，这里限定只允许在请求数据request中使用`$ 变量。处理方式先将字典格式的request，转为yaml字符串。这里不使用默认的yaml流格式，因为yaml流格式将字典转为{a: 1, b:2}而不是a: 1\nb: 2`，大括号对模板变量替换有一些影响。

request_str = yaml.dump(request, default_flow_style=False)  # 先转为字符串
if '$' in request_str:
    request_str = Template(request_str).safe_substitute(context)  # 替换${变量}为varables中的同名变量
request = yaml.safe_load(request_str)  # 重新转为字典

设置默认请求方法

由于requests.request()方法中method参数是必选参数，因此每个请求段都必选有method字段，但是笔者却总是忘记写。
这了为了可以不写method，对其添加默认值，如果请求字段中有data/json/files字段，则默认使用post，否则默认使用get。

if request.get('data') or request.get('json') or request.get('files'):
   request.setdefault('method', 'post')
else:
   request.setdefault('method', 'get')

组装baseurl

if baseurl:
    url = request.get('url')
    if not url.startswith('http'):
        request['url'] = base_url + url

通常情况下baseurl默认不带/，而url以/开头。为避免少些或多写/，也可以使用request['url'] = '/'.join((baseurl.rstrip('/'), url.lstrip('/')))

发送请求

发送请求时直接将请求数据request字典，解包放入session.request()方法中即可，注意request字典中不能有该方法不支持的参数。

print('  请求url:', request.get('url'))  # print(' 发送请求:', request)
response = session.request(**request)  # 字典解包，发送接口
print('  状态码:', response.status_code)  # print(' 响应数据:', response.text)
# 注册上下文变量
step_result = dict(
    request=request,
    response=response,
    status_code=response.status_code,
    response_text=response.text,
    response_headers=response.headers,
    response_time=response.elapsed.seconds
)
context['steps'].append(step_result)  # 保存步骤结果
context.update(step_result)  # 将最近的响应结果更新到上下文变量中

这里将响应的一些数据组成字典添加到上下文变量中，这里添加了两次。
第一次是将本步骤的结果添加到上下文的steps，这里保存了每一步的结果。
第二次是将本次请求的request,response等变量直接添加到上下文中，即最近一次请求的请求和响应结果，方便提取或断言中可以直接使用这些变量。

变量提取及处理断言

# 提取变量
extract = step.get(EXTRACT)
if extract is not None:  # 如果存在extract
    for key, value in extract.items():
        print("  提取变量:", key, value)
        # 计算value表达式，可使用的全局变量为空，可使用的局部变量为上下文context中的变量
        context[key] = eval(value, {}, context)  # 保存变量结果到局部变量中
# 处理断言
check = step.get(CHECK)
if check and isinstance(check, list):
    for line in check:
        result = eval(line, {}, context)  # 计算断言表达式，True代表成功，False代表失败
        print("  处理断言:", line, "结果:", "PASS" if result else "FAIL")

extact段为一个字典，key为要保存的变量名，value是一个Python表达式字符串，这里使用eval()执行Python表达式，将返回的值注册到上下文context变量中。
eval()由于可以直接将字符串按Python语句执行，是存在安全隐患的，在使用eval()时，应尽量限定其使用的全局变量和局部变量。这里限定eval解析时只运行使用context上下文中的变量。

完整代码

import os
from string import Template
from collections import ChainMap

import yaml
import requests

# 步骤定义
CONFIG = 'config'  # 配置关键字  settings
STEPS = 'tests'  # 步骤关键字  steps/teststeps/testcases

NAME = 'name'  # 名称
VAIABLES = 'variables'  # 用户自定义变量关键字
BASEURL = 'baseurl'
REQUEST = 'request'  # 请求配置,请求数据关键字
CHECK = 'verify'  # 验证关键字  check/validate/assert
EXTRACT = 'extract'   # 提取关键字 output/register
SKIP = 'skip'  # 跳过步骤关键字
TIMES = 'times'  # 循环步骤关键字  circle


def run(data):
    # 解析配置
    session = requests.session()
    config = data.get(CONFIG)
    if config:
        name = config.get(NAME)
        variables = config.get(VAIABLES, {})
        baseurl = config.get(BASEURL)
        request = config.get(REQUEST)
        if request:
            for key, value in request.items():
                session.__setattr__(key, value)
        print('执行用例:', name)

    # 上下文变量
    context = ChainMap(variables, os.environ)
    # 解析步骤
    context['steps'] = []  # 用于保存所有步骤的请求和响应, 便于跨步骤引用
    steps = data.get(STEPS) 
    for step in steps:
        step_name = step.get(NAME)
        skip = step.get(SKIP)
        times = step.get(TIMES, 1)
        request = step.get(REQUEST)
        if skip or not request:
            print(' 跳过步骤:', step_name)
            continue

        for i in range(times):
            print(' 执行步骤:', step_name, f'第{i+1}轮' if step.get(TIMES) else '')
            # 请求$变量解析
            if not request:
                continue
            request_str = yaml.dump(request, default_flow_style=False)  # 先转为字符串
            if '$' in request_str:
                request_str = Template(request_str).safe_substitute(context)  # 替换${变量}为varables中的同名变量
                request = yaml.safe_load(request_str)  # 重新转为字典
            # 设置默认请求方法
            if request.get('data') or request.get('json') or request.get('files'):
                request.setdefault('method', 'post')
            else:
                request.setdefault('method', 'get')
            # 组装baseurl
            if baseurl:
                url = request.get('url')
                if not url.startswith('http'):
                    request['url'] = '/'.join((baseurl.rstrip('/'), url.lstrip('/')))

            # 发送请求
            print('  请求url:', request.get('url'))  # print(' 发送请求:', request)
            response = session.request(**request)  # 字典解包，发送接口
            print('  状态码:', response.status_code)  # print(' 响应数据:', response.text)

            # 注册上下文变量
            step_result = dict(
                request=request,
                response=response,
                status_code=response.status_code,
                response_text=response.text,
                response_headers=response.headers,
                response_time=response.elapsed.seconds
            )
            context['steps'].append(step_result)  # 保存步骤结果
            context.update(step_result)  # 将最近的响应结果更新到上下文变量中

            # 提取变量
            extract = step.get(EXTRACT)
            if extract is not None:  # 如果存在extract
                for key, value in extract.items():
                    print("  提取变量:", key, value)
                    # 计算value表达式，可使用的全局变量为空，可使用的局部变量为RESPONSE(响应对象)
                    context[key] = eval(value, {}, context)  # 保存变量结果到上下文中
            # 处理断言
            check = step.get(CHECK)
            if check and isinstance(check, list):
                for line in check:
                    result = eval(line, {}, context)  # 计算断言表达式，True代表成功，False代表失败
                    print("  处理断言:", line, "结果:", "PASS" if result else "FAIL")
    return context['steps']


if __name__ == "__main__":
    with open('data.yml', encoding='utf-8') as f:
        data = yaml.safe_load(f)
    run(data)

执行结果：

执行用例: 测试用例
 执行步骤: 步骤1-获取百度token接口 
  请求url: https://aip.baidubce.com/oauth/2.0/token
  状态码: 200
  提取变量: token response.json()['access_token']
  处理断言: status_code == 200 结果: PASS
 执行步骤: 步骤2-百度ORC接口 
  请求url: https://aip.baidubce.com/rest/2.0/ocr/v1/general_basic?access_token=24.07107d04b4252ecead3c2be01cf52613.2592000.1587199134.282335-11767296
  状态码: 200
  处理断言: response.json()['words_result_num'] == 6 结果: PASS
 跳过步骤: 步骤3-跳过
 执行步骤: 步骤4-重复执行 第1轮
  请求url: https://httpbin.org/get
  状态码: 200
 执行步骤: 步骤4-重复执行 第2轮
  请求url: https://httpbin.org/get
  状态码: 200
 执行步骤: 步骤4-重复执行 第3轮
  请求url: https://httpbin.org/get
  状态码: 200

最后编辑于：2020.03.19 16:53:18

人面猴
序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
沈念sama阅读 203,324评论 5赞 476
死咒
序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
沈念sama阅读 85,303评论 2赞 381
救了他两次的神仙让他今天三更去死
文/潘晓璐我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
开封第一讲书人阅读 150,192评论 0赞 337
道士缉凶录：失踪的卖姜人
文/不坏的土叔我叫张陵，是天一观的道长。经常有香客问我，道长，这世上最难降的妖魔是什么？我笑而不...
开封第一讲书人阅读 54,555评论 1赞 273
港岛之恋（遗憾婚礼）
正文为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
茶点故事阅读 63,569评论 5赞 365
恶毒庶女顶嫁案：这布局不是一般人想出来的
文/花漫我一把揭开白布。她就那样静静地躺着，像睡着了一般。火红的嫁衣衬着肌肤如雪。梳的纹丝不乱的头发上，一...
开封第一讲书人阅读 48,566评论 1赞 281
城市分裂传说
那天，我揣着相机与录音，去河边找鬼。笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。我是一名探鬼主播，决...
沈念sama阅读 37,927评论 3赞 395
双鸳鸯连环套：你想象不到人心有多黑
文/苍兰香墨我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
开封第一讲书人阅读 36,583评论 0赞 257
万荣杀人案实录
序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
沈念sama阅读 40,827评论 1赞 297
护林员之死
正文独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋以下内容为张勋视角年9月15日...
茶点故事阅读 35,590评论 2赞 320
白月光启示录
正文我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
茶点故事阅读 37,669评论 1赞 329
活死人
序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
沈念sama阅读 33,365评论 4赞 318
日本核电站爆炸内幕
正文年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
茶点故事阅读 38,941评论 3赞 307
男人毒药：我在死后第九天来索命
文/蒙蒙一、第九天我趴在偏房一处隐蔽的房顶上张望。院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
开封第一讲书人阅读 29,928评论 0赞 19
一桩弑父案，背后竟有这般阴谋
文/苍兰香墨我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。一阵脚步声响...
开封第一讲书人阅读 31,159评论 1赞 259
情欲美人皮
我被黑心中介骗来泰国打工，没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。一个月前我还...
沈念sama阅读 42,880评论 2赞 349
代替公主和亲
正文我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
茶点故事阅读 42,399评论 2赞 342