commit b809a39b5d18756e6a643cb2b4e930c1e1824773 parent aea8b511fbe104a469708afbe214fe24aaf4d87d Author: Julian Descottes <jdescottes@mozilla.com> Date: Wed, 5 Nov 2025 16:17:32 +0000 Bug 1988955 - [wdspec] Add tests for network.getData with dataType request r=whimboo Differential Revision: https://phabricator.services.mozilla.com/D267828 Diffstat:
9 files changed, 343 insertions(+), 113 deletions(-)
diff --git a/testing/web-platform/tests/webdriver/tests/bidi/network/add_data_collector/data_types.py b/testing/web-platform/tests/webdriver/tests/bidi/network/add_data_collector/data_types.py @@ -0,0 +1,73 @@ +import pytest +import webdriver.bidi.error as error + +pytestmark = pytest.mark.asyncio + + +async def test_request_and_response_same_collector( + bidi_session, url, setup_collected_data +): + [request, _] = await setup_collected_data( + fetch_url=url( + "/webdriver/tests/support/http_handlers/headers.py?content=somecontent" + ), + fetch_post_data="somedata", + data_types=["request", "response"], + ) + data = await bidi_session.network.get_data(request=request, data_type="request") + assert data["type"] == "string" + assert data["value"] == "somedata" + + data = await bidi_session.network.get_data(request=request, data_type="response") + assert data["type"] == "string" + assert data["value"] == "somecontent" + + +async def test_request_and_response_different_collectors( + bidi_session, url, add_data_collector, setup_collected_data +): + request_collector = await add_data_collector( + collector_type="blob", data_types=["request"], max_encoded_data_size=1000 + ) + [request, response_collector] = await setup_collected_data( + fetch_url=url( + "/webdriver/tests/support/http_handlers/headers.py?content=somecontent" + ), + fetch_post_data="somedata", + data_types=["response"], + ) + + # Check that both request and response data can be retrieved + data = await bidi_session.network.get_data(request=request, data_type="request") + assert data["type"] == "string" + assert data["value"] == "somedata" + + data = await bidi_session.network.get_data(request=request, data_type="response") + assert data["type"] == "string" + assert data["value"] == "somecontent" + + # Check that the request data can only be retrieved with the request collector + await bidi_session.network.get_data( + request=request, data_type="request", collector=request_collector + ) + with pytest.raises(error.NoSuchNetworkDataException): + await bidi_session.network.get_data( + request=request, data_type="request", collector=response_collector + ) + + # Check that the response data can only be retrieved with the response collector + await bidi_session.network.get_data( + request=request, data_type="response", collector=response_collector + ) + with pytest.raises(error.NoSuchNetworkDataException): + await bidi_session.network.get_data( + request=request, data_type="response", collector=request_collector + ) + + # Remove the response collector + await bidi_session.network.remove_data_collector(collector=response_collector) + + # Check request data can still be retrieved, but not the response data. + await bidi_session.network.get_data(request=request, data_type="request") + with pytest.raises(error.NoSuchNetworkDataException): + await bidi_session.network.get_data(request=request, data_type="response") diff --git a/testing/web-platform/tests/webdriver/tests/bidi/network/add_data_collector/max_encoded_data_size.py b/testing/web-platform/tests/webdriver/tests/bidi/network/add_data_collector/max_encoded_data_size.py @@ -1,11 +1,12 @@ import pytest import webdriver.bidi.error as error -from .. import PAGE_EMPTY_TEXT, RESPONSE_COMPLETED_EVENT +from .. import BEFORE_REQUEST_SENT_EVENT, PAGE_EMPTY_TEXT, RESPONSE_COMPLETED_EVENT pytestmark = pytest.mark.asyncio +@pytest.mark.parametrize("data_type", ["request", "response"]) async def test_max_encoded_data_size( bidi_session, add_data_collector, @@ -14,37 +15,42 @@ async def test_max_encoded_data_size( wait_for_future_safe, url, fetch, + data_type, ): context = await bidi_session.browsing_context.create(type_hint="tab") await setup_network_test( - events=[RESPONSE_COMPLETED_EVENT], context=context["context"] + events=[BEFORE_REQUEST_SENT_EVENT, RESPONSE_COMPLETED_EVENT], + context=context["context"], ) - small_collector = await add_data_collector(max_encoded_data_size=1) - big_collector = await add_data_collector(max_encoded_data_size=100000) + small_collector = await add_data_collector( + max_encoded_data_size=1, data_types=[data_type] + ) + big_collector = await add_data_collector( + max_encoded_data_size=100000, data_types=[data_type] + ) # Trigger a request in `context` on_response_completed = wait_for_event(RESPONSE_COMPLETED_EVENT) - await fetch(url(PAGE_EMPTY_TEXT), context=context) + + post_data = "somedata" if data_type == "request" else None + await fetch(url(PAGE_EMPTY_TEXT), post_data=post_data, context=context) + event = await wait_for_future_safe(on_response_completed) request = event["request"]["request"] # request data can be retrieved with big_collector or with no collector # argument await bidi_session.network.get_data( - request=request, data_type="response", collector=big_collector - ) - await bidi_session.network.get_data( - request=request, data_type="response" + request=request, data_type=data_type, collector=big_collector ) + await bidi_session.network.get_data(request=request, data_type=data_type) with pytest.raises(error.NoSuchNetworkDataException): await bidi_session.network.get_data( - request=request, data_type="response", collector=small_collector + request=request, data_type=data_type, collector=small_collector ) # Remove big_collector and check the collected data can no longer be accessed await bidi_session.network.remove_data_collector(collector=big_collector) with pytest.raises(error.NoSuchNetworkDataException): - await bidi_session.network.get_data( - request=request, data_type="response" - ) + await bidi_session.network.get_data(request=request, data_type=data_type) diff --git a/testing/web-platform/tests/webdriver/tests/bidi/network/disown_data/data_type.py b/testing/web-platform/tests/webdriver/tests/bidi/network/disown_data/data_type.py @@ -0,0 +1,33 @@ +import pytest +import webdriver.bidi.error as error + +from .. import PAGE_EMPTY_TEXT + +pytestmark = pytest.mark.asyncio + + +@pytest.mark.parametrize("data_type", ["request", "response"]) +async def test_data_type( + bidi_session, + url, + setup_collected_data, + data_type, +): + [request, collector] = await setup_collected_data( + fetch_url=url(PAGE_EMPTY_TEXT), + fetch_post_data="somedata", + data_types=[data_type], + ) + await bidi_session.network.disown_data( + request=request, data_type=data_type, collector=collector + ) + + # Check that after calling disown data, you can no longer get or disown the + # data. + with pytest.raises(error.NoSuchNetworkDataException): + await bidi_session.network.disown_data( + request=request, data_type=data_type, collector=collector + ) + + with pytest.raises(error.NoSuchNetworkDataException): + await bidi_session.network.get_data(request=request, data_type=data_type) diff --git a/testing/web-platform/tests/webdriver/tests/bidi/network/disown_data/invalid.py b/testing/web-platform/tests/webdriver/tests/bidi/network/disown_data/invalid.py @@ -49,6 +49,28 @@ async def test_params_data_type_invalid_value(bidi_session, value): ) +@pytest.mark.parametrize( + "collector_data_type", ["request", "response"] +) +async def test_params_data_type_mismatch( + bidi_session, url, setup_collected_data, collector_data_type +): + [request, collector] = await setup_collected_data( + fetch_url=url(PAGE_EMPTY_TEXT), + data_types=[collector_data_type], + ) + + if collector_data_type == "request": + data_type = "response" + else: + data_type = "request" + + with pytest.raises(error.NoSuchNetworkDataException): + await bidi_session.network.disown_data( + request=request, data_type=data_type, collector=collector + ) + + @pytest.mark.parametrize("value", [None, False, 42, {}, []]) async def test_params_collector_invalid_type(bidi_session, value): with pytest.raises(error.InvalidArgumentException): diff --git a/testing/web-platform/tests/webdriver/tests/bidi/network/disown_data/request.py b/testing/web-platform/tests/webdriver/tests/bidi/network/disown_data/request.py @@ -1,27 +0,0 @@ -import pytest -import webdriver.bidi.error as error - -from .. import PAGE_EMPTY_TEXT - -pytestmark = pytest.mark.asyncio - - -async def test_request( - bidi_session, - url, - setup_collected_data, -): - [request, collector] = await setup_collected_data(fetch_url=url(PAGE_EMPTY_TEXT)) - await bidi_session.network.disown_data( - request=request, data_type="response", collector=collector - ) - - # Check that after calling disown data, you can no longer get or disown the - # data. - with pytest.raises(error.NoSuchNetworkDataException): - await bidi_session.network.disown_data( - request=request, data_type="response", collector=collector - ) - - with pytest.raises(error.NoSuchNetworkDataException): - await bidi_session.network.get_data(request=request, data_type="response") diff --git a/testing/web-platform/tests/webdriver/tests/bidi/network/get_data/data_type.py b/testing/web-platform/tests/webdriver/tests/bidi/network/get_data/data_type.py @@ -0,0 +1,135 @@ +import base64 +import pytest +import webdriver.bidi.error as error + +from tests.support.asserts import assert_png + +from .. import ( + BEFORE_REQUEST_SENT_EVENT, + PAGE_DATA_URL_IMAGE, + PAGE_EMPTY_IMAGE, + PAGE_EMPTY_TEXT, + PAGE_OTHER_TEXT, + RESPONSE_COMPLETED_EVENT, +) + +pytestmark = pytest.mark.asyncio + + +@pytest.mark.parametrize( + "file, expected_value", [(PAGE_EMPTY_TEXT, "empty\n"), (PAGE_OTHER_TEXT, "other\n")] +) +async def test_data_type_response_text_file( + bidi_session, url, setup_collected_data, file, expected_value +): + [request, _] = await setup_collected_data(fetch_url=url(file)) + data = await bidi_session.network.get_data(request=request, data_type="response") + + assert data["type"] == "string" + assert data["value"] == expected_value + + +async def test_data_type_response_base64_file( + bidi_session, + url, + setup_collected_data, +): + [request, _] = await setup_collected_data(fetch_url=url(PAGE_EMPTY_IMAGE)) + data = await bidi_session.network.get_data(request=request, data_type="response") + + assert data["type"] == "base64" + assert isinstance(data["value"], str) + assert_png(data["value"]) + + +async def test_data_type_response_data_scheme_text( + bidi_session, url, setup_collected_data, +): + expected_value = "loremipsum" + [request, _] = await setup_collected_data(fetch_url=f"data:text/plain,{expected_value}") + data = await bidi_session.network.get_data(request=request, data_type="response") + + assert data["type"] == "string" + assert data["value"] == expected_value + + +async def test_data_type_response_data_scheme_image( + bidi_session, url, setup_collected_data, +): + expected_value = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNkYPhfDwAChwGA60e6kgAAAABJRU5ErkJggg==" + [request, _] = await setup_collected_data(fetch_url=f"data:image/png;base64,{expected_value}") + data = await bidi_session.network.get_data(request=request, data_type="response") + + assert data["type"] == "base64" + assert data["value"] == expected_value + + +async def test_data_type_response_empty_response( + bidi_session, + inline, + setup_collected_data, +): + empty_url = inline("", doctype="js") + [request, _] = await setup_collected_data(fetch_url=empty_url) + + data = await bidi_session.network.get_data(request=request, data_type="response") + + assert data["type"] == "string" + assert data["value"] == "" + + +async def test_data_type_request_text(bidi_session, url, setup_collected_data): + [request, _] = await setup_collected_data( + fetch_url=url(PAGE_EMPTY_TEXT), + fetch_post_data="somedata", + data_types=["request"], + ) + data = await bidi_session.network.get_data(request=request, data_type="request") + + assert data["type"] == "string" + assert data["value"] == "somedata" + + +async def test_data_type_request_text_multipart(bidi_session, url, setup_collected_data): + [request, _] = await setup_collected_data( + fetch_url=url(PAGE_EMPTY_TEXT), + fetch_post_data={"foo": 1, "bar": 2}, + data_types=["request"], + ) + + data = await bidi_session.network.get_data(request=request, data_type="request") + assert data["type"] == "string" + assert 'form-data; name="foo"' in data["value"] + assert 'form-data; name="bar"' in data["value"] + + +async def test_data_type_request_image(bidi_session, url, setup_collected_data): + expected_image = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNkYPhfDwAChwGA60e6kgAAAABJRU5ErkJggg==" + [request, _] = await setup_collected_data( + fetch_url=url(PAGE_EMPTY_TEXT), + fetch_post_data={ + "test_image": { + "filename": "image.png", + "type": "image/png", + "value": expected_image, + } + }, + data_types=["request"], + ) + + data = await bidi_session.network.get_data(request=request, data_type="request") + assert data["type"] == "base64" + assert isinstance(data["value"], str) + + decoded_image = base64.b64decode(expected_image) + decoded_value = base64.b64decode(data["value"]) + assert decoded_image in decoded_value + + +async def test_data_type_request_no_postdata(bidi_session, url, setup_collected_data): + [request, _] = await setup_collected_data( + fetch_url=url(PAGE_EMPTY_TEXT), + data_types=["request"], + ) + with pytest.raises(error.NoSuchNetworkDataException): + await bidi_session.network.get_data(request=request, data_type="request") diff --git a/testing/web-platform/tests/webdriver/tests/bidi/network/get_data/invalid.py b/testing/web-platform/tests/webdriver/tests/bidi/network/get_data/invalid.py @@ -36,6 +36,25 @@ async def test_params_data_type_invalid_value( with pytest.raises(error.InvalidArgumentException): await bidi_session.network.get_data(request=request, data_type=value) +@pytest.mark.parametrize( + "collector_data_type", ["request", "response"] +) +async def test_params_data_type_mismatch( + bidi_session, url, setup_collected_data, collector_data_type +): + [request, collector] = await setup_collected_data( + fetch_url=url(PAGE_EMPTY_TEXT), + data_types=[collector_data_type], + ) + + if collector_data_type == "request": + data_type = "response" + else: + data_type = "request" + + with pytest.raises(error.NoSuchNetworkDataException): + await bidi_session.network.get_data(request=request, data_type=data_type) + @pytest.mark.parametrize("value", [False, 42, {}, []]) async def test_params_collector_invalid_type( diff --git a/testing/web-platform/tests/webdriver/tests/bidi/network/get_data/request.py b/testing/web-platform/tests/webdriver/tests/bidi/network/get_data/request.py @@ -1,64 +0,0 @@ -import pytest - -from .. import PAGE_EMPTY_IMAGE, PAGE_EMPTY_TEXT, PAGE_OTHER_TEXT - -pytestmark = pytest.mark.asyncio - - -@pytest.mark.parametrize( - "file, expected_value", [(PAGE_EMPTY_TEXT, "empty\n"), (PAGE_OTHER_TEXT, "other\n")] -) -async def test_request_text_file( - bidi_session, url, setup_collected_data, file, expected_value -): - [request, _] = await setup_collected_data(fetch_url=url(file)) - data = await bidi_session.network.get_data(request=request, data_type="response") - - assert data["type"] == "string" - assert data["value"] == expected_value - - -async def test_request_base64_file( - bidi_session, - url, - setup_collected_data, -): - [request, _] = await setup_collected_data(fetch_url=url(PAGE_EMPTY_IMAGE)) - data = await bidi_session.network.get_data(request=request, data_type="response") - - assert data["type"] == "base64" - assert isinstance(data["value"], str) - - -async def test_request_data_scheme_text( - bidi_session, url, setup_collected_response, -): - [request, _] = await setup_collected_response(fetch_url="data:text/plain,abcdefgh") - data = await bidi_session.network.get_data(request=request, data_type="response") - - assert data["type"] == "string" - assert data["value"] == "abcdefgh" - - -async def test_request_data_scheme_image( - bidi_session, url, setup_collected_response, -): - [request, _] = await setup_collected_response(fetch_url="data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNkYPhfDwAChwGA60e6kgAAAABJRU5ErkJggg==") - data = await bidi_session.network.get_data(request=request, data_type="response") - - assert data["type"] == "base64" - assert data["value"] == "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNkYPhfDwAChwGA60e6kgAAAABJRU5ErkJggg==" - - -async def test_request_empty_response( - bidi_session, - inline, - setup_collected_data, -): - empty_url = inline("", doctype="js") - [request, _] = await setup_collected_data(fetch_url=empty_url) - - data = await bidi_session.network.get_data(request=request, data_type="response") - - assert data["type"] == "string" - assert data["value"] == "" diff --git a/testing/web-platform/tests/webdriver/tests/support/fixtures_bidi.py b/testing/web-platform/tests/webdriver/tests/support/fixtures_bidi.py @@ -625,24 +625,58 @@ def domain_value(server_config): def fetch(bidi_session, top_context, configuration): """Perform a fetch from the page of the provided context, default to the top context. + + :param url: The url to fetch. + :param method: Force a specific HTTP method. defaults to "GET". + :param headers: Dictionary of request headers. + :param post_data: Request post data (forces method to "POST" if set to "GET"). + If post_data is a dictionary, FormData will be used to set + the data as multipart form data. Otherwise will be set as + as string. + :param context: BrowsingContext info object. + :param timeout_in_seconds: Timeout in seconds (defaults to 3 seconds). """ async def fetch( url, - method="GET", + method=None, headers=None, post_data=None, context=top_context, timeout_in_seconds=3, ): + if method is None: + method = "GET" if post_data is None else "POST" + method_arg = f"method: '{method}'," headers_arg = "" if headers is not None: headers_arg = f"headers: {json.dumps(headers)}," - body_arg = "" - if post_data is not None: + if post_data is None: + body_arg = "" + elif isinstance(post_data, dict): + body_arg = f"""body: (() => {{ + const formData = new FormData(); + const data = {json.dumps(post_data)}; + for(const name in data) {{ + // Handle file binary data. + if (typeof data[name] == "object") {{ + const binary = atob(data[name].value); + const bytes = new Uint8Array(binary.length); + for (let i = 0; i < binary.length; i++) {{ + bytes[i] = binary.charCodeAt(i); + }} + const blob = new Blob([bytes], {{ type: data[name].type }}); + formData.append(name, blob, data[name].filename); + }} else {{ + formData.append(name, data[name]); + }} + }} + return formData; + }})(),""" + else: body_arg = f"body: {json.dumps(post_data)}," timeout_in_seconds = timeout_in_seconds * configuration["timeout_multiplier"] @@ -739,12 +773,11 @@ async def setup_collected_data( ) on_response_completed = wait_for_event("network.responseCompleted") - if fetch_post_data is None: - method = "GET" - else: - method = "POST" - - await fetch(fetch_url, method=method, post_data=fetch_post_data, context=context) + await fetch( + fetch_url, + post_data=fetch_post_data, + context=context, + ) response_completed_event = await on_response_completed request = response_completed_event["request"]["request"] return [request, collector]