Skip to content

Filler Plugin

A pytest plugin to fill tests and generate JSON fixtures.

FixtureOutput

Bases: BaseModel

Represents the output destination for generated test fixtures.

Source code in src/pytest_plugins/filler/fixture_output.py 
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
class FixtureOutput(BaseModel):
    """Represents the output destination for generated test fixtures."""

    output_path: Path = Field(description="Directory path to store the generated test fixtures")
    flat_output: bool = Field(
        default=False,
        description="Output each test case in the directory without the folder structure",
    )
    single_fixture_per_file: bool = Field(
        default=False,
        description=(
            "Don't group fixtures in JSON files by test function; "
            "write each fixture to its own file"
        ),
    )
    clean: bool = Field(
        default=False,
        description="Clean (remove) the output directory before filling fixtures.",
    )
    generate_shared_pre: bool = Field(
        default=False,
        description="Generate shared pre-allocation state (phase 1).",
    )
    use_shared_pre: bool = Field(
        default=False,
        description="Use existing shared pre-allocation state (phase 2).",
    )

    @property
    def directory(self) -> Path:
        """Return the actual directory path where fixtures will be written."""
        return self.strip_tarball_suffix(self.output_path)

    @property
    def metadata_dir(self) -> Path:
        """Return metadata directory to store fixture meta files."""
        if self.is_stdout:
            return self.directory
        return self.directory / ".meta"

    @property
    def is_tarball(self) -> bool:
        """Return True if the output should be packaged as a tarball."""
        path = self.output_path
        return path.suffix == ".gz" and path.with_suffix("").suffix == ".tar"

    @property
    def is_stdout(self) -> bool:
        """Return True if the fixture output is configured to be stdout."""
        return self.directory.name == "stdout"

    @property
    def shared_pre_alloc_folder_path(self) -> Path:
        """Return the path for shared pre-allocation state file."""
        reorg_dir = BlockchainEngineReorgFixture.output_base_dir_name()
        return self.directory / reorg_dir / "pre_alloc"

    @staticmethod
    def strip_tarball_suffix(path: Path) -> Path:
        """Strip the '.tar.gz' suffix from the output path."""
        if str(path).endswith(".tar.gz"):
            return path.with_suffix("").with_suffix("")
        return path

    def is_directory_empty(self) -> bool:
        """Check if the output directory is empty."""
        if not self.directory.exists():
            return True

        return not any(self.directory.iterdir())

    def is_directory_usable_for_phase(self) -> bool:
        """Check if the output directory is usable for the current phase."""
        if not self.directory.exists():
            return True

        if self.generate_shared_pre:
            # Phase 1: Directory must be completely empty
            return self.is_directory_empty()
        elif self.use_shared_pre:
            # Phase 2: Only shared alloc file must exist, no other files allowed
            if not self.shared_pre_alloc_folder_path.exists():
                return False
            # Check that only the shared prealloc file exists
            existing_files = {f for f in self.directory.rglob("*") if f.is_file()}
            allowed_files = set(self.shared_pre_alloc_folder_path.rglob("*.json"))
            return existing_files == allowed_files
        else:
            # Normal filling: Directory must be empty
            return self.is_directory_empty()

    def get_directory_summary(self) -> str:
        """Return a summary of directory contents for error reporting."""
        if not self.directory.exists():
            return "directory does not exist"

        items = list(self.directory.iterdir())
        if not items:
            return "empty directory"

        dirs = [d.name for d in items if d.is_dir()]
        files = [f.name for f in items if f.is_file()]

        max_dirs = 4
        summary_parts = []
        if dirs:
            summary_parts.append(
                f"{len(dirs)} directories"
                + (
                    f" ({', '.join(dirs[:max_dirs])}"
                    + (f"... and {len(dirs) - max_dirs} more" if len(dirs) > max_dirs else "")
                    + ")"
                    if dirs
                    else ""
                )
            )
        if files:
            summary_parts.append(
                f"{len(files)} files"
                + (
                    f" ({', '.join(files[:3])}"
                    + (f"... and {len(files) - 3} more" if len(files) > 3 else "")
                    + ")"
                    if files
                    else ""
                )
            )

        return " and ".join(summary_parts)

    def create_directories(self, is_master: bool) -> None:
        """
        Create output and metadata directories if needed.

        If clean flag is set, remove and recreate the directory.
        Otherwise, verify the directory is empty before proceeding.
        """
        if self.is_stdout:
            return

        # Only the master process should delete/create directories if using pytest-xdist
        if not is_master:
            return

        if self.directory.exists() and self.clean:
            shutil.rmtree(self.directory)

        if self.directory.exists() and not self.is_directory_usable_for_phase():
            summary = self.get_directory_summary()

            if self.generate_shared_pre:
                raise ValueError(
                    f"Output directory '{self.directory}' must be completely empty for "
                    f"shared allocation generation (phase 1). Contains: {summary}. "
                    "Use --clean to remove all existing files."
                )
            elif self.use_shared_pre:
                if not self.shared_pre_alloc_folder_path.exists():
                    raise ValueError(
                        "Shared pre-allocation file not found at "
                        f"'{self.shared_pre_alloc_folder_path}'. "
                        "Run phase 1 with --generate-shared-pre first."
                    )
            else:
                raise ValueError(
                    f"Output directory '{self.directory}' is not empty. "
                    f"Contains: {summary}. Use --clean to remove all existing files "
                    "or specify a different output directory."
                )

        # Create directories
        self.directory.mkdir(parents=True, exist_ok=True)
        self.metadata_dir.mkdir(parents=True, exist_ok=True)

        # Create shared allocation directory for phase 1
        if self.generate_shared_pre:
            self.shared_pre_alloc_folder_path.parent.mkdir(parents=True, exist_ok=True)

    def create_tarball(self) -> None:
        """Create tarball of the output directory if configured to do so."""
        if not self.is_tarball:
            return

        with tarfile.open(self.output_path, "w:gz") as tar:
            for file in self.directory.rglob("*"):
                if file.suffix in {".json", ".ini"}:
                    arcname = Path("fixtures") / file.relative_to(self.directory)
                    tar.add(file, arcname=arcname)

    @classmethod
    def from_config(cls, config: pytest.Config) -> "FixtureOutput":
        """Create a FixtureOutput instance from pytest configuration."""
        return cls(
            output_path=config.getoption("output"),
            flat_output=config.getoption("flat_output"),
            single_fixture_per_file=config.getoption("single_fixture_per_file"),
            clean=config.getoption("clean"),
            generate_shared_pre=config.getoption("generate_shared_pre"),
            use_shared_pre=config.getoption("use_shared_pre"),
        )

directory: Path property

Return the actual directory path where fixtures will be written.

metadata_dir: Path property

Return metadata directory to store fixture meta files.

is_tarball: bool property

Return True if the output should be packaged as a tarball.

is_stdout: bool property

Return True if the fixture output is configured to be stdout.

shared_pre_alloc_folder_path: Path property

Return the path for shared pre-allocation state file.

strip_tarball_suffix(path) staticmethod

Strip the '.tar.gz' suffix from the output path.

Source code in src/pytest_plugins/filler/fixture_output.py 
70
71
72
73
74
75
@staticmethod
def strip_tarball_suffix(path: Path) -> Path:
    """Strip the '.tar.gz' suffix from the output path."""
    if str(path).endswith(".tar.gz"):
        return path.with_suffix("").with_suffix("")
    return path

is_directory_empty()

Check if the output directory is empty.

Source code in src/pytest_plugins/filler/fixture_output.py 
77
78
79
80
81
82
def is_directory_empty(self) -> bool:
    """Check if the output directory is empty."""
    if not self.directory.exists():
        return True

    return not any(self.directory.iterdir())

is_directory_usable_for_phase()

Check if the output directory is usable for the current phase.

Source code in src/pytest_plugins/filler/fixture_output.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def is_directory_usable_for_phase(self) -> bool:
    """Check if the output directory is usable for the current phase."""
    if not self.directory.exists():
        return True

    if self.generate_shared_pre:
        # Phase 1: Directory must be completely empty
        return self.is_directory_empty()
    elif self.use_shared_pre:
        # Phase 2: Only shared alloc file must exist, no other files allowed
        if not self.shared_pre_alloc_folder_path.exists():
            return False
        # Check that only the shared prealloc file exists
        existing_files = {f for f in self.directory.rglob("*") if f.is_file()}
        allowed_files = set(self.shared_pre_alloc_folder_path.rglob("*.json"))
        return existing_files == allowed_files
    else:
        # Normal filling: Directory must be empty
        return self.is_directory_empty()

get_directory_summary()

Return a summary of directory contents for error reporting.

Source code in src/pytest_plugins/filler/fixture_output.py 
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
def get_directory_summary(self) -> str:
    """Return a summary of directory contents for error reporting."""
    if not self.directory.exists():
        return "directory does not exist"

    items = list(self.directory.iterdir())
    if not items:
        return "empty directory"

    dirs = [d.name for d in items if d.is_dir()]
    files = [f.name for f in items if f.is_file()]

    max_dirs = 4
    summary_parts = []
    if dirs:
        summary_parts.append(
            f"{len(dirs)} directories"
            + (
                f" ({', '.join(dirs[:max_dirs])}"
                + (f"... and {len(dirs) - max_dirs} more" if len(dirs) > max_dirs else "")
                + ")"
                if dirs
                else ""
            )
        )
    if files:
        summary_parts.append(
            f"{len(files)} files"
            + (
                f" ({', '.join(files[:3])}"
                + (f"... and {len(files) - 3} more" if len(files) > 3 else "")
                + ")"
                if files
                else ""
            )
        )

    return " and ".join(summary_parts)

create_directories(is_master)

Create output and metadata directories if needed.

If clean flag is set, remove and recreate the directory. Otherwise, verify the directory is empty before proceeding.

Source code in src/pytest_plugins/filler/fixture_output.py 
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
def create_directories(self, is_master: bool) -> None:
    """
    Create output and metadata directories if needed.

    If clean flag is set, remove and recreate the directory.
    Otherwise, verify the directory is empty before proceeding.
    """
    if self.is_stdout:
        return

    # Only the master process should delete/create directories if using pytest-xdist
    if not is_master:
        return

    if self.directory.exists() and self.clean:
        shutil.rmtree(self.directory)

    if self.directory.exists() and not self.is_directory_usable_for_phase():
        summary = self.get_directory_summary()

        if self.generate_shared_pre:
            raise ValueError(
                f"Output directory '{self.directory}' must be completely empty for "
                f"shared allocation generation (phase 1). Contains: {summary}. "
                "Use --clean to remove all existing files."
            )
        elif self.use_shared_pre:
            if not self.shared_pre_alloc_folder_path.exists():
                raise ValueError(
                    "Shared pre-allocation file not found at "
                    f"'{self.shared_pre_alloc_folder_path}'. "
                    "Run phase 1 with --generate-shared-pre first."
                )
        else:
            raise ValueError(
                f"Output directory '{self.directory}' is not empty. "
                f"Contains: {summary}. Use --clean to remove all existing files "
                "or specify a different output directory."
            )

    # Create directories
    self.directory.mkdir(parents=True, exist_ok=True)
    self.metadata_dir.mkdir(parents=True, exist_ok=True)

    # Create shared allocation directory for phase 1
    if self.generate_shared_pre:
        self.shared_pre_alloc_folder_path.parent.mkdir(parents=True, exist_ok=True)

create_tarball()

Create tarball of the output directory if configured to do so.

Source code in src/pytest_plugins/filler/fixture_output.py 
191
192
193
194
195
196
197
198
199
200
def create_tarball(self) -> None:
    """Create tarball of the output directory if configured to do so."""
    if not self.is_tarball:
        return

    with tarfile.open(self.output_path, "w:gz") as tar:
        for file in self.directory.rglob("*"):
            if file.suffix in {".json", ".ini"}:
                arcname = Path("fixtures") / file.relative_to(self.directory)
                tar.add(file, arcname=arcname)

from_config(config) classmethod

Create a FixtureOutput instance from pytest configuration.

Source code in src/pytest_plugins/filler/fixture_output.py 
202
203
204
205
206
207
208
209
210
211
212
@classmethod
def from_config(cls, config: pytest.Config) -> "FixtureOutput":
    """Create a FixtureOutput instance from pytest configuration."""
    return cls(
        output_path=config.getoption("output"),
        flat_output=config.getoption("flat_output"),
        single_fixture_per_file=config.getoption("single_fixture_per_file"),
        clean=config.getoption("clean"),
        generate_shared_pre=config.getoption("generate_shared_pre"),
        use_shared_pre=config.getoption("use_shared_pre"),
    )

Top-level pytest configuration file providing: - Command-line options, - Test-fixtures that can be used by all test cases, and that modifies pytest hooks in order to fill test specs for all tests and writes the generated fixtures to file.

calculate_post_state_diff(post_state, genesis_state)

Calculate the state difference between post_state and genesis_state.

This function enables significant space savings in reorg fixtures by storing only the accounts that changed during test execution, rather than the full post-state which may contain thousands of unchanged shared accounts.

Returns an Alloc containing only the accounts that: - Changed between genesis and post state (balance, nonce, storage, code) - Were created during test execution (new accounts) - Were deleted during test execution (represented as None)

Parameters:

Name Type Description Default
post_state Alloc

Final state after test execution

 required
genesis_state Alloc

Shared genesis pre-allocation state

 required

Returns:

Type Description
Alloc

Alloc containing only the state differences for efficient storage
Source code in src/pytest_plugins/filler/filler.py 
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
def calculate_post_state_diff(post_state: Alloc, genesis_state: Alloc) -> Alloc:
    """
    Calculate the state difference between post_state and genesis_state.

    This function enables significant space savings in reorg fixtures by storing
    only the accounts that changed during test execution, rather than the full
    post-state which may contain thousands of unchanged shared accounts.

    Returns an Alloc containing only the accounts that:
    - Changed between genesis and post state (balance, nonce, storage, code)
    - Were created during test execution (new accounts)
    - Were deleted during test execution (represented as None)

    Args:
        post_state: Final state after test execution
        genesis_state: Shared genesis pre-allocation state

    Returns:
        Alloc containing only the state differences for efficient storage

    """
    diff: Dict[Address, Account | None] = {}

    # Find all addresses that exist in either state
    all_addresses = set(post_state.root.keys()) | set(genesis_state.root.keys())

    for address in all_addresses:
        genesis_account = genesis_state.root.get(address)
        post_account = post_state.root.get(address)

        # Account was deleted (exists in genesis but not in post)
        if genesis_account is not None and post_account is None:
            diff[address] = None

        # Account was created (doesn't exist in genesis but exists in post)
        elif genesis_account is None and post_account is not None:
            diff[address] = post_account

        # Account was modified (exists in both but different)
        elif genesis_account != post_account:
            diff[address] = post_account

        # Account unchanged - don't include in diff

    return Alloc(diff)

default_output_directory()

Directory (default) to store the generated test fixtures. Defined as a function to allow for easier testing.

Source code in src/pytest_plugins/filler/filler.py 
101
102
103
104
105
106
def default_output_directory() -> str:
    """
    Directory (default) to store the generated test fixtures. Defined as a
    function to allow for easier testing.
    """
    return "./fixtures"

default_html_report_file_path()

File path (default) to store the generated HTML test report. Defined as a function to allow for easier testing.

Source code in src/pytest_plugins/filler/filler.py 
109
110
111
112
113
114
def default_html_report_file_path() -> str:
    """
    File path (default) to store the generated HTML test report. Defined as a
    function to allow for easier testing.
    """
    return ".meta/report_fill.html"

pytest_addoption(parser)

Add command-line options to pytest.

Source code in src/pytest_plugins/filler/filler.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
def pytest_addoption(parser: pytest.Parser):
    """Add command-line options to pytest."""
    evm_group = parser.getgroup("evm", "Arguments defining evm executable behavior")
    evm_group.addoption(
        "--evm-bin",
        action="store",
        dest="evm_bin",
        type=Path,
        default="ethereum-spec-evm-resolver",
        help=(
            "Path to an evm executable (or name of an executable in the PATH) that provides `t8n`."
            " Default: `ethereum-spec-evm-resolver`."
        ),
    )
    evm_group.addoption(
        "--traces",
        action="store_true",
        dest="evm_collect_traces",
        default=None,
        help="Collect traces of the execution information from the transition tool.",
    )
    evm_group.addoption(
        "--verify-fixtures",
        action="store_true",
        dest="verify_fixtures",
        default=False,
        help=(
            "Verify generated fixture JSON files using geth's evm blocktest command. "
            "By default, the same evm binary as for the t8n tool is used. A different (geth) evm "
            "binary may be specified via --verify-fixtures-bin, this must be specified if filling "
            "with a non-geth t8n tool that does not support blocktest."
        ),
    )
    evm_group.addoption(
        "--verify-fixtures-bin",
        action="store",
        dest="verify_fixtures_bin",
        type=Path,
        default=None,
        help=(
            "Path to an evm executable that provides the `blocktest` command. "
            "Default: The first (geth) 'evm' entry in PATH."
        ),
    )

    test_group = parser.getgroup("tests", "Arguments defining filler location and output")
    test_group.addoption(
        "--filler-path",
        action="store",
        dest="filler_path",
        default="./tests/",
        type=Path,
        help="Path to filler directives",
    )
    test_group.addoption(
        "--output",
        action="store",
        dest="output",
        type=Path,
        default=Path(default_output_directory()),
        help=(
            "Directory path to store the generated test fixtures. Must be empty if it exists. "
            "If the specified path ends in '.tar.gz', then the specified tarball is additionally "
            "created (the fixtures are still written to the specified path without the '.tar.gz' "
            f"suffix). Can be deleted. Default: '{default_output_directory()}'."
        ),
    )
    test_group.addoption(
        "--clean",
        action="store_true",
        dest="clean",
        default=False,
        help="Clean (remove) the output directory before filling fixtures.",
    )
    test_group.addoption(
        "--flat-output",
        action="store_true",
        dest="flat_output",
        default=False,
        help="Output each test case in the directory without the folder structure.",
    )
    test_group.addoption(
        "--single-fixture-per-file",
        action="store_true",
        dest="single_fixture_per_file",
        default=False,
        help=(
            "Don't group fixtures in JSON files by test function; write each fixture to its own "
            "file. This can be used to increase the granularity of --verify-fixtures."
        ),
    )
    test_group.addoption(
        "--no-html",
        action="store_true",
        dest="disable_html",
        default=False,
        help=(
            "Don't generate an HTML test report (in the output directory). "
            "The --html flag can be used to specify a different path."
        ),
    )
    test_group.addoption(
        "--build-name",
        action="store",
        dest="build_name",
        default=None,
        type=str,
        help="Specify a build name for the fixtures.ini file, e.g., 'stable'.",
    )
    test_group.addoption(
        "--skip-index",
        action="store_false",
        dest="generate_index",
        default=True,
        help="Skip generating an index file for all produced fixtures.",
    )
    test_group.addoption(
        "--block-gas-limit",
        action="store",
        dest="block_gas_limit",
        default=EnvironmentDefaults.gas_limit,
        type=int,
        help=(
            "Default gas limit used ceiling used for blocks and tests that attempt to "
            f"consume an entire block's gas. (Default: {EnvironmentDefaults.gas_limit})"
        ),
    )
    test_group.addoption(
        "--generate-shared-pre",
        action="store_true",
        dest="generate_shared_pre",
        default=False,
        help="Generate shared pre-allocation state (phase 1 only).",
    )
    test_group.addoption(
        "--use-shared-pre",
        action="store_true",
        dest="use_shared_pre",
        default=False,
        help="Fill tests using an existing shared pre-allocation state (phase 2 only).",
    )

    debug_group = parser.getgroup("debug", "Arguments defining debug behavior")
    debug_group.addoption(
        "--evm-dump-dir",
        "--t8n-dump-dir",
        action="store",
        dest="base_dump_dir",
        default=AppConfig().DEFAULT_EVM_LOGS_DIR,
        help=(
            "Path to dump the transition tool debug output. "
            f"(Default: {AppConfig().DEFAULT_EVM_LOGS_DIR})"
        ),
    )
    debug_group.addoption(
        "--skip-evm-dump",
        "--skip-t8n-dump",
        action="store_true",
        dest="skip_dump_dir",
        default=False,
        help=("Skip dumping the the transition tool debug output."),
    )

pytest_sessionstart(session)

Initialize session-level state.

Either initialize an empty shared pre-state container for phase 1 or load the shared pre-allocation state for phase 2 execution.

Source code in src/pytest_plugins/filler/filler.py 
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
def pytest_sessionstart(session: pytest.Session):
    """
    Initialize session-level state.

    Either initialize an empty shared pre-state container for phase 1 or
    load the shared pre-allocation state for phase 2 execution.
    """
    # Initialize empty shared pre-state container for phase 1
    if session.config.getoption("generate_shared_pre"):
        session.config.shared_pre_state = SharedPreState(root={})  # type: ignore[attr-defined]

    # Load the pre-state for phase 2
    if session.config.getoption("use_shared_pre"):
        shared_pre_alloc_folder = session.config.fixture_output.shared_pre_alloc_folder_path  # type: ignore[attr-defined]
        if shared_pre_alloc_folder.exists():
            session.config.shared_pre_state = SharedPreState.from_folder(shared_pre_alloc_folder)  # type: ignore[attr-defined]
        else:
            pytest.exit(
                f"Shared pre-alloc file not found: {shared_pre_alloc_folder}. "
                "Run phase 1 with --generate-shared-alloc first.",
                returncode=pytest.ExitCode.USAGE_ERROR,
            )

pytest_configure(config)

Pytest hook called after command line options have been parsed and before test collection begins.

Couple of notes: 1. Register the plugin's custom markers and process command-line options.

Custom marker registration:
https://docs.pytest.org/en/7.1.x/how-to/writing_plugins.html#registering-custom-markers
  1. @pytest.hookimpl(tryfirst=True) is applied to ensure that this hook is called before the pytest-html plugin's pytest_configure to ensure that it uses the modified htmlpath option.
Source code in src/pytest_plugins/filler/filler.py 
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
@pytest.hookimpl(tryfirst=True)
def pytest_configure(config):
    """
    Pytest hook called after command line options have been parsed and before
    test collection begins.

    Couple of notes:
    1. Register the plugin's custom markers and process command-line options.

        Custom marker registration:
        https://docs.pytest.org/en/7.1.x/how-to/writing_plugins.html#registering-custom-markers

    2. `@pytest.hookimpl(tryfirst=True)` is applied to ensure that this hook is
        called before the pytest-html plugin's pytest_configure to ensure that
        it uses the modified `htmlpath` option.
    """
    # Modify the block gas limit if specified.
    if config.getoption("block_gas_limit"):
        EnvironmentDefaults.gas_limit = config.getoption("block_gas_limit")

    # Initialize fixture output configuration
    config.fixture_output = FixtureOutput.from_config(config)

    if is_help_or_collectonly_mode(config):
        return

    try:
        # Check whether the directory exists and is not empty; if --clean is set, it will delete it
        config.fixture_output.create_directories(is_master=not hasattr(config, "workerinput"))
    except ValueError as e:
        pytest.exit(str(e), returncode=pytest.ExitCode.USAGE_ERROR)

    if (
        not config.getoption("disable_html")
        and config.getoption("htmlpath") is None
        and not config.getoption("generate_shared_pre")
    ):
        config.option.htmlpath = config.fixture_output.directory / default_html_report_file_path()

    # Instantiate the transition tool here to check that the binary path/trace option is valid.
    # This ensures we only raise an error once, if appropriate, instead of for every test.
    t8n = TransitionTool.from_binary_path(
        binary_path=config.getoption("evm_bin"), trace=config.getoption("evm_collect_traces")
    )
    if (
        isinstance(config.getoption("numprocesses"), int)
        and config.getoption("numprocesses") > 0
        and "Besu" in str(t8n.detect_binary_pattern)
    ):
        pytest.exit(
            "The Besu t8n tool does not work well with the xdist plugin; use -n=0.",
            returncode=pytest.ExitCode.USAGE_ERROR,
        )

    if "Tools" not in config.stash[metadata_key]:
        config.stash[metadata_key]["Tools"] = {
            "t8n": t8n.version(),
        }
    else:
        config.stash[metadata_key]["Tools"]["t8n"] = t8n.version()

    args = ["fill"] + [str(arg) for arg in config.invocation_params.args]
    for i in range(len(args)):
        if " " in args[i]:
            args[i] = f'"{args[i]}"'
    command_line_args = " ".join(args)
    config.stash[metadata_key]["Command-line args"] = f"<code>{command_line_args}</code>"

pytest_report_header(config)

Add lines to pytest's console output header.

Source code in src/pytest_plugins/filler/filler.py 
374
375
376
377
378
379
380
@pytest.hookimpl(trylast=True)
def pytest_report_header(config: pytest.Config):
    """Add lines to pytest's console output header."""
    if is_help_or_collectonly_mode(config):
        return
    t8n_version = config.stash[metadata_key]["Tools"]["t8n"]
    return [(f"{t8n_version}")]

pytest_report_teststatus(report, config)

Modify test results in pytest's terminal output.

We use this:

  1. To disable test session progress report if we're writing the JSON fixtures to stdout to be read by a consume command on stdin. I.e., don't write this type of output to the console: 
    ...x...
Source code in src/pytest_plugins/filler/filler.py 
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
@pytest.hookimpl(tryfirst=True)
def pytest_report_teststatus(report, config: pytest.Config):
    """
    Modify test results in pytest's terminal output.

    We use this:

    1. To disable test session progress report if we're writing the JSON
        fixtures to stdout to be read by a consume command on stdin. I.e.,
        don't write this type of output to the console:
    ```text
    ...x...
    ```
    """
    if config.fixture_output.is_stdout:  # type: ignore[attr-defined]
        return report.outcome, "", report.outcome.upper()

pytest_terminal_summary(terminalreporter, exitstatus, config)

Modify pytest's terminal summary to emphasize that no tests were ran.

Emphasize that fixtures have only been filled; they must now be executed to actually run the tests.

Source code in src/pytest_plugins/filler/filler.py 
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
@pytest.hookimpl(hookwrapper=True, trylast=True)
def pytest_terminal_summary(
    terminalreporter: TerminalReporter, exitstatus: int, config: pytest.Config
):
    """
    Modify pytest's terminal summary to emphasize that no tests were ran.

    Emphasize that fixtures have only been filled; they must now be executed to
    actually run the tests.
    """
    yield
    if config.fixture_output.is_stdout or hasattr(config, "workerinput"):  # type: ignore[attr-defined]
        return
    stats = terminalreporter.stats
    if "passed" in stats and stats["passed"]:
        # Custom message for Phase 1 (shared pre-allocation generation)
        if config.getoption("generate_shared_pre"):
            # Generate summary stats
            shared_pre_state: SharedPreState
            if config.pluginmanager.hasplugin("xdist"):
                # Load shared pre-state from disk
                shared_pre_state = SharedPreState.from_folder(
                    config.fixture_output.shared_pre_alloc_folder_path  # type: ignore[attr-defined]
                )
            else:
                assert hasattr(config, "shared_pre_state")
                shared_pre_state = config.shared_pre_state  # type: ignore[attr-defined]

            total_groups = len(shared_pre_state.root)
            total_accounts = sum(
                group.pre_account_count for group in shared_pre_state.root.values()
            )

            terminalreporter.write_sep(
                "=",
                f" Phase 1 Complete: Generated {total_groups} shared pre-allocation groups "
                f"({total_accounts} total accounts) ",
                bold=True,
                green=True,
            )

        else:
            # Normal message for fixture generation
            # append / to indicate this is a directory
            output_dir = str(config.fixture_output.directory) + "/"  # type: ignore[attr-defined]
            terminalreporter.write_sep(
                "=",
                (
                    f' No tests executed - the test fixtures in "{output_dir}" may now be '
                    "executed against a client "
                ),
                bold=True,
                yellow=True,
            )

pytest_metadata(metadata)

Add or remove metadata to/from the pytest report.

Source code in src/pytest_plugins/filler/filler.py 
457
458
459
def pytest_metadata(metadata):
    """Add or remove metadata to/from the pytest report."""
    metadata.pop("JAVA_HOME", None)

pytest_html_results_table_header(cells)

Customize the table headers of the HTML report table.

Source code in src/pytest_plugins/filler/filler.py 
462
463
464
465
466
def pytest_html_results_table_header(cells):
    """Customize the table headers of the HTML report table."""
    cells.insert(3, '<th class="sortable" data-column-type="fixturePath">JSON Fixture File</th>')
    cells.insert(4, '<th class="sortable" data-column-type="evmDumpDir">EVM Dump Dir</th>')
    del cells[-1]  # Remove the "Links" column

pytest_html_results_table_row(report, cells)

Customize the table rows of the HTML report table.

Source code in src/pytest_plugins/filler/filler.py 
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
def pytest_html_results_table_row(report, cells):
    """Customize the table rows of the HTML report table."""
    if hasattr(report, "user_properties"):
        user_props = dict(report.user_properties)
        if (
            report.passed
            and "fixture_path_absolute" in user_props
            and "fixture_path_relative" in user_props
        ):
            fixture_path_absolute = user_props["fixture_path_absolute"]
            fixture_path_relative = user_props["fixture_path_relative"]
            fixture_path_link = (
                f'<a href="{fixture_path_absolute}" target="_blank">{fixture_path_relative}</a>'
            )
            cells.insert(3, f"<td>{fixture_path_link}</td>")
        elif report.failed:
            cells.insert(3, "<td>Fixture unavailable</td>")
        if "evm_dump_dir" in user_props:
            if user_props["evm_dump_dir"] is None:
                cells.insert(
                    4, "<td>For t8n debug info use <code>--evm-dump-dir=path --traces</code></td>"
                )
            else:
                evm_dump_dir = user_props.get("evm_dump_dir")
                if evm_dump_dir == "N/A":
                    evm_dump_entry = "N/A"
                else:
                    evm_dump_entry = f'<a href="{evm_dump_dir}" target="_blank">{evm_dump_dir}</a>'
                cells.insert(4, f"<td>{evm_dump_entry}</td>")
    del cells[-1]  # Remove the "Links" column

pytest_runtest_makereport(item, call)

Make each test's fixture json path available to the test report via user_properties.

This hook is called when each test is run and a report is being made.

Source code in src/pytest_plugins/filler/filler.py 
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_makereport(item, call):
    """
    Make each test's fixture json path available to the test report via
    user_properties.

    This hook is called when each test is run and a report is being made.
    """
    outcome = yield
    report = outcome.get_result()

    if call.when == "call":
        if hasattr(item.config, "fixture_path_absolute") and hasattr(
            item.config, "fixture_path_relative"
        ):
            report.user_properties.append(
                ("fixture_path_absolute", item.config.fixture_path_absolute)
            )
            report.user_properties.append(
                ("fixture_path_relative", item.config.fixture_path_relative)
            )
        if hasattr(item.config, "evm_dump_dir") and hasattr(item.config, "fixture_format"):
            if item.config.fixture_format in [
                "state_test",
                "blockchain_test",
                "blockchain_test_engine",
            ]:
                report.user_properties.append(("evm_dump_dir", item.config.evm_dump_dir))
            else:
                report.user_properties.append(("evm_dump_dir", "N/A"))  # not yet for EOF

pytest_html_report_title(report)

Set the HTML report title (pytest-html plugin).

Source code in src/pytest_plugins/filler/filler.py 
533
534
535
def pytest_html_report_title(report):
    """Set the HTML report title (pytest-html plugin)."""
    report.title = "Fill Test Report"

evm_bin(request)

Return configured evm tool binary path used to run t8n.

Source code in src/pytest_plugins/filler/filler.py 
538
539
540
541
@pytest.fixture(autouse=True, scope="session")
def evm_bin(request: pytest.FixtureRequest) -> Path:
    """Return configured evm tool binary path used to run t8n."""
    return request.config.getoption("evm_bin")

verify_fixtures_bin(request)

Return configured evm tool binary path used to run statetest or blocktest.

Source code in src/pytest_plugins/filler/filler.py 
544
545
546
547
548
549
550
@pytest.fixture(autouse=True, scope="session")
def verify_fixtures_bin(request: pytest.FixtureRequest) -> Path | None:
    """
    Return configured evm tool binary path used to run statetest or
    blocktest.
    """
    return request.config.getoption("verify_fixtures_bin")

t8n(request, evm_bin)

Return configured transition tool.

Source code in src/pytest_plugins/filler/filler.py 
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
@pytest.fixture(autouse=True, scope="session")
def t8n(request: pytest.FixtureRequest, evm_bin: Path) -> Generator[TransitionTool, None, None]:
    """Return configured transition tool."""
    t8n = TransitionTool.from_binary_path(
        binary_path=evm_bin, trace=request.config.getoption("evm_collect_traces")
    )
    if not t8n.exception_mapper.reliable:
        warnings.warn(
            f"The t8n tool that is currently being used to fill tests ({t8n.__class__.__name__}) "
            "does not provide reliable exception messages. This may lead to false positives when "
            "writing tests and extra care should be taken when writing tests that produce "
            "exceptions.",
            stacklevel=2,
        )
    yield t8n
    t8n.shutdown()

do_fixture_verification(request, verify_fixtures_bin)

Return True if evm statetest or evm blocktest should be ran on the generated fixture JSON files.

Source code in src/pytest_plugins/filler/filler.py 
571
572
573
574
575
576
577
578
579
580
581
582
583
584
@pytest.fixture(scope="session")
def do_fixture_verification(
    request: pytest.FixtureRequest, verify_fixtures_bin: Path | None
) -> bool:
    """
    Return True if evm statetest or evm blocktest should be ran on the
    generated fixture JSON files.
    """
    do_fixture_verification = False
    if verify_fixtures_bin:
        do_fixture_verification = True
    if request.config.getoption("verify_fixtures"):
        do_fixture_verification = True
    return do_fixture_verification

evm_fixture_verification(request, do_fixture_verification, evm_bin, verify_fixtures_bin)

Return configured evm binary for executing statetest and blocktest commands used to verify generated JSON fixtures.

Source code in src/pytest_plugins/filler/filler.py 
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
@pytest.fixture(autouse=True, scope="session")
def evm_fixture_verification(
    request: pytest.FixtureRequest,
    do_fixture_verification: bool,
    evm_bin: Path,
    verify_fixtures_bin: Path | None,
) -> Generator[FixtureConsumer | None, None, None]:
    """
    Return configured evm binary for executing statetest and blocktest
    commands used to verify generated JSON fixtures.
    """
    if not do_fixture_verification:
        yield None
        return
    reused_evm_bin = False
    if not verify_fixtures_bin and evm_bin:
        verify_fixtures_bin = evm_bin
        reused_evm_bin = True
    if not verify_fixtures_bin:
        return
    try:
        evm_fixture_verification = FixtureConsumerTool.from_binary_path(
            binary_path=Path(verify_fixtures_bin),
            trace=request.config.getoption("evm_collect_traces"),
        )
    except Exception:
        if reused_evm_bin:
            pytest.exit(
                "The binary specified in --evm-bin could not be recognized as a known "
                "FixtureConsumerTool. Either remove --verify-fixtures or set "
                "--verify-fixtures-bin to a known fixture consumer binary.",
                returncode=pytest.ExitCode.USAGE_ERROR,
            )
        else:
            pytest.exit(
                "Specified binary in --verify-fixtures-bin could not be recognized as a known "
                "FixtureConsumerTool. Please see `GethFixtureConsumer` for an example "
                "of how a new fixture consumer can be defined.",
                returncode=pytest.ExitCode.USAGE_ERROR,
            )
    yield evm_fixture_verification

base_dump_dir(request)

Path to base directory to dump the evm debug output.

Source code in src/pytest_plugins/filler/filler.py 
630
631
632
633
634
635
636
637
638
@pytest.fixture(scope="session")
def base_dump_dir(request: pytest.FixtureRequest) -> Path | None:
    """Path to base directory to dump the evm debug output."""
    if request.config.getoption("skip_dump_dir"):
        return None
    base_dump_dir_str = request.config.getoption("base_dump_dir")
    if base_dump_dir_str:
        return Path(base_dump_dir_str)
    return None

fixture_output(request)

Return the fixture output configuration.

Source code in src/pytest_plugins/filler/filler.py 
641
642
643
644
@pytest.fixture(scope="session")
def fixture_output(request: pytest.FixtureRequest) -> FixtureOutput:
    """Return the fixture output configuration."""
    return request.config.fixture_output  # type: ignore[attr-defined]

is_output_tarball(fixture_output)

Return True if the output directory is a tarball.

Source code in src/pytest_plugins/filler/filler.py 
647
648
649
650
@pytest.fixture(scope="session")
def is_output_tarball(fixture_output: FixtureOutput) -> bool:
    """Return True if the output directory is a tarball."""
    return fixture_output.is_tarball

output_dir(fixture_output)

Return directory to store the generated test fixtures.

Source code in src/pytest_plugins/filler/filler.py 
653
654
655
656
@pytest.fixture(scope="session")
def output_dir(fixture_output: FixtureOutput) -> Path:
    """Return directory to store the generated test fixtures."""
    return fixture_output.directory

create_properties_file(request, fixture_output)

Create ini file with fixture build properties in the fixture output directory.

Source code in src/pytest_plugins/filler/filler.py 
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
@pytest.fixture(scope="session", autouse=True)
def create_properties_file(request: pytest.FixtureRequest, fixture_output: FixtureOutput) -> None:
    """
    Create ini file with fixture build properties in the fixture output
    directory.
    """
    if fixture_output.is_stdout:
        return

    fixture_properties = {
        "timestamp": datetime.datetime.now().isoformat(),
    }
    if build_name := request.config.getoption("build_name"):
        fixture_properties["build"] = build_name
    if github_ref := os.getenv("GITHUB_REF"):
        fixture_properties["ref"] = github_ref
    if github_sha := os.getenv("GITHUB_SHA"):
        fixture_properties["commit"] = github_sha
    command_line_args = request.config.stash[metadata_key]["Command-line args"]
    command_line_args = command_line_args.replace("<code>", "").replace("</code>", "")
    fixture_properties["command_line_args"] = command_line_args

    config = configparser.ConfigParser(interpolation=None)
    config["fixtures"] = fixture_properties
    environment_properties = {}
    for key, val in request.config.stash[metadata_key].items():
        if key.lower() == "command-line args":
            continue
        if key.lower() in ["ci", "python", "platform"]:
            environment_properties[key] = val
        elif isinstance(val, dict):
            config[key.lower()] = val
        else:
            warnings.warn(
                f"Fixtures ini file: Skipping metadata key {key} with value {val}.", stacklevel=2
            )
    config["environment"] = environment_properties

    ini_filename = fixture_output.metadata_dir / "fixtures.ini"
    with open(ini_filename, "w") as f:
        f.write("; This file describes fixture build properties\n\n")
        config.write(f)

dump_dir_parameter_level(request, base_dump_dir, filler_path)

Directory to dump evm transition tool debug output on a test parameter level.

Example with --evm-dump-dir=/tmp/evm: -> /tmp/evm/shanghai__eip3855_push0__test_push0__test_push0_key_sstore/fork_shanghai/

Source code in src/pytest_plugins/filler/filler.py 
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
@pytest.fixture(scope="function")
def dump_dir_parameter_level(
    request: pytest.FixtureRequest, base_dump_dir: Path | None, filler_path: Path
) -> Path | None:
    """
    Directory to dump evm transition tool debug output on a test parameter
    level.

    Example with --evm-dump-dir=/tmp/evm:
    -> /tmp/evm/shanghai__eip3855_push0__test_push0__test_push0_key_sstore/fork_shanghai/
    """
    evm_dump_dir = node_to_test_info(request.node).get_dump_dir_path(
        base_dump_dir,
        filler_path,
        level="test_parameter",
    )
    # NOTE: Use str for compatibility with pytest-dist
    if evm_dump_dir:
        request.node.config.evm_dump_dir = str(evm_dump_dir)
    else:
        request.node.config.evm_dump_dir = None
    return evm_dump_dir

get_fixture_collection_scope(fixture_name, config)

Return the appropriate scope to write fixture JSON files.

See: https://docs.pytest.org/en/stable/how-to/fixtures.html#dynamic-scope

Source code in src/pytest_plugins/filler/filler.py 
727
728
729
730
731
732
733
734
735
736
737
def get_fixture_collection_scope(fixture_name, config):
    """
    Return the appropriate scope to write fixture JSON files.

    See: https://docs.pytest.org/en/stable/how-to/fixtures.html#dynamic-scope
    """
    if config.fixture_output.is_stdout:
        return "session"
    if config.fixture_output.single_fixture_per_file:
        return "function"
    return "module"

reference_spec(request)

Pytest fixture that returns the reference spec defined in a module.

See get_ref_spec_from_module.

Source code in src/pytest_plugins/filler/filler.py 
740
741
742
743
744
745
746
747
748
749
@pytest.fixture(autouse=True, scope="module")
def reference_spec(request) -> None | ReferenceSpec:
    """
    Pytest fixture that returns the reference spec defined in a module.

    See `get_ref_spec_from_module`.
    """
    if hasattr(request, "module"):
        return get_ref_spec_from_module(request.module)
    return None

fixture_collector(request, do_fixture_verification, evm_fixture_verification, filler_path, base_dump_dir, fixture_output)

Return configured fixture collector instance used for all tests in one test module.

Source code in src/pytest_plugins/filler/filler.py 
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
@pytest.fixture(scope=get_fixture_collection_scope)
def fixture_collector(
    request: pytest.FixtureRequest,
    do_fixture_verification: bool,
    evm_fixture_verification: FixtureConsumer,
    filler_path: Path,
    base_dump_dir: Path | None,
    fixture_output: FixtureOutput,
) -> Generator[FixtureCollector, None, None]:
    """
    Return configured fixture collector instance used for all tests
    in one test module.
    """
    fixture_collector = FixtureCollector(
        output_dir=fixture_output.directory,
        flat_output=fixture_output.flat_output,
        fill_static_tests=request.config.getoption("fill_static_tests_enabled"),
        single_fixture_per_file=fixture_output.single_fixture_per_file,
        filler_path=filler_path,
        base_dump_dir=base_dump_dir,
    )
    yield fixture_collector
    fixture_collector.dump_fixtures()
    if do_fixture_verification:
        fixture_collector.verify_fixture_files(evm_fixture_verification)

filler_path(request)

Return directory containing the tests to execute.

Source code in src/pytest_plugins/filler/filler.py 
779
780
781
782
@pytest.fixture(autouse=True, scope="session")
def filler_path(request: pytest.FixtureRequest) -> Path:
    """Return directory containing the tests to execute."""
    return request.config.getoption("filler_path")

node_to_test_info(node)

Return test info of the current node item.

Source code in src/pytest_plugins/filler/filler.py 
785
786
787
788
789
790
791
792
def node_to_test_info(node: pytest.Item) -> TestInfo:
    """Return test info of the current node item."""
    return TestInfo(
        name=node.name,
        id=node.nodeid,
        original_name=node.originalname,  # type: ignore
        module_path=Path(node.path),
    )

commit_hash_or_tag()

Cache the git commit hash or tag for the entire test session.

Source code in src/pytest_plugins/filler/filler.py 
795
796
797
798
@pytest.fixture(scope="session")
def commit_hash_or_tag() -> str:
    """Cache the git commit hash or tag for the entire test session."""
    return get_current_commit_hash_or_tag()

fixture_source_url(request, commit_hash_or_tag)

Return URL to the fixture source.

Source code in src/pytest_plugins/filler/filler.py 
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
@pytest.fixture(scope="function")
def fixture_source_url(
    request: pytest.FixtureRequest,
    commit_hash_or_tag: str,
) -> str:
    """Return URL to the fixture source."""
    if hasattr(request.node, "github_url"):
        return request.node.github_url
    function_line_number = request.function.__code__.co_firstlineno
    module_relative_path = os.path.relpath(request.function.__code__.co_filename)

    github_url = generate_github_url(
        module_relative_path,
        branch_or_commit_or_tag=commit_hash_or_tag,
        line_number=function_line_number,
    )
    test_module_relative_path = os.path.relpath(request.module.__file__)
    if module_relative_path != test_module_relative_path:
        # This can be the case when the test function's body only contains pass and the entire
        # test logic is implemented as a test generator from the framework.
        test_module_github_url = generate_github_url(
            test_module_relative_path,
            branch_or_commit_or_tag=commit_hash_or_tag,
        )
        github_url += f" called via `{request.node.originalname}()` in {test_module_github_url}"
    return github_url

base_test_parametrizer(cls)

Generate pytest.fixture for a given BaseTest subclass.

Implementation detail: All spec fixtures must be scoped on test function level to avoid leakage between tests.

Source code in src/pytest_plugins/filler/filler.py 
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
def base_test_parametrizer(cls: Type[BaseTest]):
    """
    Generate pytest.fixture for a given BaseTest subclass.

    Implementation detail: All spec fixtures must be scoped on test function level to avoid
    leakage between tests.
    """

    @pytest.fixture(
        scope="function",
        name=cls.pytest_parameter_name(),
    )
    def base_test_parametrizer_func(
        request: pytest.FixtureRequest,
        t8n: TransitionTool,
        fork: Fork,
        reference_spec: ReferenceSpec,
        pre: Alloc,
        output_dir: Path,
        dump_dir_parameter_level: Path | None,
        fixture_collector: FixtureCollector,
        test_case_description: str,
        fixture_source_url: str,
    ):
        """
        Fixture used to instantiate an auto-fillable BaseTest object from within
        a test function.

        Every test that defines a test filler must explicitly specify its parameter name
        (see `pytest_parameter_name` in each implementation of BaseTest) in its function
        arguments.

        When parametrize, indirect must be used along with the fixture format as value.
        """
        if hasattr(request.node, "fixture_format"):
            fixture_format = request.node.fixture_format
        else:
            fixture_format = request.param
        assert issubclass(fixture_format, BaseFixture)
        if fork is None:
            assert hasattr(request.node, "fork")
            fork = request.node.fork

        class BaseTestWrapper(cls):  # type: ignore
            def __init__(self, *args, **kwargs):
                kwargs["t8n_dump_dir"] = dump_dir_parameter_level
                if "pre" not in kwargs:
                    kwargs["pre"] = pre
                super(BaseTestWrapper, self).__init__(*args, **kwargs)
                self._request = request

                # Phase 1: Generate shared pre-state
                if fixture_format is BlockchainEngineReorgFixture and request.config.getoption(
                    "generate_shared_pre"
                ):
                    self.update_shared_pre_state(
                        request.config.shared_pre_state, fork, request.node.nodeid
                    )
                    return  # Skip fixture generation in phase 1

                # Phase 2: Use shared pre-state (only for BlockchainEngineReorgFixture)
                pre_alloc_hash = None
                if fixture_format is BlockchainEngineReorgFixture and request.config.getoption(
                    "use_shared_pre"
                ):
                    pre_alloc_hash = self.compute_shared_pre_alloc_hash(fork=fork)
                    if pre_alloc_hash not in request.config.shared_pre_state:
                        pre_alloc_path = (
                            request.config.fixture_output.shared_pre_alloc_folder_path
                            / pre_alloc_hash
                        )
                        raise ValueError(
                            f"Pre-allocation hash {pre_alloc_hash} not found in shared pre-state. "
                            f"Please check the shared pre-state file at: {pre_alloc_path}. "
                            "Make sure phase 1 (--generate-shared-pre) was run before phase 2."
                        )
                    group: SharedPreStateGroup = request.config.shared_pre_state[pre_alloc_hash]
                    self.pre = group.pre

                fixture = self.generate(
                    t8n=t8n,
                    fork=fork,
                    fixture_format=fixture_format,
                )

                # Post-process for reorg format (add pre_hash and state diff)
                if (
                    fixture_format is BlockchainEngineReorgFixture
                    and request.config.getoption("use_shared_pre")
                    and pre_alloc_hash is not None
                ):
                    fixture.pre_hash = pre_alloc_hash

                    # Calculate state diff for efficiency
                    if hasattr(fixture, "post_state") and fixture.post_state is not None:
                        group = request.config.shared_pre_state[pre_alloc_hash]
                        fixture.post_state_diff = calculate_post_state_diff(
                            fixture.post_state, group.pre
                        )

                fixture.fill_info(
                    t8n.version(),
                    test_case_description,
                    fixture_source_url=fixture_source_url,
                    ref_spec=reference_spec,
                    _info_metadata=t8n._info_metadata,
                )

                fixture_path = fixture_collector.add_fixture(
                    node_to_test_info(request.node),
                    fixture,
                )

                # NOTE: Use str for compatibility with pytest-dist
                request.node.config.fixture_path_absolute = str(fixture_path.absolute())
                request.node.config.fixture_path_relative = str(
                    fixture_path.relative_to(output_dir)
                )
                request.node.config.fixture_format = fixture_format.format_name

        return BaseTestWrapper

    return base_test_parametrizer_func

pytest_generate_tests(metafunc)

Pytest hook used to dynamically generate test cases for each fixture format a given test spec supports.

Source code in src/pytest_plugins/filler/filler.py 
 960
 961
 962
 963
 964
 965
 966
 967
 968
 969
 970
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
def pytest_generate_tests(metafunc: pytest.Metafunc):
    """
    Pytest hook used to dynamically generate test cases for each fixture format a given
    test spec supports.
    """
    for test_type in BaseTest.spec_types.values():
        if test_type.pytest_parameter_name() in metafunc.fixturenames:
            generate_shared_pre = metafunc.config.getoption("generate_shared_pre", False)
            use_shared_pre = metafunc.config.getoption("use_shared_pre", False)

            if generate_shared_pre or use_shared_pre:
                # When shared alloc flags are set, only generate BlockchainEngineReorgFixture
                supported_formats = [
                    format_item
                    for format_item in test_type.supported_fixture_formats
                    if (
                        format_item is BlockchainEngineReorgFixture
                        or (
                            isinstance(format_item, LabeledFixtureFormat)
                            and format_item.format is BlockchainEngineReorgFixture
                        )
                    )
                ]
            else:
                # Filter out BlockchainEngineReorgFixture if shared alloc flags not set
                supported_formats = [
                    format_item
                    for format_item in test_type.supported_fixture_formats
                    if not (
                        format_item is BlockchainEngineReorgFixture
                        or (
                            isinstance(format_item, LabeledFixtureFormat)
                            and format_item.format is BlockchainEngineReorgFixture
                        )
                    )
                ]

            parameters = []
            for i, format_with_or_without_label in enumerate(supported_formats):
                parameter = labeled_format_parameter_set(format_with_or_without_label)
                if i > 0:
                    parameter.marks.append(pytest.mark.derived_test)  # type: ignore
                parameters.append(parameter)
            metafunc.parametrize(
                [test_type.pytest_parameter_name()],
                parameters,
                scope="function",
                indirect=True,
            )

pytest_collection_modifyitems(config, items)

Remove pre-Paris tests parametrized to generate hive type fixtures; these can't be used in the Hive Pyspec Simulator.

Replaces the test ID for state tests that use a transition fork with the base fork.

These can't be handled in this plugins pytest_generate_tests() as the fork parametrization occurs in the forks plugin.

Source code in src/pytest_plugins/filler/filler.py 
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
def pytest_collection_modifyitems(
    config: pytest.Config, items: List[pytest.Item | pytest.Function]
):
    """
    Remove pre-Paris tests parametrized to generate hive type fixtures; these
    can't be used in the Hive Pyspec Simulator.

    Replaces the test ID for state tests that use a transition fork with the base fork.

    These can't be handled in this plugins pytest_generate_tests() as the fork
    parametrization occurs in the forks plugin.
    """
    for item in items[:]:  # use a copy of the list, as we'll be modifying it
        params: Dict[str, Any] | None = None
        if isinstance(item, pytest.Function):
            params = item.callspec.params
        elif hasattr(item, "params"):
            params = item.params
        if not params or "fork" not in params or params["fork"] is None:
            items.remove(item)
            continue
        fork: Fork = params["fork"]
        spec_type, fixture_format = get_spec_format_for_item(params)
        if isinstance(fixture_format, NotSetType):
            items.remove(item)
            continue
        assert issubclass(fixture_format, BaseFixture)
        if not fixture_format.supports_fork(fork):
            items.remove(item)
            continue
        markers = list(item.iter_markers())
        if spec_type.discard_fixture_format_by_marks(fixture_format, fork, markers):
            items.remove(item)
            continue
        for marker in markers:
            if marker.name == "fill":
                for mark in marker.args:
                    item.add_marker(mark)
        if "yul" in item.fixturenames:  # type: ignore
            item.add_marker(pytest.mark.yul_test)

        # Update test ID for state tests that use a transition fork
        if fork in get_transition_forks():
            has_state_test = any(marker.name == "state_test" for marker in markers)
            has_valid_transition = any(
                marker.name == "valid_at_transition_to" for marker in markers
            )
            if has_state_test and has_valid_transition:
                base_fork = get_transition_fork_predecessor(fork)
                item._nodeid = item._nodeid.replace(
                    f"fork_{fork.name()}",
                    f"fork_{base_fork.name()}",
                )

pytest_sessionfinish(session, exitstatus)

Perform session finish tasks.

  • Save shared pre-allocation state (phase 1)
  • Remove any lock files that may have been created.
  • Generate index file for all produced fixtures.
  • Create tarball of the output directory if the output is a tarball.
Source code in src/pytest_plugins/filler/filler.py 
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
def pytest_sessionfinish(session: pytest.Session, exitstatus: int):
    """
    Perform session finish tasks.

    - Save shared pre-allocation state (phase 1)
    - Remove any lock files that may have been created.
    - Generate index file for all produced fixtures.
    - Create tarball of the output directory if the output is a tarball.
    """
    # Save shared pre-state after phase 1
    fixture_output = session.config.fixture_output  # type: ignore[attr-defined]
    if session.config.getoption("generate_shared_pre") and hasattr(
        session.config, "shared_pre_state"
    ):
        shared_pre_alloc_folder = fixture_output.shared_pre_alloc_folder_path
        shared_pre_alloc_folder.mkdir(parents=True, exist_ok=True)
        session.config.shared_pre_state.to_folder(shared_pre_alloc_folder)
        return

    if xdist.is_xdist_worker(session):
        return

    if fixture_output.is_stdout or is_help_or_collectonly_mode(session.config):
        return

    # Remove any lock files that may have been created.
    for file in fixture_output.directory.rglob("*.lock"):
        file.unlink()

    # Generate index file for all produced fixtures.
    if session.config.getoption("generate_index") and not session.config.getoption(
        "generate_shared_pre"
    ):
        generate_fixtures_index(
            fixture_output.directory, quiet_mode=True, force_flag=False, disable_infer_format=False
        )

    # Create tarball of the output directory if the output is a tarball.
    fixture_output.create_tarball()

Pre-alloc specifically conditioned for test filling.

pytest_addoption(parser)

Add command-line options to pytest.

Source code in src/pytest_plugins/filler/pre_alloc.py 
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
def pytest_addoption(parser: pytest.Parser):
    """Add command-line options to pytest."""
    pre_alloc_group = parser.getgroup(
        "pre_alloc", "Arguments defining pre-allocation behavior during test filling."
    )

    pre_alloc_group.addoption(
        "--strict-alloc",
        action="store_true",
        dest="strict_alloc",
        default=False,
        help=("[DEBUG ONLY] Disallows deploying a contract in a predefined address."),
    )
    pre_alloc_group.addoption(
        "--ca-start",
        "--contract-address-start",
        action="store",
        dest="test_contract_start_address",
        default=f"{CONTRACT_START_ADDRESS_DEFAULT}",
        type=str,
        help="The starting address from which tests will deploy contracts.",
    )
    pre_alloc_group.addoption(
        "--ca-incr",
        "--contract-address-increment",
        action="store",
        dest="test_contract_address_increments",
        default=f"{CONTRACT_ADDRESS_INCREMENTS_DEFAULT}",
        type=str,
        help="The address increment value to each deployed contract by a test.",
    )
    pre_alloc_group.addoption(
        "--evm-code-type",
        action="store",
        dest="evm_code_type",
        default=None,
        type=EVMCodeType,
        choices=list(EVMCodeType),
        help="Type of EVM code to deploy in each test by default.",
    )

AllocMode

Bases: IntEnum

Allocation mode for the state.

Source code in src/pytest_plugins/filler/pre_alloc.py 
79
80
81
82
83
class AllocMode(IntEnum):
    """Allocation mode for the state."""

    PERMISSIVE = 0
    STRICT = 1

Alloc

Bases: Alloc

Allocation of accounts in the state, pre and post test execution.

Source code in src/pytest_plugins/filler/pre_alloc.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
class Alloc(BaseAlloc):
    """Allocation of accounts in the state, pre and post test execution."""

    _alloc_mode: AllocMode = PrivateAttr()
    _contract_address_iterator: Iterator[Address] = PrivateAttr()
    _eoa_iterator: Iterator[EOA] = PrivateAttr()
    _evm_code_type: EVMCodeType | None = PrivateAttr(None)

    def __init__(
        self,
        *args,
        alloc_mode: AllocMode,
        contract_address_iterator: Iterator[Address],
        eoa_iterator: Iterator[EOA],
        evm_code_type: EVMCodeType | None = None,
        **kwargs,
    ):
        """Initialize allocation with the given properties."""
        super().__init__(*args, **kwargs)
        self._alloc_mode = alloc_mode
        self._contract_address_iterator = contract_address_iterator
        self._eoa_iterator = eoa_iterator
        self._evm_code_type = evm_code_type

    def __setitem__(self, address: Address | FixedSizeBytesConvertible, account: Account | None):
        """Set account associated with an address."""
        if self._alloc_mode == AllocMode.STRICT:
            raise ValueError("Cannot set items in strict mode")
        super().__setitem__(address, account)

    def code_pre_processor(
        self, code: BytesConvertible, *, evm_code_type: EVMCodeType | None
    ) -> BytesConvertible:
        """Pre-processes the code before setting it."""
        if evm_code_type is None:
            evm_code_type = self._evm_code_type
        if evm_code_type == EVMCodeType.EOF_V1:
            if not isinstance(code, Container):
                if isinstance(code, Bytecode) and not code.terminating:
                    return Container.Code(code + Opcodes.STOP)
                return Container.Code(code)
        return code

    def deploy_contract(
        self,
        code: BytesConvertible,
        *,
        storage: Storage | StorageRootType | None = None,
        balance: NumberConvertible = 0,
        nonce: NumberConvertible = 1,
        address: Address | None = None,
        evm_code_type: EVMCodeType | None = None,
        label: str | None = None,
    ) -> Address:
        """
        Deploy a contract to the allocation.

        Warning: `address` parameter is a temporary solution to allow tests to hard-code the
        contract address. Do NOT use in new tests as it will be removed in the future!
        """
        if storage is None:
            storage = {}
        if address is not None:
            assert self._alloc_mode == AllocMode.PERMISSIVE, "address parameter is not supported"
            assert address not in self, f"address {address} already in allocation"
            contract_address = address
        else:
            contract_address = next(self._contract_address_iterator)

        if self._alloc_mode == AllocMode.STRICT:
            assert Number(nonce) >= 1, "impossible to deploy contract with nonce lower than one"

        super().__setitem__(
            contract_address,
            Account(
                nonce=nonce,
                balance=balance,
                code=self.code_pre_processor(code, evm_code_type=evm_code_type),
                storage=storage,
            ),
        )
        if label is None:
            # Try to deduce the label from the code
            frame = inspect.currentframe()
            if frame is not None:
                caller_frame = frame.f_back
                if caller_frame is not None:
                    code_context = inspect.getframeinfo(caller_frame).code_context
                    if code_context is not None:
                        line = code_context[0].strip()
                        if "=" in line:
                            label = line.split("=")[0].strip()

        contract_address.label = label
        return contract_address

    def fund_eoa(
        self,
        amount: NumberConvertible | None = None,
        label: str | None = None,
        storage: Storage | None = None,
        delegation: Address | Literal["Self"] | None = None,
        nonce: NumberConvertible | None = None,
    ) -> EOA:
        """
        Add a previously unused EOA to the pre-alloc with the balance specified by `amount`.

        If amount is 0, nothing will be added to the pre-alloc but a new and unique EOA will be
        returned.
        """
        eoa = next(self._eoa_iterator)
        if amount is None:
            amount = self._eoa_fund_amount_default
        if (
            Number(amount) > 0
            or storage is not None
            or delegation is not None
            or (nonce is not None and Number(nonce) > 0)
        ):
            if storage is None and delegation is None:
                nonce = Number(0 if nonce is None else nonce)
                account = Account(
                    nonce=nonce,
                    balance=amount,
                )
                if nonce > 0:
                    eoa.nonce = nonce
            else:
                # Type-4 transaction is sent to the EOA to set the storage, so the nonce must be 1
                if not isinstance(delegation, Address) and delegation == "Self":
                    delegation = eoa
                # If delegation is None but storage is not, realistically the nonce should be 2
                # because the account must have delegated to set the storage and then again to
                # reset the delegation (but can be overridden by the test for a non-realistic
                # scenario)
                real_nonce = 2 if delegation is None else 1
                nonce = Number(real_nonce if nonce is None else nonce)
                account = Account(
                    nonce=nonce,
                    balance=amount,
                    storage=storage if storage is not None else {},
                    code=DELEGATION_DESIGNATION + bytes(delegation)  # type: ignore
                    if delegation is not None
                    else b"",
                )
                eoa.nonce = nonce

            super().__setitem__(eoa, account)
        return eoa

    def fund_address(self, address: Address, amount: NumberConvertible):
        """
        Fund an address with a given amount.

        If the address is already present in the pre-alloc the amount will be
        added to its existing balance.
        """
        if address in self:
            account = self[address]
            if account is not None:
                current_balance = account.balance or 0
                account.balance = ZeroPaddedHexNumber(current_balance + Number(amount))
                return
        super().__setitem__(address, Account(balance=amount))

    def empty_account(self) -> Address:
        """
        Add a previously unused account guaranteed to be empty to the pre-alloc.

        This ensures the account has:
        - Zero balance
        - Zero nonce
        - No code
        - No storage

        This is different from precompiles or system contracts. The function does not
        send any transactions, ensuring that the account remains "empty."

        Returns:
            Address: The address of the created empty account.

        """
        eoa = next(self._eoa_iterator)

        return Address(eoa)

__init__(*args, alloc_mode, contract_address_iterator, eoa_iterator, evm_code_type=None, **kwargs)

Initialize allocation with the given properties.

Source code in src/pytest_plugins/filler/pre_alloc.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
def __init__(
    self,
    *args,
    alloc_mode: AllocMode,
    contract_address_iterator: Iterator[Address],
    eoa_iterator: Iterator[EOA],
    evm_code_type: EVMCodeType | None = None,
    **kwargs,
):
    """Initialize allocation with the given properties."""
    super().__init__(*args, **kwargs)
    self._alloc_mode = alloc_mode
    self._contract_address_iterator = contract_address_iterator
    self._eoa_iterator = eoa_iterator
    self._evm_code_type = evm_code_type

__setitem__(address, account)

Set account associated with an address.

Source code in src/pytest_plugins/filler/pre_alloc.py 
113
114
115
116
117
def __setitem__(self, address: Address | FixedSizeBytesConvertible, account: Account | None):
    """Set account associated with an address."""
    if self._alloc_mode == AllocMode.STRICT:
        raise ValueError("Cannot set items in strict mode")
    super().__setitem__(address, account)

code_pre_processor(code, *, evm_code_type)

Pre-processes the code before setting it.

Source code in src/pytest_plugins/filler/pre_alloc.py 
119
120
121
122
123
124
125
126
127
128
129
130
def code_pre_processor(
    self, code: BytesConvertible, *, evm_code_type: EVMCodeType | None
) -> BytesConvertible:
    """Pre-processes the code before setting it."""
    if evm_code_type is None:
        evm_code_type = self._evm_code_type
    if evm_code_type == EVMCodeType.EOF_V1:
        if not isinstance(code, Container):
            if isinstance(code, Bytecode) and not code.terminating:
                return Container.Code(code + Opcodes.STOP)
            return Container.Code(code)
    return code

deploy_contract(code, *, storage=None, balance=0, nonce=1, address=None, evm_code_type=None, label=None)

Deploy a contract to the allocation.

Warning: address parameter is a temporary solution to allow tests to hard-code the contract address. Do NOT use in new tests as it will be removed in the future!

Source code in src/pytest_plugins/filler/pre_alloc.py 
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
def deploy_contract(
    self,
    code: BytesConvertible,
    *,
    storage: Storage | StorageRootType | None = None,
    balance: NumberConvertible = 0,
    nonce: NumberConvertible = 1,
    address: Address | None = None,
    evm_code_type: EVMCodeType | None = None,
    label: str | None = None,
) -> Address:
    """
    Deploy a contract to the allocation.

    Warning: `address` parameter is a temporary solution to allow tests to hard-code the
    contract address. Do NOT use in new tests as it will be removed in the future!
    """
    if storage is None:
        storage = {}
    if address is not None:
        assert self._alloc_mode == AllocMode.PERMISSIVE, "address parameter is not supported"
        assert address not in self, f"address {address} already in allocation"
        contract_address = address
    else:
        contract_address = next(self._contract_address_iterator)

    if self._alloc_mode == AllocMode.STRICT:
        assert Number(nonce) >= 1, "impossible to deploy contract with nonce lower than one"

    super().__setitem__(
        contract_address,
        Account(
            nonce=nonce,
            balance=balance,
            code=self.code_pre_processor(code, evm_code_type=evm_code_type),
            storage=storage,
        ),
    )
    if label is None:
        # Try to deduce the label from the code
        frame = inspect.currentframe()
        if frame is not None:
            caller_frame = frame.f_back
            if caller_frame is not None:
                code_context = inspect.getframeinfo(caller_frame).code_context
                if code_context is not None:
                    line = code_context[0].strip()
                    if "=" in line:
                        label = line.split("=")[0].strip()

    contract_address.label = label
    return contract_address

fund_eoa(amount=None, label=None, storage=None, delegation=None, nonce=None)

Add a previously unused EOA to the pre-alloc with the balance specified by amount.

If amount is 0, nothing will be added to the pre-alloc but a new and unique EOA will be returned.

Source code in src/pytest_plugins/filler/pre_alloc.py 
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
def fund_eoa(
    self,
    amount: NumberConvertible | None = None,
    label: str | None = None,
    storage: Storage | None = None,
    delegation: Address | Literal["Self"] | None = None,
    nonce: NumberConvertible | None = None,
) -> EOA:
    """
    Add a previously unused EOA to the pre-alloc with the balance specified by `amount`.

    If amount is 0, nothing will be added to the pre-alloc but a new and unique EOA will be
    returned.
    """
    eoa = next(self._eoa_iterator)
    if amount is None:
        amount = self._eoa_fund_amount_default
    if (
        Number(amount) > 0
        or storage is not None
        or delegation is not None
        or (nonce is not None and Number(nonce) > 0)
    ):
        if storage is None and delegation is None:
            nonce = Number(0 if nonce is None else nonce)
            account = Account(
                nonce=nonce,
                balance=amount,
            )
            if nonce > 0:
                eoa.nonce = nonce
        else:
            # Type-4 transaction is sent to the EOA to set the storage, so the nonce must be 1
            if not isinstance(delegation, Address) and delegation == "Self":
                delegation = eoa
            # If delegation is None but storage is not, realistically the nonce should be 2
            # because the account must have delegated to set the storage and then again to
            # reset the delegation (but can be overridden by the test for a non-realistic
            # scenario)
            real_nonce = 2 if delegation is None else 1
            nonce = Number(real_nonce if nonce is None else nonce)
            account = Account(
                nonce=nonce,
                balance=amount,
                storage=storage if storage is not None else {},
                code=DELEGATION_DESIGNATION + bytes(delegation)  # type: ignore
                if delegation is not None
                else b"",
            )
            eoa.nonce = nonce

        super().__setitem__(eoa, account)
    return eoa

fund_address(address, amount)

Fund an address with a given amount.

If the address is already present in the pre-alloc the amount will be added to its existing balance.

Source code in src/pytest_plugins/filler/pre_alloc.py 
239
240
241
242
243
244
245
246
247
248
249
250
251
252
def fund_address(self, address: Address, amount: NumberConvertible):
    """
    Fund an address with a given amount.

    If the address is already present in the pre-alloc the amount will be
    added to its existing balance.
    """
    if address in self:
        account = self[address]
        if account is not None:
            current_balance = account.balance or 0
            account.balance = ZeroPaddedHexNumber(current_balance + Number(amount))
            return
    super().__setitem__(address, Account(balance=amount))

empty_account()

Add a previously unused account guaranteed to be empty to the pre-alloc.

This ensures the account has: - Zero balance - Zero nonce - No code - No storage

This is different from precompiles or system contracts. The function does not send any transactions, ensuring that the account remains "empty."

Returns:

Name Type Description
Address Address

The address of the created empty account.
Source code in src/pytest_plugins/filler/pre_alloc.py 
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
def empty_account(self) -> Address:
    """
    Add a previously unused account guaranteed to be empty to the pre-alloc.

    This ensures the account has:
    - Zero balance
    - Zero nonce
    - No code
    - No storage

    This is different from precompiles or system contracts. The function does not
    send any transactions, ensuring that the account remains "empty."

    Returns:
        Address: The address of the created empty account.

    """
    eoa = next(self._eoa_iterator)

    return Address(eoa)

alloc_mode(request)

Return allocation mode for the tests.

Source code in src/pytest_plugins/filler/pre_alloc.py 
276
277
278
279
280
281
@pytest.fixture(scope="session")
def alloc_mode(request: pytest.FixtureRequest) -> AllocMode:
    """Return allocation mode for the tests."""
    if request.config.getoption("strict_alloc"):
        return AllocMode.STRICT
    return AllocMode.PERMISSIVE

contract_start_address(request)

Return starting address for contract deployment.

Source code in src/pytest_plugins/filler/pre_alloc.py 
284
285
286
287
@pytest.fixture(scope="session")
def contract_start_address(request: pytest.FixtureRequest) -> int:
    """Return starting address for contract deployment."""
    return int(request.config.getoption("test_contract_start_address"), 0)

contract_address_increments(request)

Return address increment for contract deployment.

Source code in src/pytest_plugins/filler/pre_alloc.py 
290
291
292
293
@pytest.fixture(scope="session")
def contract_address_increments(request: pytest.FixtureRequest) -> int:
    """Return address increment for contract deployment."""
    return int(request.config.getoption("test_contract_address_increments"), 0)

sha256_from_string(s)

Return SHA-256 hash of a string.

Source code in src/pytest_plugins/filler/pre_alloc.py 
296
297
298
def sha256_from_string(s: str) -> int:
    """Return SHA-256 hash of a string."""
    return int.from_bytes(sha256(s.encode("utf-8")).digest(), "big")

contract_address_iterator(request, contract_start_address, contract_address_increments)

Return iterator over contract addresses with dynamic scoping.

Source code in src/pytest_plugins/filler/pre_alloc.py 
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
@pytest.fixture(scope="function")
def contract_address_iterator(
    request: pytest.FixtureRequest,
    contract_start_address: int,
    contract_address_increments: int,
) -> Iterator[Address]:
    """Return iterator over contract addresses with dynamic scoping."""
    if request.config.getoption("generate_shared_pre", default=False) or request.config.getoption(
        "use_shared_pre", default=False
    ):
        # Use a starting address that is derived from the test node
        contract_start_address = sha256_from_string(request.node.nodeid)
    return iter(
        Address((contract_start_address + (i * contract_address_increments)) % 2**160)
        for i in count()
    )

eoa_by_index(i) cached

Return EOA by index.

Source code in src/pytest_plugins/filler/pre_alloc.py 
319
320
321
322
@cache
def eoa_by_index(i: int) -> EOA:
    """Return EOA by index."""
    return EOA(key=TestPrivateKey + i if i != 1 else TestPrivateKey2, nonce=0)

eoa_iterator(request)

Return iterator over EOAs copies with dynamic scoping.

Source code in src/pytest_plugins/filler/pre_alloc.py 
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
@pytest.fixture(scope="function")
def eoa_iterator(request: pytest.FixtureRequest) -> Iterator[EOA]:
    """Return iterator over EOAs copies with dynamic scoping."""
    if request.config.getoption("generate_shared_pre", default=False) or request.config.getoption(
        "use_shared_pre", default=False
    ):
        # Use a starting address that is derived from the test node
        eoa_start_pk = sha256_from_string(request.node.nodeid)
        return iter(
            EOA(
                key=(eoa_start_pk + i)
                % 0xFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFEBAAEDCE6AF48A03BBFD25E8CD0364141,
                nonce=0,
            )
            for i in count()
        )
    return iter(eoa_by_index(i).copy() for i in count())

evm_code_type(request)

Return default EVM code type for all tests (LEGACY).

Source code in src/pytest_plugins/filler/pre_alloc.py 
344
345
346
347
348
349
350
351
@pytest.fixture(autouse=True)
def evm_code_type(request: pytest.FixtureRequest) -> EVMCodeType:
    """Return default EVM code type for all tests (LEGACY)."""
    parameter_evm_code_type = request.config.getoption("evm_code_type")
    if parameter_evm_code_type is not None:
        assert type(parameter_evm_code_type) is EVMCodeType, "Invalid EVM code type"
        return parameter_evm_code_type
    return EVMCodeType.LEGACY

pre(alloc_mode, contract_address_iterator, eoa_iterator, evm_code_type)

Return default pre allocation for all tests (Empty alloc).

Source code in src/pytest_plugins/filler/pre_alloc.py 
354
355
356
357
358
359
360
361
362
363
364
365
366
367
@pytest.fixture(scope="function")
def pre(
    alloc_mode: AllocMode,
    contract_address_iterator: Iterator[Address],
    eoa_iterator: Iterator[EOA],
    evm_code_type: EVMCodeType,
) -> Alloc:
    """Return default pre allocation for all tests (Empty alloc)."""
    return Alloc(
        alloc_mode=alloc_mode,
        contract_address_iterator=contract_address_iterator,
        eoa_iterator=eoa_iterator,
        evm_code_type=evm_code_type,
    )