? ? ? ? 本来是想从项目的业务模型来慢慢介绍到需求拆解的，但是转念一想：作为测试人员，自己应该清楚每个人都测试过好多项目，对于我们来说，项目都是大同小异的，具体到测试时都是按功能点划分的，既然这样，那我就按照写测试用例的方式将需求拆解并组织代码设计，这样更加符合测试人员的习惯。【具体流程不会一个个去讲，这样太啰嗦了，会调出典型的来解读，其他的都大同小异】
一、下面4点太长可不看，可以看后面具体会实现哪些功能
1、测试数据准备和清理，即tear_up和tear_down
? ? ? ? 即要模拟创建出需要的虚拟用户（设备），并在测试结束时删除掉这些数据，之所以要介绍一遍，是因为我的常用作法是将这个过程单独拎出来，不放在整个测试流程中，因为不同项目，测试数据的创建时间不同，且测试一遍后也没必要立即清理，所以建议放在单独文件中，与测试代码隔离开
2、登陆过程，即一个典型的挑战登录（即第一次登陆会拿到一个值，然后按照约定的方式重新计算，再请求一次）过程，不同的是拿到的token过期时间有点快，需要1h内通过一个接口拿回一个新token，老token将会失效，该token不光在请求内带上，一些接口的url就是用token拼接的，说到这，如果要用jmeter实现的话，当然能做到，但是必然要花费一番功夫的
3、由于要接受到服务器的主动推送，所以后端要求客户端得建立websoket连接，因此需要建立连接，收发消息
4、需要调用sdk来发送视频流，这个则涉及到python调用c接口的问题，当然这个也可以是等远程调用，考虑到这个十分耗费服务器资源，可以将其放到其他服务器，将接口封装后以常用的服务接口形式暴露，这样就可以当做一个服务器的接口处理即可，当然这个服务需要一定的耐心去完成
上面说了一堆看起来十分长的废话，实际上也只是整个工程一小部分，最终目的还是为了引出下面的代码设计和需求拆解：
1、我们要实现一个可以定时刷新token的功能(开一个线程定时刷新)，并且在任意地方获取到这个token（一个简单的get_token函数）
2、一个稳定的websoket连接，而且要考虑到某些模拟用户意外断开连接后的处理
3、对于耗费服务器资源或者不好在python工程中处理的东西，可以将繁琐的东西转换成简单的接口暴露出来，对于我们来说，看做一个测试流程的接口即可
二、避开一些常见的坑
1、locust版本问题，如果你搜索其他博客，会发现一些代码demo会报错，这个确实是locust的问题，哪有更新版本将一些关键的类名和方法名直接改了，在这里也不列举了，直接用新版或者指定老版本安装即可（我用的是新版，locust -V? locust 2.1.0）
2、接下来都是和版本无关的
# 继承HttpUser（Locust User FastHttpUser）的类则代表一个用户
class UserInfo(FastHttpUser):
sn_queue = queue.Queue()
for i in range(0, 400):
sn_queue.put_nowait(str(999999900000 + int(i)))
tasks = [MockUserTask]#?通过@task来定义一个行为，用户可以包含多个行为
# 开户销户放到单独的模块
# 业务流程1:登录并定时获取新的token，要用单独一个线程刷新token
# 将user的fasthttp client作参数传给其他类中，可以将代码分离或者复用之前的requests模块的业务
@task
def test_get_refresh_token_task(self):
self.box_sn = self.user.sn_queue.get()
box_refresh_token_cfg = {"server_ip": self.cfg_mes[0],
"user_name": self.box_sn,
"token_refresh_time": self.cfg_mes[4],
"http_client": self.client}
t_token = get_box_id_token_t.GetIdAndRefreshTokenThread(**box_refresh_token_cfg)
t_token.setDaemon(False)
t_token.start()
self.t_get_refresh_token = t_token
# 要确保能获取到token才能执行后面的业务，fasthttpclient的默认超时是60s，可以检查60s后没有拿到token就终止这个用户
@task
def test_check_token_time_out(self):
tim = 0
while True:
if self.t_get_refresh_token.getBoxToken() != '':
print(self.box_sn + ' getBoxToken sucess')
break
if tim == 60:
print(self.box_sn + " getBoxToken time_out")
# 终止定时获取token的线程
self.t_get_refresh_token.setStopFlag(True)
events.request_failure.fire(request_type='http', name='get_token', response_time=60, response_length=0, exception='fail')
time.sleep(int(self.cfg_mes[5]) * 60)
self.interrupt()
else:
time.sleep(1)
tim = tim + 1#?@task中的行为可以是异步也可以是同步的（同步的则会阻塞直到结束或者return）
#?web界面中?Number?of?users?to?simulate:?用户数?
#?Spawn?rate?(users?to?start?per?second?(or?stop)):每秒用户上线速度?小数如0.1?则代表间隔10秒上线一个用户
#?用户上线完毕后，会循环用户里定义的task任务
#?wait_time?=?between(30,?50)?设置task之间的运行时间间隔
#?要模拟用户只执行一次task?可以用SequentialTaskSet?在最后一个task?定义一个time.sleep到测试结束
#?指定主机?locust?-f?test.py?--master??指定测试机?本机的话直接带worker即可?目的是为了使用多核cpu?locust?-f?test.py?--worker
#?locust?-f?test.py?--web-host=127.0.0.1?运行locust的机器为多网卡时可以指定网卡
这些东西一下子感觉有点杂，实际都是常用的模块，至于怎么用，会在后面具体实现中单独讲解的
三、代码设计解读
（备注：代码工程会在检查后确保不会有其他项目信息后上传）
?整个工程目录如图所示
1、我的项目是模拟一些box，那么我将这个box有关的接口放在一起，即项目中的一些业务接口
2、sdk则是上面说到的需要将其做成一个接口给测试程序调用的那个，即第三方模块
3、test_model_global 则是都要用的东西，比如配置文件读取，挑战登陆的流程，项目中要求的一些封装，自己对于request请求的封装（例如对接口返回值的解析，自动带上token的请求等等）
4、test_thread 测试线程，主要是对于需要一直要做的事务用一个线程单独一直运行，例如刷token，websoket消息的收发
这样一来，比起jmeter，确实更像一个可以维护，多人参与的工程了