feat: upstream provider and forked network tools [APE-1485] (ApeWorX#…

…1714)

src/ape/api/__init__.py

@@ -10,7 +10,13 @@
 from .config import ConfigDict, ConfigEnum, PluginConfig
 from .convert import ConverterAPI
 from .explorers import ExplorerAPI
-from .networks import EcosystemAPI, NetworkAPI, ProviderContextManager, create_network_type
+from .networks import (
+    EcosystemAPI,
+    ForkedNetworkAPI,
+    NetworkAPI,
+    ProviderContextManager,
+    create_network_type,
+)
 from .projects import DependencyAPI, ProjectAPI
 from .providers import (
     BlockAPI,
@@ -36,6 +42,7 @@
     "DependencyAPI",
     "EcosystemAPI",
     "ExplorerAPI",
+    "ForkedNetworkAPI",
     "ImpersonatedAccount",
     "PluginConfig",
     "ProjectAPI",

src/ape/api/networks.py

@@ -48,7 +48,7 @@
 
 if TYPE_CHECKING:
     from .explorers import ExplorerAPI
-    from .providers import BlockAPI, ProviderAPI
+    from .providers import BlockAPI, ProviderAPI, UpstreamProvider
     from .transactions import ReceiptAPI, TransactionAPI
 
 
@@ -1046,6 +1046,51 @@ def verify_chain_id(self, chain_id: int):
             raise NetworkMismatchError(chain_id, self)
 
 
+class ForkedNetworkAPI(NetworkAPI):
+    @property
+    def upstream_network(self) -> NetworkAPI:
+        """
+        The network being forked.
+        """
+        network_name = self.name.replace("-fork", "")
+        return self.ecosystem.get_network(network_name)
+
+    @property
+    def upstream_provider(self) -> "UpstreamProvider":
+        """
+        The provider used when requesting data before the local fork.
+        Set this in your config under the network settings.
+        When not set, will attempt to use the default provider, if one
+        exists.
+        """
+
+        config_choice = self._network_config.get("upstream_provider")
+        if provider_name := config_choice or self.upstream_network.default_provider:
+            return self.get_provider(provider_name)
+
+        raise NetworkError(f"Upstream network '{self.upstream_network}' has no providers.")
+
+    @property
+    def upstream_chain_id(self) -> int:
+        """
+        The chain Id of the upstream network.
+        For example, when on ``mainnet-fork``, this should always
+        return the chain ID for ``mainnet``. Some providers may use
+        a different chain ID for forked networks while some do not.
+        This property should ALWAYS be that of the forked network, regardless.
+        """
+        return self.upstream_network.chain_id
+
+    def use_upstream_provider(self) -> ProviderContextManager:
+        """
+        Connect to the upstream provider.
+
+        Returns:
+            :class:`~ape.api.networks.ProviderContextManager`
+        """
+        return self.upstream_network.use_provider(self.upstream_provider.name)
+
+
 def create_network_type(chain_id: int, network_id: int) -> Type[NetworkAPI]:
     """
     Easily create a :class:`~ape.api.networks.NetworkAPI` subclass.

src/ape_ethereum/__init__.py

@@ -1,5 +1,5 @@
 from ape import plugins
-from ape.api import NetworkAPI, create_network_type
+from ape.api import ForkedNetworkAPI, NetworkAPI, create_network_type
 from ape.api.networks import LOCAL_NETWORK_NAME
 
 from ._converters import WeiConversions
@@ -25,7 +25,7 @@ def ecosystems():
 def networks():
     for network_name, network_params in NETWORKS.items():
         yield "ethereum", network_name, create_network_type(*network_params)
-        yield "ethereum", f"{network_name}-fork", NetworkAPI
+        yield "ethereum", f"{network_name}-fork", ForkedNetworkAPI
 
     # NOTE: This works for local providers, as they get chain_id from themselves
     yield "ethereum", LOCAL_NETWORK_NAME, NetworkAPI

src/ape_ethereum/ecosystem.py

@@ -131,23 +131,34 @@ def validate_gas_limit(cls, value):
         raise ValueError(f"Invalid gas limit '{value}'")
 
 
-def _create_local_config(default_provider: Optional[str] = None, **kwargs) -> NetworkConfig:
+class ForkedNetworkConfig(NetworkConfig):
+    upstream_provider: Optional[str] = None
+    """
+    The provider to use as the upstream-provider for this forked network.
+    """
+
+
+def _create_local_config(
+    default_provider: Optional[str] = None, use_fork: bool = False, **kwargs
+) -> NetworkConfig:
     return _create_config(
         base_fee_multiplier=1.0,
         default_provider=default_provider,
         gas_limit="max",
         required_confirmations=0,
         transaction_acceptance_timeout=DEFAULT_LOCAL_TRANSACTION_ACCEPTANCE_TIMEOUT,
+        cls=ForkedNetworkConfig if use_fork else NetworkConfig,
         **kwargs,
     )
 
 
 def _create_config(
     required_confirmations: int = 2,
     base_fee_multiplier: float = DEFAULT_LIVE_NETWORK_BASE_FEE_MULTIPLIER,
+    cls: Type[NetworkConfig] = NetworkConfig,
     **kwargs,
 ) -> NetworkConfig:
-    return NetworkConfig(
+    return cls(
         base_fee_multiplier=base_fee_multiplier,
         required_confirmations=required_confirmations,
         **kwargs,
@@ -156,11 +167,11 @@ def _create_config(
 
 class EthereumConfig(PluginConfig):
     mainnet: NetworkConfig = _create_config(block_time=13)
-    mainnet_fork: NetworkConfig = _create_local_config()
+    mainnet_fork: ForkedNetworkConfig = _create_local_config(use_fork=True)
     goerli: NetworkConfig = _create_config(block_time=15)
-    goerli_fork: NetworkConfig = _create_local_config()
+    goerli_fork: ForkedNetworkConfig = _create_local_config(use_fork=True)
     sepolia: NetworkConfig = _create_config(block_time=15)
-    sepolia_fork: NetworkConfig = _create_local_config()
+    sepolia_fork: ForkedNetworkConfig = _create_local_config(use_fork=True)
     local: NetworkConfig = _create_local_config(default_provider="test")
     default_network: str = LOCAL_NETWORK_NAME
 

tests/functional/test_network_api.py

@@ -34,3 +34,9 @@ def test_gas_limits(networks, config, project_with_source_files_contract):
 def test_base_fee_multiplier(networks):
     assert networks.ethereum.mainnet.base_fee_multiplier == 1.4
     assert networks.ethereum.local.base_fee_multiplier == 1.0
+
+
+def test_forked_networks(ethereum):
+    mainnet_fork = ethereum.mainnet_fork
+    assert mainnet_fork.upstream_network.name == "mainnet"
+    assert mainnet_fork.upstream_chain_id == 1