How to use test_set_extra_http_headers_should_throw_for_non_string_header_values 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_network.py

Source: test_network.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
17from asyncio.futures import Future
18from typing import Dict, List, cast
19
20import pytest
21
22from playwright.async_api import Error, Page, Request, Response
23
24
25async def test_request_fulfill(page, server):
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.fulfill(body="Text")
37
38    await page.route(
39        "**/empty.html",
40        lambda route, request: asyncio.create_task(handle_request(route, request)),
41    )
42
43    response = await page.goto(server.EMPTY_PAGE)
44    assert response.ok
45    assert await response.text() == "Text"
46
47
48async def test_request_continue(page, server):
49    async def handle_request(route, request, intercepted):
50        intercepted.append(True)
51        await route.continue_()
52
53    intercepted = list()
54    await page.route(
55        "**/*",
56        lambda route, request: asyncio.create_task(
57            handle_request(route, request, intercepted)
58        ),
59    )
60
61    response = await page.goto(server.EMPTY_PAGE)
62    assert response.ok
63    assert intercepted == [True]
64    assert await page.title() == ""
65
66
67async def test_page_events_request_should_fire_for_navigation_requests(
68    page: Page, server
69):
70    requests = []
71    page.on("request", lambda r: requests.append(r))
72    await page.goto(server.EMPTY_PAGE)
73    assert len(requests) == 1
74
75
76async def test_page_events_request_should_fire_for_iframes(page, server, utils):
77    requests = []
78    page.on("request", lambda r: requests.append(r))
79    await page.goto(server.EMPTY_PAGE)
80    await utils.attach_frame(page, "frame1", server.EMPTY_PAGE)
81    assert len(requests) == 2
82
83
84async def test_page_events_request_should_fire_for_fetches(page, server):
85    requests = []
86    page.on("request", lambda r: requests.append(r))
87    await page.goto(server.EMPTY_PAGE)
88    await page.evaluate('() => fetch("/empty.html")')
89    assert len(requests) == 2
90
91
92async def test_page_events_request_should_report_requests_and_responses_handled_by_service_worker(
93    page: Page, server
94):
95    await page.goto(server.PREFIX + "/serviceworkers/fetchdummy/sw.html")
96    await page.evaluate("() => window.activationPromise")
97    sw_response = None
98    async with page.expect_request("**/*") as request_info:
99        sw_response = await page.evaluate('() => fetchDummy("foo")')
100    request = await request_info.value
101    assert sw_response == "responseFromServiceWorker:foo"
102    assert request.url == server.PREFIX + "/serviceworkers/fetchdummy/foo"
103    response = await request.response()
104    assert response.url == server.PREFIX + "/serviceworkers/fetchdummy/foo"
105    assert await response.text() == "responseFromServiceWorker:foo"
106
107
108async def test_request_frame_should_work_for_main_frame_navigation_request(
109    page, server
110):
111    requests = []
112    page.on("request", lambda r: requests.append(r))
113    await page.goto(server.EMPTY_PAGE)
114    assert len(requests) == 1
115    assert requests[0].frame == page.main_frame
116
117
118async def test_request_frame_should_work_for_subframe_navigation_request(
119    page, server, utils
120):
121    await page.goto(server.EMPTY_PAGE)
122    requests = []
123    page.on("request", lambda r: requests.append(r))
124    await utils.attach_frame(page, "frame1", server.EMPTY_PAGE)
125    assert len(requests) == 1
126    assert requests[0].frame == page.frames[1]
127
128
129async def test_request_frame_should_work_for_fetch_requests(page, server):
130    await page.goto(server.EMPTY_PAGE)
131    requests: List[Request] = []
132    page.on("request", lambda r: requests.append(r))
133    await page.evaluate('() => fetch("/digits/1.png")')
134    requests = [r for r in requests if "favicon" not in r.url]
135    assert len(requests) == 1
136    assert requests[0].frame == page.main_frame
137
138
139async def test_request_headers_should_work(
140    page, server, is_chromium, is_firefox, is_webkit
141):
142    response = await page.goto(server.EMPTY_PAGE)
143    if is_chromium:
144        assert "Chrome" in response.request.headers["user-agent"]
145    elif is_firefox:
146        assert "Firefox" in response.request.headers["user-agent"]
147    elif is_webkit:
148        assert "WebKit" in response.request.headers["user-agent"]
149
150
151async def test_request_headers_should_get_the_same_headers_as_the_server(
152    page: Page, server, is_webkit, is_win
153):
154    server_request_headers_future: Future[Dict[str, str]] = asyncio.Future()
155
156    def handle(request):
157        normalized_headers = {
158            key.decode().lower(): value[0].decode()
159            for key, value in request.requestHeaders.getAllRawHeaders()
160        }
161        server_request_headers_future.set_result(normalized_headers)
162        request.write(b"done")
163        request.finish()
164
165    server.set_route("/empty.html", handle)
166    response = await page.goto(server.EMPTY_PAGE)
167    server_headers = await server_request_headers_future
168    if is_webkit and is_win:
169        # Curl does not show accept-encoding and accept-language
170        server_headers.pop("accept-encoding")
171        server_headers.pop("accept-language")
172    assert cast(Response, response).request.headers == server_headers
173
174
175async def test_request_headers_should_get_the_same_headers_as_the_server_cors(
176    page: Page, server, is_webkit, is_win
177):
178    await page.goto(server.PREFIX + "/empty.html")
179    server_request_headers_future: Future[Dict[str, str]] = asyncio.Future()
180
181    def handle_something(request):
182        normalized_headers = {
183            key.decode().lower(): value[0].decode()
184            for key, value in request.requestHeaders.getAllRawHeaders()
185        }
186        server_request_headers_future.set_result(normalized_headers)
187        request.setHeader("Access-Control-Allow-Origin", "*")
188        request.write(b"done")
189        request.finish()
190
191    server.set_route("/something", handle_something)
192
193    text = None
194    async with page.expect_request("**/*") as request_info:
195        text = await page.evaluate(
196            """async url => {
197                const data = await fetch(url);
198                return data.text();
199            }""",
200            server.CROSS_PROCESS_PREFIX + "/something",
201        )
202    request = await request_info.value
203    assert text == "done"
204    server_headers = await server_request_headers_future
205    if is_webkit and is_win:
206        # Curl does not show accept-encoding and accept-language
207        server_headers.pop("accept-encoding")
208        server_headers.pop("accept-language")
209    assert request.headers == server_headers
210
211
212async def test_response_headers_should_work(page, server):
213    server.set_route("/empty.html", lambda r: (r.setHeader("foo", "bar"), r.finish()))
214
215    response = await page.goto(server.EMPTY_PAGE)
216    assert response.headers["foo"] == "bar"
217
218
219async def test_request_post_data_should_work(page, server):
220    await page.goto(server.EMPTY_PAGE)
221    server.set_route("/post", lambda r: r.finish())
222    requests = []
223    page.on("request", lambda r: requests.append(r))
224    await page.evaluate(
225        '() => fetch("./post", { method: "POST", body: JSON.stringify({foo: "bar"})})'
226    )
227    assert len(requests) == 1
228    assert requests[0].post_data == '{"foo":"bar"}'
229
230
231async def test_request_post_data__should_be_undefined_when_there_is_no_post_data(
232    page, server
233):
234    response = await page.goto(server.EMPTY_PAGE)
235    assert response.request.post_data is None
236
237
238async def test_should_parse_the_json_post_data(page, server):
239    await page.goto(server.EMPTY_PAGE)
240    server.set_route("/post", lambda req: req.finish())
241    requests = []
242    page.on("request", lambda r: requests.append(r))
243    await page.evaluate(
244        """() => fetch('./post', { method: 'POST', body: JSON.stringify({ foo: 'bar' }) })"""
245    )
246    assert len(requests) == 1
247    assert requests[0].post_data_json == {"foo": "bar"}
248
249
250async def test_should_parse_the_data_if_content_type_is_form_urlencoded(page, server):
251    await page.goto(server.EMPTY_PAGE)
252    server.set_route("/post", lambda req: req.finish())
253    requests = []
254    page.on("request", lambda r: requests.append(r))
255    await page.set_content(
256        """<form method='POST' action='/post'><input type='text' name='foo' value='bar'><input type='number' name='baz' value='123'><input type='submit'></form>"""
257    )
258    await page.click("input[type=submit]")
259    assert len(requests) == 1
260    assert requests[0].post_data_json == {"foo": "bar", "baz": "123"}
261
262
263async def test_should_be_undefined_when_there_is_no_post_data(page, server):
264    response = await page.goto(server.EMPTY_PAGE)
265    assert response.request.post_data_json is None
266
267
268async def test_should_work_with_binary_post_data(page, server):
269    await page.goto(server.EMPTY_PAGE)
270    server.set_route("/post", lambda req: req.finish())
271    requests = []
272    page.on("request", lambda r: requests.append(r))
273    await page.evaluate(
274        """async () => {
275        await fetch('./post', { method: 'POST', body: new Uint8Array(Array.from(Array(256).keys())) })
276    }"""
277    )
278    assert len(requests) == 1
279    buffer = requests[0].post_data_buffer
280    assert len(buffer) == 256
281    for i in range(256):
282        assert buffer[i] == i
283
284
285async def test_should_work_with_binary_post_data_and_interception(page, server):
286    await page.goto(server.EMPTY_PAGE)
287    server.set_route("/post", lambda req: req.finish())
288    requests = []
289    await page.route("/post", lambda route: asyncio.ensure_future(route.continue_()))
290    page.on("request", lambda r: requests.append(r))
291    await page.evaluate(
292        """async () => {
293        await fetch('./post', { method: 'POST', body: new Uint8Array(Array.from(Array(256).keys())) })
294    }"""
295    )
296    assert len(requests) == 1
297    buffer = requests[0].post_data_buffer
298    assert len(buffer) == 256
299    for i in range(256):
300        assert buffer[i] == i
301
302
303async def test_response_text_should_work(page, server):
304    response = await page.goto(server.PREFIX + "/simple.json")
305    assert await response.text() == '{"foo": "bar"}\n'
306
307
308async def test_response_text_should_return_uncompressed_text(page, server):
309    server.enable_gzip("/simple.json")
310    response = await page.goto(server.PREFIX + "/simple.json")
311    assert response.headers["content-encoding"] == "gzip"
312    assert await response.text() == '{"foo": "bar"}\n'
313
314
315async def test_response_text_should_throw_when_requesting_body_of_redirected_response(
316    page, server
317):
318    server.set_redirect("/foo.html", "/empty.html")
319    response = await page.goto(server.PREFIX + "/foo.html")
320    redirected_from = response.request.redirected_from
321    assert redirected_from
322    redirected = await redirected_from.response()
323    assert redirected.status == 302
324    error = None
325    try:
326        await redirected.text()
327    except Error as exc:
328        error = exc
329    assert "Response body is unavailable for redirect responses" in error.message
330
331
332async def test_response_json_should_work(page, server):
333    response = await page.goto(server.PREFIX + "/simple.json")
334    assert await response.json() == {"foo": "bar"}
335
336
337async def test_response_body_should_work(page, server, assetdir):
338    response = await page.goto(server.PREFIX + "/pptr.png")
339    with open(
340        assetdir / "pptr.png",
341        "rb",
342    ) as fd:
343        assert fd.read() == await response.body()
344
345
346async def test_response_body_should_work_with_compression(page, server, assetdir):
347    server.enable_gzip("/pptr.png")
348    response = await page.goto(server.PREFIX + "/pptr.png")
349    with open(
350        assetdir / "pptr.png",
351        "rb",
352    ) as fd:
353        assert fd.read() == await response.body()
354
355
356async def test_response_status_text_should_work(page, server):
357    server.set_route("/cool", lambda r: (r.setResponseCode(200, b"cool!"), r.finish()))
358
359    response = await page.goto(server.PREFIX + "/cool")
360    assert response.status_text == "cool!"
361
362
363async def test_request_resource_type_should_return_event_source(page, server):
364    SSE_MESSAGE = {"foo": "bar"}
365    # 1. Setup server-sent events on server that immediately sends a message to the client.
366    server.set_route(
367        "/sse",
368        lambda r: (
369            r.setHeader("Content-Type", "text/event-stream"),
370            r.setHeader("Connection", "keep-alive"),
371            r.setHeader("Cache-Control", "no-cache"),
372            r.setResponseCode(200),
373            r.write(f"data: {json.dumps(SSE_MESSAGE)}\n\n".encode()),
374            r.finish(),
375        ),
376    )
377
378    # 2. Subscribe to page request events.
379    await page.goto(server.EMPTY_PAGE)
380    requests = []
381    page.on("request", lambda r: requests.append(r))
382    # 3. Connect to EventSource in browser and return first message.
383    assert (
384        await page.evaluate(
385            """() => {
386      const eventSource = new EventSource('/sse');
387      return new Promise(resolve => {
388        eventSource.onmessage = e => resolve(JSON.parse(e.data));
389      });
390    }"""
391        )
392        == SSE_MESSAGE
393    )
394    assert requests[0].resource_type == "eventsource"
395
396
397async def test_network_events_request(page, server):
398    requests = []
399    page.on("request", lambda r: requests.append(r))
400    await page.goto(server.EMPTY_PAGE)
401    assert len(requests) == 1
402    assert requests[0].url == server.EMPTY_PAGE
403    assert requests[0].resource_type == "document"
404    assert requests[0].method == "GET"
405    assert await requests[0].response()
406    assert requests[0].frame == page.main_frame
407    assert requests[0].frame.url == server.EMPTY_PAGE
408
409
410async def test_network_events_response(page, server):
411    responses = []
412    page.on("response", lambda r: responses.append(r))
413    await page.goto(server.EMPTY_PAGE)
414    assert len(responses) == 1
415    assert responses[0].url == server.EMPTY_PAGE
416    assert responses[0].status == 200
417    assert responses[0].ok
418    assert responses[0].request
419
420
421async def test_network_events_request_failed(
422    page, server, is_chromium, is_webkit, is_mac, is_win
423):
424    def handle_request(request):
425        request.setHeader("Content-Type", "text/css")
426        request.transport.loseConnection()
427
428    server.set_route("/one-style.css", handle_request)
429
430    failed_requests = []
431    page.on("requestfailed", lambda request: failed_requests.append(request))
432    await page.goto(server.PREFIX + "/one-style.html")
433    assert len(failed_requests) == 1
434    assert "one-style.css" in failed_requests[0].url
435    assert await failed_requests[0].response() is None
436    assert failed_requests[0].resource_type == "stylesheet"
437    if is_chromium:
438        assert failed_requests[0].failure == "net::ERR_EMPTY_RESPONSE"
439    elif is_webkit:
440        if is_mac:
441            assert failed_requests[0].failure == "The network connection was lost."
442        elif is_win:
443            assert (
444                failed_requests[0].failure
445                == "Server returned nothing (no headers, no data)"
446            )
447        else:
448            assert failed_requests[0].failure == "Message Corrupt"
449    else:
450        assert failed_requests[0].failure == "NS_ERROR_NET_RESET"
451    assert failed_requests[0].frame
452
453
454async def test_network_events_request_finished(page, server):
455    async with page.expect_event("requestfinished") as event_info:
456        await page.goto(server.EMPTY_PAGE)
457    request = await event_info.value
458    assert request.url == server.EMPTY_PAGE
459    assert await request.response()
460    assert request.frame == page.main_frame
461    assert request.frame.url == server.EMPTY_PAGE
462
463
464async def test_network_events_should_fire_events_in_proper_order(page, server):
465    events = []
466    page.on("request", lambda request: events.append("request"))
467    page.on("response", lambda response: events.append("response"))
468    response = await page.goto(server.EMPTY_PAGE)
469    await response.finished()
470    events.append("requestfinished")
471    assert events == ["request", "response", "requestfinished"]
472
473
474async def test_network_events_should_support_redirects(page, server):
475    events = []
476    page.on("request", lambda request: events.append(f"{request.method} {request.url}"))
477    page.on(
478        "response", lambda response: events.append(f"{response.status} {response.url}")
479    )
480    page.on("requestfinished", lambda request: events.append(f"DONE {request.url}"))
481    page.on("requestfailed", lambda request: events.append(f"FAIL {request.url}"))
482    server.set_redirect("/foo.html", "/empty.html")
483    FOO_URL = server.PREFIX + "/foo.html"
484    response = await page.goto(FOO_URL)
485    await response.finished()
486    assert events == [
487        f"GET {FOO_URL}",
488        f"302 {FOO_URL}",
489        f"DONE {FOO_URL}",
490        f"GET {server.EMPTY_PAGE}",
491        f"200 {server.EMPTY_PAGE}",
492        f"DONE {server.EMPTY_PAGE}",
493    ]
494    redirected_from = response.request.redirected_from
495    assert "/foo.html" in redirected_from.url
496    assert redirected_from.redirected_from is None
497    assert redirected_from.redirected_to == response.request
498
499
500async def test_request_is_navigation_request_should_work(page, server):
501    pytest.skip(msg="test")
502    requests = {}
503
504    def handle_request(request):
505        requests[request.url().split("/").pop()] = request
506
507    page.on("request", handle_request)
508    server.set_redirect("/rrredirect", "/frames/one-frame.html")
509    await page.goto(server.PREFIX + "/rrredirect")
510    print("kek")
511    assert requests.get("rrredirect").is_navigation_request()
512    assert requests.get("one-frame.html").is_navigation_request()
513    assert requests.get("frame.html").is_navigation_request()
514    assert requests.get("script.js").is_navigation_request() is False
515    assert requests.get("style.css").is_navigation_request() is False
516
517
518async def test_request_is_navigation_request_should_work_when_navigating_to_image(
519    page, server
520):
521    requests = []
522    page.on("request", lambda r: requests.append(r))
523    await page.goto(server.PREFIX + "/pptr.png")
524    assert requests[0].is_navigation_request()
525
526
527async def test_set_extra_http_headers_should_work(page, server):
528    await page.set_extra_http_headers({"foo": "bar"})
529
530    request = (
531        await asyncio.gather(
532            server.wait_for_request("/empty.html"),
533            page.goto(server.EMPTY_PAGE),
534        )
535    )[0]
536    assert request.getHeader("foo") == "bar"
537
538
539async def test_set_extra_http_headers_should_work_with_redirects(page, server):
540    server.set_redirect("/foo.html", "/empty.html")
541    await page.set_extra_http_headers({"foo": "bar"})
542
543    request = (
544        await asyncio.gather(
545            server.wait_for_request("/empty.html"),
546            page.goto(server.PREFIX + "/foo.html"),
547        )
548    )[0]
549    assert request.getHeader("foo") == "bar"
550
551
552async def test_set_extra_http_headers_should_work_with_extra_headers_from_browser_context(
553    browser, server
554):
555    context = await browser.new_context()
556    await context.set_extra_http_headers({"foo": "bar"})
557
558    page = await context.new_page()
559    request = (
560        await asyncio.gather(
561            server.wait_for_request("/empty.html"),
562            page.goto(server.EMPTY_PAGE),
563        )
564    )[0]
565    await context.close()
566    assert request.getHeader("foo") == "bar"
567
568
569async def test_set_extra_http_headers_should_override_extra_headers_from_browser_context(
570    browser, server
571):
572    context = await browser.new_context(extra_http_headers={"fOo": "bAr", "baR": "foO"})
573
574    page = await context.new_page()
575    await page.set_extra_http_headers({"Foo": "Bar"})
576
577    request = (
578        await asyncio.gather(
579            server.wait_for_request("/empty.html"),
580            page.goto(server.EMPTY_PAGE),
581        )
582    )[0]
583    await context.close()
584    assert request.getHeader("foo") == "Bar"
585    assert request.getHeader("bar") == "foO"
586
587
588async def test_set_extra_http_headers_should_throw_for_non_string_header_values(
589    page, server
590):
591    error = None
592    try:
593        await page.set_extra_http_headers({"foo": 1})
594    except Error as exc:
595        error = exc
596    assert error.message == "headers[0].value: expected string, got number"
597
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)