Filler Plugin

A pytest plugin to fill tests and generate JSON fixtures.

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.

PhaseManager dataclass

Manages the execution phase for fixture generation.

The filler plugin supports two-phase execution for pre-allocation group generation: - Phase 1: Generate pre-allocation groups (pytest run with --generate-pre-alloc-groups).

• Phase 2: Fill fixtures using pre-allocation groups (pytest run with --use-pre-alloc-groups).

Note: These are separate pytest runs orchestrated by the CLI wrapper. Each run gets a fresh PhaseManager instance (no persistence between phases).

Source code in packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

@dataclass(kw_only=True)
class PhaseManager:
    """
    Manages the execution phase for fixture generation.

    The filler plugin supports two-phase execution for pre-allocation group
    generation:
    - Phase 1: Generate pre-allocation groups (pytest run with
        --generate-pre-alloc-groups).

    - Phase 2: Fill fixtures using pre-allocation
        groups (pytest run with --use-pre-alloc-groups).

    Note: These are separate pytest runs orchestrated by the CLI wrapper. Each
    run gets a fresh PhaseManager instance (no persistence between phases).
    """

    current_phase: FixtureFillingPhase
    previous_phases: Set[FixtureFillingPhase] = field(default_factory=set)

    @classmethod
    def from_config(cls, config: pytest.Config) -> "Self":
        """
        Create a PhaseManager from pytest configuration.

        Flag logic:
        - use_pre_alloc_groups: We're in phase 2 (FILL) after phase
                                1 (PRE_ALLOC_GENERATION).
        - generate_pre_alloc_groups or generate_all_formats:
                                We're in phase 1 (PRE_ALLOC_GENERATION). -
        Otherwise: Normal single-phase filling (FILL).

        Note: generate_all_formats triggers PRE_ALLOC_GENERATION because the
        CLI passes it to phase 1 to ensure all formats are considered for
        grouping.
        """
        generate_pre_alloc = config.getoption(
            "generate_pre_alloc_groups", False
        )
        use_pre_alloc = config.getoption("use_pre_alloc_groups", False)
        generate_all = config.getoption("generate_all_formats", False)

        if use_pre_alloc:
            # Phase 2: Using pre-generated groups
            return cls(
                current_phase=FixtureFillingPhase.FILL,
                previous_phases={FixtureFillingPhase.PRE_ALLOC_GENERATION},
            )
        elif generate_pre_alloc or generate_all:
            # Phase 1: Generating pre-allocation groups
            return cls(current_phase=FixtureFillingPhase.PRE_ALLOC_GENERATION)
        else:
            # Normal single-phase filling
            return cls(current_phase=FixtureFillingPhase.FILL)

    @property
    def is_pre_alloc_generation(self) -> bool:
        """Check if we're in the pre-allocation generation phase."""
        return self.current_phase == FixtureFillingPhase.PRE_ALLOC_GENERATION

    @property
    def is_fill_after_pre_alloc(self) -> bool:
        """Check if we're filling after pre-allocation generation."""
        return (
            self.current_phase == FixtureFillingPhase.FILL
            and FixtureFillingPhase.PRE_ALLOC_GENERATION
            in self.previous_phases
        )

    @property
    def is_single_phase_fill(self) -> bool:
        """Check if we're in single-phase fill mode (no pre-allocation)."""
        return (
            self.current_phase == FixtureFillingPhase.FILL
            and FixtureFillingPhase.PRE_ALLOC_GENERATION
            not in self.previous_phases
        )

from_config(config) classmethod

Create a PhaseManager from pytest configuration.

Flag logic: - use_pre_alloc_groups: We're in phase 2 (FILL) after phase 1 (PRE_ALLOC_GENERATION). - generate_pre_alloc_groups or generate_all_formats: We're in phase 1 (PRE_ALLOC_GENERATION). - Otherwise: Normal single-phase filling (FILL).

Note: generate_all_formats triggers PRE_ALLOC_GENERATION because the CLI passes it to phase 1 to ensure all formats are considered for grouping.

Source code in packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

@classmethod
def from_config(cls, config: pytest.Config) -> "Self":
    """
    Create a PhaseManager from pytest configuration.

    Flag logic:
    - use_pre_alloc_groups: We're in phase 2 (FILL) after phase
                            1 (PRE_ALLOC_GENERATION).
    - generate_pre_alloc_groups or generate_all_formats:
                            We're in phase 1 (PRE_ALLOC_GENERATION). -
    Otherwise: Normal single-phase filling (FILL).

    Note: generate_all_formats triggers PRE_ALLOC_GENERATION because the
    CLI passes it to phase 1 to ensure all formats are considered for
    grouping.
    """
    generate_pre_alloc = config.getoption(
        "generate_pre_alloc_groups", False
    )
    use_pre_alloc = config.getoption("use_pre_alloc_groups", False)
    generate_all = config.getoption("generate_all_formats", False)

    if use_pre_alloc:
        # Phase 2: Using pre-generated groups
        return cls(
            current_phase=FixtureFillingPhase.FILL,
            previous_phases={FixtureFillingPhase.PRE_ALLOC_GENERATION},
        )
    elif generate_pre_alloc or generate_all:
        # Phase 1: Generating pre-allocation groups
        return cls(current_phase=FixtureFillingPhase.PRE_ALLOC_GENERATION)
    else:
        # Normal single-phase filling
        return cls(current_phase=FixtureFillingPhase.FILL)

is_pre_alloc_generation property

Check if we're in the pre-allocation generation phase.

is_fill_after_pre_alloc property

Check if we're filling after pre-allocation generation.

is_single_phase_fill property

Check if we're in single-phase fill mode (no pre-allocation).

FormatSelector dataclass

Handles fixture format selection based on the current phase and format capabilities.

This class encapsulates the complex logic for determining which fixture formats should be generated in each phase of the two-phase execution model.

Source code in packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

@dataclass(kw_only=True)
class FormatSelector:
    """
    Handles fixture format selection based on the current phase and format
    capabilities.

    This class encapsulates the complex logic for determining which fixture
    formats should be generated in each phase of the two-phase execution model.
    """

    phase_manager: PhaseManager
    generate_all_formats: bool

    def should_generate(
        self, fixture_format: Type[BaseFixture] | LabeledFixtureFormat
    ) -> bool:
        """
        Determine if a fixture format should be generated in the current phase.

        Args:
            fixture_format: The fixture format to check (may be wrapped in
                LabeledFixtureFormat).

        Returns:
            True if the format should be generated in the current phase.

        """
        format_phases = fixture_format.format_phases

        if self.phase_manager.is_pre_alloc_generation:
            return self._should_generate_pre_alloc(format_phases)
        else:  # FILL phase
            return self._should_generate_fill(format_phases)

    def _should_generate_pre_alloc(
        self, format_phases: Set[FixtureFillingPhase]
    ) -> bool:
        """
        Determine if format should be generated during pre-alloc generation
        phase.
        """
        # Only generate formats that need pre-allocation groups
        return FixtureFillingPhase.PRE_ALLOC_GENERATION in format_phases

    def _should_generate_fill(
        self, format_phases: Set[FixtureFillingPhase]
    ) -> bool:
        """Determine if format should be generated during fill phase."""
        if (
            FixtureFillingPhase.PRE_ALLOC_GENERATION
            in self.phase_manager.previous_phases
        ):
            # Phase 2: After pre-alloc generation
            if self.generate_all_formats:
                # Generate all formats, including those that don't need pre-
                # alloc
                return True
            else:
                # Only generate formats that needed pre-alloc groups
                return (
                    FixtureFillingPhase.PRE_ALLOC_GENERATION in format_phases
                )
        else:
            # Single phase: Only generate fill-only formats
            return format_phases == {FixtureFillingPhase.FILL}

should_generate(fixture_format)

Determine if a fixture format should be generated in the current phase.

Parameters:

Name	Type	Description	Default
fixture_format	Type[BaseFixture] | LabeledFixtureFormat	The fixture format to check (may be wrapped in LabeledFixtureFormat).	required

Returns:

Type	Description
bool	True if the format should be generated in the current phase.

Source code in packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

def should_generate(
    self, fixture_format: Type[BaseFixture] | LabeledFixtureFormat
) -> bool:
    """
    Determine if a fixture format should be generated in the current phase.

    Args:
        fixture_format: The fixture format to check (may be wrapped in
            LabeledFixtureFormat).

    Returns:
        True if the format should be generated in the current phase.

    """
    format_phases = fixture_format.format_phases

    if self.phase_manager.is_pre_alloc_generation:
        return self._should_generate_pre_alloc(format_phases)
    else:  # FILL phase
        return self._should_generate_fill(format_phases)

FillingSession dataclass

Manages all state for a single pytest fill session.

This class serves as the single source of truth for all filler state management, including phase management, format selection, and pre-allocation groups.

Important: Each pytest run gets a fresh FillingSession instance. There is no persistence between phase 1 (generate pre-alloc) and phase 2 (use pre-alloc) except through file I/O.

Source code in packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

@dataclass(kw_only=True)
class FillingSession:
    """
    Manages all state for a single pytest fill session.

    This class serves as the single source of truth for all filler state
    management, including phase management, format selection, and
    pre-allocation groups.

    Important: Each pytest run gets a fresh FillingSession instance. There is
    no persistence between phase 1 (generate pre-alloc) and phase 2 (use
    pre-alloc) except through file I/O.
    """

    fixture_output: FixtureOutput
    phase_manager: PhaseManager
    format_selector: FormatSelector
    pre_alloc_groups: PreAllocGroups | None = None
    pre_alloc_group_builders: PreAllocGroupBuilders | None = None

    @classmethod
    def from_config(cls, config: pytest.Config) -> "Self":
        """
        Initialize a filling session from pytest configuration.

        Args:
            config: The pytest configuration object.

        """
        phase_manager = PhaseManager.from_config(config)
        instance = cls(
            fixture_output=FixtureOutput.from_config(config),
            phase_manager=phase_manager,
            format_selector=FormatSelector(
                phase_manager=phase_manager,
                generate_all_formats=config.getoption(
                    "generate_all_formats", False
                ),
            ),
            pre_alloc_groups=None,
        )

        # Initialize pre-alloc groups based on phase
        instance._initialize_pre_alloc_groups()
        return instance

    def _initialize_pre_alloc_groups(self) -> None:
        """Initialize pre-allocation groups based on the current phase."""
        if self.phase_manager.is_pre_alloc_generation:
            # Phase 1: Create empty container for collecting groups
            self.pre_alloc_group_builders = PreAllocGroupBuilders(root={})
        elif self.phase_manager.is_fill_after_pre_alloc:
            # Phase 2: Load pre-alloc groups from disk
            self._load_pre_alloc_groups_from_folder()

    def _load_pre_alloc_groups_from_folder(self) -> None:
        """Load pre-allocation groups from the output folder."""
        pre_alloc_folder = self.fixture_output.pre_alloc_groups_folder_path
        if pre_alloc_folder.exists():
            self.pre_alloc_groups = PreAllocGroups.from_folder(
                pre_alloc_folder, lazy_load=True
            )
        else:
            raise FileNotFoundError(
                f"Pre-allocation groups folder not found: {pre_alloc_folder}. "
                "Run phase 1 with --generate-pre-alloc-groups first."
            )

    def should_generate_format(
        self, fixture_format: Type[BaseFixture] | LabeledFixtureFormat
    ) -> bool:
        """
        Determine if a fixture format should be generated in the current
        session.

        Args:
            fixture_format: The fixture format to check.

        Returns:
            True if the format should be generated.

        """
        return self.format_selector.should_generate(fixture_format)

    def get_pre_alloc_group(self, hash_key: str) -> PreAllocGroup:
        """
        Get a pre-allocation group by hash.

        Args:
            hash_key: The hash of the pre-alloc group.

        Returns:
            The pre-allocation group.

        Raises:
            ValueError: If pre-alloc groups not initialized or hash not found.

        """
        if self.pre_alloc_groups is None:
            raise ValueError("Pre-allocation groups not initialized")

        if hash_key not in self.pre_alloc_groups:
            pre_alloc_path = (
                self.fixture_output.pre_alloc_groups_folder_path / hash_key
            )
            raise ValueError(
                f"Pre-allocation hash {hash_key} not found in "
                f"pre-allocation groups. Please check the file at: "
                f"{pre_alloc_path}. Make sure phase 1 "
                "(--generate-pre-alloc-groups) was run before phase 2."
            )

        return self.pre_alloc_groups[hash_key]

    def update_pre_alloc_group_builder(
        self, hash_key: str, group_builder: PreAllocGroupBuilder
    ) -> None:
        """
        Update or add a pre-allocation group.

        Args:
            hash_key: The hash of the pre-alloc group.
            group_builder: The pre-allocation group builder.

        Raises:
            ValueError: If not in pre-alloc generation phase.

        """
        if not self.phase_manager.is_pre_alloc_generation:
            raise ValueError(
                "Can only update pre-alloc groups in generation phase"
            )

        if self.pre_alloc_group_builders is None:
            self.pre_alloc_group_builders = PreAllocGroupBuilders(root={})

        self.pre_alloc_group_builders.root[hash_key] = group_builder

    def save_pre_alloc_groups(self) -> None:
        """Save pre-allocation groups to disk as partial files."""
        if self.pre_alloc_group_builders is None:
            return

        pre_alloc_folder = self.fixture_output.pre_alloc_groups_folder_path
        pre_alloc_folder.mkdir(parents=True, exist_ok=True)
        # Pass worker_id so each worker writes its own partial files
        # (no lock contention). Master merges them after all workers finish.
        self.pre_alloc_group_builders.to_folder(
            pre_alloc_folder, worker_id=_get_worker_id()
        )

from_config(config) classmethod

Initialize a filling session from pytest configuration.

Parameters:

Name	Type	Description	Default
config	Config	The pytest configuration object.	required

Source code in packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

@classmethod
def from_config(cls, config: pytest.Config) -> "Self":
    """
    Initialize a filling session from pytest configuration.

    Args:
        config: The pytest configuration object.

    """
    phase_manager = PhaseManager.from_config(config)
    instance = cls(
        fixture_output=FixtureOutput.from_config(config),
        phase_manager=phase_manager,
        format_selector=FormatSelector(
            phase_manager=phase_manager,
            generate_all_formats=config.getoption(
                "generate_all_formats", False
            ),
        ),
        pre_alloc_groups=None,
    )

    # Initialize pre-alloc groups based on phase
    instance._initialize_pre_alloc_groups()
    return instance

should_generate_format(fixture_format)

Determine if a fixture format should be generated in the current session.

Parameters:

Name	Type	Description	Default
fixture_format	Type[BaseFixture] | LabeledFixtureFormat	The fixture format to check.	required

Returns:

Type	Description
bool	True if the format should be generated.

Source code in packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

def should_generate_format(
    self, fixture_format: Type[BaseFixture] | LabeledFixtureFormat
) -> bool:
    """
    Determine if a fixture format should be generated in the current
    session.

    Args:
        fixture_format: The fixture format to check.

    Returns:
        True if the format should be generated.

    """
    return self.format_selector.should_generate(fixture_format)

get_pre_alloc_group(hash_key)

Get a pre-allocation group by hash.

Parameters:

Name	Type	Description	Default
hash_key	str	The hash of the pre-alloc group.	required

Returns:

Type	Description
PreAllocGroup	The pre-allocation group.

Raises:

Type	Description
ValueError	If pre-alloc groups not initialized or hash not found.

Source code in packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

def get_pre_alloc_group(self, hash_key: str) -> PreAllocGroup:
    """
    Get a pre-allocation group by hash.

    Args:
        hash_key: The hash of the pre-alloc group.

    Returns:
        The pre-allocation group.

    Raises:
        ValueError: If pre-alloc groups not initialized or hash not found.

    """
    if self.pre_alloc_groups is None:
        raise ValueError("Pre-allocation groups not initialized")

    if hash_key not in self.pre_alloc_groups:
        pre_alloc_path = (
            self.fixture_output.pre_alloc_groups_folder_path / hash_key
        )
        raise ValueError(
            f"Pre-allocation hash {hash_key} not found in "
            f"pre-allocation groups. Please check the file at: "
            f"{pre_alloc_path}. Make sure phase 1 "
            "(--generate-pre-alloc-groups) was run before phase 2."
        )

    return self.pre_alloc_groups[hash_key]

update_pre_alloc_group_builder(hash_key, group_builder)

Update or add a pre-allocation group.

Parameters:

Name	Type	Description	Default
hash_key	str	The hash of the pre-alloc group.	required
group_builder	PreAllocGroupBuilder	The pre-allocation group builder.	required

Raises:

Type	Description
ValueError	If not in pre-alloc generation phase.

Source code in packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

def update_pre_alloc_group_builder(
    self, hash_key: str, group_builder: PreAllocGroupBuilder
) -> None:
    """
    Update or add a pre-allocation group.

    Args:
        hash_key: The hash of the pre-alloc group.
        group_builder: The pre-allocation group builder.

    Raises:
        ValueError: If not in pre-alloc generation phase.

    """
    if not self.phase_manager.is_pre_alloc_generation:
        raise ValueError(
            "Can only update pre-alloc groups in generation phase"
        )

    if self.pre_alloc_group_builders is None:
        self.pre_alloc_group_builders = PreAllocGroupBuilders(root={})

    self.pre_alloc_group_builders.root[hash_key] = group_builder

save_pre_alloc_groups()

Save pre-allocation groups to disk as partial files.

Source code in packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

def save_pre_alloc_groups(self) -> None:
    """Save pre-allocation groups to disk as partial files."""
    if self.pre_alloc_group_builders is None:
        return

    pre_alloc_folder = self.fixture_output.pre_alloc_groups_folder_path
    pre_alloc_folder.mkdir(parents=True, exist_ok=True)
    # Pass worker_id so each worker writes its own partial files
    # (no lock contention). Master merges them after all workers finish.
    self.pre_alloc_group_builders.to_folder(
        pre_alloc_folder, worker_id=_get_worker_id()
    )

TransitionToolCacheStats dataclass

Stats for caching of the transition tool requests.

Source code in packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

@dataclass(kw_only=True)
class TransitionToolCacheStats:
    """Stats for caching of the transition tool requests."""

    key_test_hits: int = 0
    key_test_miss: int = 0
    subkey_test_hits: int = 0
    subkey_test_miss: int = 0
    unique_keys: int = 0
    _seen_keys: Set[str] = field(default_factory=set, repr=False)

    def record_key(self, key: str) -> None:
        """Record a cache key and update unique_keys count."""
        self._seen_keys.add(key)
        self.unique_keys = len(self._seen_keys)

    @property
    def expected_hits(self) -> int:
        """Number of tests expected to hit the cache."""
        total_cacheable = self.key_test_hits + self.key_test_miss
        return total_cacheable - self.unique_keys

    def to_dict(self) -> Dict[str, int]:
        """Convert stats to dict for xdist worker transfer."""
        return {
            "key_test_hits": self.key_test_hits,
            "key_test_miss": self.key_test_miss,
            "subkey_test_hits": self.subkey_test_hits,
            "subkey_test_miss": self.subkey_test_miss,
            "unique_keys": self.unique_keys,
        }

    def add(self, other: "TransitionToolCacheStats") -> None:
        """Add another stats object to this one."""
        self.key_test_hits += other.key_test_hits
        self.key_test_miss += other.key_test_miss
        self.subkey_test_hits += other.subkey_test_hits
        self.subkey_test_miss += other.subkey_test_miss
        self.unique_keys += other.unique_keys

    @classmethod
    def from_dict(cls, data: Dict[str, int]) -> "TransitionToolCacheStats":
        """Create stats from dict (xdist worker transfer)."""
        return cls(
            key_test_hits=data.get("key_test_hits", 0),
            key_test_miss=data.get("key_test_miss", 0),
            subkey_test_hits=data.get("subkey_test_hits", 0),
            subkey_test_miss=data.get("subkey_test_miss", 0),
            unique_keys=data.get("unique_keys", 0),
        )

record_key(key)

Record a cache key and update unique_keys count.

Source code in packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

def record_key(self, key: str) -> None:
    """Record a cache key and update unique_keys count."""
    self._seen_keys.add(key)
    self.unique_keys = len(self._seen_keys)

expected_hits property

Number of tests expected to hit the cache.

to_dict()

Convert stats to dict for xdist worker transfer.

Source code in packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

def to_dict(self) -> Dict[str, int]:
    """Convert stats to dict for xdist worker transfer."""
    return {
        "key_test_hits": self.key_test_hits,
        "key_test_miss": self.key_test_miss,
        "subkey_test_hits": self.subkey_test_hits,
        "subkey_test_miss": self.subkey_test_miss,
        "unique_keys": self.unique_keys,
    }

add(other)

Add another stats object to this one.

Source code in packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

def add(self, other: "TransitionToolCacheStats") -> None:
    """Add another stats object to this one."""
    self.key_test_hits += other.key_test_hits
    self.key_test_miss += other.key_test_miss
    self.subkey_test_hits += other.subkey_test_hits
    self.subkey_test_miss += other.subkey_test_miss
    self.unique_keys += other.unique_keys

from_dict(data) classmethod

Create stats from dict (xdist worker transfer).

Source code in packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

@classmethod
def from_dict(cls, data: Dict[str, int]) -> "TransitionToolCacheStats":
    """Create stats from dict (xdist worker transfer)."""
    return cls(
        key_test_hits=data.get("key_test_hits", 0),
        key_test_miss=data.get("key_test_miss", 0),
        subkey_test_hits=data.get("subkey_test_hits", 0),
        subkey_test_miss=data.get("subkey_test_miss", 0),
        unique_keys=data.get("unique_keys", 0),
    )

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 Engine X fixtures by storing only the accounts that changed during test execution, rather than the full post-state which may contain thousands of unchanged 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	Genesis pre-allocation state	required

Returns:

Type	Description
Alloc	Alloc containing only the state differences for efficient storage

Source code in packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

def calculate_post_state_diff(
    post_state: BaseAlloc, genesis_state: BaseAlloc
) -> BaseAlloc:
    """
    Calculate the state difference between post_state and genesis_state.

    This function enables significant space savings in Engine X fixtures by
    storing only the accounts that changed during test execution, rather than
    the full post-state which may contain thousands of unchanged 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: 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 BaseAlloc(diff)

default_output_directory()

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

Source code in packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

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 packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

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 packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

def pytest_addoption(parser: pytest.Parser) -> None:
    """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=None,
        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(
        "--t8n-server-url",
        action="store",
        dest="t8n_server_url",
        type=str,
        default=None,
        help=(
            "[INTERNAL USE ONLY] URL of the t8n server to use. Used by "
            "framework tests/ci; not intended for regular CLI use."
        ),
    )
    evm_group.addoption(
        "--chain-id",
        action="store",
        dest="chain_id",
        type=int,
        default=None,
        help=(
            "Specify the chain ID for the test filling. "
            f"Default: {ChainConfigDefaults.chain_id}."
        ),
    )
    evm_group.addoption(
        "--traces",
        action="store_true",
        dest="evm_collect_traces",
        default=None,
        help="Collect traces of execution info 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' suffix). Tarball output "
            "automatically enables --generate-all-formats. Can be "
            f"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(
        "--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 ceiling for blocks and tests that attempt "
            f"to consume an entire block's gas. "
            f"(Default: {EnvironmentDefaults.gas_limit})"
        ),
    )
    test_group.addoption(
        "--generate-pre-alloc-groups",
        action="store_true",
        dest="generate_pre_alloc_groups",
        default=False,
        help="Generate pre-allocation groups (phase 1 only).",
    )
    test_group.addoption(
        "--use-pre-alloc-groups",
        action="store_true",
        dest="use_pre_alloc_groups",
        default=False,
        help="Fill tests using existing pre-allocation groups (phase 2 only).",
    )
    test_group.addoption(
        "--generate-all-formats",
        action="store_true",
        dest="generate_all_formats",
        default=False,
        help=(
            "Generate all fixture formats including BlockchainEngineX. "
            "Enables two-phase execution: Phase 1 generates pre-allocation "
            "groups, phase 2 generates all supported fixture formats."
        ),
    )

    optimize_gas_group = parser.getgroup(
        "optimize gas",
        "Arguments defining test gas optimization behavior.",
    )
    optimize_gas_group.addoption(
        "--optimize-gas",
        action="store_true",
        dest="optimize_gas",
        default=False,
        help=(
            "Attempt to optimize gas used in every transaction for filled "
            "tests, then print the minimum gas at which the test still "
            "produces a correct post state and the exact same trace."
        ),
    )
    optimize_gas_group.addoption(
        "--optimize-gas-output",
        action="store",
        dest="optimize_gas_output",
        default=Path("optimize-gas-output.json"),
        type=Path,
        help=(
            "Path to the JSON file that is output to the gas optimization. "
            "Requires `--optimize-gas`."
        ),
    )
    optimize_gas_group.addoption(
        "--optimize-gas-max-gas-limit",
        action="store",
        dest="optimize_gas_max_gas_limit",
        default=None,
        type=int,
        help=(
            "Maximum gas limit for gas optimization, if reached the search "
            "will stop and fail for that test. Requires `--optimize-gas`."
        ),
    )
    optimize_gas_group.addoption(
        "--optimize-gas-post-processing",
        action="store_true",
        dest="optimize_gas_post_processing",
        default=False,
        help=(
            "Post process traces during gas optimization to account for "
            "opcodes that put the current gas in the stack, in order to "
            "remove remaining-gas from the comparison."
        ),
    )

    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=None,
        help=(
            "Path to dump the transition tool debug output. "
            "Only creates debug output when explicitly specified."
        ),
    )
    debug_group.addoption(
        "--post-verifications",
        action="store_true",
        dest="post_verifications",
        default=False,
        help=(
            "Include a postVerifications field in fixture output "
            "that records which post-state checks were "
            "performed during filling."
        ),
    )

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 packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

@pytest.hookimpl(tryfirst=True)
def pytest_configure(config: pytest.Config) -> None:
    """
    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.
    """
    # Register custom markers
    # Modify the block gas limit if specified.
    if option_was_explicitly_set(config, "--block-gas-limit"):
        EnvironmentDefaults.gas_limit = config.getoption("block_gas_limit")

    # Initialize fixture output configuration
    config.fixture_output = FixtureOutput.from_config(  # type: ignore[attr-defined]
        config
    )

    # Initialize filling session
    config.filling_session = FillingSession.from_config(  # type: ignore[attr-defined]
        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(  # type: ignore[attr-defined]
            is_master=not hasattr(config, "workerinput")
        )
    except ValueError as e:
        pytest.exit(str(e), returncode=pytest.ExitCode.USAGE_ERROR)

    # Register atexit/signal handlers for cleanup (master only, not workers).
    global _fixture_output_dir, _atexit_registered
    global _original_sigint_handler, _original_sigterm_handler
    is_xdist_worker = hasattr(config, "workerinput")
    if not config.fixture_output.is_stdout:  # type: ignore[attr-defined]
        _fixture_output_dir = config.fixture_output.directory  # type: ignore[attr-defined]
        if not _atexit_registered and not is_xdist_worker:
            atexit.register(_merge_on_exit)
            _original_sigint_handler = signal.signal(
                signal.SIGINT, _termination_handler
            )
            _original_sigterm_handler = signal.signal(
                signal.SIGTERM, _termination_handler
            )
            _atexit_registered = True

    if (
        not config.getoption("disable_html")
        and config.getoption("htmlpath") is None
        and config.filling_session.phase_manager.current_phase  # type: ignore[attr-defined]
        != FixtureFillingPhase.PRE_ALLOC_GENERATION
    ):
        config.option.htmlpath = (
            config.fixture_output.directory  # type: ignore[attr-defined]
            / default_html_report_file_path()
        )

    config.gas_optimized_tests = {}  # type: ignore[attr-defined]
    if config.getoption("optimize_gas", False):
        if config.getoption("optimize_gas_post_processing"):
            config.op_mode = (  # type: ignore[attr-defined]
                OpMode.OPTIMIZE_GAS_POST_PROCESSING
            )
        else:
            config.op_mode = OpMode.OPTIMIZE_GAS  # type: ignore[attr-defined]

    config.collect_traces = (  # type: ignore[attr-defined]
        config.getoption("evm_collect_traces")
        or config.getoption("optimize_gas", False)
    )

    # set default for --evm-dump-dir
    if (
        config.collect_traces  # type: ignore[attr-defined]
        and config.getoption("base_dump_dir")
    ) is None:
        config.option.base_dump_dir = "traces"

    # 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.
    evm_bin = config.getoption("evm_bin")
    trace = config.getoption("evm_collect_traces")
    t8n_server_url = config.getoption("t8n_server_url")
    kwargs = {
        "trace": trace,
    }
    if t8n_server_url is not None:
        kwargs["server_url"] = t8n_server_url
    if evm_bin is None:
        assert TransitionTool.default_tool is not None, (
            "No default transition tool found"
        )
        t8n = TransitionTool.default_tool(**kwargs)
    else:
        t8n = TransitionTool.from_binary_path(binary_path=evm_bin, **kwargs)

    if (
        isinstance(config.getoption("numprocesses"), int)
        and config.getoption("numprocesses") > 0
        and not t8n.supports_xdist
    ):
        pytest.exit(
            f"The {t8n.__class__.__name__} t8n tool does not work well "
            "with the xdist plugin; use -n=0.",
            returncode=pytest.ExitCode.USAGE_ERROR,
        )
    config.t8n = t8n  # type: ignore[attr-defined]

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

    # Initialize aggregated cache stats on xdist controller.
    # Controller = xdist active but not a worker.
    numprocesses = config.getoption("numprocesses", None)
    is_xdist_active = isinstance(numprocesses, int) and numprocesses > 0
    is_controller = is_xdist_active and not hasattr(config, "workerinput")
    if is_controller:
        config.t8n_cache_stats_aggregated = (  # type: ignore[attr-defined]
            TransitionToolCacheStats()
        )

    # Default chain id can be overwritten by user flag or env var
    ChainConfigDefaults.chain_id = DEFAULT_CHAIN_ID
    chain_id = config.getoption("chain_id")
    if chain_id is None:
        env_chain_id = os.environ.get("CHAIN_ID")
        if env_chain_id is not None:
            chain_id = int(env_chain_id)

    if chain_id is not None:
        ChainConfigDefaults.chain_id = int(chain_id)

pytest_report_header(config)

Add lines to pytest's console output header.

Source code in packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

@pytest.hookimpl(trylast=True)
def pytest_report_header(config: pytest.Config) -> List[str]:
    """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 packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

@pytest.hookimpl(tryfirst=True)
def pytest_report_teststatus(
    report: Any, config: pytest.Config
) -> tuple[str, str, str] | None:
    """
    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()
    return None

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 packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

@pytest.hookimpl(hookwrapper=True, trylast=True)
def pytest_terminal_summary(
    terminalreporter: TerminalReporter,
    exitstatus: int,
    config: pytest.Config,
) -> Generator[None, None, None]:
    """
    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.
    """
    del exitstatus

    yield
    if config.fixture_output.is_stdout or hasattr(config, "workerinput"):  # type: ignore[attr-defined]
        return

    # Get cache stats: try aggregated (xdist), else local (sequential)
    t8n_cache_stats: TransitionToolCacheStats | None = getattr(
        config, "t8n_cache_stats_aggregated", None
    ) or getattr(config, "transition_tool_cache_stats", None)

    if t8n_cache_stats is not None:
        expected = t8n_cache_stats.expected_hits
        actual = t8n_cache_stats.key_test_hits
        if expected > 0:
            efficiency = actual / expected * 100
            terminalreporter.write_sep(
                "=",
                (
                    f" T8n cache: {efficiency:.0f}% hit rate"
                    f" ({actual}/{expected} tests expected),"
                    f" {t8n_cache_stats.subkey_test_hits} t8n calls saved"
                ),
                bold=True,
                green=efficiency == 100,
            )
        elif t8n_cache_stats.unique_keys > 0:
            terminalreporter.write_sep(
                "=",
                (
                    f" T8n cache: {t8n_cache_stats.unique_keys} unique"
                    " test groups, no cache sharing possible"
                ),
                bold=True,
            )
    stats = terminalreporter.stats
    if "passed" in stats and stats["passed"]:
        # Custom message for Phase 1 (pre-allocation group generation)
        session_instance: FillingSession = config.filling_session  # type: ignore[attr-defined]
        if session_instance.phase_manager.is_pre_alloc_generation:
            # Generate summary stats
            # For xdist, count files and accounts without fully loading groups
            # (avoids expensive state_root computation just for summary stats)
            if config.pluginmanager.hasplugin("xdist"):
                pre_alloc_folder = (
                    config.fixture_output.pre_alloc_groups_folder_path  # type: ignore[attr-defined]
                )
                group_files = list(pre_alloc_folder.glob("*.json"))
                total_groups = len(group_files)
                # Count accounts by loading as builder (no genesis computation)
                total_accounts = 0
                for group_file in group_files:
                    builder = PreAllocGroupBuilder.model_validate_json(
                        group_file.read_text()
                    )
                    total_accounts += builder.get_pre_account_count()
            else:
                assert session_instance.pre_alloc_group_builders is not None
                total_groups = len(
                    session_instance.pre_alloc_group_builders.root
                )
                total_accounts = sum(
                    builder.get_pre_account_count()
                    for builder in session_instance.pre_alloc_group_builders.root.values()  # noqa: E501
                )

            terminalreporter.write_sep(
                "=",
                f" Phase 1 Complete: Generated {total_groups} pre-alloc "
                f"groups ({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 "
                    f'"{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 packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

def pytest_metadata(metadata: Any) -> None:
    """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 packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

def pytest_html_results_table_header(cells: Any) -> None:
    """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 packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

def pytest_html_results_table_row(report: Any, cells: Any) -> None:
    """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">'
                f"{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">'
                        f"{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 packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_makereport(
    item: Any, call: Any
) -> Generator[None, None, None]:
    """
    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()  # type: ignore[attr-defined]

    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",
                "blockchain_test_sync",
            ]:
                report.user_properties.append(
                    ("evm_dump_dir", item.config.evm_dump_dir)
                )
            else:
                report.user_properties.append(("evm_dump_dir", "N/A"))

pytest_html_report_title(report)

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

Source code in packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

def pytest_html_report_title(report: Any) -> None:
    """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 packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

@pytest.fixture(autouse=True, scope="session")
def evm_bin(request: pytest.FixtureRequest) -> Path | None:
    """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 packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

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

session_t8n(request)

Return configured transition tool for the session.

Source code in packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

@pytest.fixture(autouse=True, scope="session")
def session_t8n(
    request: pytest.FixtureRequest,
) -> Generator[TransitionTool, None, None]:
    """Return configured transition tool for the session."""
    t8n: TransitionTool = request.config.t8n  # type: ignore
    if not t8n.exception_mapper.reliable:
        t8n_name = t8n.__class__.__name__
        warnings.warn(
            f"The t8n tool being used to fill tests ({t8n_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()

get_t8n_cache_key(request)

Get the cache key to be used for the current test, if any.

Source code in packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

def get_t8n_cache_key(request: pytest.FixtureRequest) -> str | None:
    """Get the cache key to be used for the current test, if any."""
    mark: pytest.Mark = request.node.get_closest_marker(
        "transition_tool_cache_key"
    )
    if mark is not None and len(mark.args) == 1:
        return f"{strip_fixture_format_from_node(request.node)}-{mark.args[0]}"
    return None

transition_tool_cache_stats(request)

Get the transition tool cache stats.

Source code in packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

@pytest.fixture(autouse=True, scope="session")
def transition_tool_cache_stats(
    request: pytest.FixtureRequest,
) -> Generator[TransitionToolCacheStats, None, None]:
    """Get the transition tool cache stats."""
    stats = TransitionToolCacheStats()
    yield stats
    # Store stats for later access
    request.config.transition_tool_cache_stats = stats  # type: ignore[attr-defined]
    # For xdist workers, send stats to controller via workeroutput
    if hasattr(request.config, "workeroutput"):
        request.config.workeroutput["t8n_cache_stats"] = stats.to_dict()

t8n(request, session_t8n, dump_dir_parameter_level, transition_tool_cache_stats)

Set the transition tool up for the current test.

Source code in packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

@pytest.fixture(autouse=True, scope="function")
def t8n(
    request: pytest.FixtureRequest,
    session_t8n: TransitionTool,
    dump_dir_parameter_level: Path | None,
    transition_tool_cache_stats: TransitionToolCacheStats,
) -> Generator[TransitionTool, None, None]:
    """Set the transition tool up for the current test."""
    if transition_tool_cache_key := get_t8n_cache_key(request):
        # This test is allowed to cache results
        transition_tool_cache_stats.record_key(transition_tool_cache_key)
        if session_t8n.set_cache(key=transition_tool_cache_key):
            transition_tool_cache_stats.key_test_hits += 1
        else:
            transition_tool_cache_stats.key_test_miss += 1
    else:
        # Test cannot use output cache, remove it
        session_t8n.remove_cache()
    # Reset the traces
    session_t8n.reset_traces()
    session_t8n.call_counter = 0
    session_t8n.debug_dump_dir = dump_dir_parameter_level
    # TODO: Configure the transition tool to count opcodes only when required.
    session_t8n.reset_opcode_count()
    yield session_t8n
    # Only collect subkey stats for cacheable tests (non-cacheable tests
    # still interact with the OutputCache after remove_cache, producing
    # phantom misses that would skew the hit rate).
    if transition_tool_cache_key and session_t8n.output_cache is not None:
        transition_tool_cache_stats.subkey_test_hits += (
            session_t8n.output_cache.hits
        )
        transition_tool_cache_stats.subkey_test_miss += (
            session_t8n.output_cache.misses
        )
        # Reset counters to avoid double-counting (cache persists across tests)
        session_t8n.output_cache.hits = 0
        session_t8n.output_cache.misses = 0

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 packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

@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 packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

@pytest.fixture(autouse=True, scope="session")
def evm_fixture_verification(
    request: pytest.FixtureRequest,
    do_fixture_verification: bool,
    evm_bin: Path | None,
    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:
        pytest.exit(
            "--verify-fixtures requires --evm-bin or --verify-fixtures-bin "
            "to be specified.",
            returncode=pytest.ExitCode.USAGE_ERROR,
        )
    try:
        evm_fixture_verification = FixtureConsumerTool.from_binary_path(
            binary_path=Path(verify_fixtures_bin),
            trace=request.config.collect_traces,  # type: ignore[attr-defined]
        )
    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 packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

@pytest.fixture(scope="session")
def base_dump_dir(request: pytest.FixtureRequest) -> Path | None:
    """Path to base directory to dump the evm debug output."""
    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 packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

@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 packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

@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 packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

@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 packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

@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} "
                f"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_shangh ai/

Source code in packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

@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_shangh
    ai/
    """
    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 packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

def get_fixture_collection_scope(
    fixture_name: str, config: pytest.Config
) -> str:
    """
    Return the appropriate scope to write fixture JSON files.

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

    if config.fixture_output.is_stdout:  # type: ignore[attr-defined]
        return "session"
    if (
        config.fixture_output.single_fixture_per_file  # type: ignore
    ):
        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 packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

@pytest.fixture(autouse=True, scope="module")
def reference_spec(request: pytest.FixtureRequest) -> 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 packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

@pytest.fixture(scope=get_fixture_collection_scope)  # type: ignore[arg-type]
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.
    """
    # Dynamically load the 'static_filler' and 'solc' plugins if needed
    if request.config.getoption("fill_static_tests_enabled"):
        request.config.pluginmanager.import_plugin(
            "execution_testing.cli.pytest_commands.plugins.filler.static_filler"
        )
        request.config.pluginmanager.import_plugin(
            "execution_testing.cli.pytest_commands.plugins.solc.solc"
        )

    fixture_collector = FixtureCollector(
        output_dir=fixture_output.directory,
        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,
        generate_index=request.config.getoption("generate_index"),
    )
    yield fixture_collector
    try:
        # dump_fixtures() only needed for stdout mode
        fixture_collector.dump_fixtures()
        # Verify fixtures for stdout mode only (files are in memory).
        # For file mode, verification happens at session finish after merge.
        if do_fixture_verification and fixture_output.is_stdout:
            fixture_collector.verify_fixture_files(evm_fixture_verification)
    finally:
        # Always close streaming file handles, even on error
        fixture_collector.close_streaming_files()

filler_path(request)

Return directory containing the tests to execute.

Source code in packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

@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 packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

def node_to_test_info(node: pytest.Item) -> TestInfo:
    """Return test info of the current node item."""
    # Strip xdist group suffix (@groupname) that may be added during execution.
    return TestInfo(
        name=_strip_xdist_group_suffix(node.name),
        id=_strip_xdist_group_suffix(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 packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

@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 packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

@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}()` "
            f"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 packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

def base_test_parametrizer(cls: Type[BaseTest]) -> Any:
    """
    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.
    """
    cls_fixture_parameters = [
        p for p in ALL_FIXTURE_PARAMETERS if p in cls.model_fields
    ]

    @pytest.fixture(
        scope="function",
        name=cls.pytest_parameter_name(),
    )
    def base_test_parametrizer_func(
        request: pytest.FixtureRequest,
        t8n: TransitionTool,
        fork: Fork | TransitionFork,
        reference_spec: ReferenceSpec,
        pre: Alloc,
        output_dir: Path,
        fixture_collector: FixtureCollector,
        test_case_description: str,
        fixture_source_url: str,
        gas_benchmark_value: int,
        fixed_opcode_count: int | None,
        is_tx_gas_heavy_test: bool,
        is_exception_test: bool,
    ) -> Any:
        """
        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.
        """
        del fixed_opcode_count
        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
            __is_base_test_wrapper__ = True

            def __init__(self, *args: Any, **kwargs: Any) -> None:
                if "pre" not in kwargs:
                    kwargs["pre"] = pre
                if "expected_benchmark_gas_used" not in kwargs:
                    kwargs["expected_benchmark_gas_used"] = gas_benchmark_value
                kwargs["fork"] = fork
                op_mode: OpMode = request.config.op_mode  # type: ignore
                kwargs["operation_mode"] = op_mode
                kwargs["is_tx_gas_heavy_test"] = is_tx_gas_heavy_test
                kwargs["is_exception_test"] = is_exception_test
                if (
                    op_mode == OpMode.OPTIMIZE_GAS
                    or op_mode == OpMode.OPTIMIZE_GAS_POST_PROCESSING
                ):
                    kwargs["gas_optimization_max_gas_limit"] = (
                        request.config.getoption(
                            "optimize_gas_max_gas_limit", None
                        )
                    )
                kwargs |= {
                    p: request.getfixturevalue(p)
                    for p in cls_fixture_parameters
                    if p not in kwargs
                }

                super(BaseTestWrapper, self).__init__(*args, **kwargs)

                # Get the filling session from config
                session: FillingSession = request.config.filling_session  # type: ignore
                assert isinstance(session, FillingSession)

                group_salt: str | None = None
                if pre_alloc_group_marker := request.node.get_closest_marker(
                    "pre_alloc_group"
                ):
                    # Get the group name/salt from marker args
                    if pre_alloc_group_marker.args:
                        group_salt = str(pre_alloc_group_marker.args[0])
                    else:
                        # We got the marker but unspecified, pass test name
                        group_salt = _strip_xdist_group_suffix(
                            request.node.nodeid
                        )

                pre_alloc_hash: str | None = None
                # Phase 1: Generate pre-allocation groups
                if session.phase_manager.is_pre_alloc_generation:
                    # Use the original update_pre_alloc_groups method which
                    # returns the groups
                    assert session.pre_alloc_group_builders is not None
                    test_id = _strip_xdist_group_suffix(request.node.nodeid)
                    genesis_environment = self.get_genesis_environment()
                    pre_alloc_hash = pre.compute_pre_alloc_group_hash(
                        fork=fork,
                        genesis_environment=genesis_environment,
                        group_salt=group_salt,
                    )
                    session.pre_alloc_group_builders.add_test_pre(
                        pre_alloc_hash=pre_alloc_hash,
                        test_id=test_id,
                        fork=fork,
                        chain_id=ChainConfigDefaults.chain_id,
                        environment=genesis_environment,
                        pre=pre,
                    )
                    return  # Skip fixture generation in phase 1

                # Phase 2: Use pre-allocation groups (only for
                # BlockchainEngineXFixture)
                if (
                    FixtureFillingPhase.PRE_ALLOC_GENERATION
                    in fixture_format.format_phases
                ):
                    pre_alloc_hash = pre.compute_pre_alloc_group_hash(
                        fork=fork,
                        genesis_environment=self.get_genesis_environment(),
                        group_salt=group_salt,
                    )
                    group = session.get_pre_alloc_group(pre_alloc_hash)
                    self.pre = group.pre
                fill_result: FillResult | None = None
                try:
                    fill_result = self.generate(
                        t8n=t8n,
                        fixture_format=fixture_format,
                    )
                finally:
                    if (
                        self.operation_mode == OpMode.OPTIMIZE_GAS
                        or self.operation_mode
                        == OpMode.OPTIMIZE_GAS_POST_PROCESSING
                    ):
                        gas_optimized_tests = (
                            request.config.gas_optimized_tests  # type: ignore
                        )
                        assert gas_optimized_tests is not None
                        # Force adding something to the list, even if it's
                        # None, to keep track of failed tests in the output
                        # file.
                        gas_optimized_tests[request.node.nodeid] = (
                            fill_result.gas_optimization
                            if fill_result is not None
                            else None
                        )
                assert fill_result is not None
                fixture = fill_result.fixture
                if (
                    request.config.getoption("post_verifications")
                    and fill_result.post_verifications is not None
                ):
                    fixture.post_verifications = fill_result.post_verifications
                # If operation mode is benchmarking, check the gas used.
                self.validate_benchmark_gas(
                    benchmark_gas_used=fill_result.benchmark_gas_used,
                    gas_benchmark_value=gas_benchmark_value,
                )

                # Post-process for Engine X format (add pre_hash and state
                # diff)
                if (
                    FixtureFillingPhase.PRE_ALLOC_GENERATION
                    in fixture_format.format_phases
                    and pre_alloc_hash is not None
                ):
                    # TODO: This should be handled by the `generate` method
                    # of the spec.
                    assert isinstance(fixture, BlockchainEngineXFixture)
                    fixture.pre_hash = pre_alloc_hash

                    # Calculate state diff for efficiency
                    if (
                        hasattr(fixture, "post_state")
                        and fixture.post_state is not None
                    ):
                        group = session.get_pre_alloc_group(pre_alloc_hash)
                        fixture.post_state_diff = calculate_post_state_diff(
                            fixture.post_state, group.pre
                        )

                fill_metadata: Dict[str, Any] = {}
                if t8n.opcode_count is not None:
                    fill_metadata["opcode_count"] = (
                        t8n.opcode_count.model_dump()
                    )
                if fill_result.metadata:
                    fill_metadata.update(fill_result.metadata)

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

                output_subdir = resolve_fixture_subfolder(
                    list(request.node.iter_markers("fixture_subfolder"))
                )

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

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

NOTE: The static test filler does NOT use this hook. See FillerFile.collect() in ./static_filler.py for more details.

Source code in packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

def pytest_generate_tests(metafunc: pytest.Metafunc) -> None:
    """
    Pytest hook used to dynamically generate test cases for each fixture format
    a given test spec supports.

    NOTE: The static test filler does NOT use this hook. See
    FillerFile.collect() in ./static_filler.py for more details.
    """
    session: FillingSession = metafunc.config.filling_session  # type: ignore[attr-defined]
    for test_type in BaseTest.spec_types.values():
        if test_type.pytest_parameter_name() in metafunc.fixturenames:
            parameters = []
            for i, format_with_or_without_label in enumerate(
                test_type.supported_fixture_formats
            ):
                if not session.should_generate_format(
                    format_with_or_without_label
                ):
                    continue
                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 packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

@pytest.hookimpl(tryfirst=True)
def pytest_collection_modifyitems(
    config: pytest.Config, items: List[pytest.Item | pytest.Function]
) -> None:
    """
    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.
    """
    # Track specs with no fixture formats for warning message
    specs_without_fixture_formats: Dict[str, Set[str]] = {}

    items_for_removal = []
    for i, item in enumerate(items):
        item.name = item.name.strip().replace(" ", "-")
        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 "parametrized_fork" not in params
            or params["parametrized_fork"] is None
        ):
            items_for_removal.append(i)
            continue
        fork: Fork | TransitionFork = params["parametrized_fork"]
        spec_type, fixture_format = get_spec_format_for_item(params)
        if isinstance(fixture_format, NotSetType):
            items_for_removal.append(i)
            # Track this spec type and the test file
            spec_name = spec_type.__name__
            if spec_name not in specs_without_fixture_formats:
                specs_without_fixture_formats[spec_name] = set()
            # Get the test file path from the item
            test_file = (
                str(item.fspath) if hasattr(item, "fspath") else str(item.path)
            )
            specs_without_fixture_formats[spec_name].add(test_file)
            continue
        assert issubclass(fixture_format, BaseFixture)
        if not fixture_format.supports_fork(fork):
            items_for_removal.append(i)
            continue

        markers = list(item.iter_markers())

        # Both the fixture format itself and the spec filling it have a chance
        # to veto the filling of a specific format.
        if fixture_format.discard_fixture_format_by_marks(fork, markers):
            items_for_removal.append(i)
            continue
        if spec_type.discard_fixture_format_by_marks(
            fixture_format, fork, markers
        ):
            items_for_removal.append(i)
            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 = fork.transitions_from()
                item._nodeid = item._nodeid.replace(
                    f"fork_{fork.name()}",
                    f"fork_{base_fork.name()}",
                )

    for i in reversed(items_for_removal):
        items.pop(i)

    # Warn users if tests were removed because they have no fixture formats
    if specs_without_fixture_formats:
        reporter = config.pluginmanager.get_plugin("terminalreporter")
        if reporter and isinstance(reporter, TerminalReporter):
            reporter.write_line("")
            reporter.write_sep(
                "=",
                "NOTICE: Tests ignored (as expected; no fixture formats)",
                yellow=True,
                bold=True,
            )
            for spec_name, test_files in specs_without_fixture_formats.items():
                reporter.write_line(
                    f"  {spec_name} has no supported_fixture_formats "
                    f"for 'fill' command.",
                    yellow=True,
                )
                for test_file in sorted(test_files):
                    reporter.write_line(f"    - {test_file}", yellow=True)
            reporter.write_line("")
            reporter.write_line(
                "  These tests are designed for the 'execute' command, "
                "not 'fill'.",
                yellow=True,
            )
            reporter.write_line(
                "  Use 'uv run execute hive' or 'uv run execute remote' "
                "instead.",
                yellow=True,
            )
            reporter.write_sep("=", yellow=True, bold=True)
            reporter.write_line("")

    # Build base_nodeid cache and identify slow groups.
    # If ANY fixture format variant is marked slow, treat ALL variants as slow
    # to keep them grouped together for cache locality.
    item_base_nodeids: Dict[int, str] = {}
    slow_base_nodeids: set[str] = set()
    for item in items:
        base_nodeid = strip_fixture_format_from_node(item)
        item_base_nodeids[id(item)] = base_nodeid
        if item.get_closest_marker("slow") is not None:
            slow_base_nodeids.add(base_nodeid)

    # Sort items for optimal execution order:
    # 1. Slow groups first (LPT scheduling for xdist load balance)
    # 2. Related fixture formats together (cache locality)
    # 3. Cacheable formats first within a group (so non-cacheable formats
    #    don't clear the cache between two cacheable ones; e.g., for
    #    StateTest the _from_state_test labels sort engine_x between the
    #    two cacheable formats alphabetically, breaking cache hits)
    # 4. Deterministic order within groups (alphabetical by nodeid)
    def sort_key(item: pytest.Item) -> tuple[bool, str, bool, str]:
        base = item_base_nodeids[id(item)]
        is_slow = base in slow_base_nodeids
        has_cache_key = (
            item.get_closest_marker("transition_tool_cache_key") is not None
        )
        return (not is_slow, base, not has_cache_key, item.nodeid)

    items.sort(key=sort_key)

    # Group related fixture formats for cache locality with xdist.
    # Detect xdist: check for -n in original args (collection happens before
    # xdist initializes, so config.option.numprocesses is None).
    orig_args = (
        config.invocation_params.args
        if hasattr(config, "invocation_params")
        else []
    )
    is_xdist = any(arg == "-n" or arg.startswith("-n") for arg in orig_args)

    if is_xdist:
        # Add xdist_group markers for --dist=loadgroup.
        # Skip if test already has an xdist_group marker (e.g., bigmem).
        # Tests with existing markers still benefit from the cache within their
        # worker, just with potentially more interleaving.
        # IMPORTANT: Use hash for group name because loadgroup's _split_scope
        # uses rfind("]") to detect group suffix, and our base_nodeid contains
        # "]" characters which would break the detection.
        for item in items:
            if not item.get_closest_marker("xdist_group"):
                base_nodeid = item_base_nodeids[id(item)]
                h = hashlib.md5(
                    base_nodeid.encode(), usedforsecurity=False
                ).hexdigest()[:8]
                group_name = f"t8n-cache-{h}"
                item.add_marker(pytest.mark.xdist_group(name=group_name))

pytest_sessionfinish(session, exitstatus)

Perform session finish tasks.

• Save pre-allocation groups (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 packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

def pytest_sessionfinish(session: pytest.Session, exitstatus: int) -> None:
    """
    Perform session finish tasks.

    - Save pre-allocation groups (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.
    """
    logger = logging.getLogger("fill.sessionfinish")
    is_worker = xdist.is_xdist_worker(session)

    # Workers collect logs to forward to master via workeroutput
    worker_timing_logs: list[str] = []

    def _log_timing(msg: str) -> None:
        """Log with timestamp. Workers collect logs; master logs directly."""
        log_line = f"[sessionfinish] {time.strftime('%H:%M:%S')} {msg}"
        if is_worker:
            worker_timing_logs.append(log_line)
        else:
            logger.debug(log_line)

    # Log immediately when hook is entered (before any early returns)
    _log_timing(f"pytest_sessionfinish ENTERED (worker={is_worker})")

    del exitstatus

    # Save pre-allocation groups after phase 1
    fixture_output: FixtureOutput = session.config.fixture_output  # type: ignore[attr-defined]
    session_instance: FillingSession = session.config.filling_session  # type: ignore[attr-defined]
    if session_instance.phase_manager.is_pre_alloc_generation:
        _log_timing("Phase 1: saving pre-alloc groups (partial)...")
        t0 = time.time()
        session_instance.save_pre_alloc_groups()
        _log_timing(
            f"Phase 1: save_pre_alloc_groups done in {time.time() - t0:.1f}s"
        )

        # Master merges all worker partial files after all workers finish
        if not is_worker:
            _log_timing("Phase 1 (master): merging partial group files...")
            t0 = time.time()
            pre_alloc_folder = fixture_output.pre_alloc_groups_folder_path
            merge_partial_group_files(pre_alloc_folder)
            _log_timing(
                f"Phase 1 (master): merge done in {time.time() - t0:.1f}s"
            )
        else:
            # Workers: clear in-memory state to reduce memory pressure while
            # waiting for other workers to finish
            session_instance.pre_alloc_group_builders = None
            gc.collect()
            # Store timing logs for master to print when this worker finishes
            session.config.workeroutput["timing_logs"] = worker_timing_logs  # type: ignore[attr-defined] # noqa: E501

        return

    if session.config.getoption("optimize_gas", False):
        output_file = Path(session.config.getoption("optimize_gas_output"))
        lock_file_path = output_file.with_suffix(".lock")
        assert hasattr(session.config, "gas_optimized_tests")
        gas_optimized_tests: Dict[str, int] = (
            session.config.gas_optimized_tests
        )
        with FileLock(lock_file_path):
            if output_file.exists():
                gas_optimized_tests = (
                    json.loads(output_file.read_text()) | gas_optimized_tests
                )
            output_file.write_text(
                json.dumps(gas_optimized_tests, indent=2, sort_keys=True)
            )

    if is_worker:
        # Workers: clear in-memory state to reduce memory pressure while
        # waiting for other workers to finish
        session_instance.pre_alloc_groups = None
        if hasattr(session.config, "fixture_collector"):
            fc = session.config.fixture_collector
            fc.all_fixtures.clear()
            fc._fixtures_to_verify.clear()
        gc.collect()
        # Store timing logs for master to print when this worker finishes
        session.config.workeroutput["timing_logs"] = worker_timing_logs  # type: ignore[attr-defined] # noqa: E501
        return

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

    _log_timing("Finalization (master): starting...")

    # Merge partial fixture files from all workers into final JSON files
    _log_timing("merge_partial_fixture_files: starting...")
    t0 = time.time()
    merge_partial_fixture_files(fixture_output.directory)
    _log_timing(
        f"merge_partial_fixture_files: done in {time.time() - t0:.1f}s"
    )

    # Remove any lock files that may have been created.
    lock_files = list(fixture_output.directory.rglob("*.lock"))
    if lock_files:
        _log_timing(f"Removing {len(lock_files)} lock files...")
        t0 = time.time()
        for file in lock_files:
            file.unlink()
        _log_timing(f"Lock files removed in {time.time() - t0:.1f}s")

    # Verify fixtures after merge if verification is enabled
    if session.config.getoption("verify_fixtures"):
        _log_timing("_verify_fixtures_post_merge: starting...")
        t0 = time.time()
        _verify_fixtures_post_merge(session.config, fixture_output.directory)
        _log_timing(
            f"_verify_fixtures_post_merge: done in {time.time() - t0:.1f}s"
        )

    # Generate index file for all produced fixtures by merging partial indexes.
    # Only merge if partial indexes were actually written (i.e., tests produced
    # fixtures). When no tests are filled (e.g., all skipped), no partial
    # indexes exist and merge_partial_indexes should not be called.
    if (
        session.config.getoption("generate_index")
        and not session_instance.phase_manager.is_pre_alloc_generation
    ):
        meta_dir = fixture_output.directory / ".meta"
        if meta_dir.exists() and any(meta_dir.glob("partial_index*.jsonl")):
            _log_timing("merge_partial_indexes: starting...")
            t0 = time.time()
            merge_partial_indexes(fixture_output.directory, quiet_mode=True)
            _log_timing(
                f"merge_partial_indexes: done in {time.time() - t0:.1f}s"
            )

    # Create tarball of the output directory if the output is a tarball.
    if fixture_output.is_tarball:
        _log_timing("create_tarball: starting...")
        t0 = time.time()
        fixture_output.create_tarball()
        _log_timing(f"create_tarball: done in {time.time() - t0:.1f}s")

    _log_timing("Finalization (master): COMPLETE")

pytest_testnodedown(node, error)

Called on master when a worker node finishes.

Aggregate cache stats and print timing logs from the worker.

Source code in packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/filler.py

def pytest_testnodedown(node: Any, error: Any) -> None:
    """
    Called on master when a worker node finishes.

    Aggregate cache stats and print timing logs from the worker.
    """
    del error
    _aggregate_cache_stats(node)
    logger = logging.getLogger("fill.sessionfinish")
    worker_id = getattr(node, "workerinput", {}).get("workerid", "unknown")
    timing_logs = getattr(node, "workeroutput", {}).get("timing_logs", [])
    for log_line in timing_logs:
        logger.debug(f"[worker {worker_id}] {log_line}")

Pre-alloc specifically conditioned for test filling.

pytest_addoption(parser)

Add command-line options to pytest.

Source code in packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/pre_alloc.py

def pytest_addoption(parser: pytest.Parser) -> None:
    """Add command-line options to pytest."""
    pre_alloc_group = parser.getgroup(
        "pre_alloc",
        "Arguments defining pre-allocation behavior during test filling.",
    )
    # TODO: consolidate with execute/rpc/remote.py
    pre_alloc_group.addoption(
        "--rpc-endpoint",
        action="store",
        dest="rpc_endpoint",
        default=None,
        help="RPC endpoint to an execution client.",
    )

Alloc

Bases: Alloc

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

Source code in packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/pre_alloc.py

class Alloc(SharedAlloc):
    """Allocation of accounts in the state, pre and post test execution."""

    _eoa_fund_amount_default: int = PrivateAttr(10**21)
    _account_salt: Dict[Hash, int] = PrivateAttr(default_factory=dict)
    _stub_accounts: Dict[str, Account] = PrivateAttr(default_factory=dict)

    def __init__(
        self,
        *args: Any,
        fork: Fork,
        flags: AllocFlags,
        stub_accounts: Dict[str, Account] | None = None,
        stub_eoas: Dict[str, EOA] | None = None,
        **kwargs: Any,
    ) -> None:
        """Initialize the pre-alloc."""
        super().__init__(
            *args,
            fork=fork,
            flags=flags,
            stub_eoas=stub_eoas,
            **kwargs,
        )
        if stub_accounts is not None:
            self._stub_accounts = stub_accounts

    def get_next_account_salt(self, account_hash: Hash) -> int:
        """Retrieve the next salt for this account."""
        salt = self._account_salt.get(account_hash, 0)
        self._account_salt[account_hash] = salt + 1
        return salt

    def code_pre_processor(self, code: BytesConvertible) -> BytesConvertible:
        """Pre-processes the code before setting it."""
        return code

    def modified_accounts_salt(self) -> int:
        """
        Return a salt if this pre-allocation was affected by setting addresses
        to hard-coded accounts or has pre-funded addresses.

        Any modification the test does to a hard-coded address must affect
        this salt.
        """
        if (
            not self._set_addresses
            and not self._pre_funded_addresses
            and not self._hardcoded_addresses_deployed_to
            and not self._deleted_addresses
        ):
            return 0

        # Build a hashable buffer from the modified accounts.
        buffer = b""
        altered_accounts = (
            self._set_addresses
            | self._pre_funded_addresses
            | self._hardcoded_addresses_deployed_to
        )
        if altered_accounts:
            buffer += b"\0"
            for altered_account in sorted(altered_accounts):
                buffer += altered_account
                account = self[altered_account]
                assert account is not None
                buffer += account.hash()
        if self._deleted_addresses:
            buffer += b"\1"
            for deleted_address in sorted(self._deleted_addresses):
                buffer += deleted_address

        return int.from_bytes(
            hashlib.sha256(buffer).digest()[:8], byteorder="big"
        )

    def compute_pre_alloc_group_hash(
        self,
        *,
        fork: Fork | TransitionFork,
        genesis_environment: Environment,
        group_salt: str | None,
    ) -> str:
        """Hash (fork, env) in order to group tests by genesis config."""
        fork_digest = hashlib.sha256(fork.name().encode("utf-8")).digest()
        fork_hash = int.from_bytes(fork_digest[:8], byteorder="big")
        combined_hash = (
            fork_hash
            ^ hash(genesis_environment)
            ^ self.modified_accounts_salt()
        )

        # Check if this pre-allocation has a group salt
        if group_salt:
            # Add custom salt to hash
            salt_hash = hashlib.sha256(group_salt.encode("utf-8")).digest()
            salt_int = int.from_bytes(salt_hash[:8], byteorder="big")
            combined_hash = combined_hash ^ salt_int

        return f"0x{combined_hash:016x}"

    def _deterministic_deploy_contract(
        self,
        *,
        deploy_code: BytesConvertible,
        salt: Hash | int,
        initcode: BytesConvertible | None,
        storage: Storage | StorageRootType | None,
        label: str | None,
    ) -> Address:
        """
        Filler implementation of contract deployment to a deterministic
        location.
        """
        if not isinstance(deploy_code, Bytes):
            deploy_code = Bytes(deploy_code)
        if initcode is None:
            initcode = Initcode(deploy_code=deploy_code)
        elif not isinstance(initcode, Bytes):
            initcode = Bytes(initcode)
        if storage is None:
            storage = {}
        salt = Hash(salt)
        # Everything is deployed at genesis, hence `transitions_from`
        fork = self._fork.transitions_from()
        contract_address = compute_deterministic_create2_address(
            salt=salt, initcode=initcode, fork=fork
        )
        if contract_address in self:
            raise ValueError(
                f"contract address already in pre-alloc: {contract_address}"
            )
        max_code_size = fork.max_code_size()
        if len(deploy_code) > max_code_size:
            raise ValueError(
                f"code too large: {len(deploy_code)} > {max_code_size}"
            )

        fork_deterministic_factory_address = (
            fork.deterministic_factory_predeploy_address()
        )
        if (
            fork_deterministic_factory_address is None
            and DETERMINISTIC_FACTORY_ADDRESS not in self
        ):
            self.__internal_setitem__(
                DETERMINISTIC_FACTORY_ADDRESS,
                Account(
                    nonce=1,
                    code=DETERMINISTIC_FACTORY_BYTECODE,
                    storage={},
                ),
            )

        self.__internal_setitem__(
            contract_address,
            Account(
                nonce=1,
                code=deploy_code,
                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 _deploy_contract(
        self,
        code: BytesConvertible,
        *,
        storage: Storage | StorageRootType | None,
        balance: NumberConvertible,
        nonce: NumberConvertible,
        address: Address | None,
        label: str | None,
        stub: str | None,
    ) -> Address:
        """Filler implementation of contract deployment."""
        if stub is not None:
            if stub not in self._stub_accounts:
                raise ValueError(
                    f"Stub '{stub}' not found in address stubs. "
                    "Provide --address-stubs with a mapping file."
                )
            account = self._stub_accounts[stub]
        else:
            if storage is None:
                storage = {}
            code = self.code_pre_processor(code)
            code_bytes = (
                bytes(code) if not isinstance(code, (bytes, str)) else code
            )
            max_code_size = self._fork.transitions_from().max_code_size()
            assert len(code_bytes) <= max_code_size, (
                f"code too large: {len(code_bytes)} > {max_code_size}"
            )

            account = Account(
                nonce=nonce,
                balance=balance,
                code=code,
                storage=storage,
            )

        if address is not None:
            assert address not in self, (
                f"address {address} already in allocation"
            )
            contract_address = address
        else:
            account_hash = account.hash()
            salt = self.get_next_account_salt(account_hash)
            contract_address = contract_address_from_hash(account_hash, salt)

        self.__internal_setitem__(contract_address, account)
        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,
        label: str | None,
        storage: Storage | None,
        code: BytesConvertible | None,
        delegation: Address | Literal["Self"] | None,
        nonce: NumberConvertible | None,
    ) -> EOA:
        """
        Filler implementation of EOA funding.

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

        if amount is None:
            amount = self._eoa_fund_amount_default
        if (
            Number(amount) > 0
            or storage is not None
            or code is not None
            or delegation is not None
            or (nonce is not None and Number(nonce) > 0)
        ):
            if code is not None and delegation is not None:
                raise Exception(
                    "code and delegation cannot be set at the same time"
                )
            if storage is None and delegation is None and code is None:
                nonce = Number(0 if nonce is None else nonce)
                account = Account(
                    nonce=nonce,
                    balance=amount,
                )
            else:
                # Type-4 transaction is sent to the EOA to set the storage, so
                # the nonce must be 1
                if delegation is not None:
                    if (
                        not isinstance(delegation, Address)
                        and delegation == "Self"
                    ):
                        # This is a placeholder value, since we don't know
                        # the address until the end of the function.
                        code = DELEGATION_DESIGNATION + b"Self"
                    else:
                        code = DELEGATION_DESIGNATION + delegation
                elif code is not None:
                    code = Bytes(code)
                else:
                    code = b""
                # 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=code,
                )

        else:
            account = Account()

        account_hash = account.hash()
        salt = self.get_next_account_salt(account_hash)
        eoa = eoa_from_hash(account_hash, salt)

        if account.nonce > 0:
            eoa.nonce = account.nonce

        if not isinstance(delegation, Address) and delegation == "Self":
            account = account.copy(code=DELEGATION_DESIGNATION + eoa)
        if account:
            self.__internal_setitem__(eoa, account)
        return eoa

    def _fund_address(
        self,
        address: Address,
        amount: int,
        *,
        minimum_balance: bool,
    ) -> None:
        """
        Filler implementation of address funding.
        """
        del minimum_balance
        self.__internal_setitem__(address, Account(balance=amount))

    def _nonexistent_account(self) -> Address:
        """
        Filler implementation of nonexistent_account.
        """
        salt = self.get_next_account_salt(EMPTY_ACCOUNT_HASH)
        return Address(eoa_from_hash(EMPTY_ACCOUNT_HASH, salt))

__init__(*args, fork, flags, stub_accounts=None, stub_eoas=None, **kwargs)

Initialize the pre-alloc.

Source code in packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/pre_alloc.py

def __init__(
    self,
    *args: Any,
    fork: Fork,
    flags: AllocFlags,
    stub_accounts: Dict[str, Account] | None = None,
    stub_eoas: Dict[str, EOA] | None = None,
    **kwargs: Any,
) -> None:
    """Initialize the pre-alloc."""
    super().__init__(
        *args,
        fork=fork,
        flags=flags,
        stub_eoas=stub_eoas,
        **kwargs,
    )
    if stub_accounts is not None:
        self._stub_accounts = stub_accounts

get_next_account_salt(account_hash)

Retrieve the next salt for this account.

Source code in packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/pre_alloc.py

def get_next_account_salt(self, account_hash: Hash) -> int:
    """Retrieve the next salt for this account."""
    salt = self._account_salt.get(account_hash, 0)
    self._account_salt[account_hash] = salt + 1
    return salt

code_pre_processor(code)

Pre-processes the code before setting it.

Source code in packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/pre_alloc.py

def code_pre_processor(self, code: BytesConvertible) -> BytesConvertible:
    """Pre-processes the code before setting it."""
    return code

modified_accounts_salt()

Return a salt if this pre-allocation was affected by setting addresses to hard-coded accounts or has pre-funded addresses.

Any modification the test does to a hard-coded address must affect this salt.

Source code in packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/pre_alloc.py

def modified_accounts_salt(self) -> int:
    """
    Return a salt if this pre-allocation was affected by setting addresses
    to hard-coded accounts or has pre-funded addresses.

    Any modification the test does to a hard-coded address must affect
    this salt.
    """
    if (
        not self._set_addresses
        and not self._pre_funded_addresses
        and not self._hardcoded_addresses_deployed_to
        and not self._deleted_addresses
    ):
        return 0

    # Build a hashable buffer from the modified accounts.
    buffer = b""
    altered_accounts = (
        self._set_addresses
        | self._pre_funded_addresses
        | self._hardcoded_addresses_deployed_to
    )
    if altered_accounts:
        buffer += b"\0"
        for altered_account in sorted(altered_accounts):
            buffer += altered_account
            account = self[altered_account]
            assert account is not None
            buffer += account.hash()
    if self._deleted_addresses:
        buffer += b"\1"
        for deleted_address in sorted(self._deleted_addresses):
            buffer += deleted_address

    return int.from_bytes(
        hashlib.sha256(buffer).digest()[:8], byteorder="big"
    )

compute_pre_alloc_group_hash(*, fork, genesis_environment, group_salt)

Hash (fork, env) in order to group tests by genesis config.

Source code in packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/pre_alloc.py

def compute_pre_alloc_group_hash(
    self,
    *,
    fork: Fork | TransitionFork,
    genesis_environment: Environment,
    group_salt: str | None,
) -> str:
    """Hash (fork, env) in order to group tests by genesis config."""
    fork_digest = hashlib.sha256(fork.name().encode("utf-8")).digest()
    fork_hash = int.from_bytes(fork_digest[:8], byteorder="big")
    combined_hash = (
        fork_hash
        ^ hash(genesis_environment)
        ^ self.modified_accounts_salt()
    )

    # Check if this pre-allocation has a group salt
    if group_salt:
        # Add custom salt to hash
        salt_hash = hashlib.sha256(group_salt.encode("utf-8")).digest()
        salt_int = int.from_bytes(salt_hash[:8], byteorder="big")
        combined_hash = combined_hash ^ salt_int

    return f"0x{combined_hash:016x}"

sha256_from_string(s)

Return SHA-256 hash of a string.

Source code in packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/pre_alloc.py

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

node_id_for_entropy(request, fork)

Return the node id with the fixture format name and fork name stripped.

Used in cases where we are filling for pre-alloc groups, and we take the name of the test as source of entropy to get a deterministic address when generating the pre-alloc grouping.

Removing the fixture format and the fork name from the node id before hashing results in the contracts and senders addresses being the same across fixture types and forks for the same test.

Source code in packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/pre_alloc.py

@pytest.fixture(scope="function")
def node_id_for_entropy(
    request: pytest.FixtureRequest, fork: Fork | None
) -> str:
    """
    Return the node id with the fixture format name and fork name stripped.

    Used in cases where we are filling for pre-alloc groups, and we take the
    name of the test as source of entropy to get a deterministic address when
    generating the pre-alloc grouping.

    Removing the fixture format and the fork name from the node id before
    hashing results in the contracts and senders addresses being the same
    across fixture types and forks for the same test.
    """
    node_id: str = request.node.nodeid
    # Strip xdist group suffix (e.g., @t8n-cache-abc12345) so entropy is
    # deterministic regardless of whether xdist is active.
    if "@" in node_id:
        node_id = node_id.rsplit("@", 1)[0]
    if fork is None:
        # FIXME: Static tests don't have a fork, so we need to get it from the
        # node.
        assert hasattr(request.node, "fork")
        fork = request.node.fork
    for fixture_format_name in ALL_FIXTURE_FORMAT_NAMES:
        if fixture_format_name in node_id:
            parts = node_id.split("::")
            test_file_path = parts[0]
            test_name = "::".join(parts[1:])
            stripped_test_name = test_name.replace(
                fixture_format_name, ""
            ).replace(fork.name(), "")
            return f"{test_file_path}::{stripped_test_name}"
    raise Exception(f"Fixture format name not found in test {node_id}")

eoa_by_index(i) cached

Return EOA by index.

Source code in packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/pre_alloc.py

@cache
def eoa_by_index(i: int) -> EOA:
    """Return EOA by index."""
    return EOA(key=TestPrivateKey + i if i != 1 else TestPrivateKey2, nonce=0)

stub_accounts(request)

Return stub accounts pre-populated during configuration.

Source code in packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/pre_alloc.py

@pytest.fixture(scope="session")
def stub_accounts(
    request: pytest.FixtureRequest,
) -> Dict[str, Account]:
    """Return stub accounts pre-populated during configuration."""
    return request.config.stash.get(stub_accounts_key, {})

stub_eoas(request)

Return stub EOAs pre-populated during configuration.

Source code in packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/pre_alloc.py

@pytest.fixture(scope="session")
def stub_eoas(
    request: pytest.FixtureRequest,
) -> Dict[str, EOA]:
    """Return stub EOAs pre-populated during configuration."""
    return request.config.stash.get(stub_eoas_key, {})

pre(alloc_flags, fork, request, stub_accounts, stub_eoas)

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

Source code in packages/testing/src/execution_testing/cli/pytest_commands/plugins/filler/pre_alloc.py

@pytest.fixture(scope="function")
def pre(
    alloc_flags: AllocFlags,
    fork: Fork | None,
    request: pytest.FixtureRequest,
    stub_accounts: Dict[str, Account],
    stub_eoas: Dict[str, EOA],
) -> Alloc:
    """Return default pre allocation for all tests (Empty alloc)."""
    # FIXME: Static tests don't have a fork so we need to get it from the node.
    actual_fork = fork
    if actual_fork is None:
        assert hasattr(request.node, "fork")
        actual_fork = request.node.fork

    return Alloc(
        flags=alloc_flags,
        fork=actual_fork,
        stub_accounts=stub_accounts,
        stub_eoas=stub_eoas,
    )