Skip to content

EVM Transition Tool Package

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

BesuTransitionTool

Bases: TransitionTool

Besu EvmTool Transition tool frontend wrapper class.

Source code in src/evm_transition_tool/besu.py 
 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
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
class BesuTransitionTool(TransitionTool):
    """
    Besu EvmTool Transition tool frontend wrapper class.
    """

    default_binary = Path("evm")
    detect_binary_pattern = compile(r"^Hyperledger Besu evm .*$")

    binary: Path
    cached_version: Optional[str] = None
    trace: bool
    process: Optional[subprocess.Popen] = None
    server_url: str

    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 start_server(self):
        """
        Starts the t8n-server process, extracts the port, and leaves it running for future re-use.
        """
        self.process = subprocess.Popen(
            args=[
                str(self.binary),
                "t8n-server",
                "--port=0",  # OS assigned server port
            ],
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
        )

        while True:
            line = str(self.process.stdout.readline())

            if not line or "Failed to start transition server" in line:
                raise Exception("Failed starting Besu subprocess\n" + line)
            if "Transition server listening on" in line:
                port = re.search("Transition server listening on ([0-9]+)", line).group(1)
                self.server_url = f"http://localhost:{port}/"
                break

    def shutdown(self):
        """
        Stops the t8n-server process if it was started
        """
        if self.process:
            self.process.kill()

    def evaluate(
        self,
        alloc: Any,
        txs: Any,
        env: Any,
        fork_name: str,
        chain_id: int = 1,
        reward: int = 0,
        eips: Optional[List[int]] = None,
        debug_output_path: str = "",
    ) -> Tuple[Dict[str, Any], Dict[str, Any]]:
        """
        Executes `evm t8n` with the specified arguments.
        """
        if not self.process:
            self.start_server()

        if eips is not None:
            fork_name = "+".join([fork_name] + [str(eip) for eip in eips])

        if self.trace:
            raise Exception("Besu `t8n-server` does not support tracing.")

        input_json = {
            "alloc": alloc,
            "txs": txs,
            "env": env,
        }
        state_json = {
            "fork": fork_name,
            "chainid": chain_id,
            "reward": reward,
        }

        post_data = {"state": state_json, "input": input_json}

        if debug_output_path:
            post_data_string = json.dumps(post_data, indent=4)
            additional_indent = " " * 16  # for pretty indentation in t8n.sh
            indented_post_data_string = "{\n" + "\n".join(
                additional_indent + line for line in post_data_string[1:].splitlines()
            )
            t8n_script = textwrap.dedent(
                f"""\
                #!/bin/bash
                # Use $1 as t8n-server port if provided, else default to 3000
                PORT=${{1:-3000}}
                curl http://localhost:${{PORT}}/ -X POST -H "Content-Type: application/json" \\
                --data '{indented_post_data_string}'
                """  # noqa: E221
            )
            dump_files_to_directory(
                debug_output_path,
                {
                    "state.json": state_json,
                    "input/alloc.json": input_json["alloc"],
                    "input/env.json": input_json["env"],
                    "input/txs.json": input_json["txs"],
                    "t8n.sh+x": t8n_script,
                },
            )

        response = requests.post(self.server_url, json=post_data, timeout=5)
        response.raise_for_status()  # exception visible in pytest failure output
        output = response.json()

        if debug_output_path:
            dump_files_to_directory(
                debug_output_path,
                {
                    "response.txt": response.text,
                    "status_code.txt": response.status_code,
                    "time_elapsed_seconds.txt": response.elapsed.total_seconds(),
                },
            )

        if response.status_code != 200:
            raise Exception(
                f"t8n-server returned status code {response.status_code}, "
                f"response: {response.text}"
            )
        if not all([x in output for x in ["alloc", "result", "body"]]):
            raise Exception(
                "Malformed t8n output: missing 'alloc', 'result' or 'body', server response: "
                f"{response.text}"
            )

        if debug_output_path:
            dump_files_to_directory(
                debug_output_path,
                {
                    "output/alloc.json": output["alloc"],
                    "output/result.json": output["result"],
                    "output/txs.rlp": output["body"],
                },
            )

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

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

start_server()

Starts the t8n-server process, extracts the port, and leaves it running for future re-use.

Source code in src/evm_transition_tool/besu.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def start_server(self):
    """
    Starts the t8n-server process, extracts the port, and leaves it running for future re-use.
    """
    self.process = subprocess.Popen(
        args=[
            str(self.binary),
            "t8n-server",
            "--port=0",  # OS assigned server port
        ],
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
    )

    while True:
        line = str(self.process.stdout.readline())

        if not line or "Failed to start transition server" in line:
            raise Exception("Failed starting Besu subprocess\n" + line)
        if "Transition server listening on" in line:
            port = re.search("Transition server listening on ([0-9]+)", line).group(1)
            self.server_url = f"http://localhost:{port}/"
            break

shutdown()

Stops the t8n-server process if it was started

Source code in src/evm_transition_tool/besu.py 
74
75
76
77
78
79
def shutdown(self):
    """
    Stops the t8n-server process if it was started
    """
    if self.process:
        self.process.kill()

evaluate(alloc, txs, env, fork_name, chain_id=1, reward=0, eips=None, debug_output_path='')

Executes evm t8n with the specified arguments.

Source code in src/evm_transition_tool/besu.py 
 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
def evaluate(
    self,
    alloc: Any,
    txs: Any,
    env: Any,
    fork_name: str,
    chain_id: int = 1,
    reward: int = 0,
    eips: Optional[List[int]] = None,
    debug_output_path: str = "",
) -> Tuple[Dict[str, Any], Dict[str, Any]]:
    """
    Executes `evm t8n` with the specified arguments.
    """
    if not self.process:
        self.start_server()

    if eips is not None:
        fork_name = "+".join([fork_name] + [str(eip) for eip in eips])

    if self.trace:
        raise Exception("Besu `t8n-server` does not support tracing.")

    input_json = {
        "alloc": alloc,
        "txs": txs,
        "env": env,
    }
    state_json = {
        "fork": fork_name,
        "chainid": chain_id,
        "reward": reward,
    }

    post_data = {"state": state_json, "input": input_json}

    if debug_output_path:
        post_data_string = json.dumps(post_data, indent=4)
        additional_indent = " " * 16  # for pretty indentation in t8n.sh
        indented_post_data_string = "{\n" + "\n".join(
            additional_indent + line for line in post_data_string[1:].splitlines()
        )
        t8n_script = textwrap.dedent(
            f"""\
            #!/bin/bash
            # Use $1 as t8n-server port if provided, else default to 3000
            PORT=${{1:-3000}}
            curl http://localhost:${{PORT}}/ -X POST -H "Content-Type: application/json" \\
            --data '{indented_post_data_string}'
            """  # noqa: E221
        )
        dump_files_to_directory(
            debug_output_path,
            {
                "state.json": state_json,
                "input/alloc.json": input_json["alloc"],
                "input/env.json": input_json["env"],
                "input/txs.json": input_json["txs"],
                "t8n.sh+x": t8n_script,
            },
        )

    response = requests.post(self.server_url, json=post_data, timeout=5)
    response.raise_for_status()  # exception visible in pytest failure output
    output = response.json()

    if debug_output_path:
        dump_files_to_directory(
            debug_output_path,
            {
                "response.txt": response.text,
                "status_code.txt": response.status_code,
                "time_elapsed_seconds.txt": response.elapsed.total_seconds(),
            },
        )

    if response.status_code != 200:
        raise Exception(
            f"t8n-server returned status code {response.status_code}, "
            f"response: {response.text}"
        )
    if not all([x in output for x in ["alloc", "result", "body"]]):
        raise Exception(
            "Malformed t8n output: missing 'alloc', 'result' or 'body', server response: "
            f"{response.text}"
        )

    if debug_output_path:
        dump_files_to_directory(
            debug_output_path,
            {
                "output/alloc.json": output["alloc"],
                "output/result.json": output["result"],
                "output/txs.rlp": output["body"],
            },
        )

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

is_fork_supported(fork)

Returns True if the fork is supported by the tool

Source code in src/evm_transition_tool/besu.py 
180
181
182
183
184
def is_fork_supported(self, fork: Fork) -> bool:
    """
    Returns True if the fork is supported by the tool
    """
    return fork.fork() in self.help_string

EvmOneTransitionTool

Bases: TransitionTool

Evmone evmone-t8n Transition tool interface wrapper class.

Source code in src/evm_transition_tool/evmone.py 
 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
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
class EvmOneTransitionTool(TransitionTool):
    """
    Evmone `evmone-t8n` Transition tool interface 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)

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

        temp_dir = tempfile.TemporaryDirectory()
        os.mkdir(os.path.join(temp_dir.name, "input"))
        os.mkdir(os.path.join(temp_dir.name, "output"))

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

        input_paths = {
            k: os.path.join(temp_dir.name, "input", f"{k}.json") for k in input_contents.keys()
        }
        for key, file_path in input_paths.items():
            write_json_file(input_contents[key], file_path)

        output_paths = {
            output: os.path.join("output", f"{output}.json") for output in ["alloc", "result"]
        }
        output_paths["body"] = os.path.join("output", "txs.rlp")

        # 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_paths["result"],
            "--output.alloc",
            output_paths["alloc"],
            "--output.body",
            output_paths["body"],
            "--state.reward",
            str(reward),
            "--state.chainid",
            str(chain_id),
        ]

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

        result = subprocess.run(
            args,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )

        if debug_output_path:
            if os.path.exists(debug_output_path):
                shutil.rmtree(debug_output_path)
            shutil.copytree(temp_dir.name, debug_output_path)
            t8n_output_base_dir = os.path.join(debug_output_path, "t8n.sh.out")
            t8n_call = " ".join(args)
            for file_path in input_paths.values():  # update input paths
                t8n_call = t8n_call.replace(
                    os.path.dirname(file_path), os.path.join(debug_output_path, "input")
                )
            t8n_call = t8n_call.replace(  # use a new output path for basedir and outputs
                temp_dir.name,
                t8n_output_base_dir,
            )
            t8n_script = textwrap.dedent(
                f"""\
                #!/bin/bash
                rm -rf {debug_output_path}/t8n.sh.out  # hard-coded to avoid surprises
                mkdir -p {debug_output_path}/t8n.sh.out/output
                {t8n_call}
                """
            )
            dump_files_to_directory(
                debug_output_path,
                {
                    "args.py": args,
                    "returncode.txt": result.returncode,
                    "stdout.txt": result.stdout.decode(),
                    "stderr.txt": result.stderr.decode(),
                    "t8n.sh+x": t8n_script,
                },
            )

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

        for key, file_path in output_paths.items():
            output_paths[key] = os.path.join(temp_dir.name, file_path)

        output_contents = {}
        for key, file_path in output_paths.items():
            if "txs.rlp" in file_path:
                continue
            with open(file_path, "r+") as file:
                output_contents[key] = json.load(file)

        if self.trace:
            self.collect_traces(output_contents["result"]["receipts"], temp_dir, debug_output_path)

        temp_dir.cleanup()

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

    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_name, chain_id=1, reward=0, eips=None, debug_output_path='')

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

    temp_dir = tempfile.TemporaryDirectory()
    os.mkdir(os.path.join(temp_dir.name, "input"))
    os.mkdir(os.path.join(temp_dir.name, "output"))

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

    input_paths = {
        k: os.path.join(temp_dir.name, "input", f"{k}.json") for k in input_contents.keys()
    }
    for key, file_path in input_paths.items():
        write_json_file(input_contents[key], file_path)

    output_paths = {
        output: os.path.join("output", f"{output}.json") for output in ["alloc", "result"]
    }
    output_paths["body"] = os.path.join("output", "txs.rlp")

    # 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_paths["result"],
        "--output.alloc",
        output_paths["alloc"],
        "--output.body",
        output_paths["body"],
        "--state.reward",
        str(reward),
        "--state.chainid",
        str(chain_id),
    ]

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

    result = subprocess.run(
        args,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )

    if debug_output_path:
        if os.path.exists(debug_output_path):
            shutil.rmtree(debug_output_path)
        shutil.copytree(temp_dir.name, debug_output_path)
        t8n_output_base_dir = os.path.join(debug_output_path, "t8n.sh.out")
        t8n_call = " ".join(args)
        for file_path in input_paths.values():  # update input paths
            t8n_call = t8n_call.replace(
                os.path.dirname(file_path), os.path.join(debug_output_path, "input")
            )
        t8n_call = t8n_call.replace(  # use a new output path for basedir and outputs
            temp_dir.name,
            t8n_output_base_dir,
        )
        t8n_script = textwrap.dedent(
            f"""\
            #!/bin/bash
            rm -rf {debug_output_path}/t8n.sh.out  # hard-coded to avoid surprises
            mkdir -p {debug_output_path}/t8n.sh.out/output
            {t8n_call}
            """
        )
        dump_files_to_directory(
            debug_output_path,
            {
                "args.py": args,
                "returncode.txt": result.returncode,
                "stdout.txt": result.stdout.decode(),
                "stderr.txt": result.stderr.decode(),
                "t8n.sh+x": t8n_script,
            },
        )

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

    for key, file_path in output_paths.items():
        output_paths[key] = os.path.join(temp_dir.name, file_path)

    output_contents = {}
    for key, file_path in output_paths.items():
        if "txs.rlp" in file_path:
            continue
        with open(file_path, "r+") as file:
            output_contents[key] = json.load(file)

    if self.trace:
        self.collect_traces(output_contents["result"]["receipts"], temp_dir, debug_output_path)

    temp_dir.cleanup()

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

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 
173
174
175
176
177
178
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

ExecutionSpecsTransitionTool

Bases: GethTransitionTool

Ethereum Specs ethereum-spec-evm Transition tool interface wrapper class.

The behavior of this tool is almost identical to go-ethereum's evm t8n command.

note: Using the latest version of the ethereum-spec-evm tool:

As the `ethereum` package provided by `execution-specs` is a requirement of
`execution-spec-tests`, the `ethereum-spec-evm` is already installed in the
virtual environment where `execution-spec-tests` is installed
(via `pip install -e .`). Therefore, the `ethereum-spec-evm` transition tool
can be used to fill tests via:

```console
    fill --evm-bin=ethereum-spec-evm
```

To ensure you're using the latest version of `ethereum-spec-evm` you can run:

```
pip install --force-reinstall -e .
```

or

```
pip install --force-reinstall -e .[docs,lint,tests]
```

as appropriate.

note: Using a specific version of the ethereum-spec-evm tool:

1. Create a virtual environment and activate it:
    ```
    python -m venv venv-execution-specs
    source venv-execution-specs/bin/activate
    ```
2. Clone the ethereum/execution-specs repository, change working directory to it and
    retrieve the desired version of the repository:
    ```
    git clone git@github.com:ethereum/execution-specs.git
    cd execution-specs
    git checkout <version>
    ```
3. Install the packages provided by the repository:
    ```
    pip install -e .
    ```
    Check that the `ethereum-spec-evm` command is available:
    ```
    ethereum-spec-evm --help
    ```
4. Clone the ethereum/execution-specs-tests repository and change working directory to it:
    ```
    cd ..
    git clone git@github.com:ethereum/execution-spec-tests.git
    cd execution-spec-tests
    ```
5. Install the packages provided by the ethereum/execution-spec-tests repository:
    ```
    pip install -e .
    ```
6. Run the tests, specifying the `ethereum-spec-evm` command as the transition tool:
    ```
    fill --evm-bin=path/to/venv-execution-specs/ethereum-spec-evm
    ```
Source code in src/evm_transition_tool/execution_specs.py 
 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
class ExecutionSpecsTransitionTool(GethTransitionTool):
    """
    Ethereum Specs `ethereum-spec-evm` Transition tool interface wrapper class.

    The behavior of this tool is almost identical to go-ethereum's `evm t8n` command.

    note: Using the latest version of the `ethereum-spec-evm` tool:

        As the `ethereum` package provided by `execution-specs` is a requirement of
        `execution-spec-tests`, the `ethereum-spec-evm` is already installed in the
        virtual environment where `execution-spec-tests` is installed
        (via `pip install -e .`). Therefore, the `ethereum-spec-evm` transition tool
        can be used to fill tests via:

        ```console
            fill --evm-bin=ethereum-spec-evm
        ```

        To ensure you're using the latest version of `ethereum-spec-evm` you can run:

        ```
        pip install --force-reinstall -e .
        ```

        or

        ```
        pip install --force-reinstall -e .[docs,lint,tests]
        ```

        as appropriate.

    note: Using a specific version of the `ethereum-spec-evm` tool:

        1. Create a virtual environment and activate it:
            ```
            python -m venv venv-execution-specs
            source venv-execution-specs/bin/activate
            ```
        2. Clone the ethereum/execution-specs repository, change working directory to it and
            retrieve the desired version of the repository:
            ```
            git clone git@github.com:ethereum/execution-specs.git
            cd execution-specs
            git checkout <version>
            ```
        3. Install the packages provided by the repository:
            ```
            pip install -e .
            ```
            Check that the `ethereum-spec-evm` command is available:
            ```
            ethereum-spec-evm --help
            ```
        4. Clone the ethereum/execution-specs-tests repository and change working directory to it:
            ```
            cd ..
            git clone git@github.com:ethereum/execution-spec-tests.git
            cd execution-spec-tests
            ```
        5. Install the packages provided by the ethereum/execution-spec-tests repository:
            ```
            pip install -e .
            ```
        6. Run the tests, specifying the `ethereum-spec-evm` command as the transition tool:
            ```
            fill --evm-bin=path/to/venv-execution-specs/ethereum-spec-evm
            ```
    """

    default_binary = Path("ethereum-spec-evm")
    detect_binary_pattern = compile(r"^ethereum-spec-evm\b")
    statetest_subcommand: Optional[str] = None
    blocktest_subcommand: Optional[str] = None

    def is_fork_supported(self, fork: Fork) -> bool:
        """
        Returns True if the fork is supported by the tool.
        Currently, ethereum-spec-evm provides no way to determine supported forks.
        """
        return fork not in UNSUPPORTED_FORKS

is_fork_supported(fork)

Returns True if the fork is supported by the tool. Currently, ethereum-spec-evm provides no way to determine supported forks.

Source code in src/evm_transition_tool/execution_specs.py 
 96
 97
 98
 99
100
101
def is_fork_supported(self, fork: Fork) -> bool:
    """
    Returns True if the fork is supported by the tool.
    Currently, ethereum-spec-evm provides no way to determine supported forks.
    """
    return fork not in UNSUPPORTED_FORKS

GethTransitionTool

Bases: TransitionTool

Go-ethereum evm Transition tool interface wrapper class.

Source code in src/evm_transition_tool/geth.py 
 16
 17
 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
class GethTransitionTool(TransitionTool):
    """
    Go-ethereum `evm` Transition tool interface wrapper class.
    """

    default_binary = Path("evm")
    detect_binary_pattern = compile(r"^evm version\b")
    t8n_subcommand: Optional[str] = "t8n"
    statetest_subcommand: Optional[str] = "statetest"
    blocktest_subcommand: Optional[str] = "blocktest"

    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), str(self.t8n_subcommand), "--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 is_fork_supported(self, fork: Fork) -> bool:
        """
        Returns True if the fork is supported by the tool.

        If the fork is a transition fork, we want to check the fork it transitions to.
        """
        return fork.fork() in self.help_string

    def verify_fixture(
        self, fixture_format: FixtureFormats, fixture_path: Path, debug_output_path: Optional[Path]
    ):
        """
        Executes `evm [state|block]test` to verify the fixture at `fixture_path`.
        """
        command: list[str] = [str(self.binary)]

        if debug_output_path:
            command += ["--debug", "--json", "--verbosity", "100"]

        if FixtureFormats.is_state_test(fixture_format):
            assert self.statetest_subcommand, "statetest subcommand not set"
            command.append(self.statetest_subcommand)
        elif FixtureFormats.is_blockchain_test(fixture_format):
            assert self.blocktest_subcommand, "blocktest subcommand not set"
            command.append(self.blocktest_subcommand)
        else:
            raise Exception(f"Invalid test fixture format: {fixture_format}")

        command.append(str(fixture_path))

        result = subprocess.run(
            command,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )

        if debug_output_path:
            debug_fixture_path = debug_output_path / fixture_path.name
            # Use the local copy of the fixture in the debug directory
            verify_fixtures_call = " ".join(command[:-1]) + f" {debug_fixture_path}"
            verify_fixtures_script = textwrap.dedent(
                f"""\
                #!/bin/bash
                {verify_fixtures_call}
                """
            )
            dump_files_to_directory(
                str(debug_output_path),
                {
                    "verify_fixtures_args.py": command,
                    "verify_fixtures_returncode.txt": result.returncode,
                    "verify_fixtures_stdout.txt": result.stdout.decode(),
                    "verify_fixtures_stderr.txt": result.stderr.decode(),
                    "verify_fixtures.sh+x": verify_fixtures_script,
                },
            )

        if result.returncode != 0:
            raise Exception(
                f"Failed to verify fixture via: '{' '.join(command)}'. "
                f"Error: '{result.stderr.decode()}'"
            )

is_fork_supported(fork)

Returns True if the fork is supported by the tool.

If the fork is a transition fork, we want to check the fork it transitions to.

Source code in src/evm_transition_tool/geth.py 
47
48
49
50
51
52
53
def is_fork_supported(self, fork: Fork) -> bool:
    """
    Returns True if the fork is supported by the tool.

    If the fork is a transition fork, we want to check the fork it transitions to.
    """
    return fork.fork() in self.help_string

verify_fixture(fixture_format, fixture_path, debug_output_path)

Executes evm [state|block]test to verify the fixture at fixture_path.

Source code in src/evm_transition_tool/geth.py 
 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
def verify_fixture(
    self, fixture_format: FixtureFormats, fixture_path: Path, debug_output_path: Optional[Path]
):
    """
    Executes `evm [state|block]test` to verify the fixture at `fixture_path`.
    """
    command: list[str] = [str(self.binary)]

    if debug_output_path:
        command += ["--debug", "--json", "--verbosity", "100"]

    if FixtureFormats.is_state_test(fixture_format):
        assert self.statetest_subcommand, "statetest subcommand not set"
        command.append(self.statetest_subcommand)
    elif FixtureFormats.is_blockchain_test(fixture_format):
        assert self.blocktest_subcommand, "blocktest subcommand not set"
        command.append(self.blocktest_subcommand)
    else:
        raise Exception(f"Invalid test fixture format: {fixture_format}")

    command.append(str(fixture_path))

    result = subprocess.run(
        command,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )

    if debug_output_path:
        debug_fixture_path = debug_output_path / fixture_path.name
        # Use the local copy of the fixture in the debug directory
        verify_fixtures_call = " ".join(command[:-1]) + f" {debug_fixture_path}"
        verify_fixtures_script = textwrap.dedent(
            f"""\
            #!/bin/bash
            {verify_fixtures_call}
            """
        )
        dump_files_to_directory(
            str(debug_output_path),
            {
                "verify_fixtures_args.py": command,
                "verify_fixtures_returncode.txt": result.returncode,
                "verify_fixtures_stdout.txt": result.stdout.decode(),
                "verify_fixtures_stderr.txt": result.stderr.decode(),
                "verify_fixtures.sh+x": verify_fixtures_script,
            },
        )

    if result.returncode != 0:
        raise Exception(
            f"Failed to verify fixture via: '{' '.join(command)}'. "
            f"Error: '{result.stderr.decode()}'"
        )

NimbusTransitionTool

Bases: TransitionTool

Nimbus evm Transition tool interface wrapper class.

Source code in src/evm_transition_tool/nimbus.py 
16
17
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
class NimbusTransitionTool(TransitionTool):
    """
    Nimbus `evm` Transition tool interface wrapper class.
    """

    default_binary = Path("t8n")
    detect_binary_pattern = compile(r"^Nimbus-t8n\b")
    version_flag: str = "--version"

    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), "--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 version(self) -> str:
        """
        Gets `evm` binary version.
        """
        if self.cached_version is None:
            self.cached_version = re.sub(r"\x1b\[0m", "", super().version()).strip()

        return self.cached_version

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

        If the fork is a transition fork, we want to check the fork it transitions to.
        """
        return fork.fork() in self.help_string

version()

Gets evm binary version.

Source code in src/evm_transition_tool/nimbus.py 
45
46
47
48
49
50
51
52
def version(self) -> str:
    """
    Gets `evm` binary version.
    """
    if self.cached_version is None:
        self.cached_version = re.sub(r"\x1b\[0m", "", super().version()).strip()

    return self.cached_version

is_fork_supported(fork)

Returns True if the fork is supported by the tool.

If the fork is a transition fork, we want to check the fork it transitions to.

Source code in src/evm_transition_tool/nimbus.py 
54
55
56
57
58
59
60
def is_fork_supported(self, fork: Fork) -> bool:
    """
    Returns True if the fork is supported by the tool.

    If the fork is a transition fork, we want to check the fork it transitions to.
    """
    return fork.fork() in self.help_string

FixtureFormats

Bases: Enum

Helper class to define fixture formats.

Source code in src/evm_transition_tool/transition_tool.py 
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
class FixtureFormats(Enum):
    """
    Helper class to define fixture formats.
    """

    STATE_TEST = "state_test"
    STATE_TEST_HIVE = "state_test_hive"
    BLOCKCHAIN_TEST = "blockchain_test"
    BLOCKCHAIN_TEST_HIVE = "blockchain_test_hive"

    @classmethod
    def is_state_test(cls, format):  # noqa: D102
        return format in (cls.STATE_TEST, cls.STATE_TEST_HIVE)

    @classmethod
    def is_blockchain_test(cls, format):  # noqa: D102
        return format in (cls.BLOCKCHAIN_TEST, cls.BLOCKCHAIN_TEST_HIVE)

    @classmethod
    def is_hive_format(cls, format):  # noqa: D102
        return format in (cls.STATE_TEST_HIVE, cls.BLOCKCHAIN_TEST_HIVE)

    @classmethod
    def is_standard_format(cls, format):  # noqa: D102
        return format in (cls.STATE_TEST, cls.BLOCKCHAIN_TEST)

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 
 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
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
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"
    t8n_subcommand: Optional[str] = None
    statetest_subcommand: Optional[str] = None
    blocktest_subcommand: Optional[str] = None
    cached_version: Optional[str] = None

    # 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:
                result = subprocess.run(
                    [binary, version_flag], stdout=subprocess.PIPE, stderr=subprocess.PIPE
                )
                if result.returncode != 0:
                    raise Exception(f"Non-zero return code: {result.returncode}")

                if result.stderr:
                    raise Exception(f"Tool wrote to stderr: {result.stderr.decode()}")

                binary_output = ""
                if result.stdout:
                    binary_output = result.stdout.decode().strip()
            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

    def version(self) -> str:
        """
        Return name and version of tool used to state transition
        """
        if self.cached_version is None:
            result = subprocess.run(
                [str(self.binary), self.version_flag],
                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

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

    def shutdown(self):
        """
        Perform any cleanup tasks related to the tested 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 collect_traces(
        self,
        receipts: List[Any],
        temp_dir: tempfile.TemporaryDirectory,
        debug_output_path: str = "",
    ) -> None:
        """
        Collect the traces from the t8n tool output and store them in the traces list.
        """
        traces: List[List[Dict]] = []
        for i, r in enumerate(receipts):
            trace_file_name = f"trace-{i}-{r['transactionHash']}.jsonl"
            if debug_output_path:
                shutil.copy(
                    os.path.join(temp_dir.name, trace_file_name),
                    os.path.join(debug_output_path, trace_file_name),
                )
            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)

    def evaluate(
        self,
        *,
        alloc: Any,
        txs: Any,
        env: Any,
        fork_name: str,
        chain_id: int = 1,
        reward: int = 0,
        eips: Optional[List[int]] = None,
        debug_output_path: str = "",
    ) -> Tuple[Dict[str, Any], Dict[str, Any]]:
        """
        Executes `evm t8n` with the specified arguments.

        If a client's `t8n` tool varies from the default behavior, this method
        should be overridden.
        """
        if eips is not None:
            fork_name = "+".join([fork_name] + [str(eip) for eip in eips])

        if int(env["currentNumber"], 0) == 0:
            reward = -1

        command: list[str] = [str(self.binary)]
        if self.t8n_subcommand:
            command.append(self.t8n_subcommand)

        args = command + [
            "--input.alloc=stdin",
            "--input.txs=stdin",
            "--input.env=stdin",
            "--output.result=stdout",
            "--output.alloc=stdout",
            "--output.body=stdout",
            f"--state.fork={fork_name}",
            f"--state.chainid={chain_id}",
            f"--state.reward={reward}",
        ]

        if self.trace:
            temp_dir = tempfile.TemporaryDirectory()
            args.append("--trace")
            args.append(f"--output.basedir={temp_dir.name}")

        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 debug_output_path:
            t8n_call = " ".join(args)
            t8n_output_base_dir = os.path.join(debug_output_path, "t8n.sh.out")
            if self.trace:
                t8n_call = t8n_call.replace(temp_dir.name, t8n_output_base_dir)
            t8n_script = textwrap.dedent(
                f"""\
                #!/bin/bash
                rm -rf {debug_output_path}/t8n.sh.out  # hard-coded to avoid surprises
                mkdir {debug_output_path}/t8n.sh.out  # unused if tracing is not enabled
                {t8n_call} < {debug_output_path}/stdin.txt
                """
            )
            dump_files_to_directory(
                debug_output_path,
                {
                    "args.py": args,
                    "input/alloc.json": stdin["alloc"],
                    "input/env.json": stdin["env"],
                    "input/txs.json": stdin["txs"],
                    "returncode.txt": result.returncode,
                    "stdin.txt": stdin,
                    "stdout.txt": result.stdout.decode(),
                    "stderr.txt": result.stderr.decode(),
                    "t8n.sh+x": t8n_script,
                },
            )

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

        output = json.loads(result.stdout)

        if not all([x in output for x in ["alloc", "result", "body"]]):
            raise Exception("Malformed t8n output: missing 'alloc', 'result' or 'body'.")

        if debug_output_path:
            dump_files_to_directory(
                debug_output_path,
                {
                    "output/alloc.json": output["alloc"],
                    "output/result.json": output["result"],
                    "output/txs.rlp": output["body"],
                },
            )

        if self.trace:
            self.collect_traces(output["result"]["receipts"], temp_dir, debug_output_path)
            temp_dir.cleanup()

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

    def calc_state_root(
        self, *, alloc: Any, fork: Fork, debug_output_path: str = ""
    ) -> Tuple[Dict, 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"] = []

        if fork.header_excess_blob_gas_required(0, 0):
            env["currentExcessBlobGas"] = "0"

        if fork.header_beacon_root_required(0, 0):
            env[
                "parentBeaconBlockRoot"
            ] = "0x0000000000000000000000000000000000000000000000000000000000000000"

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

    def verify_fixture(
        self, fixture_format: FixtureFormats, fixture_path: Path, debug_output_path: Optional[Path]
    ):
        """
        Executes `evm [state|block]test` to verify the fixture at `fixture_path`.

        Currently only implemented by geth's evm.
        """
        raise Exception(
            "The `verify_fixture()` function is not supported by this tool. Use geth's evm tool."
        )

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

Abstract initialization method that all subclasses must implement.

Source code in src/evm_transition_tool/transition_tool.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
@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 
135
136
137
138
139
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 
141
142
143
144
145
146
@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 
148
149
150
151
152
153
@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 
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
@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:
            result = subprocess.run(
                [binary, version_flag], stdout=subprocess.PIPE, stderr=subprocess.PIPE
            )
            if result.returncode != 0:
                raise Exception(f"Non-zero return code: {result.returncode}")

            if result.stderr:
                raise Exception(f"Tool wrote to stderr: {result.stderr.decode()}")

            binary_output = ""
            if result.stdout:
                binary_output = result.stdout.decode().strip()
        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 
204
205
206
207
208
209
210
211
@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

version()

Return name and version of tool used to state transition

Source code in src/evm_transition_tool/transition_tool.py 
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
def version(self) -> str:
    """
    Return name and version of tool used to state transition
    """
    if self.cached_version is None:
        result = subprocess.run(
            [str(self.binary), self.version_flag],
            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) abstractmethod

Returns True if the fork is supported by the tool

Source code in src/evm_transition_tool/transition_tool.py 
230
231
232
233
234
235
@abstractmethod
def is_fork_supported(self, fork: Fork) -> bool:
    """
    Returns True if the fork is supported by the tool
    """
    pass

shutdown()

Perform any cleanup tasks related to the tested tool.

Source code in src/evm_transition_tool/transition_tool.py 
237
238
239
240
241
def shutdown(self):
    """
    Perform any cleanup tasks related to the tested 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 
243
244
245
246
247
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 
249
250
251
252
253
254
255
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 
257
258
259
260
261
def get_traces(self) -> List[List[List[Dict]]] | None:
    """
    Returns the accumulated traces
    """
    return self.traces

collect_traces(receipts, temp_dir, debug_output_path='')

Collect the traces from the t8n tool output and store them in the traces list.

Source code in src/evm_transition_tool/transition_tool.py 
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
def collect_traces(
    self,
    receipts: List[Any],
    temp_dir: tempfile.TemporaryDirectory,
    debug_output_path: str = "",
) -> None:
    """
    Collect the traces from the t8n tool output and store them in the traces list.
    """
    traces: List[List[Dict]] = []
    for i, r in enumerate(receipts):
        trace_file_name = f"trace-{i}-{r['transactionHash']}.jsonl"
        if debug_output_path:
            shutil.copy(
                os.path.join(temp_dir.name, trace_file_name),
                os.path.join(debug_output_path, trace_file_name),
            )
        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)

evaluate(*, alloc, txs, env, fork_name, chain_id=1, reward=0, eips=None, debug_output_path='')

Executes evm t8n with the specified arguments.

If a client's t8n tool varies from the default behavior, this method should be overridden.

Source code in src/evm_transition_tool/transition_tool.py 
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
def evaluate(
    self,
    *,
    alloc: Any,
    txs: Any,
    env: Any,
    fork_name: str,
    chain_id: int = 1,
    reward: int = 0,
    eips: Optional[List[int]] = None,
    debug_output_path: str = "",
) -> Tuple[Dict[str, Any], Dict[str, Any]]:
    """
    Executes `evm t8n` with the specified arguments.

    If a client's `t8n` tool varies from the default behavior, this method
    should be overridden.
    """
    if eips is not None:
        fork_name = "+".join([fork_name] + [str(eip) for eip in eips])

    if int(env["currentNumber"], 0) == 0:
        reward = -1

    command: list[str] = [str(self.binary)]
    if self.t8n_subcommand:
        command.append(self.t8n_subcommand)

    args = command + [
        "--input.alloc=stdin",
        "--input.txs=stdin",
        "--input.env=stdin",
        "--output.result=stdout",
        "--output.alloc=stdout",
        "--output.body=stdout",
        f"--state.fork={fork_name}",
        f"--state.chainid={chain_id}",
        f"--state.reward={reward}",
    ]

    if self.trace:
        temp_dir = tempfile.TemporaryDirectory()
        args.append("--trace")
        args.append(f"--output.basedir={temp_dir.name}")

    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 debug_output_path:
        t8n_call = " ".join(args)
        t8n_output_base_dir = os.path.join(debug_output_path, "t8n.sh.out")
        if self.trace:
            t8n_call = t8n_call.replace(temp_dir.name, t8n_output_base_dir)
        t8n_script = textwrap.dedent(
            f"""\
            #!/bin/bash
            rm -rf {debug_output_path}/t8n.sh.out  # hard-coded to avoid surprises
            mkdir {debug_output_path}/t8n.sh.out  # unused if tracing is not enabled
            {t8n_call} < {debug_output_path}/stdin.txt
            """
        )
        dump_files_to_directory(
            debug_output_path,
            {
                "args.py": args,
                "input/alloc.json": stdin["alloc"],
                "input/env.json": stdin["env"],
                "input/txs.json": stdin["txs"],
                "returncode.txt": result.returncode,
                "stdin.txt": stdin,
                "stdout.txt": result.stdout.decode(),
                "stderr.txt": result.stderr.decode(),
                "t8n.sh+x": t8n_script,
            },
        )

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

    output = json.loads(result.stdout)

    if not all([x in output for x in ["alloc", "result", "body"]]):
        raise Exception("Malformed t8n output: missing 'alloc', 'result' or 'body'.")

    if debug_output_path:
        dump_files_to_directory(
            debug_output_path,
            {
                "output/alloc.json": output["alloc"],
                "output/result.json": output["result"],
                "output/txs.rlp": output["body"],
            },
        )

    if self.trace:
        self.collect_traces(output["result"]["receipts"], temp_dir, debug_output_path)
        temp_dir.cleanup()

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

calc_state_root(*, alloc, fork, debug_output_path='')

Calculate the state root for the given alloc.

Source code in src/evm_transition_tool/transition_tool.py 
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
def calc_state_root(
    self, *, alloc: Any, fork: Fork, debug_output_path: str = ""
) -> Tuple[Dict, 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"] = []

    if fork.header_excess_blob_gas_required(0, 0):
        env["currentExcessBlobGas"] = "0"

    if fork.header_beacon_root_required(0, 0):
        env[
            "parentBeaconBlockRoot"
        ] = "0x0000000000000000000000000000000000000000000000000000000000000000"

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

verify_fixture(fixture_format, fixture_path, debug_output_path)

Executes evm [state|block]test to verify the fixture at fixture_path.

Currently only implemented by geth's evm.

Source code in src/evm_transition_tool/transition_tool.py 
441
442
443
444
445
446
447
448
449
450
451
def verify_fixture(
    self, fixture_format: FixtureFormats, fixture_path: Path, debug_output_path: Optional[Path]
):
    """
    Executes `evm [state|block]test` to verify the fixture at `fixture_path`.

    Currently only implemented by geth's evm.
    """
    raise Exception(
        "The `verify_fixture()` function is not supported by this tool. Use geth's evm tool."
    )

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 
29
30
31
32
33
34
35
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)

UnknownTransitionTool

Bases: Exception

Exception raised if an unknown t8n is encountered

Source code in src/evm_transition_tool/transition_tool.py 
23
24
25
26
class UnknownTransitionTool(Exception):
    """Exception raised if an unknown t8n is encountered"""

    pass