test_interception.py

# Copyright (c) Microsoft Corporation.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import asyncio
import json

import pytest

from playwright.async_api import Browser, BrowserContext, Error, Page, Playwright, Route
from tests.server import Server


async def test_page_route_should_intercept(page, server):
    intercepted = []

    async def handle_request(route, request):
        assert route.request == request
        assert "empty.html" in request.url
        assert request.headers["user-agent"]
        assert request.method == "GET"
        assert request.post_data is None
        assert request.is_navigation_request()
        assert request.resource_type == "document"
        assert request.frame == page.main_frame
        assert request.frame.url == "about:blank"
        await route.continue_()
        intercepted.append(True)

    await page.route("**/empty.html", handle_request)

    response = await page.goto(server.EMPTY_PAGE)
    assert response.ok
    assert len(intercepted) == 1


async def test_page_route_should_unroute(page: Page, server):
    intercepted = []

    await page.route(
        "**/*",
        lambda route: (
            intercepted.append(1),
            asyncio.create_task(route.continue_()),
        ),
    )

    await page.route(
        "**/empty.html",
        lambda route: (
            intercepted.append(2),
            asyncio.create_task(route.continue_()),
        ),
    )

    await page.route(
        "**/empty.html",
        lambda route: (
            intercepted.append(3),
            asyncio.create_task(route.continue_()),
        ),
    )

    def handler4(route):
        intercepted.append(4)
        asyncio.create_task(route.continue_())

    await page.route("**/empty.html", handler4)

    await page.goto(server.EMPTY_PAGE)
    assert intercepted == [4]

    intercepted = []
    await page.unroute("**/empty.html", handler4)
    await page.goto(server.EMPTY_PAGE)
    assert intercepted == [3]

    intercepted = []
    await page.unroute("**/empty.html")
    await page.goto(server.EMPTY_PAGE)
    assert intercepted == [1]


async def test_page_route_should_work_when_POST_is_redirected_with_302(page, server):
    server.set_redirect("/rredirect", "/empty.html")
    await page.goto(server.EMPTY_PAGE)
    await page.route("**/*", lambda route: route.continue_())
    await page.set_content(
        """
      <form action='/rredirect' method='post'>
        <input type="hidden" id="foo" name="foo" value="FOOBAR">
      </form>
    """
    )
    async with page.expect_navigation():
        await page.eval_on_selector("form", "form => form.submit()"),


# @see https://github.com/GoogleChrome/puppeteer/issues/3973
async def test_page_route_should_work_when_header_manipulation_headers_with_redirect(
    page, server
):
    server.set_redirect("/rrredirect", "/empty.html")
    await page.route(
        "**/*",
        lambda route: route.continue_(headers={**route.request.headers, "foo": "bar"}),
    )

    await page.goto(server.PREFIX + "/rrredirect")


# @see https://github.com/GoogleChrome/puppeteer/issues/4743
async def test_page_route_should_be_able_to_remove_headers(page, server):
    async def handle_request(route):
        headers = route.request.headers
        if "origin" in headers:
            del headers["origin"]
        await route.continue_(headers=headers)

    await page.route(
        "**/*",  # remove "origin" header
        handle_request,
    )

    [serverRequest, _] = await asyncio.gather(
        server.wait_for_request("/empty.html"), page.goto(server.PREFIX + "/empty.html")
    )
    assert serverRequest.getHeader("origin") is None


async def test_page_route_should_contain_referer_header(page, server):
    requests = []
    await page.route(
        "**/*",
        lambda route: (
            requests.append(route.request),
            asyncio.create_task(route.continue_()),
        ),
    )

    await page.goto(server.PREFIX + "/one-style.html")
    assert "/one-style.css" in requests[1].url
    assert "/one-style.html" in requests[1].headers["referer"]


async def test_page_route_should_properly_return_navigation_response_when_URL_has_cookies(
    context, page, server
):
    # Setup cookie.
    await page.goto(server.EMPTY_PAGE)
    await context.add_cookies(
        [{"url": server.EMPTY_PAGE, "name": "foo", "value": "bar"}]
    )

    # Setup request interception.
    await page.route("**/*", lambda route: route.continue_())
    response = await page.reload()
    assert response.status == 200


async def test_page_route_should_show_custom_HTTP_headers(page, server):
    await page.set_extra_http_headers({"foo": "bar"})

    def assert_headers(request):
        assert request.headers["foo"] == "bar"

    await page.route(
        "**/*",
        lambda route: (
            assert_headers(route.request),
            asyncio.create_task(route.continue_()),
        ),
    )

    response = await page.goto(server.EMPTY_PAGE)
    assert response.ok


# @see https://github.com/GoogleChrome/puppeteer/issues/4337
async def test_page_route_should_work_with_redirect_inside_sync_XHR(page, server):
    await page.goto(server.EMPTY_PAGE)
    server.set_redirect("/logo.png", "/pptr.png")
    await page.route("**/*", lambda route: route.continue_())
    status = await page.evaluate(
        """async() => {
      const request = new XMLHttpRequest();
      request.open('GET', '/logo.png', false);  // `false` makes the request synchronous
      request.send(null);
      return request.status;
    }"""
    )

    assert status == 200


async def test_page_route_should_work_with_custom_referer_headers(page, server):
    await page.set_extra_http_headers({"referer": server.EMPTY_PAGE})

    def assert_headers(route):
        assert route.request.headers["referer"] == server.EMPTY_PAGE

    await page.route(
        "**/*",
        lambda route: (
            assert_headers(route),
            asyncio.create_task(route.continue_()),
        ),
    )

    response = await page.goto(server.EMPTY_PAGE)
    assert response.ok


async def test_page_route_should_be_abortable(page, server):
    await page.route(r"/\.css$/", lambda route: asyncio.create_task(route.abort()))
    failed = []

    def handle_request(request):
        if request.url.includes(".css"):
            failed.append(True)

    page.on("requestfailed", handle_request)

    response = await page.goto(server.PREFIX + "/one-style.html")
    assert response.ok
    assert response.request.failure is None
    assert len(failed) == 0


async def test_page_route_should_be_abortable_with_custom_error_codes(
    page: Page, server, is_webkit, is_firefox
):
    await page.route(
        "**/*",
        lambda route: route.abort("internetdisconnected"),
    )
    failed_requests = []
    page.on("requestfailed", lambda request: failed_requests.append(request))
    with pytest.raises(Error):
        await page.goto(server.EMPTY_PAGE)
    assert len(failed_requests) == 1
    failed_request = failed_requests[0]
    if is_webkit:
        assert failed_request.failure == "Request intercepted"
    elif is_firefox:
        assert failed_request.failure == "NS_ERROR_OFFLINE"
    else:
        assert failed_request.failure == "net::ERR_INTERNET_DISCONNECTED"


async def test_page_route_should_send_referer(page, server):
    await page.set_extra_http_headers({"referer": "http://google.com/"})

    await page.route("**/*", lambda route: route.continue_())
    [request, _] = await asyncio.gather(
        server.wait_for_request("/grid.html"),
        page.goto(server.PREFIX + "/grid.html"),
    )
    assert request.getHeader("referer") == "http://google.com/"


async def test_page_route_should_fail_navigation_when_aborting_main_resource(
    page, server, is_webkit, is_firefox
):
    await page.route("**/*", lambda route: route.abort())
    with pytest.raises(Error) as exc:
        await page.goto(server.EMPTY_PAGE)
    assert exc
    if is_webkit:
        assert "Request intercepted" in exc.value.message
    elif is_firefox:
        assert "NS_ERROR_FAILURE" in exc.value.message
    else:
        assert "net::ERR_FAILED" in exc.value.message


async def test_page_route_should_not_work_with_redirects(page, server):
    intercepted = []
    await page.route(
        "**/*",
        lambda route: (
            asyncio.create_task(route.continue_()),
            intercepted.append(route.request),
        ),
    )

    server.set_redirect("/non-existing-page.html", "/non-existing-page-2.html")
    server.set_redirect("/non-existing-page-2.html", "/non-existing-page-3.html")
    server.set_redirect("/non-existing-page-3.html", "/non-existing-page-4.html")
    server.set_redirect("/non-existing-page-4.html", "/empty.html")

    response = await page.goto(server.PREFIX + "/non-existing-page.html")
    assert response.status == 200
    assert "empty.html" in response.url

    assert len(intercepted) == 1
    assert intercepted[0].resource_type == "document"
    assert intercepted[0].is_navigation_request()
    assert "/non-existing-page.html" in intercepted[0].url

    chain = []
    r = response.request
    while r:
        chain.append(r)
        assert r.is_navigation_request()
        r = r.redirected_from

    assert len(chain) == 5
    assert "/empty.html" in chain[0].url
    assert "/non-existing-page-4.html" in chain[1].url
    assert "/non-existing-page-3.html" in chain[2].url
    assert "/non-existing-page-2.html" in chain[3].url
    assert "/non-existing-page.html" in chain[4].url
    for idx, _ in enumerate(chain):
        assert chain[idx].redirected_to == (chain[idx - 1] if idx > 0 else None)


async def test_page_route_should_work_with_redirects_for_subresources(page, server):
    intercepted = []
    await page.route(
        "**/*",
        lambda route: (
            asyncio.create_task(route.continue_()),
            intercepted.append(route.request),
        ),
    )

    server.set_redirect("/one-style.css", "/two-style.css")
    server.set_redirect("/two-style.css", "/three-style.css")
    server.set_redirect("/three-style.css", "/four-style.css")
    server.set_route(
        "/four-style.css",
        lambda req: (req.write(b"body {box-sizing: border-box; }"), req.finish()),
    )

    response = await page.goto(server.PREFIX + "/one-style.html")
    assert response.status == 200
    assert "one-style.html" in response.url

    # TODO: https://github.com/microsoft/playwright/issues/12789
    assert len(intercepted) >= 2
    assert intercepted[0].resource_type == "document"
    assert "one-style.html" in intercepted[0].url

    r = intercepted[1]
    for url in [
        "/one-style.css",
        "/two-style.css",
        "/three-style.css",
        "/four-style.css",
    ]:
        assert r.resource_type == "stylesheet"
        assert url in r.url
        r = r.redirected_to
    assert r is None


async def test_page_route_should_work_with_equal_requests(page, server):
    await page.goto(server.EMPTY_PAGE)
    hits = [True]

    def handle_request(request, hits):
        request.write(str(len(hits) * 11).encode())
        request.finish()
        hits.append(True)

    server.set_route("/zzz", lambda r: handle_request(r, hits))

    spinner = []

    async def handle_route(route):
        if len(spinner) == 1:
            await route.abort()
            spinner.pop(0)
        else:
            await route.continue_()
            spinner.append(True)

    # Cancel 2nd request.
    await page.route("**/*", handle_route)

    results = []
    for idx in range(3):
        results.append(
            await page.evaluate(
                """() => fetch('/zzz').then(response => response.text()).catch(e => 'FAILED')"""
            )
        )
    assert results == ["11", "FAILED", "22"]


async def test_page_route_should_navigate_to_dataURL_and_not_fire_dataURL_requests(
    page, server
):
    requests = []
    await page.route(
        "**/*",
        lambda route: (
            requests.append(route.request),
            asyncio.create_task(route.continue_()),
        ),
    )

    data_URL = "data:text/html,<div>yo</div>"
    response = await page.goto(data_URL)
    assert response is None
    assert len(requests) == 0


async def test_page_route_should_be_able_to_fetch_dataURL_and_not_fire_dataURL_requests(
    page, server
):
    await page.goto(server.EMPTY_PAGE)
    requests = []
    await page.route(
        "**/*",
        lambda route: (
            requests.append(route.request),
            asyncio.create_task(route.continue_()),
        ),
    )

    data_URL = "data:text/html,<div>yo</div>"
    text = await page.evaluate("url => fetch(url).then(r => r.text())", data_URL)
    assert text == "<div>yo</div>"
    assert len(requests) == 0


async def test_page_route_should_navigate_to_URL_with_hash_and_and_fire_requests_without_hash(
    page, server
):
    requests = []
    await page.route(
        "**/*",
        lambda route: (
            requests.append(route.request),
            asyncio.create_task(route.continue_()),
        ),
    )

    response = await page.goto(server.EMPTY_PAGE + "#hash")
    assert response.status == 200
    assert response.url == server.EMPTY_PAGE
    assert len(requests) == 1
    assert requests[0].url == server.EMPTY_PAGE


async def test_page_route_should_work_with_encoded_server(page, server):
    # The requestWillBeSent will report encoded URL, whereas interception will
    # report URL as-is. @see crbug.com/759388
    await page.route("**/*", lambda route: route.continue_())
    response = await page.goto(server.PREFIX + "/some nonexisting page")
    assert response.status == 404


async def test_page_route_should_work_with_encoded_server___2(page, server):
    # The requestWillBeSent will report URL as-is, whereas interception will
    # report encoded URL for stylesheet. @see crbug.com/759388
    requests = []
    await page.route(
        "**/*",
        lambda route: (
            asyncio.create_task(route.continue_()),
            requests.append(route.request),
        ),
    )

    response = await page.goto(
        f"""data:text/html,<link rel="stylesheet" href="{server.PREFIX}/fonts?helvetica|arial"/>"""
    )
    assert response is None
    # TODO: https://github.com/microsoft/playwright/issues/12789
    assert len(requests) >= 1
    assert (await requests[0].response()).status == 404


async def test_page_route_should_not_throw_Invalid_Interception_Id_if_the_request_was_cancelled(
    page, server
):
    await page.set_content("<iframe></iframe>")
    route_future = asyncio.Future()
    await page.route("**/*", lambda r, _: route_future.set_result(r))

    async with page.expect_request("**/*"):
        await page.eval_on_selector(
            "iframe", """(frame, url) => frame.src = url""", server.EMPTY_PAGE
        )
    # Delete frame to cause request to be canceled.
    await page.eval_on_selector("iframe", "frame => frame.remove()")
    route = await route_future
    await route.continue_()


async def test_page_route_should_intercept_main_resource_during_cross_process_navigation(
    page, server
):
    await page.goto(server.EMPTY_PAGE)
    intercepted = []
    await page.route(
        server.CROSS_PROCESS_PREFIX + "/empty.html",
        lambda route: (
            intercepted.append(True),
            asyncio.create_task(route.continue_()),
        ),
    )

    response = await page.goto(server.CROSS_PROCESS_PREFIX + "/empty.html")
    assert response.ok
    assert len(intercepted) == 1


@pytest.mark.skip_browser("webkit")
async def test_page_route_should_create_a_redirect(page, server):
    await page.goto(server.PREFIX + "/empty.html")

    async def handle_route(route, request):
        if request.url != (server.PREFIX + "/redirect_this"):
            return await route.continue_()
        await route.fulfill(status=301, headers={"location": "/empty.html"})

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

    text = await page.evaluate(
        """async url => {
      const data = await fetch(url);
      return data.text();
    }""",
        server.PREFIX + "/redirect_this",
    )
    assert text == ""


async def test_page_route_should_support_cors_with_GET(page, server, browser_name):
    await page.goto(server.EMPTY_PAGE)

    async def handle_route(route, request):
        headers = (
            {"access-control-allow-origin": "*"}
            if request.url.endswith("allow")
            else {}
        )
        await route.fulfill(
            content_type="application/json",
            headers=headers,
            status=200,
            body=json.dumps(["electric", "gas"]),
        )

    await page.route(
        "**/cars*",
        handle_route,
    )
    # Should succeed
    resp = await page.evaluate(
        """async () => {
        const response = await fetch('https://example.com/cars?allow', { mode: 'cors' });
        return response.json();
      }"""
    )

    assert resp == ["electric", "gas"]

    # Should be rejected
    with pytest.raises(Error) as exc:
        await page.evaluate(
            """async () => {
            const response = await fetch('https://example.com/cars?reject', { mode: 'cors' });
            return response.json();
        }"""
        )
    if browser_name == "chromium":
        assert "Failed" in exc.value.message
    elif browser_name == "webkit":
        assert "TypeError" in exc.value.message
    elif browser_name == "firefox":
        assert "NetworkError" in exc.value.message


async def test_page_route_should_support_cors_with_POST(page, server):
    await page.goto(server.EMPTY_PAGE)
    await page.route(
        "**/cars",
        lambda route: route.fulfill(
            content_type="application/json",
            headers={"Access-Control-Allow-Origin": "*"},
            status=200,
            body=json.dumps(["electric", "gas"]),
        ),
    )

    resp = await page.evaluate(
        """async () => {
      const response = await fetch('https://example.com/cars', {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        mode: 'cors',
        body: JSON.stringify({ 'number': 1 })
      });
      return response.json();
    }"""
    )

    assert resp == ["electric", "gas"]


async def test_page_route_should_support_cors_for_different_methods(page, server):
    await page.goto(server.EMPTY_PAGE)
    await page.route(
        "**/cars",
        lambda route, request: route.fulfill(
            content_type="application/json",
            headers={"Access-Control-Allow-Origin": "*"},
            status=200,
            body=json.dumps([request.method, "electric", "gas"]),
        ),
    )

    # First POST
    resp = await page.evaluate(
        """async () => {
        const response = await fetch('https://example.com/cars', {
          method: 'POST',
          headers: { 'Content-Type': 'application/json' },
          mode: 'cors',
          body: JSON.stringify({ 'number': 1 })
        });
        return response.json();
      }"""
    )

    assert resp == ["POST", "electric", "gas"]
    # Then DELETE
    resp = await page.evaluate(
        """async () => {
        const response = await fetch('https://example.com/cars', {
          method: 'DELETE',
          headers: {},
          mode: 'cors',
          body: ''
        });
        return response.json();
      }"""
    )

    assert resp == ["DELETE", "electric", "gas"]


async def test_request_fulfill_should_work_a(page, server):
    await page.route(
        "**/*",
        lambda route: route.fulfill(
            status=201,
            headers={"foo": "bar"},
            content_type="text/html",
            body="Yo, page!",
        ),
    )

    response = await page.goto(server.EMPTY_PAGE)
    assert response.status == 201
    assert response.headers["foo"] == "bar"
    assert await page.evaluate("() => document.body.textContent") == "Yo, page!"


async def test_request_fulfill_should_work_with_status_code_422(page, server):
    await page.route(
        "**/*",
        lambda route: route.fulfill(status=422, body="Yo, page!"),
    )

    response = await page.goto(server.EMPTY_PAGE)
    assert response.status == 422
    assert response.status_text == "Unprocessable Entity"
    assert await page.evaluate("() => document.body.textContent") == "Yo, page!"


async def test_request_fulfill_should_allow_mocking_binary_responses(
    page: Page, server, assert_to_be_golden, assetdir
):
    await page.route(
        "**/*",
        lambda route: route.fulfill(
            content_type="image/png",
            body=(assetdir / "pptr.png").read_bytes(),
        ),
    )

    await page.evaluate(
        """PREFIX => {
      const img = document.createElement('img');
      img.src = PREFIX + '/does-not-exist.png';
      document.body.appendChild(img);
      return new Promise(fulfill => img.onload = fulfill);
    }""",
        server.PREFIX,
    )
    img = await page.query_selector("img")
    assert img
    assert_to_be_golden(await img.screenshot(), "mock-binary-response.png")


async def test_request_fulfill_should_allow_mocking_svg_with_charset(
    page, server, assert_to_be_golden
):
    await page.route(
        "**/*",
        lambda route: route.fulfill(
            content_type="image/svg+xml ; charset=utf-8",
            body='<svg width="50" height="50" version="1.1" xmlns="http://www.w3.org/2000/svg"><rect x="10" y="10" width="30" height="30" stroke="black" fill="transparent" stroke-width="5"/></svg>',
        ),
    )

    await page.evaluate(
        """PREFIX => {
      const img = document.createElement('img');
      img.src = PREFIX + '/does-not-exist.svg';
      document.body.appendChild(img);
      return new Promise((f, r) => { img.onload = f; img.onerror = r; });
    }""",
        server.PREFIX,
    )
    img = await page.query_selector("img")
    assert_to_be_golden(await img.screenshot(), "mock-svg.png")


async def test_request_fulfill_should_work_with_file_path(
    page: Page, server, assert_to_be_golden, assetdir
):
    await page.route(
        "**/*",
        lambda route: route.fulfill(
            content_type="shouldBeIgnored", path=assetdir / "pptr.png"
        ),
    )
    await page.evaluate(
        """PREFIX => {
      const img = document.createElement('img');
      img.src = PREFIX + '/does-not-exist.png';
      document.body.appendChild(img);
      return new Promise(fulfill => img.onload = fulfill);
    }""",
        server.PREFIX,
    )
    img = await page.query_selector("img")
    assert img
    assert_to_be_golden(await img.screenshot(), "mock-binary-response.png")


async def test_request_fulfill_should_stringify_intercepted_request_response_headers(
    page, server
):
    await page.route(
        "**/*",
        lambda route: route.fulfill(
            status=200, headers={"foo": True}, body="Yo, page!"
        ),
    )

    response = await page.goto(server.EMPTY_PAGE)
    assert response.status == 200
    headers = response.headers
    assert headers["foo"] == "True"
    assert await page.evaluate("() => document.body.textContent") == "Yo, page!"


async def test_request_fulfill_should_not_modify_the_headers_sent_to_the_server(
    page, server
):
    await page.goto(server.PREFIX + "/empty.html")
    interceptedRequests = []

    # this is just to enable request interception, which disables caching in chromium
    await page.route(server.PREFIX + "/unused", lambda route, req: None)

    server.set_route(
        "/something",
        lambda response: (
            interceptedRequests.append(response),
            response.setHeader("Access-Control-Allow-Origin", "*"),
            response.write(b"done"),
            response.finish(),
        ),
    )

    text = await page.evaluate(
        """async url => {
      const data = await fetch(url);
      return data.text();
    }""",
        server.CROSS_PROCESS_PREFIX + "/something",
    )
    assert text == "done"

    playwrightRequest = asyncio.Future()
    await page.route(
        server.CROSS_PROCESS_PREFIX + "/something",
        lambda route, request: (
            playwrightRequest.set_result(request),
            asyncio.create_task(route.continue_(headers={**request.headers})),
        ),
    )

    textAfterRoute = await page.evaluate(
        """async url => {
      const data = await fetch(url);
      return data.text();
    }""",
        server.CROSS_PROCESS_PREFIX + "/something",
    )
    assert textAfterRoute == "done"

    assert len(interceptedRequests) == 2
    assert (
        interceptedRequests[0].requestHeaders == interceptedRequests[1].requestHeaders
    )


async def test_request_fulfill_should_include_the_origin_header(page, server):
    await page.goto(server.PREFIX + "/empty.html")
    interceptedRequest = []
    await page.route(
        server.CROSS_PROCESS_PREFIX + "/something",
        lambda route, request: (
            interceptedRequest.append(request),
            asyncio.create_task(
                route.fulfill(
                    headers={"Access-Control-Allow-Origin": "*"},
                    content_type="text/plain",
                    body="done",
                )
            ),
        ),
    )

    text = await page.evaluate(
        """async url => {
      const data = await fetch(url);
      return data.text();
    }""",
        server.CROSS_PROCESS_PREFIX + "/something",
    )
    assert text == "done"
    assert len(interceptedRequest) == 1
    assert interceptedRequest[0].headers["origin"] == server.PREFIX


async def test_request_fulfill_should_work_with_request_interception(page, server):
    requests = {}

    async def _handle_route(route: Route):
        requests[route.request.url.split("/").pop()] = route.request
        await route.continue_()

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

    server.set_redirect("/rrredirect", "/frames/one-frame.html")
    await page.goto(server.PREFIX + "/rrredirect")
    assert requests["rrredirect"].is_navigation_request()
    assert requests["frame.html"].is_navigation_request()
    assert requests["script.js"].is_navigation_request() is False
    assert requests["style.css"].is_navigation_request() is False


async def test_Interception_should_work_with_request_interception(
    browser: Browser, https_server
):
    context = await browser.new_context(ignore_https_errors=True)
    page = await context.new_page()

    await page.route("**/*", lambda route: asyncio.ensure_future(route.continue_()))
    response = await page.goto(https_server.EMPTY_PAGE)
    assert response
    assert response.status == 200
    await context.close()


async def test_ignore_http_errors_service_worker_should_intercept_after_a_service_worker(
    page, server
):
    await page.goto(server.PREFIX + "/serviceworkers/fetchdummy/sw.html")
    await page.evaluate("() => window.activationPromise")

    # Sanity check.
    sw_response = await page.evaluate('() => fetchDummy("foo")')
    assert sw_response == "responseFromServiceWorker:foo"

    def _handle_route(route):
        asyncio.ensure_future(
            route.fulfill(
                status=200,
                content_type="text/css",
                body="responseFromInterception:" + route.request.url.split("/")[-1],
            )
        )

    await page.route("**/foo", _handle_route)

    # Page route is applied after service worker fetch event.
    sw_response2 = await page.evaluate('() => fetchDummy("foo")')
    assert sw_response2 == "responseFromServiceWorker:foo"

    # Page route is not applied to service worker initiated fetch.
    non_intercepted_response = await page.evaluate('() => fetchDummy("passthrough")')
    assert non_intercepted_response == "FAILURE: Not Found"


async def test_page_route_should_support_times_parameter(page: Page, server: Server):
    intercepted = []

    async def handle_request(route):
        await route.continue_()
        intercepted.append(True)

    await page.route("**/empty.html", handle_request, times=1)

    await page.goto(server.EMPTY_PAGE)
    await page.goto(server.EMPTY_PAGE)
    await page.goto(server.EMPTY_PAGE)
    assert len(intercepted) == 1


async def test_context_route_should_support_times_parameter(
    context: BrowserContext, page: Page, server: Server
):
    intercepted = []

    async def handle_request(route):
        await route.continue_()
        intercepted.append(True)

    await context.route("**/empty.html", handle_request, times=1)

    await page.goto(server.EMPTY_PAGE)
    await page.goto(server.EMPTY_PAGE)
    await page.goto(server.EMPTY_PAGE)
    assert len(intercepted) == 1


async def test_should_fulfill_with_global_fetch_result(
    page: Page, playwright: Playwright, server: Server
) -> None:
    async def handle_request(route: Route) -> None:
        request = await playwright.request.new_context()
        response = await request.get(server.PREFIX + "/simple.json")
        await route.fulfill(response=response)
        await request.dispose()

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

    response = await page.goto(server.EMPTY_PAGE)
    assert response
    assert response.status == 200
    assert await response.json() == {"foo": "bar"}