diff --git a/pyproject.toml b/pyproject.toml index a03b06ffcd..0bf639deef 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -89,7 +89,7 @@ tf = "cli.pytest_commands.fill:tf" checkfixtures = "cli.check_fixtures:check_fixtures" consume = "cli.pytest_commands.consume:consume" genindex = "cli.gen_index:generate_fixtures_index_cli" -gentest = "cli.gentest:make_test" +gentest = "cli.gentest:generate" pyspelling_soft_fail = "cli.tox_helpers:pyspelling" markdownlintcli2_soft_fail = "cli.tox_helpers:markdownlint" order_fixtures = "cli.order_fixtures:order_fixtures" diff --git a/src/cli/gentest.py b/src/cli/gentest.py deleted file mode 100644 index c1a9c03d3b..0000000000 --- a/src/cli/gentest.py +++ /dev/null @@ -1,346 +0,0 @@ -""" -Generate a Python blockchain test from a transaction hash. - -This script can be used to generate Python source for a blockchain test case -that replays a mainnet or testnet transaction from its transaction hash. - -Note: - -Requirements: - -1. Access to an archive node for the network where the transaction - originates. A provider may be used. -2. The transaction hash of a type 0 transaction (currently only legacy - transactions are supported). - -Example Usage: - -1. Generate a test for a transaction with hash - - ```console - uv run gentest \ - 0xa41f343be7a150b740e5c939fa4d89f3a2850dbe21715df96b612fc20d1906be \ - tests/paris/test_0xa41f.py - ``` - -2. Fill the test: - - ```console - fill --fork=Paris tests/paris/test_0xa41f.py - ``` - -Limitations: - -1. Only legacy transaction types (type 0) are currently supported. -""" - -from dataclasses import asdict, dataclass -from sys import stderr -from typing import Dict, TextIO - -import click - -from config import EnvConfig -from ethereum_test_base_types import Account, Address, Hash, ZeroPaddedHexNumber -from ethereum_test_rpc import BlockNumberType, DebugRPC, EthRPC -from ethereum_test_types import Transaction - - -@click.command() -@click.argument("transaction_hash") -@click.argument("output_file", type=click.File("w", lazy=True)) -def make_test(transaction_hash: str, output_file: TextIO): - """ - Extracts a transaction and required state from a network to make a blockchain test out of it. - - TRANSACTION_HASH is the hash of the transaction to be used. - - OUTPUT_FILE is the path to the output python script. - """ - request = RequestManager() - - print( - "Perform tx request: eth_get_transaction_by_hash(" + f"{transaction_hash}" + ")", - file=stderr, - ) - tr = request.eth_get_transaction_by_hash(Hash(transaction_hash)) - - print("Perform debug_trace_call", file=stderr) - state = request.debug_trace_call(tr) - - print("Perform eth_get_block_by_number", file=stderr) - block = request.eth_get_block_by_number(tr.block_number) - - print("Generate py test", file=stderr) - constructor = TestConstructor(PYTEST_TEMPLATE) - test = constructor.make_test_template(block, tr, state) - - output_file.write(test) - - print("Finished", file=stderr) - - -class TestConstructor: - """ - Construct .py file from test template, by replacing keywords with test data - """ - - test_template: str - - def __init__(self, test_template: str): - """ - Initialize with template - """ - self.test_template = test_template - - def _make_test_comments(self, test: str, tr_hash: str) -> str: - test = test.replace( - "$HEADLINE_COMMENT", - "gentest autogenerated test with debug_traceCall of tx.hash " + tr_hash, - ) - test = test.replace("$TEST_NAME", "test_transaction_" + tr_hash[2:]) - test = test.replace( - "$TEST_COMMENT", "gentest autogenerated test for tx.hash " + tr_hash[2:] - ) - return test - - def _make_test_environment(self, test: str, bl: "RequestManager.RemoteBlock") -> str: - env_str = "" - pad = " " - for field, value in asdict(bl).items(): - env_str += ( - f'{pad}{field}="{value}",\n' if field == "coinbase" else f"{pad}{field}={value},\n" - ) - test = test.replace("$ENV", env_str) - return test - - def _make_pre_state( - self, test: str, tr: "RequestManager.RemoteTransaction", state: Dict[Address, Account] - ) -> str: - # Print a nice .py storage pre - pad = " " - state_str = "" - for address, account in state.items(): - if isinstance(account, dict): - account_obj = Account(**account) - state_str += f' "{address}": Account(\n' - state_str += f"{pad}balance={str(account_obj.balance)},\n" - if address == tr.transaction.sender: - state_str += f"{pad}nonce={tr.transaction.nonce},\n" - else: - state_str += f"{pad}nonce={str(account_obj.nonce)},\n" - - if account_obj.code is None: - state_str += f'{pad}code="0x",\n' - else: - state_str += f'{pad}code="{str(account_obj.code)}",\n' - state_str += pad + "storage={\n" - - if account_obj.storage is not None: - for record, value in account_obj.storage.root.items(): - pad_record = ZeroPaddedHexNumber(record) - pad_value = ZeroPaddedHexNumber(value) - state_str += f'{pad} "{pad_record}" : "{pad_value}",\n' - - state_str += pad + "}\n" - state_str += " ),\n" - return test.replace("$PRE", state_str) - - def _make_transaction(self, test: str, tr: "RequestManager.RemoteTransaction") -> str: - """ - Print legacy transaction in .py - """ - pad = " " - tr_str = "" - quoted_fields_array = ["data", "to"] - hex_fields_array = ["v", "r", "s"] - legacy_fields_array = [ - "ty", - "chain_id", - "nonce", - "gas_price", - "protected", - "gas_limit", - "value", - ] - for field, value in iter(tr.transaction): - if value is None: - continue - - if field in legacy_fields_array: - tr_str += f"{pad}{field}={value},\n" - - if field in quoted_fields_array: - tr_str += f'{pad}{field}="{value}",\n' - - if field in hex_fields_array: - tr_str += f"{pad}{field}={hex(value)},\n" - - return test.replace("$TR", tr_str) - - def make_test_template( - self, - bl: "RequestManager.RemoteBlock", - tr: "RequestManager.RemoteTransaction", - state: Dict[Address, Account], - ) -> str: - """ - Prepare the .py file template - """ - test = self.test_template - test = self._make_test_comments(test, str(tr.tr_hash)) - test = self._make_test_environment(test, bl) - test = self._make_pre_state(test, tr, state) - test = self._make_transaction(test, tr) - return test - - -class RequestManager: - """ - Interface for the RPC interaction with remote node - """ - - @dataclass() - class RemoteTransaction: - """ - Remote transaction structure - """ - - block_number: int - tr_hash: Hash - transaction: Transaction - - @dataclass - class RemoteBlock: - """ - Remote block header information structure - """ - - coinbase: str - difficulty: str - gas_limit: str - number: str - timestamp: str - - node_url: str - headers: dict[str, str] - - def __init__(self): - """ - Initialize the RequestManager with specific client config. - """ - node_config = EnvConfig().remote_nodes[0] - self.node_url = node_config.node_url - headers = node_config.rpc_headers - self.rpc = EthRPC(node_config.node_url, extra_headers=headers) - self.debug_rpc = DebugRPC(node_config.node_url, extra_headers=headers) - - def eth_get_transaction_by_hash(self, transaction_hash: Hash) -> RemoteTransaction: - """ - Get transaction data. - """ - res = self.rpc.get_transaction_by_hash(transaction_hash) - block_number = res.block_number - assert block_number is not None, "Transaction does not seem to be included in any block" - - assert ( - res.ty == 0 - ), f"Transaction has type {res.ty}: Currently only type 0 transactions are supported." - - return RequestManager.RemoteTransaction( - block_number=block_number, - tr_hash=res.transaction_hash, - transaction=Transaction( - ty=res.ty, - gas_limit=res.gas_limit, - gas_price=res.gas_price, - data=res.data, - nonce=res.nonce, - sender=res.from_address, - to=res.to_address, - value=res.value, - v=res.v, - r=res.r, - s=res.s, - protected=True if res.v > 30 else False, - ), - ) - - def eth_get_block_by_number(self, block_number: BlockNumberType) -> RemoteBlock: - """ - Get block by number - """ - res = self.rpc.get_block_by_number(block_number) - - return RequestManager.RemoteBlock( - coinbase=res["miner"], - number=res["number"], - difficulty=res["difficulty"], - gas_limit=res["gasLimit"], - timestamp=res["timestamp"], - ) - - def debug_trace_call(self, tr: RemoteTransaction) -> Dict[Address, Account]: - """ - Get pre-state required for transaction - """ - return self.debug_rpc.trace_call( - { - "from": f"{str(tr.transaction.sender)}", - "to": f"{str(tr.transaction.to)}", - "data": f"{str(tr.transaction.data)}", - }, - f"{tr.block_number}", - ) - - -PYTEST_TEMPLATE = """ -\"\"\" -$HEADLINE_COMMENT -\"\"\" - -import pytest - -from ethereum_test_tools import ( - Account, - Address, - Block, - Environment, - BlockchainTestFiller, - Transaction, -) - -REFERENCE_SPEC_GIT_PATH = "N/A" -REFERENCE_SPEC_VERSION = "N/A" - - -@pytest.fixture -def env(): # noqa: D103 - return Environment( -$ENV - ) - - -@pytest.mark.valid_from("Paris") -def $TEST_NAME( - env: Environment, - blockchain_test: BlockchainTestFiller, -): - \"\"\" - $TEST_COMMENT - \"\"\" - - pre = { -$PRE - } - - post = { - } - - tx = Transaction( -$TR - ) - - blockchain_test(genesis_environment=env, pre=pre, post=post, blocks=[Block(txs=[tx])]) - -""" diff --git a/src/cli/gentest/__init__.py b/src/cli/gentest/__init__.py new file mode 100644 index 0000000000..40b6a4112b --- /dev/null +++ b/src/cli/gentest/__init__.py @@ -0,0 +1,35 @@ +""" +Generate a Python blockchain test from a transaction hash. + +This script can be used to generate Python source for a blockchain test case +that replays a mainnet or testnet transaction from its transaction hash. + +Requirements: + +1. Access to an archive node for the network where the transaction + originates. A provider may be used. +2. The transaction hash of a type 0 transaction (currently only legacy + transactions are supported). + +Example Usage: + +1. Generate a test for a transaction with hash + + ```console + uv run gentest \ + 0xa41f343be7a150b740e5c939fa4d89f3a2850dbe21715df96b612fc20d1906be \ + tests/paris/test_0xa41f.py + ``` + +2. Fill the test: + + ```console + fill --fork=Paris tests/paris/test_0xa41f.py + ``` + +Limitations: + +1. Only legacy transaction types (type 0) are currently supported. +""" + +from .cli import generate # noqa: 401 diff --git a/src/cli/gentest/cli.py b/src/cli/gentest/cli.py new file mode 100644 index 0000000000..14926299aa --- /dev/null +++ b/src/cli/gentest/cli.py @@ -0,0 +1,56 @@ +""" +CLI interface for generating blockchain test scripts. + +It extracts a specified transaction and its required state from a blockchain network +using the transaction hash and generates a Python test script based on that information. +""" + +from sys import stderr +from typing import TextIO + +import click +import jinja2 + +from ethereum_test_base_types import Hash + +from .request_manager import RPCRequest +from .test_providers import BlockchainTestProvider + +template_loader = jinja2.PackageLoader("cli.gentest") +template_env = jinja2.Environment(loader=template_loader, keep_trailing_newline=True) + + +@click.command() +@click.argument("transaction_hash") +@click.argument("output_file", type=click.File("w", lazy=True)) +def generate(transaction_hash: str, output_file: TextIO): + """ + Extracts a transaction and required state from a network to make a blockchain test out of it. + + TRANSACTION_HASH is the hash of the transaction to be used. + + OUTPUT_FILE is the path to the output python script. + """ + request = RPCRequest() + + print( + "Perform tx request: eth_get_transaction_by_hash(" + f"{transaction_hash}" + ")", + file=stderr, + ) + transaction = request.eth_get_transaction_by_hash(Hash(transaction_hash)) + + print("Perform debug_trace_call", file=stderr) + state = request.debug_trace_call(transaction) + + print("Perform eth_get_block_by_number", file=stderr) + block = request.eth_get_block_by_number(transaction.block_number) + + print("Generate py test", file=stderr) + context = BlockchainTestProvider( + block=block, transaction=transaction, state=state + ).get_context() + + template = template_env.get_template("blockchain_test/transaction.py.j2") + output_file.write(template.render(context)) + + print("Finished", file=stderr) diff --git a/src/cli/gentest/request_manager.py b/src/cli/gentest/request_manager.py new file mode 100644 index 0000000000..28774e62bc --- /dev/null +++ b/src/cli/gentest/request_manager.py @@ -0,0 +1,117 @@ +""" +A request manager Ethereum RPC calls. + +The RequestManager handles transactions and block data retrieval from a remote Ethereum node, +utilizing Pydantic models to define the structure of transactions and blocks. + +Classes: +- RequestManager: The main class for managing RPC requests and responses. +- RemoteTransaction: A Pydantic model representing a transaction retrieved from the node. +- RemoteBlock: A Pydantic model representing a block retrieved from the node. +""" + +from typing import Dict + +from pydantic import BaseModel + +from config import EnvConfig +from ethereum_test_base_types import Account, Address, Hash, HexNumber +from ethereum_test_rpc import BlockNumberType, DebugRPC, EthRPC +from ethereum_test_types import Transaction + + +class RPCRequest: + """ + Interface for the RPC interaction with remote node + """ + + class RemoteTransaction(Transaction): + """ + Model that represents a transaction. + """ + + block_number: HexNumber + tx_hash: Hash + + class RemoteBlock(BaseModel): + """ + Model that represents a block. + """ + + coinbase: str + difficulty: str + gas_limit: str + number: str + timestamp: str + + node_url: str + headers: dict[str, str] + + def __init__(self): + """ + Initialize the RequestManager with specific client config. + """ + node_config = EnvConfig().remote_nodes[0] + self.node_url = node_config.node_url + headers = node_config.rpc_headers + self.rpc = EthRPC(node_config.node_url, extra_headers=headers) + self.debug_rpc = DebugRPC(node_config.node_url, extra_headers=headers) + + def eth_get_transaction_by_hash(self, transaction_hash: Hash) -> RemoteTransaction: + """ + Get transaction data. + """ + res = self.rpc.get_transaction_by_hash(transaction_hash) + block_number = res.block_number + assert block_number is not None, "Transaction does not seem to be included in any block" + + assert ( + res.ty == 0 + ), f"Transaction has type {res.ty}: Currently only type 0 transactions are supported." + + return RPCRequest.RemoteTransaction( + block_number=block_number, + tx_hash=res.transaction_hash, + ty=res.ty, + gas_limit=res.gas_limit, + gas_price=res.gas_price, + data=res.data, + nonce=res.nonce, + sender=res.from_address, + to=res.to_address, + value=res.value, + v=res.v, + r=res.r, + s=res.s, + protected=True if res.v > 30 else False, + ) + + def eth_get_block_by_number(self, block_number: BlockNumberType) -> RemoteBlock: + """ + Get block by number + """ + res = self.rpc.get_block_by_number(block_number) + + return RPCRequest.RemoteBlock( + coinbase=res["miner"], + number=res["number"], + difficulty=res["difficulty"], + gas_limit=res["gasLimit"], + timestamp=res["timestamp"], + ) + + def debug_trace_call(self, transaction: RemoteTransaction) -> Dict[Address, Account]: + """ + Get pre-state required for transaction + """ + assert transaction.sender is not None + assert transaction.to is not None + + return self.debug_rpc.trace_call( + { + "from": transaction.sender.hex(), + "to": transaction.to.hex(), + "data": transaction.data.hex(), + }, + f"{transaction.block_number}", + ) diff --git a/src/cli/gentest/templates/blockchain_test/transaction.py.j2 b/src/cli/gentest/templates/blockchain_test/transaction.py.j2 new file mode 100644 index 0000000000..93836949a7 --- /dev/null +++ b/src/cli/gentest/templates/blockchain_test/transaction.py.j2 @@ -0,0 +1,44 @@ +""" +gentest autogenerated test with debug_traceCall of tx.hash +{{ tx_hash }} +https://etherscan.io/tx/{{tx_hash}} +""" + +from typing import Dict + +import pytest + +from ethereum_test_tools import Account, Block, BlockchainTestFiller, Environment, Transaction + +REFERENCE_SPEC_GIT_PATH = "N/A" +REFERENCE_SPEC_VERSION = "N/A" + + +@pytest.fixture +def env(): # noqa: D103 + return Environment( +{{ environment_kwargs }} + ) + + +@pytest.mark.valid_from("Paris") +def test_transaction_{{ tx_hash }}( # noqa: SC200, E501 + env: Environment, + blockchain_test: BlockchainTestFiller, +): + """ + gentest autogenerated test for tx.hash + {{ tx_hash }} + """ + pre = { +{{ pre_state_items }} + } + + post: Dict = { + } + + tx = Transaction( +{{ transaction_items }} + ) + + blockchain_test(genesis_environment=env, pre=pre, post=post, blocks=[Block(txs=[tx])]) \ No newline at end of file diff --git a/src/cli/gentest/test_providers.py b/src/cli/gentest/test_providers.py new file mode 100644 index 0000000000..0779632e45 --- /dev/null +++ b/src/cli/gentest/test_providers.py @@ -0,0 +1,117 @@ +""" +This module contains various providers which generates context required to create test scripts. + +Classes: +- BlockchainTestProvider: The BlockchainTestProvider class takes information about a block, +a transaction, and the associated state, and provides methods to generate various elements +needed for testing, such as module docstrings, test names, and pre-state items. + +Example: + provider = BlockchainTestProvider(block=block, transaction=transaction, state=state) + context = provider.get_context() +""" + +from typing import Any, Dict + +from pydantic import BaseModel + +from ethereum_test_base_types import Account, Address, ZeroPaddedHexNumber + +from .request_manager import RPCRequest + + +class BlockchainTestProvider(BaseModel): + """ + Provides context required to generate a `blockchain_test` using pytest. + """ + + block: RPCRequest.RemoteBlock + transaction: RPCRequest.RemoteTransaction + state: Dict[Address, Account] + + def _get_environment_kwargs(self) -> str: + env_str = "" + pad = " " + for field, value in self.block.dict().items(): + env_str += ( + f'{pad}{field}="{value}",\n' if field == "coinbase" else f"{pad}{field}={value},\n" + ) + + return env_str + + # TODO: Output should be dict. Formatting should happen in the template. + def _get_pre_state_items(self) -> str: + # Print a nice .py storage pre + pad = " " + state_str = "" + for address, account_obj in self.state.items(): + state_str += f' "{address}": Account(\n' + state_str += f"{pad}balance={str(account_obj.balance)},\n" + if address == self.transaction.sender: + state_str += f"{pad}nonce={self.transaction.nonce},\n" + else: + state_str += f"{pad}nonce={str(account_obj.nonce)},\n" + + if account_obj.code is None: + state_str += f'{pad}code="0x",\n' + else: + state_str += f'{pad}code="{str(account_obj.code)}",\n' + state_str += pad + "storage={\n" + + if account_obj.storage is not None: + for record, value in account_obj.storage.root.items(): + pad_record = ZeroPaddedHexNumber(record) + pad_value = ZeroPaddedHexNumber(value) + state_str += f'{pad} "{pad_record}" : "{pad_value}",\n' + + state_str += pad + "}\n" + state_str += " ),\n" + return state_str + + # TODO: Output should be dict. Formatting should happen in the template. + def _get_transaction_items(self) -> str: + """ + Print legacy transaction in .py + """ + pad = " " + tr_str = "" + quoted_fields_array = ["data", "to"] + hex_fields_array = ["v", "r", "s"] + legacy_fields_array = [ + "ty", + "chain_id", + "nonce", + "gas_price", + "protected", + "gas_limit", + "value", + ] + for field, value in iter(self.transaction): + if value is None: + continue + + if field in legacy_fields_array: + tr_str += f"{pad}{field}={value},\n" + + if field in quoted_fields_array: + tr_str += f'{pad}{field}="{value}",\n' + + if field in hex_fields_array: + tr_str += f"{pad}{field}={hex(value)},\n" + + return tr_str + + def get_context(self) -> Dict[str, Any]: + """ + Get the context for generating a blockchain test. + + Returns: + Dict[str, Any]: A dictionary containing module docstring, test name, + test docstring, environment kwargs, pre-state items, and transaction items. + """ + return { + "environment_kwargs": self._get_environment_kwargs(), + "pre_state_items": self._get_pre_state_items(), + "transaction_items": self._get_transaction_items(), + "tx_hash": self.transaction.tx_hash, + }