fix: issues causing dev messages from working on vyper 0.3.9 [APE-132…

…0] (ApeWorX#1625)

src/ape/api/compiler.py

@@ -193,7 +193,10 @@ def _create_contract_from_call(
         if "address" not in data:
             return None, calldata
 
-        addr = data["address"]
+        # NOTE: Handling when providers give us odd address values.
+        raw_addr = HexBytes(data["address"]).hex().replace("0x", "")
+        zeroes = max(40 - len(raw_addr), 0) * "0"
+        addr = f"0x{zeroes}{raw_addr}"
 
         try:
             address = self.provider.network.ecosystem.decode_address(addr)

src/ape/api/providers.py

@@ -56,6 +56,7 @@
     ContractLog,
     LogFilter,
     SnapshotID,
+    SourceTraceback,
     TraceFrame,
 )
 from ape.utils import (
@@ -702,16 +703,9 @@ def _increment_call_func_coverage_hit_count(self, txn: TransactionAPI):
         ):
             return
 
-        cov_data = self._test_runner.coverage_tracker.data
-        if not cov_data:
-            return
-
-        contract_type = self.chain_manager.contracts.get(txn.receiver)
-        if not contract_type:
-            return
-
-        contract_src = self.project_manager._create_contract_source(contract_type)
-        if not contract_src:
+        if not (contract_type := self.chain_manager.contracts.get(txn.receiver)) or not (
+            contract_src := self.project_manager._create_contract_source(contract_type)
+        ):
             return
 
         method_id = txn.data[:4]
@@ -1571,8 +1565,7 @@ def get_virtual_machine_error(self, exception: Exception, **kwargs) -> VirtualMa
         if not isinstance(err_data, dict):
             return VirtualMachineError(base_err=exception, **kwargs)
 
-        err_msg = err_data.get("message")
-        if not err_msg:
+        if not (err_msg := err_data.get("message")):
             return VirtualMachineError(base_err=exception, **kwargs)
 
         if txn is not None and "nonce too low" in str(err_msg):
@@ -1593,9 +1586,14 @@ def _handle_execution_reverted(
         txn: Optional[TransactionAPI] = None,
         trace: Optional[Iterator[TraceFrame]] = None,
         contract_address: Optional[AddressType] = None,
+        source_traceback: Optional[SourceTraceback] = None,
     ) -> ContractLogicError:
         message = str(exception).split(":")[-1].strip()
-        params: Dict = {"trace": trace, "contract_address": contract_address}
+        params: Dict = {
+            "trace": trace,
+            "contract_address": contract_address,
+            "source_traceback": source_traceback,
+        }
         no_reason = message == "execution reverted"
 
         if isinstance(exception, Web3ContractLogicError) and no_reason:

src/ape/exceptions.py

@@ -2,16 +2,13 @@
 import tempfile
 import time
 import traceback
-from collections import deque
-from functools import cached_property
 from inspect import getframeinfo, stack
 from pathlib import Path
 from types import CodeType, TracebackType
 from typing import TYPE_CHECKING, Any, Dict, Iterator, List, Optional, Union
 
 import click
 from eth_utils import humanize_hash
-from ethpm_types import ContractType
 from ethpm_types.abi import ConstructorABI, ErrorABI, MethodABI
 from rich import print as rich_print
 
@@ -180,12 +177,10 @@ def _set_tb(self):
         if not self.source_traceback and self.txn:
             self.source_traceback = _get_ape_traceback(self.txn)
 
-        src_tb = self.source_traceback
-        if src_tb is not None and self.txn is not None:
+        if (src_tb := self.source_traceback) and self.txn is not None:
             # Create a custom Pythonic traceback using lines from the sources
             # found from analyzing the trace of the transaction.
-            py_tb = _get_custom_python_traceback(self, self.txn, src_tb)
-            if py_tb:
+            if py_tb := _get_custom_python_traceback(self, self.txn, src_tb):
                 self.__traceback__ = py_tb
 
 
@@ -213,12 +208,6 @@ def __init__(
         self.txn = txn
         self.trace = trace
         self.contract_address = contract_address
-        if revert_message is None:
-            try:
-                # Attempt to use dev message as main exception message.
-                revert_message = self.dev_message
-            except Exception:
-                pass
 
         super().__init__(
             base_err=base_err,
@@ -229,11 +218,18 @@ def __init__(
             txn=txn,
         )
 
+        if revert_message is None and source_traceback is not None and (dev := self.dev_message):
+            try:
+                # Attempt to use dev message as main exception message.
+                self.message = dev
+            except Exception:
+                pass
+
     @property
     def revert_message(self):
         return self.message
 
-    @cached_property
+    @property
     def dev_message(self) -> Optional[str]:
         """
         The dev-string message of the exception.
@@ -242,109 +238,7 @@ def dev_message(self) -> Optional[str]:
             ``ValueError``: When unable to get dev message.
         """
 
-        trace = self._get_trace()
-        if len(trace) == 0:
-            raise ValueError("Missing trace.")
-
-        if address := self.address:
-            try:
-                contract_type = trace[-1].chain_manager.contracts[address]
-            except Exception as err:
-                raise ValueError(
-                    f"Could not fetch contract at {address} to check dev message."
-                ) from err
-
-        else:
-            raise ValueError("Could not fetch contract information to check dev message.")
-
-        if contract_type.pcmap is None:
-            raise ValueError("Compiler does not support source code mapping.")
-
-        pc = None
-        pcmap = contract_type.pcmap.parse()
-
-        # To find a suitable line for inspecting dev messages, we must start at the revert and work
-        # our way backwards. If the last frame's PC is in the PC map, the offending line is very
-        # likely a 'raise' statement.
-        if trace[-1].pc in pcmap:
-            pc = trace[-1].pc
-
-        # Otherwise we must traverse the trace backwards until we find our first suitable candidate.
-        else:
-            last_depth = 1
-            while len(trace) > 0:
-                frame = trace.pop()
-                if frame.depth > last_depth:
-                    # Call was made, get the new PCMap.
-                    contract_type = self._find_next_contract(trace)
-                    if not contract_type.pcmap:
-                        raise ValueError("Compiler does not support source code mapping.")
-
-                    pcmap = contract_type.pcmap.parse()
-                    last_depth += 1
-
-                if frame.pc in pcmap:
-                    pc = frame.pc
-                    break
-
-        # We were unable to find a suitable PC that matched the compiler's map.
-        if pc is None:
-            return None
-
-        offending_source = pcmap[pc]
-        if offending_source is None:
-            return None
-
-        dev_messages = contract_type.dev_messages or {}
-        if offending_source.line_start is None:
-            # Check for a `dev` field in PCMap.
-            return None if offending_source.dev is None else offending_source.dev
-
-        elif offending_source.line_start in dev_messages:
-            return dev_messages[offending_source.line_start]
-
-        elif offending_source.dev is not None:
-            return offending_source.dev
-
-        # Dev message is neither found from the compiler or from a dev-comment.
-        return None
-
-    def _get_trace(self) -> deque:
-        trace = None
-        if self.trace is None and self.txn is not None:
-            try:
-                trace = deque(self.txn.trace)
-            except APINotImplementedError as err:
-                raise ValueError(
-                    "Cannot check dev message; provider must support transaction tracing."
-                ) from err
-
-            except (ProviderError, SignatureError) as err:
-                raise ValueError("Cannot fetch transaction trace.") from err
-
-        elif self.trace is not None:
-            trace = deque(self.trace)
-
-        if not trace:
-            raise ValueError("Cannot fetch transaction trace.")
-
-        return trace
-
-    def _find_next_contract(self, trace: deque) -> ContractType:
-        msg = "Could not fetch contract at '{address}' to check dev message."
-        idx = len(trace) - 1
-        while idx >= 0:
-            frame = trace[idx]
-            if frame.contract_address:
-                ct = frame.chain_manager.contracts.get(frame.contract_address)
-                if not ct:
-                    raise ValueError(msg.format(address=frame.contract_address))
-
-                return ct
-
-            idx -= 1
-
-        raise ValueError(msg.format(address=frame.contract_address))
+        return self.source_traceback.revert_type if self.source_traceback else None
 
     @classmethod
     def from_error(cls, err: Exception):

src/ape/types/trace.py

@@ -1,6 +1,6 @@
-from itertools import tee
+from itertools import chain, tee
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, Dict, Iterator, List, Optional, Set
+from typing import TYPE_CHECKING, Any, Dict, Iterator, List, Optional, Set, Union
 
 from ethpm_types import ASTNode, BaseModel, ContractType, HexBytes
 from ethpm_types.ast import SourceLocation
@@ -489,17 +489,17 @@ class SourceTraceback(BaseModel):
     __root__: List[ControlFlow]
 
     @classmethod
-    def create(cls, contract_type: ContractType, trace: Iterator[TraceFrame], data: HexBytes):
-        source_id = contract_type.source_id
-        if not source_id:
+    def create(
+        cls,
+        contract_type: ContractType,
+        trace: Iterator[TraceFrame],
+        data: Union[HexBytes, str],
+    ):
+        trace, second_trace = tee(trace)
+        if not second_trace or not (accessor := next(second_trace, None)):
             return cls.parse_obj([])
 
-        trace, second_trace = tee(trace)
-        if second_trace:
-            accessor = next(second_trace, None)
-            if not accessor:
-                return cls.parse_obj([])
-        else:
+        if not (source_id := contract_type.source_id):
             return cls.parse_obj([])
 
         ext = f".{source_id.split('.')[-1]}"
@@ -508,7 +508,7 @@ def create(cls, contract_type: ContractType, trace: Iterator[TraceFrame], data:
 
         compiler = accessor.compiler_manager.registered_compilers[ext]
         try:
-            return compiler.trace_source(contract_type, trace, data)
+            return compiler.trace_source(contract_type, trace, HexBytes(data))
         except NotImplementedError:
             return cls.parse_obj([])
 
@@ -533,24 +533,63 @@ def __getitem__(self, idx: int) -> ControlFlow:
     def __setitem__(self, key, value):
         return self.__root__.__setitem__(key, value)
 
+    @property
+    def revert_type(self) -> Optional[str]:
+        """
+        The revert type, such as a builtin-error code or a user dev-message,
+        if there is one.
+        """
+
+        return self.statements[-1].type if self.statements[-1].type != "source" else None
+
     def append(self, __object) -> None:
+        """
+        Append the given control flow to this one.
+        """
         self.__root__.append(__object)
 
     def extend(self, __iterable) -> None:
+        """
+        Append all the control flows from the given traceback to this one.
+        """
         if not isinstance(__iterable, SourceTraceback):
             raise TypeError("Can only extend another traceback object.")
 
         self.__root__.extend(__iterable.__root__)
 
     @property
     def last(self) -> Optional[ControlFlow]:
+        """
+        The last control flow in the traceback, if there is one.
+        """
         return self.__root__[-1] if len(self.__root__) else None
 
     @property
     def execution(self) -> List[ControlFlow]:
+        """
+        All the control flows in order. Each set of statements in
+        a control flow is separated by a jump.
+        """
         return list(self.__root__)
 
+    @property
+    def statements(self) -> List[Statement]:
+        """
+        All statements from each control flow.
+        """
+        return list(chain(*[x.statements for x in self.__root__]))
+
+    @property
+    def source_statements(self) -> List[SourceStatement]:
+        """
+        All source statements from each control flow.
+        """
+        return list(chain(*[x.source_statements for x in self.__root__]))
+
     def format(self) -> str:
+        """
+        Get a formatted traceback string for displaying to users.
+        """
         if not len(self.__root__):
             # No calls.
             return ""
@@ -566,6 +605,10 @@ def format(self) -> str:
                     continue
 
                 last_depth = control_flow.depth
+                content_str = control_flow.format()
+                if not content_str.strip():
+                    continue
+
                 segment = f"{indent}{control_flow.source_header}\n{control_flow.format()}"
 
                 # Try to include next statement for display purposes.
@@ -660,6 +703,20 @@ def add_builtin_jump(
         source_path: Optional[Path] = None,
         pcs: Optional[Set[int]] = None,
     ):
+        """
+        A convenience method for appending a control flow that happened
+        from an internal compiler built-in code. See the ape-vyper plugin
+        for a usage example.
+
+        Args:
+            name (str): The name of the compiler built-in.
+            _type (str): A str describing the type of check.
+            compiler_name (str): The name of the compiler.
+            full_name (Optional[str]): A full-name ID.
+            source_path (Optional[Path]): The source file related, if there is one.
+            pcs (Optional[Set[int]]): Program counter values mapping to this check.
+        """
+        # TODO: Assess if compiler_name is needed or get rid of in v0.7.
         pcs = pcs or set()
         closure = Closure(name=name, full_name=full_name or name)
         depth = self.last.depth - 1 if self.last else 0

src/ape/utils/__init__.py

@@ -55,7 +55,7 @@
     GeneratedDevAccount,
     generate_dev_accounts,
 )
-from ape.utils.trace import TraceStyles, parse_coverage_tables, parse_gas_table
+from ape.utils.trace import USER_ASSERT_TAG, TraceStyles, parse_coverage_tables, parse_gas_table
 
 __all__ = [
     "abstractmethod",
@@ -108,5 +108,6 @@
     "TraceStyles",
     "use_temp_sys_path",
     "USER_AGENT",
+    "USER_ASSERT_TAG",
     "ZERO_ADDRESS",
 ]