Filler Plugin

A pytest plugin to fill tests and generate JSON fixtures.

FixtureOutput

Bases: BaseModel

Represents the output destination for generated test fixtures.

Source code in src/pytest_plugins/filler/fixture_output.py

class FixtureOutput(BaseModel):
    """Represents the output destination for generated test fixtures."""

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

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

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

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

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

    @property
    def pre_alloc_groups_folder_path(self) -> Path:
        """Return the path for pre-allocation groups folder."""
        engine_x_dir = BlockchainEngineXFixture.output_base_dir_name()
        return self.directory / engine_x_dir / "pre_alloc"

    @property
    def should_auto_enable_all_formats(self) -> bool:
        """Check if all formats should be auto-enabled due to tarball output."""
        return self.is_tarball

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

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

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

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

        if self.generate_pre_alloc_groups:
            # Phase 1: Directory must be completely empty
            return self.is_directory_empty()
        elif self.use_pre_alloc_groups:
            # Phase 2: Only pre-allocation groups must exist, no other files allowed
            if not self.pre_alloc_groups_folder_path.exists():
                return False
            # Check that only the pre-allocation group files exist
            existing_files = {f for f in self.directory.rglob("*") if f.is_file()}
            allowed_files = set(self.pre_alloc_groups_folder_path.rglob("*.json"))
            return existing_files == allowed_files
        else:
            # Normal filling: Directory must be empty
            return self.is_directory_empty()

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

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

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

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

        return " and ".join(summary_parts)

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

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

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

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

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

            if self.generate_pre_alloc_groups:
                raise ValueError(
                    f"Output directory '{self.directory}' must be completely empty for "
                    f"pre-allocation group generation (phase 1). Contains: {summary}. "
                    "Use --clean to remove all existing files."
                )
            elif self.use_pre_alloc_groups:
                if not self.pre_alloc_groups_folder_path.exists():
                    raise ValueError(
                        "Pre-allocation groups folder not found at "
                        f"'{self.pre_alloc_groups_folder_path}'. "
                        "Run phase 1 with --generate-pre-alloc-groups first."
                    )
            else:
                raise ValueError(
                    f"Output directory '{self.directory}' is not empty. "
                    f"Contains: {summary}. Use --clean to remove all existing files "
                    "or specify a different output directory."
                )

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

        # Create pre-allocation groups directory for phase 1
        if self.generate_pre_alloc_groups:
            self.pre_alloc_groups_folder_path.parent.mkdir(parents=True, exist_ok=True)

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

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

    @classmethod
    def from_config(cls, config: pytest.Config) -> "FixtureOutput":
        """Create a FixtureOutput instance from pytest configuration."""
        output_path = Path(config.getoption("output"))
        should_generate_all_formats = config.getoption("generate_all_formats")

        # Auto-enable --generate-all-formats for tarball output
        # Use same logic as is_tarball property
        if output_path.suffix == ".gz" and output_path.with_suffix("").suffix == ".tar":
            should_generate_all_formats = True

        return cls(
            output_path=output_path,
            flat_output=config.getoption("flat_output"),
            single_fixture_per_file=config.getoption("single_fixture_per_file"),
            clean=config.getoption("clean"),
            generate_pre_alloc_groups=config.getoption("generate_pre_alloc_groups"),
            use_pre_alloc_groups=config.getoption("use_pre_alloc_groups"),
            should_generate_all_formats=should_generate_all_formats,
        )

directory property

Return the actual directory path where fixtures will be written.

metadata_dir property

Return metadata directory to store fixture meta files.

is_tarball property

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

is_stdout property

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

pre_alloc_groups_folder_path property

Return the path for pre-allocation groups folder.

should_auto_enable_all_formats property

Check if all formats should be auto-enabled due to tarball output.

strip_tarball_suffix(path) staticmethod

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

Source code in src/pytest_plugins/filler/fixture_output.py

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

is_directory_empty()

Check if the output directory is empty.

Source code in src/pytest_plugins/filler/fixture_output.py

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

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

is_directory_usable_for_phase()

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

Source code in src/pytest_plugins/filler/fixture_output.py

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

    if self.generate_pre_alloc_groups:
        # Phase 1: Directory must be completely empty
        return self.is_directory_empty()
    elif self.use_pre_alloc_groups:
        # Phase 2: Only pre-allocation groups must exist, no other files allowed
        if not self.pre_alloc_groups_folder_path.exists():
            return False
        # Check that only the pre-allocation group files exist
        existing_files = {f for f in self.directory.rglob("*") if f.is_file()}
        allowed_files = set(self.pre_alloc_groups_folder_path.rglob("*.json"))
        return existing_files == allowed_files
    else:
        # Normal filling: Directory must be empty
        return self.is_directory_empty()

get_directory_summary()

Return a summary of directory contents for error reporting.

Source code in src/pytest_plugins/filler/fixture_output.py

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

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

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

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

    return " and ".join(summary_parts)

create_directories(is_master)

Create output and metadata directories if needed.

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

Source code in src/pytest_plugins/filler/fixture_output.py

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

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

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

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

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

        if self.generate_pre_alloc_groups:
            raise ValueError(
                f"Output directory '{self.directory}' must be completely empty for "
                f"pre-allocation group generation (phase 1). Contains: {summary}. "
                "Use --clean to remove all existing files."
            )
        elif self.use_pre_alloc_groups:
            if not self.pre_alloc_groups_folder_path.exists():
                raise ValueError(
                    "Pre-allocation groups folder not found at "
                    f"'{self.pre_alloc_groups_folder_path}'. "
                    "Run phase 1 with --generate-pre-alloc-groups first."
                )
        else:
            raise ValueError(
                f"Output directory '{self.directory}' is not empty. "
                f"Contains: {summary}. Use --clean to remove all existing files "
                "or specify a different output directory."
            )

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

    # Create pre-allocation groups directory for phase 1
    if self.generate_pre_alloc_groups:
        self.pre_alloc_groups_folder_path.parent.mkdir(parents=True, exist_ok=True)

create_tarball()

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

Source code in src/pytest_plugins/filler/fixture_output.py

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

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

from_config(config) classmethod

Create a FixtureOutput instance from pytest configuration.

Source code in src/pytest_plugins/filler/fixture_output.py

@classmethod
def from_config(cls, config: pytest.Config) -> "FixtureOutput":
    """Create a FixtureOutput instance from pytest configuration."""
    output_path = Path(config.getoption("output"))
    should_generate_all_formats = config.getoption("generate_all_formats")

    # Auto-enable --generate-all-formats for tarball output
    # Use same logic as is_tarball property
    if output_path.suffix == ".gz" and output_path.with_suffix("").suffix == ".tar":
        should_generate_all_formats = True

    return cls(
        output_path=output_path,
        flat_output=config.getoption("flat_output"),
        single_fixture_per_file=config.getoption("single_fixture_per_file"),
        clean=config.getoption("clean"),
        generate_pre_alloc_groups=config.getoption("generate_pre_alloc_groups"),
        use_pre_alloc_groups=config.getoption("use_pre_alloc_groups"),
        should_generate_all_formats=should_generate_all_formats,
    )

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 src/pytest_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 src/pytest_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 src/pytest_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 src/pytest_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 src/pytest_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

    @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_groups = PreAllocGroups(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)
        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 pre-allocation groups. "
                f"Please check the pre-allocation groups file at: {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(self, hash_key: str, group: PreAllocGroup) -> None:
        """
        Update or add a pre-allocation group.

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

        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_groups is None:
            self.pre_alloc_groups = PreAllocGroups(root={})

        self.pre_alloc_groups[hash_key] = group

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

        pre_alloc_folder = self.fixture_output.pre_alloc_groups_folder_path
        pre_alloc_folder.mkdir(parents=True, exist_ok=True)
        self.pre_alloc_groups.to_folder(pre_alloc_folder)

    def aggregate_pre_alloc_groups(self, worker_groups: PreAllocGroups) -> None:
        """
        Aggregate pre-alloc groups from a worker process (xdist support).

        Args:
            worker_groups: Pre-alloc groups from a worker process.

        """
        if self.pre_alloc_groups is None:
            self.pre_alloc_groups = PreAllocGroups(root={})

        for hash_key, group in worker_groups.root.items():
            if hash_key in self.pre_alloc_groups:
                # Merge if exists (should not happen in practice)
                existing = self.pre_alloc_groups[hash_key]
                if existing.pre != group.pre:
                    raise ValueError(
                        f"Conflicting pre-alloc groups for hash {hash_key}: "
                        f"existing={self.pre_alloc_groups[hash_key].pre}, new={group.pre}"
                    )
            else:
                self.pre_alloc_groups[hash_key] = group

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 src/pytest_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 src/pytest_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 src/pytest_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 pre-allocation groups. "
            f"Please check the pre-allocation groups file at: {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(hash_key, group)

Update or add a pre-allocation group.

Parameters:

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

Raises:

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

Source code in src/pytest_plugins/filler/filler.py

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

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

    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_groups is None:
        self.pre_alloc_groups = PreAllocGroups(root={})

    self.pre_alloc_groups[hash_key] = group

save_pre_alloc_groups()

Save pre-allocation groups to disk.

Source code in src/pytest_plugins/filler/filler.py

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

    pre_alloc_folder = self.fixture_output.pre_alloc_groups_folder_path
    pre_alloc_folder.mkdir(parents=True, exist_ok=True)
    self.pre_alloc_groups.to_folder(pre_alloc_folder)

aggregate_pre_alloc_groups(worker_groups)

Aggregate pre-alloc groups from a worker process (xdist support).

Parameters:

Name	Type	Description	Default
worker_groups	PreAllocGroups	Pre-alloc groups from a worker process.	required

Source code in src/pytest_plugins/filler/filler.py

def aggregate_pre_alloc_groups(self, worker_groups: PreAllocGroups) -> None:
    """
    Aggregate pre-alloc groups from a worker process (xdist support).

    Args:
        worker_groups: Pre-alloc groups from a worker process.

    """
    if self.pre_alloc_groups is None:
        self.pre_alloc_groups = PreAllocGroups(root={})

    for hash_key, group in worker_groups.root.items():
        if hash_key in self.pre_alloc_groups:
            # Merge if exists (should not happen in practice)
            existing = self.pre_alloc_groups[hash_key]
            if existing.pre != group.pre:
                raise ValueError(
                    f"Conflicting pre-alloc groups for hash {hash_key}: "
                    f"existing={self.pre_alloc_groups[hash_key].pre}, new={group.pre}"
                )
        else:
            self.pre_alloc_groups[hash_key] = group

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 src/pytest_plugins/filler/filler.py

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

    This function enables significant space savings in 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 Alloc(diff)

default_output_directory()

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

Source code in src/pytest_plugins/filler/filler.py

def default_output_directory() -> str:
    """
    Directory (default) to store the generated test fixtures. Defined as a
    function to allow for easier testing.
    """
    return "./fixtures"

default_html_report_file_path()

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

Source code in src/pytest_plugins/filler/filler.py

def default_html_report_file_path() -> str:
    """
    File path (default) to store the generated HTML test report. Defined as a
    function to allow for easier testing.
    """
    return ".meta/report_fill.html"

pytest_addoption(parser)

Add command-line options to pytest.

Source code in src/pytest_plugins/filler/filler.py

def pytest_addoption(parser: pytest.Parser):
    """Add command-line options to pytest."""
    evm_group = parser.getgroup("evm", "Arguments defining evm executable behavior")
    evm_group.addoption(
        "--evm-bin",
        action="store",
        dest="evm_bin",
        type=Path,
        default=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(
        "--traces",
        action="store_true",
        dest="evm_collect_traces",
        default=None,
        help="Collect traces of the execution information from the transition tool.",
    )
    evm_group.addoption(
        "--verify-fixtures",
        action="store_true",
        dest="verify_fixtures",
        default=False,
        help=(
            "Verify generated fixture JSON files using geth's evm blocktest command. "
            "By default, the same evm binary as for the t8n tool is used. A different (geth) evm "
            "binary may be specified via --verify-fixtures-bin, this must be specified if filling "
            "with a non-geth t8n tool that does not support blocktest."
        ),
    )
    evm_group.addoption(
        "--verify-fixtures-bin",
        action="store",
        dest="verify_fixtures_bin",
        type=Path,
        default=None,
        help=(
            "Path to an evm executable that provides the `blocktest` command. "
            "Default: The first (geth) 'evm' entry in PATH."
        ),
    )

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

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

pytest_configure(config)

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

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

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

1. @pytest.hookimpl(tryfirst=True) is applied to ensure that this hook is called before the pytest-html plugin's pytest_configure to ensure that it uses the modified htmlpath option.

Source code in src/pytest_plugins/filler/filler.py

@pytest.hookimpl(tryfirst=True)
def pytest_configure(config):
    """
    Pytest hook called after command line options have been parsed and before
    test collection begins.

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

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

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

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

    # Initialize filling session
    config.filling_session = FillingSession.from_config(config)

    if is_help_or_collectonly_mode(config):
        return

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

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

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

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

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

pytest_report_header(config)

Add lines to pytest's console output header.

Source code in src/pytest_plugins/filler/filler.py

@pytest.hookimpl(trylast=True)
def pytest_report_header(config: pytest.Config):
    """Add lines to pytest's console output header."""
    if is_help_or_collectonly_mode(config):
        return
    t8n_version = config.stash[metadata_key]["Tools"]["t8n"]
    return [(f"{t8n_version}")]

pytest_report_teststatus(report, config)

Modify test results in pytest's terminal output.

We use this:

1. To disable test session progress report if we're writing the JSON fixtures to stdout to be read by a consume command on stdin. I.e., don't write this type of output to the console:

   ...x...
   

Source code in src/pytest_plugins/filler/filler.py

@pytest.hookimpl(tryfirst=True)
def pytest_report_teststatus(report, config: pytest.Config):
    """
    Modify test results in pytest's terminal output.

    We use this:

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

pytest_terminal_summary(terminalreporter, exitstatus, config)

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

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

Source code in src/pytest_plugins/filler/filler.py

@pytest.hookimpl(hookwrapper=True, trylast=True)
def pytest_terminal_summary(
    terminalreporter: TerminalReporter, exitstatus: int, config: pytest.Config
):
    """
    Modify pytest's terminal summary to emphasize that no tests were ran.

    Emphasize that fixtures have only been filled; they must now be executed to
    actually run the tests.
    """
    yield
    if config.fixture_output.is_stdout or hasattr(config, "workerinput"):  # type: ignore[attr-defined]
        return
    stats = terminalreporter.stats
    if "passed" in stats and stats["passed"]:
        # Custom message for Phase 1 (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
            pre_alloc_groups: PreAllocGroups
            if config.pluginmanager.hasplugin("xdist"):
                # Load pre-allocation groups from disk
                pre_alloc_groups = PreAllocGroups.from_folder(
                    config.fixture_output.pre_alloc_groups_folder_path  # type: ignore[attr-defined]
                )
            else:
                assert session_instance.pre_alloc_groups is not None
                pre_alloc_groups = session_instance.pre_alloc_groups

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

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

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

pytest_metadata(metadata)

Add or remove metadata to/from the pytest report.

Source code in src/pytest_plugins/filler/filler.py

def pytest_metadata(metadata):
    """Add or remove metadata to/from the pytest report."""
    metadata.pop("JAVA_HOME", None)

pytest_html_results_table_header(cells)

Customize the table headers of the HTML report table.

Source code in src/pytest_plugins/filler/filler.py

def pytest_html_results_table_header(cells):
    """Customize the table headers of the HTML report table."""
    cells.insert(3, '<th class="sortable" data-column-type="fixturePath">JSON Fixture File</th>')
    cells.insert(4, '<th class="sortable" data-column-type="evmDumpDir">EVM Dump Dir</th>')
    del cells[-1]  # Remove the "Links" column

pytest_html_results_table_row(report, cells)

Customize the table rows of the HTML report table.

Source code in src/pytest_plugins/filler/filler.py

def pytest_html_results_table_row(report, cells):
    """Customize the table rows of the HTML report table."""
    if hasattr(report, "user_properties"):
        user_props = dict(report.user_properties)
        if (
            report.passed
            and "fixture_path_absolute" in user_props
            and "fixture_path_relative" in user_props
        ):
            fixture_path_absolute = user_props["fixture_path_absolute"]
            fixture_path_relative = user_props["fixture_path_relative"]
            fixture_path_link = (
                f'<a href="{fixture_path_absolute}" target="_blank">{fixture_path_relative}</a>'
            )
            cells.insert(3, f"<td>{fixture_path_link}</td>")
        elif report.failed:
            cells.insert(3, "<td>Fixture unavailable</td>")
        if "evm_dump_dir" in user_props:
            if user_props["evm_dump_dir"] is None:
                cells.insert(
                    4, "<td>For t8n debug info use <code>--evm-dump-dir=path --traces</code></td>"
                )
            else:
                evm_dump_dir = user_props.get("evm_dump_dir")
                if evm_dump_dir == "N/A":
                    evm_dump_entry = "N/A"
                else:
                    evm_dump_entry = f'<a href="{evm_dump_dir}" target="_blank">{evm_dump_dir}</a>'
                cells.insert(4, f"<td>{evm_dump_entry}</td>")
    del cells[-1]  # Remove the "Links" column

pytest_runtest_makereport(item, call)

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

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

Source code in src/pytest_plugins/filler/filler.py

@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_makereport(item, call):
    """
    Make each test's fixture json path available to the test report via
    user_properties.

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

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

pytest_html_report_title(report)

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

Source code in src/pytest_plugins/filler/filler.py

def pytest_html_report_title(report):
    """Set the HTML report title (pytest-html plugin)."""
    report.title = "Fill Test Report"

evm_bin(request)

Return configured evm tool binary path used to run t8n.

Source code in src/pytest_plugins/filler/filler.py

@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 src/pytest_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")

t8n_server_url(request)

Return configured t8n server url.

Source code in src/pytest_plugins/filler/filler.py

@pytest.fixture(autouse=True, scope="session")
def t8n_server_url(request: pytest.FixtureRequest) -> str | None:
    """Return configured t8n server url."""
    return request.config.getoption("t8n_server_url")

t8n(request, evm_bin, t8n_server_url)

Return configured transition tool.

Source code in src/pytest_plugins/filler/filler.py

@pytest.fixture(autouse=True, scope="session")
def t8n(
    request: pytest.FixtureRequest, evm_bin: Path | None, t8n_server_url: str | None
) -> Generator[TransitionTool, None, None]:
    """Return configured transition tool."""
    kwargs = {
        "trace": request.config.getoption("evm_collect_traces"),
    }
    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 not t8n.exception_mapper.reliable:
        warnings.warn(
            f"The t8n tool that is currently being used to fill tests ({t8n.__class__.__name__}) "
            "does not provide reliable exception messages. This may lead to false positives when "
            "writing tests and extra care should be taken when writing tests that produce "
            "exceptions.",
            stacklevel=2,
        )
    yield t8n
    t8n.shutdown()

do_fixture_verification(request, verify_fixtures_bin)

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

Source code in src/pytest_plugins/filler/filler.py

@pytest.fixture(scope="session")
def do_fixture_verification(
    request: pytest.FixtureRequest, verify_fixtures_bin: Path | None
) -> bool:
    """
    Return True if evm statetest or evm blocktest should be ran on the
    generated fixture JSON files.
    """
    do_fixture_verification = False
    if verify_fixtures_bin:
        do_fixture_verification = True
    if request.config.getoption("verify_fixtures"):
        do_fixture_verification = True
    return do_fixture_verification

evm_fixture_verification(request, do_fixture_verification, evm_bin, verify_fixtures_bin)

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

Source code in src/pytest_plugins/filler/filler.py

@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:
        return
    try:
        evm_fixture_verification = FixtureConsumerTool.from_binary_path(
            binary_path=Path(verify_fixtures_bin),
            trace=request.config.getoption("evm_collect_traces"),
        )
    except Exception:
        if reused_evm_bin:
            pytest.exit(
                "The binary specified in --evm-bin could not be recognized as a known "
                "FixtureConsumerTool. Either remove --verify-fixtures or set "
                "--verify-fixtures-bin to a known fixture consumer binary.",
                returncode=pytest.ExitCode.USAGE_ERROR,
            )
        else:
            pytest.exit(
                "Specified binary in --verify-fixtures-bin could not be recognized as a known "
                "FixtureConsumerTool. Please see `GethFixtureConsumer` for an example "
                "of how a new fixture consumer can be defined.",
                returncode=pytest.ExitCode.USAGE_ERROR,
            )
    yield evm_fixture_verification

base_dump_dir(request)

Path to base directory to dump the evm debug output.

Source code in src/pytest_plugins/filler/filler.py

@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 src/pytest_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 src/pytest_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 src/pytest_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 src/pytest_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} with value {val}.", stacklevel=2
            )
    config["environment"] = environment_properties

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

dump_dir_parameter_level(request, base_dump_dir, filler_path)

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

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

Source code in src/pytest_plugins/filler/filler.py

@pytest.fixture(scope="function")
def dump_dir_parameter_level(
    request: pytest.FixtureRequest, base_dump_dir: Path | None, filler_path: Path
) -> Path | None:
    """
    Directory to dump evm transition tool debug output on a test parameter
    level.

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

get_fixture_collection_scope(fixture_name, config)

Return the appropriate scope to write fixture JSON files.

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

Source code in src/pytest_plugins/filler/filler.py

def get_fixture_collection_scope(fixture_name, config):
    """
    Return the appropriate scope to write fixture JSON files.

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

reference_spec(request)

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

See get_ref_spec_from_module.

Source code in src/pytest_plugins/filler/filler.py

@pytest.fixture(autouse=True, scope="module")
def reference_spec(request) -> None | ReferenceSpec:
    """
    Pytest fixture that returns the reference spec defined in a module.

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

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

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

Source code in src/pytest_plugins/filler/filler.py

@pytest.fixture(scope=get_fixture_collection_scope)
def fixture_collector(
    request: pytest.FixtureRequest,
    do_fixture_verification: bool,
    evm_fixture_verification: FixtureConsumer,
    filler_path: Path,
    base_dump_dir: Path | None,
    fixture_output: FixtureOutput,
) -> Generator[FixtureCollector, None, None]:
    """
    Return configured fixture collector instance used for all tests
    in one test module.
    """
    # Dynamically load the 'static_filler' and 'solc' plugins if needed
    if request.config.getoption("fill_static_tests_enabled"):
        request.config.pluginmanager.import_plugin("pytest_plugins.filler.static_filler")
        request.config.pluginmanager.import_plugin("pytest_plugins.solc.solc")

    fixture_collector = FixtureCollector(
        output_dir=fixture_output.directory,
        flat_output=fixture_output.flat_output,
        fill_static_tests=request.config.getoption("fill_static_tests_enabled"),
        single_fixture_per_file=fixture_output.single_fixture_per_file,
        filler_path=filler_path,
        base_dump_dir=base_dump_dir,
    )
    yield fixture_collector
    fixture_collector.dump_fixtures()
    if do_fixture_verification:
        fixture_collector.verify_fixture_files(evm_fixture_verification)

filler_path(request)

Return directory containing the tests to execute.

Source code in src/pytest_plugins/filler/filler.py

@pytest.fixture(autouse=True, scope="session")
def filler_path(request: pytest.FixtureRequest) -> Path:
    """Return directory containing the tests to execute."""
    return request.config.getoption("filler_path")

node_to_test_info(node)

Return test info of the current node item.

Source code in src/pytest_plugins/filler/filler.py

def node_to_test_info(node: pytest.Item) -> TestInfo:
    """Return test info of the current node item."""
    return TestInfo(
        name=node.name,
        id=node.nodeid,
        original_name=node.originalname,  # type: ignore
        module_path=Path(node.path),
    )

commit_hash_or_tag()

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

Source code in src/pytest_plugins/filler/filler.py

@pytest.fixture(scope="session")
def commit_hash_or_tag() -> str:
    """Cache the git commit hash or tag for the entire test session."""
    return get_current_commit_hash_or_tag()

fixture_source_url(request, commit_hash_or_tag)

Return URL to the fixture source.

Source code in src/pytest_plugins/filler/filler.py

@pytest.fixture(scope="function")
def fixture_source_url(
    request: pytest.FixtureRequest,
    commit_hash_or_tag: str,
) -> str:
    """Return URL to the fixture source."""
    if hasattr(request.node, "github_url"):
        return request.node.github_url
    function_line_number = request.function.__code__.co_firstlineno
    module_relative_path = os.path.relpath(request.function.__code__.co_filename)

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

base_test_parametrizer(cls)

Generate pytest.fixture for a given BaseTest subclass.

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

Source code in src/pytest_plugins/filler/filler.py

def base_test_parametrizer(cls: Type[BaseTest]):
    """
    Generate pytest.fixture for a given BaseTest subclass.

    Implementation detail: All spec fixtures must be scoped on test function level to avoid
    leakage between tests.
    """
    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,
        reference_spec: ReferenceSpec,
        pre: Alloc,
        output_dir: Path,
        dump_dir_parameter_level: Path | None,
        fixture_collector: FixtureCollector,
        test_case_description: str,
        fixture_source_url: str,
        gas_benchmark_value: int,
    ):
        """
        Fixture used to instantiate an auto-fillable BaseTest object from within
        a test function.

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

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

        class BaseTestWrapper(cls):  # type: ignore
            def __init__(self, *args, **kwargs):
                kwargs["t8n_dump_dir"] = dump_dir_parameter_level
                if "pre" not in kwargs:
                    kwargs["pre"] = pre
                if "expected_benchmark_gas_used" not in kwargs:
                    kwargs["expected_benchmark_gas_used"] = gas_benchmark_value
                kwargs |= {
                    p: request.getfixturevalue(p)
                    for p in cls_fixture_parameters
                    if p not in kwargs
                }

                super(BaseTestWrapper, self).__init__(*args, **kwargs)
                self._request = request
                self._operation_mode = request.config.op_mode

                # Get the filling session from config
                session: FillingSession = request.config.filling_session  # type: ignore[attr-defined]

                # 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
                    self.update_pre_alloc_groups(
                        session.pre_alloc_groups, fork, request.node.nodeid
                    )
                    return  # Skip fixture generation in phase 1

                # Phase 2: Use pre-allocation groups (only for BlockchainEngineXFixture)
                pre_alloc_hash = None
                if FixtureFillingPhase.PRE_ALLOC_GENERATION in fixture_format.format_phases:
                    pre_alloc_hash = self.compute_pre_alloc_group_hash(fork=fork)
                    group = session.get_pre_alloc_group(pre_alloc_hash)
                    self.pre = group.pre

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

                # 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
                ):
                    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
                        )

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

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

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

        return BaseTestWrapper

    return base_test_parametrizer_func

pytest_generate_tests(metafunc)

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

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

Source code in src/pytest_plugins/filler/filler.py

def pytest_generate_tests(metafunc: pytest.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.
    """
    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 src/pytest_plugins/filler/filler.py

def pytest_collection_modifyitems(
    config: pytest.Config, items: List[pytest.Item | pytest.Function]
):
    """
    Remove pre-Paris tests parametrized to generate hive type fixtures; these
    can't be used in the Hive Pyspec Simulator.

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

    These can't be handled in this plugins pytest_generate_tests() as the fork
    parametrization occurs in the forks plugin.
    """
    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 "fork" not in params or params["fork"] is None:
            items_for_removal.append(i)
            continue
        fork: Fork = params["fork"]
        spec_type, fixture_format = get_spec_format_for_item(params)
        if isinstance(fixture_format, NotSetType):
            items_for_removal.append(i)
            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 = get_transition_fork_predecessor(fork)
                item._nodeid = item._nodeid.replace(
                    f"fork_{fork.name()}",
                    f"fork_{base_fork.name()}",
                )

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

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 src/pytest_plugins/filler/filler.py

def pytest_sessionfinish(session: pytest.Session, exitstatus: int):
    """
    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.
    """
    # 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:
        session_instance.save_pre_alloc_groups()
        return

    if xdist.is_xdist_worker(session):
        return

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

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

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

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

Pre-alloc specifically conditioned for test filling.

pytest_addoption(parser)

Add command-line options to pytest.

Source code in src/pytest_plugins/filler/pre_alloc.py

def pytest_addoption(parser: pytest.Parser):
    """Add command-line options to pytest."""
    pre_alloc_group = parser.getgroup(
        "pre_alloc", "Arguments defining pre-allocation behavior during test filling."
    )

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

AllocMode

Bases: IntEnum

Allocation mode for the state.

Source code in src/pytest_plugins/filler/pre_alloc.py

class AllocMode(IntEnum):
    """Allocation mode for the state."""

    PERMISSIVE = 0
    STRICT = 1

Alloc

Bases: Alloc

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

Source code in src/pytest_plugins/filler/pre_alloc.py

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

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

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

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

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

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

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

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

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

        contract_address.label = label
        return contract_address

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

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

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

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

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

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

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

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

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

        """
        eoa = next(self._eoa_iterator)

        return Address(eoa)

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

Initialize allocation with the given properties.

Source code in src/pytest_plugins/filler/pre_alloc.py

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

__setitem__(address, account)

Set account associated with an address.

Source code in src/pytest_plugins/filler/pre_alloc.py

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

code_pre_processor(code, *, evm_code_type)

Pre-processes the code before setting it.

Source code in src/pytest_plugins/filler/pre_alloc.py

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

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

Deploy a contract to the allocation.

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

Source code in src/pytest_plugins/filler/pre_alloc.py

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

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

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

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

    contract_address.label = label
    return contract_address

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

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

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

Source code in src/pytest_plugins/filler/pre_alloc.py

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

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

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

fund_address(address, amount)

Fund an address with a given amount.

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

Source code in src/pytest_plugins/filler/pre_alloc.py

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

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

empty_account()

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

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

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

Returns:

Name	Type	Description
Address	Address	The address of the created empty account.

Source code in src/pytest_plugins/filler/pre_alloc.py

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

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

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

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

    """
    eoa = next(self._eoa_iterator)

    return Address(eoa)

alloc_mode(request)

Return allocation mode for the tests.

Source code in src/pytest_plugins/filler/pre_alloc.py

@pytest.fixture(scope="session")
def alloc_mode(request: pytest.FixtureRequest) -> AllocMode:
    """Return allocation mode for the tests."""
    if request.config.getoption("strict_alloc"):
        return AllocMode.STRICT
    return AllocMode.PERMISSIVE

contract_start_address(request)

Return starting address for contract deployment.

Source code in src/pytest_plugins/filler/pre_alloc.py

@pytest.fixture(scope="session")
def contract_start_address(request: pytest.FixtureRequest) -> int:
    """Return starting address for contract deployment."""
    return int(request.config.getoption("test_contract_start_address"), 0)

contract_address_increments(request)

Return address increment for contract deployment.

Source code in src/pytest_plugins/filler/pre_alloc.py

@pytest.fixture(scope="session")
def contract_address_increments(request: pytest.FixtureRequest) -> int:
    """Return address increment for contract deployment."""
    return int(request.config.getoption("test_contract_address_increments"), 0)

sha256_from_string(s)

Return SHA-256 hash of a string.

Source code in src/pytest_plugins/filler/pre_alloc.py

def sha256_from_string(s: str) -> int:
    """Return SHA-256 hash of a string."""
    return int.from_bytes(sha256(s.encode("utf-8")).digest(), "big")

contract_address_iterator(request, contract_start_address, contract_address_increments)

Return iterator over contract addresses with dynamic scoping.

Source code in src/pytest_plugins/filler/pre_alloc.py

@pytest.fixture(scope="function")
def contract_address_iterator(
    request: pytest.FixtureRequest,
    contract_start_address: int,
    contract_address_increments: int,
) -> Iterator[Address]:
    """Return iterator over contract addresses with dynamic scoping."""
    if request.config.getoption(
        "generate_pre_alloc_groups", default=False
    ) or request.config.getoption("use_pre_alloc_groups", default=False):
        # Use a starting address that is derived from the test node
        contract_start_address = sha256_from_string(request.node.nodeid)
    return iter(
        Address((contract_start_address + (i * contract_address_increments)) % 2**160)
        for i in count()
    )

eoa_by_index(i) cached

Return EOA by index.

Source code in src/pytest_plugins/filler/pre_alloc.py

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

eoa_iterator(request)

Return iterator over EOAs copies with dynamic scoping.

Source code in src/pytest_plugins/filler/pre_alloc.py

@pytest.fixture(scope="function")
def eoa_iterator(request: pytest.FixtureRequest) -> Iterator[EOA]:
    """Return iterator over EOAs copies with dynamic scoping."""
    if request.config.getoption(
        "generate_pre_alloc_groups", default=False
    ) or request.config.getoption("use_pre_alloc_groups", default=False):
        # Use a starting address that is derived from the test node
        eoa_start_pk = sha256_from_string(request.node.nodeid)
        return iter(
            EOA(
                key=(eoa_start_pk + i)
                % 0xFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFEBAAEDCE6AF48A03BBFD25E8CD0364141,
                nonce=0,
            )
            for i in count()
        )
    return iter(eoa_by_index(i).copy() for i in count())

evm_code_type(request)

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

Source code in src/pytest_plugins/filler/pre_alloc.py

@pytest.fixture(autouse=True)
def evm_code_type(request: pytest.FixtureRequest) -> EVMCodeType:
    """Return default EVM code type for all tests (LEGACY)."""
    parameter_evm_code_type = request.config.getoption("evm_code_type")
    if parameter_evm_code_type is not None:
        assert type(parameter_evm_code_type) is EVMCodeType, "Invalid EVM code type"
        return parameter_evm_code_type
    return EVMCodeType.LEGACY

pre(alloc_mode, contract_address_iterator, eoa_iterator, evm_code_type)

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

Source code in src/pytest_plugins/filler/pre_alloc.py

@pytest.fixture(scope="function")
def pre(
    alloc_mode: AllocMode,
    contract_address_iterator: Iterator[Address],
    eoa_iterator: Iterator[EOA],
    evm_code_type: EVMCodeType,
) -> Alloc:
    """Return default pre allocation for all tests (Empty alloc)."""
    return Alloc(
        alloc_mode=alloc_mode,
        contract_address_iterator=contract_address_iterator,
        eoa_iterator=eoa_iterator,
        evm_code_type=evm_code_type,
    )