💜route.fulfill()

route.fulfill() 是 Playwright 路由拦截功能中的另一个核心方法，它允许你完全替代原始请求的响应，而不是让请求继续到实际的服务器。这对于模拟API响应、测试错误场景、离线测试等非常有用。

基本概念

当使用 route.fulfill() 时，Playwright 不会发送实际的网络请求，而是直接返回你提供的模拟响应。这样可以：

• 完全控制响应内容
• 模拟各种服务器状态
• 避免依赖外部服务
• 加快测试执行速度

基础用法

1. 简单的响应替换

from playwright.sync_api import sync_playwright

with sync_playwright() as p:
    browser = p.chromium.launch(headless=False)
    context = browser.new_context()
    page = context.new_page()
    
    def handle_route(route, request):
        if "/api/user" in request.url:
            # 返回模拟的用户数据
            route.fulfill(
                status=200,
                headers={"Content-Type": "application/json"},
                body='{"id": 123, "name": "Test User", "email": "test@example.com"}'
            )
        else:
            route.continue_()
    
    page.route("**/*", handle_route)
    page.goto("https://example.com")
    
    browser.close()

2. 返回静态文件内容

def handle_route(route, request):
    if request.url.endswith('.css'):
        # 返回自定义CSS
        css_content = """
        body { background-color: #f0f0f0; }
        .test-element { color: red; }
        """
        route.fulfill(
            status=200,
            headers={"Content-Type": "text/css"},
            body=css_content
        )
    else:
        route.continue_()

完整的参数列表

route.fulfill() 支持以下参数：

route.fulfill(
    status=200,                              # HTTP状态码
    headers={"Content-Type": "application/json"},  # 响应头
    body="response content",                 # 响应体（字符串）
    body_bytes=b"binary content",           # 响应体（字节）
    path="/path/to/file.json",              # 从文件读取响应
    content_type="application/json",         # 内容类型（简化设置）
    response=original_response               # 基于现有响应修改
)

高级用法

1. 模拟API响应

import json
from datetime import datetime

class APISimulator:
    def __init__(self):
        self.users = [
            {"id": 1, "name": "Alice", "email": "alice@example.com"},
            {"id": 2, "name": "Bob", "email": "bob@example.com"},
            {"id": 3, "name": "Charlie", "email": "charlie@example.com"}
        ]
        self.posts = [
            {"id": 1, "title": "First Post", "author_id": 1, "content": "Hello World!"},
            {"id": 2, "title": "Second Post", "author_id": 2, "content": "This is a test."}
        ]
    
    def handle_route(self, route, request):
        url = request.url
        method = request.method
        
        # 处理用户相关API
        if "/api/users" in url:
            if method == "GET":
                if url.endswith("/users"):
                    # 返回所有用户
                    route.fulfill(
                        status=200,
                        headers={"Content-Type": "application/json"},
                        body=json.dumps(self.users)
                    )
                else:
                    # 返回特定用户
                    user_id = int(url.split("/")[-1])
                    user = next((u for u in self.users if u["id"] == user_id), None)
                    if user:
                        route.fulfill(
                            status=200,
                            headers={"Content-Type": "application/json"},
                            body=json.dumps(user)
                        )
                    else:
                        route.fulfill(
                            status=404,
                            headers={"Content-Type": "application/json"},
                            body='{"error": "User not found"}'
                        )
            
            elif method == "POST":
                # 创建新用户
                try:
                    new_user_data = json.loads(request.post_data or "{}")
                    new_user = {
                        "id": len(self.users) + 1,
                        "name": new_user_data.get("name"),
                        "email": new_user_data.get("email")
                    }
                    self.users.append(new_user)
                    
                    route.fulfill(
                        status=201,
                        headers={"Content-Type": "application/json"},
                        body=json.dumps(new_user)
                    )
                except json.JSONDecodeError:
                    route.fulfill(
                        status=400,
                        headers={"Content-Type": "application/json"},
                        body='{"error": "Invalid JSON"}'
                    )
        
        # 处理文章API
        elif "/api/posts" in url:
            route.fulfill(
                status=200,
                headers={"Content-Type": "application/json"},
                body=json.dumps(self.posts)
            )
        
        else:
            route.continue_()

# 使用示例
simulator = APISimulator()

with sync_playwright() as p:
    browser = p.chromium.launch(headless=False)
    context = browser.new_context()
    page = context.new_page()
    
    page.route("**/*", simulator.handle_route)
    
    # 测试页面现在可以使用模拟的API
    page.goto("https://your-app.com")
    
    browser.close()

2. 模拟不同的HTTP状态码

class HTTPStatusSimulator:
    def __init__(self):
        self.request_count = 0
    
    def handle_route(self, route, request):
        self.request_count += 1
        
        if "/api/flaky-endpoint" in request.url:
            # 模拟不稳定的服务
            if self.request_count % 3 == 0:
                # 每第3个请求失败
                route.fulfill(
                    status=500,
                    headers={"Content-Type": "application/json"},
                    body='{"error": "Internal Server Error", "code": "SERVER_ERROR"}'
                )
            elif self.request_count % 5 == 0:
                # 每第5个请求超时
                route.fulfill(
                    status=408,
                    headers={"Content-Type": "application/json"},
                    body='{"error": "Request Timeout", "code": "TIMEOUT"}'
                )
            else:
                # 正常响应
                route.fulfill(
                    status=200,
                    headers={"Content-Type": "application/json"},
                    body='{"status": "success", "data": "Everything is fine"}'
                )
        
        elif "/api/auth" in request.url:
            # 模拟认证失败
            auth_header = request.headers.get("Authorization", "")
            if not auth_header or "Bearer valid-token" not in auth_header:
                route.fulfill(
                    status=401,
                    headers={"Content-Type": "application/json"},
                    body='{"error": "Unauthorized", "code": "INVALID_TOKEN"}'
                )
            else:
                route.fulfill(
                    status=200,
                    headers={"Content-Type": "application/json"},
                    body='{"user": "authenticated_user", "role": "admin"}'
                )
        
        elif "/api/rate-limited" in request.url:
            # 模拟限流
            route.fulfill(
                status=429,
                headers={
                    "Content-Type": "application/json",
                    "Retry-After": "60"
                },
                body='{"error": "Too Many Requests", "retry_after": 60}'
            )
        
        else:
            route.continue_()

# 使用示例
status_simulator = HTTPStatusSimulator()
page.route("**/api/**", status_simulator.handle_route)

3. 基于文件的响应

import os
import mimetypes

class FileBasedMockServer:
    def __init__(self, mock_dir="./mock_responses"):
        self.mock_dir = mock_dir
        self.ensure_mock_dir()
        self.create_sample_files()
    
    def ensure_mock_dir(self):
        if not os.path.exists(self.mock_dir):
            os.makedirs(self.mock_dir)
    
    def create_sample_files(self):
        # 创建示例模拟文件
        samples = {
            "users.json": '{"users": [{"id": 1, "name": "John"}, {"id": 2, "name": "Jane"}]}',
            "products.json": '{"products": [{"id": 1, "name": "Laptop", "price": 999}]}',
            "error_500.json": '{"error": "Internal Server Error", "message": "Something went wrong"}',
            "styles.css": "body { font-family: Arial; background: #f0f0f0; }",
            "index.html": "<html><body><h1>Mock Response</h1></body></html>"
        }
        
        for filename, content in samples.items():
            filepath = os.path.join(self.mock_dir, filename)
            if not os.path.exists(filepath):
                with open(filepath, 'w', encoding='utf-8') as f:
                    f.write(content)
    
    def handle_route(self, route, request):
        url = request.url
        
        # 映射URL到文件
        url_to_file = {
            "/api/users": "users.json",
            "/api/products": "products.json",
            "/api/error-test": "error_500.json",
            "/styles.css": "styles.css",
            "/mock-page": "index.html"
        }
        
        # 查找匹配的文件
        for url_pattern, filename in url_to_file.items():
            if url_pattern in url:
                filepath = os.path.join(self.mock_dir, filename)
                
                if os.path.exists(filepath):
                    # 根据文件扩展名确定内容类型
                    content_type, _ = mimetypes.guess_type(filepath)
                    if not content_type:
                        content_type = "text/plain"
                    
                    # 特殊处理错误文件
                    status_code = 500 if "error" in filename else 200
                    
                    route.fulfill(
                        status=status_code,
                        headers={"Content-Type": content_type},
                        path=filepath
                    )
                    return
        
        # 如果没有匹配的文件，继续原始请求
        route.continue_()

# 使用示例
mock_server = FileBasedMockServer()
page.route("**/*", mock_server.handle_route)

4. 动态响应生成

import random
import time
from datetime import datetime

class DynamicResponseGenerator:
    def __init__(self):
        self.session_data = {}
    
    def generate_user_data(self):
        names = ["Alice", "Bob", "Charlie", "Diana", "Eve"]
        domains = ["example.com", "test.com", "demo.com"]
        
        return {
            "id": random.randint(1, 1000),
            "name": random.choice(names),
            "email": f"{random.choice(names).lower()}@{random.choice(domains)}",
            "created_at": datetime.now().isoformat(),
            "last_active": datetime.now().isoformat(),
            "score": random.randint(0, 100)
        }
    
    def generate_analytics_data(self):
        return {
            "timestamp": datetime.now().isoformat(),
            "page_views": random.randint(100, 1000),
            "unique_visitors": random.randint(50, 500),
            "bounce_rate": round(random.uniform(0.2, 0.8), 2),
            "conversion_rate": round(random.uniform(0.01, 0.1), 3),
            "performance": {
                "load_time": round(random.uniform(0.5, 3.0), 2),
                "render_time": round(random.uniform(0.1, 1.0), 2)
            }
        }
    
    def handle_route(self, route, request):
        url = request.url
        
        if "/api/random-user" in url:
            # 生成随机用户数据
            user_data = self.generate_user_data()
            route.fulfill(
                status=200,
                headers={
                    "Content-Type": "application/json",
                    "X-Generated-At": datetime.now().isoformat()
                },
                body=json.dumps(user_data)
            )
        
        elif "/api/analytics" in url:
            # 模拟实时分析数据
            analytics = self.generate_analytics_data()
            route.fulfill(
                status=200,
                headers={
                    "Content-Type": "application/json",
                    "Cache-Control": "no-cache"
                },
                body=json.dumps(analytics)
            )
        
        elif "/api/slow-response" in url:
            # 模拟慢响应
            time.sleep(2)  # 注意：这会阻塞，实际使用中考虑异步
            route.fulfill(
                status=200,
                headers={"Content-Type": "application/json"},
                body='{"message": "This was a slow response", "delay": "2 seconds"}'
            )
        
        elif "/api/conditional" in url:
            # 基于请求参数的条件响应
            if "error=true" in url:
                route.fulfill(
                    status=400,
                    headers={"Content-Type": "application/json"},
                    body='{"error": "Simulated error condition"}'
                )
            elif "empty=true" in url:
                route.fulfill(
                    status=200,
                    headers={"Content-Type": "application/json"},
                    body='{"data": [], "total": 0}'
                )
            else:
                route.fulfill(
                    status=200,
                    headers={"Content-Type": "application/json"},
                    body='{"data": ["item1", "item2", "item3"], "total": 3}'
                )
        
        else:
            route.continue_()

# 使用示例
generator = DynamicResponseGenerator()
page.route("**/api/**", generator.handle_route)

5. 响应修改和增强

class ResponseEnhancer:
    def __init__(self):
        self.request_id_counter = 0
    
    def handle_route(self, route, request):
        self.request_id_counter += 1
        
        if "/api/" in request.url:
            # 对所有API请求添加通用的响应头和数据
            
            # 首先获取原始响应（如果需要的话）
            # 注意：这里演示的是完全替换响应的方式
            
            if "/api/enhanced" in request.url:
                # 创建增强的响应
                enhanced_response = {
                    "meta": {
                        "request_id": f"req_{self.request_id_counter}",
                        "timestamp": datetime.now().isoformat(),
                        "version": "1.0",
                        "server": "mock-server"
                    },
                    "data": {
                        "message": "This is an enhanced response",
                        "features": ["caching", "versioning", "tracking"]
                    },
                    "links": {
                        "self": request.url,
                        "documentation": "https://api-docs.example.com"
                    }
                }
                
                route.fulfill(
                    status=200,
                    headers={
                        "Content-Type": "application/json",
                        "X-Request-ID": f"req_{self.request_id_counter}",
                        "X-API-Version": "1.0",
                        "X-Rate-Limit-Remaining": "100",
                        "Cache-Control": "max-age=300"
                    },
                    body=json.dumps(enhanced_response, indent=2)
                )
            else:
                route.continue_()
        else:
            route.continue_()

# 使用示例
enhancer = ResponseEnhancer()
page.route("**/*", enhancer.handle_route)

6. 图片和二进制文件处理

import base64

class BinaryResponseHandler:
    def __init__(self):
        # 1x1像素的透明PNG图片的base64编码
        self.tiny_png = base64.b64decode(
            'iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8/5+hHgAHggJ/PchI7wAAAABJRU5ErkJggg=='
        )
    
    def handle_route(self, route, request):
        url = request.url
        
        if any(ext in url for ext in ['.png', '.jpg', '.jpeg', '.gif']):
            # 返回占位图片
            route.fulfill(
                status=200,
                headers={"Content-Type": "image/png"},
                body_bytes=self.tiny_png
            )
        
        elif url.endswith('.pdf'):
            # 返回简单的PDF内容
            pdf_content = b"%PDF-1.4\n1 0 obj\n<<\n/Type /Catalog\n/Pages 2 0 R\n>>\nendobj\n2 0 obj\n<<\n/Type /Pages\n/Kids [3 0 R]\n/Count 1\n>>\nendobj\n3 0 obj\n<<\n/Type /Page\n/Parent 2 0 R\n/MediaBox [0 0 612 792]\n>>\nendobj\nxref\n0 4\n0000000000 65535 f \n0000000009 00000 n \n0000000074 00000 n \n0000000120 00000 n \ntrailer\n<<\n/Size 4\n/Root 1 0 R\n>>\nstartxref\n174\n%%EOF"
            
            route.fulfill(
                status=200,
                headers={
                    "Content-Type": "application/pdf",
                    "Content-Disposition": "inline; filename=test.pdf"
                },
                body_bytes=pdf_content
            )
        
        elif url.endswith('.zip'):
            # 返回空的ZIP文件
            empty_zip = b'PK\x05\x06\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
            
            route.fulfill(
                status=200,
                headers={
                    "Content-Type": "application/zip",
                    "Content-Disposition": "attachment; filename=empty.zip"
                },
                body_bytes=empty_zip
            )
        
        else:
            route.continue_()

# 使用示例
binary_handler = BinaryResponseHandler()
page.route("**/*", binary_handler.handle_route)

错误处理和调试

class SafeRouteHandler:
    def __init__(self):
        self.error_count = 0
    
    def handle_route(self, route, request):
        try:
            if "/api/test" in request.url:
                # 模拟可能失败的操作
                data = self._process_request(request)
                
                route.fulfill(
                    status=200,
                    headers={"Content-Type": "application/json"},
                    body=json.dumps(data)
                )
            else:
                route.continue_()
                
        except Exception as e:
            self.error_count += 1
            print(f"Route处理错误 #{self.error_count}: {e}")
            
            # 返回错误响应而不是让整个测试失败
            route.fulfill(
                status=500,
                headers={"Content-Type": "application/json"},
                body=json.dumps({
                    "error": "Mock server error",
                    "message": str(e),
                    "error_count": self.error_count
                })
            )
    
    def _process_request(self, request):
        # 模拟一些可能出错的处理
        if "trigger_error" in request.url:
            raise ValueError("Simulated processing error")
        
        return {"status": "success", "processed_at": datetime.now().isoformat()}

# 使用示例
safe_handler = SafeRouteHandler()
page.route("**/*", safe_handler.handle_route)

性能优化

import hashlib
from functools import lru_cache

class OptimizedMockServer:
    def __init__(self):
        self.response_cache = {}
    
    @lru_cache(maxsize=100)
    def _generate_cached_response(self, cache_key):
        """使用LRU缓存来避免重复生成相同的响应"""
        return {
            "timestamp": datetime.now().isoformat(),
            "data": f"Cached response for {cache_key}",
            "cache_hit": True
        }
    
    def _get_cache_key(self, request):
        """为请求生成缓存键"""
        key_parts = [
            request.method,
            request.url,
            request.post_data or ""
        ]
        return hashlib.md5("|".join(key_parts).encode()).hexdigest()
    
    def handle_route(self, route, request):
        if "/api/cached" in request.url:
            cache_key = self._get_cache_key(request)
            
            # 检查内存缓存
            if cache_key in self.response_cache:
                cached_response = self.response_cache[cache_key]
                route.fulfill(
                    status=200,
                    headers={
                        "Content-Type": "application/json",
                        "X-Cache": "HIT"
                    },
                    body=json.dumps(cached_response)
                )
                return
            
            # 生成新响应并缓存
            response_data = self._generate_cached_response(cache_key)
            self.response_cache[cache_key] = response_data
            
            route.fulfill(
                status=200,
                headers={
                    "Content-Type": "application/json",
                    "X-Cache": "MISS"
                },
                body=json.dumps(response_data)
            )
        else:
            route.continue_()

# 使用示例
optimized_server = OptimizedMockServer()
page.route("**/api/**", optimized_server.handle_route)

实际应用场景

1. 完整的测试环境模拟

class TestEnvironmentSimulator:
    def __init__(self, config_file="test_config.json"):
        self.config = self._load_config(config_file)
        self.setup_test_data()
    
    def _load_config(self, config_file):
        # 加载测试配置
        default_config = {
            "api_delay": 0,
            "error_rate": 0,
            "features": {
                "user_management": True,
                "analytics": True,
                "notifications": False
            }
        }
        return default_config
    
    def setup_test_data(self):
        # 设置测试数据
        pass
    
    def handle_route(self, route, request):
        # 根据配置模拟不同的环境行为
        url = request.url
        
        # 添加延迟（如果配置了）
        if self.config["api_delay"] > 0:
            time.sleep(self.config["api_delay"] / 1000.0)
        
        # 模拟错误（根据错误率）
        if random.random() < self.config["error_rate"]:
            route.fulfill(
                status=500,
                headers={"Content-Type": "application/json"},
                body='{"error": "Simulated random error"}'
            )
            return
        
        # 根据功能开关返回不同的响应
        if "/api/features" in url:
            route.fulfill(
                status=200,
                headers={"Content-Type": "application/json"},
                body=json.dumps(self.config["features"])
            )
        else:
            route.continue_()

# 使用示例
test_env = TestEnvironmentSimulator()
page.route("**/*", test_env.handle_route)

注意事项

1. 响应完整性：使用 fulfill() 时必须提供完整的响应，包括状态码和适当的头部
2. 内容类型：确保设置正确的 Content-Type 头部
3. 字符编码：处理非ASCII字符时注意编码问题
4. 内存使用：大量使用缓存时注意内存消耗
5. 测试隔离：确保不同测试之间的模拟数据不会相互影响
6. 性能影响：复杂的响应生成逻辑可能影响测试性能

route.fulfill() 是一个非常强大的功能，通过合理使用可以创建完全可控的测试环境，提高测试的可靠性和独立性。