Filler Plugin

A pytest plugin that provides fixtures that fill tests and generate fixtures.

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

default_output_directory()

The default directory 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:
    """
    The default directory to store the generated test fixtures. Defined as a
    function to allow for easier testing.
    """
    return "./fixtures"

default_html_report_file_path()

The default file path 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:
    """
    The default file path to store the generated HTML test report. Defined as a
    function to allow for easier testing.
    """
    return ".meta/report_fill.html"

strip_output_tarball_suffix(output)

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

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

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

is_output_stdout(output)

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

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

def is_output_stdout(output: Path) -> bool:
    """
    Returns True if the fixture output is configured to be stdout.
    """
    return strip_output_tarball_suffix(output).name == "stdout"

pytest_addoption(parser)

Adds command-line options to pytest.

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

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

    test_group = parser.getgroup("tests", "Arguments defining filler location and output")
    test_group.addoption(
        "--filler-path",
        action="store",
        dest="filler_path",
        default="./tests/",
        type=Path,
        help="Path to filler directives",
    )
    test_group.addoption(
        "--output",
        action="store",
        dest="output",
        type=Path,
        default=Path(default_output_directory()),
        help=(
            "Directory path to store the generated test fixtures. "
            "If the specified path ends in '.tar.gz', then the specified tarball is additionally "
            "created (the fixtures are still written to the specified path without the '.tar.gz' "
            f"suffix). Can be deleted. Default: '{default_output_directory()}'."
        ),
    )
    test_group.addoption(
        "--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(
        "--index",
        action="store_true",
        dest="generate_index",
        default=False,
        help="Generate an index file for all produced fixtures.",
    )

    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="",
        help="Path to dump the transition tool debug output.",
    )

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.
    """
    if config.option.collectonly:
        return
    if not config.getoption("disable_html") and config.getoption("htmlpath") is None:
        # generate an html report by default, unless explicitly disabled
        config.option.htmlpath = (
            strip_output_tarball_suffix(config.getoption("output"))
            / default_html_report_file_path()
        )
    # Instantiate the transition tool here to check that the binary path/trace option is valid.
    # This ensures we only raise an error once, if appropriate, instead of for every test.
    t8n = TransitionTool.from_binary_path(
        binary_path=config.getoption("evm_bin"), trace=config.getoption("evm_collect_traces")
    )
    if (
        isinstance(config.getoption("numprocesses"), int)
        and config.getoption("numprocesses") > 0
        and "Besu" in str(t8n.detect_binary_pattern)
    ):
        pytest.exit(
            "The Besu t8n tool does not work well with the xdist plugin; use -n=0.",
            returncode=pytest.ExitCode.USAGE_ERROR,
        )

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

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

pytest_report_header(config)

Add lines to pytest's console output header

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

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

pytest_report_teststatus(report, config)

This hook modifies the 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

def pytest_report_teststatus(report, config: pytest.Config):
    """
    This hook modifies the 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 is_output_stdout(config.getoption("output")):
        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
    stats = terminalreporter.stats
    if "passed" in stats and stats["passed"]:
        # append / to indicate this is a directory
        output_dir = str(strip_output_tarball_suffix(config.getoption("output"))) + "/"
        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)

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

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

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

@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_makereport(item, call):
    """
    This hook is called when each test is run and a report is being made.

    Make each test's fixture json path available to the test report via
    user_properties.
    """
    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)

Returns the 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:
    """
    Returns the configured evm tool binary path used to run t8n.
    """
    return request.config.getoption("evm_bin")

verify_fixtures_bin(request)

Returns the 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:
    """
    Returns the configured evm tool binary path used to run statetest or
    blocktest.
    """
    return request.config.getoption("verify_fixtures_bin")

t8n(request, evm_bin)

Returns the 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) -> Generator[TransitionTool, None, None]:
    """
    Returns the configured transition tool.
    """
    t8n = TransitionTool.from_binary_path(
        binary_path=evm_bin, trace=request.config.getoption("evm_collect_traces")
    )
    yield t8n
    t8n.shutdown()

do_fixture_verification(request, verify_fixtures_bin)

Returns 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:
    """
    Returns 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(do_fixture_verification, evm_bin, verify_fixtures_bin)

Returns the 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(
    do_fixture_verification: bool,
    evm_bin: Path,
    verify_fixtures_bin: Path | None,
) -> Generator[TransitionTool | None, None, None]:
    """
    Returns the configured evm binary for executing statetest and blocktest
    commands used to verify generated JSON fixtures.
    """
    if not do_fixture_verification:
        yield None
        return
    if not verify_fixtures_bin and evm_bin:
        verify_fixtures_bin = evm_bin
    evm_fixture_verification = TransitionTool.from_binary_path(binary_path=verify_fixtures_bin)
    if not evm_fixture_verification.blocktest_subcommand:
        pytest.exit(
            "Only geth's evm tool is supported to verify fixtures: "
            "Either remove --verify-fixtures or set --verify-fixtures-bin to a Geth evm binary.",
            returncode=pytest.ExitCode.USAGE_ERROR,
        )
    yield evm_fixture_verification
    evm_fixture_verification.shutdown()

base_dump_dir(request)

The 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:
    """
    The 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

is_output_tarball(request)

Returns 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(request: pytest.FixtureRequest) -> bool:
    """
    Returns True if the output directory is a tarball.
    """
    output: Path = request.config.getoption("output")
    if output.suffix == ".gz" and output.with_suffix("").suffix == ".tar":
        return True
    return False

output_dir(request, is_output_tarball)

Returns the directory to store the generated test fixtures.

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

@pytest.fixture(scope="session")
def output_dir(request: pytest.FixtureRequest, is_output_tarball: bool) -> Path:
    """
    Returns the directory to store the generated test fixtures.
    """
    output = request.config.getoption("output")
    if is_output_tarball:
        return strip_output_tarball_suffix(output)
    return output

output_metadata_dir(output_dir)

Returns the metadata directory to store fixture meta files.

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

@pytest.fixture(scope="session")
def output_metadata_dir(output_dir: Path) -> Path:
    """
    Returns the metadata directory to store fixture meta files.
    """
    return output_dir / ".meta"

create_properties_file(request, output_dir, output_metadata_dir)

Creates an 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, output_dir: Path, output_metadata_dir: Path
) -> None:
    """
    Creates an ini file with fixture build properties in the fixture output
    directory.
    """
    if is_output_stdout(request.config.getoption("output")):
        return
    if not output_dir.exists():
        output_dir.mkdir(parents=True)
    if not output_metadata_dir.exists():
        output_metadata_dir.mkdir(parents=True)

    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}.")
    config["environment"] = environment_properties

    ini_filename = 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)

create_tarball(request, output_dir, is_output_tarball)

Create a tarball of json files the output directory if the configured output ends with '.tar.gz'.

Only include .json and .ini files in the archive.

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

@pytest.fixture(scope="session", autouse=True)
def create_tarball(
    request: pytest.FixtureRequest, output_dir: Path, is_output_tarball: bool
) -> Generator[None, None, None]:
    """
    Create a tarball of json files the output directory if the configured
    output ends with '.tar.gz'.

    Only include .json and .ini files in the archive.
    """
    yield
    if is_output_tarball:
        source_dir = output_dir
        tarball_filename = request.config.getoption("output")
        with tarfile.open(tarball_filename, "w:gz") as tar:
            for file in source_dir.rglob("*"):
                if file.suffix in {".json", ".ini"}:
                    arcname = Path("fixtures") / file.relative_to(source_dir)
                    tar.add(file, arcname=arcname)

dump_dir_parameter_level(request, base_dump_dir, filler_path)

The 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:
    """
    The 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 is_output_stdout(config.getoption("output")):
        return "session"
    if config.getoption("single_fixture_per_file"):
        return "function"
    return "module"

fixture_collector(request, do_fixture_verification, evm_fixture_verification, filler_path, base_dump_dir, output_dir, session_temp_folder, generate_index)

Returns the 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: TransitionTool,
    filler_path: Path,
    base_dump_dir: Path | None,
    output_dir: Path,
    session_temp_folder: Path | None,
    generate_index: bool,
) -> Generator[FixtureCollector, None, None]:
    """
    Returns the configured fixture collector instance used for all tests
    in one test module.
    """
    if session_temp_folder is not None:
        fixture_collector_count_file_name = "fixture_collector_count"
        fixture_collector_count_file = session_temp_folder / fixture_collector_count_file_name
        fixture_collector_count_file_lock = (
            session_temp_folder / f"{fixture_collector_count_file_name}.lock"
        )
        with FileLock(fixture_collector_count_file_lock):
            if fixture_collector_count_file.exists():
                with open(fixture_collector_count_file, "r") as f:
                    fixture_collector_count = int(f.read())
            else:
                fixture_collector_count = 0
            fixture_collector_count += 1
            with open(fixture_collector_count_file, "w") as f:
                f.write(str(fixture_collector_count))

    fixture_collector = FixtureCollector(
        output_dir=output_dir,
        flat_output=request.config.getoption("flat_output"),
        single_fixture_per_file=request.config.getoption("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)

    fixture_collector_count = 0
    if session_temp_folder is not None:
        with FileLock(fixture_collector_count_file_lock):
            with open(fixture_collector_count_file, "r") as f:
                fixture_collector_count = int(f.read())
            fixture_collector_count -= 1
            with open(fixture_collector_count_file, "w") as f:
                f.write(str(fixture_collector_count))
    if generate_index and fixture_collector_count == 0:
        generate_fixtures_index(
            output_dir, quiet_mode=True, force_flag=False, disable_infer_format=False
        )

filler_path(request)

Returns the 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:
    """
    Returns the directory containing the tests to execute.
    """
    return request.config.getoption("filler_path")

node_to_test_info(node)

Returns the 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:
    """
    Returns the test info of the current node item.
    """
    return TestInfo(
        name=node.name,
        id=node.nodeid,
        original_name=node.originalname,  # type: ignore
        path=Path(node.path),
    )

fixture_source_url(request)

Returns the 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) -> str:
    """
    Returns the URL to the fixture source.
    """
    function_line_number = request.function.__code__.co_firstlineno
    module_relative_path = os.path.relpath(request.module.__file__)
    hash_or_tag = get_current_commit_hash_or_tag()
    github_url = generate_github_url(
        module_relative_path, branch_or_commit_or_tag=hash_or_tag, line_number=function_line_number
    )
    return github_url

base_test_parametrizer(cls)

Generates a 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]):
    """
    Generates a pytest.fixture for a given BaseTest subclass.

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

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

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

        When parametrize, indirect must be used along with the fixture format as value.
        """
        fixture_format = request.param
        assert issubclass(fixture_format, BaseFixture)

        class BaseTestWrapper(cls):  # type: ignore
            def __init__(self, *args, **kwargs):
                kwargs["t8n_dump_dir"] = dump_dir_parameter_level
                if "pre" not in kwargs:
                    kwargs["pre"] = pre
                super(BaseTestWrapper, self).__init__(*args, **kwargs)
                fixture = self.generate(
                    request=request,
                    t8n=t8n,
                    fork=fork,
                    fixture_format=fixture_format,
                    eips=eips,
                )
                fixture.fill_info(
                    t8n.version(),
                    test_case_description,
                    fixture_source_url=fixture_source_url,
                    ref_spec=reference_spec,
                )

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

        return BaseTestWrapper

    return base_test_parametrizer_func

pytest_generate_tests(metafunc)

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

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

def pytest_generate_tests(metafunc: pytest.Metafunc):
    """
    Pytest hook used to dynamically generate test cases for each fixture format a given
    test spec supports.
    """
    for test_type in SPEC_TYPES:
        if test_type.pytest_parameter_name() in metafunc.fixturenames:
            metafunc.parametrize(
                [test_type.pytest_parameter_name()],
                [
                    pytest.param(
                        fixture_format,
                        id=fixture_format.fixture_format_name.lower(),
                        marks=[getattr(pytest.mark, fixture_format.fixture_format_name.lower())],
                    )
                    for fixture_format in test_type.supported_fixture_formats
                ],
                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.

This 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]):
    """
    Remove pre-Paris tests parametrized to generate hive type fixtures; these
    can't be used in the Hive Pyspec Simulator.

    This can't be handled in this plugins pytest_generate_tests() as the fork
    parametrization occurs in the forks plugin.
    """
    for item in items[:]:  # use a copy of the list, as we'll be modifying it
        if isinstance(item, EIPSpecTestItem):
            continue
        params: Dict[str, Any] = item.callspec.params  # type: ignore
        if "fork" not in params or params["fork"] is None:
            items.remove(item)
            continue
        fork: Fork = params["fork"]
        for spec_name in [spec_type.pytest_parameter_name() for spec_type in SPEC_TYPES]:
            if spec_name in params and not params[spec_name].supports_fork(fork):
                items.remove(item)
                break
        for marker in item.iter_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)

Pre-alloc specifically conditioned for test filling.

pytest_addoption(parser)

Adds command-line options to pytest.

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

def pytest_addoption(parser: pytest.Parser):
    """
    Adds 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,
    ):
        """
        Initializes the 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):
        """
        Sets the 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 = {},
        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 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
                nonce = Number(1 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))

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

Initializes the 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,
):
    """
    Initializes the 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)

Sets the account associated with an address.

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

def __setitem__(self, address: Address | FixedSizeBytesConvertible, account: Account | None):
    """
    Sets the 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={}, 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 = {},
    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 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
            nonce = Number(1 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))

alloc_mode(request)

Returns the 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:
    """
    Returns the allocation mode for the tests.
    """
    if request.config.getoption("strict_alloc"):
        return AllocMode.STRICT
    return AllocMode.PERMISSIVE

contract_start_address(request)

Returns the 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:
    """
    Returns the starting address for contract deployment.
    """
    return int(request.config.getoption("test_contract_start_address"), 0)

contract_address_increments(request)

Returns the 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:
    """
    Returns the address increment for contract deployment.
    """
    return int(request.config.getoption("test_contract_address_increments"), 0)

contract_address_iterator(contract_start_address, contract_address_increments)

Returns an iterator over contract addresses.

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

@pytest.fixture(scope="function")
def contract_address_iterator(
    contract_start_address: int,
    contract_address_increments: int,
) -> Iterator[Address]:
    """
    Returns an iterator over contract addresses.
    """
    return iter(
        Address(contract_start_address + (i * contract_address_increments)) for i in count()
    )

eoa_by_index(i) cached

Returns an EOA by index.

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

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

eoa_iterator()

Returns an iterator over EOAs copies.

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

@pytest.fixture(scope="function")
def eoa_iterator() -> Iterator[EOA]:
    """
    Returns an iterator over EOAs copies.
    """
    return iter(eoa_by_index(i).copy() for i in count())

evm_code_type(request)

Returns the 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:
    """
    Returns the 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)

Returns the 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:
    """
    Returns the 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,
    )