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

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
    besu_trace_dir: Optional[tempfile.TemporaryDirectory]

    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
        self.besu_trace_dir = tempfile.TemporaryDirectory() if self.trace else None

    def start_server(self):
        """
        Starts the t8n-server process, extracts the port, and leaves 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):
        """
        Stops 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,
        vkt: Any = None,
        chain_id: int = 1,
        reward: int = 0,
        eips: Optional[List[int]] = None,
        debug_output_path: str = "",
    ) -> TransitionToolOutput:
        """
        Executes `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}'
                """  # 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: 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:
        """
        Returns True if the fork is supported by the tool
        """
        return fork.transition_tool_name() 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

def start_server(self):
    """
    Starts the t8n-server process, extracts the port, and leaves 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()

Stops the t8n-server process if it was started

Source code in src/evm_transition_tool/besu.py

def shutdown(self):
    """
    Stops 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, vkt=None, 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

def evaluate(
    self,
    *,
    alloc: Alloc,
    txs: List[Transaction],
    env: Environment,
    fork: Fork,
    vkt: Any = None,
    chain_id: int = 1,
    reward: int = 0,
    eips: Optional[List[int]] = None,
    debug_output_path: str = "",
) -> TransitionToolOutput:
    """
    Executes `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}'
            """  # 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: 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)

Returns True if the fork is supported by the tool

Source code in src/evm_transition_tool/besu.py

def is_fork_supported(self, fork: Fork) -> bool:
    """
    Returns 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/evm_transition_tool/ethereumjs.py

class EthereumJSTransitionTool(TransitionTool):
    """
    EthereumJS Transition tool interface wrapper class.
    """

    default_binary = Path("ethereumjs-t8ntool.sh")
    detect_binary_pattern = 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,
    ):
        super().__init__(binary=binary, trace=trace)

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

is_fork_supported(fork)

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

Source code in src/evm_transition_tool/ethereumjs.py

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

EvmOneTransitionTool

Bases: TransitionTool

Evmone evmone-t8n Transition tool interface wrapper class.

Source code in src/evm_transition_tool/evmone.py

class EvmOneTransitionTool(TransitionTool):
    """
    Evmone `evmone-t8n` Transition tool interface wrapper class.
    """

    default_binary = Path("evmone-t8n")
    detect_binary_pattern = 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,
    ):
        super().__init__(binary=binary, trace=trace)

    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

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

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: 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/evm_transition_tool/execution_specs.py

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 = 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,
    ):
        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(
                "ethereum-spec-evm-resolver process unexpectedly returned a non-zero status code: "
                f"{e}."
            )
        except Exception as e:
            raise Exception(f"Unexpected exception calling ethereum-spec-evm-resolver: {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.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):
        """
        Stops 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:
        """
        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.

        `ethereum-spec-evm` appends newlines to forks in the help string.
        """
        return (fork.transition_tool_name() + "\n") 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/execution_specs.py

def start_server(self):
    """
    Starts the t8n-server process, extracts the port, and leaves 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()

Stops the t8n-server process if it was started.

Source code in src/evm_transition_tool/execution_specs.py

def shutdown(self):
    """
    Stops 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)

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.

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

Source code in src/evm_transition_tool/execution_specs.py

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.

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

GethTransitionTool

Bases: TransitionTool

Go-ethereum evm Transition tool interface wrapper class.

Source code in src/evm_transition_tool/geth.py

class GethTransitionTool(TransitionTool):
    """
    Go-ethereum `evm` Transition tool interface wrapper class.
    """

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

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

    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.transition_tool_name() in self.help_string

    def get_blocktest_help(self) -> str:
        """
        Return the help string for the blocktest subcommand.
        """
        args = [str(self.binary), "blocktest", "--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}.")
        return result.stdout

    def is_verifiable(
        self,
        fixture_format: FixtureFormat,
    ) -> bool:
        """
        Returns whether the fixture format is verifiable by this Geth's evm tool.
        """
        return fixture_format in {StateFixture, BlockchainFixture}

    def verify_fixture(
        self,
        fixture_format: FixtureFormat,
        fixture_path: Path,
        fixture_name: Optional[str] = None,
        debug_output_path: Optional[Path] = None,
    ):
        """
        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 fixture_format == StateFixture:
            assert self.statetest_subcommand, "statetest subcommand not set"
            command.append(self.statetest_subcommand)
        elif fixture_format == BlockchainFixture:
            assert self.blocktest_subcommand, "blocktest subcommand not set"
            command.append(self.blocktest_subcommand)
        else:
            raise Exception(f"Invalid test fixture format: {fixture_format}")

        if fixture_name and fixture_format == BlockchainFixture:
            assert isinstance(fixture_name, str), "fixture_name must be a string"
            command.append("--run")
            command.append(fixture_name)
        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 / "fixtures.json"
            # 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,
                },
            )
            shutil.copyfile(fixture_path, debug_fixture_path)

        if result.returncode != 0:
            raise Exception(
                f"EVM test failed.\n{' '.join(command)}\n\n Error:\n{result.stderr.decode()}"
            )

        if fixture_format == StateFixture:
            result_json = json.loads(result.stdout.decode())
            if not isinstance(result_json, list):
                raise Exception(f"Unexpected result from evm statetest: {result_json}")
        else:
            result_json = []  # there is no parseable format for blocktest output
        return result_json

    def _run_verkle_command(self, subcommand: str, *args: str) -> str:
        """
        Helper function to run a verkle subcommand and return the output as a string.
        """
        command = [
            str(self.binary),
            str(self.verkle_subcommand),
            subcommand,
            *args,
        ]
        result = subprocess.run(
            command,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )
        if result.returncode != 0:
            raise Exception(
                f"Failed to run verkle subcommand: '{' '.join(command)}'. "
                f"Error: '{result.stderr.decode()}'"
            )
        return result.stdout.decode().strip()

    def from_mpt_to_vkt(self, mpt_alloc: Alloc) -> VerkleTree:
        """
        Returns the verkle tree representation for an input MPT.
        """
        with tempfile.TemporaryDirectory() as temp_dir:
            alloc_path = os.path.join(temp_dir, "alloc.json")
            with open(alloc_path, "w") as f:
                json.dump(to_json(mpt_alloc), f)

            output = self._run_verkle_command("tree-keys", "--input.alloc", alloc_path)
            return VerkleTree(json.loads(output))

    def get_verkle_state_root(self, mpt_alloc: Alloc) -> bytes:
        """
        Returns the VKT state root from an input MPT.
        """
        with tempfile.TemporaryDirectory() as temp_dir:
            alloc_path = os.path.join(temp_dir, "alloc.json")
            with open(alloc_path, "w") as f:
                json.dump(to_json(mpt_alloc), f)

            hex_string = self._run_verkle_command(
                "state-root", "--input.alloc", alloc_path
            )
            return binascii.unhexlify(hex_string[2:])

    def get_verkle_single_key(
        self, address: Address, storage_slot: Optional[ZeroPaddedHexNumber] = None
    ) -> str:
        """
        Returns the VKT key for an account address or storage slot.
        """
        args = [str(address)]
        if storage_slot is not None:
            args.append(str(storage_slot))
        output = self._run_verkle_command("single-key", *args)
        return output

    def get_verkle_code_chunk_key(
        self, address: Address, code_chunk: ZeroPaddedHexNumber
    ) -> str:
        """
        Returns the VKT key of a code chunk for an account address.
        """
        output = self._run_verkle_command(
            "code-chunk-key", str(address), str(code_chunk)
        )
        return output

    def get_witness_check_mapping(
        self, witness_check: WitnessCheck
    ) -> Tuple[StateDiff, Dict[Stem, Address]]:
        """
        Returns a tuple containing:
        A) StateDiff - A pseudo StateDiff type with stems, suffixes, and current values.
        B) Dict[Stem, Address] - A mapping of stems to their associated addresses.
        """
        stem_account_mapping: Dict[Stem, Address] = {}
        state_diff: List[StemStateDiff] = []

        # Format account entries using `evm verkle single-key`
        if witness_check.account_entries:
            for address, entry, value in witness_check.account_entries:
                account_tree_key_str = self.get_verkle_single_key(address)
                tree_key = bytearray.fromhex(account_tree_key_str[2:])
                stem = Stem(bytes(tree_key[:-1]))
                stem_account_mapping[stem] = address
                suffix_state_diff = SuffixStateDiff(
                    suffix=entry,
                    current_value=value,
                )
                stem_state_diff = next(
                    (sd for sd in state_diff if sd.stem == stem), None
                )
                if stem_state_diff is None:
                    stem_state_diff = StemStateDiff(stem=stem, suffix_diffs=[])
                    state_diff.append(stem_state_diff)
                stem_state_diff.suffix_diffs.append(suffix_state_diff)

        # Format storage entries using `evm verkle single-key`
        if witness_check.storage_slots:
            for address, storage_slot, value in witness_check.storage_slots:
                storage_tree_key_str = self.get_verkle_single_key(
                    address, ZeroPaddedHexNumber(storage_slot)
                )
                tree_key = bytearray.fromhex(storage_tree_key_str[2:])
                stem = Stem(bytes(tree_key[:-1]))
                suffix = int(tree_key[-1])
                stem_account_mapping[stem] = address
                suffix_state_diff = SuffixStateDiff(suffix=suffix, current_value=value)
                stem_state_diff = next(
                    (sd for sd in state_diff if sd.stem == stem), None
                )
                if stem_state_diff is None:
                    stem_state_diff = StemStateDiff(stem=stem, suffix_diffs=[])
                    state_diff.append(stem_state_diff)
                stem_state_diff.suffix_diffs.append(suffix_state_diff)

        # Format code chunks using `evm verkle code-chunk-key`
        if witness_check.code_chunks:
            for address, code_chunk, value in witness_check.code_chunks:
                code_chunk_tree_key_str = self.get_verkle_code_chunk_key(
                    address, ZeroPaddedHexNumber(code_chunk)
                )
                tree_key = bytearray.fromhex(code_chunk_tree_key_str[2:])
                stem = Stem(bytes(tree_key[:-1]))
                suffix = int(tree_key[-1])
                stem_account_mapping[stem] = address
                suffix_state_diff = SuffixStateDiff(suffix=suffix, current_value=value)
                stem_state_diff = next(
                    (sd for sd in state_diff if sd.stem == stem), None
                )
                if stem_state_diff is None:
                    stem_state_diff = StemStateDiff(stem=stem, suffix_diffs=[])
                    state_diff.append(stem_state_diff)
                stem_state_diff.suffix_diffs.append(suffix_state_diff)

        return StateDiff(root=state_diff), stem_account_mapping

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

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.transition_tool_name() in self.help_string

get_blocktest_help()

Return the help string for the blocktest subcommand.

Source code in src/evm_transition_tool/geth.py

def get_blocktest_help(self) -> str:
    """
    Return the help string for the blocktest subcommand.
    """
    args = [str(self.binary), "blocktest", "--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}.")
    return result.stdout

is_verifiable(fixture_format)

Returns whether the fixture format is verifiable by this Geth's evm tool.

Source code in src/evm_transition_tool/geth.py

def is_verifiable(
    self,
    fixture_format: FixtureFormat,
) -> bool:
    """
    Returns whether the fixture format is verifiable by this Geth's evm tool.
    """
    return fixture_format in {StateFixture, BlockchainFixture}

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

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

Source code in src/evm_transition_tool/geth.py

def verify_fixture(
    self,
    fixture_format: FixtureFormat,
    fixture_path: Path,
    fixture_name: Optional[str] = None,
    debug_output_path: Optional[Path] = None,
):
    """
    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 fixture_format == StateFixture:
        assert self.statetest_subcommand, "statetest subcommand not set"
        command.append(self.statetest_subcommand)
    elif fixture_format == BlockchainFixture:
        assert self.blocktest_subcommand, "blocktest subcommand not set"
        command.append(self.blocktest_subcommand)
    else:
        raise Exception(f"Invalid test fixture format: {fixture_format}")

    if fixture_name and fixture_format == BlockchainFixture:
        assert isinstance(fixture_name, str), "fixture_name must be a string"
        command.append("--run")
        command.append(fixture_name)
    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 / "fixtures.json"
        # 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,
            },
        )
        shutil.copyfile(fixture_path, debug_fixture_path)

    if result.returncode != 0:
        raise Exception(
            f"EVM test failed.\n{' '.join(command)}\n\n Error:\n{result.stderr.decode()}"
        )

    if fixture_format == StateFixture:
        result_json = json.loads(result.stdout.decode())
        if not isinstance(result_json, list):
            raise Exception(f"Unexpected result from evm statetest: {result_json}")
    else:
        result_json = []  # there is no parseable format for blocktest output
    return result_json

from_mpt_to_vkt(mpt_alloc)

Returns the verkle tree representation for an input MPT.

Source code in src/evm_transition_tool/geth.py

def from_mpt_to_vkt(self, mpt_alloc: Alloc) -> VerkleTree:
    """
    Returns the verkle tree representation for an input MPT.
    """
    with tempfile.TemporaryDirectory() as temp_dir:
        alloc_path = os.path.join(temp_dir, "alloc.json")
        with open(alloc_path, "w") as f:
            json.dump(to_json(mpt_alloc), f)

        output = self._run_verkle_command("tree-keys", "--input.alloc", alloc_path)
        return VerkleTree(json.loads(output))

get_verkle_state_root(mpt_alloc)

Returns the VKT state root from an input MPT.

Source code in src/evm_transition_tool/geth.py

def get_verkle_state_root(self, mpt_alloc: Alloc) -> bytes:
    """
    Returns the VKT state root from an input MPT.
    """
    with tempfile.TemporaryDirectory() as temp_dir:
        alloc_path = os.path.join(temp_dir, "alloc.json")
        with open(alloc_path, "w") as f:
            json.dump(to_json(mpt_alloc), f)

        hex_string = self._run_verkle_command(
            "state-root", "--input.alloc", alloc_path
        )
        return binascii.unhexlify(hex_string[2:])

get_verkle_single_key(address, storage_slot=None)

Returns the VKT key for an account address or storage slot.

Source code in src/evm_transition_tool/geth.py

def get_verkle_single_key(
    self, address: Address, storage_slot: Optional[ZeroPaddedHexNumber] = None
) -> str:
    """
    Returns the VKT key for an account address or storage slot.
    """
    args = [str(address)]
    if storage_slot is not None:
        args.append(str(storage_slot))
    output = self._run_verkle_command("single-key", *args)
    return output

get_verkle_code_chunk_key(address, code_chunk)

Returns the VKT key of a code chunk for an account address.

Source code in src/evm_transition_tool/geth.py

def get_verkle_code_chunk_key(
    self, address: Address, code_chunk: ZeroPaddedHexNumber
) -> str:
    """
    Returns the VKT key of a code chunk for an account address.
    """
    output = self._run_verkle_command(
        "code-chunk-key", str(address), str(code_chunk)
    )
    return output

get_witness_check_mapping(witness_check)

Returns a tuple containing: A) StateDiff - A pseudo StateDiff type with stems, suffixes, and current values. B) Dict[Stem, Address] - A mapping of stems to their associated addresses.

Source code in src/evm_transition_tool/geth.py

def get_witness_check_mapping(
    self, witness_check: WitnessCheck
) -> Tuple[StateDiff, Dict[Stem, Address]]:
    """
    Returns a tuple containing:
    A) StateDiff - A pseudo StateDiff type with stems, suffixes, and current values.
    B) Dict[Stem, Address] - A mapping of stems to their associated addresses.
    """
    stem_account_mapping: Dict[Stem, Address] = {}
    state_diff: List[StemStateDiff] = []

    # Format account entries using `evm verkle single-key`
    if witness_check.account_entries:
        for address, entry, value in witness_check.account_entries:
            account_tree_key_str = self.get_verkle_single_key(address)
            tree_key = bytearray.fromhex(account_tree_key_str[2:])
            stem = Stem(bytes(tree_key[:-1]))
            stem_account_mapping[stem] = address
            suffix_state_diff = SuffixStateDiff(
                suffix=entry,
                current_value=value,
            )
            stem_state_diff = next(
                (sd for sd in state_diff if sd.stem == stem), None
            )
            if stem_state_diff is None:
                stem_state_diff = StemStateDiff(stem=stem, suffix_diffs=[])
                state_diff.append(stem_state_diff)
            stem_state_diff.suffix_diffs.append(suffix_state_diff)

    # Format storage entries using `evm verkle single-key`
    if witness_check.storage_slots:
        for address, storage_slot, value in witness_check.storage_slots:
            storage_tree_key_str = self.get_verkle_single_key(
                address, ZeroPaddedHexNumber(storage_slot)
            )
            tree_key = bytearray.fromhex(storage_tree_key_str[2:])
            stem = Stem(bytes(tree_key[:-1]))
            suffix = int(tree_key[-1])
            stem_account_mapping[stem] = address
            suffix_state_diff = SuffixStateDiff(suffix=suffix, current_value=value)
            stem_state_diff = next(
                (sd for sd in state_diff if sd.stem == stem), None
            )
            if stem_state_diff is None:
                stem_state_diff = StemStateDiff(stem=stem, suffix_diffs=[])
                state_diff.append(stem_state_diff)
            stem_state_diff.suffix_diffs.append(suffix_state_diff)

    # Format code chunks using `evm verkle code-chunk-key`
    if witness_check.code_chunks:
        for address, code_chunk, value in witness_check.code_chunks:
            code_chunk_tree_key_str = self.get_verkle_code_chunk_key(
                address, ZeroPaddedHexNumber(code_chunk)
            )
            tree_key = bytearray.fromhex(code_chunk_tree_key_str[2:])
            stem = Stem(bytes(tree_key[:-1]))
            suffix = int(tree_key[-1])
            stem_account_mapping[stem] = address
            suffix_state_diff = SuffixStateDiff(suffix=suffix, current_value=value)
            stem_state_diff = next(
                (sd for sd in state_diff if sd.stem == stem), None
            )
            if stem_state_diff is None:
                stem_state_diff = StemStateDiff(stem=stem, suffix_diffs=[])
                state_diff.append(stem_state_diff)
            stem_state_diff.suffix_diffs.append(suffix_state_diff)

    return StateDiff(root=state_diff), stem_account_mapping

NimbusTransitionTool

Bases: TransitionTool

Nimbus evm Transition tool interface wrapper class.

Source code in src/evm_transition_tool/nimbus.py

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.transition_tool_name() in self.help_string

version()

Gets evm binary version.

Source code in src/evm_transition_tool/nimbus.py

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

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.transition_tool_name() in self.help_string

TransitionTool

Bases: FixtureVerifier

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

Source code in src/evm_transition_tool/transition_tool.py

class TransitionTool(FixtureVerifier):
    """
    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
    verkle_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

    # 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 start_server(self):
        """
        Starts the t8n-server process, extracts the port, and leaves it running for future re-use.
        """
        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[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 = field(default=1)
        reward: int = field(default=0)
        vkt: Optional[VerkleTree] = None

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

    def _evaluate_filesystem(
        self,
        *,
        t8n_data: TransitionToolData,
        debug_output_path: str = "",
    ) -> TransitionToolOutput:
        """
        Executes 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")

        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),
        ]

        # TODO: Verkle specific logic, update when Verkle fork is confirmed.
        if t8n_data.fork_name == "Verkle":
            output_paths.update(
                {
                    "vkt": os.path.join("output", "vkt.json"),
                    "witness": os.path.join("output", "witness.json"),
                }
            )
            args.extend(
                [
                    "--output.vkt",
                    output_paths["vkt"],
                    "--output.witness",
                    output_paths["witness"],
                ]
            )
            if t8n_data.vkt is not None:
                args.extend(["--input.vkt", input_paths["vkt"]])

        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 _evaluate_server(
        self,
        *,
        t8n_data: TransitionToolData,
        debug_output_path: str = "",
    ) -> TransitionToolOutput:
        """
        Executes the transition tool sending inputs and outputs via a server.
        """
        input_contents = t8n_data.to_input()
        input_json = input_contents.model_dump(mode="json", **model_dump_config)
        post_data = {
            "state": {
                "fork": t8n_data.fork_name,
                "chainid": t8n_data.chain_id,
                "reward": t8n_data.reward,
            },
            "input": input_json,
        }

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

        response = Session().post(self.server_url, json=post_data, timeout=20)
        response.raise_for_status()  # exception visible in pytest failure output
        if response.status_code != 200:
            raise Exception(
                f"t8n-server returned status code {response.status_code}, "
                f"response: {response.text}"
            )

        output: TransitionToolOutput = TransitionToolOutput.model_validate(
            response.json()
        )

        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:
        """
        Executes 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:
            files_to_dump = {
                "output/alloc.json": output.alloc,
                "output/result.json": output.result,
                "output/txs.rlp": str(output.body),
            }
            # Only dump verkle if present
            if output.vkt:
                files_to_dump["output/vkt.json"] = output.vkt

            dump_files_to_directory(
                debug_output_path,
                files_to_dump,
            )

        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.t8n_subcommand:
            command.append(self.t8n_subcommand)

        args = command + [
            "--input.alloc=stdin",
            "--input.txs=stdin",
            "--input.env=stdin",
        ]

        args += [
            "--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}",
        ]

        # TODO: Verkle specific logic, update when Verkle fork is confirmed.
        if t8n_data.fork_name == "Verkle":
            args.append("--output.vkt=stdout")
            if t8n_data.vkt is not None:
                args.append("--input.vkt=stdin")

        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
            """
        )

        files_to_dump = {
            "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,
        }
        # Only dump verkle if present
        if stdin.vkt:
            files_to_dump["input/vkt.json"] = stdin.vkt

        dump_files_to_directory(
            debug_output_path,
            files_to_dump,
        )

    def evaluate(
        self,
        *,
        alloc: Alloc,
        txs: List[Transaction],
        env: Environment,
        fork: Fork,
        chain_id: int = 1,
        reward: int = 0,
        vkt: Optional[VerkleTree] = None,
        eips: Optional[List[int]] = None,
        debug_output_path: str = "",
    ) -> TransitionToolOutput:
        """
        Executes 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.
        """
        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])
        if env.number == 0:
            reward = -1
        t8n_data = self.TransitionToolData(
            alloc=alloc,
            txs=txs,
            env=env,
            fork_name=fork_name,
            vkt=vkt,
            chain_id=chain_id,
            reward=reward,
        )

        if self.t8n_use_server:
            return self._evaluate_server(
                t8n_data=t8n_data, debug_output_path=debug_output_path
            )

        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,
        )

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

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

    def from_mpt_to_vkt(self, mpt_alloc: Alloc) -> VerkleTree:
        """
        Returns the verkle tree representation for an input MPT.
        """
        raise NotImplementedError(
            "The `from_mpt_to_vkt()` function is not supported by this tool. Use geth's evm tool."
        )

    def get_verkle_state_root(self, mpt_alloc: Alloc) -> bytes:
        """
        Returns the VKT state root from an input MPT.
        """
        raise NotImplementedError(
            "The `get_verkle_state_root()` function is not supported by this tool. Use geth's evm"
            " tool."
        )

    def get_verkle_single_key(
        self, address: Address, storage_slot: Optional[ZeroPaddedHexNumber] = None
    ) -> str:
        """
        Returns the VKT key for an account address or storage slot.
        """
        raise NotImplementedError(
            "The `get_verkle_single_key()` function is not supported by this tool. Use geth's evm"
            " tool."
        )

    def get_verkle_code_chunk_key(
        self, address: Address, code_chunk: ZeroPaddedHexNumber
    ) -> str:
        """
        Returns the VKT key of a code chunk for an account address.
        """
        raise NotImplementedError(
            "The `get_verkle_code_chunk_key()` function is not supported by this tool. Use geth's"
            " evm tool."
        )

    def get_witness_check_mapping(
        self, witness_check: WitnessCheck
    ) -> Tuple[StateDiff, Dict[Stem, Address]]:
        """
        Returns a tuple containing:
        A) StateDiff - A pseudo StateDiff type with stems, suffixes, and current values.
        B) Dict[Stem, Address] - A mapping of stems to their associated addresses.
        """
        raise NotImplementedError(
            "The `get_witness_check_mapping()` 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

@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

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

@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

@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

@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

@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

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

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

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/transition_tool.py

def start_server(self):
    """
    Starts the t8n-server process, extracts the port, and leaves it running for future re-use.
    """
    pass

shutdown()

Perform any cleanup tasks related to the tested tool.

Source code in src/evm_transition_tool/transition_tool.py

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

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

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

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

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/evm_transition_tool/transition_tool.py

@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 = field(default=1)
    reward: int = field(default=0)
    vkt: Optional[VerkleTree] = None

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

to_input()

Convert the data to a TransactionToolInput object

Source code in src/evm_transition_tool/transition_tool.py

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

construct_args_stream(t8n_data, temp_dir)

Construct arguments for t8n interaction via streams

Source code in src/evm_transition_tool/transition_tool.py

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.t8n_subcommand:
        command.append(self.t8n_subcommand)

    args = command + [
        "--input.alloc=stdin",
        "--input.txs=stdin",
        "--input.env=stdin",
    ]

    args += [
        "--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}",
    ]

    # TODO: Verkle specific logic, update when Verkle fork is confirmed.
    if t8n_data.fork_name == "Verkle":
        args.append("--output.vkt=stdout")
        if t8n_data.vkt is not None:
            args.append("--input.vkt=stdin")

    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/evm_transition_tool/transition_tool.py

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
        """
    )

    files_to_dump = {
        "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,
    }
    # Only dump verkle if present
    if stdin.vkt:
        files_to_dump["input/vkt.json"] = stdin.vkt

    dump_files_to_directory(
        debug_output_path,
        files_to_dump,
    )

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

Executes 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/evm_transition_tool/transition_tool.py

def evaluate(
    self,
    *,
    alloc: Alloc,
    txs: List[Transaction],
    env: Environment,
    fork: Fork,
    chain_id: int = 1,
    reward: int = 0,
    vkt: Optional[VerkleTree] = None,
    eips: Optional[List[int]] = None,
    debug_output_path: str = "",
) -> TransitionToolOutput:
    """
    Executes 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.
    """
    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])
    if env.number == 0:
        reward = -1
    t8n_data = self.TransitionToolData(
        alloc=alloc,
        txs=txs,
        env=env,
        fork_name=fork_name,
        vkt=vkt,
        chain_id=chain_id,
        reward=reward,
    )

    if self.t8n_use_server:
        return self._evaluate_server(
            t8n_data=t8n_data, debug_output_path=debug_output_path
        )

    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,
    )

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

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

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

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

from_mpt_to_vkt(mpt_alloc)

Returns the verkle tree representation for an input MPT.

Source code in src/evm_transition_tool/transition_tool.py

def from_mpt_to_vkt(self, mpt_alloc: Alloc) -> VerkleTree:
    """
    Returns the verkle tree representation for an input MPT.
    """
    raise NotImplementedError(
        "The `from_mpt_to_vkt()` function is not supported by this tool. Use geth's evm tool."
    )

get_verkle_state_root(mpt_alloc)

Returns the VKT state root from an input MPT.

Source code in src/evm_transition_tool/transition_tool.py

def get_verkle_state_root(self, mpt_alloc: Alloc) -> bytes:
    """
    Returns the VKT state root from an input MPT.
    """
    raise NotImplementedError(
        "The `get_verkle_state_root()` function is not supported by this tool. Use geth's evm"
        " tool."
    )

get_verkle_single_key(address, storage_slot=None)

Returns the VKT key for an account address or storage slot.

Source code in src/evm_transition_tool/transition_tool.py

def get_verkle_single_key(
    self, address: Address, storage_slot: Optional[ZeroPaddedHexNumber] = None
) -> str:
    """
    Returns the VKT key for an account address or storage slot.
    """
    raise NotImplementedError(
        "The `get_verkle_single_key()` function is not supported by this tool. Use geth's evm"
        " tool."
    )

get_verkle_code_chunk_key(address, code_chunk)

Returns the VKT key of a code chunk for an account address.

Source code in src/evm_transition_tool/transition_tool.py

def get_verkle_code_chunk_key(
    self, address: Address, code_chunk: ZeroPaddedHexNumber
) -> str:
    """
    Returns the VKT key of a code chunk for an account address.
    """
    raise NotImplementedError(
        "The `get_verkle_code_chunk_key()` function is not supported by this tool. Use geth's"
        " evm tool."
    )

get_witness_check_mapping(witness_check)

Returns a tuple containing: A) StateDiff - A pseudo StateDiff type with stems, suffixes, and current values. B) Dict[Stem, Address] - A mapping of stems to their associated addresses.

Source code in src/evm_transition_tool/transition_tool.py

def get_witness_check_mapping(
    self, witness_check: WitnessCheck
) -> Tuple[StateDiff, Dict[Stem, Address]]:
    """
    Returns a tuple containing:
    A) StateDiff - A pseudo StateDiff type with stems, suffixes, and current values.
    B) Dict[Stem, Address] - A mapping of stems to their associated addresses.
    """
    raise NotImplementedError(
        "The `get_witness_check_mapping()` 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

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

class UnknownTransitionTool(Exception):
    """Exception raised if an unknown t8n is encountered"""

    pass

Result

Bases: CamelModel

Result of a t8n

Source code in src/evm_transition_tool/types.py

class Result(CamelModel):
    """
    Result of a t8n
    """

    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_root: Hash | None = None
    deposit_requests: List[DepositRequest] | None = None
    withdrawal_requests: List[WithdrawalRequest] | None = None
    consolidation_requests: List[ConsolidationRequest] | None = None

    # Verkle fields within the result
    verkle_conversion_address: Address | None = Field(None, alias="currentConversionAddress")
    verkle_conversion_slot_hash: Hash | None = Field(None, alias="currentConversionSlotHash")
    verkle_conversion_started: bool | None = Field(None, alias="currentConversionStarted")
    verkle_conversion_ended: bool | None = Field(None, alias="currentConversionEnded")
    verkle_conversion_storage_processed: bool | None = Field(
        None,
        alias="currentConversionStorageProcessed",
    )

    verkle_proof: VerkleProof | None = None
    state_diff: StateDiff | None = None

TransitionToolOutput

Bases: CamelModel

Transition tool output

Source code in src/evm_transition_tool/types.py

class TransitionToolOutput(CamelModel):
    """
    Transition tool output
    """

    alloc: Alloc
    result: Result
    body: Bytes | None = None
    vkt: VerkleTree | None = None
    witness: Witness | None = None