接口自动化测试-L3

一、 整体结构响应断言

1、 针对与“大响应数据”如何断言

  • 针对主要且少量的业务字段断言。
  • 其他字段不做数据正确性断言,只做类型与整体结构的校验。
  • 与前面的版本进行 diff,对比差异化的地方。

2、 JSONSchema 简介

  • 使用 JSON 格式编写的
  • 可以用来定义校验 JSON 数据的结构
  • 可以用来校验 JSON 数据的一致性
  • 可以用来校验 API 接口请求和响应

3、 JSONSchema 整体结构响应断言

  • 预先生成对应结构的 Schema。
  • 将实际获取到的响应与生成的 Schema 进行对比。

4、 JSONSchema 的生成

  • 通过界面工具生成。
  • 通过第三方库生成。
  • 通过命令行工具生成。

5、 JSONSchema 的生成效果

// # 预期的 JSON 文档结构
{
  "name": "Hogwarts",
  "Courses": ["Mock", "Docker"]
}

生成的jsonschema

// jsonschema
// type表示类型
// properties代表属性,有name,Courses,且强调的是属性的类型,items表示array里面元素的类型
// required代表必填,name,Courses均为必填的
{
  "$schema": "http://json-schema.org/draft-06/schema#",
  "$ref": "#/definitions/Welcome",
  "definitions": {
    "Welcome": {
      "type": "object",
      "additionalProperties": false,
      "properties": {
        "name": {
          "type": "string"
        },
        "Courses": {
          "type": "array",
          "items": {
            "type": "string"
          }
        }
      },
      "required": ["Courses", "name"],
      "title": "Welcome"
    }
  }
}

6、 界面工具生成

  • 复制 JSON 数据
  • 粘贴到在线生成工具中
  • 自动生成 JSON Schema 数据
  • JSON Schema 在线生成工具:https://app.quicktype.io

7、 第三方库生成

  • 安装:pip install genson
  • 调用方法生成对应的 JSONSchema 数据结构。
from genson import SchemaBuilder

def generate_jsonschema(obj):
    # 实例化 SchemaBuilder 类
    builder = SchemaBuilder()
    # 调用 add_object 方法,将要转换成jsonschema的数据传入进去,python对象obj,格式为字典
    builder.add_object(obj)
    # 转换成 schema 数据
    return builder.to_schema()

# 验证是否成功生成
def test_generate_jsonschema():
    print(generate_jsonschema({"name": 1}))

8、 JSONSchema 验证

  • 安装:pip install jsonschema
  • 调用 validate() 进行验证。
import json
from genson import SchemaBuilder
from jsonschema.validators import validate


def generate_jsonschema(obj):
    # 实例化 SchemaBuilder 类
    builder = SchemaBuilder()
    # 调用 add_object 方法,将要转换成jsonschema的数据传入进去,python对象obj,格式为字典
    builder.add_object(obj)
    # 转换成 schema 数据
    return builder.to_schema()


# 将 schema 数据存入文件中
def generate_jsonschema_file(obj, file_path):
    # 生成 jsonschema 的结果数据赋给 json_schema_data
    json_schema_data = generate_jsonschema(obj)
    with open(file_path, "w") as f:
        # json_schema_data 需要传的对象,f 文件形式
        json.dump(json_schema_data, f)


# 将 schema 数据存入文件中,文件为validate.json
def test_generate_jsonschema_file():
    generate_jsonschema_file({"name": 1}, "validate.json")


"""
对比 python 对象与生成的 JSONSchame 的结构是否一致
"""


# data_obj 被验证的数据,即响应的源数据,schema=schema指定schema结构,即验证的格式
def validate_schema(data_obj, schema):
    # 不直接抛出错误,直接返回是否成功的标志 T 或 F
    try:
        validate(data_obj, schema=schema)
        return True
    except Exception as e:
        # 打印异常,也可以使用日志形式
        # log()
        print(f"结构体验证失败,失败原因是{e}")
        return False


def test_validate_schema():
    # 打开文件validate.json
    with open("validate.json") as f:
        json_schema_data = json.load(f)
    # 直接断言是否成功,举例:输入错误的类型 "a",name的值的类型为整型
    assert validate_schema({"name": "a"}, json_schema_data)

9、 JSONSchema 二次封装

  • 生成JSONSchema
  • 验证JSONSchema

封装为工具类,创建文件 jsonschema_utils.py

import json
from genson import SchemaBuilder
from jsonschema.validators import validate


# 通常调用类,通过实例化类 json_utils = JSONSchemaUtils(),然后 json_utils.方法 来调用
# 这里可以通过定义为类方法,实现 JSONSchemaUtils.方法 来直接调用,方法里需要添加 cls 参数代表类本身
class JSONSchemaUtils:
    @classmethod
    def validate_schema_by_file(cls, data_obj, schema_file):
        # 打开文件
        with open(schema_file) as f:
            # 读取文件
            schema_data = json.load(f)
        # 验证 python 对象 data_obj 与生成的 JSONSchame 的 schema_data 结构是否一致
        # 返回结果
        return cls.validate_schema(data_obj, schema_data)

    @classmethod
    def validate_schema(cls, data_obj, schema):
        """
        通过 schema 验证数据
        """
        # 不直接抛出错误,直接返回是否成功的标志 T 或 F
        try:
            validate(data_obj, schema=schema)
            return True
        except Exception as e:
            # 打印异常,可以使用日志形式
            # log()
            print(f"结构体验证失败,失败原因是{e}")
            return False

    @classmethod
    def generate_jsonschema_by_file(cls, obj, file_path):
        json_schema_data = cls.generate_jsonschema(obj)
        with open(file_path, "w") as f:
            # json_schema_data 需要传的对象,f 文件形式
            json.dump(json_schema_data, f)

    @classmethod
    def generate_jsonschema(cls, obj):
        """
        生成对应的 jsonschema 数据
        """
        # 实例化 SchemaBuilder 类
        builder = SchemaBuilder()
        # 调用 add_object 方法,将要转换成jsonschema的数据传入进去,python对象obj,格式为字典
        builder.add_object(obj)
        # 转换成 schema 数据
        return builder.to_schema()

调用工具类,创建文件 test_res_jsonschema.py

import requests
from interface.test_jsonschema.jsonschema_utils import JSONSchemaUtils


def test_httpbin_generate_schema():
    """
    针对 httpbin 的响应信息,生成对应的 schema 结构文件
    """
    # 发起请求,获取响应信息
    r = requests.get("https://httpbin.ceshiren.com/get")
    # r.json() 提取json格式的响应内容,生成 schema 响应结构存入文件 httpbin.json
    JSONSchemaUtils.generate_jsonschema_by_file(r.json(), "httpbin.json")


def test_httpbin_req():
    # 获取响应信息
    r = requests.get("https://httpbin.ceshiren.com/get")
    # 把响应信息 r.json() 传入到验证方法里,打开前面生成的 schema 响应结构文件 httpbin.json
    validate_res = JSONSchemaUtils.validate_schema_by_file(r.json(), "httpbin.json")
    # 断言,验证结构是否一致
    assert validate_res == True

二、数据库操作与断言

1、 接口测试响应验证

  • 通过接口响应值
  • 通过查询数据库信息辅助验证

2、 接口测试数据清理

  • 通过 Delete 接口删除
  • 自动化测试使用干净的测试环境,每次自动化测试执行完成之前或之后做数据还原。

3、 数据库操作注意事项

  • 表结构复杂,随便删除数据会影响测试,甚至会导致系统出现异常

4、 接口自动化测试常用的数据库操作

  • 连接与配置
  • 查询数据与断言

5、 数据库封装

  • 封装数据库配置
  • 封装 sql 查询操作
  • 调用方法执行 sql 语句
import pymysql


# 封装建立连接的对象
def get_conn():
    conn = pymysql.connect(
        host="litemall.hogwarts.ceshiren.com",
        port=13306,
        user="test",
        password="test123456",
        database="litemall",
        charset="utf8mb4"
    )
    return conn


# 执行sql语句
def execute_sql(sql):
    connect = get_conn()
    cursor = connect.cursor()
    cursor.execute(sql)  # 执行SQL
    record = cursor.fetchone()  # 查询记录
    return record


if __name__ == '__main__':
    # 执行sql语句查询user123这个用户的购物车有一个名称为 hogwarts1 的商品
    execute_sql("select * from litemall_cart where "
                "user_id=1 and deleted=0 and "
                "goods_name='hogwarts1'")

6、 查询数据与数据库断言

# 查询查询user123这个用户的购物车有一个名称为 hogwarts1 的商品
sql_res = execute_sql("select * from litemall_cart where "
            "user_id=1 and deleted=0 and "
            "goods_name='hogwarts1'")

assert sql_res

三、接口鉴权的多种情况与解决方案

1、 接口鉴权是什么

  • 身份认证



2、 接口鉴权通用的解决方案

  • 认证信息的获取
  • 认证信息的携带

image

3、 后端接口鉴权常用方法

image

4、 cookie 鉴权

  • cookie 的获取(根据接口文档获取)
  • 发送携带 cookie 的请求
    • 直接通过 cookies 参数
    • 通过 Session() 对象
import requests


class TestCookieVerify:
    def setup_class(self):
        self.proxy = {
            "http": "http://127.0.0.1:8888",
            "https": "http://127.0.0.1:8888"
        }

    # 简单场景,直接写入cookie
    def test_cookies(self):
        """
        在有确定的cookies信息的情况下,可以直接使用cookies参数
        :return:
        """
        r = requests.get("https://httpbin.ceshiren.com/cookies", cookies={"hello": "ok"}, proxies=self.proxy, verify=False)
        print(r.json())

    def test_cookies_with_out_session(self):
        """
        1、获取cookie
        2、set cookie 设置cookie
        3、再次获取cookie,查看是否设置成功
        :return:
        """
        #
        r1 = requests.get("https://httpbin.ceshiren.com/cookies", headers={"hello": "1"}, proxies=self.proxy, verify=False)
        print(f"第1次的响应值为{r1.json()}")
        r2 = requests.get("https://httpbin.ceshiren.com/cookies/set/username/123456", headers={"hello": "2"}, proxies=self.proxy, verify=False)
        print(f"第2次的响应值为{r2.json()}")
        r3 = requests.get("https://httpbin.ceshiren.com/cookies", headers={"hello": "3"}, proxies=self.proxy, verify=False)
        print(f"第3次的响应值为{r3.json()}")

    # 使用session对象
    def test_cookies_session(self):
        """
        # 获取session 的实例,需要通过Session()保持会话,
        # 即为认证之后,之后所有的实例都会携带cookie
        # 可以模仿用户在浏览器的操作
        :return:
        """

        # 实例化session
        req = requests.session()
        r1 = req.get("https://httpbin.ceshiren.com/cookies/set/username/123456", headers={"hello": "1"}, proxies=self.proxy, verify=False)
        print(f"第1次的响应值为{r1.json()}")
        r2 = req.get("https://httpbin.ceshiren.com/cookies", headers={"hello": "2"}, proxies=self.proxy, verify=False)
        print(f"第2次的响应值为{r2.json()}")
        r3 = req.get("https://httpbin.ceshiren.com/cookies", headers={"hello": "3"}, proxies=self.proxy, verify=False)
        print(f"第3次的响应值为{r3.json()}")

5、 token 鉴权

  • token 的获取(根据接口文档获取)
  • 发送携带 token 的请求(根据接口文档获取)
import requests


class TestTokenVerify:
    def setup_class(self):
        self.proxy = {
            "http": "http://127.0.0.1:8888",
            "https": "http://127.0.0.1:8888"
        }

    def test_token(self):
        user_data = {"username": "admin123", "password": "admin123", "code": ""}
        r = requests.post("http://litemall.hogwarts.ceshiren.com/admin/auth/login", json=user_data, proxies=self.proxy, verify=False)
        # print(r.json())
        # 获取token值
        token = r.json()["data"]["token"]
        print(token)
        # 这个接口的token是放在请求头里的
        r2 = requests.get("https://litemall.hogwarts.ceshiren.com/admin/profile/nnotice", headers={"X-Litemall-Admin-Token":  token})
        print(r2.json())

6、 auth 鉴权

  • 在基本 HTTP 身份验证中,请求包含格式为 的标头字段Authorization: Basic
  • 其中credentials是 ID 和密码的Base64编码,由单个冒号连接:。

import requests
from requests.auth import HTTPBasicAuth


class TestAuthVerify:
    def setup_class(self):
        self.proxy = {
            "http": "http://127.0.0.1:8888",
            "https": "http://127.0.0.1:8888"
        }

    def test_basic_auth(self):
        # 表示访问一个需要BasicAuth认证的路径
        # username=用户名,password=密码
        # 如果不使用basic auth 则会失败
        r = requests.get("https://httpbin.ceshiren.com/basic-auth/user/abc",
                         proxies=self.proxy, verify=False, auth=HTTPBasicAuth("user", "abc"))

四、电子商城接口自动化测试实战

1、接口测试流程

image

2、 商城业务场

  • 商品上架
  • 商品查询
  • 加入购物车

image

3、 接口测试用例设计思路

4、 添加购物车流程脚本编写

  • 上架商品
  • 查询商品列表,获取商品ID
  • 查询商品详情,获取商品库存ID
  • 加入购物车
思路

image

import requests


class TestLitemall:
    def setup_class(self):
        # B端token获取
        login_b_url = "https://litemall.hogwarts.ceshiren.com/admin/auth/login"
        login_b_data = {"username": "hogwarts", "password": "test12345", "code": ""}
        login_b_r = requests.post(login_b_url, json=login_b_data)
        # 获取B端 token,定义为实例变量
        # self.token 的声明要在用例之前完成,使用 setup_class 提前完成变量的声明
        self.token_b = login_b_r.json()["data"]["token"]
        # C端token获取
        login_c_url = "https://litemall.hogwarts.ceshiren.com/wx/auth/login"
        login_c_data = {"username": "user123", "password": "user123"}
        login_c_r = requests.post(login_c_url, json=login_c_data)
        # 获取C端 token,定义为实例变量
        self.token_c = login_c_r.json()["data"]["token"]

    def teardown_class(self):
        pass

    # 上架商品接口
    def test_add_goods(self):
        create_url = "https://litemall.hogwarts.ceshiren.com/admin/goods/create"
        goods_data = {"goods": {"picUrl": "", "gallery": [], "isHot": False, "isNew": True, "isOnSale": True,
                                 "goodsSn": "23052901", "name": "0529goods01"},
                       "specifications": [{"specification": "规格", "value": "标准", "picUrl": ""}],
                       "products": [{"id": 0, "specifications": ["标准"], "price": "59", "number": "59", "url": ""}],
                       "attributes": []}
        create_r = requests.post(create_url, json=goods_data, headers={"X-Litemall-Admin-Token": self.token_b})
        # 打印响应体内容
        print(create_r.json())

    def test_add_cart(self):
        cart_url = "https://litemall.hogwarts.ceshiren.com/wx/cart/add"
        # 待优化:goodsId和productId是写死的
        cart_data = {"goodsId": 1434787, "number": 1, "productId": 254034}
        cart_r = requests.post(cart_url, json=cart_data, headers={"X-Litemall-Token": self.token_c})
        # 打印响应体内容
        print(cart_r)

脚本优化

  • 参数化
    • 使用pytest parametrize装饰器实现商品名称的参数化
  • 添加日志
    • 新建日志配置
    • 在用例中使用配置好的日志实例
  • 数据清理
    • 在用例执行完成之后调用删除接口完成数据清理
  • 报告展示
    • 安装allure相关依赖
# 生成报告信息
pytest -vs test_litemall_l3.py --alluredir=./reports/
# 生成报告在线服务,查看报告
allure serve ./reports/ 

完整代码

import json
import allure
import pytest
import requests
from interface.utils.log_utils import logger


@allure.feature("电子商城接口测试")
class TestLitemall:
    def setup_class(self):
        # ========= 管理端token获取
        login_b_url = "https://litemall.hogwarts.ceshiren.com/admin/auth/login"
        login_b_data = {"username": "hogwarts", "password": "test12345", "code": ""}
        login_b_r = requests.post(login_b_url, json=login_b_data)
        # 获取B端 token,定义为实例变量
        # self.token 的声明要在用例之前完成,使用 setup_class 提前完成变量的声明
        self.token_b = login_b_r.json()["data"]["token"]

        # ========= 用户端token获取
        login_c_url = "https://litemall.hogwarts.ceshiren.com/wx/auth/login"
        login_c_data = {"username": "user123", "password": "user123"}
        login_c_r = requests.post(login_c_url, json=login_c_data)
        # 获取C端 token,定义为实例变量
        self.token_c = login_c_r.json()["data"]["token"]

    def teardown(self):
        delete_url = "https://litemall.hogwarts.ceshiren.com/admin/goods/delete"
        data = {"id": self.goods_id}
        delete_r = requests.post(delete_url, json=data, headers={"X-Litemall-Admin-Token": self.token_b})
        # 添加日志,indent 缩进,编码格式 ensure_ascii,默认为True
        logger.info(f"删除商品接口的响应信息为{json.dumps(delete_r.json(), indent=2, ensure_ascii=False)}")

    @allure.story("电子商城上架商品并加入购物车")
    # 参数化 goods_name
    @pytest.mark.parametrize("goods_name", ["0529goods01", "0529goods02"])
    def test_add_goods(self, goods_name):
        # ========= 上架商品接口
        # goods_name = "0529goods"
        create_url = "https://litemall.hogwarts.ceshiren.com/admin/goods/create"
        goods_data = {"goods": {"picUrl": "", "gallery": [], "isHot": False, "isNew": True, "isOnSale": True,
                                 "goodsSn": "23052901", "name": goods_name},
                       "specifications": [{"specification": "规格", "value": "标准", "picUrl": ""}],
                       "products": [{"id": 0, "specifications": ["标准"], "price": "59", "number": "59", "url": ""}],
                       "attributes": []}
        create_r = requests.post(create_url, json=goods_data, headers={"X-Litemall-Admin-Token": self.token_b})
        # 添加日志
        # logger.debug(f"上架商品接口的响应信息为{create_r.json()}")
        # 日志优化,indent 缩进,编码格式 ensure_ascii,默认为True
        logger.info(f"上架商品接口的响应信息为{json.dumps(create_r.json(), indent=2 ,ensure_ascii=False)}")

        # ========= 获取商品列表接口
        goods_list_url = "https://litemall.hogwarts.ceshiren.com/admin/goods/list"
        goods_list_data = {
            "name": goods_name,
            "order": "desc",
            "sort": "add_time"
        }
        # get请求,使用params传参
        goods_list_r = requests.get(goods_list_url, params=goods_list_data, headers={"X-Litemall-Admin-Token": self.token_b})
        # 提取 self.goods_id
        self.goods_id = goods_list_r.json()["data"]["list"][0]["id"]
        # 添加日志,indent 缩进,编码格式 ensure_ascii,默认为True
        logger.info(f"获取商品列表接口的响应信息为{json.dumps(goods_list_r.json(), indent=2, ensure_ascii=False)}")

        # ========= 获取商品详情接口
        goods_detail_url = "https://litemall.hogwarts.ceshiren.com/admin/goods/detail"
        goods_detail_r = requests.get(goods_detail_url, params={"id": self.goods_id}, headers={"X-Litemall-Admin-Token": self.token_b})
        # 提取 product_id
        product_id = goods_detail_r.json()["data"]["products"][0]["id"]
        # 添加日志,indent 缩进,编码格式 ensure_ascii,默认为True
        logger.info(f"获取商品详情接口的响应信息为{json.dumps(goods_detail_r.json(), indent=2, ensure_ascii=False)}")

        # ========= C端添加购物车
        cart_url = "https://litemall.hogwarts.ceshiren.com/wx/cart/add"
        cart_data = {"goodsId": self.goods_id, "number": 1, "productId": product_id}
        cart_r = requests.post(cart_url, json=cart_data, headers={"X-Litemall-Token": self.token_c})
        # 打印响应体内容
        res = cart_r.json()
        print(res)
        # 添加日志,indent 缩进,编码格式 ensure_ascii,默认为True
        logger.info(f"添加购物车接口的响应信息为{json.dumps(cart_r.json(), indent=2, ensure_ascii=False)}")
        assert cart_r.json()["errmsg"] == "成功"