From aa9c50e7b4d3edabeef0a204ff1153482fb8ee91 Mon Sep 17 00:00:00 2001 From: SuhrudhC <152193534+SuhrudhC@users.noreply.github.com> Date: Tue, 6 Oct 2026 10:58:38 -0400 Subject: [PATCH 1/2] Fix zero pool timeout waiting indefinitely --- openaq/core/transport.py | 2 +- tests/unit/test_transport.py | 24 ++++++++++++++++++++++++ 2 files changed, 25 insertions(+), 1 deletion(-) diff --git a/openaq/core/transport.py b/openaq/core/transport.py index bd7baafd..b4f6d121 100644 --- a/openaq/core/transport.py +++ b/openaq/core/transport.py @@ -285,7 +285,7 @@ def acquire(self, host: str, pool_timeout: float | None = None) -> PooledConnect Raises: TimeoutError: If no connection becomes available within ``pool_timeout`` seconds. """ - deadline = (time.monotonic() + pool_timeout) if pool_timeout else None + deadline = time.monotonic() + pool_timeout if pool_timeout is not None else None with self._has_capacity: while True: diff --git a/tests/unit/test_transport.py b/tests/unit/test_transport.py index 352fa41a..633a9e6e 100644 --- a/tests/unit/test_transport.py +++ b/tests/unit/test_transport.py @@ -232,6 +232,30 @@ def test_raises_on_pool_timeout(self): with pytest.raises(TimeoutError, match="Connection pool exhausted"): pool.acquire("api.openaq.org", pool_timeout=0.05) + def test_zero_pool_timeout_raises_without_waiting_when_full(self): + pool = self.make_pool(max_connections=1) + pc = pool.acquire("api.openaq.org") + try: + with mock.patch.object( + pool._has_capacity, + "wait", + side_effect=AssertionError("A zero pool timeout must not wait"), + ) as wait: + with pytest.raises(TimeoutError, match="Connection pool exhausted"): + pool.acquire("api.openaq.org", pool_timeout=0.0) + wait.assert_not_called() + finally: + pool.release(pc, discard=True) + + def test_zero_pool_timeout_acquires_available_capacity(self): + pool = self.make_pool(max_connections=1) + pc = pool.acquire("api.openaq.org", pool_timeout=0.0) + pool.release(pc) + try: + assert pool.acquire("api.openaq.org", pool_timeout=0.0) is pc + finally: + pool.release(pc, discard=True) + def test_close_all_clears_pool(self): pool = self.make_pool() pc = pool.acquire("api.openaq.org") From 8767a1b52fc9a1256d8cd274c1b8e488ed55395b Mon Sep 17 00:00:00 2001 From: SuhrudhC <152193534+SuhrudhC@users.noreply.github.com> Date: Tue, 6 Oct 2026 15:54:12 -0400 Subject: [PATCH 2/2] test: exercise pool timeouts through concurrent client requests --- tests/unit/test_client.py | 73 ++++++++++++++++++++++++++++++++++++++- 1 file changed, 72 insertions(+), 1 deletion(-) diff --git a/tests/unit/test_client.py b/tests/unit/test_client.py index 858af873..e382edea 100644 --- a/tests/unit/test_client.py +++ b/tests/unit/test_client.py @@ -1,6 +1,8 @@ import http import os import platform +import threading +from concurrent.futures import ThreadPoolExecutor from datetime import datetime, timedelta from pathlib import Path from unittest.mock import MagicMock, Mock, mock_open, patch @@ -8,13 +10,16 @@ import pytest from freezegun import freeze_time -from openaq import __version__ +from openaq import TimeoutError, __version__ from openaq.client import OpenAQ, _check_api_key, _get_openaq_config, _has_toml from openaq.core.exceptions import ApiKeyMissingError, RateLimitError from openaq.core.transport import ( DEFAULT_LIMITS, DEFAULT_TIMEOUT, Headers, + Limits, + Timeout, + Transport, ) from .mocks import MockTransport @@ -22,6 +27,72 @@ USER_AGENT = f"openaq-python-{__version__}-{platform.python_version()}" +@pytest.mark.parametrize("pool_timeout", [0.0, None]) +def test_client_pool_timeout_under_concurrent_requests(pool_timeout): + # Custom transport configuration is an internal test hook, not a public + # OpenAQ timeout option. Only the network connection is mocked here. + request_started = threading.Event() + release_response = threading.Event() + pool_wait_started = threading.Event() + raw_response = Mock() + raw_response.status = 200 + raw_response.msg = http.client.HTTPMessage() + raw_response.msg["x-ratelimit-limit"] = "60" + raw_response.msg["x-ratelimit-remaining"] = "59" + raw_response.msg["x-ratelimit-reset"] = "60" + raw_response.read.return_value = ( + b'{"meta":{"name":"openaq-api","website":"/","page":1,' + b'"limit":1000,"found":0},"results":[]}' + ) + + def delayed_response(): + request_started.set() + if not release_response.wait(timeout=5): + raise AssertionError("Test did not release the first response") + return raw_response + + transport = Transport( + timeout=Timeout(pool=pool_timeout), limits=Limits(max_connections=1) + ) + original_wait = transport._pool._has_capacity.wait + + def record_pool_wait(timeout=None): + pool_wait_started.set() + return original_wait(timeout=timeout) + + with patch("openaq.core.transport.http.client.HTTPSConnection") as connection: + connection.return_value.getresponse.side_effect = delayed_response + with patch.object( + transport._pool._has_capacity, "wait", side_effect=record_pool_wait + ): + with OpenAQ( + api_key="f" * 64, + base_url="https://sdk-test.invalid/v3/", + _transport=transport, + ) as client: + with ThreadPoolExecutor(max_workers=2) as executor: + first = executor.submit(client.countries.list) + try: + assert request_started.wait(timeout=2) + second = executor.submit(client.countries.list) + if pool_timeout == 0.0: + with pytest.raises( + TimeoutError, match="Connection pool exhausted" + ): + second.result(timeout=2) + assert not pool_wait_started.is_set() + else: + assert pool_wait_started.wait(timeout=2) + assert not second.done() + finally: + release_response.set() + assert first.result(timeout=2).results == [] + if pool_timeout is None: + assert second.result(timeout=2).results == [] + # A later client request succeeds after the pool is released. + assert client.countries.list().results == [] + + @pytest.mark.parametrize( "api_key, expected_exception", [