Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
27 changes: 2 additions & 25 deletions src/lean_spec/subspecs/networking/peer.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,13 +2,12 @@

from __future__ import annotations

from dataclasses import dataclass, field
from time import time
from dataclasses import dataclass

from .enr import ENR
from .reqresp import Status
from .transport import PeerId
from .types import ConnectionState, Direction, ForkDigest, Multiaddr
from .types import ConnectionState, Direction, Multiaddr


@dataclass(slots=True)
Expand Down Expand Up @@ -38,9 +37,6 @@ class PeerInfo:
address: Multiaddr | None = None
"""Last known network address for this peer."""

last_seen: float = field(default_factory=time)
"""Unix timestamp of last successful interaction."""

enr: ENR | None = None
"""Cached ENR from peer configuration. Contains eth2 fork_digest for compatibility checks."""

Expand All @@ -50,22 +46,3 @@ class PeerInfo:
def is_connected(self) -> bool:
"""Check if peer has an active connection."""
return self.state == ConnectionState.CONNECTED

def update_last_seen(self) -> None:
"""Update the last seen timestamp to now."""
self.last_seen = time()

@property
def fork_digest(self) -> ForkDigest | None:
"""
Get the peer's fork_digest from cached ENR.

Returns:
4-byte fork_digest or None if ENR/eth2 data unavailable.
"""
if self.enr is None:
return None
eth2_data = self.enr.eth2_data
if eth2_data is None:
return None
return eth2_data.fork_digest
72 changes: 2 additions & 70 deletions tests/lean_spec/subspecs/networking/test_peer.py
Original file line number Diff line number Diff line change
@@ -1,15 +1,10 @@
"""Tests for minimal peer module."""

import time

from lean_spec.subspecs.networking import PeerId
from lean_spec.subspecs.networking.enr import ENR
from lean_spec.subspecs.networking.enr.eth2 import FAR_FUTURE_EPOCH
from lean_spec.subspecs.networking.enr.keys import EnrKey
from lean_spec.subspecs.networking.peer import PeerInfo
from lean_spec.subspecs.networking.reqresp import Status
from lean_spec.subspecs.networking.types import ConnectionState, Direction, Multiaddr, SeqNumber
from lean_spec.types import Bytes32, Bytes64, Checkpoint, Slot
from lean_spec.subspecs.networking.types import ConnectionState, Direction, Multiaddr
from lean_spec.types import Bytes32, Checkpoint, Slot


def peer(name: str) -> PeerId:
Expand Down Expand Up @@ -77,58 +72,6 @@ def test_is_connected(self) -> None:
info.state = ConnectionState.DISCONNECTING
assert not info.is_connected()

def test_update_last_seen(self) -> None:
"""update_last_seen() updates timestamp."""
info = PeerInfo(peer_id=peer("test"))
original_time = info.last_seen

# Small delay to ensure time difference
time.sleep(0.01)

info.update_last_seen()
assert info.last_seen > original_time


class TestPeerInfoForkDigest:
"""Tests for PeerInfo fork_digest property."""

def _make_enr_with_eth2(self, fork_digest_bytes: bytes) -> ENR:
"""Create a minimal ENR with eth2 data for testing."""
# Create eth2 bytes: fork_digest(4) + next_fork_version(4) + next_fork_epoch(8)
eth2_bytes = (
fork_digest_bytes + fork_digest_bytes + int(FAR_FUTURE_EPOCH).to_bytes(8, "little")
)
return ENR(
signature=Bytes64(b"\x00" * 64),
seq=SeqNumber(1),
pairs={EnrKey("eth2"): eth2_bytes, EnrKey("id"): b"v4"},
)

def test_fork_digest_none_without_enr(self) -> None:
"""fork_digest returns None when no ENR is set."""
info = PeerInfo(peer_id=peer("test"))
assert info.fork_digest is None

def test_fork_digest_none_without_eth2(self) -> None:
"""fork_digest returns None when ENR has no eth2 data."""
# ENR without eth2 key
enr = ENR(
signature=Bytes64(b"\x00" * 64),
seq=SeqNumber(1),
pairs={EnrKey("id"): b"v4"},
)
info = PeerInfo(peer_id=peer("test"), enr=enr)
assert info.fork_digest is None

def test_fork_digest_returns_bytes(self) -> None:
"""fork_digest returns 4-byte fork_digest from ENR eth2 data."""
fork_bytes = b"\x12\x34\x56\x78"
enr = self._make_enr_with_eth2(fork_bytes)
info = PeerInfo(peer_id=peer("test"), enr=enr)

assert info.fork_digest is not None
assert bytes(info.fork_digest) == fork_bytes

def test_enr_and_status_fields(self) -> None:
"""Test that enr and status fields exist and default to None."""
info = PeerInfo(peer_id=peer("test"))
Expand All @@ -151,14 +94,3 @@ def test_status_can_be_set(self) -> None:
assert info.status is not None
assert info.status.finalized.slot == Slot(100)
assert info.status.head.slot == Slot(200)

def test_update_last_seen_updates_timestamp(self) -> None:
"""Test that update_last_seen updates the last_seen timestamp."""
info = PeerInfo(peer_id=peer("test"))
original_time = info.last_seen

# Brief delay
time.sleep(0.01)

info.update_last_seen()
assert info.last_seen > original_time
Loading