接口自动化测试-L4

[1845543565_5289]

一、接口加密与解密

1、 环境准备

• 对响应加密的接口。对它发起一个get请求后，得到一个加密过后的响应信息。(如果有可用的加密过的接口以及了解它的解密方法，可以跳过)
• 准备一个加密文件
• 使用python命令在有加密文件的所在目录启动一个服务
• 访问该网站

2、 原理

• 在得到响应后对响应做解密处理：
  • 如果知道使用的是哪个通用加密算法的话，可以自行解决。
  • 如果不了解对应的加密算法的话，可以让研发提供加解密的lib。
  • 如果既不是通用加密算法、研发也无法提供加解密的lib的话，可以让加密方提供远程解析服务，这样算法仍然是保密的。

"""
调用python自带的base64，直接对返回的响应做解密，即可得到解密后的响应。
封装对于不同算法的处理方法。
"""
import requests
import base64
import json


def test_encode():
    url = "http://127.0.0.1:9999/demo.txt"
    r = requests.get(url)
    print(r.text)
    # content 获取一个二进制的响应结果，对加密内容进行解密，使用json.loads()优化格式
    res = json.loads(base64.b64decode(r.content))
    print(res)

（1）封装对于不同算法的处理方法

"""
调用python自带的base64，直接对返回的响应做解密，即可得到解密后的响应。
封装对于不同算法的处理方法。
"""
import requests
import base64
import json


class ApiRequest:
    def send(self, data:dict):
        res = requests.request(data["method"], data["url"], headers=data["headers"])
        if data["encoding"] == "base64":
            return json.loads(base64.b64decode(res.content))

        # 把加密过后的响应值发给第三方服务，让第三方去做解密然后返回解密过后的信息，private是自定义的，不是固定
        elif data["encoding"] == "private":
            return requests.post("url", data=res.content)

（2）测试封装的方法，新建py文件

from interface import test_api_request

class TestApiRequest:
    res_data = {
        "method": "get",
        "url": "http://127.0.0.1:9999/demo.txt",
        "headers": None,
        "encoding": "base64"
    }

    def test_send(self):
        # 实例化对象
        ar = test_api_request.ApiRequest()
        print(ar.send(self.res_data))

二、 多套被测环境

1、 多环境介绍

环境	使用场景	备注
dev	开发环境	开发自测
QA	测试环境	QA日常测试
preprod	预发布环境	回归测试、产品验测试
prod	线上环境	用户使用的环境

2、 多套被测环境切换的意义和价值

• 访问信息： 不同环境的域名或ip都不一样，部分产品Host也会有区别
• 配置信息： DB、Redis、ES等中间件的配置信息不同环境也不一样
• 每条用例的url都是写死的，一旦切换环境，所有的用例都要修改。

3、 实现目标

• 全局控制，一键切换
• 可维护性和扩展性强，可以应对不断演进的环境变化。
  • 环境管理
  • 环境切换

4、 环境管理

• 使用环境管理文件
  • yaml
  • ini
  • 常量类
• 使用不同的文件管理不同的环境
• 在接口用例中只指定path，不指定url

5、环境切换

• 通过环境变量进行切换
• 通过命令行参数进行切换

6、 通过环境变量进行切换

• 设置环境变量
• 读取环境变量

# mac设置环境变量，尽量不直接使用env，env可能会和底层环境变量重名，可以使用interface_env等
export interface_env=dev
# windows 设置环境变量
set interface_env=dev

（1）新建test.yaml文件

# 测试环境的配置文件
base_url: https://httpbin.org/

（2）新建dev.yaml文件

# 开发环境的配置文件
base_url: https://httpbin.ceshiren.com/

（3）代码

import os
import requests
import yaml


# 设置临时环境变量
# Mac： export interface_env="test"
# windows：set interface_env="test"
# window如果使用的是powershell，需要使用$env:interface_env="test" 命令设置环境变量
class TestMulitiEnv:
    def setup_class(self):
        # 在接口用例中只指定path，不指定url
        # 从yaml文件读取数据，env读取出来是一个字典
        # 优化点：第一种方式：从环境变量读取名称为inter_env的配置环境，default给默认值
        path_env = os.getenv("inter_env", default="test")
        env = yaml.safe_load(open(f"{path_env}.yaml", encoding="utf-8"))
        # env = yaml.safe_load(open("dev.yaml", encoding="utf-8"))
        self.base_url = env["base_url"]

    def test_devenv(self):
        """
        验证是否为开发环境
        :return:
        """
        path = "get"
        r = requests.get(self.base_url + path)
        # print(r.json())
        # 假设httpbin.ceshiren.com是开发环境，则断言当前请求是否是向"开发环境"发起的
        assert r.json()["headers"]["Host"] == "httpbin.ceshiren.com"

    def test_testenv(self):
        """
        验证是否为测试环境
        :return:
        """
        path = "get"
        r = requests.get(self.base_url + path)
        # r = requests.get("https://httpbin.org/get")
        assert r.json()["headers"]["Host"] == "httpbin.org"

（4）终端执行用例

# 设置开发环境变量
export interface_env="dev"
pytest test_muliti_env.py 

# 设置测试环境变量
export interface_env="test"
pytest test_muliti_env.py 

7、 使用命令行进行切换

（1）conftest.py文件

# 声明变量 global_env
global_env = {}


# hook 函数，是添加命令行参数使用
def pytest_addoption(parser):
    # group 将下面所有的 option都展示在这个group下。
    mygroup = parser.getgroup("hogwarts")
    # 注册一个命令行选项
    mygroup.addoption("--env",
    # 参数的默认值
    default='test',
    # 存储的变量
    dest='env',
    # 参数的描述信息
    help='设置接口自动化测试默认的环境'
     )


# 获取设置对命令行参数，并传递给一个变量global_env
def pytest_configure(config):
    default_ev = config.getoption("--env")
    tmp = {"env": default_ev}
    global_env.update(tmp)

（2）代码

import requests
import yaml
from interface.conftest import global_env


class TestMulitiEnvByOption:
    def setup_class(self):
        path_env = global_env.get("env")
        env = yaml.safe_load(open(f"{path_env}.yaml", encoding="utf-8"))
        self.base_url = env["base_url"]

    def test_devenv(self):
        """
        验证是否为开发环境
        :return:
        """
        path = "get"
        r = requests.get(self.base_url + path)
        # print(r.json())
        # 假设httpbin.ceshiren.com是开发环境，则断言当前请求是否是向"开发环境"发起的
        assert r.json()["headers"]["Host"] == "httpbin.ceshiren.com"

    def test_testenv(self):
        """
        验证是否为测试环境
        :return:
        """
        path = "get"
        r = requests.get(self.base_url + path)
        # r = requests.get("https://httpbin.org/get")
        assert r.json()["headers"]["Host"] == "httpbin.org"

（3）终端执行

pytest test_muliti_by_option_env.py --env=test
pytest test_muliti_by_option_env.py --env=dev

三、多响应类型封装设计

1、 多协议封装应用场景

• 问题：

  • 响应值不统一
    • json
    • xml
  • 断言比较困难

• 解决方案：

  • 获得的响应信息全部转换为结构化的数据进行处理

图一：未优化前

import requests
import xmltodict


def test_xml_to_dict():
    res = requests.get("https://www.nasa.gov/rss/dyn/lg_image_of_the_day.rss")
    # 注意，打印res的texts属性
    print(res.text)
    # 转换为python的dict格式
    dict_res = xmltodict.parse(res.text)
    print(dict_res)

图二：优化后的

"""
多协议封装：对响应值做二次封装，可以使用统一提取方式完成断言
"""
import requests
import xmltodict
from requests import Response


def test_response_to_dict():
    # res = requests.get("https://www.nasa.gov/rss/dyn/lg_image_of_the_day.rss")
    res = requests.get("https://httpbin.ceshiren.com/get")
    final_res = response_to_dict(res)
    # 断言响应值是否为dict类型的格式
    assert isinstance(final_res, dict)


def response_to_dict(response: Response):
    # 获取text属性
    res_text = response.text
    # 判断响应文本信息是否以 <?xml 开头
    if res_text.startswith("<?xml"):
        final_dict = xmltodict.parse(res_text)
    elif res_text.startswith("<!DOCTYPE html>"):
        final_dict = "html"
    # 如果是json 则返回 json 格式
    else:
        final_dict = response.json()
    return final_dict

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

1、问题

• 可维护性差：一个 api 发生变化，需要修改用例文件
• 可读性差：无法从代码中看出来明确的业务逻辑
• 断言能力差：响应内容只能一层一层提取

2、 架构优化设计

（1）goods.py文件

import requests
from intermediate.apis.base_api import BaseApi


# 优化1：接口里面直接使用了 requests
# 解决方案：在base_api 中添加公共的 send 方法，通过继承来调用
# 优化2：大量重复的base_url
# 在构造函数，实例化base_url， 并且在send方法， 就直接做拼接，就无需重复编写self.base_url
# 优化3：token的获取也放在 BaseApi 的构造函数中
class Goods(BaseApi):

    def create(self, goods_data):
        create_url = "admin/goods/create"
        # r = requests.post(create_url, json=goods_data, headers={"X-Litemall-Admin-Token": self.token})
        r = self.send("post", create_url, json=goods_data)
        return r

    def list(self, goods_name, order="desc", sort="add_time"):
        goods_list_url = "admin/goods/list"
        goods_list_data = {
            "name": goods_name,
            "order": order,
            "sort": sort
        }
        # r = requests.get(goods_list_url, params=goods_list_data, headers={"X-Litemall-Admin-Token": self.token})
        r = self.send("get", goods_list_url, params=goods_list_data)
        return r

    def detail(self, goods_id):
        goods_detail_url = "admin/goods/detail"
        # r = requests.get(goods_detail_url, params={"id": goods_id}, headers={"X-Litemall-Admin-Token": self.token})
        r = self.send("get", goods_detail_url, params={"id": goods_id})
        return r

    def detele(self, goods_id):
        delete_url = "admin/goods/delete"
        data = {"id": goods_id}
        r = self.send("post", delete_url, json=data)
        return r

（2）cart.py文件

from intermediate.apis.base_api import BaseApi


class Cart(BaseApi):
    def add(self, goods_id, product_id):
        cart_url = "wx/cart/add"
        cart_data = {"goodsId": goods_id, "number": 1, "productId": product_id}
        r = self.send("post", cart_url, json=cart_data)
        return r

（3）base_api.py文件

import json
import requests
from intermediate.utils.log_utils import logger


class BaseApi:
    # 如果有多角色，就做判断，如果没有就不做，即默认传None
    def __init__(self, base_url, role=None):
        self.base_url = base_url
        # 获取对应的角色信息
        if role:
            self.role = role

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

    def __set_token(self, request_infos):
        """
        优化点：大部分接口都需要设置token
        解决方案：
        1、在发起接口请求之前，就获取token信息
        2、获取token信息之后，塞入到请求信息之中
            除了 method 和 url 之外，所有到其他信息，包括 header、params 等其他的参数，都会塞入至 kwargs 不定长参数中
            cart_api（发起请求） —> 调用send方法 —> （send方法控制kwargs）获取 kwargs —> 在 kwargs 中塞入 header（鉴权）信息

        优化点：有两个token如何解决
            不同的角色有不同的token信息
        :return:
        """
        # ======管理端token获取
        admin_url = "admin/auth/login"
        admin_data = {"username": "hogwarts", "password": "test12345", "code": ""}
        admin_r = requests.request("post", self.base_url + admin_url, json=admin_data)
        # 获取B端 token，定义为实例变量
        # B端和C端token的key值不一样，可以将 self.token_b 变为键值对，加入相应的key
        self.token = {"X-Litemall-Admin-Token": admin_r.json()["data"]["token"]}

        # ======用户端token获取
        client_url = "wx/auth/login"
        client_data = {"username": "user123", "password": "user123"}
        client_r = requests.request("post", self.base_url + client_url, json=client_data)
        # 获取C端 token，定义为实例变量
        self.client_token = {"X-Litemall-Admin-Token": client_r.json()["data"]["token"]}

        # 如果是 admin，那么就塞入 admin 的 token，如果是其他，那么就塞入其他的 token
        # 在 test_cart 测试用例文件中，实例化时，需要传一个角色的值，确认是否是admin角色还是其他角色
        if self.role == "admin":
            self.final_token = self.token
        else:
            self.final_token = self.client_token
        # 获取headers，如果请求本身有头信息，那么就把token信息更新（添加）进去
        # if request_infos.get("headers"):
        #     request_infos["headers"]["X-Litemall-Admin-Token"] = self.token_b
        # else:
        #     request_infos["headers"] = {"X-Litemall-Admin-Token": self.token_b}
        if request_infos.get("headers"):
            request_infos["headers"].update(self.final_token)
        else:
            request_infos["headers"] = self.final_token

        return request_infos

    def send(self, method, url, **kwargs):
        self.__set_token(kwargs)
        r = requests.request(method, self.base_url + url, **kwargs)
        logger.debug(f"{url}接口的响应值为：{json.dumps(r.json(), indent=2, ensure_ascii=False)}")
        return r.json()

（4）test_cart.py文件

import pytest
from intermediate.apis.admin.goods import Goods
from intermediate.apis.wx.cart import Cart


@pytest.mark.parametrize("goods_name", ["0529goods01", "0529goods02"])
class TestCart:
    def setup_class(self):
        self.goods = Goods("https://litemall.hogwarts.ceshiren.com/", "admin")
        self.cart = Cart("https://litemall.hogwarts.ceshiren.com/", "client")

    def test_add_cart(self, goods_name):
        """
            添加购物车的步骤：
            1、上架商品
            2、获取商品列表
            3、获取商品详情
            4、添加购物车
        """
        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": []}

        self.goods.create(goods_data)
        goods_list_r = self.goods.list(goods_name)
        self.goods_id = goods_list_r["data"]["list"][0]["id"]
        goods_detail_r = self.goods.detail(self.goods_id)
        product_id = goods_detail_r["data"]["products"][0]["id"]
        res = self.cart.add(self.goods_id, product_id)
        # 将delete放在用例中执行：1、调用方便；2、测试类中的每个方法并非都会添加goods数据；
        # 如果部分用例不添加，则相当于这个步骤，不能在每个测试用例执行完成之后执行，只能在单个用例执行完之后执行
        self.goods.detele(self.goods_id)
        # 断言
        assert res["errmsg"] == "成功"

（5）log_utils.py文件

# 日志配置
import logging
# 创建logger实例
logger = logging.getLogger('simple_example')
# 设置日志级别
logger.setLevel(logging.DEBUG)
# 流处理器
ch = logging.StreamHandler()
ch.setLevel(logging.DEBUG)
# 日志打印格式
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
# 添加格式配置
ch.setFormatter(formatter)
# 添加日志配置
logger.addHandler(ch)

3、 添加领域模型

（1）新增domain文件夹

（2）新建goods_domain.py文件

class GoodsDomain:
    """
    抽象的概念，不做具体实现
    代表是某一种业务模型
    """
    def detele_by_name(self, name):
        # 如果子类 Goods 没有重写父类方法，是调不通的
        # 调用子类方法
        goods_list_r = self.list(name)
        goods_id = goods_list_r["data"]["list"][0]["id"]
        self.detele(goods_id)

（3）goods.py文件变更

（4）test_cart.py文件变更

from intermediate.apis.wx.cart import Cart


# 不使用 self.goods.detele(self.goods_id)方法
# 使用 self.goods.detele_by_name(goods_name) 方法调用delete接口
@pytest.mark.parametrize("goods_name", ["0529goods01", "0529goods02"])
class TestCart:
    def setup_class(self):
        self.goods = Goods("https://litemall.hogwarts.ceshiren.com/", "admin")
        self.cart = Cart("https://litemall.hogwarts.ceshiren.com/", "client")

    def test_add_cart(self, goods_name):
        """
            添加购物车的步骤：
            1、上架商品
            2、获取商品列表
            3、获取商品详情
            4、添加购物车
        """
        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": []}

        self.goods.create(goods_data)
        goods_list_r = self.goods.list(goods_name)
        self.goods_id = goods_list_r["data"]["list"][0]["id"]
        goods_detail_r = self.goods.detail(self.goods_id)
        product_id = goods_detail_r["data"]["products"][0]["id"]
        res = self.cart.add(self.goods_id, product_id)
        # 将delete放在用例中执行：1、调用方便；2、测试类中的每个方法并非都会添加goods数据；
        # 如果部分用例不添加，则相当于这个步骤，不能在每个测试用例执行完成之后执行，只能在单个用例执行完之后执行
        # self.goods.detele(self.goods_id)
        self.goods.detele_by_name(goods_name)
        # 断言
        assert res["errmsg"] == "成功"