Add helpers and tests

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

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"

src/choreographer/protocol/devtools_async_helpers.py

@@ -0,0 +1,135 @@
+"""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 browser.create_tab(url)
+    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():
+                # 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)
+        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)

tests/test_devtools_async_helpers.py

@@ -0,0 +1,92 @@
+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__)
+
+
+# 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."""
+    _logger.info("testing create_and_wait...")
+
+    # Count tabs before
+    initial_tab_count = len(browser.tabs)
+
+    # Create a simple HTML page as a data URL
+    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)
+    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_url)
+
+    # 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=0.5)
+
+
+@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("")
+
+    # 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 = "<html><body><h1>Navigation Test 1</h1></body></html>"
+    data_url1 = f"data:text/html,{html_content1}"
+
+    html_content2 = "<html><body><h1>Navigation Test 2</h1></body></html>"
+    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=0.5)