diff --git a/multiaddr/codecs/memory.py b/multiaddr/codecs/memory.py new file mode 100644 index 0000000..844ad43 --- /dev/null +++ b/multiaddr/codecs/memory.py @@ -0,0 +1,30 @@ +import struct +from typing import Any + +from ..codecs import CodecBase +from ..exceptions import BinaryParseError + +SIZE = 64 +IS_PATH = False + + +class Codec(CodecBase): + SIZE = SIZE + IS_PATH = IS_PATH + + def to_bytes(self, proto: Any, string: str) -> bytes: + # Parse as unsigned 64-bit int + value = int(string, 10) + if value < 0 or value > 0xFFFFFFFFFFFFFFFF: + raise ValueError("Value out of range for uint64") + return struct.pack(">Q", value) # big-endian uint64 + + def to_string(self, proto: Any, buf: bytes) -> str: + if len(buf) != 8: + raise BinaryParseError("Expected 8 bytes for uint64", buf, "memory") + value = struct.unpack(">Q", buf)[0] + return str(value) + + def memory_validate(self, b: bytes) -> None: + if len(b) != 8: + raise ValueError(f"Invalid length: must be exactly 8 bytes, got {len(b)}") diff --git a/multiaddr/protocols.py b/multiaddr/protocols.py index eb25f63..f32b74d 100644 --- a/multiaddr/protocols.py +++ b/multiaddr/protocols.py @@ -78,6 +78,7 @@ P_SNI = 0x01C1 P_NOISE = 0x01C6 P_WEBTRANSPORT = 0x01D1 +P_MEMORY = 0x309 class Protocol: @@ -166,6 +167,7 @@ def __repr__(self) -> str: Protocol(P_P2P_CIRCUIT, "p2p-circuit", None), Protocol(P_WEBTRANSPORT, "webtransport", None), Protocol(P_UNIX, "unix", "fspath"), + Protocol(P_MEMORY, "memory", "memory"), ] diff --git a/newsfragments/92.feature.rst b/newsfragments/92.feature.rst new file mode 100644 index 0000000..1d5e306 --- /dev/null +++ b/newsfragments/92.feature.rst @@ -0,0 +1 @@ +Integrated the memory protocol, in reference with go-libp2p diff --git a/tests/test_multiaddr.py b/tests/test_multiaddr.py index 2426ab5..369c08c 100644 --- a/tests/test_multiaddr.py +++ b/tests/test_multiaddr.py @@ -1,5 +1,6 @@ import pytest +from multiaddr import protocols from multiaddr.exceptions import ( BinaryParseError, ProtocolLookupError, @@ -802,3 +803,25 @@ def test_decapsulate_code(): # No-op on empty ma3 = Multiaddr("") assert str(ma3.decapsulate_code(P_TCP)) == "" + + +def test_memory_protocol_integration(): + ma = Multiaddr("/memory/12345") + assert str(ma) == "/memory/12345" + assert len(ma.protocols()) == 1 + assert ma.protocols()[0].name == "memory" # type: ignore + assert ma.value_for_protocol(777) == "12345" + + # Binary rountrip + binary = ma.to_bytes() + reconstructed = Multiaddr(binary) + assert str(reconstructed) == str(ma) + + +def test_memory_protocol_properties(): + proto = protocols.Protocol(protocols.P_MEMORY, "memory", "memory") + assert proto.size == 64 # 8 bytes/ 64 bits + assert not proto.path # Not a path protocol + assert proto.code == 777 + assert proto.name == "memory" + assert proto.codec == "memory" diff --git a/tests/test_protocols.py b/tests/test_protocols.py index fabfede..f7b8bed 100644 --- a/tests/test_protocols.py +++ b/tests/test_protocols.py @@ -1,7 +1,9 @@ import pytest import varint -from multiaddr import exceptions, protocols +from multiaddr import Multiaddr, exceptions, protocols +from multiaddr.codecs import memory +from multiaddr.exceptions import BinaryParseError def test_code_to_varint(): @@ -180,3 +182,90 @@ def test_add_protocol_lock(valid_params): def test_protocol_repr(): proto = protocols.protocol_with_name("ip4") assert "Protocol(code=4, name='ip4', codec='ip4')" == repr(proto) + + +def test_to_bytes_and_to_string_roundtrip(): + codec = memory.Codec() + + # some valid values + for val in [0, 1, 42, 2**32, 2**64 - 1]: + s = str(val) + b = codec.to_bytes(None, s) + # must be exactly 8 bytes + assert isinstance(b, bytes) + assert len(b) == 8 + # roundtrip back to string + out = codec.to_string(None, b) + assert out == s + + +def test_invalid_string_to_bytes(): + codec = memory.Codec() + + # not a number + with pytest.raises(ValueError): + codec.to_bytes(None, "abc") + + # negative number + with pytest.raises(ValueError): + codec.to_bytes(None, "-1") + + # too large + with pytest.raises(ValueError): + codec.to_bytes(None, str(2**64)) + + +def test_invalid_bytes_to_string(): + codec = memory.Codec() + + # too short + with pytest.raises(BinaryParseError): + codec.to_string(None, b"\x00\x01") + + # too long + with pytest.raises(BinaryParseError): + codec.to_string(None, b"\x00" * 9) + + +def test_specific_encoding(): + codec = memory.Codec() + + # 42 encoded in big-endian + expected_bytes = b"\x00\x00\x00\x00\x00\x00\x00*" + assert codec.to_bytes(None, "42") == expected_bytes + assert codec.to_string(None, expected_bytes) == "42" + + +def test_memory_validate_function(): + # Directly test the helper + codec = memory.Codec() + + # Valid case + codec.memory_validate(b"\x00" * 8) # should not raise + + # Invalid length + with pytest.raises(ValueError): + codec.memory_validate(b"\x00" * 7) + + +def test_memory_integration_edge_values(): + # Minimum (0) + ma0 = Multiaddr("/memory/0") + assert str(ma0) == "/memory/0" + assert ma0.value_for_protocol(777) == "0" + + # Maximum (2**64 - 1) + max_val = str(2**64 - 1) + mamax = Multiaddr(f"/memory/{max_val}") + assert str(mamax) == f"/memory/{max_val}" + assert mamax.value_for_protocol(777) == max_val + + +def test_memory_integration_invalid_values(): + # Negative number + with pytest.raises(ValueError): + Multiaddr("/memory/-1") + + # Too large (overflow > uint64) + with pytest.raises(ValueError): + Multiaddr(f"/memory/{2**64}")