degenbot.fork ============= .. py:module:: degenbot.fork .. autoapi-nested-parse:: Anvil fork management utilities for local test chains. A thin companion shell (ADR-005 three-layer Python layer) over the rust core `degenbot._ffi.AnvilFork` PyO3 seam (`degenbot_fork::AnvilFork`). The rust core owns the spawned anvil subprocess (via `alloy::node_bindings::Anvil`) + a connected alloy `DynProvider` (over IPC) + the 12 anvil dev-RPC methods. This Python shell: - carries the anvil launch configuration in :class:`ForkLaunchConfig` (the fork target stays a direct ``fork_url`` / ``fork_block`` argument); - re-uses rust-driven dev-method invocations (`mine` / `reset` / `set_snapshot` / `return_to_snapshot` / `set_balance` / ...); - re-exposes general RPC against the forked node as `self.provider` — a `degenbot._ffi.AlloyProvider` constructed over the fork's HTTP endpoint (replaces the legacy `self.w3` Web3 handle; the rust-owned core's dev-RPC runs over its own IPC `DynProvider`, while the companion general-RPC provider is HTTP so borrowed consumers never hold an alloy pubsub). The breaking change vs the pre-0.7 Python `AnvilFork` is `self.w3` → `self.provider` (and the dropped capture-file / middlewares / free-port management). `_FFIAnvilFork` exposes the IPC path the rust core resolved for the spawned anvil process. Former callers using `fork.w3.eth.X` are migrated to `fork.provider.X` — the legacy attribute is gone. Package Contents ---------------- .. py:exception:: DegenbotValueError(*, message: str | None = None) Bases: :py:obj:`DegenbotError` DegenbotValueError error. .. py:exception:: AnvilError(method: str, error: str) Bases: :py:obj:`degenbot.exceptions.base.DegenbotError` Raised on errors resulting from failed calls to Anvil via JSON-RPC. This exception is specifically for errors that occur when making RPC calls to an Anvil instance, such as invalid method calls, parameter errors, or other Anvil-specific failures. .. py:attribute:: method .. py:attribute:: error .. py:data:: logger .. py:class:: AlloyProvider(rpc_url: str, max_retries: int = 10, max_blocks_per_request: int = 5000, chain_id: int | None = None) Bases: :py:obj:`_AlloyEndpointMixin`, :py:obj:`_AlloyQueryMixin`, :py:obj:`_AlloyIntrospectionMixin` High-performance Ethereum RPC provider using Alloy. Backs log fetching and basic RPC calls with a Rust HTTP client using connection pooling for optimal performance. The public surface is assembled from the query mixins; this class owns construction and teardown of the wrapped Rust provider. :param rpc_url: HTTP/HTTPS endpoint URL :param max_retries: Maximum retry attempts (default: 10) :param max_blocks_per_request: Maximum logs per request (default: 5000) :param chain_id: Chain to bind the endpoint to. The Rust core reads ``eth_chainId`` once at construction and raises :class:`ValueError` when the endpoint serves another chain; ``None`` constructs the provider with no binding. .. rubric:: Example >>> provider = AlloyProvider("https://eth-mainnet.example.com") >>> >>> # Properties >>> chain_id = provider.chain_id >>> block_number = provider.block_number >>> >>> # Methods >>> block = provider.get_block(18_000_000) >>> logs = provider.get_logs(from_block=18_000_000, to_block=18_010_000) >>> code = provider.get_code("0x...") >>> result = provider.call("0x...", calldata) .. py:type:: BlockNumber :canonical: int .. py:type:: ValidatedUint256 :canonical: Annotated[int, Field(ge=MIN_UINT256, le=MAX_UINT256)] .. py:class:: ForkLaunchConfig Non-target launch knobs for the rust-owned anvil subprocess. The fork target (``fork_url`` / ``fork_block``) stays a direct :class:`AnvilFork` argument because nearly every caller names it; this carrier groups the anvil-CLI, mining, and post-spawn state-override families that most callers leave at their defaults. Values forward to the rust core unchanged (``localhost`` becomes its ``host`` key; ``ipc_path`` is stringified). .. py:attribute:: localhost :type: str :value: '127.0.0.1' .. py:attribute:: fork_transaction_hash :type: str | None :value: None .. py:attribute:: mining_mode :type: Literal['auto', 'interval', 'none'] :value: 'auto' .. py:attribute:: mining_interval :type: int | None :value: None .. py:attribute:: storage_caching :type: bool :value: True .. py:attribute:: base_fee :type: int | None :value: None .. py:attribute:: ipc_path :type: pathlib.Path | None :value: None .. py:attribute:: mnemonic :type: str :value: 'patient rude simple dog close planet oval animal hunt sketch suspect slim' .. py:attribute:: chain_id :type: int | None :value: None .. py:attribute:: balance_overrides :type: collections.abc.Iterable[tuple[degenbot.types.chain.HexAddress, int]] | None :value: None .. py:attribute:: bytecode_overrides :type: collections.abc.Iterable[tuple[degenbot.types.chain.HexAddress, bytes]] | None :value: None .. py:attribute:: nonce_overrides :type: collections.abc.Iterable[tuple[degenbot.types.chain.HexAddress, int]] | None :value: None .. py:attribute:: storage_overrides :type: collections.abc.Iterable[tuple[degenbot.types.chain.HexAddress | bytes, int, degenbot.types.chain.HexStr | bytes | int]] | None :value: None .. py:attribute:: coinbase :type: degenbot.types.chain.HexAddress | None :value: None .. py:attribute:: anvil_opts :type: list[str] | None :value: None .. py:class:: AnvilFork(*, fork_url: str | None = None, fork_block: degenbot.types.aliases.BlockNumber | None = None, launch: ForkLaunchConfig | None = None) A rust-owned Anvil fork subprocess + IPC `DynProvider` + dev-RPC surface. Companion shell (ADR-005 Python layer) over `degenbot._ffi.AnvilFork` (the PyO3 seam on `degenbot_fork::AnvilFork`). The rust core owns the subprocess lifecycle (drop = kill) + an alloy `DynProvider` (IPC transport) + the 12 anvil dev-RPC methods. The shell carries the legacy constructor knobs in :class:`ForkLaunchConfig`, preserves the legacy dev-method names, raises the legacy `AnvilError` exception subclass on rust-side RPC failures, and exposes general RPC through the `AlloyProvider` pyclass at :attr:`provider` (replacing the legacy `self.w3` Web3 handle — that handle is gone; all callers now use ``fork.provider.get_block_number()`` etc.). The rust core (`degenbot_fork::AnvilFork`) owns: - the anvil subprocess lifecycle (drop = kill); - a connected alloy `DynProvider` over IPC (transport); - the 12 dev-RPC methods (``evm_mine``/``anvil_reset``/``evm_snapshot``/ ``evm_revert``/``anvil_set_balance``/...), exposed on this shell as the legacy Python names (``mine``/``reset``/``set_snapshot``/ ``return_to_snapshot``/``set_balance``/...). The Python shell keeps: - the legacy constructor knobs (grouped in :class:`ForkLaunchConfig`) + the multi-override queueing (``balance_overrides`` / ``bytecode_overrides`` / ``nonce_overrides`` / ``storage_overrides`` / ``coinbase``) — applied post-spawn via the dev-method delegations; - the legacy Python dev-method names so existing API surface doesn't reshuffle (the rust core's name for `set_snapshot` is `snapshot`, `return_to_snapshot` → `revert`, `set_storage` → `set_storage_at`, `set_next_base_fee` → `set_next_base_fee` — these translations happen in-shell); - the legacy `AnvilError` exception on rust RPC failures (translated from rust's `PyRuntimeError` at the boundary, so the legacy ``AnvilError(method=..., error=...)`` shape + the exception-subclass identity stay intact); - the `@validate_call` + `ValidatedUint256` annotations on `set_balance` / `set_next_base_fee` / `set_next_block_timestamp` so input-range validation raises pydantic `ValidationError` before the rust round-trip (matches the legacy pre-0.7 surface). Dropped (no longer needed now that subprocess lifecycle is rust-owned): the `--auto-impersonate` CLI arg list construction, the `_setup_process` / `_setup_w3` / `_close_ipc_socket` block, the `socket`-based free-port allocation, the `tenacity` spawn retry, the `subprocess.Popen` capture-file management (`capture_path` / `preserve_capture` / `*_capture_filename` properties), the `web3.IPCProvider` client + `middleware_onion.inject(middleware)` plumbing (alloy's `DynProvider` is the only transport layer under the rust core), the `ipc_provider_kwargs` shim, and `http_url` / `ws_url` / `port` / `ipc_filename` (anvil-CLI control knobs that the rust core controls directly now — the IPC path is the only socket surfaced back, via :attr:`ipc_path`). The one breaking change for callers: ``self.w3`` is gone. Use :attr:`provider` (a `degenbot._ffi.AlloyProvider`) for general RPC. .. py:property:: provider :type: degenbot.provider.AlloyProvider The general-RPC `AlloyProvider` bound to the fork over HTTP. Use for retry-aware ``eth_call`` / ``get_logs`` against the in-memory fork (mirrors the legacy ``self.w3`` handle). HTTP (not the rust core's IPC `DynProvider`) so that consumers which borrow it never hold an alloy IPC/WS pubsub that would reconnect into a dead socket when the fork closes. Raising once the fork is closed guarantees a provider over a dead subprocess is never silently handed out. :raises AnvilError: If the fork was closed. .. py:property:: fork_url :type: str | None Fork URL the spawned anvil process forks from, or `None` for in-memory. .. py:property:: ipc_path :type: str Resolved IPC socket path the spawned anvil process listens on. Use to construct a second `AlloyProvider` (or other transport) against the same forked node. Replaces the legacy ``AnvilFork.ipc_filename`` (a `pathlib.Path` derived from a user-supplied `ipc_path` parent + the allocated HTTP port — that bookkeeping is gone; the rust core owns the IPC socket directly + reports the resolved path here). :raises AnvilError: If the fork was closed. .. py:property:: http_url :type: str HTTP endpoint of the spawned anvil node (e.g. ``http://127.0.0.1:PORT``). Use to construct a second ``AlloyProvider`` over the rust-owned anvil subprocess when the IPC-bound `provider` is not enough (e.g. a plain HTTP transport from another Python process). .. py:property:: ws_url :type: str WebSocket endpoint of the spawned anvil node (e.g. ``ws://127.0.0.1:PORT``). .. py:property:: port :type: int TCP port the spawned anvil node listens on. .. py:method:: close() -> None Drop the rust-owned subprocess + IPC handle. The rust core (`degenbot_fork::AnvilFork`) owns the subprocess via alloy's `AnvilInstance`; dropping the `_FFIAnvilFork` pyclass handle terminates the anvil process + closes the IPC transport. This method just clears the Python-side reference so the GC finalizes the rust handle — there is no separate IPC-socket cleanup needed (alloy's `AnvilInstance` `Drop` impl handles it). .. py:method:: mine() -> None Mine a single block (`evm_mine`). :raises AnvilError: If the RPC call fails. .. py:method:: reset(fork_url: str | None = None, block_number: degenbot.types.aliases.BlockNumber | None = None, transaction_hash: str | None = None) -> None Fork from a new endpoint, block number, or transaction hash. Resetting to a new block number only is done in-place (no subprocess relaunch). Resetting to a new endpoint or from a transaction hash tears down the existing rust-owned subprocess + spawns a new one with the new fork parameters. Slower than in-place. :raises AnvilError: If the in-place RPC call fails. :raises DegenbotValueError: If no reset options are provided. .. py:method:: set_balance(address: str, balance: degenbot.validation.evm_values.ValidatedUint256) -> None Set balance (`anvil_setBalance`). :raises AnvilError: If the RPC call fails. .. py:method:: set_code(address: str, bytecode: bytes) -> None Set code (`anvil_setCode`). :raises AnvilError: If the RPC call fails. .. py:method:: set_coinbase(address: str) -> None Set coinbase (`anvil_setCoinbase`). :raises AnvilError: If the RPC call fails. .. py:method:: set_block_timestamp_interval(interval: int) -> None Set block timestamp interval (`anvil_setBlockTimestampInterval`). :raises AnvilError: If the RPC call fails. .. py:method:: set_next_base_fee(fee: degenbot.validation.evm_values.ValidatedUint256) -> None Set the next block base fee (`anvil_setNextBlockBaseFeePerGas`). :raises AnvilError: If the RPC call fails. .. py:method:: set_next_block_timestamp(timestamp: degenbot.validation.evm_values.ValidatedUint256) -> None Set the next block timestamp (`evm_setNextBlockTimestamp`). :raises AnvilError: If the RPC call fails. .. py:method:: set_nonce(address: str, nonce: int) -> None Set nonce (`anvil_setNonce`). :raises AnvilError: If the RPC call fails. .. py:method:: set_storage(address: degenbot.types.chain.HexAddress | bytes, position: int, value: degenbot.types.chain.HexStr | bytes | int) -> None Set storage (`anvil_setStorageAt`). :raises AnvilError: If the RPC call fails.