标签：name temp 项目 self token dict user 测数据

压测流程

首先启动 locust 压测脚本
然后启动bus查分模拟脚本
收集数据
压测结束，清理数据

采集的数据为：

请求相关数据，如响应时间，请求总数据量
资源相关，请求时pod的数量以及实时cpu，内存消耗
请求数量数量，总请求数量，时间分布
apm请求记录，查询请求具体耗时
数据库信息，记录网络连接数变化
locust请求端数据总计，错误数量，RPS，耗时统计等

单测查询接口

单独测试查询结果的接口，要求：返回时间不超过1s，如果超过1s则添加pod，测试极限压力

查分实际场景模拟
在查分时有查询接口和添加准考证，而且还会有查分bus。查实多个接口调用下的各项参数指标。

数据准备
用户：cettest20~cettest79 共60个
获取用户token，保存在redis中，后续请求需要携带token发起请求。
注意：该接口只有staging环境才能使用

import requests
import redis

pool = redis.ConnectionPool(host='localhost', port=6379, decode_responses=True)
r = redis.Redis(connection_pool=pool)


token_dict = {}
url = "https://apiv3.shanbay.com/bayuser/login"
for i in range(20,80):
    user_name = f"cettest{str(i)}"
    
    payload = {
        "account": user_name,
        "code_2fa":"",
        "password": "123456"
    }
    headers = {"X-API-TOKEN": "IBILiaNGtorE"}
    res = requests.post(url, json=payload, headers=headers)
    if 200 == res.status_code:
        print(user_name)
        token_dict[user_name] = res.cookies.get("auth_token")
    r.hmset("cettest_user_token_dict", token_dict)


print(token_dict)
token_dict_redis = r.hgetall("cettest_user_token_dict")
print(token_dict_redis)

locust 压测脚本

locust_test.py

import time
import random
import redis
from locust import HttpUser, TaskSet, task, constant_throughput


class UserBehavior(HttpUser):
    wait_time = constant_throughput(2)

    def on_start(self):
        pool = redis.ConnectionPool(host='localhost', port=6379, decode_responses=True)
        r = redis.Redis(connection_pool=pool)

        # cettest20~cettest79 共60个测试账号的token
        self.token_dict = r.hgetall("cettest_user_token_dict")  # 字典 {user_name: token ...}
        self.token_list = [token for token in self.token_dict.values()]
        self.user_list = []
        for i in range(20, 80):
            temp = {}
            temp["name"] = f"cettest{str(i)}"
            temp["ticket"] = str(random.randint(5,9))*15
            temp["phone_num"] = 11111111111
            temp["source_plan_id"] = random.randint(19,20)
            self.user_list.append(temp)

    @task(4)
    def get_user_ticket(self):
        url = "https://apiv3.shanbay.com/tickethub/ticket/user?code=2021-12-cet"
        cookies = {"auth_token": random.choice(self.token_list)}
        self.client.get(url, cookies=cookies)
    
    @task(2)
    def post_user_ticket(self):
        url = "https://apiv3.shanbay.com/tickethub/ticket/user?code=2021-12-cet&is_in_shanbay=1"
        user = random.choice(self.user_list)
        payload = {
            "channel": 0,
            "name": user["name"],
            "phone_num": user["phone_num"],
            "source_plan_id": user["source_plan_id"],
            "ticket": user["ticket"]
        }
        cookies = {"auth_token": self.token_dict[user["name"]]}
        self.client.post(url, cookies=cookies, json=payload)

    @task(1)
    def put_user_ticket(self):
        url = "https://apiv3.shanbay.com/tickethub/ticket/user?code=2021-12-cet&is_in_shanbay=1"
        user = random.choice(self.user_list)
        payload = {
            "name": user["name"],
            "phone_num": user["phone_num"],
            "ticket": user["ticket"]
        }
        cookies = {"auth_token": self.token_dict[user["name"]]}
        self.client.put(url, cookies=cookies, json=payload)

启动脚本：

locust -f  locust_test.py

访问：
127.0.0.1:8098 填写用户数和每秒增长数
RPS计算：代码中的 wait_time 是一次请求的间隔时间，为1表示请求一次间隔1s，RPS为1。用户数指开启的协程数。所以RPS = wait_time * 用户数
bus 查分模拟脚本
过程：查询出post请求创建的用户，调用查分结果的bus写入分数。
数据特征：所有测试用户的手机号都是 11111111111 学校 college为 "压测学院"

import time
import random
from app import models as am
from app.buses import send_user_grades


user_tickets = am.UserTicket.select().where(am.UserTicket.phone_num=="bQ8AqNMnFh8pLkOX5tXg9A==")

while True: 
    for user in user_tickets:
        print(f"{user.id}{user.name}")
        user_list = []
        temp = dict()
        temp["name"] = user.name
        temp["ticket"] = user.ticket
        temp["status"] = 0
        temp["reading"] = random.randint(100,300)
        temp["listening"] = random.randint(100,300)
        temp["writing"] = random.randint(100,300)
        temp["total"] = random.randint(300,500)
        temp["school_name"] = "压测学院"
        temp["exam_level"] = random.randint(0,1)
        user_list.append(temp)
        send_user_grades.delay(user_list)
    time.sleep(5)
    print("*" * 20)

数据清除脚本
根据数据特征清除测试数据

from app import models as am
user_tickets = am.UserTicket.select().where(am.UserTicket.phone_num=="bQ8AqNMnFh8pLkOX5tXg9A==", am.UserTicket.college=="压测学院").count()

print(f"将要删除的用户数{user_tickets}")
delete_users = am.UserTicket.delete().where(am.UserTicket.phone_num=="bQ8AqNMnFh8pLkOX5tXg9A==", am.UserTicket.college=="压测学院").execute()
print("删除的用户{delete_users}")