How to use test_ignore_http_errors_service_worker_should_intercept_after_a_service_worker method in Playwright Python

Best Python code snippet using playwright-python

Run Playwright Python automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

test_interception.py

Source: test_interception.py Github

copy
1# Copyright (c) Microsoft Corporation.
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7# http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
14
15import asyncio
16import json
17
18import pytest
19
20from playwright.async_api import Browser, Error, Page, Route
21
22
23async def test_page_route_should_intercept(page, server):
24    intercepted = []
25
26    async def handle_request(route, request):
27        assert route.request == request
28        assert "empty.html" in request.url
29        assert request.headers["user-agent"]
30        assert request.method == "GET"
31        assert request.post_data is None
32        assert request.is_navigation_request()
33        assert request.resource_type == "document"
34        assert request.frame == page.main_frame
35        assert request.frame.url == "about:blank"
36        await route.continue_()
37        intercepted.append(True)
38
39    await page.route("**/empty.html", handle_request)
40
41    response = await page.goto(server.EMPTY_PAGE)
42    assert response.ok
43    assert len(intercepted) == 1
44
45
46async def test_page_route_should_unroute(page: Page, server):
47    intercepted = []
48
49    def handler1(route):
50        intercepted.append(1)
51        asyncio.create_task(route.continue_())
52
53    await page.route("**/empty.html", handler1)
54    await page.route(
55        "**/empty.html",
56        lambda route: (
57            intercepted.append(2),  # type: ignore
58            asyncio.create_task(route.continue_()),
59        ),
60    )
61
62    await page.route(
63        "**/empty.html",
64        lambda route: (
65            intercepted.append(3),  # type: ignore
66            asyncio.create_task(route.continue_()),
67        ),
68    )
69
70    await page.route(
71        "**/*",
72        lambda route: (
73            intercepted.append(4),  # type: ignore
74            asyncio.create_task(route.continue_()),
75        ),
76    )
77
78    await page.goto(server.EMPTY_PAGE)
79    assert intercepted == [1]
80
81    intercepted = []
82    await page.unroute("**/empty.html", handler1)
83    await page.goto(server.EMPTY_PAGE)
84    assert intercepted == [2]
85
86    intercepted = []
87    await page.unroute("**/empty.html")
88    await page.goto(server.EMPTY_PAGE)
89    assert intercepted == [4]
90
91
92async def test_page_route_should_work_when_POST_is_redirected_with_302(page, server):
93    server.set_redirect("/rredirect", "/empty.html")
94    await page.goto(server.EMPTY_PAGE)
95    await page.route("**/*", lambda route: route.continue_())
96    await page.set_content(
97        """
98      <form action='/rredirect' method='post'>
99        <input type="hidden" id="foo" name="foo" value="FOOBAR">
100      </form>
101    """
102    )
103    async with page.expect_navigation():
104        await page.eval_on_selector("form", "form => form.submit()"),
105
106
107# @see https://github.com/GoogleChrome/puppeteer/issues/3973
108async def test_page_route_should_work_when_header_manipulation_headers_with_redirect(
109    page, server
110):
111    server.set_redirect("/rrredirect", "/empty.html")
112    await page.route(
113        "**/*",
114        lambda route: route.continue_(headers={**route.request.headers, "foo": "bar"}),
115    )
116
117    await page.goto(server.PREFIX + "/rrredirect")
118
119
120# @see https://github.com/GoogleChrome/puppeteer/issues/4743
121async def test_page_route_should_be_able_to_remove_headers(page, server):
122    async def handle_request(route):
123        headers = route.request.headers
124        if "origin" in headers:
125            del headers["origin"]
126        await route.continue_(headers=headers)
127
128    await page.route(
129        "**/*",  # remove "origin" header
130        handle_request,
131    )
132
133    [serverRequest, _] = await asyncio.gather(
134        server.wait_for_request("/empty.html"), page.goto(server.PREFIX + "/empty.html")
135    )
136    assert serverRequest.getHeader("origin") is None
137
138
139async def test_page_route_should_contain_referer_header(page, server):
140    requests = []
141    await page.route(
142        "**/*",
143        lambda route: (
144            requests.append(route.request),
145            asyncio.create_task(route.continue_()),
146        ),
147    )
148
149    await page.goto(server.PREFIX + "/one-style.html")
150    assert "/one-style.css" in requests[1].url
151    assert "/one-style.html" in requests[1].headers["referer"]
152
153
154async def test_page_route_should_properly_return_navigation_response_when_URL_has_cookies(
155    context, page, server
156):
157    # Setup cookie.
158    await page.goto(server.EMPTY_PAGE)
159    await context.add_cookies(
160        [{"url": server.EMPTY_PAGE, "name": "foo", "value": "bar"}]
161    )
162
163    # Setup request interception.
164    await page.route("**/*", lambda route: route.continue_())
165    response = await page.reload()
166    assert response.status == 200
167
168
169async def test_page_route_should_show_custom_HTTP_headers(page, server):
170    await page.set_extra_http_headers({"foo": "bar"})
171
172    def assert_headers(request):
173        assert request.headers["foo"] == "bar"
174
175    await page.route(
176        "**/*",
177        lambda route: (
178            assert_headers(route.request),
179            asyncio.create_task(route.continue_()),
180        ),
181    )
182
183    response = await page.goto(server.EMPTY_PAGE)
184    assert response.ok
185
186
187# @see https://github.com/GoogleChrome/puppeteer/issues/4337
188async def test_page_route_should_work_with_redirect_inside_sync_XHR(page, server):
189    await page.goto(server.EMPTY_PAGE)
190    server.set_redirect("/logo.png", "/pptr.png")
191    await page.route("**/*", lambda route: route.continue_())
192    status = await page.evaluate(
193        """async() => {
194      const request = new XMLHttpRequest();
195      request.open('GET', '/logo.png', false);  // `false` makes the request synchronous
196      request.send(null);
197      return request.status;
198    }"""
199    )
200
201    assert status == 200
202
203
204async def test_page_route_should_work_with_custom_referer_headers(page, server):
205    await page.set_extra_http_headers({"referer": server.EMPTY_PAGE})
206
207    def assert_headers(route):
208        assert route.request.headers["referer"] == server.EMPTY_PAGE
209
210    await page.route(
211        "**/*",
212        lambda route: (
213            assert_headers(route),
214            asyncio.create_task(route.continue_()),
215        ),
216    )
217
218    response = await page.goto(server.EMPTY_PAGE)
219    assert response.ok
220
221
222async def test_page_route_should_be_abortable(page, server):
223    await page.route(r"/\.css$/", lambda route: asyncio.create_task(route.abort()))
224    failed = []
225
226    def handle_request(request):
227        if request.url.includes(".css"):
228            failed.append(True)
229
230    page.on("requestfailed", handle_request)
231
232    response = await page.goto(server.PREFIX + "/one-style.html")
233    assert response.ok
234    assert response.request.failure is None
235    assert len(failed) == 0
236
237
238async def test_page_route_should_be_abortable_with_custom_error_codes(
239    page: Page, server, is_webkit, is_firefox
240):
241    await page.route(
242        "**/*",
243        lambda route: route.abort("internetdisconnected"),
244    )
245    failed_requests = []
246    page.on("requestfailed", lambda request: failed_requests.append(request))
247    with pytest.raises(Error):
248        await page.goto(server.EMPTY_PAGE)
249    assert len(failed_requests) == 1
250    failed_request = failed_requests[0]
251    if is_webkit:
252        assert failed_request.failure == "Request intercepted"
253    elif is_firefox:
254        assert failed_request.failure == "NS_ERROR_OFFLINE"
255    else:
256        assert failed_request.failure == "net::ERR_INTERNET_DISCONNECTED"
257
258
259async def test_page_route_should_send_referer(page, server):
260    await page.set_extra_http_headers({"referer": "http://google.com/"})
261
262    await page.route("**/*", lambda route: route.continue_())
263    [request, _] = await asyncio.gather(
264        server.wait_for_request("/grid.html"),
265        page.goto(server.PREFIX + "/grid.html"),
266    )
267    assert request.getHeader("referer") == "http://google.com/"
268
269
270async def test_page_route_should_fail_navigation_when_aborting_main_resource(
271    page, server, is_webkit, is_firefox
272):
273    await page.route("**/*", lambda route: route.abort())
274    with pytest.raises(Error) as exc:
275        await page.goto(server.EMPTY_PAGE)
276    assert exc
277    if is_webkit:
278        assert "Request intercepted" in exc.value.message
279    elif is_firefox:
280        assert "NS_ERROR_FAILURE" in exc.value.message
281    else:
282        assert "net::ERR_FAILED" in exc.value.message
283
284
285async def test_page_route_should_not_work_with_redirects(page, server):
286    intercepted = []
287    await page.route(
288        "**/*",
289        lambda route: (
290            asyncio.create_task(route.continue_()),
291            intercepted.append(route.request),
292        ),
293    )
294
295    server.set_redirect("/non-existing-page.html", "/non-existing-page-2.html")
296    server.set_redirect("/non-existing-page-2.html", "/non-existing-page-3.html")
297    server.set_redirect("/non-existing-page-3.html", "/non-existing-page-4.html")
298    server.set_redirect("/non-existing-page-4.html", "/empty.html")
299
300    response = await page.goto(server.PREFIX + "/non-existing-page.html")
301    assert response.status == 200
302    assert "empty.html" in response.url
303
304    assert len(intercepted) == 1
305    assert intercepted[0].resource_type == "document"
306    assert intercepted[0].is_navigation_request()
307    assert "/non-existing-page.html" in intercepted[0].url
308
309    chain = []
310    r = response.request
311    while r:
312        chain.append(r)
313        assert r.is_navigation_request()
314        r = r.redirected_from
315
316    assert len(chain) == 5
317    assert "/empty.html" in chain[0].url
318    assert "/non-existing-page-4.html" in chain[1].url
319    assert "/non-existing-page-3.html" in chain[2].url
320    assert "/non-existing-page-2.html" in chain[3].url
321    assert "/non-existing-page.html" in chain[4].url
322    for idx, _ in enumerate(chain):
323        assert chain[idx].redirected_to == (chain[idx - 1] if idx > 0 else None)
324
325
326async def test_page_route_should_work_with_redirects_for_subresources(page, server):
327    intercepted = []
328    await page.route(
329        "**/*",
330        lambda route: (
331            asyncio.create_task(route.continue_()),
332            intercepted.append(route.request),
333        ),
334    )
335
336    server.set_redirect("/one-style.css", "/two-style.css")
337    server.set_redirect("/two-style.css", "/three-style.css")
338    server.set_redirect("/three-style.css", "/four-style.css")
339    server.set_route(
340        "/four-style.css",
341        lambda req: (req.write(b"body {box-sizing: border-box; }"), req.finish()),
342    )
343
344    response = await page.goto(server.PREFIX + "/one-style.html")
345    assert response.status == 200
346    assert "one-style.html" in response.url
347
348    assert len(intercepted) == 2
349    assert intercepted[0].resource_type == "document"
350    assert "one-style.html" in intercepted[0].url
351
352    r = intercepted[1]
353    for url in [
354        "/one-style.css",
355        "/two-style.css",
356        "/three-style.css",
357        "/four-style.css",
358    ]:
359        assert r.resource_type == "stylesheet"
360        assert url in r.url
361        r = r.redirected_to
362    assert r is None
363
364
365async def test_page_route_should_work_with_equal_requests(page, server):
366    await page.goto(server.EMPTY_PAGE)
367    hits = [True]
368
369    def handle_request(request, hits):
370        request.write(str(len(hits) * 11).encode())
371        request.finish()
372        hits.append(True)
373
374    server.set_route("/zzz", lambda r: handle_request(r, hits))
375
376    spinner = []
377
378    async def handle_route(route):
379        if len(spinner) == 1:
380            await route.abort()
381            spinner.pop(0)
382        else:
383            await route.continue_()
384            spinner.append(True)
385
386    # Cancel 2nd request.
387    await page.route("**/*", handle_route)
388
389    results = []
390    for idx in range(3):
391        results.append(
392            await page.evaluate(
393                """() => fetch('/zzz').then(response => response.text()).catch(e => 'FAILED')"""
394            )
395        )
396    assert results == ["11", "FAILED", "22"]
397
398
399async def test_page_route_should_navigate_to_dataURL_and_not_fire_dataURL_requests(
400    page, server
401):
402    requests = []
403    await page.route(
404        "**/*",
405        lambda route: (
406            requests.append(route.request),
407            asyncio.create_task(route.continue_()),
408        ),
409    )
410
411    data_URL = "data:text/html,<div>yo</div>"
412    response = await page.goto(data_URL)
413    assert response is None
414    assert len(requests) == 0
415
416
417async def test_page_route_should_be_able_to_fetch_dataURL_and_not_fire_dataURL_requests(
418    page, server
419):
420    await page.goto(server.EMPTY_PAGE)
421    requests = []
422    await page.route(
423        "**/*",
424        lambda route: (
425            requests.append(route.request),
426            asyncio.create_task(route.continue_()),
427        ),
428    )
429
430    data_URL = "data:text/html,<div>yo</div>"
431    text = await page.evaluate("url => fetch(url).then(r => r.text())", data_URL)
432    assert text == "<div>yo</div>"
433    assert len(requests) == 0
434
435
436async def test_page_route_should_navigate_to_URL_with_hash_and_and_fire_requests_without_hash(
437    page, server
438):
439    requests = []
440    await page.route(
441        "**/*",
442        lambda route: (
443            requests.append(route.request),
444            asyncio.create_task(route.continue_()),
445        ),
446    )
447
448    response = await page.goto(server.EMPTY_PAGE + "#hash")
449    assert response.status == 200
450    assert response.url == server.EMPTY_PAGE
451    assert len(requests) == 1
452    assert requests[0].url == server.EMPTY_PAGE
453
454
455async def test_page_route_should_work_with_encoded_server(page, server):
456    # The requestWillBeSent will report encoded URL, whereas interception will
457    # report URL as-is. @see crbug.com/759388
458    await page.route("**/*", lambda route: route.continue_())
459    response = await page.goto(server.PREFIX + "/some nonexisting page")
460    assert response.status == 404
461
462
463async def test_page_route_should_work_with_encoded_server___2(page, server):
464    # The requestWillBeSent will report URL as-is, whereas interception will
465    # report encoded URL for stylesheet. @see crbug.com/759388
466    requests = []
467    await page.route(
468        "**/*",
469        lambda route: (
470            asyncio.create_task(route.continue_()),
471            requests.append(route.request),
472        ),
473    )
474
475    response = await page.goto(
476        f"""data:text/html,<link rel="stylesheet" href="{server.PREFIX}/fonts?helvetica|arial"/>"""
477    )
478    assert response is None
479    assert len(requests) == 1
480    assert (await requests[0].response()).status == 404
481
482
483async def test_page_route_should_not_throw_Invalid_Interception_Id_if_the_request_was_cancelled(
484    page, server
485):
486    await page.set_content("<iframe></iframe>")
487    route_future = asyncio.Future()
488    await page.route("**/*", lambda r, _: route_future.set_result(r))
489
490    async with page.expect_request("**/*"):
491        await page.eval_on_selector(
492            "iframe", """(frame, url) => frame.src = url""", server.EMPTY_PAGE
493        )
494    # Delete frame to cause request to be canceled.
495    await page.eval_on_selector("iframe", "frame => frame.remove()")
496    route = await route_future
497    await route.continue_()
498
499
500async def test_page_route_should_intercept_main_resource_during_cross_process_navigation(
501    page, server
502):
503    await page.goto(server.EMPTY_PAGE)
504    intercepted = []
505    await page.route(
506        server.CROSS_PROCESS_PREFIX + "/empty.html",
507        lambda route: (
508            intercepted.append(True),
509            asyncio.create_task(route.continue_()),
510        ),
511    )
512
513    response = await page.goto(server.CROSS_PROCESS_PREFIX + "/empty.html")
514    assert response.ok
515    assert len(intercepted) == 1
516
517
518@pytest.mark.skip_browser("webkit")
519async def test_page_route_should_create_a_redirect(page, server):
520    await page.goto(server.PREFIX + "/empty.html")
521
522    async def handle_route(route, request):
523        if request.url != (server.PREFIX + "/redirect_this"):
524            return await route.continue_()
525        await route.fulfill(status=301, headers={"location": "/empty.html"})
526
527    await page.route(
528        "**/*",
529        handle_route,
530    )
531
532    text = await page.evaluate(
533        """async url => {
534      const data = await fetch(url);
535      return data.text();
536    }""",
537        server.PREFIX + "/redirect_this",
538    )
539    assert text == ""
540
541
542async def test_page_route_should_support_cors_with_GET(page, server):
543    await page.goto(server.EMPTY_PAGE)
544
545    async def handle_route(route, request):
546        headers = (
547            {"access-control-allow-origin": "*"}
548            if request.url.endswith("allow")
549            else {}
550        )
551        await route.fulfill(
552            content_type="application/json",
553            headers=headers,
554            status=200,
555            body=json.dumps(["electric", "gas"]),
556        )
557
558    await page.route(
559        "**/cars*",
560        handle_route,
561    )
562    # Should succeed
563    resp = await page.evaluate(
564        """async () => {
565        const response = await fetch('https://example.com/cars?allow', { mode: 'cors' });
566        return response.json();
567      }"""
568    )
569
570    assert resp == ["electric", "gas"]
571
572    # Should be rejected
573    with pytest.raises(Error) as exc:
574        await page.evaluate(
575            """async () => {
576            const response = await fetch('https://example.com/cars?reject', { mode: 'cors' });
577            return response.json();
578        }"""
579        )
580    assert "failed" in exc.value.message
581
582
583async def test_page_route_should_support_cors_with_POST(page, server):
584    await page.goto(server.EMPTY_PAGE)
585    await page.route(
586        "**/cars",
587        lambda route: route.fulfill(
588            content_type="application/json",
589            headers={"Access-Control-Allow-Origin": "*"},
590            status=200,
591            body=json.dumps(["electric", "gas"]),
592        ),
593    )
594
595    resp = await page.evaluate(
596        """async () => {
597      const response = await fetch('https://example.com/cars', {
598        method: 'POST',
599        headers: { 'Content-Type': 'application/json' },
600        mode: 'cors',
601        body: JSON.stringify({ 'number': 1 })
602      });
603      return response.json();
604    }"""
605    )
606
607    assert resp == ["electric", "gas"]
608
609
610async def test_page_route_should_support_cors_for_different_methods(page, server):
611    await page.goto(server.EMPTY_PAGE)
612    await page.route(
613        "**/cars",
614        lambda route, request: route.fulfill(
615            content_type="application/json",
616            headers={"Access-Control-Allow-Origin": "*"},
617            status=200,
618            body=json.dumps([request.method, "electric", "gas"]),
619        ),
620    )
621
622    # First POST
623    resp = await page.evaluate(
624        """async () => {
625        const response = await fetch('https://example.com/cars', {
626          method: 'POST',
627          headers: { 'Content-Type': 'application/json' },
628          mode: 'cors',
629          body: JSON.stringify({ 'number': 1 })
630        });
631        return response.json();
632      }"""
633    )
634
635    assert resp == ["POST", "electric", "gas"]
636    # Then DELETE
637    resp = await page.evaluate(
638        """async () => {
639        const response = await fetch('https://example.com/cars', {
640          method: 'DELETE',
641          headers: {},
642          mode: 'cors',
643          body: ''
644        });
645        return response.json();
646      }"""
647    )
648
649    assert resp == ["DELETE", "electric", "gas"]
650
651
652async def test_request_fulfill_should_work_a(page, server):
653    await page.route(
654        "**/*",
655        lambda route: route.fulfill(
656            status=201,
657            headers={"foo": "bar"},
658            content_type="text/html",
659            body="Yo, page!",
660        ),
661    )
662
663    response = await page.goto(server.EMPTY_PAGE)
664    assert response.status == 201
665    assert response.headers["foo"] == "bar"
666    assert await page.evaluate("() => document.body.textContent") == "Yo, page!"
667
668
669async def test_request_fulfill_should_work_with_status_code_422(page, server):
670    await page.route(
671        "**/*",
672        lambda route: route.fulfill(status=422, body="Yo, page!"),
673    )
674
675    response = await page.goto(server.EMPTY_PAGE)
676    assert response.status == 422
677    assert response.status_text == "Unprocessable Entity"
678    assert await page.evaluate("() => document.body.textContent") == "Yo, page!"
679
680
681async def test_request_fulfill_should_allow_mocking_binary_responses(
682    page: Page, server, assert_to_be_golden, assetdir
683):
684    await page.route(
685        "**/*",
686        lambda route: route.fulfill(
687            content_type="image/png",
688            body=(assetdir / "pptr.png").read_bytes(),
689        ),
690    )
691
692    await page.evaluate(
693        """PREFIX => {
694      const img = document.createElement('img');
695      img.src = PREFIX + '/does-not-exist.png';
696      document.body.appendChild(img);
697      return new Promise(fulfill => img.onload = fulfill);
698    }""",
699        server.PREFIX,
700    )
701    img = await page.query_selector("img")
702    assert img
703    assert_to_be_golden(await img.screenshot(), "mock-binary-response.png")
704
705
706async def test_request_fulfill_should_allow_mocking_svg_with_charset(
707    page, server, assert_to_be_golden
708):
709    await page.route(
710        "**/*",
711        lambda route: route.fulfill(
712            content_type="image/svg+xml ; charset=utf-8",
713            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>',
714        ),
715    )
716
717    await page.evaluate(
718        """PREFIX => {
719      const img = document.createElement('img');
720      img.src = PREFIX + '/does-not-exist.svg';
721      document.body.appendChild(img);
722      return new Promise((f, r) => { img.onload = f; img.onerror = r; });
723    }""",
724        server.PREFIX,
725    )
726    img = await page.query_selector("img")
727    assert_to_be_golden(await img.screenshot(), "mock-svg.png")
728
729
730async def test_request_fulfill_should_work_with_file_path(
731    page: Page, server, assert_to_be_golden, assetdir
732):
733    await page.route(
734        "**/*",
735        lambda route: route.fulfill(
736            content_type="shouldBeIgnored", path=assetdir / "pptr.png"
737        ),
738    )
739    await page.evaluate(
740        """PREFIX => {
741      const img = document.createElement('img');
742      img.src = PREFIX + '/does-not-exist.png';
743      document.body.appendChild(img);
744      return new Promise(fulfill => img.onload = fulfill);
745    }""",
746        server.PREFIX,
747    )
748    img = await page.query_selector("img")
749    assert img
750    assert_to_be_golden(await img.screenshot(), "mock-binary-response.png")
751
752
753async def test_request_fulfill_should_stringify_intercepted_request_response_headers(
754    page, server
755):
756    await page.route(
757        "**/*",
758        lambda route: route.fulfill(
759            status=200, headers={"foo": True}, body="Yo, page!"
760        ),
761    )
762
763    response = await page.goto(server.EMPTY_PAGE)
764    assert response.status == 200
765    headers = response.headers
766    assert headers["foo"] == "True"
767    assert await page.evaluate("() => document.body.textContent") == "Yo, page!"
768
769
770async def test_request_fulfill_should_not_modify_the_headers_sent_to_the_server(
771    page, server
772):
773    await page.goto(server.PREFIX + "/empty.html")
774    interceptedRequests = []
775
776    # this is just to enable request interception, which disables caching in chromium
777    await page.route(server.PREFIX + "/unused", lambda route, req: None)
778
779    server.set_route(
780        "/something",
781        lambda response: (
782            interceptedRequests.append(response),
783            response.setHeader("Access-Control-Allow-Origin", "*"),
784            response.write(b"done"),
785            response.finish(),
786        ),
787    )
788
789    text = await page.evaluate(
790        """async url => {
791      const data = await fetch(url);
792      return data.text();
793    }""",
794        server.CROSS_PROCESS_PREFIX + "/something",
795    )
796    assert text == "done"
797
798    playwrightRequest = asyncio.Future()
799    await page.route(
800        server.CROSS_PROCESS_PREFIX + "/something",
801        lambda route, request: (
802            playwrightRequest.set_result(request),
803            asyncio.create_task(route.continue_(headers={**request.headers})),
804        ),
805    )
806
807    textAfterRoute = await page.evaluate(
808        """async url => {
809      const data = await fetch(url);
810      return data.text();
811    }""",
812        server.CROSS_PROCESS_PREFIX + "/something",
813    )
814    assert textAfterRoute == "done"
815
816    assert len(interceptedRequests) == 2
817    assert (
818        interceptedRequests[0].requestHeaders == interceptedRequests[1].requestHeaders
819    )
820
821
822async def test_request_fulfill_should_include_the_origin_header(page, server):
823    await page.goto(server.PREFIX + "/empty.html")
824    interceptedRequest = []
825    await page.route(
826        server.CROSS_PROCESS_PREFIX + "/something",
827        lambda route, request: (
828            interceptedRequest.append(request),
829            asyncio.create_task(
830                route.fulfill(
831                    headers={"Access-Control-Allow-Origin": "*"},
832                    content_type="text/plain",
833                    body="done",
834                )
835            ),
836        ),
837    )
838
839    text = await page.evaluate(
840        """async url => {
841      const data = await fetch(url);
842      return data.text();
843    }""",
844        server.CROSS_PROCESS_PREFIX + "/something",
845    )
846    assert text == "done"
847    assert len(interceptedRequest) == 1
848    assert interceptedRequest[0].headers["origin"] == server.PREFIX
849
850
851async def test_request_fulfill_should_work_with_request_interception(page, server):
852    requests = {}
853
854    async def _handle_route(route: Route):
855        requests[route.request.url.split("/").pop()] = route.request
856        await route.continue_()
857
858    await page.route("**/*", _handle_route)
859
860    server.set_redirect("/rrredirect", "/frames/one-frame.html")
861    await page.goto(server.PREFIX + "/rrredirect")
862    assert requests["rrredirect"].is_navigation_request()
863    assert requests["frame.html"].is_navigation_request()
864    assert requests["script.js"].is_navigation_request() is False
865    assert requests["style.css"].is_navigation_request() is False
866
867
868async def test_Interception_should_work_with_request_interception(
869    browser: Browser, https_server
870):
871    context = await browser.new_context(ignore_https_errors=True)
872    page = await context.new_page()
873
874    await page.route("**/*", lambda route: asyncio.ensure_future(route.continue_()))
875    response = await page.goto(https_server.EMPTY_PAGE)
876    assert response
877    assert response.status == 200
878    await context.close()
879
880
881async def test_ignore_http_errors_service_worker_should_intercept_after_a_service_worker(
882    page, server
883):
884    await page.goto(server.PREFIX + "/serviceworkers/fetchdummy/sw.html")
885    await page.evaluate("() => window.activationPromise")
886
887    # Sanity check.
888    sw_response = await page.evaluate('() => fetchDummy("foo")')
889    assert sw_response == "responseFromServiceWorker:foo"
890
891    def _handle_route(route):
892        asyncio.ensure_future(
893            route.fulfill(
894                status=200,
895                content_type="text/css",
896                body="responseFromInterception:" + route.request.url.split("/")[-1],
897            )
898        )
899
900    await page.route("**/foo", _handle_route)
901
902    # Page route is applied after service worker fetch event.
903    sw_response2 = await page.evaluate('() => fetchDummy("foo")')
904    assert sw_response2 == "responseFromServiceWorker:foo"
905
906    # Page route is not applied to service worker initiated fetch.
907    non_intercepted_response = await page.evaluate('() => fetchDummy("passthrough")')
908    assert non_intercepted_response == "FAILURE: Not Found"
909
Full Screen

Accelerate Your Automation Test Cycles With LambdaTest

Leverage LambdaTest’s cloud-based platform to execute your automation tests in parallel and trim down your test execution time significantly. Your first 100 automation testing minutes are on us.

Try LambdaTest

Run Python Tests on LambdaTest Cloud Grid

Execute automation tests with Playwright Python on a cloud-based Grid of 3000+ real browsers and operating systems for both web and mobile applications.

Test now for Free
LambdaTestX

We use cookies to give you the best experience. Cookies help to provide a more personalized experience and relevant advertising for you, and web analytics for us. Learn More in our Cookies policy, Privacy & Terms of service

Allow Cookie
Sarah

I hope you find the best code examples for your project.

If you want to accelerate automated browser testing, try LambdaTest. Your first 100 automation testing minutes are FREE.

Sarah Elson (Product & Growth Lead)