chromium/third_party/blink/web_tests/external/wpt/webdriver/tests/bidi/network/add_intercept/url_patterns.py

import asyncio

import pytest

from .. import assert_before_request_sent_event, 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(
    wait_for_event,
    subscribe_events,
    new_tab,
    add_intercept,
    fetch,
    substitute_host,
    wait_for_future_safe,
    patterns,
    url_template,
):
    await subscribe_events(events=[BEFORE_REQUEST_SENT_EVENT], contexts=[new_tab["context"]])

    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(BEFORE_REQUEST_SENT_EVENT)
    asyncio.ensure_future(fetch(substitute_host(url_template), context=new_tab))
    event = await wait_for_future_safe(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(
    wait_for_event,
    subscribe_events,
    new_tab,
    add_intercept,
    fetch,
    substitute_host,
    wait_for_future_safe,
    pattern,
    url_template,
):
    await subscribe_events(events=[BEFORE_REQUEST_SENT_EVENT], contexts=[new_tab["context"]])

    for key in pattern:
        pattern[key] = substitute_host(pattern[key])

    pattern.update({"type": "pattern"})

    await add_intercept(phases=["beforeRequestSent"], url_patterns=[pattern])

    on_network_event = wait_for_event(BEFORE_REQUEST_SENT_EVENT)
    asyncio.ensure_future(fetch(substitute_host(url_template), context=new_tab))
    event = await wait_for_future_safe(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(
    wait_for_event,
    subscribe_events,
    new_tab,
    add_intercept,
    fetch,
    substitute_host,
    wait_for_future_safe,
    pattern,
    url_template,
):
    await subscribe_events(events=[BEFORE_REQUEST_SENT_EVENT], contexts=[new_tab["context"]])

    intercept = await add_intercept(
        phases=["beforeRequestSent"],
        url_patterns=[{"type": "string", "pattern": substitute_host(pattern)}],
    )

    on_network_event = wait_for_event(BEFORE_REQUEST_SENT_EVENT)
    asyncio.ensure_future(fetch(substitute_host(url_template), context=new_tab))
    event = await wait_for_future_safe(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(
    wait_for_event,
    subscribe_events,
    new_tab,
    add_intercept,
    fetch,
    substitute_host,
    wait_for_future_safe,
    pattern,
    url_template,
):
    await subscribe_events(events=[BEFORE_REQUEST_SENT_EVENT], contexts=[new_tab["context"]])

    await add_intercept(
        phases=["beforeRequestSent"],
        url_patterns=[{"type": "string", "pattern": substitute_host(pattern)}],
    )

    on_network_event = wait_for_event(BEFORE_REQUEST_SENT_EVENT)
    asyncio.ensure_future(fetch(substitute_host(url_template), context=new_tab))
    event = await wait_for_future_safe(on_network_event)

    assert_before_request_sent_event(event, is_blocked=False)