Skip to content

Ethereum Test Fixtures package

Ethereum test fixture format definitions.

BaseFixture

Bases: CamelModel

Represents a base Ethereum test fixture of any type.

Source code in src/ethereum_test_fixtures/base.py 
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
class BaseFixture(CamelModel):
    """Represents a base Ethereum test fixture of any type."""

    # Base Fixture class properties
    formats: ClassVar[Dict[str, Type["BaseFixture"]]] = {}
    formats_type_adapter: ClassVar[TypeAdapter]

    info: Dict[str, Dict[str, Any] | str] = Field(default_factory=dict, alias="_info")

    # Fixture format properties
    format_name: ClassVar[str] = ""
    output_file_extension: ClassVar[str] = ".json"
    description: ClassVar[str] = "Unknown fixture format; it has not been set."

    @classmethod
    def output_base_dir_name(cls) -> str:
        """Return name of the subdirectory where this type of fixture should be dumped to."""
        return cls.format_name.replace("test", "tests")

    @classmethod
    def __pydantic_init_subclass__(cls, **kwargs):
        """
        Register all subclasses of BaseFixture with a fixture format name set
        as possible fixture formats.
        """
        if cls.format_name:
            # Register the new fixture format
            BaseFixture.formats[cls.format_name] = cls
            if len(BaseFixture.formats) > 1:
                BaseFixture.formats_type_adapter = TypeAdapter(
                    Annotated[
                        Union[
                            tuple(
                                Annotated[fixture_format, Tag(format_name)]
                                for (
                                    format_name,
                                    fixture_format,
                                ) in BaseFixture.formats.items()
                            )
                        ],
                        Discriminator(fixture_format_discriminator),
                    ]
                )
            else:
                BaseFixture.formats_type_adapter = TypeAdapter(cls)

    @model_validator(mode="wrap")
    @classmethod
    def _parse_into_subclass(cls, v: Any, handler: ValidatorFunctionWrapHandler) -> "BaseFixture":
        """Parse the fixture into the correct subclass."""
        if cls is BaseFixture:
            return BaseFixture.formats_type_adapter.validate_python(v)
        return handler(v)

    @cached_property
    def json_dict(self) -> Dict[str, Any]:
        """Returns the JSON representation of the fixture."""
        return self.model_dump(mode="json", by_alias=True, exclude_none=True, exclude={"info"})

    @cached_property
    def hash(self) -> str:
        """Returns the hash of the fixture."""
        json_str = json.dumps(self.json_dict, sort_keys=True, separators=(",", ":"))
        h = hashlib.sha256(json_str.encode("utf-8")).hexdigest()
        return f"0x{h}"

    def json_dict_with_info(self, hash_only: bool = False) -> Dict[str, Any]:
        """Return JSON representation of the fixture with the info field."""
        dict_with_info = self.json_dict.copy()
        dict_with_info["_info"] = {"hash": self.hash}
        if not hash_only:
            dict_with_info["_info"].update(self.info)
        return dict_with_info

    def fill_info(
        self,
        t8n_version: str,
        test_case_description: str,
        fixture_source_url: str,
        ref_spec: ReferenceSpec | None,
        _info_metadata: Dict[str, Any],
    ):
        """Fill the info field for this fixture."""
        if "comment" not in self.info:
            self.info["comment"] = "`execution-spec-tests` generated test"
        self.info["filling-transition-tool"] = t8n_version
        self.info["description"] = test_case_description
        self.info["url"] = fixture_source_url
        self.info["fixture-format"] = self.format_name
        if ref_spec is not None:
            ref_spec.write_info(self.info)
        if _info_metadata:
            self.info.update(_info_metadata)

    def get_fork(self) -> Fork | None:
        """Return fork of the fixture as a string."""
        raise NotImplementedError

    @classmethod
    def supports_fork(cls, fork: Fork) -> bool:
        """
        Return whether the fixture can be generated for the given fork.

        By default, all fixtures support all forks.
        """
        return True

output_base_dir_name() classmethod

Return name of the subdirectory where this type of fixture should be dumped to.

Source code in src/ethereum_test_fixtures/base.py 
55
56
57
58
@classmethod
def output_base_dir_name(cls) -> str:
    """Return name of the subdirectory where this type of fixture should be dumped to."""
    return cls.format_name.replace("test", "tests")

__pydantic_init_subclass__(**kwargs) classmethod

Register all subclasses of BaseFixture with a fixture format name set as possible fixture formats.

Source code in src/ethereum_test_fixtures/base.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
@classmethod
def __pydantic_init_subclass__(cls, **kwargs):
    """
    Register all subclasses of BaseFixture with a fixture format name set
    as possible fixture formats.
    """
    if cls.format_name:
        # Register the new fixture format
        BaseFixture.formats[cls.format_name] = cls
        if len(BaseFixture.formats) > 1:
            BaseFixture.formats_type_adapter = TypeAdapter(
                Annotated[
                    Union[
                        tuple(
                            Annotated[fixture_format, Tag(format_name)]
                            for (
                                format_name,
                                fixture_format,
                            ) in BaseFixture.formats.items()
                        )
                    ],
                    Discriminator(fixture_format_discriminator),
                ]
            )
        else:
            BaseFixture.formats_type_adapter = TypeAdapter(cls)

json_dict: Dict[str, Any] cached property

Returns the JSON representation of the fixture.

hash: str cached property

Returns the hash of the fixture.

json_dict_with_info(hash_only=False)

Return JSON representation of the fixture with the info field.

Source code in src/ethereum_test_fixtures/base.py 
107
108
109
110
111
112
113
def json_dict_with_info(self, hash_only: bool = False) -> Dict[str, Any]:
    """Return JSON representation of the fixture with the info field."""
    dict_with_info = self.json_dict.copy()
    dict_with_info["_info"] = {"hash": self.hash}
    if not hash_only:
        dict_with_info["_info"].update(self.info)
    return dict_with_info

fill_info(t8n_version, test_case_description, fixture_source_url, ref_spec, _info_metadata)

Fill the info field for this fixture.

Source code in src/ethereum_test_fixtures/base.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
def fill_info(
    self,
    t8n_version: str,
    test_case_description: str,
    fixture_source_url: str,
    ref_spec: ReferenceSpec | None,
    _info_metadata: Dict[str, Any],
):
    """Fill the info field for this fixture."""
    if "comment" not in self.info:
        self.info["comment"] = "`execution-spec-tests` generated test"
    self.info["filling-transition-tool"] = t8n_version
    self.info["description"] = test_case_description
    self.info["url"] = fixture_source_url
    self.info["fixture-format"] = self.format_name
    if ref_spec is not None:
        ref_spec.write_info(self.info)
    if _info_metadata:
        self.info.update(_info_metadata)

get_fork()

Return fork of the fixture as a string.

Source code in src/ethereum_test_fixtures/base.py 
135
136
137
def get_fork(self) -> Fork | None:
    """Return fork of the fixture as a string."""
    raise NotImplementedError

supports_fork(fork) classmethod

Return whether the fixture can be generated for the given fork.

By default, all fixtures support all forks.

Source code in src/ethereum_test_fixtures/base.py 
139
140
141
142
143
144
145
146
@classmethod
def supports_fork(cls, fork: Fork) -> bool:
    """
    Return whether the fixture can be generated for the given fork.

    By default, all fixtures support all forks.
    """
    return True

LabeledFixtureFormat

Represents a fixture format with a custom label.

This label will be used in the test id and also will be added as a marker to the generated test case when filling the test.

Source code in src/ethereum_test_fixtures/base.py 
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
class LabeledFixtureFormat:
    """
    Represents a fixture format with a custom label.

    This label will be used in the test id and also will be added as a marker to the
    generated test case when filling the test.
    """

    format: Type[BaseFixture]
    label: str
    description: str

    registered_labels: ClassVar[Dict[str, "LabeledFixtureFormat"]] = {}

    def __init__(
        self,
        fixture_format: "Type[BaseFixture] | LabeledFixtureFormat",
        label: str,
        description: str,
    ):
        """Initialize the fixture format with a custom label."""
        self.format = (
            fixture_format.format
            if isinstance(fixture_format, LabeledFixtureFormat)
            else fixture_format
        )
        self.label = label
        self.description = description
        if label not in LabeledFixtureFormat.registered_labels:
            LabeledFixtureFormat.registered_labels[label] = self

    @property
    def format_name(self) -> str:
        """Get the execute format name."""
        return self.format.format_name

    def __eq__(self, other: Any) -> bool:
        """
        Check if two labeled fixture formats are equal.

        If the other object is a FixtureFormat type, the format of the labeled fixture
        format will be compared with the format of the other object.
        """
        if isinstance(other, LabeledFixtureFormat):
            return self.format == other.format
        if isinstance(other, type) and issubclass(other, BaseFixture):
            return self.format == other
        return False

__init__(fixture_format, label, description)

Initialize the fixture format with a custom label.

Source code in src/ethereum_test_fixtures/base.py 
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
def __init__(
    self,
    fixture_format: "Type[BaseFixture] | LabeledFixtureFormat",
    label: str,
    description: str,
):
    """Initialize the fixture format with a custom label."""
    self.format = (
        fixture_format.format
        if isinstance(fixture_format, LabeledFixtureFormat)
        else fixture_format
    )
    self.label = label
    self.description = description
    if label not in LabeledFixtureFormat.registered_labels:
        LabeledFixtureFormat.registered_labels[label] = self

format_name: str property

Get the execute format name.

__eq__(other)

Check if two labeled fixture formats are equal.

If the other object is a FixtureFormat type, the format of the labeled fixture format will be compared with the format of the other object.

Source code in src/ethereum_test_fixtures/base.py 
185
186
187
188
189
190
191
192
193
194
195
196
def __eq__(self, other: Any) -> bool:
    """
    Check if two labeled fixture formats are equal.

    If the other object is a FixtureFormat type, the format of the labeled fixture
    format will be compared with the format of the other object.
    """
    if isinstance(other, LabeledFixtureFormat):
        return self.format == other.format
    if isinstance(other, type) and issubclass(other, BaseFixture):
        return self.format == other
    return False

BlockchainEngineFixture

Bases: BlockchainEngineFixtureCommon

Engine specific test fixture information.

Source code in src/ethereum_test_fixtures/blockchain.py 
548
549
550
551
552
553
554
555
556
557
558
559
class BlockchainEngineFixture(BlockchainEngineFixtureCommon):
    """Engine specific test fixture information."""

    format_name: ClassVar[str] = "blockchain_test_engine"
    description: ClassVar[str] = (
        "Tests that generate a blockchain test fixture in Engine API format."
    )
    pre: Alloc
    genesis: FixtureHeader = Field(..., alias="genesisBlockHeader")
    post_state: Alloc | None = Field(None)
    payloads: List[FixtureEngineNewPayload] = Field(..., alias="engineNewPayloads")
    sync_payload: FixtureEngineNewPayload | None = None

BlockchainEngineFixtureCommon

Bases: BaseFixture

Base blockchain test fixture model for Engine API based execution.

Similar to BlockchainFixtureCommon but excludes the 'pre' field to avoid duplicating large pre-allocations when using shared genesis approaches.

Source code in src/ethereum_test_fixtures/blockchain.py 
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
@post_state_validator()
class BlockchainEngineFixtureCommon(BaseFixture):
    """
    Base blockchain test fixture model for Engine API based execution.

    Similar to BlockchainFixtureCommon but excludes the 'pre' field to avoid
    duplicating large pre-allocations when using shared genesis approaches.
    """

    fork: Fork = Field(..., alias="network")
    post_state_hash: Hash | None = Field(None)
    last_block_hash: Hash = Field(..., alias="lastblockhash")  # FIXME: lastBlockHash
    config: FixtureConfig

    def get_fork(self) -> Fork | None:
        """Return fixture's `Fork`."""
        return self.fork

    @classmethod
    def supports_fork(cls, fork: Fork) -> bool:
        """
        Return whether the fixture can be generated for the given fork.

        The Engine API is available only on Paris and afterwards.
        """
        return fork >= Paris

get_fork()

Return fixture's Fork.

Source code in src/ethereum_test_fixtures/blockchain.py 
534
535
536
def get_fork(self) -> Fork | None:
    """Return fixture's `Fork`."""
    return self.fork

supports_fork(fork) classmethod

Return whether the fixture can be generated for the given fork.

The Engine API is available only on Paris and afterwards.

Source code in src/ethereum_test_fixtures/blockchain.py 
538
539
540
541
542
543
544
545
@classmethod
def supports_fork(cls, fork: Fork) -> bool:
    """
    Return whether the fixture can be generated for the given fork.

    The Engine API is available only on Paris and afterwards.
    """
    return fork >= Paris

BlockchainEngineReorgFixture

Bases: BlockchainEngineFixtureCommon

Engine reorg specific test fixture information.

Uses shared pre-allocations and blockchain reorganization for efficient test execution without client restarts.

Source code in src/ethereum_test_fixtures/blockchain.py 
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
@post_state_validator(alternate_field="post_state_diff")
class BlockchainEngineReorgFixture(BlockchainEngineFixtureCommon):
    """
    Engine reorg specific test fixture information.

    Uses shared pre-allocations and blockchain reorganization for efficient
    test execution without client restarts.
    """

    format_name: ClassVar[str] = "blockchain_test_engine_reorg"
    description: ClassVar[str] = (
        "Tests that generate a blockchain test fixture for use with a shared pre-state and engine "
        "reorg execution."
    )

    pre_hash: str
    """Hash of the shared pre-allocation group this test belongs to."""

    post_state_diff: Alloc | None = None
    """State difference from genesis after test execution (efficiency optimization)."""

    payloads: List[FixtureEngineNewPayload] = Field(..., alias="engineNewPayloads")
    """Engine API payloads for blockchain execution."""

    sync_payload: FixtureEngineNewPayload | None = None
    """Optional sync payload for blockchain synchronization."""

pre_hash: str instance-attribute

Hash of the shared pre-allocation group this test belongs to.

post_state_diff: Alloc | None = None class-attribute instance-attribute

State difference from genesis after test execution (efficiency optimization).

payloads: List[FixtureEngineNewPayload] = Field(..., alias='engineNewPayloads') class-attribute instance-attribute

Engine API payloads for blockchain execution.

sync_payload: FixtureEngineNewPayload | None = None class-attribute instance-attribute

Optional sync payload for blockchain synchronization.

BlockchainFixture

Bases: BlockchainFixtureCommon

Cross-client specific blockchain test model use in JSON fixtures.

Source code in src/ethereum_test_fixtures/blockchain.py 
509
510
511
512
513
514
515
516
517
class BlockchainFixture(BlockchainFixtureCommon):
    """Cross-client specific blockchain test model use in JSON fixtures."""

    format_name: ClassVar[str] = "blockchain_test"
    description: ClassVar[str] = "Tests that generate a blockchain test fixture."

    genesis_rlp: Bytes = Field(..., alias="genesisRLP")
    blocks: List[FixtureBlock | InvalidFixtureBlock]
    seal_engine: Literal["NoProof"] = Field("NoProof")

BlockchainFixtureCommon

Bases: BaseFixture

Base blockchain test fixture model.

Source code in src/ethereum_test_fixtures/blockchain.py 
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
@post_state_validator()
class BlockchainFixtureCommon(BaseFixture):
    """Base blockchain test fixture model."""

    fork: Fork = Field(..., alias="network")
    genesis: FixtureHeader = Field(..., alias="genesisBlockHeader")
    pre: Alloc
    post_state: Alloc | None = Field(None)
    post_state_hash: Hash | None = Field(None)
    last_block_hash: Hash = Field(..., alias="lastblockhash")  # FIXME: lastBlockHash
    config: FixtureConfig

    @model_validator(mode="before")
    @classmethod
    def config_defaults_for_backwards_compatibility(cls, data: Any) -> Any:
        """
        Check if the config field is populated, otherwise use the root-level field values for
        backwards compatibility.
        """
        if isinstance(data, dict):
            if "config" not in data:
                data["config"] = {}
            if isinstance(data["config"], dict):
                if "network" not in data["config"]:
                    data["config"]["network"] = data["network"]
                if "chainid" not in data["config"]:
                    data["config"]["chainid"] = "0x01"
        return data

    def get_fork(self) -> Fork | None:
        """Return fork of the fixture as a string."""
        return self.fork

config_defaults_for_backwards_compatibility(data) classmethod

Check if the config field is populated, otherwise use the root-level field values for backwards compatibility.

Source code in src/ethereum_test_fixtures/blockchain.py 
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
@model_validator(mode="before")
@classmethod
def config_defaults_for_backwards_compatibility(cls, data: Any) -> Any:
    """
    Check if the config field is populated, otherwise use the root-level field values for
    backwards compatibility.
    """
    if isinstance(data, dict):
        if "config" not in data:
            data["config"] = {}
        if isinstance(data["config"], dict):
            if "network" not in data["config"]:
                data["config"]["network"] = data["network"]
            if "chainid" not in data["config"]:
                data["config"]["chainid"] = "0x01"
    return data

get_fork()

Return fork of the fixture as a string.

Source code in src/ethereum_test_fixtures/blockchain.py 
504
505
506
def get_fork(self) -> Fork | None:
    """Return fork of the fixture as a string."""
    return self.fork

FixtureCollector dataclass

Collects all fixtures generated by the test cases.

Source code in src/ethereum_test_fixtures/collector.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
@dataclass(kw_only=True)
class FixtureCollector:
    """Collects all fixtures generated by the test cases."""

    output_dir: Path
    flat_output: bool
    fill_static_tests: bool
    single_fixture_per_file: bool
    filler_path: Path
    base_dump_dir: Optional[Path] = None

    # Internal state
    all_fixtures: Dict[Path, Fixtures] = field(default_factory=dict)
    json_path_to_test_item: Dict[Path, TestInfo] = field(default_factory=dict)

    def get_fixture_basename(self, info: TestInfo) -> Path:
        """Return basename of the fixture file for a given test case."""
        if self.flat_output:
            if self.single_fixture_per_file:
                return Path(info.get_single_test_name(mode="test"))
            return Path(info.get_single_test_name(mode="module"))
        else:
            module_relative_output_dir = info.get_module_relative_output_dir(self.filler_path)

            # Each legacy test filler has only 1 test per file if it's a !state test!
            # So no need to create directory Add11/add11.json it can be plain add11.json
            if self.fill_static_tests:
                return module_relative_output_dir.parent / info.original_name

            if self.single_fixture_per_file:
                return module_relative_output_dir / info.get_single_test_name(mode="test")
            return module_relative_output_dir / info.get_single_test_name(mode="module")

    def add_fixture(self, info: TestInfo, fixture: BaseFixture) -> Path:
        """Add fixture to the list of fixtures of a given test case."""
        fixture_basename = self.get_fixture_basename(info)

        fixture_path = (
            self.output_dir
            / fixture.output_base_dir_name()
            / fixture_basename.with_suffix(fixture.output_file_extension)
        )
        if fixture_path not in self.all_fixtures.keys():  # relevant when we group by test function
            self.all_fixtures[fixture_path] = Fixtures(root={})
            self.json_path_to_test_item[fixture_path] = info

        self.all_fixtures[fixture_path][info.get_id()] = fixture

        return fixture_path

    def dump_fixtures(self) -> None:
        """Dump all collected fixtures to their respective files."""
        if self.output_dir.name == "stdout":
            combined_fixtures = {
                k: to_json(v) for fixture in self.all_fixtures.values() for k, v in fixture.items()
            }
            json.dump(combined_fixtures, sys.stdout, indent=4)
            return
        os.makedirs(self.output_dir, exist_ok=True)
        for fixture_path, fixtures in self.all_fixtures.items():
            os.makedirs(fixture_path.parent, exist_ok=True)
            if len({fixture.__class__ for fixture in fixtures.values()}) != 1:
                raise TypeError("All fixtures in a single file must have the same format.")
            fixtures.collect_into_file(fixture_path)

    def verify_fixture_files(self, evm_fixture_verification: FixtureConsumer) -> None:
        """Run `evm [state|block]test` on each fixture."""
        for fixture_path, name_fixture_dict in self.all_fixtures.items():
            for _fixture_name, fixture in name_fixture_dict.items():
                if evm_fixture_verification.can_consume(fixture.__class__):
                    info = self.json_path_to_test_item[fixture_path]
                    consume_direct_dump_dir = self._get_consume_direct_dump_dir(info)
                    evm_fixture_verification.consume_fixture(
                        fixture.__class__,
                        fixture_path,
                        fixture_name=None,
                        debug_output_path=consume_direct_dump_dir,
                    )

    def _get_consume_direct_dump_dir(
        self,
        info: TestInfo,
    ):
        """
        Directory to dump the current test function's fixture.json and fixture
        verification debug output.
        """
        if not self.base_dump_dir:
            return None
        if self.single_fixture_per_file:
            return info.get_dump_dir_path(
                self.base_dump_dir, self.filler_path, level="test_parameter"
            )
        else:
            return info.get_dump_dir_path(
                self.base_dump_dir, self.filler_path, level="test_function"
            )

get_fixture_basename(info)

Return basename of the fixture file for a given test case.

Source code in src/ethereum_test_fixtures/collector.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
def get_fixture_basename(self, info: TestInfo) -> Path:
    """Return basename of the fixture file for a given test case."""
    if self.flat_output:
        if self.single_fixture_per_file:
            return Path(info.get_single_test_name(mode="test"))
        return Path(info.get_single_test_name(mode="module"))
    else:
        module_relative_output_dir = info.get_module_relative_output_dir(self.filler_path)

        # Each legacy test filler has only 1 test per file if it's a !state test!
        # So no need to create directory Add11/add11.json it can be plain add11.json
        if self.fill_static_tests:
            return module_relative_output_dir.parent / info.original_name

        if self.single_fixture_per_file:
            return module_relative_output_dir / info.get_single_test_name(mode="test")
        return module_relative_output_dir / info.get_single_test_name(mode="module")

add_fixture(info, fixture)

Add fixture to the list of fixtures of a given test case.

Source code in src/ethereum_test_fixtures/collector.py 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
def add_fixture(self, info: TestInfo, fixture: BaseFixture) -> Path:
    """Add fixture to the list of fixtures of a given test case."""
    fixture_basename = self.get_fixture_basename(info)

    fixture_path = (
        self.output_dir
        / fixture.output_base_dir_name()
        / fixture_basename.with_suffix(fixture.output_file_extension)
    )
    if fixture_path not in self.all_fixtures.keys():  # relevant when we group by test function
        self.all_fixtures[fixture_path] = Fixtures(root={})
        self.json_path_to_test_item[fixture_path] = info

    self.all_fixtures[fixture_path][info.get_id()] = fixture

    return fixture_path

dump_fixtures()

Dump all collected fixtures to their respective files.

Source code in src/ethereum_test_fixtures/collector.py 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
def dump_fixtures(self) -> None:
    """Dump all collected fixtures to their respective files."""
    if self.output_dir.name == "stdout":
        combined_fixtures = {
            k: to_json(v) for fixture in self.all_fixtures.values() for k, v in fixture.items()
        }
        json.dump(combined_fixtures, sys.stdout, indent=4)
        return
    os.makedirs(self.output_dir, exist_ok=True)
    for fixture_path, fixtures in self.all_fixtures.items():
        os.makedirs(fixture_path.parent, exist_ok=True)
        if len({fixture.__class__ for fixture in fixtures.values()}) != 1:
            raise TypeError("All fixtures in a single file must have the same format.")
        fixtures.collect_into_file(fixture_path)

verify_fixture_files(evm_fixture_verification)

Run evm [state|block]test on each fixture.

Source code in src/ethereum_test_fixtures/collector.py 
171
172
173
174
175
176
177
178
179
180
181
182
183
def verify_fixture_files(self, evm_fixture_verification: FixtureConsumer) -> None:
    """Run `evm [state|block]test` on each fixture."""
    for fixture_path, name_fixture_dict in self.all_fixtures.items():
        for _fixture_name, fixture in name_fixture_dict.items():
            if evm_fixture_verification.can_consume(fixture.__class__):
                info = self.json_path_to_test_item[fixture_path]
                consume_direct_dump_dir = self._get_consume_direct_dump_dir(info)
                evm_fixture_verification.consume_fixture(
                    fixture.__class__,
                    fixture_path,
                    fixture_name=None,
                    debug_output_path=consume_direct_dump_dir,
                )

TestInfo dataclass

Contains test information from the current node.

Source code in src/ethereum_test_fixtures/collector.py 
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
@dataclass(kw_only=True)
class TestInfo:
    """Contains test information from the current node."""

    name: str  # pytest: Item.name, e.g. test_paris_one[fork_Paris-state_test]
    id: str  # pytest: Item.nodeid, e.g. tests/paris/test_module_paris.py::test_paris_one[...]
    original_name: str  # pytest: Item.originalname, e.g. test_paris_one
    module_path: Path  # pytest: Item.path, e.g. .../tests/paris/test_module_paris.py

    test_prefix: ClassVar[str] = "test_"  # Python test prefix
    filler_suffix: ClassVar[str] = "Filler"  # Static test suffix

    @classmethod
    def strip_test_name(cls, name: str) -> str:
        """Remove test prefix from a python test case name."""
        if name.startswith(cls.test_prefix):
            return name[len(cls.test_prefix) :]
        if name.endswith(cls.filler_suffix):
            return name[: -len(cls.filler_suffix)]
        return name

    def get_name_and_parameters(self) -> Tuple[str, str]:
        """
        Convert test name to a tuple containing the test name and test parameters.

        Example:
        test_push0_key_sstore[fork_Shanghai] -> test_push0_key_sstore, fork_Shanghai

        """
        test_name, parameters = self.name.split("[")
        return test_name, re.sub(r"[\[\-]", "_", parameters).replace("]", "")

    def get_single_test_name(self, mode: Literal["module", "test"] = "module") -> str:
        """Convert test name to a single test name."""
        if mode == "module":
            # Use the module name as the test name
            return self.strip_test_name(self.original_name)
        elif mode == "test":
            # Mix the module name and the test name/arguments
            test_name, test_parameters = self.get_name_and_parameters()
            test_name = self.strip_test_name(test_name)
            return f"{test_name}__{test_parameters}"

    def get_dump_dir_path(
        self,
        base_dump_dir: Optional[Path],
        filler_path: Path,
        level: Literal["test_module", "test_function", "test_parameter"] = "test_parameter",
    ) -> Optional[Path]:
        """Path to dump the debug output as defined by the level to dump at."""
        if not base_dump_dir:
            return None
        test_module_relative_dir = self.get_module_relative_output_dir(filler_path)
        if level == "test_module":
            return Path(base_dump_dir) / Path(str(test_module_relative_dir).replace(os.sep, "__"))
        test_name, test_parameter_string = self.get_name_and_parameters()
        flat_path = f"{str(test_module_relative_dir).replace(os.sep, '__')}__{test_name}"
        if level == "test_function":
            return Path(base_dump_dir) / flat_path
        elif level == "test_parameter":
            return Path(base_dump_dir) / flat_path / test_parameter_string
        raise Exception("Unexpected level.")

    def get_id(self) -> str:
        """Return the test id."""
        return self.id

    def get_module_relative_output_dir(self, filler_path: Path) -> Path:
        """
        Return a directory name for the provided test_module (relative to the
        base ./tests directory) that can be used for output (within the
        configured fixtures output path or the base_dump_dir directory).

        Example:
        tests/shanghai/eip3855_push0/test_push0.py -> shanghai/eip3855_push0/test_push0

        """
        basename = self.module_path.with_suffix("").absolute()
        basename_relative = basename.relative_to(
            os.path.commonpath([filler_path.absolute(), basename])
        )
        module_path = basename_relative.parent / self.strip_test_name(basename_relative.stem)
        return module_path

strip_test_name(name) classmethod

Remove test prefix from a python test case name.

Source code in src/ethereum_test_fixtures/collector.py 
33
34
35
36
37
38
39
40
@classmethod
def strip_test_name(cls, name: str) -> str:
    """Remove test prefix from a python test case name."""
    if name.startswith(cls.test_prefix):
        return name[len(cls.test_prefix) :]
    if name.endswith(cls.filler_suffix):
        return name[: -len(cls.filler_suffix)]
    return name

get_name_and_parameters()

Convert test name to a tuple containing the test name and test parameters.

Example: test_push0_key_sstore[fork_Shanghai] -> test_push0_key_sstore, fork_Shanghai

Source code in src/ethereum_test_fixtures/collector.py 
42
43
44
45
46
47
48
49
50
51
def get_name_and_parameters(self) -> Tuple[str, str]:
    """
    Convert test name to a tuple containing the test name and test parameters.

    Example:
    test_push0_key_sstore[fork_Shanghai] -> test_push0_key_sstore, fork_Shanghai

    """
    test_name, parameters = self.name.split("[")
    return test_name, re.sub(r"[\[\-]", "_", parameters).replace("]", "")

get_single_test_name(mode='module')

Convert test name to a single test name.

Source code in src/ethereum_test_fixtures/collector.py 
53
54
55
56
57
58
59
60
61
62
def get_single_test_name(self, mode: Literal["module", "test"] = "module") -> str:
    """Convert test name to a single test name."""
    if mode == "module":
        # Use the module name as the test name
        return self.strip_test_name(self.original_name)
    elif mode == "test":
        # Mix the module name and the test name/arguments
        test_name, test_parameters = self.get_name_and_parameters()
        test_name = self.strip_test_name(test_name)
        return f"{test_name}__{test_parameters}"

get_dump_dir_path(base_dump_dir, filler_path, level='test_parameter')

Path to dump the debug output as defined by the level to dump at.

Source code in src/ethereum_test_fixtures/collector.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def get_dump_dir_path(
    self,
    base_dump_dir: Optional[Path],
    filler_path: Path,
    level: Literal["test_module", "test_function", "test_parameter"] = "test_parameter",
) -> Optional[Path]:
    """Path to dump the debug output as defined by the level to dump at."""
    if not base_dump_dir:
        return None
    test_module_relative_dir = self.get_module_relative_output_dir(filler_path)
    if level == "test_module":
        return Path(base_dump_dir) / Path(str(test_module_relative_dir).replace(os.sep, "__"))
    test_name, test_parameter_string = self.get_name_and_parameters()
    flat_path = f"{str(test_module_relative_dir).replace(os.sep, '__')}__{test_name}"
    if level == "test_function":
        return Path(base_dump_dir) / flat_path
    elif level == "test_parameter":
        return Path(base_dump_dir) / flat_path / test_parameter_string
    raise Exception("Unexpected level.")

get_id()

Return the test id.

Source code in src/ethereum_test_fixtures/collector.py 
84
85
86
def get_id(self) -> str:
    """Return the test id."""
    return self.id

get_module_relative_output_dir(filler_path)

Return a directory name for the provided test_module (relative to the base ./tests directory) that can be used for output (within the configured fixtures output path or the base_dump_dir directory).

Example: tests/shanghai/eip3855_push0/test_push0.py -> shanghai/eip3855_push0/test_push0

Source code in src/ethereum_test_fixtures/collector.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def get_module_relative_output_dir(self, filler_path: Path) -> Path:
    """
    Return a directory name for the provided test_module (relative to the
    base ./tests directory) that can be used for output (within the
    configured fixtures output path or the base_dump_dir directory).

    Example:
    tests/shanghai/eip3855_push0/test_push0.py -> shanghai/eip3855_push0/test_push0

    """
    basename = self.module_path.with_suffix("").absolute()
    basename_relative = basename.relative_to(
        os.path.commonpath([filler_path.absolute(), basename])
    )
    module_path = basename_relative.parent / self.strip_test_name(basename_relative.stem)
    return module_path

FixtureConsumer

Bases: ABC

Abstract class for verifying Ethereum test fixtures.

Source code in src/ethereum_test_fixtures/consume.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
class FixtureConsumer(ABC):
    """Abstract class for verifying Ethereum test fixtures."""

    fixture_formats: List[FixtureFormat]

    def can_consume(
        self,
        fixture_format: FixtureFormat,
    ) -> bool:
        """Return whether the fixture format is consumable by this consumer."""
        return fixture_format in self.fixture_formats

    @abstractmethod
    def consume_fixture(
        self,
        fixture_format: FixtureFormat,
        fixture_path: Path,
        fixture_name: str | None = None,
        debug_output_path: Path | None = None,
    ):
        """Test the client with the specified fixture using its direct consumer interface."""
        raise NotImplementedError(
            "The `consume_fixture()` function is not supported by this tool."
        )

can_consume(fixture_format)

Return whether the fixture format is consumable by this consumer.

Source code in src/ethereum_test_fixtures/consume.py 
22
23
24
25
26
27
def can_consume(
    self,
    fixture_format: FixtureFormat,
) -> bool:
    """Return whether the fixture format is consumable by this consumer."""
    return fixture_format in self.fixture_formats

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

Test the client with the specified fixture using its direct consumer interface.

Source code in src/ethereum_test_fixtures/consume.py 
29
30
31
32
33
34
35
36
37
38
39
40
@abstractmethod
def consume_fixture(
    self,
    fixture_format: FixtureFormat,
    fixture_path: Path,
    fixture_name: str | None = None,
    debug_output_path: Path | None = None,
):
    """Test the client with the specified fixture using its direct consumer interface."""
    raise NotImplementedError(
        "The `consume_fixture()` function is not supported by this tool."
    )

EOFFixture

Bases: BaseFixture

Fixture for a single EOFTest.

Source code in src/ethereum_test_fixtures/eof.py 
41
42
43
44
45
46
47
48
49
50
51
class EOFFixture(BaseFixture):
    """Fixture for a single EOFTest."""

    format_name: ClassVar[str] = "eof_test"
    description: ClassVar[str] = "Tests that generate an EOF test fixture."

    vectors: Mapping[Number, Vector]

    def get_fork(self) -> Fork | None:
        """Return fork of the fixture as a string."""
        return None

get_fork()

Return fork of the fixture as a string.

Source code in src/ethereum_test_fixtures/eof.py 
49
50
51
def get_fork(self) -> Fork | None:
    """Return fork of the fixture as a string."""
    return None

SharedPreState

Bases: EthereumTestRootModel

Root model mapping pre-state hashes to test groups.

Source code in src/ethereum_test_fixtures/shared_alloc.py 
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
class SharedPreState(EthereumTestRootModel):
    """Root model mapping pre-state hashes to test groups."""

    root: Dict[str, SharedPreStateGroup]

    def __setitem__(self, key: str, value: Any):
        """Set item in root dict."""
        self.root[key] = value

    @classmethod
    def from_folder(cls, folder: Path) -> "SharedPreState":
        """Create SharedPreState from a folder of pre-allocation files."""
        data = {}
        for file in folder.glob("*.json"):
            with open(file) as f:
                data[file.stem] = SharedPreStateGroup.model_validate_json(f.read())
        return cls(root=data)

    def to_folder(self, folder: Path) -> None:
        """Save SharedPreState to a folder of pre-allocation files."""
        for key, value in self.root.items():
            value.to_file(folder / f"{key}.json")

    def __getitem__(self, item):
        """Get item from root dict."""
        return self.root[item]

    def __iter__(self):
        """Iterate over root dict."""
        return iter(self.root)

    def __contains__(self, item):
        """Check if item in root dict."""
        return item in self.root

    def __len__(self):
        """Get length of root dict."""
        return len(self.root)

    def keys(self):
        """Get keys from root dict."""
        return self.root.keys()

    def values(self):
        """Get values from root dict."""
        return self.root.values()

    def items(self):
        """Get items from root dict."""
        return self.root.items()

__setitem__(key, value)

Set item in root dict.

Source code in src/ethereum_test_fixtures/shared_alloc.py 
76
77
78
def __setitem__(self, key: str, value: Any):
    """Set item in root dict."""
    self.root[key] = value

from_folder(folder) classmethod

Create SharedPreState from a folder of pre-allocation files.

Source code in src/ethereum_test_fixtures/shared_alloc.py 
80
81
82
83
84
85
86
87
@classmethod
def from_folder(cls, folder: Path) -> "SharedPreState":
    """Create SharedPreState from a folder of pre-allocation files."""
    data = {}
    for file in folder.glob("*.json"):
        with open(file) as f:
            data[file.stem] = SharedPreStateGroup.model_validate_json(f.read())
    return cls(root=data)

to_folder(folder)

Save SharedPreState to a folder of pre-allocation files.

Source code in src/ethereum_test_fixtures/shared_alloc.py 
89
90
91
92
def to_folder(self, folder: Path) -> None:
    """Save SharedPreState to a folder of pre-allocation files."""
    for key, value in self.root.items():
        value.to_file(folder / f"{key}.json")

__getitem__(item)

Get item from root dict.

Source code in src/ethereum_test_fixtures/shared_alloc.py 
94
95
96
def __getitem__(self, item):
    """Get item from root dict."""
    return self.root[item]

__iter__()

Iterate over root dict.

Source code in src/ethereum_test_fixtures/shared_alloc.py 
 98
 99
100
def __iter__(self):
    """Iterate over root dict."""
    return iter(self.root)

__contains__(item)

Check if item in root dict.

Source code in src/ethereum_test_fixtures/shared_alloc.py 
102
103
104
def __contains__(self, item):
    """Check if item in root dict."""
    return item in self.root

__len__()

Get length of root dict.

Source code in src/ethereum_test_fixtures/shared_alloc.py 
106
107
108
def __len__(self):
    """Get length of root dict."""
    return len(self.root)

keys()

Get keys from root dict.

Source code in src/ethereum_test_fixtures/shared_alloc.py 
110
111
112
def keys(self):
    """Get keys from root dict."""
    return self.root.keys()

values()

Get values from root dict.

Source code in src/ethereum_test_fixtures/shared_alloc.py 
114
115
116
def values(self):
    """Get values from root dict."""
    return self.root.values()

items()

Get items from root dict.

Source code in src/ethereum_test_fixtures/shared_alloc.py 
118
119
120
def items(self):
    """Get items from root dict."""
    return self.root.items()

SharedPreStateGroup

Bases: CamelModel

Shared pre-state group for tests with identical Environment and fork values.

Groups tests by a hash of their fixture Environment and fork to enable shared pre-allocation optimization.

Source code in src/ethereum_test_fixtures/shared_alloc.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
class SharedPreStateGroup(CamelModel):
    """
    Shared pre-state group for tests with identical Environment and fork values.

    Groups tests by a hash of their fixture Environment and fork to enable
    shared pre-allocation optimization.
    """

    model_config = {"populate_by_name": True}  # Allow both field names and aliases

    test_count: int = Field(0, description="Number of tests in this group")
    pre_account_count: int = Field(0, description="Number of accounts in the pre-allocation")
    test_ids: List[str] = Field(default_factory=list, alias="testIds")
    environment: Environment = Field(..., description="Grouping environment for this test group")
    fork: Fork = Field(..., alias="network")
    pre: Alloc

    def model_post_init(self, __context):
        """Post-init hook to ensure pre is not None."""
        super().model_post_init(__context)

        self.pre = Alloc.merge(
            Alloc.model_validate(self.fork.pre_allocation_blockchain()),
            self.pre,
        )

    @computed_field  # type: ignore[misc]
    def genesis(self) -> FixtureHeader:
        """Get the genesis header for this group."""
        return FixtureHeader.genesis(
            self.fork,
            self.environment.set_fork_requirements(self.fork),
            self.pre.state_root(),
        )

    def to_file(self, file: Path) -> None:
        """Save SharedPreStateGroup to a file."""
        lock_file_path = file.with_suffix(".lock")
        with FileLock(lock_file_path):
            if file.exists():
                with open(file, "r") as f:
                    previous_shared_pre_state_group = SharedPreStateGroup.model_validate_json(
                        f.read()
                    )
                    for account in previous_shared_pre_state_group.pre:
                        if account not in self.pre:
                            self.pre[account] = previous_shared_pre_state_group.pre[account]
                    self.pre_account_count += previous_shared_pre_state_group.pre_account_count
                    self.test_count += previous_shared_pre_state_group.test_count
                    self.test_ids.extend(previous_shared_pre_state_group.test_ids)

            with open(file, "w") as f:
                f.write(self.model_dump_json(by_alias=True, exclude_none=True, indent=2))

model_post_init(__context)

Post-init hook to ensure pre is not None.

Source code in src/ethereum_test_fixtures/shared_alloc.py 
33
34
35
36
37
38
39
40
def model_post_init(self, __context):
    """Post-init hook to ensure pre is not None."""
    super().model_post_init(__context)

    self.pre = Alloc.merge(
        Alloc.model_validate(self.fork.pre_allocation_blockchain()),
        self.pre,
    )

genesis()

Get the genesis header for this group.

Source code in src/ethereum_test_fixtures/shared_alloc.py 
42
43
44
45
46
47
48
49
@computed_field  # type: ignore[misc]
def genesis(self) -> FixtureHeader:
    """Get the genesis header for this group."""
    return FixtureHeader.genesis(
        self.fork,
        self.environment.set_fork_requirements(self.fork),
        self.pre.state_root(),
    )

to_file(file)

Save SharedPreStateGroup to a file.

Source code in src/ethereum_test_fixtures/shared_alloc.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def to_file(self, file: Path) -> None:
    """Save SharedPreStateGroup to a file."""
    lock_file_path = file.with_suffix(".lock")
    with FileLock(lock_file_path):
        if file.exists():
            with open(file, "r") as f:
                previous_shared_pre_state_group = SharedPreStateGroup.model_validate_json(
                    f.read()
                )
                for account in previous_shared_pre_state_group.pre:
                    if account not in self.pre:
                        self.pre[account] = previous_shared_pre_state_group.pre[account]
                self.pre_account_count += previous_shared_pre_state_group.pre_account_count
                self.test_count += previous_shared_pre_state_group.test_count
                self.test_ids.extend(previous_shared_pre_state_group.test_ids)

        with open(file, "w") as f:
            f.write(self.model_dump_json(by_alias=True, exclude_none=True, indent=2))

StateFixture

Bases: BaseFixture

Fixture for a single StateTest.

Source code in src/ethereum_test_fixtures/state.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
class StateFixture(BaseFixture):
    """Fixture for a single StateTest."""

    format_name: ClassVar[str] = "state_test"
    description: ClassVar[str] = "Tests that generate a state test fixture."

    env: FixtureEnvironment
    pre: Alloc
    transaction: FixtureTransaction
    post: Mapping[Fork, List[FixtureForkPost]]
    config: FixtureConfig

    def get_fork(self) -> Fork | None:
        """Return fork of the fixture as a string."""
        forks = list(self.post.keys())
        assert len(forks) == 1, "Expected state test fixture with single fork"
        return forks[0]

get_fork()

Return fork of the fixture as a string.

Source code in src/ethereum_test_fixtures/state.py 
104
105
106
107
108
def get_fork(self) -> Fork | None:
    """Return fork of the fixture as a string."""
    forks = list(self.post.keys())
    assert len(forks) == 1, "Expected state test fixture with single fork"
    return forks[0]

TransactionFixture

Bases: BaseFixture

Fixture for a single TransactionTest.

Source code in src/ethereum_test_fixtures/transaction.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
class TransactionFixture(BaseFixture):
    """Fixture for a single TransactionTest."""

    format_name: ClassVar[str] = "transaction_test"
    description: ClassVar[str] = "Tests that generate a transaction test fixture."

    result: Mapping[Fork, FixtureResult]
    transaction: Bytes = Field(..., alias="txbytes")

    def get_fork(self) -> Fork | None:
        """Return the fork of the fixture as a string."""
        forks = list(self.result.keys())
        assert len(forks) == 1, "Expected transaction test fixture with single fork"
        return forks[0]

get_fork()

Return the fork of the fixture as a string.

Source code in src/ethereum_test_fixtures/transaction.py 
32
33
34
35
36
def get_fork(self) -> Fork | None:
    """Return the fork of the fixture as a string."""
    forks = list(self.result.keys())
    assert len(forks) == 1, "Expected transaction test fixture with single fork"
    return forks[0]