Skip to content

refactor(testing): fold api endpoint response logic into the test spec #1155

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
tcoratger merged 1 commit into from
Jun 18, 2026
Merged

refactor(testing): fold api endpoint response logic into the test spec #1155

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
335 changes: 146 additions & 189 deletions packages/testing/src/consensus_testing/test_fixtures/api_endpoint.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,16 +1,39 @@
"""API endpoint response conformance fixtures."""

from collections.abc import Callable
from typing import Any, ClassVar

from consensus_testing.genesis import build_anchor
from consensus_testing.test_fixtures.base import BaseConsensusFixture, BaseTestSpec
from lean_spec.base import StrictBaseModel
from lean_spec.node.metrics.registry import registry as metrics_registry
from lean_spec.spec.forks import Slot
from lean_spec.spec.forks.lstar import Store
from lean_spec.spec.forks.lstar.spec import LstarSpec
from lean_spec.spec.ssz import Uint64

REQUIRED_METRIC_NAMES = [
"lean_node_info",
"lean_node_start_time_seconds",
"lean_head_slot",
"lean_current_slot",
"lean_safe_target_slot",
"lean_fork_choice_block_processing_time_seconds",
"lean_attestations_valid_total",
"lean_attestations_invalid_total",
"lean_attestation_validation_time_seconds",
"lean_fork_choice_reorgs_total",
"lean_fork_choice_reorg_depth",
"lean_attestation_aggregate_coverage_validators",
"lean_attestation_aggregate_coverage_subnets",
"lean_attestation_aggregate_coverage_diff_validators",
"lean_latest_justified_slot",
"lean_latest_finalized_slot",
"lean_state_transition_time_seconds",
"lean_validators_count",
"lean_connected_peers",
]
"""Metric names every client must expose. Changing this list is a cross-client surface change."""


class EndpointResponseContract(StrictBaseModel):
"""The status, content type, and body one endpoint handler expects a client to return."""
Expand All @@ -25,183 +48,6 @@ class EndpointResponseContract(StrictBaseModel):
"""Expected response payload: a JSON object or a hex SSZ string."""


EndpointHandler = Callable[[Store, "ApiEndpointTest"], EndpointResponseContract]
"""Uniform signature for endpoint response builders, so the dispatch table has one call site."""


def _build_store(num_validators: int, genesis_time: int, anchor_slot: int = 0) -> Store:
"""
Build a deterministic store, genesis-only or an empty chain advanced to the anchor slot.

No attestations are injected, so justification and finalization stay at genesis.
"""
fork = LstarSpec()
# Walk the chain from genesis using empty blocks; slot 0 returns the genesis pair unchanged.
state, block = build_anchor(
fork=fork,
num_validators=num_validators,
anchor_slot=Slot(anchor_slot),
genesis_time=Uint64(genesis_time),
)
# No validator identity — fixture only reads store data, never signs.
return fork.create_store(state, block, validator_index=None)


def _health_response(_store: Store, _fixture: "ApiEndpointTest") -> EndpointResponseContract:
"""Static liveness check, independent of consensus state."""
return EndpointResponseContract(
status_code=200,
content_type="application/json",
body={"status": "healthy", "service": "lean-rpc-api"},
)


def _justified_response(store: Store, _fixture: "ApiEndpointTest") -> EndpointResponseContract:
"""Latest justified checkpoint as slot and root, with the root varying by validator count."""
return EndpointResponseContract(
status_code=200,
content_type="application/json",
body={
"slot": int(store.latest_justified.slot),
"root": "0x" + store.latest_justified.root.hex(),
},
)


def _finalized_state_response(
store: Store, _fixture: "ApiEndpointTest"
) -> EndpointResponseContract:
"""Full SSZ-encoded finalized state as hex bytes."""
state = store.states[store.latest_finalized.root]
return EndpointResponseContract(
status_code=200,
content_type="application/octet-stream",
body="0x" + state.encode_bytes().hex(),
)


def _fork_choice_response(store: Store, _fixture: "ApiEndpointTest") -> EndpointResponseContract:
"""Fork choice tree: blocks with weights, head, checkpoints, validator count."""
weights = LstarSpec().compute_block_weights(store)

# Only post-finalization blocks are relevant to head selection.
nodes = [
{
"root": "0x" + root.hex(),
"slot": int(block.slot),
"parent_root": "0x" + block.parent_root.hex(),
"proposer_index": int(block.proposer_index),
"weight": weights.get(root, 0),
}
for root, block in store.blocks.items()
if block.slot >= store.latest_finalized.slot
]

# Validator count from head state (most current view).
head_state = store.states.get(store.head)
return EndpointResponseContract(
status_code=200,
content_type="application/json",
body={
"nodes": nodes,
"head": "0x" + store.head.hex(),
"justified": {
"slot": int(store.latest_justified.slot),
"root": "0x" + store.latest_justified.root.hex(),
},
"finalized": {
"slot": int(store.latest_finalized.slot),
"root": "0x" + store.latest_finalized.root.hex(),
},
"safe_target": "0x" + store.safe_target.hex(),
"validator_count": len(head_state.validators) if head_state is not None else 0,
},
)


def _aggregator_status_response(
_store: Store, fixture: "ApiEndpointTest"
) -> EndpointResponseContract:
"""Current aggregator role as seeded by initial_is_aggregator."""
return EndpointResponseContract(
status_code=200,
content_type="application/json",
body={"is_aggregator": fixture.initial_is_aggregator},
)


def _metrics_response(_store: Store, _fixture: "ApiEndpointTest") -> EndpointResponseContract:
"""
Prometheus-format metrics scrape.

The body is dynamic, so the fixture pins only status, content type, and the metric names.
"""
from lean_spec.node.metrics.registry import registry as metrics_registry

# Enumerated from the metrics spec; changing the list is a cross-client surface change.
required_metric_names = [
"lean_node_info",
"lean_node_start_time_seconds",
"lean_head_slot",
"lean_current_slot",
"lean_safe_target_slot",
"lean_fork_choice_block_processing_time_seconds",
"lean_attestations_valid_total",
"lean_attestations_invalid_total",
"lean_attestation_validation_time_seconds",
"lean_fork_choice_reorgs_total",
"lean_fork_choice_reorg_depth",
"lean_attestation_aggregate_coverage_validators",
"lean_attestation_aggregate_coverage_subnets",
"lean_attestation_aggregate_coverage_diff_validators",
"lean_latest_justified_slot",
"lean_latest_finalized_slot",
"lean_state_transition_time_seconds",
"lean_validators_count",
"lean_connected_peers",
]
# Touch the registry so its removal trips this fixture instead of failing silently.
assert metrics_registry is not None
return EndpointResponseContract(
status_code=200,
content_type="text/plain; version=0.0.4; charset=utf-8",
body={"required_metric_names": required_metric_names},
)


def _aggregator_toggle_response(
_store: Store, fixture: "ApiEndpointTest"
) -> EndpointResponseContract:
"""Expected response after toggling the aggregator role, with the new and previous values."""
body = fixture.request_body
if not isinstance(body, dict) or not isinstance(body.get("enabled"), bool):
raise ValueError(
"POST /lean/v0/admin/aggregator fixture requires request_body "
"with a boolean 'enabled' field"
)
new_value = body["enabled"]
return EndpointResponseContract(
status_code=200,
content_type="application/json",
body={
"is_aggregator": new_value,
"previous": fixture.initial_is_aggregator,
},
)


_ENDPOINT_HANDLERS: dict[tuple[str, str], EndpointHandler] = {
("GET", "/lean/v0/health"): _health_response,
("GET", "/lean/v0/checkpoints/justified"): _justified_response,
("GET", "/lean/v0/states/finalized"): _finalized_state_response,
("GET", "/lean/v0/fork_choice"): _fork_choice_response,
("GET", "/lean/v0/admin/aggregator"): _aggregator_status_response,
("POST", "/lean/v0/admin/aggregator"): _aggregator_toggle_response,
("GET", "/metrics"): _metrics_response,
}
"""Maps (method, path) tuples to response builders."""


class ApiEndpointFixture(BaseConsensusFixture):
"""Emitted vector for API endpoint response conformance."""

Expand Down Expand Up @@ -259,23 +105,134 @@ class ApiEndpointTest(BaseTestSpec):

def generate(self) -> ApiEndpointFixture:
"""Build genesis store, compute expected response, emit the vector."""
handler = _ENDPOINT_HANDLERS.get((self.method, self.endpoint))
if handler is None:
raise ValueError(f"Unknown endpoint: {self.method} {self.endpoint}")

store = _build_store(
# Build a deterministic store: genesis-only, or an empty chain advanced to the anchor slot.
# No attestations are injected, so justification and finalization stay at genesis.
fork = LstarSpec()
# Walk the chain from genesis using empty blocks; slot 0 returns the genesis pair unchanged.
state, block = build_anchor(
fork=fork,
num_validators=self.genesis_params.get("numValidators", 4),
genesis_time=self.genesis_params.get("genesisTime", 0),
anchor_slot=self.genesis_params.get("anchorSlot", 0),
anchor_slot=Slot(self.genesis_params.get("anchorSlot", 0)),
genesis_time=Uint64(self.genesis_params.get("genesisTime", 0)),
)
response_contract = handler(store, self)
# No validator identity — fixture only reads store data, never signs.
store = fork.create_store(state, block, validator_index=None)

response = self._expected_response(store)
return ApiEndpointFixture(
endpoint=self.endpoint,
method=self.method,
genesis_params=self.genesis_params,
request_body=self.request_body,
initial_is_aggregator=self.initial_is_aggregator,
expected_status_code=response_contract.status_code,
expected_content_type=response_contract.content_type,
expected_body=response_contract.body,
expected_status_code=response.status_code,
expected_content_type=response.content_type,
expected_body=response.body,
)

def _expected_response(self, store: Store) -> EndpointResponseContract:
"""Compute the response a conforming client must return for the route under test."""
match (self.method, self.endpoint):
case ("GET", "/lean/v0/health"):
# Static liveness check, independent of consensus state.
return EndpointResponseContract(
status_code=200,
content_type="application/json",
body={"status": "healthy", "service": "lean-rpc-api"},
)

case ("GET", "/lean/v0/checkpoints/justified"):
# Latest justified checkpoint; the root varies with validator count.
return EndpointResponseContract(
status_code=200,
content_type="application/json",
body={
"slot": int(store.latest_justified.slot),
"root": "0x" + store.latest_justified.root.hex(),
},
)

case ("GET", "/lean/v0/states/finalized"):
# Full SSZ-encoded finalized state as hex bytes.
finalized_state = store.states[store.latest_finalized.root]
return EndpointResponseContract(
status_code=200,
content_type="application/octet-stream",
body="0x" + finalized_state.encode_bytes().hex(),
)

case ("GET", "/lean/v0/fork_choice"):
# Fork choice tree: blocks with weights, head, checkpoints, validator count.
weights = LstarSpec().compute_block_weights(store)

# Only post-finalization blocks are relevant to head selection.
nodes = [
{
"root": "0x" + root.hex(),
"slot": int(block.slot),
"parent_root": "0x" + block.parent_root.hex(),
"proposer_index": int(block.proposer_index),
"weight": weights.get(root, 0),
}
for root, block in store.blocks.items()
if block.slot >= store.latest_finalized.slot
]

# The head always has a stored state, so a missing one is a broken invariant.
head_state = store.states[store.head]
return EndpointResponseContract(
status_code=200,
content_type="application/json",
body={
"nodes": nodes,
"head": "0x" + store.head.hex(),
"justified": {
"slot": int(store.latest_justified.slot),
"root": "0x" + store.latest_justified.root.hex(),
},
"finalized": {
"slot": int(store.latest_finalized.slot),
"root": "0x" + store.latest_finalized.root.hex(),
},
"safe_target": "0x" + store.safe_target.hex(),
"validator_count": len(head_state.validators),
},
)

case ("GET", "/lean/v0/admin/aggregator"):
# Current aggregator role as seeded by the spec.
return EndpointResponseContract(
status_code=200,
content_type="application/json",
body={"is_aggregator": self.initial_is_aggregator},
)

case ("POST", "/lean/v0/admin/aggregator"):
# Toggling reports the new aggregator value and the previous one.
body = self.request_body
if not isinstance(body, dict) or not isinstance(body.get("enabled"), bool):
raise ValueError(
"POST /lean/v0/admin/aggregator fixture requires request_body "
"with a boolean 'enabled' field"
)
return EndpointResponseContract(
status_code=200,
content_type="application/json",
body={
"is_aggregator": body["enabled"],
"previous": self.initial_is_aggregator,
},
)

case ("GET", "/metrics"):
# The body is dynamic, so pin only status, content type, and the metric names.
# Touch the registry so its removal trips this fixture instead of failing silently.
assert metrics_registry is not None
return EndpointResponseContract(
status_code=200,
content_type="text/plain; version=0.0.4; charset=utf-8",
body={"required_metric_names": REQUIRED_METRIC_NAMES},
)

case _:
raise ValueError(f"Unknown endpoint: {self.method} {self.endpoint}")
Loading