test(networking): add codec test vectors for varint, gossip topics, and message IDs

packages/testing/src/consensus_testing/__init__.py

@@ -7,6 +7,7 @@
 from .test_fixtures import (
     BaseConsensusFixture,
     ForkChoiceTest,
+    NetworkingCodecTest,
     SSZTest,
     StateTransitionTest,
     VerifySignaturesTest,
@@ -32,6 +33,7 @@
 ForkChoiceTestFiller = Type[ForkChoiceTest]
 VerifySignaturesTestFiller = Type[VerifySignaturesTest]
 SSZTestFiller = Type[SSZTest]
+NetworkingCodecTestFiller = Type[NetworkingCodecTest]
 
 __all__ = [
     # Public API
@@ -48,6 +50,7 @@
     "ForkChoiceTest",
     "VerifySignaturesTest",
     "SSZTest",
+    "NetworkingCodecTest",
     # Test types
     "BaseForkChoiceStep",
     "TickStep",
@@ -64,4 +67,5 @@
     "ForkChoiceTestFiller",
     "VerifySignaturesTestFiller",
     "SSZTestFiller",
+    "NetworkingCodecTestFiller",
 ]

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

@@ -2,6 +2,7 @@
 
 from .base import BaseConsensusFixture
 from .fork_choice import ForkChoiceTest
+from .networking_codec import NetworkingCodecTest
 from .ssz import SSZTest
 from .state_transition import StateTransitionTest
 from .verify_signatures import VerifySignaturesTest
@@ -12,4 +13,5 @@
     "ForkChoiceTest",
     "VerifySignaturesTest",
     "SSZTest",
+    "NetworkingCodecTest",
 ]

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

@@ -0,0 +1,100 @@
+"""Networking codec test fixture for wire-format conformance testing."""
+
+from typing import Any, ClassVar
+
+from lean_spec.subspecs.containers.validator import SubnetId
+from lean_spec.subspecs.networking.gossipsub.message import GossipsubMessage
+from lean_spec.subspecs.networking.gossipsub.topic import GossipTopic, TopicKind
+from lean_spec.subspecs.networking.varint import decode_varint, encode_varint
+
+from .base import BaseConsensusFixture
+
+
+def _to_hex(data: bytes) -> str:
+    """Format raw bytes as a 0x-prefixed hex string."""
+    return "0x" + data.hex()
+
+
+def _from_hex(hex_str: str) -> bytes:
+    """Parse a 0x-prefixed hex string into raw bytes."""
+    return bytes.fromhex(hex_str.removeprefix("0x"))
+
+
+class NetworkingCodecTest(BaseConsensusFixture):
+    """Fixture for networking wire-format conformance.
+
+    Verifies encode/decode roundtrips for networking codecs.
+
+    JSON output: codecName, input, output.
+    """
+
+    format_name: ClassVar[str] = "networking_codec"
+    description: ClassVar[str] = "Tests networking codec encode/decode roundtrip"
+
+    codec_name: str
+    """Codec under test: varint, gossip_topic, or gossip_message_id."""
+
+    input: dict[str, Any]
+    """Codec-specific input parameters."""
+
+    output: dict[str, Any] = {}
+    """Computed output. Filled by make_fixture."""
+
+    def make_fixture(self) -> "NetworkingCodecTest":
+        """Dispatch to the codec handler and produce computed output.
+
+        Returns:
+            A copy of this fixture with output populated.
+
+        Raises:
+            ValueError: If codec_name is unknown.
+        """
+        match self.codec_name:
+            case "varint":
+                output = self._make_varint()
+            case "gossip_topic":
+                output = self._make_gossip_topic()
+            case "gossip_message_id":
+                output = self._make_gossip_message_id()
+            case _:
+                raise ValueError(f"Unknown codec: {self.codec_name}")
+        return self.model_copy(update={"output": output})
+
+    def _make_varint(self) -> dict[str, Any]:
+        """Encode a value as LEB128, decode it back, assert roundtrip."""
+        value = self.input["value"]
+        encoded = encode_varint(value)
+
+        # Decode must recover the original value and consume all bytes.
+        decoded, byte_length = decode_varint(encoded)
+        assert decoded == value, f"Varint roundtrip: {value} -> {encoded.hex()} -> {decoded}"
+        assert byte_length == len(encoded), f"Length: {byte_length} != {len(encoded)}"
+
+        return {"encoded": _to_hex(encoded), "byteLength": byte_length}
+
+    def _make_gossip_topic(self) -> dict[str, Any]:
+        """Build a topic string from components, parse it back, assert roundtrip."""
+        kind = TopicKind(self.input["kind"])
+        fork_digest = self.input["forkDigest"]
+        raw_subnet = self.input.get("subnetId")
+        subnet_id = SubnetId(raw_subnet) if raw_subnet is not None else None
+
+        topic = GossipTopic(kind=kind, fork_digest=fork_digest, subnet_id=subnet_id)
+        topic_string = topic.to_topic_id()
+
+        # Parse the string back to verify it reconstructs the same topic.
+        parsed = GossipTopic.from_string(topic_string)
+        assert parsed == topic, f"Topic roundtrip: {topic} -> {topic_string!r} -> {parsed}"
+
+        return {"topicString": topic_string}
+
+    def _make_gossip_message_id(self) -> dict[str, Any]:
+        """Compute a 20-byte gossipsub message ID from topic, data, and domain."""
+        topic = _from_hex(self.input["topic"])
+        data = _from_hex(self.input["data"])
+        domain = _from_hex(self.input["domain"])
+
+        # SHA256(domain + uint64_le(len(topic)) + topic + data)[:20]
+        message_id = GossipsubMessage.compute_id(topic, data, domain=domain)
+
+        return {"messageId": _to_hex(message_id)}

tests/consensus/devnet/networking/test_gossip_message_id.py

@@ -0,0 +1,123 @@
+"""Test vectors for gossipsub message ID computation."""
+
+import pytest
+from consensus_testing import NetworkingCodecTestFiller
+
+pytestmark = pytest.mark.valid_until("Devnet")
+
+VALID_SNAPPY = "0x01000000"
+"""Domain prefix for messages where snappy decompression succeeded."""
+
+INVALID_SNAPPY = "0x00000000"
+"""Domain prefix for messages where snappy decompression failed."""
+
+BLOCK_TOPIC = "0x" + b"/leanconsensus/0x12345678/block/ssz_snappy".hex()
+"""Hex-encoded topic bytes for a typical block topic string."""
+
+
+# --- Valid snappy domain ---
+
+
+def test_message_id_valid_snappy(networking_codec: NetworkingCodecTestFiller) -> None:
+    """Message ID with valid-snappy domain and typical block data."""
+    networking_codec(
+        codec_name="gossip_message_id",
+        input={
+            "topic": BLOCK_TOPIC,
+            "data": "0xdeadbeef",
+            "domain": VALID_SNAPPY,
+        },
+    )
+
+
+def test_message_id_valid_snappy_large_payload(networking_codec: NetworkingCodecTestFiller) -> None:
+    """Message ID with valid-snappy domain and a larger payload."""
+    networking_codec(
+        codec_name="gossip_message_id",
+        input={
+            "topic": BLOCK_TOPIC,
+            "data": "0x" + "ab" * 256,
+            "domain": VALID_SNAPPY,
+        },
+    )
+
+
+# --- Invalid snappy domain ---
+
+
+def test_message_id_invalid_snappy(networking_codec: NetworkingCodecTestFiller) -> None:
+    """Message ID with invalid-snappy domain and same data as the valid test."""
+    networking_codec(
+        codec_name="gossip_message_id",
+        input={
+            "topic": BLOCK_TOPIC,
+            "data": "0xdeadbeef",
+            "domain": INVALID_SNAPPY,
+        },
+    )
+
+
+# --- Empty inputs ---
+
+
+def test_message_id_empty_data(networking_codec: NetworkingCodecTestFiller) -> None:
+    """Message ID with empty payload."""
+    networking_codec(
+        codec_name="gossip_message_id",
+        input={
+            "topic": BLOCK_TOPIC,
+            "data": "0x",
+            "domain": VALID_SNAPPY,
+        },
+    )
+
+
+def test_message_id_empty_topic(networking_codec: NetworkingCodecTestFiller) -> None:
+    """Message ID with empty topic string."""
+    networking_codec(
+        codec_name="gossip_message_id",
+        input={
+            "topic": "0x",
+            "data": "0xdeadbeef",
+            "domain": VALID_SNAPPY,
+        },
+    )
+
+
+def test_message_id_both_empty(networking_codec: NetworkingCodecTestFiller) -> None:
+    """Message ID with both topic and data empty."""
+    networking_codec(
+        codec_name="gossip_message_id",
+        input={
+            "topic": "0x",
+            "data": "0x",
+            "domain": VALID_SNAPPY,
+        },
+    )
+
+
+# --- Domain differentiation ---
+
+
+def test_message_id_domain_changes_id(networking_codec: NetworkingCodecTestFiller) -> None:
+    """Same topic and data with invalid-snappy domain produces a different ID."""
+    networking_codec(
+        codec_name="gossip_message_id",
+        input={
+            "topic": "0x" + b"test-topic".hex(),
+            "data": "0x" + b"hello world".hex(),
+            "domain": INVALID_SNAPPY,
+        },
+    )
+
+
+def test_message_id_domain_valid_same_data(networking_codec: NetworkingCodecTestFiller) -> None:
+    """Same topic and data with valid-snappy domain for cross-reference."""
+    networking_codec(
+        codec_name="gossip_message_id",
+        input={
+            "topic": "0x" + b"test-topic".hex(),
+            "data": "0x" + b"hello world".hex(),
+            "domain": VALID_SNAPPY,
+        },
+    )

tests/consensus/devnet/networking/test_gossip_topic.py

@@ -0,0 +1,85 @@
+"""Test vectors for gossipsub topic string encoding and parsing."""
+
+import pytest
+from consensus_testing import NetworkingCodecTestFiller
+
+pytestmark = pytest.mark.valid_until("Devnet")
+
+FORK_DIGEST = "0x12345678"
+"""Arbitrary fork digest used across topic tests."""
+
+
+# --- Block topics ---
+
+
+def test_block_topic(networking_codec: NetworkingCodecTestFiller) -> None:
+    """Block topic with typical fork digest."""
+    networking_codec(
+        codec_name="gossip_topic",
+        input={"kind": "block", "forkDigest": FORK_DIGEST},
+    )
+
+
+def test_block_topic_different_digest(networking_codec: NetworkingCodecTestFiller) -> None:
+    """Block topic with a different fork digest to verify digest embedding."""
+    networking_codec(
+        codec_name="gossip_topic",
+        input={"kind": "block", "forkDigest": "0xaabbccdd"},
+    )
+
+
+# --- Aggregation topics ---
+
+
+def test_aggregation_topic(networking_codec: NetworkingCodecTestFiller) -> None:
+    """Committee aggregation topic."""
+    networking_codec(
+        codec_name="gossip_topic",
+        input={"kind": "aggregation", "forkDigest": FORK_DIGEST},
+    )
+
+
+# --- Attestation subnet topics ---
+
+
+def test_attestation_subnet_zero(networking_codec: NetworkingCodecTestFiller) -> None:
+    """Attestation subnet 0. First subnet ID."""
+    networking_codec(
+        codec_name="gossip_topic",
+        input={"kind": "attestation", "forkDigest": FORK_DIGEST, "subnetId": 0},
+    )
+
+
+def test_attestation_subnet_seven(networking_codec: NetworkingCodecTestFiller) -> None:
+    """Attestation subnet 7. Mid-range subnet ID."""
+    networking_codec(
+        codec_name="gossip_topic",
+        input={"kind": "attestation", "forkDigest": FORK_DIGEST, "subnetId": 7},
+    )
+
+
+def test_attestation_subnet_63(networking_codec: NetworkingCodecTestFiller) -> None:
+    """Attestation subnet 63. Last subnet in a 64-subnet network."""
+    networking_codec(
+        codec_name="gossip_topic",
+        input={"kind": "attestation", "forkDigest": FORK_DIGEST, "subnetId": 63},
+    )
+
+
+# --- Edge cases ---
+
+
+def test_block_topic_zero_digest(networking_codec: NetworkingCodecTestFiller) -> None:
+    """Block topic with all-zero fork digest."""
+    networking_codec(
+        codec_name="gossip_topic",
+        input={"kind": "block", "forkDigest": "0x00000000"},
+    )
+
+
+def test_block_topic_max_digest(networking_codec: NetworkingCodecTestFiller) -> None:
+    """Block topic with all-0xff fork digest."""
+    networking_codec(
+        codec_name="gossip_topic",
+        input={"kind": "block", "forkDigest": "0xffffffff"},
+    )