AnvilForkPool

Documentation for eth_defi.testing.anvil_fork_pool.AnvilForkPool Python class.

class AnvilForkPool

Bases: object

Registry of shared Anvil forks keyed by launch configuration.

One AnvilLaunch is created per distinct launch configuration and reused for every caller (on the same xdist worker) that requests it. Call close_all() once to tear every launch down.

Intended for read-only fork tests in its current form; see the module warning for the mutating-test caveat.

Attributes summary

launches

Cached launches keyed by (rpc_url, fork_block_number, sorted launch kwargs).

Methods summary

__init__([launches])

close_all()

Tear down every launched Anvil process.

get_launch(rpc_url, fork_block_number, ...)

Return a shared Anvil launch for this exact launch configuration.

get_web3(rpc_url, fork_block_number, *[, ...])

Return a fresh Web3 pointed at a shared Anvil fork.
launches: dict[tuple, eth_defi.provider.anvil.AnvilLaunch]

Cached launches keyed by (rpc_url, fork_block_number, sorted launch kwargs).

get_launch(rpc_url, fork_block_number, **launch_kwargs)

Return a shared Anvil launch for this exact launch configuration.

Launches Anvil lazily on the first request for a configuration and returns the cached process on every subsequent request.

Parameters

  • rpc_url (str) – Upstream archive JSON-RPC URL to fork from.

  • fork_block_number (int) – Fixed block to fork at. Required — a mutable chain tip cannot be shared safely.

  • launch_kwargs (Any) – Any other state-affecting fork_network_anvil arguments; they are part of the cache key so incompatible configs never share a process. When proxy_multiple_upstream is omitted, launch_anvil’s default applies (True — the bounded, fail-fast automatic failover proxy; see the module docstring); pass an explicit RPCProxyConfig, RPCProxy, or False to override.

Returns

The shared AnvilLaunch.

Return type

eth_defi.provider.anvil.AnvilLaunch

get_web3(rpc_url, fork_block_number, *, web3_retries=0, web3_http_timeout=(3.0, 60.0), **launch_kwargs)

Return a fresh Web3 pointed at a shared Anvil fork.

The underlying Anvil process is shared via get_launch(); the returned web3.Web3 object itself is created per call and is not shared.

Parameters

  • rpc_url (str) – Upstream archive JSON-RPC URL to fork from.

  • fork_block_number (int) – Fixed block to fork at.

  • web3_retries (int) – Number of outer retries against the local Anvil endpoint. Keep this low because the inner RPC proxy already performs provider failover.

  • web3_http_timeout (tuple[float, float]) – Connect and read timeout for local Anvil requests. Override this for a known slow cold-cache operation.

  • launch_kwargs (Any) – Additional fork_network_anvil arguments (part of the cache key).

Returns

A web3.Web3 connected to the shared Anvil RPC endpoint.

Return type

web3.main.Web3

close_all()

Tear down every launched Anvil process.

Every launch is attempted even if an earlier one fails to close, so a single teardown error cannot leak the remaining processes.

Return type

None

__init__(launches=<factory>)
Parameters

launches (dict[tuple, eth_defi.provider.anvil.AnvilLaunch]) –

Return type

None