web自动化测试进阶篇03 ———自动化并发测试应用

 
 

😏作者简介：博主是一位测试管理者，同时也是一名对外企业兼职讲师。
📡主页地址：【Austin_zhai】
🙆目的与景愿：旨在于能帮助更多的测试行业人员提升软硬技能，分享行业相关最新信息。
💎声明：博主日常工作较为繁忙，文章会不定期更新，各类行业或职场问题欢迎大家私信，有空必回。

 
 

1. 目的

  web自动化测试作为软件自动化测试领域中绕不过去的一个“香饽饽”，通常都会作为广大测试从业者的首选学习对象，相较于C/S架构的自动化来说，B/S有着其无法忽视的诸多优势，从行业发展趋、研发模式特点、测试工具支持，其整体的完整生态已经远远超过了C/S架构方面的测试价值。

  在我们的测试项目中，如果被测对象有高并发、多用户协作的业务场景，那么作为测试团队来说，一般都会需要利用到并发测试的方式来进行测试介入。而作为众多常见测试种类中的一种，并发测试的重要性也自不必多说，那么我们是否可以利用之前所学的web自动化框架来进行日常的并发测试能？答案自然是肯定的，今天就由博主来为大家详细的介绍一下web自动化测试框架如何在并发测试中大展拳脚。

 
 

2. 基本思路

  做过并发测试的同学应该都知道，并发测试的作用无非就是验证一些功能在高并发的业务场景中，但其实它的适用场景远远不止这些，比如一些多用户协作、资源抢占或同步的业务场景中都会需要利用并发测试来验证其性能与健壮性。

  既然知道我们做并发测试是为了满足以上这些场景的测试需求，那么将自动化测试结合在其中的意义又是什么呢？这个其实和我们在黑盒测试中加入自动化测试的出发点与理念较为相同，但又稍稍的有着一些差别。加入自动化测试的并发测试其中的效率问题自然不必多提，做过自动化测试的同学应该都是感同身受的，另外对于一般的并发测试来说，改为自动化之后的测试结果准确性也是有着不少的提高，试想在同样的测试条件下，人与机器的执行准确率子不用多说，我们的自动化测试脚本本身就是“没有感情的测试执行机”，不受情绪、环境、状态的影响，可以忠诚的完成我们交予的任何测试用例。而我们的测试人员就可以抽出精力来从事一些并发报告分析与定位的动作。

  还有一点是需要大家明白的是，我们的UI自动化并发测试的基本思路，当你的被测对象（UI）存在需要进行上述所说的一些业务场景测试的情况下，我们才会进行并发测试。那么我们的测试关注点一般不会再去确认被测功能的正确性了，因为这一步我们已经在黑盒中或UI自动化中已经完成了。我们在自动化并发测试中需要关注的则是UI界面的业务并发操作，比如模拟大量用户同时进行操作某些页面元素等等，简而言之，就是检查并发状态下被测对象页面的交互与操作响应。

 
 

3. 主体流程

  对于我们的自动化并发测试来说，流程中的部分流程节点与自动化UI测试本身没有太大的区别，所以掌握基本的测试流程是先决条件，这个对于广大接触过自动化测试的测试同学来说基本不是问题。这里会着重介绍自动化并发测试特有的一些关键节点与具体执行内容。

 

3.1 场景定义

  这个阶段测试同学需要与产品一起确定被测对象的哪些业务需要并发测试进行介入，描述并定义相关的并发业务场景，这里需要注意的是，并发场景不单单包括上述所说的一些测试需求，主要还是看产品与测试对于产品的理解程度。比如一些秒杀、在线文档、热门话题讨论、支付系统、多人对战游戏等等，这些被测对象的业务场景基本都是常见的并发测试介入场景。至于这里说的理解程度，无非就是在某个垂直领域中各自对于被测对象的行业业务与具体介入场景的理解与测试需求转化能力。那么根据业务理解加之系统特点，制定合适的并发测试场景。其中最基本的就是确定需要模拟的并发用户数量、并发操作的类型以及测试的时间范围等。

 

3.2 用例设计

  有了之前的具体场景设计，测试同学就可以进行对应的并发测试用例的设计工作了，这一步其实与手工测试用例的设计理念没有特别大的区别，具体还是将之前的并发业务场景进行进一步的细化，比如每个并发测试用例具体包含多个用户同时执行，具体的操作步骤是什么。至于测试用例的覆盖范围依旧与手工测试用例一致，必须覆盖对应业务流程和交互，以验证系统在并发场景下的正确性和性能。这里再次强调，并发测试中的正确性验证是基于手工测试已经完成的前提下，我们关注的是在一定量级的用户并发操作下，功能仍旧可以保证其正确性和良好性能表现，千万不可将并发测试的结果代替黑盒测试，两者的结果并不能划等号。

 

3.3 工具选择

  在这一阶段，我们就需要根据产品自身的软件架构特性与并发场景的测试需求来进行对应自动化工具与框架的选择了。因为我们这边介绍的是web产品，那基本就逃不开java+selenium或者python+selenium这样的工具与框架。基于之前的习惯，我们这次仍旧用python+selenium的组合来进行后续的介绍。这里还是啰嗦一句，各自的实际项目还是需要根据实际现状进行有效选择，切勿盲目抄作业。

 

3.4 脚本编写

  实现阶段的重要性自不必要多说了吧，这里与一般的UI自动化脚本编写不同的点在于，你需要将并发这个概念加入其中，也就是我们行业里日常所说的多线程编程。脚本中会包含多个用户同时执行的操作步骤，并需要考虑到并发访问的资源竞争和同步等问题。比如使用python，那么我们可以使用语言提供的并发控制和线程管理功能，确保执行时测试用例在并发测试环境中被正确执行。

 

3.5 环境搭建

  其实这一步倒不用太多的介绍，测试环境的搭建依旧是大家熟门熟路的内容，这里需要关注的是测试客户端侧的环境搭建与配置，另外就是你的被测对象的硬件环境是否可以支持测试中并发的负担，这里不推荐把环境的硬件条件配置的和黑盒测试的时候一样，虽然对于测试环境来说无需像生产环境那样一摸一样，但如果因为硬件条件不足的因素而导致并发测试的结果发生偏差的话，那就明显是一件得不偿失的事情了。

 

3.6 测试执行

  脚本或者框架调试完之后就可以正式的执行测试了，这一步基本就靠程序全自动运行了，那么作为测试同学来说需要做的事情就是全程观察并记录系统的响应时间、资源使用情况、错误和异常情况等。如果以上的这些动作你们也做了集成，那么就当我没说。

 

3.7 分析优化

  这一步是所有并发测试中真正重要的一环，也是对于广大测试同学来说考验与自身能力提升最大的一个阶段。分析测试报告数据自不必多说，本身就是测试团队的分内工作，是否可以根据报告数据找到性能瓶颈并给出自己的优化建议，对于测试同学来说才是真正有挑战的事项。如果你单纯的以为有开发会做这些事情，那么我只能说你是白白错失了许多让自己提升核心竞争力的大好机会。千万不要固化的认为软件不是测试开发出来的，所有相关的代码逻辑与性能表现是事不关己。一个能真正看懂报告数据并给出性能调优建议的测试，在当下的IT行业中有着怎样的价值，如果硬要举个例子的话，我只能说是降维打击。

 

3.8 持续优化

  这里的持续优化有两层意思，一个是我们的对应测试脚本与框架，另一个就是我们的被测对象。并发测试并不是一蹴而就的，自然他的测试结果与生命周期也不是考一次测试或调优就可以说大功告成的。作为测试同学来说，被测对象上线、投放至市场之后的用户反馈就是我们日后持续优化的方向与重要参考。当然，对应的并发测试场景也不会是一尘不变，后续发生需求变更而导致部分重构或推翻的情况也需要测试团队及时对脚本或框架做出相应的改变。

 
 

4. 脚本实操

  基于以上的说的一些基本思路与执行流程，我们接下来还是把重心放在对应的并发测试脚本的编写上来。这里再强调一遍，编写基础的并发测试脚本还是需要大家有较为扎实的web自动化测试脚本功底，我们可不能在脚本中再去犯一些基础的编程错误了，比如元素定位不准确，逻辑较为混乱，因为在并发测试脚本中，我们需要着重的关注脚本的并发效率与实际被测对象的承受能力，更要去关注一些资源的同步与抢占问题，所以可以说我们的自动化并发测试脚本是在原有的web自动化测试基础上更上一层。

  这里的演示实例博主还是用金融行业中的基金交易的业务场景来进行说明。交易系统与支付系统其实都是需要进行大量的并发测试来验证其性能与健壮性的，像这种与钱相关的业务系统与场景，都不能出一丁点问题，不然面对的可能就是各种公司与用户的财产损失，那性质可是相当严重的。所以在确保其业务场景中的所有单点功能没有问题的前提下，我们就可以利用并发测试来模拟日常用户的大批量业务并发操作，以此来确保被测对象的良好质量表现。

 

import logging
import pytest
import threading
import pymysql
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.chrome.service import Service
from concurrent.futures import ThreadPoolExecutor

@pytest.fixture(scope='session')
def driver(request):
    driver = webdriver.Chrome(service=Service('/opt/rtc_uat/driver/chromedriver'))
    request.addfinalizer(driver.quit)
    return driver

def get_user_credentials_db():
    # 从数据库读取用户的凭证信息
    connection = pymysql.connect(host='172.20.30.130', user='username', password='password', database='rtc_base')
    cursor = connection.cursor()
    query = "SELECT username, password FROM tb_fund_base_account"
    cursor.execute(query)
    result = cursor.fetchall()
    cursor.close()
    connection.close()
    return result

def get_fund_information_db():
    # 从数据库读取基金信息
    connection = pymysql.connect(host='172.20.30.130', user='username', password='password', database='rtc_trade')
    cursor = connection.cursor()
    query = "SELECT code, amount, trade_type FROM tb_fund_base_info"
    cursor.execute(query)
    result = cursor.fetchall()
    cursor.close()
    connection.close()
    return result

def login_and_authenticate(driver, credentials):
    # 登录鉴权
    username = credentials[0]
    password = credentials[1]
    driver.get('http://www.test.com/login/')
    username_input = driver.find_element(By.ID, 'username')
    username_input.send_keys(username)
    password_input = driver.find_element(By.ID, 'password')
    password_input.send_keys(password)
    submit_button = driver.find_element(By.ID, 'login-button')
    submit_button.click()
    try:
        driver.find_element(By.ID, 'authentication-success')
        confirm_button = driver.find_element(By.ID, 'base-annou-confirm-btn')
    	submit_button.click()
    except NoSuchElementException as e:
        logging.error(f"用户认证失败，用户名：{username}，错误信息: {str(e)}")
        raise

def purchase_and_redeem_fund(driver, fund_info):
    # 基金购买与赎回逻辑
    fund_code = fund_info[0]
    amount = fund_info[1]
    trade_type = fund_info[2]
    driver.get('http://www.test.com/fund/trade/')
    fund_code_input = driver.find_element(By.ID, 'fund-code')
    fund_code_input.send_keys(fund_code)
    amount_input = driver.find_element(By.ID, 'amount')
    amount_input.send_keys(amount)
    submit_button = driver.find_element(By.ID, 'buy-button')
    submit_button.click()
	try:
        driver.find_element(By.ID, 'authentication-success')
        confirm_button = driver.find_element(By.ID, 'base-annou-confirm-btn')
    	submit_button.click()
    except NoSuchElementException as e:
    	if trade_type == 1:
        	logging.error(f"基金购买失败，基金名：{fund_id}，错误信息: {str(e)}")
        elif trade_type == 2:
        	logging.error(f"基金赎回失败，基金名：{fund_id}，错误信息: {str(e)}")
        else:
        	logging.error(f"交易类型错误，请检查数据库对应记录字段")
        raise

def test_concurrent_operations(driver):
    # 获取对应信息
    user_credentials = get_user_credentials_db()
    fund_information = get_fund_information_db()
    # 创建线程池
    max_workers = 100
    executor = ThreadPoolExecutor(max_workers=max_workers)
    # 提交并发任务
    futures = []
    for credentials, fund_info in zip(user_credentials, fund_information):
        future = executor.submit(concurrent_operation, driver, credentials, fund_info)
        futures.append(future)
    # 等待所有任务完成
    for future in futures:
        future.result()

def concurrent_operation(driver, credentials, fund_info):
    # 执行对应方法
    login_and_authenticate(driver, credentials)
    purchase_and_redeem_fund(driver, fund_info)

# 执行并发测试
test_concurrent_operations(driver)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105

  在上面的并发测试脚本中，博主只演示了最基本的编写思路，对于并发测试中的并发需求，这里使用了python中的线程池来对线程进行管理，在concurrent.futures模块中可以利用ThreadPoolExecutor来实现我们的测试需求。当然python中不仅仅只有这一种方法可以实现，具体的还是根据各自的习惯与团队规范来选择。

 
 

5. 后话

  限于篇幅的关系，关于自动化并发测试的内容我们就先介绍这些，如果后续大家还有需要的话请私信或留言告知到我，我会增加篇幅来介绍更多自动化并发测试的相关内容。

  另外，在我们执行并发测试的时候需要注意一下相关测试数据的隔离问题，因为这个与一般的UI自动化测试不同，我们是在大量业务操作同时并发操作的前提下去进行测试与数据对比的，所以我们需要确保每个用户的测试数据应该是独立的，互不干扰。至于使用数据工厂还是其他类型的数据生成方法，这个就是见仁见智的问题了。还有就是如果测试脚本执行的过程中出现了线程相关的报错，注意检查各个业务操作的顺序，例如，确保在进行下一步操作之前，前一步操作已经完成，避免出现竞争条件和数据不一致的情况。因为并发测试涉及多个线程同时执行操作，所以需要确保线程之间的同步和协调。这个也是每个测试同学编写与调试自动化测试并发测试脚本的必过课题。