From 26cbc19d7ddfc5face9d7da3c3416a1d9716ea3c Mon Sep 17 00:00:00 2001 From: Julian Descottes Date: Thu, 7 Sep 2023 18:44:17 +0000 Subject: [PATCH] Bug 1848156 - [wdspec] Add basic wdspec test for add intercept r=webdriver-reviewers,whimboo Depends on D185457 Differential Revision: https://phabricator.services.mozilla.com/D185458 --- .../url_patterns_tentative.py.ini | 28 +++ .../network/add_intercept/add_intercept.py | 173 +++++++++++++ .../bidi/network/add_intercept/phases.py | 124 +++++++++ .../add_intercept/url_patterns_tentative.py | 235 ++++++++++++++++++ .../webdriver/tests/bidi/network/conftest.py | 29 +++ .../bidi/network/remove_intercept/invalid.py | 4 +- .../remove_intercept/remove_intercept.py | 99 ++++++++ .../tests/bidi/network/support/other.txt | 1 + 8 files changed, 691 insertions(+), 2 deletions(-) create mode 100644 testing/web-platform/meta/webdriver/tests/bidi/network/add_intercept/url_patterns_tentative.py.ini create mode 100644 testing/web-platform/tests/webdriver/tests/bidi/network/add_intercept/add_intercept.py create mode 100644 testing/web-platform/tests/webdriver/tests/bidi/network/add_intercept/phases.py create mode 100644 testing/web-platform/tests/webdriver/tests/bidi/network/add_intercept/url_patterns_tentative.py create mode 100644 testing/web-platform/tests/webdriver/tests/bidi/network/remove_intercept/remove_intercept.py create mode 100644 testing/web-platform/tests/webdriver/tests/bidi/network/support/other.txt diff --git a/testing/web-platform/meta/webdriver/tests/bidi/network/add_intercept/url_patterns_tentative.py.ini b/testing/web-platform/meta/webdriver/tests/bidi/network/add_intercept/url_patterns_tentative.py.ini new file mode 100644 index 000000000000..864050985d6c --- /dev/null +++ b/testing/web-platform/meta/webdriver/tests/bidi/network/add_intercept/url_patterns_tentative.py.ini @@ -0,0 +1,28 @@ +[url_patterns_tentative.py] + [test_pattern_patterns_matching[patterns7-https://{wpt_host}:1234/\]] + bug: 1849743 + expected: FAIL + + [test_pattern_patterns_matching[patterns8-https://{wpt_host}/path\]] + bug: 1849743 + expected: FAIL + + [test_pattern_patterns_matching[patterns11-https://{wpt_host}:1234/\]] + bug: 1849743 + expected: FAIL + + [test_pattern_patterns_matching[patterns12-https://{wpt_host}/path\]] + bug: 1849743 + expected: FAIL + + [test_pattern_patterns_matching[patterns16-https://{wpt_host}:1234/\]] + bug: 1849743 + expected: FAIL + + [test_pattern_patterns_matching[patterns17-https://{wpt_host}/path\]] + bug: 1849743 + expected: FAIL + + [test_pattern_patterns_matching[patterns25-https://{wpt_host}/\]] + bug: 1849743 + expected: FAIL diff --git a/testing/web-platform/tests/webdriver/tests/bidi/network/add_intercept/add_intercept.py b/testing/web-platform/tests/webdriver/tests/bidi/network/add_intercept/add_intercept.py new file mode 100644 index 000000000000..b68cf2789b93 --- /dev/null +++ b/testing/web-platform/tests/webdriver/tests/bidi/network/add_intercept/add_intercept.py @@ -0,0 +1,173 @@ +import asyncio +import uuid + +import pytest +from webdriver.bidi.modules.script import ScriptEvaluateResultException + +from .. import ( + assert_before_request_sent_event, + assert_response_event, +) + +PAGE_EMPTY_HTML = "/webdriver/tests/bidi/network/support/empty.html" +PAGE_EMPTY_TEXT = "/webdriver/tests/bidi/network/support/empty.txt" +PAGE_OTHER_TEXT = "/webdriver/tests/bidi/network/support/other.txt" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("phase", ["beforeRequestSent", "responseStarted"]) +async def test_other_context( + bidi_session, + wait_for_event, + url, + top_context, + add_intercept, + fetch, + setup_network_test, + phase, +): + # Subscribe to network events only in top_context + network_events = await setup_network_test( + events=[ + "network.beforeRequestSent", + "network.responseStarted", + "network.responseCompleted", + ], + contexts=[top_context["context"]], + ) + + # Create another tab, where network events are not monitored. + other_context = await bidi_session.browsing_context.create(type_hint="tab") + await bidi_session.browsing_context.navigate( + context=other_context["context"], url=url(PAGE_EMPTY_HTML), wait="complete" + ) + + # Add an intercept. + text_url = url(PAGE_EMPTY_TEXT) + intercept = await add_intercept( + phases=["beforeRequestSent"], + url_patterns=[{"type": "string", "pattern": text_url}], + ) + + # Request to top_context should be blocked and throw a ScriptEvaluateResultException + # from the AbortController. + with pytest.raises(ScriptEvaluateResultException): + await fetch(text_url, context=top_context) + + # Request to other_context should not be blocked. + await fetch(text_url, context=other_context) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("phase", ["beforeRequestSent", "responseStarted"]) +async def test_other_url( + bidi_session, + wait_for_event, + url, + top_context, + add_intercept, + fetch, + setup_network_test, + phase, +): + network_events = await setup_network_test( + events=[ + "network.beforeRequestSent", + "network.responseStarted", + "network.responseCompleted", + ], + ) + + # Add an intercept. + text_url = url(PAGE_EMPTY_TEXT) + intercept = await add_intercept( + phases=[phase], + url_patterns=[{"type": "string", "pattern": text_url}], + ) + + # Request to PAGE_EMPTY_TEXT should be blocked and throw a ScriptEvaluateResultException + # from the AbortController. + with pytest.raises(ScriptEvaluateResultException): + await fetch(text_url) + + # Request to PAGE_OTHER_TEXT should not be blocked. + await fetch(url(PAGE_OTHER_TEXT)) + + +@pytest.mark.asyncio +async def test_return_value(bidi_session, add_intercept): + intercept = await add_intercept(phases=["beforeRequestSent"], url_patterns=[]) + + assert isinstance(intercept, str) + uuid.UUID(hex=intercept) + + +@pytest.mark.asyncio +async def test_two_intercepts( + bidi_session, + wait_for_event, + url, + top_context, + add_intercept, + fetch, + setup_network_test, +): + network_events = await setup_network_test( + events=[ + "network.beforeRequestSent", + "network.responseStarted", + "network.responseCompleted", + ], + ) + + # Add a string intercept to catch requests to PAGE_EMPTY_TEXT. + text_url = url(PAGE_EMPTY_TEXT) + string_intercept = await add_intercept( + phases=["beforeRequestSent"], + url_patterns=[{"type": "string", "pattern": text_url}], + ) + # Add a second intercept to catch all requests. + global_intercept = await add_intercept( + phases=["beforeRequestSent"], + url_patterns=[], + ) + + # Perform a request to PAGE_EMPTY_TEXT, which should match both intercepts + on_network_event = wait_for_event("network.beforeRequestSent") + asyncio.ensure_future(fetch(text_url)) + event = await on_network_event + + assert_before_request_sent_event( + event, is_blocked=True, intercepts=[string_intercept, global_intercept] + ) + + # Perform a request to PAGE_OTHER_TEXT, which should only match one intercept + other_url = url(PAGE_OTHER_TEXT) + + on_network_event = wait_for_event("network.beforeRequestSent") + asyncio.ensure_future(fetch(other_url)) + event = await on_network_event + + assert_before_request_sent_event( + event, is_blocked=True, intercepts=[global_intercept] + ) + + # Remove the global intercept, requests to PAGE_OTHER_TEXT should no longer + # be blocked. + await bidi_session.network.remove_intercept(intercept=global_intercept) + await fetch(other_url) + + # Requests to PAGE_EMPTY_TEXT should still be blocked, but only by one + # intercept. + on_network_event = wait_for_event("network.beforeRequestSent") + asyncio.ensure_future(fetch(text_url)) + event = await on_network_event + + assert_before_request_sent_event( + event, is_blocked=True, intercepts=[string_intercept] + ) + + # Remove the string intercept, requests to PAGE_EMPTY_TEXT should no longer + # be blocked. + await bidi_session.network.remove_intercept(intercept=string_intercept) + await fetch(text_url) diff --git a/testing/web-platform/tests/webdriver/tests/bidi/network/add_intercept/phases.py b/testing/web-platform/tests/webdriver/tests/bidi/network/add_intercept/phases.py new file mode 100644 index 000000000000..7dd13a46ac22 --- /dev/null +++ b/testing/web-platform/tests/webdriver/tests/bidi/network/add_intercept/phases.py @@ -0,0 +1,124 @@ +import asyncio + +import pytest +from webdriver.bidi.modules.script import ScriptEvaluateResultException + +from .. import ( + assert_before_request_sent_event, + assert_response_event, +) + +PAGE_EMPTY_TEXT = "/webdriver/tests/bidi/network/support/empty.txt" + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "phases, intercepted_phase", + [ + (["beforeRequestSent"], "beforeRequestSent"), + (["responseStarted"], "responseStarted"), + (["beforeRequestSent", "responseStarted"], "beforeRequestSent"), + (["responseStarted", "beforeRequestSent"], "beforeRequestSent"), + (["beforeRequestSent", "beforeRequestSent"], "beforeRequestSent"), + ], +) +async def test_request_response_phases( + bidi_session, + wait_for_event, + url, + setup_network_test, + add_intercept, + fetch, + phases, + intercepted_phase, +): + network_events = await setup_network_test( + events=[ + "network.beforeRequestSent", + "network.responseStarted", + "network.responseCompleted", + ] + ) + before_request_sent_events = network_events["network.beforeRequestSent"] + response_started_events = network_events["network.responseStarted"] + response_completed_events = network_events["network.responseCompleted"] + + text_url = url(PAGE_EMPTY_TEXT) + intercept = await add_intercept( + phases=phases, + url_patterns=[{"type": "string", "pattern": text_url}], + ) + + assert isinstance(intercept, str) + + on_network_event = wait_for_event(f"network.{intercepted_phase}") + + # Request to top_context should be blocked and throw a ScriptEvaluateResultException + # from the AbortController. + with pytest.raises(ScriptEvaluateResultException): + await fetch(text_url) + + await on_network_event + expected_request = {"method": "GET", "url": text_url} + + if intercepted_phase == "beforeRequestSent": + assert len(before_request_sent_events) == 1 + assert len(response_started_events) == 0 + assert_before_request_sent_event( + before_request_sent_events[0], + expected_request=expected_request, + is_blocked=True, + intercepts=[intercept], + ) + elif intercepted_phase == "responseStarted": + assert len(before_request_sent_events) == 1 + assert len(response_started_events) == 1 + assert_before_request_sent_event( + before_request_sent_events[0], + expected_request=expected_request, + is_blocked=False, + ) + assert_response_event( + response_started_events[0], + expected_request=expected_request, + is_blocked=True, + intercepts=[intercept], + ) + + # Check that we did not receive response completed events. + assert len(response_completed_events) == 0 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("phase", ["beforeRequestSent", "responseStarted"]) +async def test_not_listening_to_phase_event( + bidi_session, + wait_for_event, + url, + top_context, + setup_network_test, + add_intercept, + fetch, + phase, +): + events = [ + "network.beforeRequestSent", + "network.responseStarted", + "network.responseCompleted", + ] + + # Remove the event corresponding to the intercept phase from the monitored + # events. + events.remove(f"network.{phase}") + + network_events = await setup_network_test(events=events) + + # Add an intercept without listening to the corresponding network event + text_url = url(PAGE_EMPTY_TEXT) + intercept = await add_intercept( + phases=[phase], + url_patterns=[{"type": "string", "pattern": text_url}], + ) + + # Request should not be blocked. + await fetch(text_url) diff --git a/testing/web-platform/tests/webdriver/tests/bidi/network/add_intercept/url_patterns_tentative.py b/testing/web-platform/tests/webdriver/tests/bidi/network/add_intercept/url_patterns_tentative.py new file mode 100644 index 000000000000..f32bc8df91d8 --- /dev/null +++ b/testing/web-platform/tests/webdriver/tests/bidi/network/add_intercept/url_patterns_tentative.py @@ -0,0 +1,235 @@ +import asyncio + +import pytest +import pytest_asyncio + +from .. import assert_before_request_sent_event + + +@pytest.fixture +def substitute_host(server_config): + """This test will perform various requests which should not reach the + external network. All strings refering to a domain will define it as a + placeholder which needs to be dynamically replaced by a value from the + current server configuration""" + + def substitute_host(str): + wpt_host = server_config["browser_host"] + return str.format( + wpt_host=wpt_host, + wpt_host_upper=wpt_host.upper(), + ) + + return substitute_host + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "patterns, url_template", + [ + ([], "https://{wpt_host}"), + ([], "https://{wpt_host}/"), + ([], "https://{wpt_host}:1234/"), + ([], "https://{wpt_host}/path"), + ([], "https://{wpt_host}/?search"), + ([{},], "https://{wpt_host}"), + ([{},], "https://{wpt_host}/"), + ([{},], "https://{wpt_host}:1234/"), + ([{},], "https://{wpt_host}/path"), + ([{},], "https://{wpt_host}/?search"), + ([{"protocol": "https"},], "https://{wpt_host}/"), + ([{"protocol": "https"},], "https://{wpt_host}:1234/"), + ([{"protocol": "https"},], "https://{wpt_host}/path"), + ([{"protocol": "https"},], "https://{wpt_host}/?search"), + ([{"protocol": "HTTPS"},], "https://{wpt_host}/"), + ([{"hostname": "{wpt_host}"},], "https://{wpt_host}/"), + ([{"hostname": "{wpt_host}"},], "https://{wpt_host}:1234/"), + ([{"hostname": "{wpt_host}"},], "https://{wpt_host}/path"), + ([{"hostname": "{wpt_host}"},], "https://{wpt_host}/?search"), + ([{"hostname": "{wpt_host}"},], "https://{wpt_host_upper}/"), + ([{"hostname": "{wpt_host_upper}"},], "https://{wpt_host}/"), + ([{"port": "1234"},], "https://{wpt_host}:1234/"), + ([{"pathname": ""},], "https://{wpt_host}"), + ([{"pathname": ""},], "https://{wpt_host}/"), + ([{"pathname": "path"},], "https://{wpt_host}/path"), + ([{"search": ""},], "https://{wpt_host}/"), + ([{"search": ""},], "https://{wpt_host}/?"), + ([{"search": "search"},], "https://{wpt_host}/?search"), + ], +) +async def test_pattern_patterns_matching( + bidi_session, + wait_for_event, + setup_network_test, + add_intercept, + fetch, + substitute_host, + patterns, + url_template, +): + network_events = await setup_network_test( + events=[ + "network.beforeRequestSent", + ] + ) + + for pattern in patterns: + for key in pattern: + pattern[key] = substitute_host(pattern[key]) + + pattern.update({"type": "pattern"}) + + intercept = await add_intercept(phases=["beforeRequestSent"], url_patterns=patterns) + + on_network_event = wait_for_event("network.beforeRequestSent") + asyncio.ensure_future(fetch(substitute_host(url_template))) + event = await on_network_event + + assert_before_request_sent_event(event, is_blocked=True, intercepts=[intercept]) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "pattern, url_template", + [ + ({"protocol": "http"}, "https://{wpt_host}/"), + ({"hostname": "abc.{wpt_host}"}, "https://{wpt_host}/"), + ({"hostname": "web-platform"}, "https://{wpt_host}/"), + ({"hostname": "web-platform.com"}, "https://{wpt_host}/"), + ({"port": "443"}, "https://{wpt_host}:1234/"), + ({"port": "1234"}, "https://{wpt_host}/"), + ({"pathname": ""}, "https://{wpt_host}/path"), + ({"pathname": "path"}, "https://{wpt_host}/"), + ({"pathname": "path"}, "https://{wpt_host}/path/"), + ({"pathname": "path"}, "https://{wpt_host}/other/path"), + ({"pathname": "path"}, "https://{wpt_host}/path/continued"), + ({"search": ""}, "https://{wpt_host}/?search"), + ({"search": "search"}, "https://{wpt_host}/?other"), + ], +) +async def test_pattern_patterns_not_matching( + bidi_session, + wait_for_event, + setup_network_test, + add_intercept, + fetch, + substitute_host, + pattern, + url_template, +): + network_events = await setup_network_test( + events=[ + "network.beforeRequestSent", + ] + ) + + for key in pattern: + pattern[key] = substitute_host(pattern[key]) + + pattern.update({"type": "pattern"}) + + intercept = await add_intercept(phases=["beforeRequestSent"], url_patterns=[pattern]) + + on_network_event = wait_for_event("network.beforeRequestSent") + asyncio.ensure_future(fetch(substitute_host(url_template))) + event = await on_network_event + + assert_before_request_sent_event(event, is_blocked=False) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "pattern, url_template", + [ + ("https://{wpt_host}/", "https://{wpt_host}/"), + ("https://{wpt_host}", "https://{wpt_host}/"), + ("https://{wpt_host}/", "https://{wpt_host}"), + ("HTTPS://{wpt_host}/", "https://{wpt_host}/"), + ("https://{wpt_host}/", "HTTPS://{wpt_host}/"), + ("https://{wpt_host_upper}/", "https://{wpt_host}/"), + ("https://{wpt_host}/", "https://{wpt_host_upper}/"), + ("https://user:password@{wpt_host}/", "https://{wpt_host}/"), + ("https://{wpt_host}/", "https://{wpt_host}:443/"), + ("https://{wpt_host}:443/", "https://{wpt_host}/"), + ("https://{wpt_host}:443/", "https://{wpt_host}:443/"), + ("https://{wpt_host}:1234/", "https://{wpt_host}:1234/"), + ("https://{wpt_host}/path", "https://{wpt_host}/path"), + ("https://{wpt_host}/?search", "https://{wpt_host}/?search"), + ("https://{wpt_host}/#ref", "https://{wpt_host}/"), + ("https://{wpt_host}/", "https://{wpt_host}/#ref"), + ("https://{wpt_host}/#ref1", "https://{wpt_host}/#ref2"), + ], +) +async def test_string_patterns_matching( + bidi_session, + wait_for_event, + url, + setup_network_test, + add_intercept, + fetch, + substitute_host, + pattern, + url_template, +): + network_events = await setup_network_test( + events=[ + "network.beforeRequestSent", + ] + ) + + intercept = await add_intercept( + phases=["beforeRequestSent"], + url_patterns=[{"type": "string", "pattern": substitute_host(pattern)}], + ) + + on_network_event = wait_for_event("network.beforeRequestSent") + asyncio.ensure_future(fetch(substitute_host(url_template))) + event = await on_network_event + + assert_before_request_sent_event(event, is_blocked=True, intercepts=[intercept]) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "pattern, url_template", + [ + ("https://{wpt_host}/", "https://some.other.host/"), + ("https://{wpt_host}:1234/", "https://{wpt_host}:5678/"), + ("https://{wpt_host}/", "https://{wpt_host}:5678/"), + ("https://{wpt_host}/path", "https://{wpt_host}/other/path"), + ("https://{wpt_host}/path", "https://{wpt_host}/path/continued"), + ("https://{wpt_host}/pathcase", "https://{wpt_host}/PATHCASE"), + ("https://{wpt_host}/?searchcase", "https://{wpt_host}/?SEARCHCASE"), + ("https://{wpt_host}/?key", "https://{wpt_host}/?otherkey"), + ("https://{wpt_host}/?key", "https://{wpt_host}/?key=value"), + ("https://{wpt_host}/?a=b&c=d", "https://{wpt_host}/?c=d&a=b"), + ("https://{wpt_host}/??", "https://{wpt_host}/?"), + ], +) +async def test_string_patterns_not_matching( + bidi_session, + wait_for_event, + setup_network_test, + add_intercept, + fetch, + substitute_host, + pattern, + url_template, +): + network_events = await setup_network_test( + events=[ + "network.beforeRequestSent", + ] + ) + + + intercept = await add_intercept( + phases=["beforeRequestSent"], + url_patterns=[{"type": "string", "pattern": substitute_host(pattern)}], + ) + + on_network_event = wait_for_event("network.beforeRequestSent") + asyncio.ensure_future(fetch(substitute_host(url_template))) + event = await on_network_event + + assert_before_request_sent_event(event, is_blocked=False) diff --git a/testing/web-platform/tests/webdriver/tests/bidi/network/conftest.py b/testing/web-platform/tests/webdriver/tests/bidi/network/conftest.py index 924d5a34c6d4..9f74cc67cc64 100644 --- a/testing/web-platform/tests/webdriver/tests/bidi/network/conftest.py +++ b/testing/web-platform/tests/webdriver/tests/bidi/network/conftest.py @@ -3,6 +3,7 @@ import json import pytest import pytest_asyncio +from webdriver.bidi.error import NoSuchInterceptException from webdriver.bidi.modules.script import ContextTarget RESPONSE_COMPLETED_EVENT = "network.responseCompleted" @@ -10,6 +11,34 @@ RESPONSE_COMPLETED_EVENT = "network.responseCompleted" PAGE_EMPTY_HTML = "/webdriver/tests/bidi/network/support/empty.html" +@pytest_asyncio.fixture +async def add_intercept(bidi_session): + """Add a network intercept for the provided phases and url patterns, and + ensure the intercept is removed at the end of the test.""" + + intercepts = [] + async def add_intercept(phases, url_patterns): + nonlocal intercepts + intercept = await bidi_session.network.add_intercept( + phases=phases, + url_patterns=url_patterns, + ) + intercepts.append(intercept) + + return intercept + + yield add_intercept + + # Remove all added intercepts at the end of the test + for intercept in intercepts: + try: + await bidi_session.network.remove_intercept(intercept=intercept) + except (NoSuchInterceptException): + # Ignore exceptions in case a specific intercept was already removed + # during the test. + pass + + @pytest.fixture def fetch(bidi_session, top_context, configuration): """Perform a fetch from the page of the provided context, default to the diff --git a/testing/web-platform/tests/webdriver/tests/bidi/network/remove_intercept/invalid.py b/testing/web-platform/tests/webdriver/tests/bidi/network/remove_intercept/invalid.py index d6063d22ec39..4b3526bfc37d 100644 --- a/testing/web-platform/tests/webdriver/tests/bidi/network/remove_intercept/invalid.py +++ b/testing/web-platform/tests/webdriver/tests/bidi/network/remove_intercept/invalid.py @@ -16,8 +16,8 @@ async def test_params_intercept_invalid_value(bidi_session, value): await bidi_session.network.remove_intercept(intercept=value) -async def test_params_intercept_removed_intercept(bidi_session): - intercept = await bidi_session.network.add_intercept( +async def test_params_intercept_removed_intercept(bidi_session, add_intercept): + intercept = await add_intercept( phases=["beforeRequestSent"], url_patterns=[{"type": "string", "pattern": "https://example.com"}], ) diff --git a/testing/web-platform/tests/webdriver/tests/bidi/network/remove_intercept/remove_intercept.py b/testing/web-platform/tests/webdriver/tests/bidi/network/remove_intercept/remove_intercept.py new file mode 100644 index 000000000000..dbb439a9e52d --- /dev/null +++ b/testing/web-platform/tests/webdriver/tests/bidi/network/remove_intercept/remove_intercept.py @@ -0,0 +1,99 @@ +# META: timeout=long + +import asyncio + +import pytest +from webdriver.bidi.modules.script import ScriptEvaluateResultException + +from .. import ( + assert_before_request_sent_event, + assert_response_event, +) + +PAGE_EMPTY_HTML = "/webdriver/tests/bidi/network/support/empty.html" +PAGE_EMPTY_TEXT = "/webdriver/tests/bidi/network/support/empty.txt" +PAGE_OTHER_TEXT = "/webdriver/tests/bidi/network/support/other.txt" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("phase", ["beforeRequestSent", "responseStarted"]) +async def test_remove_intercept( + bidi_session, wait_for_event, url, setup_network_test, add_intercept, fetch, phase +): + network_events = await setup_network_test( + events=[ + "network.beforeRequestSent", + "network.responseStarted", + "network.responseCompleted", + ] + ) + before_request_sent_events = network_events["network.beforeRequestSent"] + response_started_events = network_events["network.responseStarted"] + response_completed_events = network_events["network.responseCompleted"] + + text_url = url(PAGE_EMPTY_TEXT) + intercept = await add_intercept( + phases=[phase], + url_patterns=[{"type": "string", "pattern": text_url}], + ) + + on_network_event = wait_for_event(f"network.{phase}") + + # Request to top_context should be blocked and throw a ScriptEvaluateResultException + # from the AbortController. + with pytest.raises(ScriptEvaluateResultException): + await fetch(text_url) + + await on_network_event + + assert len(before_request_sent_events) == 1 + + if phase == "beforeRequestSent": + assert len(response_started_events) == 0 + assert_before_request_sent_event( + before_request_sent_events[0], is_blocked=True, intercepts=[intercept] + ) + elif phase == "responseStarted": + assert len(response_started_events) == 1 + assert_before_request_sent_event( + before_request_sent_events[0], is_blocked=False + ) + assert_response_event( + response_started_events[0], is_blocked=True, intercepts=[intercept] + ) + + # Check that we did not receive response completed events. + assert len(response_completed_events) == 0 + + # Remove the intercept + await bidi_session.network.remove_intercept(intercept=intercept) + + # The next request should not be blocked + on_response_completed = wait_for_event(f"network.responseCompleted") + await fetch(text_url) + await on_response_completed + + # Assert the network events have the expected interception properties + assert len(before_request_sent_events) == 2 + assert_before_request_sent_event(before_request_sent_events[1], is_blocked=False) + + if phase == "beforeRequestSent": + assert len(response_started_events) == 1 + assert_response_event(response_started_events[0], is_blocked=False) + elif phase == "responseStarted": + assert len(response_started_events) == 2 + assert_response_event(response_started_events[1], is_blocked=False) + + assert len(response_completed_events) == 1 + assert_response_event(response_completed_events[0], is_blocked=False) + + +@pytest.mark.asyncio +async def test_return_value(bidi_session, add_intercept): + intercept = await add_intercept( + phases=["beforeRequestSent"], + url_patterns=[], + ) + + result = await bidi_session.network.remove_intercept(intercept=intercept) + assert result == {} diff --git a/testing/web-platform/tests/webdriver/tests/bidi/network/support/other.txt b/testing/web-platform/tests/webdriver/tests/bidi/network/support/other.txt new file mode 100644 index 000000000000..e45c9c2666d4 --- /dev/null +++ b/testing/web-platform/tests/webdriver/tests/bidi/network/support/other.txt @@ -0,0 +1 @@ +other