[wdspec] Add wdspec tests for getData

Differential Revision: https://phabricator.services.mozilla.com/D259540

bugzilla-url: https://bugzilla.mozilla.org/show_bug.cgi?id=1971780
gecko-commit: b9d86d414f984e4ad602f7c5b30dcb078d2097b2
gecko-reviewers: webdriver-reviewers, Sasha

Author: juliandescottes

tools/webdriver/webdriver/bidi/modules/network.py

@@ -293,3 +293,28 @@ def set_cache_behavior(
             params["contexts"] = contexts
 
         return params
+
+    @command
+    def get_data(
+            self,
+            request: str,
+            data_type: str,
+            collector: Optional[str] = None,
+            disown: Optional[bool] = None) -> Mapping[str, Any]:
+        params: MutableMapping[str, Any] = {
+            "request": request,
+            "dataType": data_type,
+        }
+
+        if collector is not None:
+            params["collector"] = collector
+
+        if disown is not None:
+            params["disown"] = disown
+
+        return params
+
+    @get_data.result
+    def _get_data(self, result: Mapping[str, Any]) -> Any:
+        assert result["bytes"] is not None
+        return result["bytes"]

webdriver/tests/bidi/network/get_data/__init__.py

@@ -0,0 +1,40 @@
+import pytest
+import webdriver.bidi.error as error
+
+from .. import PAGE_EMPTY_TEXT
+
+pytestmark = pytest.mark.asyncio
+
+
+async def test_single_collector(bidi_session, url, setup_collected_response):
+    [request, collector] = await setup_collected_response(
+        fetch_url=url(PAGE_EMPTY_TEXT)
+    )
+    data = await bidi_session.network.get_data(
+        request=request, data_type="response", collector=collector
+    )
+
+    assert data["type"] == "string"
+    assert data["value"] == "empty\n"
+
+
+async def test_several_collectors(
+    bidi_session, url, add_data_collector, setup_collected_response
+):
+    collector = await add_data_collector(
+        collector_type="blob", data_types=["response"], max_encoded_data_size=1000
+    )
+    [request, other_collector] = await setup_collected_response(
+        fetch_url=url(PAGE_EMPTY_TEXT)
+    )
+
+    data = await bidi_session.network.get_data(
+        request=request, data_type="response", collector=collector
+    )
+    assert data["type"] == "string"
+    assert data["value"] == "empty\n"
+
+    data_from_other_collector = await bidi_session.network.get_data(
+        request=request, data_type="response", collector=other_collector
+    )
+    assert data_from_other_collector == data

webdriver/tests/bidi/network/get_data/collector.py

@@ -0,0 +1,60 @@
+import pytest
+import webdriver.bidi.error as error
+
+from .. import PAGE_EMPTY_TEXT
+
+pytestmark = pytest.mark.asyncio
+
+
+@pytest.mark.parametrize("disown", [True, False])
+async def test_single_collector(bidi_session, url, setup_collected_response, disown):
+    [request, collector] = await setup_collected_response(
+        fetch_url=url(PAGE_EMPTY_TEXT)
+    )
+    data = await bidi_session.network.get_data(
+        request=request, data_type="response", collector=collector, disown=disown
+    )
+    # Assert the data retrieved with the disown parameter is still valid
+    assert data["type"] == "string"
+    assert data["value"] == "empty\n"
+
+    if disown:
+        with pytest.raises(error.NoSuchNetworkDataException):
+            await bidi_session.network.get_data(request=request, data_type="response")
+    else:
+        data_again = await bidi_session.network.get_data(
+            request=request, data_type="response"
+        )
+        assert data_again == data
+
+
+async def test_several_collectors(
+    bidi_session, url, add_data_collector, setup_collected_response
+):
+    collector = await add_data_collector(
+        collector_type="blob", data_types=["response"], max_encoded_data_size=1000
+    )
+    [request, other_collector] = await setup_collected_response(
+        fetch_url=url(PAGE_EMPTY_TEXT)
+    )
+
+    data = await bidi_session.network.get_data(
+        request=request, data_type="response", collector=collector, disown=True
+    )
+
+    # Check that the data can no longer be retrieved for the disowned collector
+    with pytest.raises(error.NoSuchNetworkDataException):
+        await bidi_session.network.get_data(
+            request=request, data_type="response", collector=collector
+        )
+
+    # Check that the data can still be retrieved for the other collector or
+    # without any collector parameter.
+    data_from_other_collector = await bidi_session.network.get_data(
+        request=request, data_type="response", collector=other_collector
+    )
+    assert data_from_other_collector == data
+
+    data_without_collector = await bidi_session.network.get_data(
+        request=request, data_type="response"
+    )

webdriver/tests/bidi/network/get_data/disown.py

@@ -0,0 +1,81 @@
+import pytest
+import webdriver.bidi.error as error
+
+from .. import PAGE_EMPTY_TEXT
+
+pytestmark = pytest.mark.asyncio
+
+
+@pytest.mark.parametrize("value", [None, False, 42, {}, []])
+async def test_params_request_invalid_type(bidi_session, value):
+    with pytest.raises(error.InvalidArgumentException):
+        await bidi_session.network.get_data(request=value, data_type="response")
+
+
+async def test_params_request_non_existent(bidi_session):
+    with pytest.raises(error.NoSuchNetworkDataException):
+        await bidi_session.network.get_data(
+            request="does_not_exist", data_type="response"
+        )
+
+
+@pytest.mark.parametrize("value", [None, False, 42, {}, []])
+async def test_params_data_type_invalid_type(
+    bidi_session, url, setup_collected_response, value
+):
+    [request, _] = await setup_collected_response(fetch_url=url(PAGE_EMPTY_TEXT))
+    with pytest.raises(error.InvalidArgumentException):
+        await bidi_session.network.get_data(request=request, data_type=value)
+
+
+@pytest.mark.parametrize("value", ["", "invalid"])
+async def test_params_data_type_invalid_value(
+    bidi_session, url, setup_collected_response, value
+):
+    [request, _] = await setup_collected_response(fetch_url=url(PAGE_EMPTY_TEXT))
+    with pytest.raises(error.InvalidArgumentException):
+        await bidi_session.network.get_data(request=request, data_type=value)
+
+
+@pytest.mark.parametrize("value", [False, 42, {}, []])
+async def test_params_collector_invalid_type(
+    bidi_session, url, setup_collected_response, value
+):
+    [request, _] = await setup_collected_response(fetch_url=url(PAGE_EMPTY_TEXT))
+    with pytest.raises(error.InvalidArgumentException):
+        await bidi_session.network.get_data(
+            request=request, data_type="response", collector=value
+        )
+
+
+async def test_params_collector_non_existent(
+    bidi_session, url, setup_collected_response
+):
+    [request, _] = await setup_collected_response(fetch_url=url(PAGE_EMPTY_TEXT))
+    with pytest.raises(error.NoSuchNetworkCollectorException):
+        await bidi_session.network.get_data(
+            request=request, data_type="response", collector="does_not_exist"
+        )
+
+
+@pytest.mark.parametrize("value", ["", "true", "false", 42, {}, []])
+async def test_params_disown_invalid_type(
+    bidi_session, url, setup_collected_response, value
+):
+    [request, _] = await setup_collected_response(fetch_url=url(PAGE_EMPTY_TEXT))
+
+    with pytest.raises(error.InvalidArgumentException):
+        await bidi_session.network.get_data(
+            request=request, data_type="response", disown=value
+        )
+
+
+async def test_params_disown_true_without_collector(
+    bidi_session, url, setup_collected_response
+):
+    [request, _] = await setup_collected_response(fetch_url=url(PAGE_EMPTY_TEXT))
+    with pytest.raises(error.InvalidArgumentException):
+        await bidi_session.network.get_data(
+            request=request, data_type="response", disown=True
+        )
+

webdriver/tests/bidi/network/get_data/invalid.py

@@ -0,0 +1,29 @@
+import pytest
+import webdriver.bidi.error as error
+
+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_response, file, expected_value
+):
+    [request, _] = await setup_collected_response(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_response,
+):
+    [request, _] = await setup_collected_response(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)