import pytest
from tests.support.sync import AsyncPoll
from webdriver.error import TimeoutException
pytestmark = pytest.mark.asyncio
USER_PROMPT_CLOSED_EVENT = "browsingContext.userPromptClosed"
USER_PROMPT_OPENED_EVENT = "browsingContext.userPromptOpened"
@pytest.mark.capabilities({"unhandledPromptBehavior": {'default': 'ignore'}})
async def test_unsubscribe(
bidi_session, inline, new_tab, wait_for_event, wait_for_future_safe
):
await bidi_session.session.subscribe(
events=[USER_PROMPT_CLOSED_EVENT, USER_PROMPT_OPENED_EVENT]
)
await bidi_session.session.unsubscribe(events=[USER_PROMPT_CLOSED_EVENT])
on_entry = wait_for_event("browsingContext.userPromptOpened")
await bidi_session.browsing_context.navigate(
context=new_tab["context"],
url=inline("<script>window.alert('test')</script>"),
)
# Wait for the alert to open
await wait_for_future_safe(on_entry)
# Track all received browsingContext.userPromptClosed events in the events array
events = []
async def on_event(method, data):
events.append(data)
remove_listener = bidi_session.add_event_listener(
USER_PROMPT_CLOSED_EVENT, on_event
)
await bidi_session.browsing_context.handle_user_prompt(context=new_tab["context"])
wait = AsyncPoll(bidi_session, timeout=0.5)
with pytest.raises(TimeoutException):
await wait.until(lambda _: len(events) > 0)
remove_listener()
@pytest.mark.capabilities({"unhandledPromptBehavior": {'default': 'ignore'}})
async def test_prompt_type_alert(
bidi_session,
subscribe_events,
inline,
new_tab,
wait_for_event,
wait_for_future_safe,
):
await subscribe_events(events=[USER_PROMPT_CLOSED_EVENT, USER_PROMPT_OPENED_EVENT])
on_prompt_opened = wait_for_event(USER_PROMPT_OPENED_EVENT)
await bidi_session.browsing_context.navigate(
context=new_tab["context"],
url=inline("<script>window.alert('test')</script>"),
)
# Wait for the prompt to open.
await wait_for_future_safe(on_prompt_opened)
on_prompt_closed = wait_for_event(USER_PROMPT_CLOSED_EVENT)
await bidi_session.browsing_context.handle_user_prompt(context=new_tab["context"])
event = await wait_for_future_safe(on_prompt_closed)
assert event == {
"context": new_tab["context"],
"accepted": True,
"type": "alert",
}
@pytest.mark.capabilities({"unhandledPromptBehavior": {'default': 'ignore'}})
@pytest.mark.parametrize("accept", [True, False])
async def test_prompt_type_confirm(
bidi_session,
subscribe_events,
inline,
new_tab,
wait_for_event,
wait_for_future_safe,
accept,
):
await subscribe_events(events=[USER_PROMPT_CLOSED_EVENT, USER_PROMPT_OPENED_EVENT])
on_prompt_opened = wait_for_event(USER_PROMPT_OPENED_EVENT)
await bidi_session.browsing_context.navigate(
context=new_tab["context"],
url=inline("<script>window.confirm('test')</script>"),
)
# Wait for the prompt to open.
await wait_for_future_safe(on_prompt_opened)
on_prompt_closed = wait_for_event(USER_PROMPT_CLOSED_EVENT)
await bidi_session.browsing_context.handle_user_prompt(
context=new_tab["context"], accept=accept
)
event = await wait_for_future_safe(on_prompt_closed)
assert event == {
"context": new_tab["context"],
"accepted": accept,
"type": "confirm",
}
@pytest.mark.capabilities({"unhandledPromptBehavior": {'default': 'ignore'}})
@pytest.mark.parametrize("accept", [True, False])
async def test_prompt_type_prompt(
bidi_session,
subscribe_events,
inline,
new_tab,
wait_for_event,
wait_for_future_safe,
accept,
):
await subscribe_events(events=[USER_PROMPT_CLOSED_EVENT, USER_PROMPT_OPENED_EVENT])
on_prompt_opened = wait_for_event(USER_PROMPT_OPENED_EVENT)
await bidi_session.browsing_context.navigate(
context=new_tab["context"],
url=inline("<script>window.prompt('Enter Your Name: ')</script>"),
)
# Wait for the prompt to open.
await wait_for_future_safe(on_prompt_opened)
on_prompt_closed = wait_for_event(USER_PROMPT_CLOSED_EVENT)
test_user_text = "Test"
await bidi_session.browsing_context.handle_user_prompt(
context=new_tab["context"], accept=accept, user_text=test_user_text
)
event = await wait_for_future_safe(on_prompt_closed)
if accept is True:
assert event == {
"context": new_tab["context"],
"accepted": accept,
"type": "prompt",
"userText": test_user_text,
}
else:
assert event == {
"context": new_tab["context"],
"accepted": accept,
"type": "prompt",
}
@pytest.mark.capabilities({"unhandledPromptBehavior": {'default': 'ignore'}})
async def test_prompt_with_defaults(
bidi_session,
subscribe_events,
inline,
new_tab,
wait_for_event,
wait_for_future_safe,
):
await subscribe_events(events=[USER_PROMPT_CLOSED_EVENT, USER_PROMPT_OPENED_EVENT])
on_prompt_opened = wait_for_event(USER_PROMPT_OPENED_EVENT)
await bidi_session.browsing_context.navigate(
context=new_tab["context"],
url=inline("<script>window.prompt('Enter Your Name: ')</script>"),
)
# Wait for the prompt to open.
await wait_for_future_safe(on_prompt_opened)
on_prompt_closed = wait_for_event(USER_PROMPT_CLOSED_EVENT)
await bidi_session.browsing_context.handle_user_prompt(context=new_tab["context"])
event = await wait_for_future_safe(on_prompt_closed)
assert event == {
"context": new_tab["context"],
"accepted": True,
"type": "prompt",
}
@pytest.mark.capabilities({"unhandledPromptBehavior": {'default': 'ignore'}})
@pytest.mark.parametrize("type_hint", ["tab", "window"])
async def test_subscribe_to_one_context(
bidi_session,
subscribe_events,
inline,
wait_for_event,
wait_for_future_safe,
type_hint,
):
new_context = await bidi_session.browsing_context.create(type_hint=type_hint)
# Subscribe to open events for all contexts.
await subscribe_events(events=[USER_PROMPT_OPENED_EVENT])
# Subscribe to close events for only one context.
await subscribe_events(
events=[USER_PROMPT_CLOSED_EVENT],
contexts=[new_context["context"]],
)
# Track all received browsingContext.userPromptClosed events in the events array
events = []
async def on_event(method, data):
events.append(data)
remove_listener = bidi_session.add_event_listener(
USER_PROMPT_CLOSED_EVENT, on_event
)
on_prompt_opened = wait_for_event(USER_PROMPT_OPENED_EVENT)
another_new_context = await bidi_session.browsing_context.create(
type_hint=type_hint
)
# Open a prompt in the different context.
await bidi_session.browsing_context.navigate(
context=another_new_context["context"],
url=inline("<script>window.alert('second tab')</script>"),
)
await wait_for_future_safe(on_prompt_opened)
await bidi_session.browsing_context.handle_user_prompt(
context=another_new_context["context"]
)
# Make sure we don't receive this event.
wait = AsyncPoll(bidi_session, timeout=0.5)
with pytest.raises(TimeoutException):
await wait.until(lambda _: len(events) > 0)
on_prompt_opened = wait_for_event(USER_PROMPT_OPENED_EVENT)
on_prompt_closed = wait_for_event(USER_PROMPT_CLOSED_EVENT)
# Open a prompt in the subscribed context.
await bidi_session.browsing_context.navigate(
context=new_context["context"],
url=inline("<script>window.alert('first tab')</script>"),
)
await wait_for_future_safe(on_prompt_opened)
await bidi_session.browsing_context.handle_user_prompt(
context=new_context["context"]
)
event = await wait_for_future_safe(on_prompt_closed)
assert event == {
"context": new_context["context"],
"accepted": True,
"type": "alert",
}
remove_listener()
await bidi_session.browsing_context.close(context=new_context["context"])
await bidi_session.browsing_context.close(context=another_new_context["context"])
@pytest.mark.capabilities({"unhandledPromptBehavior": {'default': 'ignore'}})
async def test_iframe(
bidi_session,
new_tab,
inline,
test_origin,
subscribe_events,
wait_for_event,
wait_for_future_safe,
):
await subscribe_events(events=[USER_PROMPT_CLOSED_EVENT, USER_PROMPT_OPENED_EVENT])
on_prompt_opened = wait_for_event(USER_PROMPT_OPENED_EVENT)
on_prompt_closed = wait_for_event(USER_PROMPT_CLOSED_EVENT)
await bidi_session.browsing_context.navigate(
context=new_tab["context"],
url=inline(f"<iframe src='{test_origin}'>"),
wait="complete",
)
contexts = await bidi_session.browsing_context.get_tree(root=new_tab["context"])
assert len(contexts) == 1
assert len(contexts[0]["children"]) == 1
frame = contexts[0]["children"][0]
await bidi_session.browsing_context.navigate(
context=frame["context"],
url=inline("<script>window.alert('in iframe')</script>"),
)
await wait_for_future_safe(on_prompt_opened)
await bidi_session.browsing_context.handle_user_prompt(context=frame["context"])
event = await wait_for_future_safe(on_prompt_closed)
assert event == {
"context": new_tab["context"],
"accepted": True,
"type": "alert",
}