您是一名 QA 测试人员,渴望深入进行性能测试,而不需要广泛的编程专业知识吗?在本文中,我们将探索一种可供非程序员使用Python对云应用程序 API 进行负载测试的可行方法。无需复杂编码即可进行负载测试 - 了解常规 QA 测试人员如何使用 Python 来查找严重错误并发现潜在的性能瓶颈。
性能测试是确保您的应用程序能够满足实际需求的一个关键方面。我将尝试解释我的方法和为管理浏览器的云服务进行负载测试而设计的 Python 脚本。
负载测试场景想象一个负责管理浏览器配置文件(用于网页抓取的浏览器)的云服务。用户通过 API 与服务交互以创建、启动、停止、删除等配置文件。我的 Python 脚本模拟了这种情况,通过重复执行这些操作来向云服务施加负载。
# Dependencies import asyncio import httpx # Configuration API_HOST = 'https://cloud.io' API_KEY = 'qatest' API_HEADERS = { "x-cloud-api-token": API_KEY, "Content-Type": "application/json" } CYCLES_COUNT = 3 # Browser profile configuration data_start = { "proxy": "http://127.0.0.1:8080", "browser_settings": {"inactive_kill_timeout": 120} }
get_profiles
函数从服务中检索现有的浏览器配置文件,模拟用户请求信息的场景。 async def get_profiles(cl: httpx.AsyncClient): resp = await cl.get(f'{API_HOST}/profiles', params={'page_len': 10, 'page': 0}, headers=API_HEADERS) return resp.json()
async def start_profile(cl: httpx.AsyncClient, uuid): resp = await cl.post(f'{API_HOST}/profiles/{id}/start', json=data_start, headers=API_HEADERS) if error := resp.json().get('error'): print(f'Profile {id} not started with error {error}')
async def stop_profile(cl: httpx.AsyncClient, uuid): resp = await cl.post(f'{API_HOST}/profiles/{id}/stop', headers=API_HEADERS) if error := resp.json().get('error'): print(f'Profile {id} not stopped with error {error}') async def delete_profile(cl: httpx.AsyncClient, uuid): resp = await cl.delete(f'{API_HOST}/profiles/{id}', headers=API_HEADERS) if error := resp.json().get('error'): print(f'Profile {id} not stopped with error {error}')
for conn in cl._transport._pool.connections: if conn._connection._state.value != 1: continue print(f'Connection in progress: {conn}')
主要功能编排负载测试周期,迭代配置文件并执行异步任务。每个周期代表模拟的用户交互、创建、使用和删除浏览器配置文件。
async def main(): async with httpx.AsyncClient(timeout=httpx.Timeout(timeout=300)) as cl: for _ in range(CYCLES_COUNT): profiles = await get_profiles(cl) start_tasks = [asyncio.create_task(start_profile(cl, profile['id'])) for profile in profiles] await asyncio.gather(*start_tasks) active_browsers = await get_active_profiles(cl) stop_tasks = [asyncio.create_task(stop_profile(cl, active_browser['id'])) for active_browser in active_browsers['data']] await asyncio.gather(*stop_tasks) profiles = await get_profiles(cl) del_tasks = [asyncio.create_task(delete_profile(cl, profile['id'])) for profile in profiles] await asyncio.gather(*del_tasks) # Monitor active connections for insights into load impact
该脚本展示了 QA 为其应用程序定制负载测试场景的基础。通过自定义周期数、调整用户交互以及修改脚本以适应特定的 API 端点,测试人员可以深入了解应用程序在不同负载下的性能。在这里,您将需要必要的监控工具来获取有关服务器状态的信息、评估服务器负载以及跟踪资源利用率和日志。利用Grafana、Kibana、Prometheus等工具进行全面监控。此外,请密切关注脚本收到的响应,确保对应用程序的性能进行全面评估。这种方法对于有效的负载测试和性能分析非常有价值。
此外,为了进行更真实的负载模拟,请考虑在浏览器中打开特定页面。虽然我个人在浏览器中使用了起始页,但您还可以探索 Pyppeteer 或 Playwright 等选项来打开多个选项卡并浏览各个页面。这种方法增强了负载测试场景的真实性,与用户与应用程序的交互非常相似。
# Attempt to connect to the browser using the provided profile URL try: browser = await connect({'browserWSEndpoint': browser_url, 'defaultViewport': None}) except Exception as e: # Handle connection errors and print a message print(f'Error occurred when connecting to the browser: {str(e)}') return # Create a new page in the connected browser page = await browser.newPage() # Introduce a brief delay to ensure the page is ready await asyncio.sleep(2) # Set the viewport dimensions for the page width, height = 1920, 1080 await page.setViewport({'width': width, 'height': height}) # Try to navigate to a specific URL try: await page.goto('https://{your_website}') # Wait for 10 seconds to simulate user interaction await page.waitFor(10000) # Introduce another delay for additional stability await asyncio.sleep(5) except pyppeteer.errors.PageError as e: # Handle page navigation errors and print a message print(f'Error occurred during page navigation: {str(e)}') # Attempt to take a screenshot of the page try: await page.screenshot(path='screen.png', fullPage=True) # Print a success message if the screenshot is captured successfully print('Screenshot taken successfully.') except Exception as e: # Handle screenshot capture errors and print a message print(f'Error occurred during taking a screenshot: {str(e)}')
Python 的异步功能与 HTTP 库相结合,使其成为用于负载测试基于云的系统的多功能工具。此示例可以作为 QA 工程师在负载测试尝试中了解 Python 强大功能的起点。
笔记
在我的场景中,所描述的脚本被证明是强大且有影响力的。它是识别和解决许多问题的有用工具。该脚本的积极性在查明关键问题、促进有效调试以及引领无缝和改进的用户体验方面还可以,这对于 QA 来说非常好。
接下来,我将简要讨论另一个利用 Python 多处理模块的脚本。此方法旨在通过同时执行测试脚本的多个实例来增强负载生成。多处理的主要目标是并行执行脚本,从而实现与服务的同时交互。此方法与前面讨论的异步方法形成对比,在异步方法中,任务是按顺序执行但同时管理的。这更像是具有相同请求的垃圾邮件/DDOS,但它也可能非常有用。
def get_profiles(): response = requests.get(url=f"{api}", params=PARAMS, headers=headers) return response
def start_profiles(list_of_profiles_uuids): for uuid in list_of_profiles_uuids: # ... (API request to start profile) def stop_profiles(internal_uuids): for uuid in internal_uuids: # ... (API request to stop profile) def run_script(): start_profiles(get_profile_ids()) stop_profiles(list_of_ids)
if __name__ == "__main__": for runs in range(0, 5): processes = [] for i in range(20): p = multiprocessing.Process(target=run_script) processes.append(p) p.start() for p in processes: p.join()
多处理为负载测试应用程序提供了一种策略。它允许 QA 工程师根据其应用程序的独特特征尝试不同的方法。虽然异步测试可以提高管理并发任务的效率,但多处理在并行整个测试过程方面表现出色。您可以选择最适合其特定负载测试目标和应用程序要求的方法。
快速提醒:
该演示旨在以初学者友好的方式介绍基本概念,强调 Python 对于涉足性能测试的 QA 测试人员的简单性。
如果您遇到编程挑战,请毫不犹豫地用 google 搜索并询问同事,使用 ChatGPT 或类似工具,并使用 GitHub Copilot 在编写负载测试脚本时获得额外帮助。
也发布在这里。