feat(consume): consume block production hive simulator

src/cli/pytest_commands/consume.py

@@ -47,6 +47,10 @@ def get_command_logic_test_paths(command_name: str) -> List[Path]:
         command_logic_test_paths = [
             base_path / "simulators" / "simulator_logic" / "test_via_sync.py"
         ]
+    elif command_name == "production":
+        command_logic_test_paths = [
+            base_path / "simulators" / "simulator_logic" / "test_via_production.py"
+        ]
     elif command_name == "direct":
         command_logic_test_paths = [base_path / "direct" / "test_via_direct.py"]
     else:
@@ -116,6 +120,12 @@ def sync() -> None:
     pass
 
 
+@consume_command(is_hive=True)
+def production() -> None:
+    """Client builds blocks from mempool transactions (tests block production)."""
+    pass
+
+
 @consume.command(
     context_settings={"ignore_unknown_options": True},
 )

src/cli/pytest_commands/processors.py

@@ -104,6 +104,8 @@ def process_args(self, args: List[str]) -> List[str]:
             modified_args.extend(["-p", "pytest_plugins.consume.simulators.engine.conftest"])
         elif self.command_name == "sync":
             modified_args.extend(["-p", "pytest_plugins.consume.simulators.sync.conftest"])
+        elif self.command_name == "production":
+            modified_args.extend(["-p", "pytest_plugins.consume.simulators.production.conftest"])
         elif self.command_name == "rlp":
             modified_args.extend(["-p", "pytest_plugins.consume.simulators.rlp.conftest"])
         else:

src/pytest_plugins/consume/simulators/base.py

@@ -27,7 +27,11 @@ def check_live_port(test_suite_name: str) -> Literal[8545, 8551]:
     """Port used by hive to check for liveness of the client."""
     if test_suite_name == "eest/consume-rlp":
         return 8545
-    elif test_suite_name in {"eest/consume-engine", "eest/consume-sync"}:
+    elif test_suite_name in {
+        "eest/consume-engine",
+        "eest/consume-sync",
+        "eest/consume-production",
+    }:
         return 8551
     raise ValueError(
         f"Unexpected test suite name '{test_suite_name}' while setting HIVE_CHECK_LIVE_PORT."

src/pytest_plugins/consume/simulators/production/__init__.py

@@ -0,0 +1,3 @@
+"""
+Consume Production simulator. Tests block PRODUCTION (building) instead of validation.
+"""

src/pytest_plugins/consume/simulators/production/conftest.py

@@ -0,0 +1,127 @@
+"""
+Pytest fixtures for the `consume production` simulator.
+
+Tests block PRODUCTION (not just validation) by having clients build blocks
+from mempool transactions using forkchoiceUpdated + getPayload.
+"""
+
+import io
+from typing import Mapping
+
+import pytest
+from hive.client import Client
+
+from ethereum_test_exceptions import ExceptionMapper
+from ethereum_test_fixtures import BlockchainEngineFixture
+from ethereum_test_rpc import EngineRPC
+
+pytest_plugins = (
+    "pytest_plugins.pytest_hive.pytest_hive",
+    "pytest_plugins.consume.simulators.base",
+    "pytest_plugins.consume.simulators.single_test_client",
+    "pytest_plugins.consume.simulators.test_case_description",
+    "pytest_plugins.consume.simulators.timing_data",
+    "pytest_plugins.consume.simulators.exceptions",
+)
+
+
+def pytest_configure(config: pytest.Config) -> None:
+    """Set the supported fixture formats for the production simulator."""
+    config.supported_fixture_formats = [BlockchainEngineFixture]  # type: ignore[attr-defined]
+
+
+def pytest_collection_modifyitems(items: list[pytest.Item]) -> None:
+    """
+    Filter out tests that don't meet production simulator requirements.
+
+    Requirements:
+    - Must have exactly one transaction per payload (no multi-tx blocks)
+    - Payload must be valid (we're testing production, not validation)
+    """
+    for item in items:
+        if not hasattr(item, "callspec"):
+            continue
+
+        # Only process if this is a production test
+        if "test_blockchain_via_production" not in item.nodeid:
+            continue
+
+        # Get the fixture from parameters
+        fixture = item.callspec.params.get("fixture")
+        if not isinstance(fixture, BlockchainEngineFixture):
+            continue
+
+        # Filter: only single-transaction payloads
+        has_multi_tx_payload = False
+        has_invalid_payload = False
+        has_zero_tx_payload = False
+
+        for payload in fixture.payloads:
+            # Count transactions in this payload
+            tx_count = len(payload.params[0].transactions)
+
+            if tx_count == 0:
+                has_zero_tx_payload = True
+                break
+
+            if tx_count > 1:
+                has_multi_tx_payload = True
+                break
+
+            # Skip invalid payloads (we test production, not validation)
+            if not payload.valid():
+                has_invalid_payload = True
+                break
+
+        if has_zero_tx_payload:
+            item.add_marker(
+                pytest.mark.skip(
+                    reason="Production simulator: zero-transaction payloads not supported"
+                )
+            )
+        elif has_multi_tx_payload:
+            item.add_marker(
+                pytest.mark.skip(
+                    reason="Production simulator: multi-transaction payloads not supported"
+                )
+            )
+        elif has_invalid_payload:
+            item.add_marker(
+                pytest.mark.skip(reason="Production simulator: only tests valid block production")
+            )
+
+
+@pytest.fixture(scope="function")
+def engine_rpc(client: Client, client_exception_mapper: ExceptionMapper | None) -> EngineRPC:
+    """Initialize engine RPC client for the execution client under test."""
+    if client_exception_mapper:
+        return EngineRPC(
+            f"http://{client.ip}:8551",
+            response_validation_context={
+                "exception_mapper": client_exception_mapper,
+            },
+        )
+    return EngineRPC(f"http://{client.ip}:8551")
+
+
+@pytest.fixture(scope="module")
+def test_suite_name() -> str:
+    """The name of the hive test suite used in this simulator."""
+    return "eest/consume-production"
+
+
+@pytest.fixture(scope="module")
+def test_suite_description() -> str:
+    """The description of the hive test suite used in this simulator."""
+    return (
+        "Test block PRODUCTION (not validation) by having clients build blocks from "
+        "mempool transactions using forkchoiceUpdated + getPayload flow."
+    )
+
+
+@pytest.fixture(scope="function")
+def client_files(buffered_genesis: io.BufferedReader) -> Mapping[str, io.BufferedReader]:
+    """Define the files that hive will start the client with."""
+    files = {}
+    files["/genesis.json"] = buffered_genesis
+    return files

src/pytest_plugins/consume/simulators/production/helpers/__init__.py

@@ -0,0 +1 @@
+"""Helper functions for production simulator."""

src/pytest_plugins/consume/simulators/production/helpers/block_building.py

@@ -0,0 +1,62 @@
+"""Helper functions for block production testing."""
+
+import time
+from typing import Any
+
+from ethereum_test_base_types import Bytes, Hash
+from ethereum_test_rpc import EthRPC
+
+
+def wait_for_transaction_in_mempool(
+    eth_rpc: EthRPC,
+    tx_hash: Hash,
+    timeout: int = 10,
+    poll_interval: float = 0.1,
+) -> bool:
+    """
+    Wait for a transaction to appear in the mempool.
+
+    Returns True if transaction found, False if timeout reached.
+    """
+    start = time.time()
+    while time.time() - start < timeout:
+        try:
+            tx = eth_rpc.get_transaction_by_hash(tx_hash)
+            if tx is not None:
+                return True
+        except Exception:
+            pass
+        time.sleep(poll_interval)
+
+    return False
+
+
+def wait_for_payload_ready(
+    engine_rpc: Any,
+    payload_id: Bytes,
+    get_payload_version: int,
+    timeout: float = 5.0,
+    poll_interval: float = 0.1,
+) -> Any:
+    """
+    Poll until payload is ready to be retrieved.
+
+    Returns the built payload response when ready.
+    Raises TimeoutError if not ready within timeout.
+    """
+    start = time.time()
+    last_exception = None
+
+    while time.time() - start < timeout:
+        try:
+            built_payload_response = engine_rpc.get_payload(
+                payload_id=payload_id,
+                version=get_payload_version,
+            )
+            return built_payload_response
+        except Exception as e:
+            last_exception = e
+            time.sleep(poll_interval)
+
+    elapsed = time.time() - start
+    raise TimeoutError(f"Payload not ready after {elapsed:.2f}s. Last error: {last_exception}")