Skip to content

EVM Transition Tool Package

Library of Python wrappers for the different implementations of transition tools.

UnknownTransitionTool

Bases: Exception

Exception raised if an unknown t8n is encountered

Source code in src/evm_transition_tool/transition_tool.py 
16
17
18
19
class UnknownTransitionTool(Exception):
    """Exception raised if an unknown t8n is encountered"""

    pass

GethTransitionTool

Bases: TransitionTool

Go-ethereum evm Transition tool frontend wrapper class.

Source code in src/evm_transition_tool/geth.py 
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
class GethTransitionTool(TransitionTool):
    """
    Go-ethereum `evm` Transition tool frontend wrapper class.
    """

    default_binary = Path("evm")
    detect_binary_pattern = compile(r"^evm version\b")

    binary: Path
    cached_version: Optional[str] = None
    trace: bool

    def __init__(
        self,
        *,
        binary: Optional[Path] = None,
        trace: bool = False,
    ):
        super().__init__(binary=binary, trace=trace)
        args = [str(self.binary), "t8n", "--help"]
        try:
            result = subprocess.run(args, capture_output=True, text=True)
        except subprocess.CalledProcessError as e:
            raise Exception("evm process unexpectedly returned a non-zero status code: " f"{e}.")
        except Exception as e:
            raise Exception(f"Unexpected exception calling evm tool: {e}.")
        self.help_string = result.stdout

    def evaluate(
        self,
        alloc: Any,
        txs: Any,
        env: Any,
        fork: Fork,
        chain_id: int = 1,
        reward: int = 0,
        eips: Optional[List[int]] = None,
    ) -> Tuple[Dict[str, Any], Dict[str, Any]]:
        """
        Executes `evm t8n` with the specified arguments.
        """
        fork_name = fork.name()
        if eips is not None:
            fork_name = "+".join([fork_name] + [str(eip) for eip in eips])

        temp_dir = tempfile.TemporaryDirectory()

        if int(env["currentNumber"], 0) == 0:
            reward = -1
        args = [
            str(self.binary),
            "t8n",
            "--input.alloc=stdin",
            "--input.txs=stdin",
            "--input.env=stdin",
            "--output.result=stdout",
            "--output.alloc=stdout",
            "--output.body=txs.rlp",
            f"--output.basedir={temp_dir.name}",
            f"--state.fork={fork_name}",
            f"--state.chainid={chain_id}",
            f"--state.reward={reward}",
        ]

        if self.trace:
            args.append("--trace")

        stdin = {
            "alloc": alloc,
            "txs": txs,
            "env": env,
        }

        encoded_input = str.encode(json.dumps(stdin))
        result = subprocess.run(
            args,
            input=encoded_input,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )

        if result.returncode != 0:
            raise Exception("failed to evaluate: " + result.stderr.decode())

        output = json.loads(result.stdout)

        if "alloc" not in output or "result" not in output:
            raise Exception("malformed result")

        if self.trace:
            receipts: List[Any] = output["result"]["receipts"]
            traces: List[List[Dict]] = []
            for i, r in enumerate(receipts):
                h = r["transactionHash"]
                trace_file_name = f"trace-{i}-{h}.jsonl"
                with open(os.path.join(temp_dir.name, trace_file_name), "r") as trace_file:
                    tx_traces: List[Dict] = []
                    for trace_line in trace_file.readlines():
                        tx_traces.append(json.loads(trace_line))
                    traces.append(tx_traces)
            self.append_traces(traces)

        temp_dir.cleanup()

        return output["alloc"], output["result"]

    def version(self) -> str:
        """
        Gets `evm` binary version.
        """
        if self.cached_version is None:
            result = subprocess.run(
                [str(self.binary), "-v"],
                stdout=subprocess.PIPE,
            )

            if result.returncode != 0:
                raise Exception("failed to evaluate: " + result.stderr.decode())

            self.cached_version = result.stdout.decode().strip()

        return self.cached_version

    def is_fork_supported(self, fork: Fork) -> bool:
        """
        Returns True if the fork is supported by the tool
        """
        return fork().name() in self.help_string

evaluate(alloc, txs, env, fork, chain_id=1, reward=0, eips=None)

Executes evm t8n with the specified arguments.

Source code in src/evm_transition_tool/geth.py 
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
def evaluate(
    self,
    alloc: Any,
    txs: Any,
    env: Any,
    fork: Fork,
    chain_id: int = 1,
    reward: int = 0,
    eips: Optional[List[int]] = None,
) -> Tuple[Dict[str, Any], Dict[str, Any]]:
    """
    Executes `evm t8n` with the specified arguments.
    """
    fork_name = fork.name()
    if eips is not None:
        fork_name = "+".join([fork_name] + [str(eip) for eip in eips])

    temp_dir = tempfile.TemporaryDirectory()

    if int(env["currentNumber"], 0) == 0:
        reward = -1
    args = [
        str(self.binary),
        "t8n",
        "--input.alloc=stdin",
        "--input.txs=stdin",
        "--input.env=stdin",
        "--output.result=stdout",
        "--output.alloc=stdout",
        "--output.body=txs.rlp",
        f"--output.basedir={temp_dir.name}",
        f"--state.fork={fork_name}",
        f"--state.chainid={chain_id}",
        f"--state.reward={reward}",
    ]

    if self.trace:
        args.append("--trace")

    stdin = {
        "alloc": alloc,
        "txs": txs,
        "env": env,
    }

    encoded_input = str.encode(json.dumps(stdin))
    result = subprocess.run(
        args,
        input=encoded_input,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )

    if result.returncode != 0:
        raise Exception("failed to evaluate: " + result.stderr.decode())

    output = json.loads(result.stdout)

    if "alloc" not in output or "result" not in output:
        raise Exception("malformed result")

    if self.trace:
        receipts: List[Any] = output["result"]["receipts"]
        traces: List[List[Dict]] = []
        for i, r in enumerate(receipts):
            h = r["transactionHash"]
            trace_file_name = f"trace-{i}-{h}.jsonl"
            with open(os.path.join(temp_dir.name, trace_file_name), "r") as trace_file:
                tx_traces: List[Dict] = []
                for trace_line in trace_file.readlines():
                    tx_traces.append(json.loads(trace_line))
                traces.append(tx_traces)
        self.append_traces(traces)

    temp_dir.cleanup()

    return output["alloc"], output["result"]

version()

Gets evm binary version.

Source code in src/evm_transition_tool/geth.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
def version(self) -> str:
    """
    Gets `evm` binary version.
    """
    if self.cached_version is None:
        result = subprocess.run(
            [str(self.binary), "-v"],
            stdout=subprocess.PIPE,
        )

        if result.returncode != 0:
            raise Exception("failed to evaluate: " + result.stderr.decode())

        self.cached_version = result.stdout.decode().strip()

    return self.cached_version

is_fork_supported(fork)

Returns True if the fork is supported by the tool

Source code in src/evm_transition_tool/geth.py 
141
142
143
144
145
def is_fork_supported(self, fork: Fork) -> bool:
    """
    Returns True if the fork is supported by the tool
    """
    return fork().name() in self.help_string

TransitionToolNotFoundInPath

Bases: Exception

Exception raised if the specified t8n tool is not found in the path

Source code in src/evm_transition_tool/transition_tool.py 
22
23
24
25
26
27
28
class TransitionToolNotFoundInPath(Exception):
    """Exception raised if the specified t8n tool is not found in the path"""

    def __init__(self, message="The transition tool was not found in the path", binary=None):
        if binary:
            message = f"{message} ({binary})"
        super().__init__(message)

EvmOneTransitionTool

Bases: TransitionTool

Evmone evmone-t8n Transition tool frontend wrapper class.

Source code in src/evm_transition_tool/evmone.py 
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
class EvmOneTransitionTool(TransitionTool):
    """
    Evmone `evmone-t8n` Transition tool frontend wrapper class.
    """

    default_binary = Path("evmone-t8n")
    detect_binary_pattern = compile(r"^evmone-t8n\b")

    binary: Path
    cached_version: Optional[str] = None
    trace: bool

    def __init__(
        self,
        *,
        binary: Optional[Path] = None,
        trace: bool = False,
    ):
        super().__init__(binary=binary, trace=trace)
        if self.trace:
            raise Exception("`evmone-t8n` does not support tracing.")

    def evaluate(
        self,
        alloc: Any,
        txs: Any,
        env: Any,
        fork: Fork,
        chain_id: int = 1,
        reward: int = 0,
        eips: Optional[List[int]] = None,
    ) -> Tuple[Dict[str, Any], Dict[str, Any]]:
        """
        Executes `evmone-t8n` with the specified arguments.
        """
        fork_name = fork.name()
        if eips is not None:
            fork_name = "+".join([fork_name] + [str(eip) for eip in eips])

        temp_dir = tempfile.TemporaryDirectory()

        input_contents = {
            "alloc": alloc,
            "env": env,
            "txs": txs,
        }
        input_paths = {
            k: os.path.join(temp_dir.name, f"input_{k}.json") for k in input_contents.keys()
        }
        for key, val in input_contents.items():
            file_path = os.path.join(temp_dir.name, f"input_{key}.json")
            write_json_file(val, file_path)

        # Construct args for evmone-t8n binary
        args = [
            str(self.binary),
            "--state.fork",
            fork_name,
            "--input.alloc",
            input_paths["alloc"],
            "--input.env",
            input_paths["env"],
            "--input.txs",
            input_paths["txs"],
            "--output.basedir",
            temp_dir.name,
            "--output.result",
            "output_result.json",
            "--output.alloc",
            "output_alloc.json",
            "--output.body",
            "txs.rlp",
            "--state.reward",
            str(reward),
            "--state.chainid",
            str(chain_id),
        ]
        result = subprocess.run(
            args,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )

        if result.returncode != 0:
            raise Exception("failed to evaluate: " + result.stderr.decode())

        output_paths = {
            "alloc": os.path.join(temp_dir.name, "output_alloc.json"),
            "result": os.path.join(temp_dir.name, "output_result.json"),
        }

        output_contents = {}
        for key, file_path in output_paths.items():
            with open(file_path, "r+") as file:
                contents = json.load(file)
                file.seek(0)
                json.dump(contents, file, ensure_ascii=False, indent=4)
                file.truncate()
                output_contents[key] = contents

        temp_dir.cleanup()

        return output_contents["alloc"], output_contents["result"]

    def version(self) -> str:
        """
        Gets `evmone-t8n` binary version.
        """
        if self.cached_version is None:
            result = subprocess.run(
                [str(self.binary), "-v"],
                stdout=subprocess.PIPE,
            )

            if result.returncode != 0:
                raise Exception("failed to evaluate: " + result.stderr.decode())

            self.cached_version = result.stdout.decode().strip()

        return self.cached_version

    def is_fork_supported(self, fork: Fork) -> bool:
        """
        Returns True if the fork is supported by the tool.
        Currently, evmone-t8n provides no way to determine supported forks.
        """
        return True

evaluate(alloc, txs, env, fork, chain_id=1, reward=0, eips=None)

Executes evmone-t8n with the specified arguments.

Source code in src/evm_transition_tool/evmone.py 
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
def evaluate(
    self,
    alloc: Any,
    txs: Any,
    env: Any,
    fork: Fork,
    chain_id: int = 1,
    reward: int = 0,
    eips: Optional[List[int]] = None,
) -> Tuple[Dict[str, Any], Dict[str, Any]]:
    """
    Executes `evmone-t8n` with the specified arguments.
    """
    fork_name = fork.name()
    if eips is not None:
        fork_name = "+".join([fork_name] + [str(eip) for eip in eips])

    temp_dir = tempfile.TemporaryDirectory()

    input_contents = {
        "alloc": alloc,
        "env": env,
        "txs": txs,
    }
    input_paths = {
        k: os.path.join(temp_dir.name, f"input_{k}.json") for k in input_contents.keys()
    }
    for key, val in input_contents.items():
        file_path = os.path.join(temp_dir.name, f"input_{key}.json")
        write_json_file(val, file_path)

    # Construct args for evmone-t8n binary
    args = [
        str(self.binary),
        "--state.fork",
        fork_name,
        "--input.alloc",
        input_paths["alloc"],
        "--input.env",
        input_paths["env"],
        "--input.txs",
        input_paths["txs"],
        "--output.basedir",
        temp_dir.name,
        "--output.result",
        "output_result.json",
        "--output.alloc",
        "output_alloc.json",
        "--output.body",
        "txs.rlp",
        "--state.reward",
        str(reward),
        "--state.chainid",
        str(chain_id),
    ]
    result = subprocess.run(
        args,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )

    if result.returncode != 0:
        raise Exception("failed to evaluate: " + result.stderr.decode())

    output_paths = {
        "alloc": os.path.join(temp_dir.name, "output_alloc.json"),
        "result": os.path.join(temp_dir.name, "output_result.json"),
    }

    output_contents = {}
    for key, file_path in output_paths.items():
        with open(file_path, "r+") as file:
            contents = json.load(file)
            file.seek(0)
            json.dump(contents, file, ensure_ascii=False, indent=4)
            file.truncate()
            output_contents[key] = contents

    temp_dir.cleanup()

    return output_contents["alloc"], output_contents["result"]

version()

Gets evmone-t8n binary version.

Source code in src/evm_transition_tool/evmone.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
def version(self) -> str:
    """
    Gets `evmone-t8n` binary version.
    """
    if self.cached_version is None:
        result = subprocess.run(
            [str(self.binary), "-v"],
            stdout=subprocess.PIPE,
        )

        if result.returncode != 0:
            raise Exception("failed to evaluate: " + result.stderr.decode())

        self.cached_version = result.stdout.decode().strip()

    return self.cached_version

is_fork_supported(fork)

Returns True if the fork is supported by the tool. Currently, evmone-t8n provides no way to determine supported forks.

Source code in src/evm_transition_tool/evmone.py 
146
147
148
149
150
151
def is_fork_supported(self, fork: Fork) -> bool:
    """
    Returns True if the fork is supported by the tool.
    Currently, evmone-t8n provides no way to determine supported forks.
    """
    return True

TransitionTool

Transition tool abstract base class which should be inherited by all transition tool implementations.

Source code in src/evm_transition_tool/transition_tool.py 
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
class TransitionTool:
    """
    Transition tool abstract base class which should be inherited by all transition tool
    implementations.
    """

    traces: List[List[List[Dict]]] | None = None

    registered_tools: List[Type["TransitionTool"]] = []
    default_tool: Optional[Type["TransitionTool"]] = None
    default_binary: Path
    detect_binary_pattern: Pattern
    version_flag: str = "-v"

    # Abstract methods that each tool must implement

    @abstractmethod
    def __init__(
        self,
        *,
        binary: Optional[Path] = None,
        trace: bool = False,
    ):
        """
        Abstract initialization method that all subclasses must implement.
        """
        if binary is None:
            binary = self.default_binary
        else:
            # improve behavior of which by resolving the path: ~/relative paths don't work
            resolved_path = Path(os.path.expanduser(binary)).resolve()
            if resolved_path.exists():
                binary = resolved_path
        binary = shutil.which(binary)  # type: ignore
        if not binary:
            raise TransitionToolNotFoundInPath(binary=binary)
        self.binary = Path(binary)
        self.trace = trace

    def __init_subclass__(cls):
        """
        Registers all subclasses of TransitionTool as possible tools.
        """
        TransitionTool.register_tool(cls)

    @classmethod
    def register_tool(cls, tool_subclass: Type["TransitionTool"]):
        """
        Registers a given subclass as tool option.
        """
        cls.registered_tools.append(tool_subclass)

    @classmethod
    def set_default_tool(cls, tool_subclass: Type["TransitionTool"]):
        """
        Registers the default tool subclass.
        """
        cls.default_tool = tool_subclass

    @classmethod
    def from_binary_path(cls, *, binary_path: Optional[Path], **kwargs) -> "TransitionTool":
        """
        Instantiates the appropriate TransitionTool subclass derived from the
        tool's binary path.
        """
        assert cls.default_tool is not None, "default transition tool was never set"

        if binary_path is None:
            return cls.default_tool(binary=binary_path, **kwargs)

        resolved_path = Path(os.path.expanduser(binary_path)).resolve()
        if resolved_path.exists():
            binary_path = resolved_path
        binary = shutil.which(binary_path)  # type: ignore

        if not binary:
            raise TransitionToolNotFoundInPath(binary=binary)

        binary = Path(binary)

        # Group the tools by version flag, so we only have to call the tool once for all the
        # classes that share the same version flag
        for version_flag, subclasses in groupby(
            cls.registered_tools, key=lambda x: x.version_flag
        ):
            try:
                with os.popen(f"{binary} {version_flag}") as f:
                    binary_output = f.read()
            except Exception:
                # If the tool doesn't support the version flag,
                # we'll get an non-zero exit code.
                continue
            for subclass in subclasses:
                if subclass.detect_binary(binary_output):
                    return subclass(binary=binary, **kwargs)

        raise UnknownTransitionTool(f"Unknown transition tool binary: {binary_path}")

    @classmethod
    def detect_binary(cls, binary_output: str) -> bool:
        """
        Returns True if the binary matches the tool
        """
        assert cls.detect_binary_pattern is not None

        return cls.detect_binary_pattern.match(binary_output) is not None

    @abstractmethod
    def evaluate(
        self,
        alloc: Any,
        txs: Any,
        env: Any,
        fork: Fork,
        chain_id: int = 1,
        reward: int = 0,
        eips: Optional[List[int]] = None,
    ) -> Tuple[Dict[str, Any], Dict[str, Any]]:
        """
        Simulate a state transition with specified parameters
        """
        pass

    @abstractmethod
    def version(self) -> str:
        """
        Return name and version of tool used to state transition
        """
        pass

    @abstractmethod
    def is_fork_supported(self, fork: Fork) -> bool:
        """
        Returns True if the fork is supported by the tool
        """
        pass

    def reset_traces(self):
        """
        Resets the internal trace storage for a new test to begin
        """
        self.traces = None

    def append_traces(self, new_traces: List[List[Dict]]):
        """
        Appends a list of traces of a state transition to the current list
        """
        if self.traces is None:
            self.traces = []
        self.traces.append(new_traces)

    def get_traces(self) -> List[List[List[Dict]]] | None:
        """
        Returns the accumulated traces
        """
        return self.traces

    def calc_state_root(self, alloc: Any, fork: Fork) -> bytes:
        """
        Calculate the state root for the given `alloc`.
        """
        env: Dict[str, Any] = {
            "currentCoinbase": "0x0000000000000000000000000000000000000000",
            "currentDifficulty": "0x0",
            "currentGasLimit": "0x0",
            "currentNumber": "0",
            "currentTimestamp": "0",
        }

        if fork.header_base_fee_required(0, 0):
            env["currentBaseFee"] = "7"

        if fork.header_prev_randao_required(0, 0):
            env["currentRandom"] = "0"

        if fork.header_withdrawals_required(0, 0):
            env["withdrawals"] = []

        _, result = self.evaluate(alloc, [], env, fork)
        state_root = result.get("stateRoot")
        if state_root is None or not isinstance(state_root, str):
            raise Exception("Unable to calculate state root")
        return bytes.fromhex(state_root[2:])

    def calc_withdrawals_root(self, withdrawals: Any, fork: Fork) -> bytes:
        """
        Calculate the state root for the given `alloc`.
        """
        if type(withdrawals) is list and len(withdrawals) == 0:
            # Optimize returning the empty root immediately
            return bytes.fromhex(
                "56e81f171bcc55a6ff8345e692c0f86e5b48e01b996cadc001622fb5e363b421"
            )

        env: Dict[str, Any] = {
            "currentCoinbase": "0x0000000000000000000000000000000000000000",
            "currentDifficulty": "0x0",
            "currentGasLimit": "0x0",
            "currentNumber": "0",
            "currentTimestamp": "0",
            "withdrawals": withdrawals,
        }

        if fork.header_base_fee_required(0, 0):
            env["currentBaseFee"] = "7"

        if fork.header_prev_randao_required(0, 0):
            env["currentRandom"] = "0"

        if fork.header_excess_data_gas_required(0, 0):
            env["currentExcessDataGas"] = "0"

        _, result = self.evaluate({}, [], env, fork)
        withdrawals_root = result.get("withdrawalsRoot")
        if withdrawals_root is None:
            raise Exception(
                "Unable to calculate withdrawals root: no value returned from transition tool"
            )
        if type(withdrawals_root) is not str:
            raise Exception(
                "Unable to calculate withdrawals root: "
                + "incorrect type returned from transition tool: "
                + f"{withdrawals_root}"
            )
        return bytes.fromhex(withdrawals_root[2:])

__init__(*, binary=None, trace=False) abstractmethod

Abstract initialization method that all subclasses must implement.

Source code in src/evm_transition_tool/transition_tool.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
@abstractmethod
def __init__(
    self,
    *,
    binary: Optional[Path] = None,
    trace: bool = False,
):
    """
    Abstract initialization method that all subclasses must implement.
    """
    if binary is None:
        binary = self.default_binary
    else:
        # improve behavior of which by resolving the path: ~/relative paths don't work
        resolved_path = Path(os.path.expanduser(binary)).resolve()
        if resolved_path.exists():
            binary = resolved_path
    binary = shutil.which(binary)  # type: ignore
    if not binary:
        raise TransitionToolNotFoundInPath(binary=binary)
    self.binary = Path(binary)
    self.trace = trace

__init_subclass__()

Registers all subclasses of TransitionTool as possible tools.

Source code in src/evm_transition_tool/transition_tool.py 
70
71
72
73
74
def __init_subclass__(cls):
    """
    Registers all subclasses of TransitionTool as possible tools.
    """
    TransitionTool.register_tool(cls)

register_tool(tool_subclass) classmethod

Registers a given subclass as tool option.

Source code in src/evm_transition_tool/transition_tool.py 
76
77
78
79
80
81
@classmethod
def register_tool(cls, tool_subclass: Type["TransitionTool"]):
    """
    Registers a given subclass as tool option.
    """
    cls.registered_tools.append(tool_subclass)

set_default_tool(tool_subclass) classmethod

Registers the default tool subclass.

Source code in src/evm_transition_tool/transition_tool.py 
83
84
85
86
87
88
@classmethod
def set_default_tool(cls, tool_subclass: Type["TransitionTool"]):
    """
    Registers the default tool subclass.
    """
    cls.default_tool = tool_subclass

from_binary_path(*, binary_path, **kwargs) classmethod

Instantiates the appropriate TransitionTool subclass derived from the tool's binary path.

Source code in src/evm_transition_tool/transition_tool.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
@classmethod
def from_binary_path(cls, *, binary_path: Optional[Path], **kwargs) -> "TransitionTool":
    """
    Instantiates the appropriate TransitionTool subclass derived from the
    tool's binary path.
    """
    assert cls.default_tool is not None, "default transition tool was never set"

    if binary_path is None:
        return cls.default_tool(binary=binary_path, **kwargs)

    resolved_path = Path(os.path.expanduser(binary_path)).resolve()
    if resolved_path.exists():
        binary_path = resolved_path
    binary = shutil.which(binary_path)  # type: ignore

    if not binary:
        raise TransitionToolNotFoundInPath(binary=binary)

    binary = Path(binary)

    # Group the tools by version flag, so we only have to call the tool once for all the
    # classes that share the same version flag
    for version_flag, subclasses in groupby(
        cls.registered_tools, key=lambda x: x.version_flag
    ):
        try:
            with os.popen(f"{binary} {version_flag}") as f:
                binary_output = f.read()
        except Exception:
            # If the tool doesn't support the version flag,
            # we'll get an non-zero exit code.
            continue
        for subclass in subclasses:
            if subclass.detect_binary(binary_output):
                return subclass(binary=binary, **kwargs)

    raise UnknownTransitionTool(f"Unknown transition tool binary: {binary_path}")

detect_binary(binary_output) classmethod

Returns True if the binary matches the tool

Source code in src/evm_transition_tool/transition_tool.py 
129
130
131
132
133
134
135
136
@classmethod
def detect_binary(cls, binary_output: str) -> bool:
    """
    Returns True if the binary matches the tool
    """
    assert cls.detect_binary_pattern is not None

    return cls.detect_binary_pattern.match(binary_output) is not None

evaluate(alloc, txs, env, fork, chain_id=1, reward=0, eips=None) abstractmethod

Simulate a state transition with specified parameters

Source code in src/evm_transition_tool/transition_tool.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
@abstractmethod
def evaluate(
    self,
    alloc: Any,
    txs: Any,
    env: Any,
    fork: Fork,
    chain_id: int = 1,
    reward: int = 0,
    eips: Optional[List[int]] = None,
) -> Tuple[Dict[str, Any], Dict[str, Any]]:
    """
    Simulate a state transition with specified parameters
    """
    pass

version() abstractmethod

Return name and version of tool used to state transition

Source code in src/evm_transition_tool/transition_tool.py 
154
155
156
157
158
159
@abstractmethod
def version(self) -> str:
    """
    Return name and version of tool used to state transition
    """
    pass

is_fork_supported(fork) abstractmethod

Returns True if the fork is supported by the tool

Source code in src/evm_transition_tool/transition_tool.py 
161
162
163
164
165
166
@abstractmethod
def is_fork_supported(self, fork: Fork) -> bool:
    """
    Returns True if the fork is supported by the tool
    """
    pass

reset_traces()

Resets the internal trace storage for a new test to begin

Source code in src/evm_transition_tool/transition_tool.py 
168
169
170
171
172
def reset_traces(self):
    """
    Resets the internal trace storage for a new test to begin
    """
    self.traces = None

append_traces(new_traces)

Appends a list of traces of a state transition to the current list

Source code in src/evm_transition_tool/transition_tool.py 
174
175
176
177
178
179
180
def append_traces(self, new_traces: List[List[Dict]]):
    """
    Appends a list of traces of a state transition to the current list
    """
    if self.traces is None:
        self.traces = []
    self.traces.append(new_traces)

get_traces()

Returns the accumulated traces

Source code in src/evm_transition_tool/transition_tool.py 
182
183
184
185
186
def get_traces(self) -> List[List[List[Dict]]] | None:
    """
    Returns the accumulated traces
    """
    return self.traces

calc_state_root(alloc, fork)

Calculate the state root for the given alloc.

Source code in src/evm_transition_tool/transition_tool.py 
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
def calc_state_root(self, alloc: Any, fork: Fork) -> bytes:
    """
    Calculate the state root for the given `alloc`.
    """
    env: Dict[str, Any] = {
        "currentCoinbase": "0x0000000000000000000000000000000000000000",
        "currentDifficulty": "0x0",
        "currentGasLimit": "0x0",
        "currentNumber": "0",
        "currentTimestamp": "0",
    }

    if fork.header_base_fee_required(0, 0):
        env["currentBaseFee"] = "7"

    if fork.header_prev_randao_required(0, 0):
        env["currentRandom"] = "0"

    if fork.header_withdrawals_required(0, 0):
        env["withdrawals"] = []

    _, result = self.evaluate(alloc, [], env, fork)
    state_root = result.get("stateRoot")
    if state_root is None or not isinstance(state_root, str):
        raise Exception("Unable to calculate state root")
    return bytes.fromhex(state_root[2:])

calc_withdrawals_root(withdrawals, fork)

Calculate the state root for the given alloc.

Source code in src/evm_transition_tool/transition_tool.py 
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
def calc_withdrawals_root(self, withdrawals: Any, fork: Fork) -> bytes:
    """
    Calculate the state root for the given `alloc`.
    """
    if type(withdrawals) is list and len(withdrawals) == 0:
        # Optimize returning the empty root immediately
        return bytes.fromhex(
            "56e81f171bcc55a6ff8345e692c0f86e5b48e01b996cadc001622fb5e363b421"
        )

    env: Dict[str, Any] = {
        "currentCoinbase": "0x0000000000000000000000000000000000000000",
        "currentDifficulty": "0x0",
        "currentGasLimit": "0x0",
        "currentNumber": "0",
        "currentTimestamp": "0",
        "withdrawals": withdrawals,
    }

    if fork.header_base_fee_required(0, 0):
        env["currentBaseFee"] = "7"

    if fork.header_prev_randao_required(0, 0):
        env["currentRandom"] = "0"

    if fork.header_excess_data_gas_required(0, 0):
        env["currentExcessDataGas"] = "0"

    _, result = self.evaluate({}, [], env, fork)
    withdrawals_root = result.get("withdrawalsRoot")
    if withdrawals_root is None:
        raise Exception(
            "Unable to calculate withdrawals root: no value returned from transition tool"
        )
    if type(withdrawals_root) is not str:
        raise Exception(
            "Unable to calculate withdrawals root: "
            + "incorrect type returned from transition tool: "
            + f"{withdrawals_root}"
        )
    return bytes.fromhex(withdrawals_root[2:])