From db6fd5fe1a3ed49c96509326edb5dd3dd6a0d839 Mon Sep 17 00:00:00 2001 From: Andrew Pikul Date: Tue, 18 Nov 2025 19:35:48 -0500 Subject: [PATCH 1/4] Add helpers and tests --- .../protocol/devtools_async_helpers.py | 133 ++++++++++++++++++ tests/test_devtools_async_helpers.py | 84 +++++++++++ 2 files changed, 217 insertions(+) create mode 100644 src/choreographer/protocol/devtools_async_helpers.py create mode 100644 tests/test_devtools_async_helpers.py diff --git a/src/choreographer/protocol/devtools_async_helpers.py b/src/choreographer/protocol/devtools_async_helpers.py new file mode 100644 index 00000000..8304cb80 --- /dev/null +++ b/src/choreographer/protocol/devtools_async_helpers.py @@ -0,0 +1,133 @@ +"""Async helper functions for common Chrome DevTools Protocol patterns.""" + +from __future__ import annotations + +import asyncio +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from choreographer import Browser, Tab + + from . import BrowserResponse + + +async def create_and_wait( + browser: Browser, + url: str = "", + *, + timeout: float = 30.0, +) -> Tab: + """ + Create a new tab and wait for it to load. + + Args: + browser: Browser instance + url: URL to navigate to (default: blank page) + timeout: Seconds to wait for page load (default: 30.0) + + Returns: + The created Tab + + """ + tab = await asyncio.wait_for(browser.create_tab(url), timeout=timeout) + temp_session = await tab.create_session() + + try: + load_future = temp_session.subscribe_once("Page.loadEventFired") + await temp_session.send_command("Page.enable") + await temp_session.send_command("Runtime.enable") + + if url: + try: + await asyncio.wait_for(load_future, timeout=timeout) + except (asyncio.TimeoutError, asyncio.CancelledError, TimeoutError): + # Stop the page load when timeout occurs + await temp_session.send_command("Page.stopLoading") + raise + finally: + await tab.close_session(temp_session.session_id) + + return tab + + +async def navigate_and_wait( + tab: Tab, + url: str, + *, + timeout: float = 30.0, +) -> Tab: + """ + Navigate an existing tab to a URL and wait for it to load. + + Args: + tab: Tab to navigate + url: URL to navigate to + timeout: Seconds to wait for page load (default: 30.0) + + Returns: + The Tab after navigation completes + + """ + temp_session = await tab.create_session() + + try: + await temp_session.send_command("Page.enable") + await temp_session.send_command("Runtime.enable") + load_future = temp_session.subscribe_once("Page.loadEventFired") + try: + + async def _freezers(): + await temp_session.send_command("Page.navigate", params={"url": url}) + await load_future + + await asyncio.wait_for(_freezers(), timeout=timeout) + except (asyncio.TimeoutError, asyncio.CancelledError, TimeoutError): + # Stop the navigation when timeout occurs + await temp_session.send_command("Page.stopLoading") + raise + finally: + await tab.close_session(temp_session.session_id) + + return tab + + +async def execute_js_and_wait( + tab: Tab, + expression: str, + *, + timeout: float = 30.0, +) -> BrowserResponse: + """ + Execute JavaScript in a tab and return the result. + + Args: + tab: Tab to execute JavaScript in + expression: JavaScript expression to evaluate + timeout: Seconds to wait for execution (default: 30.0) + + Returns: + Response dict from Runtime.evaluate with 'result' and optional + 'exceptionDetails' + + """ + temp_session = await tab.create_session() + + try: + await temp_session.send_command("Page.enable") + await temp_session.send_command("Runtime.enable") + + response = await asyncio.wait_for( + temp_session.send_command( + "Runtime.evaluate", + params={ + "expression": expression, + "awaitPromise": True, + "returnByValue": True, + }, + ), + timeout=timeout, + ) + + return response + finally: + await tab.close_session(temp_session.session_id) diff --git a/tests/test_devtools_async_helpers.py b/tests/test_devtools_async_helpers.py new file mode 100644 index 00000000..dd5e57c4 --- /dev/null +++ b/tests/test_devtools_async_helpers.py @@ -0,0 +1,84 @@ +import asyncio + +import logistro +import pytest + +from choreographer.protocol.devtools_async_helpers import ( + create_and_wait, + execute_js_and_wait, + navigate_and_wait, +) + +pytestmark = pytest.mark.asyncio(loop_scope="function") + +_logger = logistro.getLogger(__name__) + + +@pytest.mark.asyncio +async def test_create_and_wait(browser): + """Test create_and_wait with both valid data URL and blank URL.""" + _logger.info("testing create_and_wait...") + + # Count tabs before + initial_tab_count = len(browser.tabs) + + # Create a simple HTML page as a data URL + html_content = "

Test Page

" + data_url = f"data:text/html,{html_content}" + + # Test 1: Create tab with data URL - should succeed + tab1 = await create_and_wait(browser, url=data_url, timeout=5.0) + assert tab1 is not None + + # Verify the page loaded correctly using execute_js_and_wait + result = await execute_js_and_wait(tab1, "window.location.href", timeout=5.0) + location = result["result"]["result"]["value"] + assert location.startswith("data:text/html") + + # Test 2: Create tab without URL - should succeed (blank page) + tab2 = await create_and_wait(browser, url="", timeout=5.0) + assert tab2 is not None + + # Verify we have 2 more tabs + final_tab_count = len(browser.tabs) + assert final_tab_count == initial_tab_count + 2 + + # Test 3: Create tab with bad URL that won't load - should timeout + with pytest.raises(asyncio.TimeoutError): + await create_and_wait(browser, url="http://192.0.2.1:9999", timeout=1.0) + + +@pytest.mark.asyncio +async def test_navigate_and_wait(browser): + """Test navigate_and_wait with both valid data URL and bad URL.""" + _logger.info("testing navigate_and_wait...") + # Create two blank tabs first + tab = await browser.create_tab("") + + # Create a data URL with identifiable content + html_content1 = "

Navigation Test 1

" + data_url1 = f"data:text/html,{html_content1}" + + html_content2 = "

Navigation Test 2

" + data_url2 = f"data:text/html,{html_content2}" + + # Test 1: Navigate first tab to valid data URL - should succeed + result_tab1 = await navigate_and_wait(tab, url=data_url1, timeout=5.0) + assert result_tab1 is tab + + # Verify the navigation succeeded using execute_js_and_wait + result = await execute_js_and_wait(tab, "window.location.href", timeout=5.0) + location = result["result"]["result"]["value"] + assert location.startswith("data:text/html") + + # Test 2: Navigate second tab to another valid data URL - should succeed + result_tab2 = await navigate_and_wait(tab, url=data_url2, timeout=5.0) + assert result_tab2 is tab + + # Verify the navigation succeeded + result = await execute_js_and_wait(tab, "window.location.href", timeout=5.0) + location = result["result"]["result"]["value"] + assert location.startswith("data:text/html") + # Test 3: Navigate to bad URL that won't load - should timeout + with pytest.raises(asyncio.TimeoutError): + await navigate_and_wait(tab, url="http://192.0.2.1:9999", timeout=1.0) From cbeb58f932c2394f041606623ffa0f08c1ff4549 Mon Sep 17 00:00:00 2001 From: Andrew Pikul Date: Tue, 18 Nov 2025 19:38:36 -0500 Subject: [PATCH 2/4] Update changelog --- CHANGELOG.txt | 1 + 1 file changed, 1 insertion(+) diff --git a/CHANGELOG.txt b/CHANGELOG.txt index cd77c98a..cbca8b02 100644 --- a/CHANGELOG.txt +++ b/CHANGELOG.txt @@ -1,3 +1,4 @@ +- Add a set of helper functions to await for tab loading and send javascript v1.2.1 - Use custom threadpool for functions that could be running during shutdown: Python's stdlib threadpool isn't available during interpreter shutdown, nor From d5a3edfadea2dd22f5ddef09b55e00a05f48ae1e Mon Sep 17 00:00:00 2001 From: Andrew Pikul Date: Wed, 19 Nov 2025 12:09:58 -0500 Subject: [PATCH 3/4] Adjust poe settings --- pyproject.toml | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index f5c293ae..eefa7ee0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -109,8 +109,9 @@ asyncio_default_fixture_loop_scope = "function" log_cli = false addopts = "--import-mode=append" +# tell poe to use the env we give it, otherwise it detects uv and overrides flags [tool.poe] -executor.type = "virtualenv" +executor.type = "simple" [tool.poe.tasks] test_proc = "pytest --log-level=1 -W error -n auto -v -rfE --capture=fd tests/test_process.py" From b3ba42316860b8a8ab1165981b13a387aad0348a Mon Sep 17 00:00:00 2001 From: Andrew Pikul Date: Wed, 19 Nov 2025 12:46:27 -0500 Subject: [PATCH 4/4] Resolve chrome errata: 1. Chrome's security model isn't fully functional when you boot, so starting tabs right away like choreographer does might sometimes work and sometimes not. In particular, data urls can behave very differently depending on how long you wait to create a tab with one: 2. That only applies to Target.createTarget, data urls are fine in all instances with page.Navigate. 3. Thanks to AI for poitning out there would be a race condition in early boots, would really like to know how it knew that though. Don't see how I ever could have resolved that on my own. 4. Page.navigate will freeze itself on a bad resolution, where as creating a tab will only show the bad resolution if you are awaiting a page loaded event. --- .../protocol/devtools_async_helpers.py | 4 +++- tests/test_devtools_async_helpers.py | 18 +++++++++++++----- 2 files changed, 16 insertions(+), 6 deletions(-) diff --git a/src/choreographer/protocol/devtools_async_helpers.py b/src/choreographer/protocol/devtools_async_helpers.py index 8304cb80..a85ac8f0 100644 --- a/src/choreographer/protocol/devtools_async_helpers.py +++ b/src/choreographer/protocol/devtools_async_helpers.py @@ -29,7 +29,7 @@ async def create_and_wait( The created Tab """ - tab = await asyncio.wait_for(browser.create_tab(url), timeout=timeout) + tab = await browser.create_tab(url) temp_session = await tab.create_session() try: @@ -77,7 +77,9 @@ async def navigate_and_wait( try: async def _freezers(): + # If no resolve, will freeze await temp_session.send_command("Page.navigate", params={"url": url}) + # Can freeze if resolve bad await load_future await asyncio.wait_for(_freezers(), timeout=timeout) diff --git a/tests/test_devtools_async_helpers.py b/tests/test_devtools_async_helpers.py index dd5e57c4..309518ec 100644 --- a/tests/test_devtools_async_helpers.py +++ b/tests/test_devtools_async_helpers.py @@ -14,6 +14,11 @@ _logger = logistro.getLogger(__name__) +# Errata: don't use data urls, whether or not they load is variable +# depends on how long chrome has been open for, how they were entered, +# etc + + @pytest.mark.asyncio async def test_create_and_wait(browser): """Test create_and_wait with both valid data URL and blank URL.""" @@ -23,8 +28,7 @@ async def test_create_and_wait(browser): initial_tab_count = len(browser.tabs) # Create a simple HTML page as a data URL - html_content = "

Test Page

" - data_url = f"data:text/html,{html_content}" + data_url = "chrome://version" # Test 1: Create tab with data URL - should succeed tab1 = await create_and_wait(browser, url=data_url, timeout=5.0) @@ -33,7 +37,7 @@ async def test_create_and_wait(browser): # Verify the page loaded correctly using execute_js_and_wait result = await execute_js_and_wait(tab1, "window.location.href", timeout=5.0) location = result["result"]["result"]["value"] - assert location.startswith("data:text/html") + assert location.startswith(data_url) # Test 2: Create tab without URL - should succeed (blank page) tab2 = await create_and_wait(browser, url="", timeout=5.0) @@ -45,7 +49,7 @@ async def test_create_and_wait(browser): # Test 3: Create tab with bad URL that won't load - should timeout with pytest.raises(asyncio.TimeoutError): - await create_and_wait(browser, url="http://192.0.2.1:9999", timeout=1.0) + await create_and_wait(browser, url="http://192.0.2.1:9999", timeout=0.5) @pytest.mark.asyncio @@ -55,6 +59,10 @@ async def test_navigate_and_wait(browser): # Create two blank tabs first tab = await browser.create_tab("") + # navigating to dataurls seems to be fine right now, + # but if one day you have an error here, + # change to the strategy above + # Create a data URL with identifiable content html_content1 = "

Navigation Test 1

" data_url1 = f"data:text/html,{html_content1}" @@ -81,4 +89,4 @@ async def test_navigate_and_wait(browser): assert location.startswith("data:text/html") # Test 3: Navigate to bad URL that won't load - should timeout with pytest.raises(asyncio.TimeoutError): - await navigate_and_wait(tab, url="http://192.0.2.1:9999", timeout=1.0) + await navigate_and_wait(tab, url="http://192.0.2.1:9999", timeout=0.5)