python

关注公众号 jb51net

关闭
首页 > 脚本专栏 > python > Playwright网络流量监控

Playwright实现网络流量监控与修改指南

作者：蛋仔聊测试

Playwright 提供了强大的网络流量控制能力,可以拦截、修改和分析所有 HTTP/HTTPS 请求,下面我们就来看看如何使用Playwright监控网络流量吧

Playwright 提供了强大的网络流量控制能力，可以拦截、修改和分析所有 HTTP/HTTPS 请求。下面我将详细介绍各种使用方法和底层原理。

一、基本网络监控

1. 记录所有请求

from playwright.sync_api import sync_playwright

with sync_playwright() as p:
    browser = p.chromium.launch()
    page = browser.new_page()
    
    # 监听请求和响应事件
    def print_request(request):
        print(f">> Request: {request.method} {request.url}")
    
    def print_response(response):
        print(f"<< Response: {response.status} {response.url}")
    
    page.on("request", print_request)
    page.on("response", print_response)
    
    page.goto("https://example.com")
    
    browser.close()

2. 获取请求详细信息

def log_request(request):
    print(f"""
    Request: {request.method} {request.url}
    Headers: {request.headers}
    Post Data: {request.post_data}
    Resource Type: {request.resource_type}
    Navigation: {request.is_navigation_request()}
    """)

page.on("request", log_request)

二、网络请求修改

1. 修改请求头

async def modify_headers(route):
    headers = route.request.headers.copy()
    headers["x-custom-header"] = "my-value"
    await route.continue_(headers=headers)

await page.route("**/*", modify_headers)

2. 修改请求体

async def modify_post_data(route):
    if route.request.method == "POST":
        post_data = route.request.post_data
        modified_data = post_data.replace("old", "new")
        await route.continue_(post_data=modified_data)
    else:
        await route.continue_()

await page.route("**/api/**", modify_post_data)

3. 拦截并返回模拟响应

async def mock_response(route):
    await route.fulfill(
        status=200,
        content_type="application/json",
        body=json.dumps({"data": "mocked"})
    )

await page.route("**/api/data", mock_response)

三、高级网络控制

1. 延迟响应

async def delay_response(route):
    await asyncio.sleep(2)  # 延迟2秒
    await route.continue_()

await page.route("**/*.css", delay_response)

2. 阻止特定请求

async def block_analytics(route):
    if "google-analytics" in route.request.url:
        await route.abort()
    else:
        await route.continue_()

await page.route("**/*", block_analytics)

3. 修改响应

async def modify_response(route):
    response = await route.fetch()
    body = await response.text()
    modified_body = body.replace("Original", "Modified")
    await route.fulfill(
        response=response,
        body=modified_body
    )

await page.route("**/api/content", modify_response)

四、底层运行原理

1. 整体架构

+-------------------+     +-------------------+     +-------------------+
|   Playwright      |     |   Browser         |     |   Remote Server   |
|   Python Client   |<--->|   (Chromium/etc)  |<--->|   (example.com)   |
+-------------------+     +-------------------+     +-------------------+
        |                       ^    ^
        | CDP (Chrome DevTools) |    |
        | WebSocket Connection |    | Actual Network Traffic
        v                       |    |
+-------------------+           |    |
|   Network Proxy   |-----------+    |
|   Layer           |----------------+
+-------------------+

2. 请求生命周期

1.初始化阶段:

2.请求拦截流程:

sequenceDiagram
    participant Page
    participant Playwright
    participant Browser
    
    Page->>Browser: 发起请求
    Browser->>Playwright: 触发请求拦截 (Fetch.requestPaused)
    Playwright->>Python: 调用注册的路由处理器
    alt 继续请求
        Python->>Playwright: route.continue_()
        Playwright->>Browser: 继续请求到服务器
    else 模拟响应
        Python->>Playwright: route.fulfill()
        Playwright->>Browser: 返回模拟响应
    else 中止请求
        Python->>Playwright: route.abort()
        Playwright->>Browser: 中止请求
    end

3.修改请求流程:

Playwright 使用中间人技术拦截请求

可以修改的部分包括:

4.响应处理流程:

对于 route.fetch() 操作:

对于 route.fulfill() 操作:

3. 关键技术点

1.请求匹配系统:

2.内存管理:

3.安全机制:

4.性能优化:

五、实际应用示例

1. API 测试模拟

async def test_api(page):
    async def handle_api(route):
        if route.request.method == "POST":
            await route.fulfill(
                status=201,
                json={"id": 123, "status": "created"}
            )
        else:
            await route.continue_()
    
    await page.route("**/api/users*", handle_api)
    
    # 测试代码
    await page.goto("https://app.example.com")
    await page.click("#create-user")
    assert await page.text_content(".status") == "User 123 created"

2. 性能分析

async def analyze_performance(page):
    requests = []
    
    def log_request(request):
        requests.append({
            "url": request.url,
            "start": time.time(),
            "type": request.resource_type
        })
    
    def log_response(response):
        for req in requests:
            if req["url"] == response.url:
                req["duration"] = time.time() - req["start"]
                req["status"] = response.status
    
    page.on("request", log_request)
    page.on("response", log_response)
    
    await page.goto("https://example.com")
    
    # 输出性能报告
    slow_requests = [r for r in requests if r.get("duration", 0) > 1]
    print(f"Slow requests: {len(slow_requests)}/{len(requests)}")

3. 认证处理

async def handle_auth(page):
    await page.route("**/api/**", lambda route: route.continue_(
        headers={**route.request.headers, "Authorization": "Bearer xyz"}
    ))
    
    await page.goto("https://app.example.com")
    # 所有API请求会自动带上认证头

六、最佳实践

1.精确路由匹配:

2.清理路由:

# 添加路由
await page.route("**/api/**", handler)

# 测试完成后移除
await page.unroute("**/api/**", handler)

3.错误处理:

async def safe_handler(route):
    try:
        await route.continue_()
    except Exception as e:
        print(f"Failed to handle {route.request.url}: {e}")
        await route.abort()

4.性能敏感操作:

5.调试技巧:

# 打印未处理的请求
page.on("request", lambda r: print("Unhandled:", r.url))

Playwright 的网络拦截 API 提供了对浏览器网络活动的完全控制，使得测试复杂 web 应用变得更加容易，特别是在需要模拟各种网络条件或测试边缘情况时。理解其底层原理有助于更有效地使用这些功能。

到此这篇关于Playwright实现网络流量监控与修改指南的文章就介绍到这了,更多相关Playwright网络流量监控内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:
阅读全文