Add CDPConnection.wait_for event helper with tests

Co-authored-by: P4x-ng <P4X-ng@users.noreply.github.com>

Author: cursoragent

IMPLEMENTATION.md

@@ -28,6 +28,7 @@ This implementation addresses the issue requesting I/O support and command multi
 ### 4. Event Handling
 - Events dispatched to an asyncio.Queue
 - Async iterator interface for consumption
+- `wait_for()` helper for waiting on one event type with optional predicate filtering
 - Non-blocking get method available
 - Automatic event parsing using existing event registry
 
@@ -101,8 +102,9 @@ class CDPConnection:
 2. **Command execution** - Success, error, timeout
 3. **Multiplexing** - Multiple concurrent commands
 4. **Event handling** - Async iterator, non-blocking get
-5. **Error handling** - Connection errors, command errors
-6. **Resource cleanup** - Pending commands cancelled on close
+5. **Event waiting helper** - Type matching, predicate filtering, timeout behavior
+6. **Error handling** - Connection errors, command errors
+7. **Resource cleanup** - Pending commands cancelled on close
 
 ### Mock Strategy
 

README.md

@@ -57,7 +57,7 @@ async def main():
     # Connect to a Chrome DevTools Protocol endpoint
     async with CDPConnection("ws://localhost:9222/devtools/page/YOUR_PAGE_ID") as conn:
         # Navigate to a URL
-        frame_id, loader_id, error = await conn.execute(
+        frame_id, *_ = await conn.execute(
             page.navigate(url="https://example.com")
         )
         print(f"Navigated to example.com, frame_id: {frame_id}")
@@ -71,10 +71,24 @@ asyncio.run(main())
 - **JSON-RPC Framing**: Automatic message ID assignment and request/response matching
 - **Command Multiplexing**: Execute multiple commands concurrently with proper tracking
 - **Event Handling**: Async iterator for receiving browser events
+- **Event Waiting Helpers**: Wait for specific events with optional filtering
 - **Error Handling**: Comprehensive error handling with typed exceptions
 
 See the [examples directory](examples/) for more usage patterns.
 
+### Waiting for a specific event
+
+```python
+from cdp import page
+
+async with CDPConnection("ws://localhost:9222/devtools/page/YOUR_PAGE_ID") as conn:
+    await conn.execute(page.enable())
+    await conn.execute(page.navigate(url="https://example.com"))
+
+    load_event = await conn.wait_for(page.LoadEventFired, timeout=5.0)
+    print(load_event.timestamp)
+```
+
 ## Sans-I/O Mode (Original)
 
 For users who prefer to manage their own I/O:
@@ -100,25 +114,6 @@ For detailed API documentation, see:
 - `cdp.<domain>` - Type wrappers for each CDP domain (e.g., `cdp.page`, `cdp.network`, `cdp.runtime`)
 - Each domain module provides types, commands, and events for that CDP domain
 
-## Contributing
-
-We welcome contributions! Please see our [Contributing Guide](CONTRIBUTING.md) for details on:
-
-- Setting up your development environment
-- Running tests and type checking
-- Submitting pull requests
-- Reporting issues
-
-Please also read our [Code of Conduct](CODE_OF_CONDUCT.md) before contributing.
-
-## Security
-
-For information about reporting security vulnerabilities, please see our [Security Policy](SECURITY.md).
-
-## License
-
-This project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.
-
 ## API Reference
 
 The library provides Python wrappers for all Chrome DevTools Protocol domains:
@@ -145,12 +140,16 @@ All CDP types, commands, and events are fully typed with Python type hints, prov
 ## Contributing
 
 We welcome contributions! Please see our [Contributing Guide](CONTRIBUTING.md) for details on:
-- How to report bugs and request features
-- Development setup and workflow
-- Coding standards and testing requirements
-- Pull request process
+- Setting up your development environment
+- Running tests and type checking
+- Submitting pull requests
+- Reporting issues
+
+Please also read our [Code of Conduct](CODE_OF_CONDUCT.md) before contributing.
+
+## Security
 
-For questions or discussions, feel free to open an issue on GitHub.
+For information about reporting security vulnerabilities, please see our [Security Policy](SECURITY.md).
 
 ## License
 

cdp/connection.py

@@ -8,6 +8,7 @@
 
 from __future__ import annotations
 import asyncio
+from collections import deque
 import json
 import logging
 import typing
@@ -25,6 +26,7 @@
 
 
 logger = logging.getLogger(__name__)
+EventT = typing.TypeVar('EventT')
 
 
 class CDPError(Exception):
@@ -96,6 +98,7 @@ def __init__(self, url: str, timeout: float = 30.0):
         self._next_command_id = 1
         self._pending_commands: typing.Dict[int, PendingCommand] = {}
         self._event_queue: asyncio.Queue = asyncio.Queue()
+        self._event_buffer: typing.Deque[typing.Any] = deque()
         self._recv_task: typing.Optional[asyncio.Task] = None
         self._closed = False
 
@@ -216,6 +219,26 @@ async def _handle_event(self, data: T_JSON_DICT) -> None:
             await self._event_queue.put(event)
         except Exception as e:
             logger.error(f"Failed to parse event: {e}")
+
+    def _restore_deferred_events(self, events: typing.List[typing.Any]) -> None:
+        """Put deferred events back at the front of the buffer."""
+        for event in reversed(events):
+            self._event_buffer.appendleft(event)
+
+    async def _next_event(self, timeout: typing.Optional[float] = None) -> typing.Any:
+        """
+        Read the next event from the local buffer or queue.
+
+        Buffered events are always consumed first to preserve ordering for events
+        temporarily skipped by ``wait_for``.
+        """
+        if self._event_buffer:
+            return self._event_buffer.popleft()
+
+        if timeout is None:
+            return await self._event_queue.get()
+
+        return await asyncio.wait_for(self._event_queue.get(), timeout=timeout)
 
     async def execute(
         self,
@@ -311,13 +334,67 @@ async def listen(self) -> typing.AsyncIterator[typing.Any]:
         """
         while not self._closed:
             try:
-                event = await asyncio.wait_for(self._event_queue.get(), timeout=1.0)
+                event = await self._next_event(timeout=1.0)
                 yield event
             except asyncio.TimeoutError:
                 # Check if connection is still alive
                 if self._closed:
                     break
                 continue
+
+    async def wait_for(
+        self,
+        event_type: typing.Type[EventT],
+        timeout: typing.Optional[float] = None,
+        predicate: typing.Optional[typing.Callable[[EventT], bool]] = None,
+    ) -> EventT:
+        """
+        Wait for the next event matching a type (and optional predicate).
+
+        Non-matching events are not discarded; they are restored and remain
+        available to ``listen()`` and ``get_event_nowait()``.
+
+        Args:
+            event_type: Event class to match (e.g. ``page.LoadEventFired``).
+            timeout: Maximum seconds to wait for a matching event.
+            predicate: Optional callback for additional filtering.
+
+        Returns:
+            The first matching event instance.
+
+        Raises:
+            asyncio.TimeoutError: If no matching event arrives in time.
+            CDPConnectionError: If the connection is closed while waiting.
+        """
+        deferred: typing.List[typing.Any] = []
+        deadline: typing.Optional[float] = None
+        if timeout is not None:
+            deadline = asyncio.get_running_loop().time() + timeout
+
+        try:
+            while True:
+                if (
+                    self._closed
+                    and not self._event_buffer
+                    and self._event_queue.empty()
+                ):
+                    raise CDPConnectionError("Connection closed while waiting for event")
+
+                remaining: typing.Optional[float] = None
+                if deadline is not None:
+                    remaining = deadline - asyncio.get_running_loop().time()
+                    if remaining <= 0:
+                        raise asyncio.TimeoutError()
+
+                event = await self._next_event(timeout=remaining)
+                if isinstance(event, event_type) and (
+                    predicate is None or predicate(event)
+                ):
+                    return event
+
+                deferred.append(event)
+        finally:
+            self._restore_deferred_events(deferred)
 
     def get_event_nowait(self) -> typing.Optional[typing.Any]:
         """
@@ -327,6 +404,8 @@ def get_event_nowait(self) -> typing.Optional[typing.Any]:
             A CDP event object, or None if no events are available
         """
         try:
+            if self._event_buffer:
+                return self._event_buffer.popleft()
             return self._event_queue.get_nowait()
         except asyncio.QueueEmpty:
             return None

docs/connection.md

@@ -21,7 +21,7 @@ async def main():
     # Connect using async context manager
     async with CDPConnection("ws://localhost:9222/devtools/page/YOUR_PAGE_ID") as conn:
         # Execute a command
-        frame_id, loader_id, error = await conn.execute(
+        frame_id, *_ = await conn.execute(
             page.navigate(url="https://example.com")
         )
 
@@ -112,6 +112,17 @@ async with CDPConnection(url) as conn:
             print(f"Navigated to {event.frame.url}")
 ```
 
+Wait for one specific event with optional filtering:
+
+```python
+async with CDPConnection(url) as conn:
+    await conn.execute(page.enable())
+    await conn.execute(page.navigate(url="https://example.com"))
+
+    load_event = await conn.wait_for(page.LoadEventFired, timeout=5.0)
+    print(load_event.timestamp)
+```
+
 You can also get events without blocking:
 
 ```python
@@ -161,6 +172,7 @@ class CDPConnection:
     async def close(self) -> None
     async def execute(self, cmd, timeout: Optional[float] = None) -> Any
     async def listen(self) -> AsyncIterator[Any]
+    async def wait_for(self, event_type, timeout: Optional[float] = None, predicate=None) -> Any
     def get_event_nowait(self) -> Optional[Any]
 
     @property

examples/connection_example.py

@@ -27,13 +27,14 @@ async def basic_example():
         await conn.execute(page.enable())
 
         # Navigate to a URL
-        frame_id, loader_id, error = await conn.execute(
+        frame_id, *_ = await conn.execute(
             page.navigate(url="https://example.com")
         )
         print(f"Navigated to example.com, frame_id: {frame_id}")
 
-        # Wait a bit for the page to load
-        await asyncio.sleep(2)
+        # Wait for the next page load event
+        load_event = await conn.wait_for(page.LoadEventFired, timeout=5.0)
+        print(f"Load event timestamp: {load_event.timestamp}")
 
         # Evaluate some JavaScript
         result, exception = await conn.execute(
@@ -61,15 +62,9 @@ async def event_handling_example():
             conn.execute(page.navigate(url="https://example.com"))
         )
 
-        # Listen for events while navigation is in progress
-        event_count = 0
-        async for event in conn.listen():
-            print(f"Received event: {type(event).__name__}")
-            event_count += 1
-            
-            # Stop after receiving a few events
-            if event_count >= 3:
-                break
+        # Wait for a specific event while navigation is in progress
+        load_event = await conn.wait_for(page.LoadEventFired, timeout=10.0)
+        print(f"Received {type(load_event).__name__} at {load_event.timestamp}")
 
         # Wait for navigation to complete
         await nav_task

test/test_connection.py

@@ -287,6 +287,85 @@ async def test_get_event_nowait():
             assert isinstance(received_event, page.LoadEventFired)
 
 
+@pytest.mark.asyncio
+async def test_wait_for_returns_matching_event_and_preserves_non_matching():
+    """wait_for should return a match and keep skipped events available."""
+    mock_ws = MockWebSocket()
+
+    mock_ws.queue_message({
+        "method": "Page.frameStartedLoading",
+        "params": {"frameId": "frame-1"},
+    })
+    mock_ws.queue_message({
+        "method": "Page.loadEventFired",
+        "params": {"timestamp": 123.0},
+    })
+
+    with patch('cdp.connection.websockets.connect', new_callable=AsyncMock) as mock_connect:
+        mock_connect.return_value = mock_ws
+
+        async with CDPConnection("ws://localhost:9222/test") as conn:
+            matched = await conn.wait_for(page.LoadEventFired, timeout=1.0)
+            assert isinstance(matched, page.LoadEventFired)
+            assert matched.timestamp == 123.0
+
+            skipped = conn.get_event_nowait()
+            assert isinstance(skipped, page.FrameStartedLoading)
+            assert skipped.frame_id == page.FrameId("frame-1")
+
+
+@pytest.mark.asyncio
+async def test_wait_for_supports_predicate_filtering():
+    """wait_for should support filtering matching event instances."""
+    mock_ws = MockWebSocket()
+
+    mock_ws.queue_message({
+        "method": "Page.loadEventFired",
+        "params": {"timestamp": 1.0},
+    })
+    mock_ws.queue_message({
+        "method": "Page.loadEventFired",
+        "params": {"timestamp": 2.0},
+    })
+
+    with patch('cdp.connection.websockets.connect', new_callable=AsyncMock) as mock_connect:
+        mock_connect.return_value = mock_ws
+
+        async with CDPConnection("ws://localhost:9222/test") as conn:
+            matched = await conn.wait_for(
+                page.LoadEventFired,
+                timeout=1.0,
+                predicate=lambda event: event.timestamp > 1.5,
+            )
+            assert matched.timestamp == 2.0
+
+            skipped = conn.get_event_nowait()
+            assert isinstance(skipped, page.LoadEventFired)
+            assert skipped.timestamp == 1.0
+
+
+@pytest.mark.asyncio
+async def test_wait_for_timeout_restores_skipped_events():
+    """wait_for timeout should not drop events that did not match."""
+    mock_ws = MockWebSocket()
+
+    mock_ws.queue_message({
+        "method": "Page.frameStartedLoading",
+        "params": {"frameId": "frame-timeout"},
+    })
+
+    with patch('cdp.connection.websockets.connect', new_callable=AsyncMock) as mock_connect:
+        mock_connect.return_value = mock_ws
+
+        async with CDPConnection("ws://localhost:9222/test") as conn:
+            with pytest.raises(asyncio.TimeoutError):
+                await conn.wait_for(page.LoadEventFired, timeout=0.1)
+
+            skipped = conn.get_event_nowait()
+            assert isinstance(skipped, page.FrameStartedLoading)
+            assert skipped.frame_id == page.FrameId("frame-timeout")
+
+
 @pytest.mark.asyncio
 async def test_pending_command_count():
     """Test tracking pending command count."""