Added memory protocol

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)}")

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"),
 ]
 
 

newsfragments/92.feature.rst

@@ -0,0 +1 @@
+Integrated the memory protocol, in reference with go-libp2p

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"

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}")