Add wrong-slot STF test with filler support

packages/testing/src/consensus_testing/test_fixtures/state_transition.py

@@ -61,8 +61,8 @@ class StateTransitionTest(BaseConsensusFixture):
     """
     The filled Blocks, processed through the specs.
 
-    This is a private attribute not part of the model schema. Tests cannot set this.
-    The framework populates it during make_fixture().
+    This is a private attribute not part of the model schema. Tests cannot set
+    this. The framework populates it during make_fixture().
     """
 
     post: StateExpectation | None = None
@@ -76,6 +76,9 @@ class StateTransitionTest(BaseConsensusFixture):
     expect_exception: type[Exception] | None = None
     """Expected exception type for invalid tests."""
 
+    expect_exception_message: str | None = None
+    """Expected exception message for invalid tests."""
+
     @field_serializer("blocks", when_used="json")
     def serialize_blocks(self, value: List[BlockSpec]) -> List[dict[str, Any]]:
         """
@@ -139,11 +142,15 @@ def make_fixture(self) -> "StateTransitionTest":
                 filled_blocks.append(block)
 
                 # Use cached state if available, otherwise run state transition
-                state = (
-                    cached_state
-                    if cached_state is not None
-                    else state.state_transition(block=block, valid_signatures=True)
-                )
+                if cached_state is not None:
+                    state = cached_state
+                elif getattr(block_spec, "skip_slot_processing", False):
+                    state = state.process_block(block)
+                else:
+                    state = state.state_transition(
+                        block=block,
+                        valid_signatures=True,
+                    )
 
             actual_post_state = state
         except (AssertionError, ValueError) as e:
@@ -168,6 +175,12 @@ def make_fixture(self) -> "StateTransitionTest":
                     f"Expected {self.expect_exception.__name__} "
                     f"but got {type(exception_raised).__name__}: {exception_raised}"
                 )
+            if self.expect_exception_message is not None:
+                if str(exception_raised) != self.expect_exception_message:
+                    raise AssertionError(
+                        f"Expected exception message '{self.expect_exception_message}' "
+                        f"but got '{exception_raised}'"
+                    )
 
         # Validate post-state expectations if provided
         if self.post is not None and actual_post_state is not None:
@@ -198,12 +211,16 @@ def _build_block_from_spec(self, spec: BlockSpec, state: State) -> tuple[Block,
         # Use provided proposer index or compute it
         proposer_index = spec.proposer_index or Uint64(int(spec.slot) % len(state.validators))
 
-        # Use provided parent root or compute it
+        temp_state: State | None = None
+        if not spec.skip_slot_processing:
+            temp_state = state.process_slots(spec.slot)
+
+        # Use provided parent_root or compute it
         if spec.parent_root is not None:
             parent_root = spec.parent_root
         else:
-            temp_state = state.process_slots(spec.slot)
-            parent_root = hash_tree_root(temp_state.latest_block_header)
+            source_state = temp_state if temp_state is not None else state
+            parent_root = hash_tree_root(source_state.latest_block_header)
 
         # Extract attestations from body if provided
         aggregated_attestations = (
@@ -221,25 +238,24 @@ def _build_block_from_spec(self, spec: BlockSpec, state: State) -> tuple[Block,
             )
             return block, None
 
-        # For invalid tests, return incomplete block without processing
-        if self.expect_exception is not None:
-            block = Block(
-                slot=spec.slot,
-                proposer_index=proposer_index,
-                parent_root=parent_root,
-                state_root=Bytes32.zero(),
-                body=spec.body or BlockBody(attestations=aggregated_attestations),
-            )
-            return block, None
+        temp_block = Block(
+            slot=spec.slot,
+            proposer_index=proposer_index,
+            parent_root=parent_root,
+            state_root=Bytes32.zero(),
+            body=spec.body or BlockBody(attestations=aggregated_attestations),
+        )
+
+        if self.expect_exception is not None or spec.skip_slot_processing:
+            return temp_block, None
 
-        # Build the block using the state for standard case
-        #
         # Convert aggregated attestations to plain attestations to build block
         plain_attestations = [
             Attestation(validator_id=vid, data=agg.data)
             for agg in aggregated_attestations
             for vid in agg.aggregation_bits.to_validator_indices()
         ]
+
         block, post_state, _, _ = state.build_block(
             slot=spec.slot,
             proposer_index=proposer_index,

packages/testing/src/consensus_testing/test_types/block_spec.py

@@ -110,3 +110,10 @@ class BlockSpec(CamelModel):
     Defaults to True (valid signature).
     If False, the proposer attestation will be given a dummy/invalid signature.
     """
+
+    skip_slot_processing: bool = False
+    """
+    If True, the state transition fixture skips automatic slot advancement before
+    processing this block. Useful for tests that intentionally exercise slot
+    mismatch failures.
+    """

tests/consensus/devnet/state_transition/test_block_processing.py

@@ -273,6 +273,44 @@ def test_block_with_invalid_state_root(
     )
 
 
+def test_block_with_wrong_slot(state_transition_test: StateTransitionTestFiller) -> None:
+    """
+    Test that blocks with mismatched slot are rejected.
+
+    Scenario
+    --------
+    Attempt to process a block at slot 1, but the block claims to be
+    at slot 2.
+
+    Expected Behavior
+    -----------------
+    Block processing fails with AssertionError: "Block slot mismatch"
+
+    Why This Matters
+    ----------------
+    Ensures temporal consistency:
+    - Blocks can't lie about their slot
+    - Prevents time manipulation attacks
+    - Maintains protocol timing integrity
+    - Essential for slot-based consensus
+    """
+    pre_state = generate_pre_state()
+    pre_state = pre_state.process_slots(Slot(1))
+
+    state_transition_test(
+        pre=pre_state,
+        blocks=[
+            BlockSpec(
+                slot=Slot(2),
+                skip_slot_processing=True,
+            ),
+        ],
+        post=None,
+        expect_exception=AssertionError,
+        expect_exception_message="Block slot mismatch",
+    )
+
+
 def test_block_extends_deep_chain(
     state_transition_test: StateTransitionTestFiller,
 ) -> None: