Add fork choice store pruning test

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

@@ -289,6 +289,7 @@ def make_fixture(self) -> Self:
                         store = store.on_gossip_attestation(
                             signed_attestation,
                             scheme=LEAN_ENV_TO_SCHEMES[self.lean_env],
+                            is_aggregator=step.is_aggregator,
                         )
 
                     case GossipAggregatedAttestationStep():

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

@@ -149,6 +149,15 @@ class AttestationStep(BaseForkChoiceStep):
     The framework fills in the attestation data and signature during make_fixture().
     """
 
+    is_aggregator: bool = False
+    """
+    Whether the node holds the aggregator role for this attestation.
+
+    Only aggregator nodes store gossip signatures in the raw signature pool.
+    Defaults to False so existing fillers preserve the behavior where gossip
+    attestations are validated but not stored.
+    """
+
     _filled_attestation: SignedAttestation | None = PrivateAttr(default=None)
     """The filled SignedAttestation, processed through the spec."""
 

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

@@ -180,6 +180,30 @@ class StoreChecks(CamelModel):
     attestation_checks: list[AttestationCheck] | None = None
     """Optional list of attestation content checks for specific validators."""
 
+    attestation_signature_target_slots: list[Slot] | None = None
+    """
+    Expected target slots present in attestation_signatures.
+
+    Compares the exact set of target checkpoint slots keyed in the raw gossip
+    signature map, independent of how many validators signed each target.
+    """
+
+    latest_new_aggregated_target_slots: list[Slot] | None = None
+    """
+    Expected target slots present in latest_new_aggregated_payloads.
+
+    Compares the exact set of target checkpoint slots keyed in the pending
+    aggregated proof map.
+    """
+
+    latest_known_aggregated_target_slots: list[Slot] | None = None
+    """
+    Expected target slots present in latest_known_aggregated_payloads.
+
+    Compares the exact set of target checkpoint slots keyed in the accepted
+    aggregated proof map.
+    """
+
     block_attestation_count: int | None = None
     """
     Expected number of aggregated attestations in the block body.
@@ -351,6 +375,24 @@ def _resolve(label: str) -> Bytes32:
                     )
                 check.validate_attestation(extracted[check.validator], label, step_index)
 
+        if "attestation_signature_target_slots" in fields:
+            assert self.attestation_signature_target_slots is not None
+            actual = sorted({data.target.slot for data in store.attestation_signatures})
+            expected = sorted(self.attestation_signature_target_slots)
+            _check("attestation_signatures.target_slots", actual, expected)
+
+        if "latest_new_aggregated_target_slots" in fields:
+            assert self.latest_new_aggregated_target_slots is not None
+            actual = sorted({data.target.slot for data in store.latest_new_aggregated_payloads})
+            expected = sorted(self.latest_new_aggregated_target_slots)
+            _check("latest_new_aggregated_payloads.target_slots", actual, expected)
+
+        if "latest_known_aggregated_target_slots" in fields:
+            assert self.latest_known_aggregated_target_slots is not None
+            actual = sorted({data.target.slot for data in store.latest_known_aggregated_payloads})
+            expected = sorted(self.latest_known_aggregated_target_slots)
+            _check("latest_known_aggregated_payloads.target_slots", actual, expected)
+
         # Block body attestation count
         if "block_attestation_count" in fields:
             if filled_block is None: