Handle flaky tests a bit better:

Properly close tasks related to each test before ending tests:

- Properly cancel and close all tasks for persistent connection provider tests
- Clear caches as a cleanup for all persistent connection provider tests

Increase dev period to allow `replace_transaction` tests more time:

- Re-running these tests with flaky does not help, hey will end up failing
  every try. Increase the dev period to allow `replace_transaction` tests
  more time and remove the flaky decorator.

- Lower request timeout for tests. No test should need more than 10s.

Author: fselmo

tests/integration/go_ethereum/conftest.py

@@ -94,7 +94,7 @@ def base_geth_command_arguments(geth_binary, datadir):
         datadir,
         "--dev",
         "--dev.period",
-        "1",
+        "2",
         "--password",
         os.path.join(datadir, "keystore", "pw.txt"),
     )

tests/integration/go_ethereum/test_goethereum_http.py

@@ -1,5 +1,8 @@
 import pytest
 
+from aiohttp import (
+    ClientTimeout,
+)
 import pytest_asyncio
 
 from tests.utils import (
@@ -70,7 +73,7 @@ def geth_command_arguments(rpc_port, base_geth_command_arguments, get_geth_versi
 @pytest.fixture(scope="module")
 def w3(geth_process, endpoint_uri):
     wait_for_http(endpoint_uri)
-    return Web3(Web3.HTTPProvider(endpoint_uri))
+    return Web3(Web3.HTTPProvider(endpoint_uri, request_kwargs={"timeout": 10}))
 
 
 class TestGoEthereumWeb3ModuleTest(GoEthereumWeb3ModuleTest):
@@ -119,7 +122,9 @@ class TestGoEthereumTxPoolModuleTest(GoEthereumTxPoolModuleTest):
 @pytest_asyncio.fixture(scope="module")
 async def async_w3(geth_process, endpoint_uri):
     await wait_for_aiohttp(endpoint_uri)
-    _w3 = AsyncWeb3(AsyncHTTPProvider(endpoint_uri))
+    _w3 = AsyncWeb3(
+        AsyncHTTPProvider(endpoint_uri, request_kwargs={"timeout": ClientTimeout(10)})
+    )
     return _w3
 
 

tests/integration/go_ethereum/test_goethereum_ipc.py

@@ -56,7 +56,7 @@ def geth_ipc_path(datadir):
 @pytest.fixture(scope="module")
 def w3(geth_process, geth_ipc_path):
     wait_for_socket(geth_ipc_path)
-    return Web3(Web3.IPCProvider(geth_ipc_path, timeout=30))
+    return Web3(Web3.IPCProvider(geth_ipc_path, timeout=10))
 
 
 class TestGoEthereumWeb3ModuleTest(GoEthereumWeb3ModuleTest):
@@ -101,7 +101,7 @@ def test_admin_start_stop_ws(self, w3: "Web3") -> None:
 @pytest_asyncio.fixture(scope="module")
 async def async_w3(geth_process, geth_ipc_path):
     await wait_for_async_socket(geth_ipc_path)
-    async with AsyncWeb3(AsyncIPCProvider(geth_ipc_path)) as _aw3:
+    async with AsyncWeb3(AsyncIPCProvider(geth_ipc_path, request_timeout=10)) as _aw3:
         yield _aw3
 
 

tests/integration/go_ethereum/test_goethereum_ws/test_async_await_w3.py

@@ -27,9 +27,8 @@
 @pytest_asyncio.fixture(scope="module")
 async def async_w3(geth_process, endpoint_uri):
     await wait_for_aiohttp(endpoint_uri)
-
     # await the persistent connection itself
-    return await AsyncWeb3(WebSocketProvider(endpoint_uri))
+    return await AsyncWeb3(WebSocketProvider(endpoint_uri, request_timeout=10))
 
 
 class TestGoEthereumAsyncWeb3ModuleTest(GoEthereumAsyncWeb3ModuleTest):

tests/integration/go_ethereum/test_goethereum_ws/test_async_ctx_manager_w3.py

@@ -27,9 +27,8 @@
 @pytest_asyncio.fixture(scope="module")
 async def async_w3(geth_process, endpoint_uri):
     await wait_for_aiohttp(endpoint_uri)
-
     # async context manager pattern
-    async with AsyncWeb3(WebSocketProvider(endpoint_uri)) as w3:
+    async with AsyncWeb3(WebSocketProvider(endpoint_uri, request_timeout=10)) as w3:
         yield w3
 
 

tests/integration/go_ethereum/test_goethereum_ws/test_async_iterator_w3.py

@@ -27,9 +27,8 @@
 @pytest_asyncio.fixture(scope="module")
 async def async_w3(geth_process, endpoint_uri):
     await wait_for_aiohttp(endpoint_uri)
-
     # async iterator pattern
-    async for w3 in AsyncWeb3(WebSocketProvider(endpoint_uri)):
+    async for w3 in AsyncWeb3(WebSocketProvider(endpoint_uri, request_timeout=10)):
         return w3
 
 

web3/_utils/module_testing/eth_module.py

@@ -58,7 +58,6 @@
     assert_contains_log,
     async_mock_offchain_lookup_request_response,
     flaky_geth_dev_mining,
-    flaky_with_xfail_on_exception,
     mock_offchain_lookup_request_response,
 )
 from web3._utils.module_testing.utils import (
@@ -77,7 +76,6 @@
     MultipleFailedRequests,
     NameNotFound,
     OffchainLookup,
-    RequestTimedOut,
     TimeExhausted,
     TooManyRequests,
     TransactionNotFound,
@@ -2390,10 +2388,6 @@ async def test_async_eth_replace_transaction_gas_price_too_low(
         with pytest.raises(Web3ValueError):
             await async_w3.eth.replace_transaction(txn_hash, txn_params)
 
-    @flaky_with_xfail_on_exception(
-        reason="Very flaky on CI runs, hard to reproduce locally.",
-        exception=RequestTimedOut,
-    )
     @pytest.mark.asyncio
     async def test_async_eth_replace_transaction_gas_price_defaulting_minimum(
         self, async_w3: "AsyncWeb3", async_keyfile_account_address: ChecksumAddress
@@ -2417,10 +2411,6 @@ async def test_async_eth_replace_transaction_gas_price_defaulting_minimum(
             gas_price * 1.125
         )  # minimum gas price
 
-    @flaky_with_xfail_on_exception(
-        reason="Very flaky on CI runs, hard to reproduce locally.",
-        exception=RequestTimedOut,
-    )
     @pytest.mark.asyncio
     async def test_async_eth_replace_transaction_gas_price_defaulting_strategy_higher(
         self, async_w3: "AsyncWeb3", async_keyfile_account_address: ChecksumAddress
@@ -2436,7 +2426,7 @@ async def test_async_eth_replace_transaction_gas_price_defaulting_strategy_highe
 
         two_gwei_in_wei = async_w3.to_wei(2, "gwei")
 
-        def higher_gas_price_strategy(async_w3: "AsyncWeb3", txn: TxParams) -> Wei:
+        def higher_gas_price_strategy(_async_w3: "AsyncWeb3", _txn: TxParams) -> Wei:
             return two_gwei_in_wei
 
         async_w3.eth.set_gas_price_strategy(higher_gas_price_strategy)
@@ -2449,16 +2439,11 @@ def higher_gas_price_strategy(async_w3: "AsyncWeb3", txn: TxParams) -> Wei:
         )  # Strategy provides higher gas price
         async_w3.eth.set_gas_price_strategy(None)  # reset strategy
 
-    @flaky_with_xfail_on_exception(
-        reason="Very flaky on CI runs, hard to reproduce locally.",
-        exception=RequestTimedOut,
-    )
     @pytest.mark.asyncio
     async def test_async_eth_replace_transaction_gas_price_defaulting_strategy_lower(
         self, async_w3: "AsyncWeb3", async_keyfile_account_address: ChecksumAddress
     ) -> None:
         gas_price = async_w3.to_wei(2, "gwei")
-
         txn_params: TxParams = {
             "from": async_keyfile_account_address,
             "to": async_keyfile_account_address,

web3/_utils/module_testing/persistent_connection_provider.py

@@ -7,6 +7,7 @@
     TYPE_CHECKING,
     Any,
     Dict,
+    Generator,
     List,
     Tuple,
     Union,
@@ -174,7 +175,7 @@ async def emit_contract_event(
     acct: ChecksumAddress,
     contract_function: "AsyncContractFunction",
     args: Any = (),
-    delay: float = 0.1,
+    delay: float = 0.25,
 ) -> None:
     await asyncio.sleep(delay)
     tx_hash = await contract_function(*args).transact({"from": acct})
@@ -213,7 +214,21 @@ def assert_no_subscriptions_left(sub_container: "SubscriptionContainer") -> None
     assert len(sub_container.handler_subscriptions) == 0
 
 
+async def clean_up_task(task: "asyncio.Task[Any]") -> None:
+    task.cancel()
+    try:
+        await task
+    except asyncio.CancelledError:
+        pass
+
+
 class PersistentConnectionProviderTest:
+    @pytest.fixture(autouse=True)
+    def clear_caches(self, async_w3: AsyncWeb3) -> Generator[None, None, None]:
+        yield
+        async_w3.provider._request_processor.clear_caches()
+        async_w3.subscription_manager.total_handler_calls = 0
+
     @staticmethod
     async def seed_transactions_to_geth(
         async_w3: AsyncWeb3,
@@ -318,9 +333,6 @@ async def test_async_eth_subscribe_syncing_mocked(
             async_w3.subscription_manager._subscription_container
         )
 
-        # cleanup
-        async_w3.provider._request_processor.clear_caches()
-
     @pytest.mark.asyncio
     async def test_async_eth_subscribe_new_heads(self, async_w3: AsyncWeb3) -> None:
         sub_id = await async_w3.eth.subscribe("newHeads")
@@ -365,11 +377,8 @@ async def test_async_eth_subscribe_creates_and_handles_new_heads_subscription_ty
         assert sub_manager.total_handler_calls == 1
         assert sub.handler_call_count == 1
 
-        # cleanup
-        sub_manager.total_handler_calls = 0
-
     @pytest.mark.asyncio
-    async def test_async_eth_subscribe_new_and_process_pending_tx_true(
+    async def test_async_eth_subscribe_process_pending_tx_true(
         self,
         async_w3: AsyncWeb3,
     ) -> None:
@@ -409,7 +418,7 @@ async def test_async_eth_subscribe_new_and_process_pending_tx_true(
         )
         async_w3.provider._request_processor.clear_caches()
         await async_w3.eth.wait_for_transaction_receipt(tx_hash)
-        tx_seeder_task.cancel()
+        await clean_up_task(tx_seeder_task)
 
     @pytest.mark.asyncio
     async def test_async_eth_subscribe_and_process_pending_tx_false(
@@ -446,7 +455,7 @@ async def test_async_eth_subscribe_and_process_pending_tx_false(
             async_w3.subscription_manager._subscription_container
         )
         await async_w3.eth.wait_for_transaction_receipt(tx_hash)
-        tx_seeder_task.cancel()
+        await clean_up_task(tx_seeder_task)
 
     @pytest.mark.asyncio
     async def test_async_eth_subscribe_creates_and_handles_pending_tx_subscription_type(
@@ -472,9 +481,8 @@ async def test_async_eth_subscribe_creates_and_handles_pending_tx_subscription_t
         accts = await async_w3.eth.accounts
         acct = accts[0]
         tx_seeder_task = asyncio.create_task(
-            self.seed_transactions_to_geth(async_w3, acct, num_txs=1, delay=0.5)
+            self.seed_transactions_to_geth(async_w3, acct)
         )
-
         await sub_manager.handle_subscriptions()
 
         assert pending_tx_handler_test.passed
@@ -485,7 +493,7 @@ async def test_async_eth_subscribe_creates_and_handles_pending_tx_subscription_t
 
         # cleanup
         sub_manager.total_handler_calls = 0
-        tx_seeder_task.cancel()
+        await clean_up_task(tx_seeder_task)
 
     @pytest.mark.asyncio
     async def test_async_eth_subscribe_and_process_logs(
@@ -523,7 +531,7 @@ async def test_async_eth_subscribe_and_process_logs(
 
         assert await async_w3.eth.unsubscribe(sub_id)
         assert len(async_w3.subscription_manager.subscriptions) == 0
-        emit_event_task.cancel()
+        await clean_up_task(emit_event_task)
 
     @pytest.mark.asyncio
     async def test_async_eth_subscribe_creates_and_handles_logs_subscription_type(
@@ -571,7 +579,7 @@ async def test_async_eth_subscribe_creates_and_handles_logs_subscription_type(
 
         # cleanup
         sub_manager.total_handler_calls = 0
-        emit_event_task.cancel()
+        await clean_up_task(emit_event_task)
 
     @pytest.mark.asyncio
     async def test_async_extradata_poa_middleware_on_eth_subscription(
@@ -731,7 +739,6 @@ async def test_async_subscription_manager_subscribes_to_many_subscriptions(
         subs = sub_manager.subscriptions
 
         await sub_manager.handle_subscriptions()
-        emit_event_task.cancel()
 
         # assert unsubscribed and removed subscriptions
         assert len(sub_manager.subscriptions) == 0
@@ -745,6 +752,7 @@ async def test_async_subscription_manager_subscribes_to_many_subscriptions(
 
         # cleanup
         sub_manager.total_handler_calls = 0
+        await clean_up_task(emit_event_task)
 
     @pytest.mark.asyncio
     async def test_subscription_handler_context(self, async_w3: AsyncWeb3) -> None:
@@ -764,9 +772,10 @@ async def test_sub_handler(
             assert handler_context.str2 == "bar"
 
             handler_context.handler_test.passed = True
-            await handler_context.subscription.unsubscribe()
+            unsubscribed = await handler_context.subscription.unsubscribe()
+            assert unsubscribed
 
-        await async_w3.eth.subscribe(
+        subscribed = await async_w3.eth.subscribe(
             "newHeads",
             label="foo",
             handler=test_sub_handler,
@@ -779,6 +788,7 @@ async def test_sub_handler(
                 "handler_test": handler_test,
             },
         )
+        assert is_hexstr(subscribed)
 
         sub_manager = async_w3.subscription_manager
 
@@ -853,13 +863,17 @@ async def unsubscribe_subs(
         sub_manager = async_w3.subscription_manager
         sub1 = NewHeadsSubscription(label="foo", handler=idle_handler)
         sub2 = LogsSubscription(label="bar", handler=idle_handler)
+
         await sub_manager.subscribe([sub1, sub2])
+
         assert sub_manager.subscriptions == [sub1, sub2]
 
-        asyncio.create_task(unsubscribe_subs([sub1, sub2]))
+        unsubscribe_task = asyncio.create_task(unsubscribe_subs([sub1, sub2]))
         # With no subscriptions in the queue, ``handle_subscriptions`` should hang
         # indefinitely. Test that when the last subscription is unsubscribed from,
         # the method breaks out of the loop. This is done via a raised
         # ``SubscriptionProcessingFinished`` within the ``TaskReliantQueue``.
         await sub_manager.handle_subscriptions()
+
         assert_no_subscriptions_left(sub_manager._subscription_container)
+        await clean_up_task(unsubscribe_task)