You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

Playwright Python同步版request事件操作未完成测试已执行

解决Playwright同步版请求事件处理与主流程同步问题

针对你遇到的请求事件处理未完成主流程就推进的问题,这里提供两种实用的同步方案,无需切换到异步API:

方案1:用threading.Event实现线程同步

Playwright同步版中,request事件的处理函数是在后台线程执行的,通过threading.Event可以让主测试线程等待事件处理完成后再继续:

import pytest
import re
from threading import Event
from playwright.sync_api import Page

def test_odata_query_validation(page: Page):
    # 初始化同步事件与验证结果变量
    process_completed = Event()
    odata_filter_is_valid = False
    # 替换为你的ODATA过滤语句正则
    filter_pattern = re.compile(r'your_odata_regex_here')

    def handle_target_request(request):
        nonlocal odata_filter_is_valid
        # 只处理目标API请求,避免无关请求干扰
        if "your_target_api_url_keyword" in request.url:
            odata_filter = request.headers.get("OData-Filter")
            if odata_filter:
                # 执行你的正则提取与验证逻辑
                matched_items = filter_pattern.findall(odata_filter)
                odata_filter_is_valid = len(matched_items) == 1  # 示例验证条件
        # 标记事件处理完成
        process_completed.set()

    # 绑定请求监听事件
    page.on("request", handle_target_request)

    # 执行触发API请求的操作(如点击按钮、提交表单)
    page.click("#trigger-api-action-btn")

    # 等待事件处理完成,设置超时防止无限阻塞
    assert process_completed.wait(timeout=10000), "请求处理超时"

    # 执行后续验证
    assert odata_filter_is_valid, "ODATA过滤语句不符合预期"

    # 解绑监听器,避免影响其他测试用例
    page.remove_listener("request", handle_target_request)

方案2:改用page.expect_request()主动等待请求(推荐)

这种方式更贴合Playwright的同步设计逻辑,直接主动等待目标请求出现,拿到请求对象后同步完成验证,无需依赖后台线程:

import pytest
import re
from playwright.sync_api import Page

def test_odata_query_validation(page: Page):
    filter_pattern = re.compile(r'your_odata_regex_here')
    odata_filter_is_valid = False

    # 包裹触发操作,同时等待目标请求
    with page.expect_request("**/your-target-api-endpoint**") as request_context:
        # 执行触发API请求的操作
        page.fill("#search-input", "test-keyword")
        page.click("#submit-search-btn")
    
    # 获取请求对象,主流程会阻塞到请求出现
    target_request = request_context.value
    odata_filter = target_request.headers.get("OData-Filter")
    
    if odata_filter:
        matched_items = filter_pattern.findall(odata_filter)
        odata_filter_is_valid = len(matched_items) == 1  # 示例验证条件

    # 后续验证逻辑
    assert odata_filter_is_valid, "ODATA过滤语句验证失败"

为什么之前的方法无效?

  • page.wait_for_timeout()是固定时长等待,无法精准匹配处理完成时机,还会引入冗余等待;
  • 直接使用threading库但未做正确的同步控制(如未用Event/Semaphore),无法让主线程感知后台处理的完成状态。

内容的提问来源于stack exchange,提问作者Fangolin

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.07.29 04:35:41