Skip to content

Ethereum CLIs 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/ethereum_clis/clis/besu.py 
 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
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
class BesuTransitionTool(TransitionTool):
    """Besu EvmTool Transition tool frontend wrapper class."""

    default_binary = Path("evm")
    detect_binary_pattern = re.compile(r"^Hyperledger Besu evm .*$")
    binary: Path
    cached_version: Optional[str] = None
    trace: bool
    process: Optional[subprocess.Popen] = None
    server_url: str
    besu_trace_dir: Optional[tempfile.TemporaryDirectory]

    def __init__(
        self,
        *,
        binary: Optional[Path] = None,
        trace: bool = False,
    ):
        """Initialize the BesuTransitionTool class."""
        super().__init__(exception_mapper=BesuExceptionMapper(), 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(
                f"evm process unexpectedly returned a non-zero status code: {e}."
            ) from e
        except Exception as e:
            raise Exception(f"Unexpected exception calling evm tool: {e}.") from e
        self.help_string = result.stdout
        self.besu_trace_dir = tempfile.TemporaryDirectory() if self.trace else None

    def start_server(self):
        """
        Start the t8n-server process, extract the port, and leave it running
        for future re-use.
        """
        args = [
            str(self.binary),
            "t8n-server",
            "--port=0",  # OS assigned server port
        ]

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

        self.process = subprocess.Popen(
            args=args,
            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 (\\d+)", line).group(1)
                self.server_url = f"http://localhost:{port}/"
                break

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

    def evaluate(
        self,
        *,
        alloc: Alloc,
        txs: List[Transaction],
        env: Environment,
        fork: Fork,
        chain_id: int,
        reward: int,
        blob_schedule: BlobSchedule | None = None,
        eips: Optional[List[int]] = None,
        debug_output_path: str = "",
        state_test: bool = False,
        slow_request: bool = False,
    ) -> TransitionToolOutput:
        """Execute `evm t8n` with the specified arguments."""
        if not self.process:
            self.start_server()

        fork_name = fork.transition_tool_name(
            block_number=env.number,
            timestamp=env.timestamp,
        )
        if eips is not None:
            fork_name = "+".join([fork_name] + [str(eip) for eip in eips])

        input_json = TransitionToolInput(
            alloc=alloc,
            txs=txs,
            env=env,
        ).model_dump(mode="json", **model_dump_config)

        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}'
                """
            )
            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: TransitionToolOutput = TransitionToolOutput.model_validate(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 debug_output_path:
            dump_files_to_directory(
                debug_output_path,
                {
                    "output/alloc.json": output.alloc.model_dump(mode="json", **model_dump_config),
                    "output/result.json": output.result.model_dump(
                        mode="json", **model_dump_config
                    ),
                    "output/txs.rlp": str(output.body),
                },
            )

        if self.trace and self.besu_trace_dir:
            self.collect_traces(output.result.receipts, self.besu_trace_dir, debug_output_path)
            for i, r in enumerate(output.result.receipts):
                trace_file_name = f"trace-{i}-{r.transaction_hash}.jsonl"
                os.remove(os.path.join(self.besu_trace_dir.name, trace_file_name))

        return output

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

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

Initialize the BesuTransitionTool class.

Source code in src/ethereum_clis/clis/besu.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def __init__(
    self,
    *,
    binary: Optional[Path] = None,
    trace: bool = False,
):
    """Initialize the BesuTransitionTool class."""
    super().__init__(exception_mapper=BesuExceptionMapper(), 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(
            f"evm process unexpectedly returned a non-zero status code: {e}."
        ) from e
    except Exception as e:
        raise Exception(f"Unexpected exception calling evm tool: {e}.") from e
    self.help_string = result.stdout
    self.besu_trace_dir = tempfile.TemporaryDirectory() if self.trace else None

start_server()

Start the t8n-server process, extract the port, and leave it running for future re-use.

Source code in src/ethereum_clis/clis/besu.py 
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
def start_server(self):
    """
    Start the t8n-server process, extract the port, and leave it running
    for future re-use.
    """
    args = [
        str(self.binary),
        "t8n-server",
        "--port=0",  # OS assigned server port
    ]

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

    self.process = subprocess.Popen(
        args=args,
        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 (\\d+)", line).group(1)
            self.server_url = f"http://localhost:{port}/"
            break

shutdown()

Stop the t8n-server process if it was started.

Source code in src/ethereum_clis/clis/besu.py 
91
92
93
94
95
96
def shutdown(self):
    """Stop the t8n-server process if it was started."""
    if self.process:
        self.process.kill()
    if self.besu_trace_dir:
        self.besu_trace_dir.cleanup()

evaluate(*, alloc, txs, env, fork, chain_id, reward, blob_schedule=None, eips=None, debug_output_path='', state_test=False, slow_request=False)

Execute evm t8n with the specified arguments.

Source code in src/ethereum_clis/clis/besu.py 
 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
def evaluate(
    self,
    *,
    alloc: Alloc,
    txs: List[Transaction],
    env: Environment,
    fork: Fork,
    chain_id: int,
    reward: int,
    blob_schedule: BlobSchedule | None = None,
    eips: Optional[List[int]] = None,
    debug_output_path: str = "",
    state_test: bool = False,
    slow_request: bool = False,
) -> TransitionToolOutput:
    """Execute `evm t8n` with the specified arguments."""
    if not self.process:
        self.start_server()

    fork_name = fork.transition_tool_name(
        block_number=env.number,
        timestamp=env.timestamp,
    )
    if eips is not None:
        fork_name = "+".join([fork_name] + [str(eip) for eip in eips])

    input_json = TransitionToolInput(
        alloc=alloc,
        txs=txs,
        env=env,
    ).model_dump(mode="json", **model_dump_config)

    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}'
            """
        )
        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: TransitionToolOutput = TransitionToolOutput.model_validate(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 debug_output_path:
        dump_files_to_directory(
            debug_output_path,
            {
                "output/alloc.json": output.alloc.model_dump(mode="json", **model_dump_config),
                "output/result.json": output.result.model_dump(
                    mode="json", **model_dump_config
                ),
                "output/txs.rlp": str(output.body),
            },
        )

    if self.trace and self.besu_trace_dir:
        self.collect_traces(output.result.receipts, self.besu_trace_dir, debug_output_path)
        for i, r in enumerate(output.result.receipts):
            trace_file_name = f"trace-{i}-{r.transaction_hash}.jsonl"
            os.remove(os.path.join(self.besu_trace_dir.name, trace_file_name))

    return output

is_fork_supported(fork)

Return True if the fork is supported by the tool.

Source code in src/ethereum_clis/clis/besu.py 
204
205
206
def is_fork_supported(self, fork: Fork) -> bool:
    """Return True if the fork is supported by the tool."""
    return fork.transition_tool_name() in self.help_string

EthereumJSTransitionTool

Bases: TransitionTool

EthereumJS Transition tool interface wrapper class.

Source code in src/ethereum_clis/clis/ethereumjs.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
class EthereumJSTransitionTool(TransitionTool):
    """EthereumJS Transition tool interface wrapper class."""

    default_binary = Path("ethereumjs-t8ntool.sh")
    detect_binary_pattern = re.compile(r"^ethereumjs t8n\b")
    version_flag: str = "--version"
    t8n_use_stream = False

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

    def __init__(
        self,
        *,
        binary: Optional[Path] = None,
        trace: bool = False,
    ):
        """Initialize the EthereumJS Transition tool interface."""
        super().__init__(exception_mapper=EthereumJSExceptionMapper(), binary=binary, trace=trace)

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

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

Initialize the EthereumJS Transition tool interface.

Source code in src/ethereum_clis/clis/ethereumjs.py 
30
31
32
33
34
35
36
37
def __init__(
    self,
    *,
    binary: Optional[Path] = None,
    trace: bool = False,
):
    """Initialize the EthereumJS Transition tool interface."""
    super().__init__(exception_mapper=EthereumJSExceptionMapper(), binary=binary, trace=trace)

is_fork_supported(fork)

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

Source code in src/ethereum_clis/clis/ethereumjs.py 
39
40
41
42
43
44
def is_fork_supported(self, fork: Fork) -> bool:
    """
    Return True if the fork is supported by the tool.
    Currently, EthereumJS-t8n provides no way to determine supported forks.
    """
    return True

EvmoneExceptionMapper

Bases: ExceptionMapper

Translate between EEST exceptions and error strings returned by Evmone.

Source code in src/ethereum_clis/clis/evmone.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
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
class EvmoneExceptionMapper(ExceptionMapper):
    """Translate between EEST exceptions and error strings returned by Evmone."""

    @property
    def _mapping_data(self):
        return [
            ExceptionMessage(
                TransactionException.TYPE_4_TX_CONTRACT_CREATION,
                "set code transaction must ",
            ),
            ExceptionMessage(
                TransactionException.TYPE_4_INVALID_AUTHORITY_SIGNATURE,
                "invalid authorization signature",
            ),
            ExceptionMessage(
                TransactionException.TYPE_4_INVALID_AUTHORITY_SIGNATURE_S_TOO_HIGH,
                "authorization signature s value too high",
            ),
            ExceptionMessage(
                TransactionException.TYPE_4_EMPTY_AUTHORIZATION_LIST,
                "empty authorization list",
            ),
            ExceptionMessage(
                TransactionException.INTRINSIC_GAS_TOO_LOW,
                "intrinsic gas too low",
            ),
            ExceptionMessage(
                TransactionException.TYPE_3_TX_MAX_BLOB_GAS_ALLOWANCE_EXCEEDED,
                "lob gas limit exceeded",
            ),
            ExceptionMessage(
                TransactionException.INITCODE_SIZE_EXCEEDED,
                "max initcode size exceeded",
            ),
            ExceptionMessage(
                TransactionException.INSUFFICIENT_ACCOUNT_FUNDS,
                "insufficient funds for gas * price + value",
            ),
            ExceptionMessage(
                TransactionException.INSUFFICIENT_MAX_FEE_PER_GAS,
                "max fee per gas less than block base fee",
            ),
            ExceptionMessage(
                TransactionException.INSUFFICIENT_MAX_FEE_PER_BLOB_GAS,
                "fee per gas less than block base fee",
            ),
            ExceptionMessage(
                TransactionException.TYPE_3_TX_PRE_FORK,
                "transaction type not supported",
            ),
            ExceptionMessage(
                TransactionException.TYPE_3_TX_INVALID_BLOB_VERSIONED_HASH,
                "invalid blob hash version",
            ),
            ExceptionMessage(
                TransactionException.TYPE_3_TX_BLOB_COUNT_EXCEEDED,
                "blob gas limit exceeded",
            ),
            ExceptionMessage(
                TransactionException.TYPE_3_TX_ZERO_BLOBS,
                "empty blob hashes list",
            ),
            ExceptionMessage(
                TransactionException.NONCE_MISMATCH_TOO_LOW,
                "nonce too low",
            ),
            ExceptionMessage(
                TransactionException.NONCE_MISMATCH_TOO_HIGH,
                "nonce too high",
            ),
            # TODO EVMONE needs to differentiate when the section is missing in the header or body
            ExceptionMessage(EOFException.MISSING_STOP_OPCODE, "err: no_terminating_instruction"),
            ExceptionMessage(EOFException.MISSING_CODE_HEADER, "err: code_section_missing"),
            ExceptionMessage(EOFException.MISSING_TYPE_HEADER, "err: type_section_missing"),
            # TODO EVMONE these exceptions are too similar, this leeds to ambiguity
            ExceptionMessage(EOFException.MISSING_TERMINATOR, "err: header_terminator_missing"),
            ExceptionMessage(
                EOFException.MISSING_HEADERS_TERMINATOR, "err: section_headers_not_terminated"
            ),
            ExceptionMessage(EOFException.INVALID_VERSION, "err: eof_version_unknown"),
            ExceptionMessage(
                EOFException.INVALID_NON_RETURNING_FLAG, "err: invalid_non_returning_flag"
            ),
            ExceptionMessage(EOFException.INVALID_MAGIC, "err: invalid_prefix"),
            ExceptionMessage(
                EOFException.INVALID_FIRST_SECTION_TYPE, "err: invalid_first_section_type"
            ),
            ExceptionMessage(
                EOFException.INVALID_SECTION_BODIES_SIZE, "err: invalid_section_bodies_size"
            ),
            ExceptionMessage(
                EOFException.INVALID_TYPE_SECTION_SIZE, "err: invalid_type_section_size"
            ),
            ExceptionMessage(EOFException.INCOMPLETE_SECTION_SIZE, "err: incomplete_section_size"),
            ExceptionMessage(
                EOFException.INCOMPLETE_SECTION_NUMBER, "err: incomplete_section_number"
            ),
            ExceptionMessage(EOFException.TOO_MANY_CODE_SECTIONS, "err: too_many_code_sections"),
            ExceptionMessage(EOFException.ZERO_SECTION_SIZE, "err: zero_section_size"),
            ExceptionMessage(EOFException.MISSING_DATA_SECTION, "err: data_section_missing"),
            ExceptionMessage(EOFException.UNDEFINED_INSTRUCTION, "err: undefined_instruction"),
            ExceptionMessage(
                EOFException.INPUTS_OUTPUTS_NUM_ABOVE_LIMIT, "err: inputs_outputs_num_above_limit"
            ),
            ExceptionMessage(
                EOFException.UNREACHABLE_INSTRUCTIONS, "err: unreachable_instructions"
            ),
            ExceptionMessage(
                EOFException.INVALID_RJUMP_DESTINATION, "err: invalid_rjump_destination"
            ),
            ExceptionMessage(
                EOFException.UNREACHABLE_CODE_SECTIONS, "err: unreachable_code_sections"
            ),
            ExceptionMessage(EOFException.STACK_UNDERFLOW, "err: stack_underflow"),
            ExceptionMessage(EOFException.STACK_OVERFLOW, "err: stack_overflow"),
            ExceptionMessage(
                EOFException.MAX_STACK_HEIGHT_ABOVE_LIMIT, "err: max_stack_height_above_limit"
            ),
            ExceptionMessage(
                EOFException.STACK_HIGHER_THAN_OUTPUTS, "err: stack_higher_than_outputs_required"
            ),
            ExceptionMessage(
                EOFException.JUMPF_DESTINATION_INCOMPATIBLE_OUTPUTS,
                "err: jumpf_destination_incompatible_outputs",
            ),
            ExceptionMessage(
                EOFException.INVALID_MAX_STACK_HEIGHT, "err: invalid_max_stack_height"
            ),
            ExceptionMessage(EOFException.INVALID_DATALOADN_INDEX, "err: invalid_dataloadn_index"),
            ExceptionMessage(EOFException.TRUNCATED_INSTRUCTION, "err: truncated_instruction"),
            ExceptionMessage(
                EOFException.TOPLEVEL_CONTAINER_TRUNCATED, "err: toplevel_container_truncated"
            ),
            ExceptionMessage(EOFException.ORPHAN_SUBCONTAINER, "err: unreferenced_subcontainer"),
            ExceptionMessage(
                EOFException.CONTAINER_SIZE_ABOVE_LIMIT, "err: container_size_above_limit"
            ),
            ExceptionMessage(
                EOFException.INVALID_CONTAINER_SECTION_INDEX,
                "err: invalid_container_section_index",
            ),
            ExceptionMessage(
                EOFException.INCOMPATIBLE_CONTAINER_KIND, "err: incompatible_container_kind"
            ),
            ExceptionMessage(EOFException.STACK_HEIGHT_MISMATCH, "err: stack_height_mismatch"),
            ExceptionMessage(EOFException.TOO_MANY_CONTAINERS, "err: too_many_container_sections"),
            ExceptionMessage(
                EOFException.INVALID_CODE_SECTION_INDEX, "err: invalid_code_section_index"
            ),
            ExceptionMessage(
                EOFException.CALLF_TO_NON_RETURNING, "err: callf_to_non_returning_function"
            ),
            ExceptionMessage(
                EOFException.EOFCREATE_WITH_TRUNCATED_CONTAINER,
                "err: eofcreate_with_truncated_container",
            ),
        ]

EvmOneTransitionTool

Bases: TransitionTool

Evmone evmone-t8n Transition tool interface wrapper class.

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

    default_binary = Path("evmone-t8n")
    detect_binary_pattern = re.compile(r"^evmone-t8n\b")
    t8n_use_stream = False

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

    def __init__(
        self,
        *,
        binary: Optional[Path] = None,
        trace: bool = False,
    ):
        """Initialize the Evmone Transition tool interface."""
        super().__init__(exception_mapper=EvmoneExceptionMapper(), binary=binary, trace=trace)

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

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

Initialize the Evmone Transition tool interface.

Source code in src/ethereum_clis/clis/evmone.py 
29
30
31
32
33
34
35
36
def __init__(
    self,
    *,
    binary: Optional[Path] = None,
    trace: bool = False,
):
    """Initialize the Evmone Transition tool interface."""
    super().__init__(exception_mapper=EvmoneExceptionMapper(), binary=binary, trace=trace)

is_fork_supported(fork)

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

Source code in src/ethereum_clis/clis/evmone.py 
38
39
40
41
42
43
def is_fork_supported(self, fork: Fork) -> bool:
    """
    Return True if the fork is supported by the tool.
    Currently, evmone-t8n provides no way to determine supported forks.
    """
    return True

ExecutionSpecsTransitionTool

Bases: TransitionTool

Ethereum Specs EVM Resolver ethereum-spec-evm-resolver Transition Tool wrapper class.

ethereum-spec-evm-resolver is installed by default for execution-spec-tests: 

uv run fill --evm-bin=ethereum-spec-evm-resolver

To use a specific version of the ethereum-spec-evm-resolver tool, update it to the desired version in pyproject.toml.

The ethereum-spec-evm-resolver tool essentially wraps around the EELS evm daemon. It can handle requests for different EVM forks, even when those forks are implemented by different versions of EELS hosted in different places.

Source code in src/ethereum_clis/clis/execution_specs.py 
 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
class ExecutionSpecsTransitionTool(TransitionTool):
    """
    Ethereum Specs EVM Resolver `ethereum-spec-evm-resolver` Transition Tool wrapper class.

    `ethereum-spec-evm-resolver` is installed by default for `execution-spec-tests`:
    ```console
    uv run fill --evm-bin=ethereum-spec-evm-resolver
    ```

    To use a specific version of the `ethereum-spec-evm-resolver` tool, update it to the
    desired version in `pyproject.toml`.

    The `ethereum-spec-evm-resolver` tool essentially wraps around the EELS evm daemon. It can
    handle requests for different EVM forks, even when those forks are implemented by different
    versions of EELS hosted in different places.
    """

    default_binary = Path("ethereum-spec-evm-resolver")
    detect_binary_pattern = re.compile(r"^ethereum-spec-evm-resolver\b")
    t8n_use_server: bool = True
    server_dir: Optional[TemporaryDirectory] = None

    def __init__(
        self,
        *,
        binary: Optional[Path] = None,
        trace: bool = False,
    ):
        """Initialize the Ethereum Specs EVM Resolver Transition Tool interface."""
        os.environ.setdefault("NO_PROXY", "*")  # Disable proxy for local connections
        super().__init__(
            exception_mapper=ExecutionSpecsExceptionMapper(), 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(
                "ethereum-spec-evm-resolver process unexpectedly returned a non-zero status code: "
                f"{e}."
            ) from e
        except Exception as e:
            raise Exception(
                f"Unexpected exception calling ethereum-spec-evm-resolver: {e}."
            ) from e
        self.help_string = result.stdout

    def start_server(self):
        """
        Start the t8n-server process, extract the port, and leave it running
        for future re-use.
        """
        self.server_dir = TemporaryDirectory()
        self.server_file_path = Path(self.server_dir.name) / "t8n.sock"
        replaced_str = str(self.server_file_path).replace("/", "%2F")
        self.server_url = f"http+unix://{replaced_str}/"
        self.process = subprocess.Popen(
            args=[
                str(self.binary),
                "daemon",
                "--uds",
                self.server_file_path,
            ],
        )
        start = time.time()
        while True:
            if self.server_file_path.exists():
                break
            if time.time() - start > DAEMON_STARTUP_TIMEOUT_SECONDS:
                raise Exception("Failed starting ethereum-spec-evm subprocess")
            time.sleep(0)  # yield to other processes

    def shutdown(self):
        """Stop the t8n-server process if it was started."""
        if self.process:
            self.process.terminate()
        if self.server_dir:
            self.server_dir.cleanup()
            self.server_dir = None

    def is_fork_supported(self, fork: Fork) -> bool:
        """
        Return 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.

        `ethereum-spec-evm` appends newlines to forks in the help string.
        """
        return (fork.transition_tool_name() + "\n") in self.help_string

    def _generate_post_args(
        self, t8n_data: TransitionTool.TransitionToolData
    ) -> Dict[str, List[str] | str]:
        """
        Generate the arguments for the POST request to the t8n-server.

        EELS T8N expects `--state-test` when running a state test.
        """
        return {"arg": "--state-test"} if t8n_data.state_test else {}

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

Initialize the Ethereum Specs EVM Resolver Transition Tool interface.

Source code in src/ethereum_clis/clis/execution_specs.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def __init__(
    self,
    *,
    binary: Optional[Path] = None,
    trace: bool = False,
):
    """Initialize the Ethereum Specs EVM Resolver Transition Tool interface."""
    os.environ.setdefault("NO_PROXY", "*")  # Disable proxy for local connections
    super().__init__(
        exception_mapper=ExecutionSpecsExceptionMapper(), 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(
            "ethereum-spec-evm-resolver process unexpectedly returned a non-zero status code: "
            f"{e}."
        ) from e
    except Exception as e:
        raise Exception(
            f"Unexpected exception calling ethereum-spec-evm-resolver: {e}."
        ) from e
    self.help_string = result.stdout

start_server()

Start the t8n-server process, extract the port, and leave it running for future re-use.

Source code in src/ethereum_clis/clis/execution_specs.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def start_server(self):
    """
    Start the t8n-server process, extract the port, and leave it running
    for future re-use.
    """
    self.server_dir = TemporaryDirectory()
    self.server_file_path = Path(self.server_dir.name) / "t8n.sock"
    replaced_str = str(self.server_file_path).replace("/", "%2F")
    self.server_url = f"http+unix://{replaced_str}/"
    self.process = subprocess.Popen(
        args=[
            str(self.binary),
            "daemon",
            "--uds",
            self.server_file_path,
        ],
    )
    start = time.time()
    while True:
        if self.server_file_path.exists():
            break
        if time.time() - start > DAEMON_STARTUP_TIMEOUT_SECONDS:
            raise Exception("Failed starting ethereum-spec-evm subprocess")
        time.sleep(0)  # yield to other processes

shutdown()

Stop the t8n-server process if it was started.

Source code in src/ethereum_clis/clis/execution_specs.py 
100
101
102
103
104
105
106
def shutdown(self):
    """Stop the t8n-server process if it was started."""
    if self.process:
        self.process.terminate()
    if self.server_dir:
        self.server_dir.cleanup()
        self.server_dir = None

is_fork_supported(fork)

Return 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.

ethereum-spec-evm appends newlines to forks in the help string.

Source code in src/ethereum_clis/clis/execution_specs.py 
108
109
110
111
112
113
114
115
116
def is_fork_supported(self, fork: Fork) -> bool:
    """
    Return 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.

    `ethereum-spec-evm` appends newlines to forks in the help string.
    """
    return (fork.transition_tool_name() + "\n") in self.help_string

GethFixtureConsumer

Bases: GethEvm, FixtureConsumerTool

Geth's implementation of the fixture consumer.

Source code in src/ethereum_clis/clis/geth.py 
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
class GethFixtureConsumer(
    GethEvm,
    FixtureConsumerTool,
    fixture_formats=[StateFixture, BlockchainFixture],
):
    """Geth's implementation of the fixture consumer."""

    def consume_blockchain_test(
        self,
        fixture_path: Path,
        fixture_name: Optional[str] = None,
        debug_output_path: Optional[Path] = None,
    ):
        """
        Consume a single blockchain test.

        The `evm blocktest` command takes the `--run` argument which can be used to select a
        specific fixture from the fixture file when executing.
        """
        subcommand = "blocktest"
        global_options = []
        subcommand_options = []
        if debug_output_path:
            global_options += ["--verbosity", "100"]
            subcommand_options += ["--trace"]

        if fixture_name:
            subcommand_options += ["--run", re.escape(fixture_name)]

        command = (
            [str(self.binary)]
            + global_options
            + [subcommand]
            + subcommand_options
            + [str(fixture_path)]
        )

        result = self._run_command(command)

        if debug_output_path:
            self._consume_debug_dump(command, result, fixture_path, debug_output_path)

        if result.returncode != 0:
            raise Exception(
                f"Unexpected exit code:\n{' '.join(command)}\n\n Error:\n{result.stderr}"
            )

    @cache  # noqa
    def consume_state_test_file(
        self,
        fixture_path: Path,
        debug_output_path: Optional[Path] = None,
    ) -> List[Dict[str, Any]]:
        """
        Consume an entire state test file.

        The `evm statetest` will always execute all the tests contained in a file without the
        possibility of selecting a single test, so this function is cached in order to only call
        the command once and `consume_state_test` can simply select the result that
        was requested.
        """
        subcommand = "statetest"
        global_options: List[str] = []
        subcommand_options: List[str] = []
        if debug_output_path:
            global_options += ["--verbosity", "100"]
            subcommand_options += ["--trace"]

        command = (
            [str(self.binary)]
            + global_options
            + [subcommand]
            + subcommand_options
            + [str(fixture_path)]
        )
        result = self._run_command(command)

        if debug_output_path:
            self._consume_debug_dump(command, result, fixture_path, debug_output_path)

        if result.returncode != 0:
            raise Exception(
                f"Unexpected exit code:\n{' '.join(command)}\n\n Error:\n{result.stderr}"
            )

        result_json = json.loads(result.stdout)
        if not isinstance(result_json, list):
            raise Exception(f"Unexpected result from evm statetest: {result_json}")
        return result_json

    def consume_state_test(
        self,
        fixture_path: Path,
        fixture_name: Optional[str] = None,
        debug_output_path: Optional[Path] = None,
    ):
        """
        Consume a single state test.

        Uses the cached result from `consume_state_test_file` in order to not call the command
        every time an select a single result from there.
        """
        file_results = self.consume_state_test_file(
            fixture_path=fixture_path,
            debug_output_path=debug_output_path,
        )
        if fixture_name:
            test_result = [
                test_result for test_result in file_results if test_result["name"] == fixture_name
            ]
            assert len(test_result) < 2, f"Multiple test results for {fixture_name}"
            assert len(test_result) == 1, f"Test result for {fixture_name} missing"
            assert test_result[0]["pass"], f"State test failed: {test_result[0]['error']}"
        else:
            if any(not test_result["pass"] for test_result in file_results):
                exception_text = "State test failed: \n" + "\n".join(
                    f"{test_result['name']}: " + test_result["error"]
                    for test_result in file_results
                    if not test_result["pass"]
                )
                raise Exception(exception_text)

    def consume_fixture(
        self,
        fixture_format: FixtureFormat,
        fixture_path: Path,
        fixture_name: Optional[str] = None,
        debug_output_path: Optional[Path] = None,
    ):
        """Execute the appropriate geth fixture consumer for the fixture at `fixture_path`."""
        if fixture_format == BlockchainFixture:
            self.consume_blockchain_test(
                fixture_path=fixture_path,
                fixture_name=fixture_name,
                debug_output_path=debug_output_path,
            )
        elif fixture_format == StateFixture:
            self.consume_state_test(
                fixture_path=fixture_path,
                fixture_name=fixture_name,
                debug_output_path=debug_output_path,
            )
        else:
            raise Exception(
                f"Fixture format {fixture_format.format_name} not supported by {self.binary}"
            )

consume_blockchain_test(fixture_path, fixture_name=None, debug_output_path=None)

Consume a single blockchain test.

The evm blocktest command takes the --run argument which can be used to select a specific fixture from the fixture file when executing.

Source code in src/ethereum_clis/clis/geth.py 
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
def consume_blockchain_test(
    self,
    fixture_path: Path,
    fixture_name: Optional[str] = None,
    debug_output_path: Optional[Path] = None,
):
    """
    Consume a single blockchain test.

    The `evm blocktest` command takes the `--run` argument which can be used to select a
    specific fixture from the fixture file when executing.
    """
    subcommand = "blocktest"
    global_options = []
    subcommand_options = []
    if debug_output_path:
        global_options += ["--verbosity", "100"]
        subcommand_options += ["--trace"]

    if fixture_name:
        subcommand_options += ["--run", re.escape(fixture_name)]

    command = (
        [str(self.binary)]
        + global_options
        + [subcommand]
        + subcommand_options
        + [str(fixture_path)]
    )

    result = self._run_command(command)

    if debug_output_path:
        self._consume_debug_dump(command, result, fixture_path, debug_output_path)

    if result.returncode != 0:
        raise Exception(
            f"Unexpected exit code:\n{' '.join(command)}\n\n Error:\n{result.stderr}"
        )

consume_state_test_file(fixture_path, debug_output_path=None) cached

Consume an entire state test file.

The evm statetest will always execute all the tests contained in a file without the possibility of selecting a single test, so this function is cached in order to only call the command once and consume_state_test can simply select the result that was requested.

Source code in src/ethereum_clis/clis/geth.py 
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
@cache  # noqa
def consume_state_test_file(
    self,
    fixture_path: Path,
    debug_output_path: Optional[Path] = None,
) -> List[Dict[str, Any]]:
    """
    Consume an entire state test file.

    The `evm statetest` will always execute all the tests contained in a file without the
    possibility of selecting a single test, so this function is cached in order to only call
    the command once and `consume_state_test` can simply select the result that
    was requested.
    """
    subcommand = "statetest"
    global_options: List[str] = []
    subcommand_options: List[str] = []
    if debug_output_path:
        global_options += ["--verbosity", "100"]
        subcommand_options += ["--trace"]

    command = (
        [str(self.binary)]
        + global_options
        + [subcommand]
        + subcommand_options
        + [str(fixture_path)]
    )
    result = self._run_command(command)

    if debug_output_path:
        self._consume_debug_dump(command, result, fixture_path, debug_output_path)

    if result.returncode != 0:
        raise Exception(
            f"Unexpected exit code:\n{' '.join(command)}\n\n Error:\n{result.stderr}"
        )

    result_json = json.loads(result.stdout)
    if not isinstance(result_json, list):
        raise Exception(f"Unexpected result from evm statetest: {result_json}")
    return result_json

consume_state_test(fixture_path, fixture_name=None, debug_output_path=None)

Consume a single state test.

Uses the cached result from consume_state_test_file in order to not call the command every time an select a single result from there.

Source code in src/ethereum_clis/clis/geth.py 
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
def consume_state_test(
    self,
    fixture_path: Path,
    fixture_name: Optional[str] = None,
    debug_output_path: Optional[Path] = None,
):
    """
    Consume a single state test.

    Uses the cached result from `consume_state_test_file` in order to not call the command
    every time an select a single result from there.
    """
    file_results = self.consume_state_test_file(
        fixture_path=fixture_path,
        debug_output_path=debug_output_path,
    )
    if fixture_name:
        test_result = [
            test_result for test_result in file_results if test_result["name"] == fixture_name
        ]
        assert len(test_result) < 2, f"Multiple test results for {fixture_name}"
        assert len(test_result) == 1, f"Test result for {fixture_name} missing"
        assert test_result[0]["pass"], f"State test failed: {test_result[0]['error']}"
    else:
        if any(not test_result["pass"] for test_result in file_results):
            exception_text = "State test failed: \n" + "\n".join(
                f"{test_result['name']}: " + test_result["error"]
                for test_result in file_results
                if not test_result["pass"]
            )
            raise Exception(exception_text)

consume_fixture(fixture_format, fixture_path, fixture_name=None, debug_output_path=None)

Execute the appropriate geth fixture consumer for the fixture at fixture_path.

Source code in src/ethereum_clis/clis/geth.py 
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
def consume_fixture(
    self,
    fixture_format: FixtureFormat,
    fixture_path: Path,
    fixture_name: Optional[str] = None,
    debug_output_path: Optional[Path] = None,
):
    """Execute the appropriate geth fixture consumer for the fixture at `fixture_path`."""
    if fixture_format == BlockchainFixture:
        self.consume_blockchain_test(
            fixture_path=fixture_path,
            fixture_name=fixture_name,
            debug_output_path=debug_output_path,
        )
    elif fixture_format == StateFixture:
        self.consume_state_test(
            fixture_path=fixture_path,
            fixture_name=fixture_name,
            debug_output_path=debug_output_path,
        )
    else:
        raise Exception(
            f"Fixture format {fixture_format.format_name} not supported by {self.binary}"
        )

GethTransitionTool

Bases: GethEvm, TransitionTool

go-ethereum evm Transition tool interface wrapper class.

Source code in src/ethereum_clis/clis/geth.py 
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
class GethTransitionTool(GethEvm, TransitionTool):
    """go-ethereum `evm` Transition tool interface wrapper class."""

    subcommand: Optional[str] = "t8n"
    trace: bool
    t8n_use_stream = True

    def __init__(self, *, binary: Path, trace: bool = False):
        """Initialize the GethTransitionTool class."""
        super().__init__(binary=binary, trace=trace)
        help_command = [str(self.binary), str(self.subcommand), "--help"]
        result = self._run_command(help_command)
        self.help_string = result.stdout

    def is_fork_supported(self, fork: Fork) -> bool:
        """
        Return 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.transition_tool_name() in self.help_string

__init__(*, binary, trace=False)

Initialize the GethTransitionTool class.

Source code in src/ethereum_clis/clis/geth.py 
230
231
232
233
234
235
def __init__(self, *, binary: Path, trace: bool = False):
    """Initialize the GethTransitionTool class."""
    super().__init__(binary=binary, trace=trace)
    help_command = [str(self.binary), str(self.subcommand), "--help"]
    result = self._run_command(help_command)
    self.help_string = result.stdout

is_fork_supported(fork)

Return 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/ethereum_clis/clis/geth.py 
237
238
239
240
241
242
243
def is_fork_supported(self, fork: Fork) -> bool:
    """
    Return 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.transition_tool_name() in self.help_string

NimbusTransitionTool

Bases: TransitionTool

Nimbus evm Transition tool interface wrapper class.

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

    default_binary = Path("t8n")
    detect_binary_pattern = re.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,
    ):
        """Initialize the Nimbus Transition tool interface."""
        super().__init__(exception_mapper=NimbusExceptionMapper(), 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(
                f"evm process unexpectedly returned a non-zero status code: {e}."
            ) from e
        except Exception as e:
            raise Exception(f"Unexpected exception calling evm tool: {e}.") from e
        self.help_string = result.stdout

    def version(self) -> str:
        """Get `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:
        """
        Return 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.transition_tool_name() in self.help_string

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

Initialize the Nimbus Transition tool interface.

Source code in src/ethereum_clis/clis/nimbus.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
def __init__(
    self,
    *,
    binary: Optional[Path] = None,
    trace: bool = False,
):
    """Initialize the Nimbus Transition tool interface."""
    super().__init__(exception_mapper=NimbusExceptionMapper(), 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(
            f"evm process unexpectedly returned a non-zero status code: {e}."
        ) from e
    except Exception as e:
        raise Exception(f"Unexpected exception calling evm tool: {e}.") from e
    self.help_string = result.stdout

version()

Get evm binary version.

Source code in src/ethereum_clis/clis/nimbus.py 
49
50
51
52
53
54
def version(self) -> str:
    """Get `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)

Return 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/ethereum_clis/clis/nimbus.py 
56
57
58
59
60
61
62
def is_fork_supported(self, fork: Fork) -> bool:
    """
    Return 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.transition_tool_name() in self.help_string

CLINotFoundInPathError

Bases: Exception

Exception raised if the specified CLI binary is not found in the path.

Source code in src/ethereum_clis/ethereum_cli.py 
18
19
20
21
22
23
24
25
class CLINotFoundInPathError(Exception):
    """Exception raised if the specified CLI binary is not found in the path."""

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

__init__(message='The CLI binary was not found in the path', binary=None)

Initialize the exception.

Source code in src/ethereum_clis/ethereum_cli.py 
21
22
23
24
25
def __init__(self, message="The CLI binary was not found in the path", binary=None):
    """Initialize the exception."""
    if binary:
        message = f"{message} ({binary})"
    super().__init__(message)

UnknownCLIError

Bases: Exception

Exception raised if an unknown CLI is encountered.

Source code in src/ethereum_clis/ethereum_cli.py 
12
13
14
15
class UnknownCLIError(Exception):
    """Exception raised if an unknown CLI is encountered."""

    pass

FixtureConsumerTool

Bases: FixtureConsumer, EthereumCLI

Fixture consumer tool abstract base class which should be inherited by all fixture consumer tool implementations.

Source code in src/ethereum_clis/fixture_consumer_tool.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
class FixtureConsumerTool(FixtureConsumer, EthereumCLI):
    """
    Fixture consumer tool abstract base class which should be inherited by all fixture consumer
    tool implementations.
    """

    registered_tools: List[Type["FixtureConsumerTool"]] = []
    default_tool: Type["FixtureConsumerTool"] | None = None

    def __init_subclass__(cls, *, fixture_formats: List[FixtureFormat]):
        """Register all subclasses of FixtureConsumerTool as possible tools."""
        FixtureConsumerTool.register_tool(cls)
        cls.fixture_formats = fixture_formats

__init_subclass__(*, fixture_formats)

Register all subclasses of FixtureConsumerTool as possible tools.

Source code in src/ethereum_clis/fixture_consumer_tool.py 
19
20
21
22
def __init_subclass__(cls, *, fixture_formats: List[FixtureFormat]):
    """Register all subclasses of FixtureConsumerTool as possible tools."""
    FixtureConsumerTool.register_tool(cls)
    cls.fixture_formats = fixture_formats

TransitionTool

Bases: EthereumCLI

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

Source code in src/ethereum_clis/transition_tool.py 
 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
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
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
class TransitionTool(EthereumCLI):
    """
    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

    subcommand: Optional[str] = None
    cached_version: Optional[str] = None
    t8n_use_stream: bool = False
    t8n_use_server: bool = False
    server_url: str
    process: Optional[subprocess.Popen] = None

    @abstractmethod
    def __init__(
        self,
        *,
        exception_mapper: ExceptionMapper,
        binary: Optional[Path] = None,
        trace: bool = False,
    ):
        """Abstract initialization method that all subclasses must implement."""
        self.exception_mapper = exception_mapper
        super().__init__(binary=binary)
        self.trace = trace
        self._info_metadata: Optional[Dict[str, Any]] = {}

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

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

    def start_server(self):
        """
        Start the t8n-server process, extract the port, and leave it running
        for future re-use.
        """
        pass

    def shutdown(self):
        """Perform any cleanup tasks related to the tested tool."""
        pass

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

    def append_traces(self, new_traces: List[List[Dict]]):
        """Append 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:
        """Return the accumulated traces."""
        return self.traces

    def collect_traces(
        self,
        receipts: List[TransactionReceipt],
        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.transaction_hash}.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)

    @dataclass
    class TransitionToolData:
        """Transition tool files and data to pass between methods."""

        alloc: Alloc
        txs: List[Transaction]
        env: Environment
        fork_name: str
        chain_id: int
        reward: int
        blob_schedule: BlobSchedule | None
        state_test: bool

        def to_input(self) -> TransitionToolInput:
            """Convert the data to a TransactionToolInput object."""
            return TransitionToolInput(
                alloc=self.alloc,
                txs=self.txs,
                env=self.env,
            )

        def get_request_data(self) -> TransitionToolRequest:
            """Convert the data to a TransitionToolRequest object."""
            return TransitionToolRequest(
                state=TransitionToolContext(
                    fork=self.fork_name,
                    chain_id=self.chain_id,
                    reward=self.reward,
                    blob_schedule=self.blob_schedule,
                ),
                input=self.to_input(),
            )

    def _evaluate_filesystem(
        self,
        *,
        t8n_data: TransitionToolData,
        debug_output_path: str = "",
    ) -> TransitionToolOutput:
        """Execute a transition tool using the filesystem for its inputs and outputs."""
        temp_dir = tempfile.TemporaryDirectory()
        os.mkdir(os.path.join(temp_dir.name, "input"))
        os.mkdir(os.path.join(temp_dir.name, "output"))

        input_contents = t8n_data.to_input().model_dump(mode="json", **model_dump_config)

        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",
            t8n_data.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(t8n_data.reward),
            "--state.chainid",
            str(t8n_data.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)
        output = TransitionToolOutput(**output_contents)
        if self.trace:
            self.collect_traces(output.result.receipts, temp_dir, debug_output_path)

        temp_dir.cleanup()

        return output

    def _server_post(
        self,
        data: Dict[str, Any],
        timeout: int,
        url_args: Optional[Dict[str, List[str] | str]] = None,
        retries: int = 5,
    ) -> Response:
        """Send a POST request to the t8n-server and return the response."""
        if url_args is None:
            url_args = {}
        post_delay = 0.1
        while True:
            try:
                response = Session().post(
                    f"{self.server_url}?{urlencode(url_args, doseq=True)}",
                    json=data,
                    timeout=timeout,
                )
                break
            except RequestsConnectionError as e:
                retries -= 1
                if retries == 0:
                    raise e
                time.sleep(post_delay)
                post_delay *= 2
        response.raise_for_status()
        if response.status_code != 200:
            raise Exception(
                f"t8n-server returned status code {response.status_code}, "
                f"response: {response.text}"
            )
        return response

    def _generate_post_args(self, t8n_data: TransitionToolData) -> Dict[str, List[str] | str]:
        """Generate the arguments for the POST request to the t8n-server."""
        return {}

    def _evaluate_server(
        self,
        *,
        t8n_data: TransitionToolData,
        debug_output_path: str = "",
        timeout: int,
    ) -> TransitionToolOutput:
        """Execute the transition tool sending inputs and outputs via a server."""
        request_data = t8n_data.get_request_data()
        request_data_json = request_data.model_dump(mode="json", **model_dump_config)

        temp_dir = tempfile.TemporaryDirectory()
        request_data_json["trace"] = self.trace
        if self.trace:
            request_data_json["output-basedir"] = temp_dir.name

        if debug_output_path:
            request_info = (
                f"Server URL: {self.server_url}\n\n"
                f"Request Data:\n{json.dumps(request_data_json, indent=2)}\n"
            )
            dump_files_to_directory(
                debug_output_path,
                {
                    "input/alloc.json": request_data.input.alloc,
                    "input/env.json": request_data.input.env,
                    "input/txs.json": [
                        tx.model_dump(mode="json", **model_dump_config)
                        for tx in request_data.input.txs
                    ],
                    "request_info.txt": request_info,
                },
            )

        response = self._server_post(
            data=request_data_json, url_args=self._generate_post_args(t8n_data), timeout=timeout
        )
        response_json = response.json()

        # pop optional test ``_info`` metadata from response, if present
        self._info_metadata = response_json.pop("_info_metadata", {})

        output: TransitionToolOutput = TransitionToolOutput.model_validate(response_json)

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

        if debug_output_path:
            response_info = (
                f"Status Code: {response.status_code}\n\n"
                f"Headers:\n{json.dumps(dict(response.headers), indent=2)}\n\n"
                f"Content:\n{response.text}\n"
            )
            dump_files_to_directory(
                debug_output_path,
                {
                    "output/alloc.json": output.alloc,
                    "output/result.json": output.result,
                    "output/txs.rlp": str(output.body),
                    "response_info.txt": response_info,
                },
            )

        return output

    def _evaluate_stream(
        self,
        *,
        t8n_data: TransitionToolData,
        debug_output_path: str = "",
    ) -> TransitionToolOutput:
        """Execute a transition tool using stdin and stdout for its inputs and outputs."""
        temp_dir = tempfile.TemporaryDirectory()
        args = self.construct_args_stream(t8n_data, temp_dir)

        stdin = t8n_data.to_input()

        result = subprocess.run(
            args,
            input=stdin.model_dump_json(**model_dump_config).encode(),
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )

        self.dump_debug_stream(debug_output_path, temp_dir, stdin, args, result)

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

        output: TransitionToolOutput = TransitionToolOutput.model_validate_json(result.stdout)

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

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

        return output

    def construct_args_stream(
        self, t8n_data: TransitionToolData, temp_dir: tempfile.TemporaryDirectory
    ) -> List[str]:
        """Construct arguments for t8n interaction via streams."""
        command: list[str] = [str(self.binary)]
        if self.subcommand:
            command.append(self.subcommand)

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

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

    def dump_debug_stream(
        self,
        debug_output_path: str,
        temp_dir: tempfile.TemporaryDirectory,
        stdin: TransitionToolInput,
        args: List[str],
        result: subprocess.CompletedProcess,
    ):
        """Export debug files if requested when interacting with t8n via streams."""
        if not debug_output_path:
            return

        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": [
                    tx.model_dump(mode="json", **model_dump_config) for tx in stdin.txs
                ],
                "returncode.txt": result.returncode,
                "stdin.txt": stdin,
                "stdout.txt": result.stdout.decode(),
                "stderr.txt": result.stderr.decode(),
                "t8n.sh+x": t8n_script,
            },
        )

    def evaluate(
        self,
        *,
        alloc: Alloc,
        txs: List[Transaction],
        env: Environment,
        fork: Fork,
        chain_id: int,
        reward: int,
        blob_schedule: BlobSchedule | None,
        eips: Optional[List[int]] = None,
        debug_output_path: str = "",
        state_test: bool = False,
        slow_request: bool = False,
    ) -> TransitionToolOutput:
        """
        Execute the relevant evaluate method as required by the `t8n` tool.

        If a client's `t8n` tool varies from the default behavior, this method
        can be overridden.
        """
        fork_name = fork.transition_tool_name(
            block_number=env.number,
            timestamp=env.timestamp,
        )
        if eips is not None:
            fork_name = "+".join([fork_name] + [str(eip) for eip in eips])
        if env.number == 0:
            reward = -1
        t8n_data = self.TransitionToolData(
            alloc=alloc,
            txs=txs,
            env=env,
            fork_name=fork_name,
            chain_id=chain_id,
            reward=reward,
            blob_schedule=blob_schedule,
            state_test=state_test,
        )

        if self.t8n_use_server:
            if not self.process:
                self.start_server()
            return self._evaluate_server(
                t8n_data=t8n_data,
                debug_output_path=debug_output_path,
                timeout=SLOW_REQUEST_TIMEOUT if slow_request else NORMAL_SERVER_TIMEOUT,
            )

        if self.t8n_use_stream:
            return self._evaluate_stream(t8n_data=t8n_data, debug_output_path=debug_output_path)

        return self._evaluate_filesystem(
            t8n_data=t8n_data,
            debug_output_path=debug_output_path,
        )

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

Abstract initialization method that all subclasses must implement.

Source code in src/ethereum_clis/transition_tool.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
@abstractmethod
def __init__(
    self,
    *,
    exception_mapper: ExceptionMapper,
    binary: Optional[Path] = None,
    trace: bool = False,
):
    """Abstract initialization method that all subclasses must implement."""
    self.exception_mapper = exception_mapper
    super().__init__(binary=binary)
    self.trace = trace
    self._info_metadata: Optional[Dict[str, Any]] = {}

__init_subclass__()

Register all subclasses of TransitionTool as possible tools.

Source code in src/ethereum_clis/transition_tool.py 
73
74
75
def __init_subclass__(cls):
    """Register all subclasses of TransitionTool as possible tools."""
    TransitionTool.register_tool(cls)

is_fork_supported(fork) abstractmethod

Return True if the fork is supported by the tool.

Source code in src/ethereum_clis/transition_tool.py 
77
78
79
80
@abstractmethod
def is_fork_supported(self, fork: Fork) -> bool:
    """Return True if the fork is supported by the tool."""
    pass

start_server()

Start the t8n-server process, extract the port, and leave it running for future re-use.

Source code in src/ethereum_clis/transition_tool.py 
82
83
84
85
86
87
def start_server(self):
    """
    Start the t8n-server process, extract the port, and leave it running
    for future re-use.
    """
    pass

shutdown()

Perform any cleanup tasks related to the tested tool.

Source code in src/ethereum_clis/transition_tool.py 
89
90
91
def shutdown(self):
    """Perform any cleanup tasks related to the tested tool."""
    pass

reset_traces()

Reset the internal trace storage for a new test to begin.

Source code in src/ethereum_clis/transition_tool.py 
93
94
95
def reset_traces(self):
    """Reset the internal trace storage for a new test to begin."""
    self.traces = None

append_traces(new_traces)

Append a list of traces of a state transition to the current list.

Source code in src/ethereum_clis/transition_tool.py 
 97
 98
 99
100
101
def append_traces(self, new_traces: List[List[Dict]]):
    """Append 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()

Return the accumulated traces.

Source code in src/ethereum_clis/transition_tool.py 
103
104
105
def get_traces(self) -> List[List[List[Dict]]] | None:
    """Return 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/ethereum_clis/transition_tool.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
def collect_traces(
    self,
    receipts: List[TransactionReceipt],
    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.transaction_hash}.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)

TransitionToolData dataclass

Transition tool files and data to pass between methods.

Source code in src/ethereum_clis/transition_tool.py 
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
@dataclass
class TransitionToolData:
    """Transition tool files and data to pass between methods."""

    alloc: Alloc
    txs: List[Transaction]
    env: Environment
    fork_name: str
    chain_id: int
    reward: int
    blob_schedule: BlobSchedule | None
    state_test: bool

    def to_input(self) -> TransitionToolInput:
        """Convert the data to a TransactionToolInput object."""
        return TransitionToolInput(
            alloc=self.alloc,
            txs=self.txs,
            env=self.env,
        )

    def get_request_data(self) -> TransitionToolRequest:
        """Convert the data to a TransitionToolRequest object."""
        return TransitionToolRequest(
            state=TransitionToolContext(
                fork=self.fork_name,
                chain_id=self.chain_id,
                reward=self.reward,
                blob_schedule=self.blob_schedule,
            ),
            input=self.to_input(),
        )

to_input()

Convert the data to a TransactionToolInput object.

Source code in src/ethereum_clis/transition_tool.py 
142
143
144
145
146
147
148
def to_input(self) -> TransitionToolInput:
    """Convert the data to a TransactionToolInput object."""
    return TransitionToolInput(
        alloc=self.alloc,
        txs=self.txs,
        env=self.env,
    )

get_request_data()

Convert the data to a TransitionToolRequest object.

Source code in src/ethereum_clis/transition_tool.py 
150
151
152
153
154
155
156
157
158
159
160
def get_request_data(self) -> TransitionToolRequest:
    """Convert the data to a TransitionToolRequest object."""
    return TransitionToolRequest(
        state=TransitionToolContext(
            fork=self.fork_name,
            chain_id=self.chain_id,
            reward=self.reward,
            blob_schedule=self.blob_schedule,
        ),
        input=self.to_input(),
    )

construct_args_stream(t8n_data, temp_dir)

Construct arguments for t8n interaction via streams.

Source code in src/ethereum_clis/transition_tool.py 
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
def construct_args_stream(
    self, t8n_data: TransitionToolData, temp_dir: tempfile.TemporaryDirectory
) -> List[str]:
    """Construct arguments for t8n interaction via streams."""
    command: list[str] = [str(self.binary)]
    if self.subcommand:
        command.append(self.subcommand)

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

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

dump_debug_stream(debug_output_path, temp_dir, stdin, args, result)

Export debug files if requested when interacting with t8n via streams.

Source code in src/ethereum_clis/transition_tool.py 
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
def dump_debug_stream(
    self,
    debug_output_path: str,
    temp_dir: tempfile.TemporaryDirectory,
    stdin: TransitionToolInput,
    args: List[str],
    result: subprocess.CompletedProcess,
):
    """Export debug files if requested when interacting with t8n via streams."""
    if not debug_output_path:
        return

    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": [
                tx.model_dump(mode="json", **model_dump_config) for tx in stdin.txs
            ],
            "returncode.txt": result.returncode,
            "stdin.txt": stdin,
            "stdout.txt": result.stdout.decode(),
            "stderr.txt": result.stderr.decode(),
            "t8n.sh+x": t8n_script,
        },
    )

evaluate(*, alloc, txs, env, fork, chain_id, reward, blob_schedule, eips=None, debug_output_path='', state_test=False, slow_request=False)

Execute the relevant evaluate method as required by the t8n tool.

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

Source code in src/ethereum_clis/transition_tool.py 
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
def evaluate(
    self,
    *,
    alloc: Alloc,
    txs: List[Transaction],
    env: Environment,
    fork: Fork,
    chain_id: int,
    reward: int,
    blob_schedule: BlobSchedule | None,
    eips: Optional[List[int]] = None,
    debug_output_path: str = "",
    state_test: bool = False,
    slow_request: bool = False,
) -> TransitionToolOutput:
    """
    Execute the relevant evaluate method as required by the `t8n` tool.

    If a client's `t8n` tool varies from the default behavior, this method
    can be overridden.
    """
    fork_name = fork.transition_tool_name(
        block_number=env.number,
        timestamp=env.timestamp,
    )
    if eips is not None:
        fork_name = "+".join([fork_name] + [str(eip) for eip in eips])
    if env.number == 0:
        reward = -1
    t8n_data = self.TransitionToolData(
        alloc=alloc,
        txs=txs,
        env=env,
        fork_name=fork_name,
        chain_id=chain_id,
        reward=reward,
        blob_schedule=blob_schedule,
        state_test=state_test,
    )

    if self.t8n_use_server:
        if not self.process:
            self.start_server()
        return self._evaluate_server(
            t8n_data=t8n_data,
            debug_output_path=debug_output_path,
            timeout=SLOW_REQUEST_TIMEOUT if slow_request else NORMAL_SERVER_TIMEOUT,
        )

    if self.t8n_use_stream:
        return self._evaluate_stream(t8n_data=t8n_data, debug_output_path=debug_output_path)

    return self._evaluate_filesystem(
        t8n_data=t8n_data,
        debug_output_path=debug_output_path,
    )

Result

Bases: CamelModel

Result of a transition tool output.

Source code in src/ethereum_clis/types.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
class Result(CamelModel):
    """Result of a transition tool output."""

    state_root: Hash
    ommers_hash: Hash | None = Field(None, validation_alias="sha3Uncles")
    transactions_trie: Hash = Field(..., alias="txRoot")
    receipts_root: Hash
    logs_hash: Hash
    logs_bloom: Bloom
    receipts: List[TransactionReceipt]
    rejected_transactions: List[RejectedTransaction] = Field(
        default_factory=list, alias="rejected"
    )
    difficulty: HexNumber | None = Field(None, alias="currentDifficulty")
    gas_used: HexNumber
    base_fee_per_gas: HexNumber | None = Field(None, alias="currentBaseFee")
    withdrawals_root: Hash | None = None
    excess_blob_gas: HexNumber | None = Field(None, alias="currentExcessBlobGas")
    blob_gas_used: HexNumber | None = None
    requests_hash: Hash | None = None
    requests: List[Bytes] | None = None

TransitionToolOutput

Bases: CamelModel

Transition tool output.

Source code in src/ethereum_clis/types.py 
49
50
51
52
53
54
class TransitionToolOutput(CamelModel):
    """Transition tool output."""

    alloc: Alloc
    result: Result
    body: Bytes | None = None