"福利中心"接口自动化实现：requests+pytest框架

为了提高测试和回归效率，我们老大提议推进服务端核心功能的自动化测试。我们用户组涉及的服务端逻辑较多，经常需要抓包看接口请求，观察接口响应数据，确认redis的生成，数据的落库等等，所以做业务逻辑接口自动化是我们项目组的不二选择。确认好大方向后，于是乎：安排！

首先选择一个相对独立稳定，日常测试步骤较复杂的项目“安卓福利中心”。然后对福利中心逻辑进行梳理，经查阅多篇wiki文档，多次“打扰”开发确认逻辑点后，终于整理成档。
选择适合的项目框架。我选择了pytest，理由是看到过一个python鄙视链：pytest 鄙视 > unittest 鄙视 > robotframework 鄙视 > 记流水账鄙视 > "hello world"小白。当然要走在潮流的前沿了。

一、pytest 框架介绍

pytest 是 python 语言中一款强大的单元测试框架，与 python 自带的 unittest 测试框架类似，但是比 unittest 框架使用起来更简洁，更高效。根据 pytest 的官方网站介绍，它具有如下特点：

简单灵活，容易上手，文档丰富
支持参数化，支持运行由nose、unittest编写的测试用例
测试用例的skip和xfail处理，可以跳过指定用例，或对某些预期失败的case标记成失败
可以很好的和jenkins集成，实现持续集成
能够支持简单的单元测试和复杂的功能测试，还可以用来做selenium/appnium等自动化测试、接口自动化测试（pytest+requests）
pytest具有丰富的第三方插件，并且可以自定义扩展，例如：pytest-rerunfailures（失败case重复执行）、pytest-xdist（多CPU分发）等等

1、pytest的安装与基本使用，以及丰富的内容请查看官方文档：https://docs.pytest.org/en/latest/reference/，后文仅介绍常用知识点

2、pytest用例运行规则（更多pytest命令行参数，可使用 pytest -h 或者 pytest -help 查看）

当我们使用 pytest 框架编写用例时，一定要严格按照命名规范去编写用例，这样框架才能准确地区分哪些是用例需要执行，哪些不是用例不需要执行。

用例设计原则：

文件名以 test_*.py 和 *_test.py 的文件
以 test_开头的函数
以 Test 开头的类，并且不能带有 init 方法
以 test_开头的方法
所有的包 package 必项要有__init__.py 文件
断言使用 assert

3、fixture介绍

fixture 是 pytest 的核心功能，也是亮点功能。fixture 的目的是提供一个固定基线，在该基线上测试可以可靠地重复地执行。fixture 提供了区别于传统单元测试（setup/teardown）有显著改进：

测试fixture有明确的名称，通过在函数/模块/类或者整个项目中激活来使用
测试fixture是模块化的实现，使用fixture名即可触发特定的fixture，每个fixture都可以互相调用
fixture的范围从简单的单元扩展到复杂的功能测试，允许根据配置和组件选项对fixture和测试用例进行参数化，或者跨函数 function、类class、模块module或整个测试会话session范围进行测试

@pytest.fixture(scope="function", params=None, autouse=False, ids=None, name=None):
    """
    使用装饰器标记fixture的功能
    :arg scope: scope 有四个级别参数 "function" (默认), "class(类)", "module(模块)" or "session(会话)".
    :arg params: 一个可选的参数列表，它将导致多个参数调用fixture功能和所有测试使用它
    :arg autouse: 如果为True，则为所有测试激活fixture； 如果为False（默认值）则显式需要参考来激活fixture
    :arg ids: 每个字符串id的列表，每个字符串对应于params 这样他们就是测试ID的一部分，如果没有提供ID它们将从params自动生成
    :arg name: fixture的名称。 这默认为装饰函数的名称
    """

fixture参数传入实现setup 与 yield实现teardown

实现场景：用例1操作1，用例2需要先登录，用例3不需要登录，用例4需要先登录

#coding:utf-8
import pytest

@pytest.fixture()             # 不带参数时默认scope="function"
def login():
    print("先登录")            # 用例前置操作，相当于setup
    yield                     # fixture里用yield来唤醒teardown的执行
    print("操作完成后退出登录")  # 用例后置操作

@pytest.fixture()
def unlogin():
    print("不登录")


def test_one():                # 什么也不传入
    print("用例1：操作1")

def test_two(login):           # 传入login
    print("用例2：登录后，操作2")

def test_three(unlogin):       # 传入unlogin
    print("用例3：不登录，操作3")

def test_four(login):          # 传入login
    print("用例4：登录后，操作4")

运行结果：

fixture之conftest.py

1. 上面案例是在同一个.py文件中，多个用例调用一个登陆功能，如果有多个.py的文件都需要调用这个登陆功能的话，那就不能把登陆写到用例里面去了。
此时应该要有一个配置文件，单独管理一些预置的操作场景，pytest里面默认读取conftest.py里面的配置。

conftest.py配置需要注意以下几点：

conftest.py配置脚本名称是固定的，不能更改名称
conftest.py与运行的用例要在同一个pakage下，并且有__init__.py文件
不需要import导入conftest.py，pytest用例会自动查找

2. fixture能实现多个.py 跨文件共享前置。例如：单独运行以下test_fix1.py和test_fix2.py都能调用到 login() 方法

# __init__.py
# conftest.py 模块
    # coding:utf-8
    import pytest

    @pytest.fixture()
    def login():
        print("先登录")


#test_fix1.py 模块

    # coding:utf-8
    import pytest
    
    def test_one():   
        print("用例1：不登录，操作1")
    
    def test_two(login):     # 传入login
        print("用例2：登录后，操作2")


# test_fix2.py 模块

    # coding:utf-8
    import pytest
    
    def test_three(login):   # 传入login
        print("用例3：登录后，操作3")

4、assert 断言

断言是写自动化测试基本最重要的一步，一个用例没有断言，就失去了自动化测试的意义。简单来说就是将实际结果与期望结果对比，符合预期就测试 pass，不符合预期就测试 failed。

pytest 里面断言实际上就是 python 里面的 assert 断言方法，常用的有以下几种：

assert xx 判断 xx 为真
assert not xx 判断 xx 不为真
assert a in b 判断 b 包含 a
assert a not in b 判断 b 不包含 a
assert a == b 判断 a 等于 b
assert a != b 判断 a 不等于

5、配置文件 pytest.ini

pytest 配置文件可以改变 pytest 的运行方式，它是一个固定的文件（ pytest.ini ），读取配置信息，按指定的方式去运行。pytest.ini 一般放在项目的根目录下。

常用配置项

1. addopts：addopts 参数可以更改默认命令行选顷，当我们在 cmd 输入指令去执行用例的时候非常有用。

例如：生成allure测试报告时，指令比较长 > pytest -p no:ngswarni --alluredir ./report/allure_raw --clean-alluredir ，每次执行都需要输入，却又总是记不住，于是可以将指令添加到 pytest.ini 里，然后在cmd窗口直接输入 pytest，它就能默认带上这些参数了

2. mark 标记（上图已有标记案例）

pytest 支持自定义标记，自定义标记可以把一个项目划分多个模块，然后指定模块名称执行，即在运行测试用例的时候可以有选择地运行某些用例。例如：app 自动化的时候，如果想 android 和 ios 公用一套代码时，就可以使用标记功能，标记哪些是 android 用例，哪些是 ios 用例，运行代码时指定 mark 名称运行就可以了。

标记规则：

使用 @pytest.mark.markname 进行标记，markname 无限制
可标记范围有：测试函数、测试类、测试方法
同一个测试函数、测试类、测试方法可同时拥有多个不同的标记

二、requests: 让 HTTP 服务人类

Requests是用python语言编写，基于urllib，采用 Apache2 Licensed 开源协议的 HTTP 库。requests比urllib更加方便，可以节约大量的工作，完全满足对 HTTP 的测试需求。requests的使用可参考文档：https://cn.python-requests.org/zh_CN/latest/

三、requests+pytest 接口自动化项目案例

在实际工作中，请求接口，查看接口响应数据，校验数据库数据的正确生成与落库是我们常有的操作。将相对稳定的项目编写成接口自动化脚本，能有效提高日常测试效率，或回归效率。例如“福利中心”接口自动化完成后，在福利中心转go需求时起到了不错的效果，在日常修改相关功能点后一次性回归时也节约了大量时间。

在此将分享对七猫免费小说app“福利中心”功能编写接口自动化脚本案例。本文仅介绍一个用例的编写，一个用例调通后，其它用例只需要同理走相应的自逻辑即可。

1. 环境准备：python 3.6（选择相对稳定的版本）+ pytest 框架 + requests库 + allure报告

# 更多第三方库，可根据项目情况灵活安装

# 安装pytest
>pip install pytest --index-url https://pypi.douban.com/simple   #指定国内豆瓣源安装更快速
# 安装allure-pytest，生成测试报告的插件
>pip install allure-pytest --index-url https://pypi.douban.com/simple

2. 项目结构介绍

3. 使用APP时，福利中心的功能都需要先登录。所以先编写安卓登录接口请求，使用requests库知识点（此登录接口不便暴露，不展示具体代码）

4. 写好一些公共方法，例如各类型数据库的连接方式和使用方法等，在此仅举例redis的连接，其增删改查使用object对象自带的封装方法

5. 在cases文件夹下新建一个 conftest.py ，使用pytest的fixture功能装饰函数，写好登录保持会话和不需要登录的前置，方便case包下各模块需要时直接调用

6. 开始编写用例

实现功能（每日阅读福利）：登录状态下，阅读达到指定时长，进入福利中心，可领取相应金币的阅读奖励，vip身份奖励翻倍。然后再观看小视频，能领取看视频的金币奖励。

# 在 case 包下新建test_welfare.py 模块，编写用例内容如下：

# coding: utf-8
import pytest
import urllib3
urllib3.disable_warnings()  # 忽略警告
from common.connect_redis import RedisConnect                                           # 导入redis的连接
from common.com_func import getUid, dataCleanRead,isVip, assertVodieTaskid, assertCoin  # 导入一些公共方法
from api.welfare import *                                                               # 导入福利中心各接口方法
import time
import allure                                                                           # 导入allure库, 可生成allure报告



@allure.story('Android每日阅读福利 - 是vip')  # allue模块功能，@allure.story 每个用例的用户故事，便于查看报告内容
def test_getDailyReadingRewardVip(login_android_rj):
    """
    每日阅读福利 - taskId: 42, 43, 44, 45, 46, 47 ; 是vip时，阅读奖励翻倍
    isVip=1, {10min +70*2, 30min +80*2, 60min +90*2, 90min +100*2, 120min +120*2, 180min +150*2}
    """
    s = login_android_rj                   # 使用被fixture装饰过的函数，让用例保持登录会话
    uid = getUid()                         # getUid()是根据登录时的手机号，获取账号uid。默认使用自己的手机号，这里不再传参
    con12 = RedisConnect(db=12).connect()  # 连接redis（机房任务中心服务项目-db12）
    dataCleanRead()                        # 阅读数据前置处理,且到达阅读时长上限
    isVip()                                # 修改账号为vip身份
    getTaskList(s)                         # 请求福利中心任务列表接口

    a = con12.smembers('m:u:o:t:l:%s:%s' % (uid, time.strftime("%Y%m%d")))    #  查询已完成的任务key
    list = ['42', '43', '44', '45', '46', '47']
    for i in list:            # 循环遍历任务id的 list
        assert i in a         # 断言已完成的key中生成任务taskid

    dict = {'42': 140, '43': 160, '44': 180, '45': 200, '46': 240, '47': 300}
    for key, values in dict.items():
        r = readTaskGetReward(s, key)                       # 请求阅读完领取金币奖励接口
        assert int(r['data']['reward_cash']) == values      # 断言接口返回的值与预期的值相等
        assertCoin(int(r['data']['reward_cash']))           # 金币落库校验

        r = readTaskGetReward(s, key)                       # 再次请求阅读完领取金币奖励接口，确保接口有防止重复领金币情况
        assert r['errors']['title'] == "此奖励已领取，不可重复领取"

    b = con12.smembers('m:u:g:t:l:%s:%s' % (uid, time.strftime("%Y%m%d")))    # 查询已领取的任务key
    for i in list:
        assert i in b                                 # 领取金币后，断言已领取的key中有生成任务tsakid
        r = readTaskVideo(s, i)                       # 领取金币后观看视频获取金币，看完视频后，请求获取奖励接口
        assert r == {"data": {"result": "1"}}         # 断言接口返回与期望值相等
        r = readTaskVideo(s, i)                       # 重复请求获取奖励接口，确保接口有防止重复领金币情况
        assert r['errors']['title'] == "此奖励已领取，不可重复领取"

    getTaskList(s)               # 请求福利中心任务列表接口
    assertVodieTaskid()          # 校验已完成的key中生成看视频任务taskid

用例中使用到的接口和方法，按规范放置到规定位置，方便统一管理与查找，例如以上用例中使用到的各接口和方法代码如下：

# api 包下 welfare.py 模块，存放的福利中心所需接口请求，方便用例中使用时直接调用

import requests
import urllib3
urllib3.disable_warnings()  # 忽略警告


def getTaskList(s):
    """
    福利中心任务列表接口
    :param s: s = login_android_fixture
    :return: 请求接口后的json返回
    """
    url = "https://xiaoshuo.wtzw.com/api/v2/task/get-task-list?open_push=1&apiVersion=20190309143259-1.9"
    r = s.get(url, verify=False)    # 设置verify=False, 让 Requests 忽略对 SSL证书的验证
    return r.json()


def readTaskGetReward(s, x):
    """
    每日阅读任务，阅读完领取金币奖励接口
    :param s: s = login_android_fixture
    :param x: 任务id
    :return: 请求对应任务id领取奖励后json返回
    """
    url = "https://xiaoshuo.wtzw.com/api/v1/read-task/get-reward?task_id=%s&apiVersion=20190309143259-1.9" % x
    r = s.get(url, verify=False)
    return r.json()


def readTaskVideo(s, y):
    """
    每日阅读任务后观看视频，获取金币奖励接口
    :param s: s = login_android_fixture
    :param y: 任务id
    :return: 请求对应任务id领取奖励后json返回
    """
    url = "https://xiaoshuo.wtzw.com/api/v1/read-task/video?task_id=%s&apiVersion=20190309143259-1.9" % y
    r = s.get(url, verify=False)
    return r.json()

# common包下 com_func.py 模块，存放一些公共方法，方便各模块需要使用时直接导入后调用

import urllib3
urllib3.disable_warnings()  # 忽略警告
from common.connect_redis import RedisConnect
from common.connect_mysql import MysqlConnect
from common.connect_mongdb import MongdbConnect
from conf.config import mysql_info, drds_qimao_all_test, user_phone
import pymongo
import datetime
import time


def getUid(phone=user_phone):   
    """
    根据手机号查询用户uid
    :param phone: 手机号，默认phone=user_phone, user_phone是写在配置文件中的常用手机号 
    :return: 手机号对应的用户 uid
    """
    db = MysqlConnect(drds_qimao_all_test, database="qimao_user")                    # 连接数据库
    sql = "SELECT * FROM `qimao_user`.`user_phone` WHERE `phone` = '%s'; " % phone   # 编写需要执行的sql
    res = db.select(sql)                                                             # 调用MysqlConnect类中的select方法执行查询sql
    return res[0]['uid']                                                             # 从查询结果中取出uid并返回


def dataCleanRead(phone=user_phone):
    """
    每日阅读福利数据前置操作
    :param phone: 用户手机号，便于根据手机号确认 uid
    :return:空
    """
    con11 = RedisConnect(db=11).connect()  # 连接redis（时长服务-db11）
    con12 = RedisConnect(db=12).connect()  # 连接redis（机房任务中心服务项目-db12）
    con11.delete('h:d:s:t:d:d:%s:%s' % (getUid(phone), time.strftime("%Y%m%d"))) # 删除今日阅读key
    con12.delete('m:u:o:t:l:%s:%s' % (getUid(phone), time.strftime("%Y%m%d")))   # 删除已完成任务key
    con12.delete('m:u:g:t:l:%s:%s' % (getUid(phone), time.strftime("%Y%m%d")))   # 删除已领取任务key
    con11.hset('h:d:s:t:d:d:%s:%s' % (getUid(phone), time.strftime("%Y%m%d")), 'tod_rd', 10800)   # 新增今日阅读key，给字段 tod_rd 赋值为 10800（180分钟，完成所有阅读任务的下限）
    return


def isVip(phone=user_phone):
    """
    修改用户数据为会员 时间戳：7955078400（2222-02-02 00:00:00）
    :param phone: 用户手机号，便于根据手机号确认 uid
    :return:空
    """
    con8 = RedisConnect(db=8).connect()                 # 连接redis（用户中心-db8）
    old_info = con8.get('s:u:b:i:%s' % getUid(phone))   # 查询key s:u:b:i:uid
    if 'vip' in old_info:                               # 如果查询结果中包含vip字段，则不需要处理
        pass
    else:                                               # 如果查询结果中不包含vip，则在末尾添加vip字段并赋值
        new_info = old_info[:(len(old_info)-1)]+',"vip":7955078400}'
        con8.set('s:u:b:i:%s' % getUid(phone), str(new_info))
    return


def assertVodieTaskid(phone=user_phone):
    """
    每日阅读任务，校验观看小视频后，m:u:o:t:l:%s:%s 中有生成看视频任务taskid
    :param phone: 用户手机号，便于根据手机号确认 uid
    :return:
    """
    con12 = RedisConnect(db=12).connect()                                              # 连接redis（机房任务中心服务项目-db12）
    list = ['read_task_video_42', 'read_task_video_43', 'read_task_video_44', 'read_task_video_45', 'read_task_video_46', 'read_task_video_47']  # 将观看小视频的taskid写入list
    c = con12.smembers('m:u:o:t:l:%s:%s' % (getUid(phone), time.strftime("%Y%m%d")))   #  查询已完成的任务key
    for i in list:
        assert i in c                                                                  # 循环遍历小视频的taskid，断言已完成的key中生成看小视频任务taskid


def assertCoin(expectcoin, phone=user_phone):
    """
    金币校验
    :param expectcoin:  期望的金币数
    :param phone: 用户手机号，便于根据手机号确认 uid
    :return:
    """
    con10 = RedisConnect(db=10).connect()                            # 连接redis（金币服务-db10）
    ava_c = con10.hget('h:c:s:u:c:d:%s' % getUid(phone), 'ava_c')    # 获取用户redis目前的可利用金币

    res = MongdbConnect('free_book_coin_detail_data').opencol('coin_detail_more_%s' % (getUid(phone) % 500))   # 连接用户的mongdb详细金币集合
    cousor = res.find({"uid": getUid(phone)}).sort("_id", pymongo.DESCENDING).limit(1)                         # 根据uid获取最新的一条数据
    for i in cousor:
        coin = i['coin']                 # 获取新数据实际的金币额
        ava_coin = i['current_coin']     # 获取用户mongdb目前的可用金币
        assert coin == expectcoin        # 校验写入数据库的金币额等于期望的金币额
        assert ava_coin == int(ava_c)    # 校验新增金币详情后，mongdb和redis 的可用金币额相等
    return

一个用例调通之后，就可以根据功能逐步编写更多的用例，当用例越来越多时，可根据pytest功能灵活使用，例如将用例放至一个测试类中，让用例看上去更简介，执行更高效。

编写完“福利中心”整个项目功能测试点后，生成的allure报告展示：