wip

moccasin/_sys_path_and_config_setup.py

@@ -62,7 +62,10 @@ def _setup_network_and_account_from_args(
         )
 
     if mox_account:
-        boa.env.add_account(mox_account, force_eoa=True)
+        if fork:
+            boa.env.eoa = mox_account.address
+        else:
+            boa.env.add_account(mox_account, force_eoa=True)
     if boa.env.eoa is None:
         logger.warning(
             "No default EOA account found. Please add an account to the environment before attempting a transaction."

moccasin/config.py

@@ -57,10 +57,7 @@ def _create_env(self) -> _AnyEnv:
         from boa_zksync import ZksyncEnv
 
         if self.is_fork:
-            self._network_env = Env()
-            self._network_env = cast(_AnyEnv, self._network_env)
-            boa.fork(self.url)  # This won't work for ZKSync?
-            self._network_env.nickname = self.name
+            self.set_fork(self.is_fork)
         else:
             if self.is_zksync:
                 self._network_env = ZksyncEnv(EthereumRPC(self.url), nickname=self.name)
@@ -70,9 +67,17 @@ def _create_env(self) -> _AnyEnv:
                 )
         return self._network_env
 
-    def get_or_create_env(self) -> _AnyEnv:
+    def set_fork(self, is_fork: bool):
+        if self.is_fork != is_fork:
+            self._network_env = Env()
+            self._network_env = cast(_AnyEnv, self._network_env)
+            boa.fork(self.url)  # This won't work for ZKSync?
+            self._network_env.nickname = self.name
+
+    def get_or_create_env(self, is_fork: bool) -> _AnyEnv:
         import boa
 
+        self.set_fork(is_fork)
         if self._network_env:
             boa.set_env(self._network_env)
             return self._network_env
@@ -81,6 +86,7 @@ def get_or_create_env(self) -> _AnyEnv:
         return new_env
 
     # TODO this function is way too big
+    # manifest_contract()
     def get_or_deploy_contract(
         self,
         contract_name: str,
@@ -116,6 +122,7 @@ def get_or_deploy_contract(
         Returns:
             VyperContract: The deployed contract instance, or a blank contract if the contract is not found.
         """
+        # TODO: Have this funcion save this to the contracts dict
         moccasin_contract = self.contracts.get(
             contract_name, MoccasinContract("blank_moccasin_contract_name")
         )
@@ -332,15 +339,15 @@ def get_network_by_name(self, alias: str) -> Network:
     def set_active_network(self, name_or_url: str | Network, is_fork: bool = False):
         env_to_set: _AnyEnv
         if isinstance(name_or_url, Network):
-            env_to_set = name_or_url.get_or_create_env()
+            env_to_set = name_or_url.get_or_create_env(is_fork)
             self._networks[name_or_url.name] = env_to_set
         else:
             if name_or_url.startswith("http"):
                 new_network = self._create_custom_network(name_or_url, is_fork=is_fork)
-                env_to_set = new_network.get_or_create_env()
+                env_to_set = new_network.get_or_create_env(is_fork)
             else:
                 if name_or_url in self._networks:
-                    env_to_set = self._networks[name_or_url].get_or_create_env()
+                    env_to_set = self._networks[name_or_url].get_or_create_env(is_fork)
                 else:
                     raise ValueError(
                         f"Network {name_or_url} not found. Please pass a valid URL/RPC or valid network name."

moccasin/moccasin_account.py

@@ -7,18 +7,22 @@
 from eth_utils import to_bytes
 from moccasin.commands.wallet import decrypt_key
 from typing import cast
+from moccasin.logging import logger
+from eth_typing import ChecksumAddress
 
 
-class MoccasinAccount:
+class MoccasinAccount(LocalAccount):
     def __init__(
         self,
         private_key: str | bytes | None = None,
         keystore_path_or_account_name: Path | str | None = None,
         password: str = None,
         password_file_path: Path = None,
     ):
-        self._private_key = private_key
-        self._local_account: LocalAccount | None = None
+        self._private_key = None
+        self._publicapi = Account()
+        self._address = None
+
         if private_key:
             private_key = to_bytes(hexstr=private_key)
         private_key = cast(bytes, private_key)
@@ -31,32 +35,38 @@ def __init__(
             private_key = self.unlock(
                 password=password, password_file_path=password_file_path
             )
-        if not private_key:
-            raise Warning("Be sure to call unlock before trying to send a transaction.")
-        self._local_account = LocalAccount(PrivateKey(private_key), Account())
 
-    def __getattr__(self, name):
-        if self._local_account is not None:
-            try:
-                return getattr(self._local_account, name)
-            except AttributeError:
-                pass
-        raise AttributeError(
-            f"'{self.__class__.__name__}' object has no attribute '{name}'"
-        )
+        if private_key:
+            self._init_key(private_key)
+        else:
+            logger.warning(
+                "Be sure to call unlock before trying to send a transaction."
+            )
+
+    @property
+    def private_key(self) -> bytes:
+        return self.key
+
+    @property
+    def address(self) -> ChecksumAddress | None:
+        if self.private_key:
+            return PrivateKey(self.private_key).public_key.to_checksum_address()
+        return None
+
+    def _init_key(self, private_key: bytes):
+        private_key_converted = PrivateKey(private_key)
+        self._address = private_key_converted.public_key.to_checksum_address()
+        key_raw: bytes = private_key_converted.to_bytes()
+        self._private_key = key_raw
+        self._key_obj: PrivateKey = private_key_converted
 
     def set_keystore_path(self, keystore_path: Path | str):
         if isinstance(keystore_path, str):
             keystore_path = DEFAULT_KEYSTORES_PATH.joinpath(Path(keystore_path))
         self.keystore_path = keystore_path
 
-    def set_private_key(self, private_key: str | HexBytes):
-        self._private_key = (
-            private_key if isinstance(private_key, HexBytes) else HexBytes(private_key)
-        )
-
     def unlocked(self) -> bool:
-        return self._private_key is not None
+        return self.private_key is not None
 
     def unlock(
         self,
@@ -71,10 +81,11 @@ def unlock(
                 raise Exception(
                     "No keystore path provided. Set it with set_keystore_path (path)"
                 )
-            self._private_key = decrypt_key(
+            decrypted_key = decrypt_key(
                 self.keystore_path.stem,
                 password=password,
                 password_file_path=password_file_path,
                 keystores_path=self.keystore_path.parent,
             )
-        return cast(HexBytes, self._private_key)
+            self._init_key(decrypted_key)
+        return cast(HexBytes, self.private_key)

tests/cli/test_cli_run.py

@@ -96,20 +96,16 @@ def test_run_fork_should_not_send_transactions(
     current_dir = Path.cwd()
     os.chdir(COMPLEX_PROJECT_PATH)
     try:
-        w3 = Web3(Web3.HTTPProvider(ANVIL_URL))
-        starting_block = w3.eth.get_block("latest").number
         result = subprocess.run(
-            [mox_path, "run", "deploy", "--fork", "--network", "fake_chain"],
+            [mox_path, "run", "deploy", "--fork", "--network", "anvil"],
             check=True,
             capture_output=True,
             text=True,
         )
-        ending_block = w3.eth.get_block("latest").number
     finally:
         os.chdir(current_dir)
     assert "Ending count:  1" in result.stdout
     assert "tx broadcasted" not in result.stdout
-    assert starting_block == ending_block
     assert result.returncode == 0
 
 
@@ -125,7 +121,6 @@ def test_deploy_via_config_get_or_deploy_contract(mox_path, set_fake_chain_rpc):
         )
     finally:
         os.chdir(current_dir)
-    breakpoint()
 
 
 def test_deploy_from_loaded_state_network_via_config(

tests/conftest.py

@@ -118,7 +118,7 @@ def installation_cleanup_dependencies():
         shutil.rmtree(created_folder_path)
 
 
-@pytest.fixture(scope="session")
+@pytest.fixture()
 def anvil_process():
-    with AnvilProcess(args=["--state-path", ANVIL_STORED_STATE_PATH]):
+    with AnvilProcess(args=["--load-state", str(ANVIL_STORED_STATE_PATH), "-b", "1"]):
         yield