如何用pytest-playwright对Dash应用做多用户并发负载测试

背景 #

我有一个极简Dash应用：

from dash import Dash, html, callback
from dash.dependencies import Input, Output
import dash_bootstrap_components as dbc
from time import sleep

app = Dash(__name__)

app.layout = html.Div([
    dbc.Button('Click me', id='button'),
    html.Div(id='output')
])


@callback(
    Output('output', 'children'),
    Input('button', 'n_clicks')
)
def click_and_sleep(n_clicks):
    sleep(1)
    return f'Button clicked {n_clicks} times'


if __name__ == '__main__':
    app.run_server(debug=True, port=8000)

对应的pytest-playwright测试用例如下：

from playwright.sync_api import Page, expect

def test_app(page: Page):
    page.goto("http://localhost:8000/")
    page.wait_for_selector("#output")
    for n in range(3):
        page.click("#button")
        expect(page.locator("#output")).to_have_text(f"Button clicked {n+1} times")

核心问题：是否可以基于此设置模拟多用户同时操作，对我的应用进行负载测试？

编辑1：尝试多用户装饰器但串行执行 #

基于思路，我编写了生成n个用户的page工厂：

def page_factory(
    num_users: int,
) -> list[Page]:
    pages = []
    for _ in range(num_users):
        context = browser.new_context()
        page = context.new_page()
        pages.append(page)
    return pages

以及包裹测试函数的多用户装饰器：

def run_with_multiple_users(num_users: int) -> Callable:
    def decorator(test_func: Callable) -> Callable:
        @wraps(test_func)
        def wrapper(*args, **kwargs):
            pages = page_factory(num_users)
            for page in pages:
                test_func(page, *args, **kwargs)
                page.close()

        return wrapper

    return decorator

使用方式：

@run_with_multiple_users(num_users=10)
def test_app(...)
   ...

执行命令：pytest -k test_app -n 10

问题：浏览器打开了10个标签页，但测试是串行执行，没有同时进行。

编辑2：尝试多进程遇到序列化错误 #

改用多进程替代装饰器内的循环：

with multiprocessing.Pool(processes=multiprocessing.cpu_count()) as pool:
   pool.map(partial(test_func, *args, **kwargs), pages)

报错：

FAILED test_basic.py::test_app - _pickle.PicklingError: Can't pickle <function test_app at 0x7851ce61eb60>: it's not the same object as test_basic.test_app

解决方案 #

1. 用pytest-xdist原生并行能力（最简单） #

放弃自定义装饰器，让每个测试进程独立创建浏览器上下文和Page，直接利用pytest-xdist的-n参数实现并行：

from playwright.sync_api import sync_playwright, expect

def test_app():
    with sync_playwright() as p:
        browser = p.chromium.launch()
        page = browser.new_page()
        page.goto("http://localhost:8000/")
        page.wait_for_selector("#output")
        for n in range(3):
            page.click("#button")
            expect(page.locator("#output")).to_have_text(f"Button clicked {n+1} times")
        browser.close()

执行命令：pytest -k test_app -n 10，10个进程会同时启动各自的浏览器实例，模拟10个用户并行操作。

2. 用Playwright异步API实现单进程内并行 #

如果想在单个进程内模拟多用户并行，使用Playwright异步API结合asyncio（同步API本身是阻塞的，无法并行）：

import asyncio
from playwright.async_api import async_playwright, expect

async def single_user_flow():
    async with async_playwright() as p:
        browser = await p.chromium.launch()
        page = await browser.new_page()
        await page.goto("http://localhost:8000/")
        await page.wait_for_selector("#output")
        for n in range(3):
            await page.click("#button")
            await expect(page.locator("#output")).to_have_text(f"Button clicked {n+1} times")
        await browser.close()

async def test_multiple_users():
    # 模拟10个用户同时执行
    tasks = [single_user_flow() for _ in range(10)]
    await asyncio.gather(*tasks)

def test_app():
    asyncio.run(test_multiple_users())

执行这个测试用例，会在单个进程内同时启动10个浏览器上下文，实现多用户并行操作。

3. 修复多进程序列化问题 #

如果坚持用多进程方案，不要传递无法序列化的Page对象，让每个进程独立创建浏览器和Page：

import multiprocessing
from playwright.sync_api import sync_playwright, expect

def single_user_test():
    with sync_playwright() as p:
        browser = p.chromium.launch()
        page = browser.new_page()
        page.goto("http://localhost:8000/")
        page.wait_for_selector("#output")
        for n in range(3):
            page.click("#button")
            expect(page.locator("#output")).to_have_text(f"Button clicked {n+1} times")
        browser.close()

def run_with_multiple_users(num_users: int):
    with multiprocessing.Pool(processes=num_users) as pool:
        pool.map(lambda _: single_user_test(), range(num_users))

def test_app():
    run_with_multiple_users(num_users=10)

每个进程独立初始化浏览器环境，避免了序列化Page实例的问题，同时实现并行执行。

内容的提问来源于stack exchange，提问作者Luggie