diff --git a/Makefile b/Makefile index b35560d..63fdbc6 100644 --- a/Makefile +++ b/Makefile @@ -1,3 +1,3 @@ style: - ruff check --fix + ruff check --select I --fix ruff format diff --git a/src/httpx_pycurl/__init__.py b/src/httpx_pycurl/__init__.py index fa85a02..9bc45b8 100644 --- a/src/httpx_pycurl/__init__.py +++ b/src/httpx_pycurl/__init__.py @@ -1,11 +1,9 @@ from __future__ import annotations -from .curl import AsyncCurl from .sync_transport import PyCurlTransport from .transport import AsyncPyCurlTransport __all__ = [ - "AsyncCurl", "PyCurlTransport", "AsyncPyCurlTransport", ] diff --git a/src/httpx_pycurl/curl.py b/src/httpx_pycurl/curl.py deleted file mode 100644 index 695ac24..0000000 --- a/src/httpx_pycurl/curl.py +++ /dev/null @@ -1,334 +0,0 @@ -""" -Low-level async wrapper around pycurl.CurlMulti. - -AsyncCurl manages a CurlMulti handle and wires socket/timer callbacks -into an asyncio event loop. Caller is responsible for initializing curl -handle and translating response to a higher-level framework; this only -handles transfer lifecycle and completion signaling. -""" - -from __future__ import annotations - -import asyncio -import logging -from dataclasses import dataclass - -import pycurl - -log = logging.getLogger(__name__) - - -@dataclass -class PerformHandle: - """Handle to a transfer in progress. - - Allows non-blocking initiation of a transfer with completion - tracked via a future that can be awaited later. - """ - - curl: pycurl.Curl - completion_future: asyncio.Future[None] - - -class AsyncCurl: - """Manages async execution of pycurl Curl handles via CurlMulti. - - Takes pre-configured Curl objects and drives them to completion using - an asyncio event loop. Uses a non-blocking perform() API that returns - a handle immediately, allowing the caller to decide when to wait for - completion. - """ - - def __init__(self, loop: asyncio.AbstractEventLoop | None = None): - """Initialize AsyncCurl. - - Args: - loop: Optional event loop. If None, detected from get_running_loop(). - AsyncCurl is one-time-use and initializes multi/loop eagerly. - - Raises: - RuntimeError: If no event loop is available and none provided. - """ - # Get loop eagerly - if loop is None: - try: - loop = asyncio.get_running_loop() - except RuntimeError: - raise RuntimeError( - "AsyncCurl requires asyncio event loop; call from async context or " - "pass loop to __init__" - ) - self._loop = loop - self._closed = False - - # Create multi handle immediately - self._multi = pycurl.CurlMulti() - self._multi.setopt(pycurl.M_SOCKETFUNCTION, self._socket_callback) - self._multi.setopt(pycurl.M_TIMERFUNCTION, self._timer_callback) - - # Track in-flight transfers: curl handle -> Future - self._transfers: dict[pycurl.Curl, asyncio.Future] = {} - - # Socket management - self._socket_watch: dict[int, int] = {} - - # Timer management - self._timer_handle: asyncio.TimerHandle | None = None - - def setopt(self, option, value) -> None: - """ - Forward options to our CurlMulti() instance. - """ - return self._multi.setopt(option, value) - - def perform(self, curl: pycurl.Curl) -> PerformHandle: - """Start a transfer without blocking. - - Initiates the transfer and returns a handle immediately. The handle's - completion_future can be awaited later to wait for the transfer to complete. - - Args: - curl: Pre-configured pycurl.Curl handle. - - Returns: - PerformHandle with curl and completion_future. - - Raises: - RuntimeError: If AsyncCurl is closed. - """ - if self._closed: - raise RuntimeError("AsyncCurl is closed") - - # Create future for this transfer - future: asyncio.Future[None] = self._loop.create_future() - self._transfers[curl] = future - - # Add to multi handle (this registers socket callbacks) - self._multi.add_handle(curl) - - # Trigger initial processing - self._drive_socket(pycurl.SOCKET_TIMEOUT, 0) - - # Return handle immediately (non-blocking) - return PerformHandle(curl=curl, completion_future=future) - - async def wait_for_completion(self, handle: PerformHandle) -> pycurl.Curl: - """Wait for a transfer to complete. - - Args: - handle: PerformHandle returned from perform(). - - Returns: - The curl handle on success. - - Raises: - pycurl.error: If the transfer fails. - """ - try: - await handle.completion_future - return handle.curl - except Exception: - # Remove transfer tracking on error - self._transfers.pop(handle.curl, None) - # Don't close the handle - let caller decide - raise - - async def aclose(self) -> None: - """Close and clean up resources. - - Can be called multiple times safely. - """ - self._closed = True - self._cancel_timer() - self._cleanup_sockets() - - if self._multi is not None: - # Remove all handles before closing multi - for curl in list(self._transfers.keys()): - try: - self._multi.remove_handle(curl) - except Exception: - pass - # Just clean up the multi object - try: - self._multi.close() - except Exception as e: - log.exception("Error closing CurlMulti: %s", e) - self._multi = None - - self._transfers.clear() - - async def __aenter__(self) -> AsyncCurl: - """Async context manager entry.""" - return self - - async def __aexit__(self, exc_type, exc_val, exc_tb) -> None: - """Async context manager exit.""" - await self.aclose() - - def _register_socket(self, fd: int, what: int) -> None: - """Register socket with event loop based on event mask.""" - current_mode = self._socket_watch.get(fd) - - if what == pycurl.POLL_REMOVE: - if current_mode is not None: - try: - self._loop.remove_reader(fd) - except (ValueError, RuntimeError): - pass - try: - self._loop.remove_writer(fd) - except (ValueError, RuntimeError): - pass - self._socket_watch.pop(fd, None) - return - - # Skip if mode hasn't changed - if current_mode == what: - return - - # Clean up previous registration (only necessary if mode changed) - try: - self._loop.remove_reader(fd) - except (ValueError, RuntimeError): - pass - try: - self._loop.remove_writer(fd) - except (ValueError, RuntimeError): - pass - - self._socket_watch[fd] = what - - try: - if what in {pycurl.POLL_IN, pycurl.POLL_INOUT}: - self._loop.add_reader(fd, self._on_socket_readable, fd) - if what in {pycurl.POLL_OUT, pycurl.POLL_INOUT}: - self._loop.add_writer(fd, self._on_socket_writable, fd) - except (OSError, ValueError, RuntimeError) as e: - log.warning("Failed to register socket %d: %s", fd, e) - self._socket_watch.pop(fd, None) - try: - self._loop.remove_reader(fd) - except (ValueError, RuntimeError): - pass - try: - self._loop.remove_writer(fd) - except (ValueError, RuntimeError): - pass - - def _socket_callback( - self, what: int, fd: int, multi: pycurl.CurlMulti, socketp: object - ) -> int: - """Called by libcurl when socket status changes.""" - self._register_socket(fd, what) - return 0 - - def _timer_callback(self, timeout_ms: int) -> int: - """Called by libcurl to schedule next timeout.""" - if self._closed: - log.debug("_timer_callback(timeout_ms=%s) after close.", timeout_ms) - self._schedule_timeout(timeout_ms) - return 0 - - def _cleanup_sockets(self) -> None: - """Unregister all sockets from event loop.""" - for fd in list(self._socket_watch): - try: - self._loop.remove_reader(fd) - self._loop.remove_writer(fd) - except Exception: - pass - - self._socket_watch.clear() - - def _schedule_timeout(self, timeout_ms: int) -> None: - """Schedule or reschedule a timeout callback.""" - self._cancel_timer() - - if timeout_ms < 0: - return - - if timeout_ms == 0: - self._loop.call_soon(self._on_timeout) - else: - self._timer_handle = self._loop.call_later( - timeout_ms / 1000.0, self._on_timeout - ) - - def _cancel_timer(self) -> None: - """Cancel any pending timer.""" - if self._timer_handle is not None: - self._timer_handle.cancel() - self._timer_handle = None - - def _on_socket_readable(self, fd: int) -> None: - """Called when socket is readable.""" - self._drive_socket(fd, pycurl.CSELECT_IN) - - def _on_socket_writable(self, fd: int) -> None: - """Called when socket is writable.""" - self._drive_socket(fd, pycurl.CSELECT_OUT) - - def _on_timeout(self) -> None: - """Called when timer expires.""" - self._timer_handle = None - self._drive_socket(pycurl.SOCKET_TIMEOUT, 0) - - def _drive_socket(self, sock_fd: int, event_mask: int) -> None: - """Process socket activity in the multi handle.""" - # Call socket_action until no more immediate work - if self._closed: - # normal on shutdown, ignore; curl cleans its own fd's. - # .socket_action() can no longer succeed. - log.debug( - "_drive_socket(sock_fd=%s, event_mask=%s) after close", - sock_fd, - event_mask, - ) - return - - while True: - status, _running = self._multi.socket_action(sock_fd, event_mask) - if status != pycurl.E_CALL_MULTI_PERFORM: - break - - # Drain completed transfers - self._drain_info_read() - - def _drain_info_read(self) -> None: - """Process completed and failed transfers from multi handle.""" - while True: - queued, successful, failed = self._multi.info_read() - - for curl in successful: - self._complete_transfer(curl, None, None) - - for curl, code, message in failed: - self._complete_transfer(curl, code, message) - - if queued == 0: - break - - def _complete_transfer( - self, curl: pycurl.Curl, error_code: int | None, error_message: str | None - ) -> None: - """Mark a transfer as complete (success or failure).""" - future = self._transfers.pop(curl, None) - if future is None: - return - - if error_code is not None: - # Transfer failed - exc = pycurl.error(error_code, error_message or "Unknown error") - if not future.done(): - future.set_exception(exc) - else: - # Transfer succeeded - if not future.done(): - future.set_result(None) - - # Remove from multi handle - try: - self._multi.remove_handle(curl) - except Exception as e: - log.warning("Error removing handle from multi: %s", e) diff --git a/src/httpx_pycurl/transport.py b/src/httpx_pycurl/transport.py index 6cad996..7bf8eea 100644 --- a/src/httpx_pycurl/transport.py +++ b/src/httpx_pycurl/transport.py @@ -10,8 +10,7 @@ import certifi import httpx import pycurl - -from .curl import AsyncCurl, PerformHandle +from pycurl import AsyncCurlMulti if TYPE_CHECKING: from collections.abc import AsyncIterator, Iterable, Iterator @@ -425,12 +424,12 @@ def __init__( self._cainfo = cainfo or certifi.where() self._stream_response = stream_response - self._curl: AsyncCurl | None = None + self._multi: AsyncCurlMulti | None = None self._closed = False self._loop: asyncio.AbstractEventLoop | None = None - # Track in-flight transfers - self._transfers: dict[pycurl.Curl, _Transfer] = {} + # Track in-flight transfers: curl handle -> (future, _Transfer) + self._transfers: dict[pycurl.Curl, tuple[asyncio.Future, _Transfer]] = {} async def handle_async_request(self, request: httpx.Request) -> httpx.Response: if self._closed: @@ -451,11 +450,11 @@ async def handle_async_request(self, request: httpx.Request) -> httpx.Response: "AsyncPyCurlTransport must be used from one event loop" ) - # Lazily initialize AsyncCurl on first request - if self._curl is None: - self._curl = AsyncCurl(loop) + # Lazily initialize AsyncCurlMulti on first request + if self._multi is None: + self._multi = AsyncCurlMulti() # since curl 7.30.0 (2013): - self._curl.setopt(pycurl.M_MAX_TOTAL_CONNECTIONS, self._max_connections) + self._multi.setopt(pycurl.M_MAX_TOTAL_CONNECTIONS, self._max_connections) context = _TransferContext( response_body=SpooledTemporaryFile(max_size=1024 * 1024) @@ -493,19 +492,18 @@ async def handle_async_request(self, request: httpx.Request) -> httpx.Response: ) # Start transfer (non-blocking, returns immediately) - handle = self._curl.perform(curl) - self._transfers[curl] = _Transfer( - request, context, handle.completion_future - ) + future = self._multi.add_handle(curl) + self._transfers[curl] = (future, _Transfer(request, context, future)) # --- Streaming path: return as soon as headers arrive --- if async_stream is not None: # Race the headers-ready event against the raw completion future. # We do NOT wrap the completion future in wait_for_completion here # because cancelling that coroutine would discard the future's result. + future, _ = self._transfers[curl] headers_task = asyncio.ensure_future(context.headers_ready.wait()) await asyncio.wait( - [headers_task, handle.completion_future], + [headers_task, future], return_when=asyncio.FIRST_COMPLETED, ) # Always cancel / clean up the headers_task; it's ephemeral. @@ -515,12 +513,12 @@ async def handle_async_request(self, request: httpx.Request) -> httpx.Response: except (asyncio.CancelledError, Exception): pass - if handle.completion_future.done(): + if future.done(): # Transfer finished before (or at same time as) headers. self._transfers.pop(curl, None) perform_error: httpx.TransportError | None = None try: - handle.completion_future.result() + future.result() except pycurl.error as error: code, message = error.args perform_error = _map_pycurl_error(code, str(message), curl) @@ -532,10 +530,10 @@ async def handle_async_request(self, request: httpx.Request) -> httpx.Response: raise perform_error else: # Headers ready; transfer still running. - # _finish_streaming takes ownership of the handle. + # _finish_streaming takes ownership of the future. _finalize_transfer(curl, context) asyncio.ensure_future( - self._finish_streaming(handle, context, async_stream, curl) + self._finish_streaming(future, context, async_stream, curl) ) return httpx.Response( @@ -550,7 +548,8 @@ async def handle_async_request(self, request: httpx.Request) -> httpx.Response: # --- Non-streaming path: buffer everything, then return --- perform_error = None try: - await self._curl.wait_for_completion(handle) + future, _ = self._transfers[curl] + await future except pycurl.error as error: code, message = error.args perform_error = _map_pycurl_error(code, str(message), curl) @@ -590,7 +589,7 @@ async def handle_async_request(self, request: httpx.Request) -> httpx.Response: async def _finish_streaming( self, - handle: PerformHandle, + future: asyncio.Future, context: _TransferContext, async_stream: _AsyncQueueStream, curl: pycurl.Curl, @@ -601,7 +600,7 @@ async def _finish_streaming( consumer of async_stream unblocks. Handles errors by queuing them. """ try: - await self._curl.wait_for_completion(handle) + await future except pycurl.error as error: code, message = error.args perform_error = _map_pycurl_error(code, str(message), curl) @@ -620,12 +619,17 @@ async def aclose(self): return self._closed = True - # Close AsyncCurl if it was initialized - if self._curl is not None: - await self._curl.aclose() + # Close AsyncCurlMulti if it was initialized + if self._multi is not None: + try: + await self._multi.aclose() + except Exception: + pass + self._multi = None # Clean up any remaining transfers - for curl, transfer in list(self._transfers.items()): + for curl, (future, transfer) in list(self._transfers.items()): + future.cancel() transfer.context.response_body.close() try: curl.close() diff --git a/tests/nginx/nginx.conf b/tests/nginx/nginx.conf index f4bb761..f91fcf4 100644 --- a/tests/nginx/nginx.conf +++ b/tests/nginx/nginx.conf @@ -232,8 +232,9 @@ http { ssl_session_cache shared:SSL:10m; ssl_session_timeout 10m; + # All tests use :8443 and none use the http:// version! server { - listen localhost:8080; + listen localhost:8081; server_name localhost; location / { diff --git a/tests/test_curl.py b/tests/test_curl.py deleted file mode 100644 index f86894d..0000000 --- a/tests/test_curl.py +++ /dev/null @@ -1,471 +0,0 @@ -"""Tests for low-level AsyncCurl implementation.""" - -from __future__ import annotations - -import asyncio -import logging -from unittest.mock import MagicMock, patch - -import pycurl -import pytest - -from httpx_pycurl.curl import AsyncCurl - - -@pytest.mark.asyncio -async def test_asynccurl_init_with_event_loop(): - """Test AsyncCurl initialization with explicit event loop.""" - loop = asyncio.get_event_loop() - curl = AsyncCurl(loop=loop) - assert curl._loop is loop - assert curl._multi is not None - assert not curl._closed - await curl.aclose() - - -@pytest.mark.asyncio -async def test_asynccurl_init_without_event_loop(): - """Test AsyncCurl initialization detects running loop.""" - curl = AsyncCurl() - assert curl._loop is asyncio.get_running_loop() - assert curl._multi is not None - await curl.aclose() - - -def test_asynccurl_init_no_running_loop(): - """Test AsyncCurl initialization fails without event loop outside async context.""" - with pytest.raises(RuntimeError, match="AsyncCurl requires asyncio event loop"): - AsyncCurl() - - -@pytest.mark.asyncio -async def test_asynccurl_context_manager(): - """Test AsyncCurl as async context manager.""" - async with AsyncCurl() as curl: - assert curl._loop is asyncio.get_running_loop() - assert curl._multi is not None - assert not curl._closed - - assert curl._closed - - -@pytest.mark.asyncio -async def test_asynccurl_aclose_idempotent(): - """Test aclose can be called multiple times safely.""" - curl = AsyncCurl() - await curl.aclose() - await curl.aclose() # Should not raise - assert curl._closed - - -@pytest.mark.asyncio -async def test_asynccurl_perform_when_closed(): - """Test perform raises RuntimeError when closed.""" - curl = AsyncCurl() - await curl.aclose() - - handle = pycurl.Curl() - with pytest.raises(RuntimeError, match="AsyncCurl is closed"): - curl.perform(handle) - - -@pytest.mark.asyncio -async def test_asynccurl_socket_callback(): - """Test _socket_callback is registered.""" - curl = AsyncCurl() - # Verify callback is set - assert curl._socket_callback is not None - - # Mock the _register_socket to avoid real socket operations - with patch.object(curl, "_register_socket") as mock_register: - result = curl._socket_callback(pycurl.POLL_REMOVE, -1, curl._multi, None) - assert result == 0 - mock_register.assert_called_once_with(-1, pycurl.POLL_REMOVE) - - await curl.aclose() - - -@pytest.mark.asyncio -async def test_asynccurl_timer_callback(): - """Test _timer_callback is registered.""" - curl = AsyncCurl() - # Verify callback is set - assert curl._timer_callback is not None - result = curl._timer_callback(0) - assert result == 0 - - # Cancel the scheduled timer - curl._cancel_timer() - await curl.aclose() - - -@pytest.mark.asyncio -async def test_asynccurl_perform_with_simple_handle(): - """Test perform with a simple curl handle.""" - curl_obj = AsyncCurl() - - # Create a simple curl handle (doesn't actually make a request) - handle = pycurl.Curl() - - # Set up the handle to do a simple mock operation - handle.setopt(pycurl.URL, "http://httpbin.org/get") - handle.setopt(pycurl.TIMEOUT, 2) - - try: - # perform() returns immediately with a PerformHandle; wait_for_completion() is awaitable - perform_handle = curl_obj.perform(handle) - assert perform_handle.curl is handle - # Wait for the transfer to complete - result = await asyncio.wait_for( - curl_obj.wait_for_completion(perform_handle), timeout=10 - ) - assert result is handle - finally: - await curl_obj.aclose() - handle.close() - - -@pytest.mark.asyncio -async def test_asynccurl_socket_registration_removal(): - """Test socket registration and removal.""" - curl = AsyncCurl() - - # Test registering a socket with POLL_REMOVE removes it - curl._register_socket(999, pycurl.POLL_REMOVE) - assert 999 not in curl._socket_watch - - await curl.aclose() - - -@pytest.mark.asyncio -async def test_asynccurl_cleanup_sockets(): - """Test socket cleanup.""" - curl = AsyncCurl() - - # Add a fake socket watch (we can't register real sockets without real I/O) - curl._socket_watch[999] = pycurl.POLL_IN - - # Clean up should not raise - curl._cleanup_sockets() - assert len(curl._socket_watch) == 0 - - await curl.aclose() - - -@pytest.mark.asyncio -async def test_asynccurl_timer_scheduling(): - """Test timer scheduling and cancellation.""" - curl = AsyncCurl() - - # Schedule a timeout - curl._schedule_timeout(100) - assert curl._timer_handle is not None - - # Cancel it - curl._cancel_timer() - assert curl._timer_handle is None - - # Schedule a zero timeout (call_soon) - curl._schedule_timeout(0) - assert curl._timer_handle is None # call_soon doesn't return a handle - - # Negative timeout should not schedule - curl._schedule_timeout(-1) - assert curl._timer_handle is None - - await curl.aclose() - - -@pytest.mark.asyncio -async def test_asynccurl_drive_socket(): - """Test _drive_socket processes socket actions.""" - curl = AsyncCurl() - - # Mock the multi handle to avoid actual I/O - with patch.object(curl._multi, "socket_action") as mock_action: - with patch.object(curl._multi, "info_read") as mock_info: - mock_action.return_value = (pycurl.E_OK, 0) - mock_info.return_value = (0, [], []) - - # Should not raise - curl._drive_socket(pycurl.SOCKET_TIMEOUT, 0) - - await curl.aclose() - - -@pytest.mark.asyncio -async def test_asynccurl_drain_with_successful_transfers(): - """Test _drain_info_read processes successful transfers.""" - curl_obj = AsyncCurl() - - # Create a mock curl handle - handle = MagicMock() - - # Create a future and add to transfers - future: asyncio.Future[None] = curl_obj._loop.create_future() - curl_obj._transfers[handle] = future - - # Replace multi with mock - mock_multi = MagicMock() - mock_multi.info_read.side_effect = [(1, [handle], []), (0, [], [])] - curl_obj._multi = mock_multi - - curl_obj._drain_info_read() - - # Future should be resolved - assert future.done() - assert future.exception() is None - - # Don't await aclose since we replaced _multi with a mock - curl_obj._closed = True - - -@pytest.mark.asyncio -async def test_asynccurl_drain_with_failed_transfers(): - """Test _drain_info_read processes failed transfers.""" - curl_obj = AsyncCurl() - - # Create a mock curl handle - handle = MagicMock() - - # Create a future and add to transfers - future: asyncio.Future[None] = curl_obj._loop.create_future() - curl_obj._transfers[handle] = future - - # Replace multi with mock - mock_multi = MagicMock() - mock_multi.info_read.side_effect = [ - (1, [], [(handle, 52, "CURLE_GOT_NOTHING")]), - (0, [], []), - ] - curl_obj._multi = mock_multi - - curl_obj._drain_info_read() - - # Future should have exception - assert future.done() - assert isinstance(future.exception(), pycurl.error) - - # Don't await aclose since we replaced _multi with a mock - curl_obj._closed = True - - -@pytest.mark.asyncio -async def test_asynccurl_complete_transfer_unknown_handle(): - """Test _complete_transfer with unknown handle doesn't crash.""" - curl_obj = AsyncCurl() - handle = MagicMock() - - # Should not raise even if handle is not tracked - curl_obj._complete_transfer(handle, None, None) - - await curl_obj.aclose() - - -@pytest.mark.asyncio -async def test_asynccurl_on_socket_readable(): - """Test _on_socket_readable calls _drive_socket.""" - curl = AsyncCurl() - - with patch.object(curl, "_drive_socket") as mock_drive: - curl._on_socket_readable(999) - mock_drive.assert_called_once_with(999, pycurl.CSELECT_IN) - - await curl.aclose() - - -@pytest.mark.asyncio -async def test_asynccurl_on_socket_writable(): - """Test _on_socket_writable calls _drive_socket.""" - curl = AsyncCurl() - - with patch.object(curl, "_drive_socket") as mock_drive: - curl._on_socket_writable(999) - mock_drive.assert_called_once_with(999, pycurl.CSELECT_OUT) - - await curl.aclose() - - -@pytest.mark.asyncio -async def test_asynccurl_on_timeout(): - """Test _on_timeout calls _drive_socket.""" - curl = AsyncCurl() - - with patch.object(curl, "_drive_socket") as mock_drive: - curl._on_timeout() - mock_drive.assert_called_once_with(pycurl.SOCKET_TIMEOUT, 0) - - assert curl._timer_handle is None - await curl.aclose() - - -@pytest.mark.asyncio -async def test_asynccurl_register_socket_poll_in(): - """Test socket registration with POLL_IN.""" - curl = AsyncCurl() - - with patch.object(curl._loop, "add_reader") as mock_add: - with patch.object(curl._loop, "add_writer") as mock_add_w: - curl._register_socket(999, pycurl.POLL_IN) - mock_add.assert_called_once() - mock_add_w.assert_not_called() - - await curl.aclose() - - -@pytest.mark.asyncio -async def test_asynccurl_register_socket_poll_out(): - """Test socket registration with POLL_OUT.""" - curl = AsyncCurl() - - with patch.object(curl._loop, "add_reader") as mock_add: - with patch.object(curl._loop, "add_writer") as mock_add_w: - curl._register_socket(999, pycurl.POLL_OUT) - mock_add.assert_not_called() - mock_add_w.assert_called_once() - - await curl.aclose() - - -@pytest.mark.asyncio -async def test_asynccurl_register_socket_poll_inout(): - """Test socket registration with POLL_INOUT.""" - curl = AsyncCurl() - - with patch.object(curl._loop, "add_reader") as mock_add: - with patch.object(curl._loop, "add_writer") as mock_add_w: - curl._register_socket(999, pycurl.POLL_INOUT) - mock_add.assert_called_once() - mock_add_w.assert_called_once() - - await curl.aclose() - - -@pytest.mark.asyncio -async def test_asynccurl_register_socket_os_error(): - """Test socket registration handles OSError.""" - curl = AsyncCurl() - - with patch.object(curl._loop, "add_reader", side_effect=OSError("mock error")): - with patch.object(curl._loop, "remove_reader"): - with patch.object(curl._loop, "remove_writer"): - # Should not raise - curl._register_socket(999, pycurl.POLL_IN) - assert 999 not in curl._socket_watch - - await curl.aclose() - - -@pytest.mark.asyncio -async def test_asynccurl_aclose_with_multi_error(caplog): - """Test aclose handles errors when closing multi.""" - curl = AsyncCurl() - - # Replace with a mock that raises on close() - mock_multi = MagicMock() - mock_multi.close.side_effect = Exception("mock error") - curl._multi = mock_multi - - with caplog.at_level(logging.ERROR): - await curl.aclose() - - # Check that an exception was logged - assert ( - len([r for r in caplog.records if "Error closing CurlMulti" in r.message]) > 0 - ) - - -@pytest.mark.asyncio -async def test_asynccurl_complete_transfer_with_error_remove_error(caplog): - """Test _complete_transfer handles error removing handle from multi.""" - curl_obj = AsyncCurl() - - handle = MagicMock() - future: asyncio.Future[None] = curl_obj._loop.create_future() - curl_obj._transfers[handle] = future - - with caplog.at_level(logging.WARNING): - with patch.object( - curl_obj._multi, "remove_handle", side_effect=Exception("mock error") - ): - curl_obj._complete_transfer(handle, None, None) - assert "Error removing handle from multi" in caplog.text - - await curl_obj.aclose() - - -@pytest.mark.asyncio -async def test_asynccurl_socket_watch_updated(): - """Test socket watch dictionary is properly maintained.""" - curl = AsyncCurl() - - with patch.object(curl._loop, "add_reader"): - with patch.object(curl._loop, "add_writer"): - curl._register_socket(999, pycurl.POLL_IN) - assert 999 in curl._socket_watch - assert curl._socket_watch[999] == pycurl.POLL_IN - - await curl.aclose() - - -@pytest.mark.asyncio -async def test_asynccurl_multi_socket_action_e_call_multi_perform(): - """Test _drive_socket with multiple socket_action calls.""" - curl = AsyncCurl() - - # Create a mock multi object to replace the real one - mock_multi = MagicMock() - mock_multi.socket_action.side_effect = [ - (pycurl.E_CALL_MULTI_PERFORM, 0), - (pycurl.E_OK, 0), - ] - mock_multi.info_read.return_value = (0, [], []) - - curl._multi = mock_multi - - curl._drive_socket(pycurl.SOCKET_TIMEOUT, 0) - assert mock_multi.socket_action.call_count == 2 - - await curl.aclose() - - -@pytest.mark.asyncio -async def test_asynccurl_complete_transfer_already_done(): - """Test _complete_transfer when future is already done.""" - curl_obj = AsyncCurl() - - handle = MagicMock() - future: asyncio.Future[None] = curl_obj._loop.create_future() - future.set_result(None) # Already done - curl_obj._transfers[handle] = future - - mock_multi = MagicMock() - curl_obj._multi = mock_multi - - # Should not raise, should not set_result again - curl_obj._complete_transfer(handle, None, None) - - curl_obj._closed = True - - -@pytest.mark.asyncio -async def test_asynccurl_cleanup_sockets_with_error(): - """Test _cleanup_sockets handles errors removing sockets.""" - curl = AsyncCurl() - - # Add socket watches - curl._socket_watch[999] = pycurl.POLL_IN - curl._socket_watch[1000] = pycurl.POLL_OUT - - # Make removal raise an exception - with patch.object(curl._loop, "remove_reader", side_effect=Exception("mock error")): - with patch.object( - curl._loop, "remove_writer", side_effect=Exception("mock error") - ): - # Should not raise - curl._cleanup_sockets() - - # Socket watch should still be cleared - assert len(curl._socket_watch) == 0 - - await curl.aclose()