[wdspec] add network interception invalidation tests for all remaining methods

Synced from #42667

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

bugzilla-url: https://bugzilla.mozilla.org/show_bug.cgi?id=1870032
gecko-commit: d3a29d02b3dd8d8cfbfff6fcad00d94226e102e8
gecko-reviewers: webdriver-reviewers, whimboo

Author: juliandescottes

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

@@ -3,6 +3,11 @@
 from ._module import BidiModule, command
 
 
+class AuthCredentials(Dict[str, Any]):
+    def __init__(self, username: str, password: str):
+        dict.__init__(self, type="password", username=username, password=password)
+
+
 class URLPatternPattern(Dict[str, Any]):
     def __init__(
         self,
@@ -57,11 +62,87 @@ def _add_intercept(self, result: Mapping[str, Any]) -> Any:
         assert result["intercept"] is not None
         return result["intercept"]
 
+    @command
+    def continue_with_auth(
+        self,
+        request: str,
+        action: str,
+        credentials: Optional[AuthCredentials] = None
+    ) -> Mapping[str, Any]:
+        params: MutableMapping[str, Any] = {
+            "request": request,
+            "action": action,
+        }
+
+        if action == "provideCredentials" and credentials is not None:
+            params["credentials"] = credentials
+
+        return params
+
+    @command
+    def continue_request(self,
+                         request: str,
+                         method: Optional[str] = None,
+                         url: Optional[str] = None) -> Mapping[str, Any]:
+        params: MutableMapping[str, Any] = {
+            "request": request,
+        }
+
+        if method is not None:
+            params["method"] = method
+
+        if url is not None:
+            params["url"] = url
+
+        # TODO: Add support for missing parameters: body, cookies, headers
+
+        return params
+
+    @command
+    def continue_response(
+            self,
+            request: str,
+            reason_phrase: Optional[str] = None,
+            status_code: Optional[int] = None) -> Mapping[str, Any]:
+        params: MutableMapping[str, Any] = {
+            "request": request,
+        }
+
+        if reason_phrase is not None:
+            params["reasonPhrase"] = reason_phrase
+
+        if status_code is not None:
+            params["statusCode"] = status_code
+
+        # TODO: Add support for missing parameters: body, credentials, headers
+
+        return params
+
     @command
     def fail_request(self, request: str) -> Mapping[str, Any]:
         params: MutableMapping[str, Any] = {"request": request}
         return params
 
+    @command
+    def provide_response(
+            self,
+            request: str,
+            reason_phrase: Optional[str] = None,
+            status_code: Optional[int] = None) -> Mapping[str, Any]:
+        params: MutableMapping[str, Any] = {
+            "request": request,
+        }
+
+        if reason_phrase is not None:
+            params["reasonPhrase"] = reason_phrase
+
+        if status_code is not None:
+            params["statusCode"] = status_code
+
+        # TODO: Add support for missing parameters: body, cookies, headers
+
+        return params
+
     @command
     def remove_intercept(self, intercept: str) -> Mapping[str, Any]:
         params: MutableMapping[str, Any] = {"intercept": intercept}

webdriver/tests/bidi/network/conftest.py

@@ -1,12 +1,13 @@
 import json
 
+import asyncio
 import pytest
 import pytest_asyncio
 
 from webdriver.bidi.error import NoSuchInterceptException
 from webdriver.bidi.modules.script import ContextTarget
 
-from . import PAGE_EMPTY_HTML, RESPONSE_COMPLETED_EVENT
+from . import PAGE_EMPTY_HTML, PAGE_EMPTY_TEXT, RESPONSE_COMPLETED_EVENT
 
 
 @pytest_asyncio.fixture
@@ -124,3 +125,36 @@ async def on_event(method, data, event=event):
     # cleanup
     for remove_listener in listeners:
         remove_listener()
+
+
+@pytest_asyncio.fixture
+async def setup_blocked_request(
+    setup_network_test, url, add_intercept, fetch, wait_for_event
+):
+    async def setup_blocked_request(phase):
+        await setup_network_test(events=[f"network.{phase}"])
+
+        if phase == "authRequired":
+            blocked_url = url(
+                "/webdriver/tests/support/http_handlers/authentication.py?realm=testrealm"
+            )
+        else:
+            blocked_url = url(PAGE_EMPTY_TEXT)
+
+        await add_intercept(
+            phases=[phase],
+            url_patterns=[
+                {
+                    "type": "string",
+                    "pattern": blocked_url,
+                }
+            ],
+        )
+
+        asyncio.ensure_future(fetch(blocked_url))
+        event = await wait_for_event(f"network.{phase}")
+        request = event["request"]["request"]
+
+        return request
+
+    return setup_blocked_request

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

@@ -0,0 +1,67 @@
+import pytest
+import webdriver.bidi.error as error
+
+from .. import PAGE_EMPTY_TEXT, RESPONSE_COMPLETED_EVENT
+
+pytestmark = pytest.mark.asyncio
+
+
+@pytest.mark.parametrize("value", [False, 42, {}, []])
+async def test_params_method_invalid_type(setup_blocked_request, bidi_session, value):
+    request = await setup_blocked_request("beforeRequestSent")
+
+    with pytest.raises(error.InvalidArgumentException):
+        await bidi_session.network.continue_request(request=request, method=value)
+
+
+@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.continue_request(request=value)
+
+
+@pytest.mark.parametrize("value", ["", "foo"])
+async def test_params_request_invalid_value(bidi_session, value):
+    with pytest.raises(error.NoSuchRequestException):
+        await bidi_session.network.continue_request(request=value)
+
+
+async def test_params_request_no_such_request(
+    bidi_session, setup_network_test, wait_for_event, fetch, url
+):
+    await setup_network_test(
+        events=[
+            RESPONSE_COMPLETED_EVENT,
+        ]
+    )
+    on_response_completed = wait_for_event(RESPONSE_COMPLETED_EVENT)
+
+    text_url = url(PAGE_EMPTY_TEXT)
+    await fetch(text_url)
+
+    response_completed_event = await on_response_completed
+    request = response_completed_event["request"]["request"]
+
+    with pytest.raises(error.NoSuchRequestException):
+        await bidi_session.network.continue_request(request=request)
+
+
+@pytest.mark.parametrize("value", [False, 42, {}, []])
+async def test_params_url_invalid_type(setup_blocked_request, bidi_session, value):
+    request = await setup_blocked_request("beforeRequestSent")
+
+    with pytest.raises(error.InvalidArgumentException):
+        await bidi_session.network.continue_request(request=request, url=value)
+
+
+@pytest.mark.parametrize("protocol", ["http", "https"])
+@pytest.mark.parametrize("value", [":invalid", "#invalid"])
+async def test_params_url_invalid_value(
+    setup_blocked_request, bidi_session, protocol, value
+):
+    request = await setup_blocked_request("beforeRequestSent")
+
+    with pytest.raises(error.InvalidArgumentException):
+        await bidi_session.network.continue_request(
+            request=request, url=f"{protocol}://{value}"
+        )

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

@@ -0,0 +1,77 @@
+import pytest
+import webdriver.bidi.error as error
+
+from .. import PAGE_EMPTY_TEXT, RESPONSE_COMPLETED_EVENT
+
+pytestmark = pytest.mark.asyncio
+
+
+async def test_params_request_invalid_phase(setup_blocked_request, bidi_session):
+    request = await setup_blocked_request("beforeRequestSent")
+
+    with pytest.raises(error.InvalidArgumentException):
+        await bidi_session.network.continue_response(request=request)
+
+
+@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.continue_response(request=value)
+
+
+@pytest.mark.parametrize("value", ["", "foo"])
+async def test_params_request_invalid_value(bidi_session, value):
+    with pytest.raises(error.NoSuchRequestException):
+        await bidi_session.network.continue_response(request=value)
+
+
+async def test_params_request_no_such_request(
+    bidi_session, setup_network_test, wait_for_event, fetch, url
+):
+    await setup_network_test(
+        events=[
+            RESPONSE_COMPLETED_EVENT,
+        ]
+    )
+    on_response_completed = wait_for_event(RESPONSE_COMPLETED_EVENT)
+
+    text_url = url(PAGE_EMPTY_TEXT)
+    await fetch(text_url)
+
+    response_completed_event = await on_response_completed
+    request = response_completed_event["request"]["request"]
+
+    with pytest.raises(error.NoSuchRequestException):
+        await bidi_session.network.continue_response(request=request)
+
+
+@pytest.mark.parametrize("value", [False, 42, {}, []])
+async def test_params_reason_phrase_invalid_type(
+    setup_blocked_request, bidi_session, value
+):
+    request = await setup_blocked_request("responseStarted")
+
+    with pytest.raises(error.InvalidArgumentException):
+        await bidi_session.network.continue_response(
+            request=request, reason_phrase=value
+        )
+
+
+@pytest.mark.parametrize("value", [False, "foo", {}, []])
+async def test_params_status_code_invalid_type(
+    setup_blocked_request, bidi_session, value
+):
+    request = await setup_blocked_request("responseStarted")
+
+    with pytest.raises(error.InvalidArgumentException):
+        await bidi_session.network.continue_response(request=request, status_code=value)
+
+
+@pytest.mark.parametrize("value", [-1, 4.3])
+async def test_params_status_code_invalid_value(
+    setup_blocked_request, bidi_session, value
+):
+    request = await setup_blocked_request("responseStarted")
+
+    with pytest.raises(error.InvalidArgumentException):
+        await bidi_session.network.continue_response(request=request, status_code=value)