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
This commit is contained in:
Julian Descottes 2023-09-07 18:44:17 +00:00
parent ca5cafac8e
commit 26cbc19d7d
8 changed files with 691 additions and 2 deletions

View file

@ -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

View file

@ -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)

View file

@ -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)

View file

@ -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)

View file

@ -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

View file

@ -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"}],
)

View file

@ -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 == {}