aamt工具：介绍–专注于接口自动化

正式版源码：

aamt简介

aamt是Api AutoMation Test的首字母缩写，是一款基于pytest测试框架的测试工具，集成了各种实用的第三方包和优秀的自动化测试设计思想，帮你快速实现自动化项目落地。

快速入门

安装

安装最新版本

pip install aamt

指定版本安装

pip install aamt==0.2.5

升级aamt

pip install -U aamt
# 大写U

验证安装成功：

aamt -V
# 大写V

新建项目脚手架

aamt startproject demo

带上-venv参数，可创建单个项目的Python虚拟环境，并在该项目的虚拟环境中安装tep：

aamt startproject demo -venv

外网速度慢，pandas可能安装失败，推荐用国内镜像

pip --default-timeout=6000 install -i https://pypi.tuna.tsinghua.edu.cn/simple aamt

目录结构说明

api ：各模块接口文件

data：用例yaml数据

case：测试用例；

fixtures：Pytest fixture，自动导入；

common：个人封装 比如邮件、mysql、project项目各种路径、token信息读取等

report：测试报告；

conftest.py：Pytest挂载；

file：待上传的文件目录；

utils：工具包；

pytest.ini：Pytest配置；

main.py：执行入口；

requirements.txt：项目依赖；

项目结构说明.txt ；

用例集

在case目录下将测试用例按功能模块分成多个用例集：

case
  offer_doctor
	test_offer_doctor.py
  offer_nurse
	test_offer_nurse.py
  brand
	test_brand.py

测试用例

必须遵循用例解耦原则，每条用例都是单独可运行的。用例由3个文件组成，一个存放不同功能模块的api接口，一个存放test用例，一个文件存放纯粹的yaml数据 ：

api
  offer_doctor
  	offer_need_api.py
case
  offer_doctor
  	test_offer_doctor.py
data
  offer_doctor
    offer_doctor.yaml

  1. 单独把api摘出来，是为了方便不能的功能模块使用这个api接口，直接导入使用即可

  2. 用例只需要安装流程组装不同的业务逻辑

  3. yaml文件里是参数化数据

一条测试用例由鉴权信息、多个测试步骤和多个断言组成：

@allure.title("采购商部分入库后（入库数量3）,供应商只能同步到的入库数据为3")
def test(xuefeng1_offer_doctor1_need, xuefeng2_offer_nurse1_receive):
    case_vars.put("token", login["token"])
    cache = TepCache(env_vars=env_vars, case_vars=case_vars)
    # 供货商领取需求订单
    xuefeng2_offer_nurse1_receive.receive_offer()
    # pytest原生断言
    assert 1==1 
    # 供货商发货
    xuefeng2_offer_nurse1_receive.nurse_ship()
    # pytest原生断言
    assert 1==1
    # 采购商入库
    xuefeng1_offer_doctor1_need.add_inbound()
    # pytest原生断言 
    assert 1==1 
    # 供货商同步入库数据
    xuefeng2_offer_nurse1_receive.nurse_sync_inboundQuantity()
    # pytest原生断言 下文有详细介绍
    assert 1==1

xuefeng1_offer_doctor1_need、xuefeng2_offer_nurse1_receive 为 session 级别的 fixture

# conftest.py

@pytest.fixture(scope="session")
def xuefeng2_offer_nurse1_receive():
    return Nurse_receive("xuefeng2_nurse1_token")

具体的存放位置，前期可以先写在不同功能模块的test*.py文件里；后期根据情况，可以统一存放在case目录下，如下

image-20230130170609361

用例运行时 项目里的 fixture 会自动导入，要注意的是 fixture 不要重名，否则可能导入的fixture并不是你想要的

测试步骤

测试步骤采用了 @allure.step('xx') , 比如发货功能中的上传运单接口 被我封装为一个函数 它的步骤描述就是供货商上传运单。

步骤的实现函数定义在api文件中的功能模块类 Nurse_receive中：

# nurse_need_api.py

class Nurse_receive(HttpClient):

	def __init__(self, token='xuefeng2_nurse1_token'):
		# 超类继承
		super().__init__()
        # 从配置文件里 读取最新的token文件
        self.token = Operate_token().read_token(section='Token', key=token)
        
	@allure.step('采购商填运单发货')
    def nurse_ship(amounts=[2, 3]):
    	method = 'post'
        url = '/offer/shipment/create'
        # url处理方法
        url = self.get_full_url(url, h=self.host)
        
        # 调用上传运单方法
        trackNo = self.upload_waybill(file_path)
        
        # 组装数据
        list_= []
        for amount in amounts:
            dic_ = {}
            dic_['trackNo'] = trackNo
            dic_['amount'] = str(amount)
            list_.append(dic_)
		
		# 通过缓存池保存数据，方便测试用例其他步骤断言
        self.cache.put('shipmentHistoryList', list_)
        
        body = {"takeOfferId":takeOfferId, "tracks":list_}
        ret = self.send(url=url, body=body, method=method, x_token=self.token)
        assert jmespath.search('code',ret) < 400, f'供货商发货 接口报错：{ret}'
        
     @allure.step('供货商上传运单')   
     def upload_waybill(file_path):
		method = 'post'
        url = '/offer/shipment/create'
        url = self.get_full_url(url, h=self.host)
        body = {"takeOfferId":takeOfferId, "trackNo":trackNos}
        ret = self.send(url=url, body=body, method=method, x_token=self.token)
        assert jmespath.search('code',ret) < 400, f'供货商上传运单 接口报错：{ret}'
        retrun jmespath.search('result',ret)

  1. Nurse_receive类 中有多个 不同小功能的 实现方法

  2. Nurse_receive 类 通过 super().__init__() 继承了 HttpClient的初始化方法，有能力的朋友可以自行重写

  3. HttpClient的初始化方法里有 自动实例了一个 缓存变量池、自动读取了一个yaml配置文件里的环境变量数据（包含 host域名、账号密码、mysql数据库等信息）

  4. HttpClient里还对 requests库 进行了二次封装，参照postman发送请求时的操作，通过入参body_type 即可控制 requests请求 json、文件以及文件流、表单数据，极大提升书写效率

  5. HttpClient里还对 url 进行了二次封装，书写接口请求时 只需要填写接口路径, 再调用 get_full_url 即可实现多场景的完整url。

测试数据

测试数据有两种存在形式

  1. 对于业务链路比较长的业务场景，做参数化会比较繁琐，工作过程中这类冒烟用例的测试数据 我是直接写在 test_offer_doctor.py

  2. 对于比较简单的接口，测试人员时间允许的情况下 可以讲测试数据写在yaml里，用于参数化，覆盖更多的比如边界值、异常等场景

测试标题

测试标题采用了@allure.title("")：

@allure.title('验证采购商发offer功能正常')
def test(xuefeng1_offer_doctor1_need):

变量

环境变量：在resources/env_vars下不同环境的yaml文件里预填变量，在resources/aamt.ini中激活某个环境，模块api.py的功能类Nurse_receive已经继承HttpClient类（会自动读取环境变量），可使用 self.host、self.env_vars_data 直接引用：

# nurse_need_api.py

class Nurse_receive(HttpClient):

	def __init__(self, token='xuefeng2_nurse1_token'):
		# 超类继承
		super().__init__()
		
	def aaa():
		# 直接引用 环境变量
		self.env_vars_data
		# 直接引用 域名
		self.host
		# 默认header头
		self.default_header

用例变量：在用例中引入case_vars fixture，不同步骤函数的数据通过case_vars.put()、case_vars.get()传递,

当用例中下一个步骤的断言 需要用到 上一个步骤的数据（入参或者响应），可通过各个功能类中的缓存变量池 self.cache来获取，存到case_vars中，随用随取。

@allure.title("从登录到下单支付")
def test(case_vars):
    case_vars.put("token", login["token"])
    cache = TepCache(env_vars=env_vars, case_vars=case_vars)

    Step("搜索商品", step_search_sku, cache)
    Step("添加购物车", step_add_cart, cache)


def step_search_sku(cache: TepCache):
    url = cache.env_vars["domain"] + "/searchSku"
    headers = {"token": cache.case_vars.get("token")}
    body = data("查询SKU")

    response = request("get", url=url, headers=headers, params=body)
    assert response.status_code < 400

    cache.case_vars.put("skuId", response.jsonpath("$.skuId"))
    cache.case_vars.put("skuPrice", response.jsonpath("$.price"))

def step_add_cart(cache: TepCache):
    url = cache.env_vars["domain"] + "/addCart"
    headers = {"token": cache.case_vars.get("token")}
    body = data("添加购物车")
    body["skuId"] = cache.case_vars.get("skuId")

    response = request("post", url=url, headers=headers, json=body)
    assert response.status_code < 400

接口关联

如上所述，通过想HttpClient类中的缓存变量池cache实现了不同的接口间的关联，上一个接口的响应，提取后存入self.cache，下一个接口的入参，从self.cache取值。

数据提取

utils/http_client.py封装了requests.Response，添加了jsonpath方法，支持简单取值：

使用 jmespath或者jsonpath 取值

import jmespath
# 响应：ret结果
code = jmespath.search('code',ret)

断言

采用Python原生的assert断言。16种常用断言如下：

import allure


@allure.title("等于")
def test_assert_equal():
    assert 1 == 1


@allure.title("不等于")
def test_assert_not_equal():
    assert 1 != 2


@allure.title("大于")
def test_assert_greater_than():
    assert 2 > 1


@allure.title("小于")
def test_assert_less_than():
    assert 1 < 2


@allure.title("大于等于")
def test_assert_less_or_equals():
    assert 2 >= 1
    assert 2 >= 2


@allure.title("小于等于")
def test_assert_greater_or_equals():
    assert 1 <= 2
    assert 1 <= 1


@allure.title("长度相等")
def test_assert_length_equal():
    assert len("abc") == len("123")


@allure.title("长度大于")
def test_assert_length_greater_than():
    assert len("hello") > len("123")


@allure.title("长度小于")
def test_assert_length_less_than():
    assert len("hi") < len("123")


@allure.title("长度大于等于")
def test_assert_length_greater_or_equals():
    assert len("hello") >= len("123")
    assert len("123") >= len("123")


@allure.title("长度小于等于")
def test_assert_length_less_or_equals():
    assert len("123") <= len("hello")
    assert len("123") <= len("123")


@allure.title("字符串相等")
def test_assert_string_equals():
    assert "dongfanger" == "dongfanger"


@allure.title("以...开头")
def test_assert_startswith():
    assert "dongfanger".startswith("don")


@allure.title("以...结尾")
def test_assert_startswith():
    assert "dongfanger".endswith("er")


@allure.title("正则匹配")
def test_assert_regex_match():
    import re
    assert re.findall(r"don.*er", "dongfanger")


@allure.title("包含")
def test_assert_contains():
    assert "fang" in "dongfanger"
    assert 2 in [2, 3]
    assert "x" in {"x": "y"}.keys()


@allure.title("类型匹配")
def test_assert_type_match():
    assert isinstance(1, int)
    assert isinstance(0.2, float)
    assert isinstance(True, bool)
    assert isinstance(3e+26j, complex)
    assert isinstance("hi", str)
    assert isinstance([1, 2], list)
    assert isinstance((1, 2), tuple)
    assert isinstance({"a", "b", "c"}, set)
    assert isinstance({"x": 1}, dict)

测试报告

allure下载地址： https://github.com/allure-framework/allure2/releases

解压后将bin目录添加到系统环境变量Path。

可以 执行以下命令

pytest --aamt-reports

或者 运行项目根路径main.py

image-20230130194450349

就能一键生成Allure测试报告，并且会把请求入参和响应出参，记录在测试报告中。

身份鉴权

封装好了HttpClient，有时候调试或者写爬虫，并不是也不想写test用例的情况下 进行接口调用，我特意将登录token信息 保存在了ini文件里，接口调用的时候 只需在实例功能模块类的时候 传入 不同账号的token标识Nurse_receive("xuefeng2_nurse1_token")

http://biji.51automate.cn/blogs/img/20230130191502.png

起初的时候 会有顾虑，每次调用都要实例化吗？ 当然没必要如此，我们只需要将一个个功能模块的实例统一存放在一个conftest.py 里

image-20230130192425665

借助pytest的fixture夹具，编写用例的时候就可以 重复 调用，即插即用。如下

http://biji.51automate.cn/blogs/img/20230130193145.png

用例执行

串行

使用pytest命令即可执行用例。

如果想要生成测试报告，使用以下命令

pytest --aamt-reports

并行

使用pytest -n auto，由pytest-xdist提供支持。

pytest -n auto --aamt-reports

特色功能

fixtures自动导入

不是必须在conftest.py里面定义fixture。只要在fixtures目录下，创建以fixture_开头的文件，fixture都会自动加载到pytest中 。

这样做的还有一个好处

不同的测试人员，维护自己的fixture_xx.py文件，在push代码时，可以高效的处理冲突，方便维护。

image-20230131114647493

分布式环境下 全局只登录一次, 解决token复用问题

预置了fixtures/fixture_xf.py登录接口，且全局仅执行一次，解决token复用问题：

import pytest

from api.logintoken import Login_after

@pytest.fixture(scope="session", autouse=True)
def login_doctor1(aamt_context_manager, env_vars_data):
    	"""
        aamt_context_manager 可兼容pytest-xdist分布式执行的上下文管理器
        即便分布式执行用例，该login只会在整个运行期间执行一次
        """
    def produce_expensive_data(variable):
        # 这里的登录方法 ，抽离出来，1、方便换不同的账号登录 2、遇到前后台不同的方法登录，也可以有效的区分		开，统一管理。
        return Login_after(variable['xuefeng_doctor1'], 'xuefeng_doctor1_token').login_()
    
    return aamt_context_manager(produce_expensive_data, env_vars_data)

aamt的fixture.py 内置了 封装了aamt_context_manager fixture，可兼容pytest-xdist 分布式执行的上下文管理器，即便多个session 也只会执行一次登录，具体实现方法后面会详细介绍。

常用封装

project.py，项目基本信息，比如根目录路径。

import os
from loguru import logger


class Config:
    # 全局项目根目录
    # root_dir = os.path.dirname(os.path.dirname(os.getcwd()))
    root_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
    # logger.info(f'全局项目根目录:{root_dir}')

    data_dir = os.path.join(root_dir, "data")
    file_dir = os.path.join(root_dir, "file")

emailhelper.py 封装了 offer365、qq、163等不同服务商发送邮件的方法

mysqlhelper.py 封装了 增删改查的方法，可快速上手。

工具包

fastapi_mock.py，示例应用。

import uvicorn
from fastapi import FastAPI, Request

app = FastAPI()


@app.post("/login")
async def login(req: Request):
    body = await req.json()
    if body["username"] == "dongfanger" and body["password"] == "123456":
        return {"token": "de2e3ffu29"}
    return ""


@app.get("/searchSku")
def search_sku(req: Request):
    if req.headers.get("token") == "de2e3ffu29" and req.query_params.get("skuName") == "电子书":
        return {"skuId": "222", "price": "2.3"}
    return ""


@app.post("/addCart")
async def add_cart(req: Request):
    body = await req.json()
    if req.headers.get("token") == "de2e3ffu29" and body["skuId"] == "222":
        return {"skuId": "222", "price": "2.3", "skuNum": "3", "totalPrice": "6.9"}
    return ""


@app.post("/order")
async def order(req: Request):
    body = await req.json()
    if req.headers.get("token") == "de2e3ffu29" and body["skuId"] == "222":
        return {"orderId": "333"}
    return ""


@app.post("/pay")
async def pay(req: Request):
    body = await req.json()
    if req.headers.get("token") == "de2e3ffu29" and body["orderId"] == "333":
        return {"success": "true"}
    return ""


if __name__ == '__main__':
    uvicorn.run("fastapi_mock:app", host="127.0.0.1", port=5000)

mitm.py，流量录制，做的不是很好，将就看看。

#!/usr/bin/python
# encoding=utf-8

# mitmproxy录制流量自动生成用例

import os
import time

from mitmproxy import ctx

project_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
tests_dir = os.path.join(project_dir, "tests")
# tests/mitm
mitm_dir = os.path.join(tests_dir, "mitm")
if not os.path.exists(mitm_dir):
    os.mkdir(mitm_dir)
# 当前时间作为文件名
filename = f'test_{time.strftime("%Y%m%d_%H%M%S", time.localtime())}.py'
case_file = os.path.join(mitm_dir, filename)
# 生成用例文件
template = """import allure
from utils.http_client import request


@allure.title("")
def test(env_vars):
"""
if not os.path.exists(case_file):
    with open(case_file, "w", encoding="utf8") as fw:
        fw.write(template)


class Record:
    def __init__(self, domains):
        self.domains = domains

    def response(self, flow):
        if self.match(flow.request.url):
            # method
            method = flow.request.method.lower()
            ctx.log.error(method)
            # url
            url = flow.request.url
            ctx.log.error(url)
            # headers
            headers = dict(flow.request.headers)
            ctx.log.error(headers)
            # body
            body = flow.request.text or {}
            ctx.log.error(body)
            with open(case_file, "a", encoding="utf8") as fa:
                fa.write(self.step(method, url, headers, body))

    def match(self, url):
        if not self.domains:
            ctx.log.error("必须配置过滤域名")
            exit(-1)
        for domain in self.domains:
            if domain in url:
                return True
        return False

    def step(self, method, url, headers, body):
        if method == "get":
            body_grammar = f"params={body}"
        else:
            body_grammar = f"json={body}"
        return f"""
    # 描述
    # 数据
    # 请求
    response = request(
        "{method}",
        url="{url}",
        headers={headers},
        {body_grammar}
    )
    # 提取
    # 断言
    assert response.status_code < 400
"""


# ==================================配置开始==================================
addons = [
    Record(
        # 过滤域名
        [
            "http://www.httpbin.org",
            "http://127.0.0.1:5000"
        ],
    )
]
# ==================================配置结束==================================

"""
==================================命令说明开始==================================
# 正向代理（需要手动打开代理）
mitmdump -s mitm.py
# 反向代理
mitmdump -s mitm.py --mode reverse:http://127.0.0.1:5000 --listen-host 127.0.0.1 --listen-port 8000
==================================命令说明结束==================================
"""

client.py，requests库二次封装，当然有基础的朋友也可以使用原生库

#!/usr/bin/python
# encoding=utf-8

import decimal
import json
import time

import allure
import jsonpath
import requests
import urllib3
from loguru import logger
from requests import Response

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


def request(method, url, **kwargs):
    template = """\n
Request URL: {}
Request Method: {}
Request Headers: {}
Request Payload: {}
Status Code: {}
Response: {}
Elapsed: {}
"""
    start = time.process_time()
    response = requests.request(method, url, **kwargs)  # requests.request原生用法
    end = time.process_time()
    elapsed = str(decimal.Decimal("%.3f" % float(end - start))) + "s"
    headers = kwargs.get("headers", {})
    kwargs.pop("headers")
    payload = kwargs
    log = template.format(url, method, json.dumps(headers), json.dumps(payload), response.status_code, response.text,
                          elapsed)
    logger.info(log)
    allure.attach(log, f'request & response', allure.attachment_type.TEXT)
    return TepResponse(response)


class TepResponse(Response):
    """
    二次封装requests.Response，添加额外方法
    """

    def __init__(self, response):
        super().__init__()
        for k, v in response.__dict__.items():
            self.__dict__[k] = v

    def jsonpath(self, expr):
        """
        此处强制取第一个值，便于简单取值
        如果复杂取值，建议直接jsonpath原生用法
        """
        return jsonpath.jsonpath(self.json(), expr)[0]

关于aamt的更多技术细节，请在源码中一探究竟吧。也可以添加微信xiaobangzhu007，随时与我联系。

后续

aamt小工具后期将会根据实际业务场景进行微调，结构上不会做大的改动。