feat: allow forking custom networks automatically (ApeWorX#2145)

docs/userguides/networks.md

@@ -67,13 +67,15 @@ Here are some examples of changes L2 plugins make that allow improved support fo
 
 Here is a list of all L2 network plugins supported by Ape:
 
-| Name              | GitHub Path                                                               |
+| Name              | GitHub                                                                    |
 | ----------------- | ------------------------------------------------------------------------- |
-| ape-avalanche     | [ApeWorX/ape-avalanche](https://github.com/ApeWorX/ape-avalanche)         |
 | ape-arbitrum      | [ApeWorX/ape-arbitrum](https://github.com/ApeWorX/ape-arbitrum)           |
+| ape-avalanche     | [ApeWorX/ape-avalanche](https://github.com/ApeWorX/ape-avalanche)         |
 | ape-base          | [ApeWorX/ape-base](https://github.com/ApeWorX/ape-base)                   |
+| ape-blast         | [ApeWorX/ape-base](https://github.com/ApeWorX/ape-blast)                  |
+| ape-bsc           | [ApeWorX/ape-base](https://github.com/ApeWorX/ape-bsc)                    |
 | ape-fantom        | [ApeWorX/ape-fantom](https://github.com/ApeWorX/ape-fantom)               |
-| ape-optmism       | [ApeWorX/ape-optimism](https://github.com/ApeWorX/ape-optimism)           |
+| ape-optimism      | [ApeWorX/ape-optimism](https://github.com/ApeWorX/ape-optimism)           |
 | ape-polygon       | [ApeWorX/ape-polygon](https://github.com/ApeWorX/ape-polygon)             |
 | ape-polygon-zkevm | [ApeWorX/ape-polygon-zkevm](https://github.com/ApeWorX/ape-polygon-zkevm) |
 
@@ -178,6 +180,28 @@ node:
 
 Now, when using `ethereum:apenet:node`, it will connect to the RPC URL `https://apenet.example.com/rpc`.
 
+#### Forking Custom Networks
+
+You can fork custom networks using providers that support forking, such as `ape-foundry` or `ape-hardhat`.
+To fork a custom network, first ensure the custom network is set-up by following the sections above.
+Once you can successfully connect to a custom network in Ape, you can fork it.
+
+To fork the network, launch an Ape command with the `--network` option with your custom network name suffixed with `-fork` and use one of the forking providers (such as `ape-foundry`):
+
+```
+ape <cmd> --network shibarium:puppynet-fork:foundry
+```
+
+Configure the forked network in the plugin the same way you configure other forked networks:
+
+```yaml
+foundry:
+  fork:
+    shibarium:
+      puppynet:
+        block_number: 500
+```
+
 #### Explorer URL
 
 To configure explorer URLs for your custom network, use the explorer's plugin config.

src/ape/api/networks.py

@@ -237,20 +237,44 @@ def networks(self) -> dict[str, "NetworkAPI"]:
         networks = {**self._networks_from_plugins}
 
         # Include configured custom networks.
-        custom_networks = [
+        custom_networks: list = [
             n
             for n in self.config_manager.get_config("networks").custom
             if (n.ecosystem or self.network_manager.default_ecosystem.name) == self.name
         ]
+
+        # Ensure forks are added automatically for custom networks.
+        forked_custom_networks = []
+        for net in custom_networks:
+            if net.name.endswith("-fork"):
+                # Already a fork.
+                continue
+
+            fork_network_name = f"{net.name}-fork"
+            if any(x.name == fork_network_name for x in custom_networks):
+                # The forked version of this network is already known.
+                continue
+
+            # Create a forked network mirroring the custom network.
+            forked_net = net.model_copy(deep=True)
+            forked_net.name = fork_network_name
+            forked_custom_networks.append(forked_net)
+
+        # NOTE: Forked custom networks are still custom networks.
+        custom_networks.extend(forked_custom_networks)
+
         for custom_net in custom_networks:
             if custom_net.name in networks:
                 raise NetworkError(
                     f"More than one network named '{custom_net.name}' in ecosystem '{self.name}'."
                 )
 
+            is_fork = custom_net.is_fork
             network_data = custom_net.model_dump(by_alias=True, exclude=("default_provider",))
             network_data["ecosystem"] = self
-            network_type = create_network_type(custom_net.chain_id, custom_net.chain_id)
+            network_type = create_network_type(
+                custom_net.chain_id, custom_net.chain_id, is_fork=is_fork
+            )
             network_api = network_type.model_validate(network_data)
             network_api._default_provider = custom_net.default_provider
             network_api._is_custom = True
@@ -758,6 +782,17 @@ def disconnect_all(self):
         self.connected_providers = {}
 
 
+def _set_provider(provider: "ProviderAPI") -> "ProviderAPI":
+    connection_id = provider.connection_id
+    if connection_id in ProviderContextManager.connected_providers:
+        # Likely multi-chain testing or utilizing multiple on-going connections.
+        provider = ProviderContextManager.connected_providers[connection_id]
+        if not provider.is_connected:
+            provider.connect()
+
+    return provider
+
+
 class NetworkAPI(BaseInterfaceModel):
     """
     A wrapper around a provider for a specific ecosystem.
@@ -1037,26 +1072,27 @@ def get_provider(
             provider_settings["ipc_path"] = provider_name
             provider_name = "node"
 
+        # If it can fork Ethereum (and we are asking for it) assume it can fork this one.
+        # TODO: Refactor this approach to work for custom-forked non-EVM networks.
+        common_forking_providers = self.network_manager.ethereum.mainnet_fork.providers
+
         if provider_name in self.providers:
             provider = self.providers[provider_name](provider_settings=provider_settings)
-            connection_id = provider.connection_id
-            if connection_id in ProviderContextManager.connected_providers:
-                # Likely multi-chain testing or utilizing multiple on-going connections.
-                provider = ProviderContextManager.connected_providers[connection_id]
-                if not provider.is_connected:
-                    provider.connect()
+            return _set_provider(provider)
 
-                return provider
-
-            return provider
-
-        else:
-            raise ProviderNotFoundError(
-                provider_name,
-                network=self.name,
-                ecosystem=self.ecosystem.name,
-                options=self.providers,
+        elif self.is_fork and provider_name in common_forking_providers:
+            provider = common_forking_providers[provider_name](
+                provider_settings=provider_settings,
+                network=self,
             )
+            return _set_provider(provider)
+
+        raise ProviderNotFoundError(
+            provider_name,
+            network=self.name,
+            ecosystem=self.ecosystem.name,
+            options=self.providers,
+        )
 
     def use_provider(
         self,
@@ -1276,12 +1312,13 @@ def use_upstream_provider(self) -> ProviderContextManager:
         return self.upstream_network.use_provider(self.upstream_provider)
 
 
-def create_network_type(chain_id: int, network_id: int) -> type[NetworkAPI]:
+def create_network_type(chain_id: int, network_id: int, is_fork: bool = False) -> type[NetworkAPI]:
     """
     Easily create a :class:`~ape.api.networks.NetworkAPI` subclass.
     """
+    BaseNetwork = ForkedNetworkAPI if is_fork else NetworkAPI
 
-    class network_def(NetworkAPI):
+    class network_def(BaseNetwork):  # type: ignore
         @property
         def chain_id(self) -> int:
             return chain_id

src/ape/cli/choices.py

@@ -335,15 +335,15 @@ def __init__(
 
         self.base_type = base_type
         self.callback = callback
-        super().__init__(
-            get_networks(ecosystem=ecosystem, network=network, provider=provider), case_sensitive
-        )
+        networks = get_networks(ecosystem=ecosystem, network=network, provider=provider)
+        super().__init__(networks, case_sensitive)
 
     def get_metavar(self, param):
         return "[ecosystem-name][:[network-name][:[provider-name]]]"
 
     def convert(self, value: Any, param: Optional[Parameter], ctx: Optional[Context]) -> Any:
         choice: Optional[Union[str, ProviderAPI]]
+        networks = ManagerAccessMixin.network_manager
         if not value:
             choice = None
 
@@ -359,23 +359,26 @@ def convert(self, value: Any, param: Optional[Parameter], ctx: Optional[Context]
             try:
                 # Validate result.
                 choice = super().convert(value, param, ctx)
-            except BadParameter as err:
-                # If an error was not raised for some reason, raise a simpler error.
-                # NOTE: Still avoid showing the massive network options list.
-                raise click.BadParameter(
-                    "Invalid network choice. Use `ape networks list` to see options."
-                ) from err
+            except BadParameter:
+                # Attempt to get the provider anyway.
+                # Sometimes, depending on the provider, it'll still work.
+                # (as-is the case for custom-forked networks).
+                try:
+                    choice = networks.get_provider_from_choice(network_choice=value)
+                except Exception as err:
+                    # If an error was not raised for some reason, raise a simpler error.
+                    # NOTE: Still avoid showing the massive network options list.
+                    raise click.BadParameter(
+                        "Invalid network choice. Use `ape networks list` to see options."
+                    ) from err
 
         if (
             choice not in (None, _NONE_NETWORK)
             and isinstance(choice, str)
             and issubclass(self.base_type, ProviderAPI)
         ):
             # Return the provider.
-
-            choice = ManagerAccessMixin.network_manager.get_provider_from_choice(
-                network_choice=value
-            )
+            choice = networks.get_provider_from_choice(network_choice=value)
 
         return self.callback(ctx, param, choice) if self.callback else choice
 

src/ape/cli/options.py

@@ -296,7 +296,7 @@ def callback(ctx, param, value):
             # No network kwargs are used. No need for partial wrapper.
             wrapped_f = f
 
-        # Use NetworkChoice option.    Raises:
+        # Use NetworkChoice option.
         kwargs["type"] = None
 
         # Set this to false to avoid click passing in a str value for network.

src/ape_networks/__init__.py

@@ -27,6 +27,13 @@ class CustomNetwork(PluginConfig):
     request_header: dict = {}
     """The HTTP request header."""
 
+    @property
+    def is_fork(self) -> bool:
+        """
+        ``True`` when the name of the network ends in ``"-fork"``.
+        """
+        return self.name.endswith("-fork")
+
 
 class NetworksConfig(PluginConfig):
     custom: list[CustomNetwork] = []

tests/functional/test_cli.py

@@ -301,7 +301,6 @@ def cmd(network):
 def test_network_option_specify_custom_network(
     runner, project, custom_networks_config_dict, network_name
 ):
-    network_part = ("--network", f"ethereum:{network_name}:node")
     with project.temp_config(**custom_networks_config_dict):
         # NOTE: Also testing network filter with a custom network
         #  But this is also required to work around LRU cache
@@ -312,10 +311,12 @@ def test_network_option_specify_custom_network(
         @click.command()
         @network_option(network=network_name)
         def cmd(network):
-            click.echo(f"Value is '{network.name}'")
+            click.echo(f"Value is '{getattr(network, 'name', network)}'")
 
-        result = runner.invoke(cmd, network_part)
+        result = runner.invoke(cmd, ("--network", f"ethereum:{network_name}:node"))
         assert f"Value is '{network_name}'" in result.output
+        result = runner.invoke(cmd, ("--network", f"ethereum:{network_name}-fork:node"))
+        assert f"Value is '{network_name}-fork'" in result.output
 
 
 def test_account_option(runner, keyfile_account):

tests/functional/test_ecosystem.py

@@ -9,7 +9,7 @@
 from ethpm_types.abi import ABIType, EventABI, MethodABI
 from evm_trace import CallTreeNode, CallType
 
-from ape.api.networks import LOCAL_NETWORK_NAME, NetworkAPI
+from ape.api.networks import LOCAL_NETWORK_NAME, ForkedNetworkAPI, NetworkAPI
 from ape.exceptions import CustomError, DecodingError, NetworkError, NetworkNotFoundError
 from ape.types import AddressType
 from ape.utils import DEFAULT_LOCAL_TRANSACTION_ACCEPTANCE_TIMEOUT
@@ -868,9 +868,22 @@ def test_set_default_network_not_exists(ethereum):
 
 def test_networks(ethereum):
     actual = ethereum.networks
-    for net in ("sepolia", "mainnet", LOCAL_NETWORK_NAME):
+
+    with_forks = ("sepolia", "mainnet")
+    without_forks = (LOCAL_NETWORK_NAME,)
+
+    def assert_net(
+        net: str,
+    ):
         assert net in actual
-        assert isinstance(actual[net], NetworkAPI)
+        base_class = ForkedNetworkAPI if net.endswith("-fork") else NetworkAPI
+        assert isinstance(actual[net], base_class)
+
+    for net in with_forks:
+        assert_net(net)
+        assert_net(f"{net}-fork")
+    for net in without_forks:
+        assert_net(net)
 
 
 def test_networks_includes_custom_networks(
@@ -884,6 +897,9 @@ def test_networks_includes_custom_networks(
             LOCAL_NETWORK_NAME,
             custom_network_name_0,
             custom_network_name_1,
+            # Show custom networks are auto-forked!
+            f"{custom_network_name_0}-fork",
+            f"{custom_network_name_1}-fork",
         ):
             assert net in actual
             assert isinstance(actual[net], NetworkAPI)

tests/functional/test_network_api.py

@@ -3,7 +3,8 @@
 
 import pytest
 
-from ape.api import ProviderAPI
+from ape.api.networks import ForkedNetworkAPI, NetworkAPI, create_network_type
+from ape.api.providers import ProviderAPI
 from ape.exceptions import NetworkError, ProviderNotFoundError
 from ape_ethereum import EthereumConfig
 from ape_ethereum.transactions import TransactionType
@@ -181,3 +182,16 @@ def test_use_provider_previously_used_and_not_connected(eth_tester_provider):
     eth_tester_provider.disconnect()
     with network.use_provider("test") as provider:
         assert provider.is_connected
+
+
+def test_create_network_type():
+    chain_id = 123321123321123321
+    actual = create_network_type(chain_id, chain_id)
+    assert issubclass(actual, NetworkAPI)
+
+
+def test_create_network_type_fork():
+    chain_id = 123321123321123322
+    actual = create_network_type(chain_id, chain_id, is_fork=True)
+    assert issubclass(actual, NetworkAPI)
+    assert issubclass(actual, ForkedNetworkAPI)

tests/functional/test_network_manager.py

@@ -4,6 +4,7 @@
 import pytest
 
 from ape.api import EcosystemAPI
+from ape.api.networks import LOCAL_NETWORK_NAME
 from ape.exceptions import NetworkError, ProviderNotFoundError
 from ape.utils import DEFAULT_TEST_CHAIN_ID
 
@@ -388,3 +389,14 @@ def test_getitem(networks):
     ethereum = networks["ethereum"]
     assert ethereum.name == "ethereum"
     assert isinstance(ethereum, EcosystemAPI)
+
+
+def test_network_names(networks, custom_networks_config_dict, project):
+    data = copy.deepcopy(custom_networks_config_dict)
+    with project.temp_config(**data):
+        actual = networks.network_names
+        assert LOCAL_NETWORK_NAME in actual
+        assert "mainnet" in actual  # Normal
+        assert "mainnet-fork" in actual  # Forked
+        assert "apenet" in actual  # Custom
+        assert "apenet-fork" in actual  # Custom forked