Merge pull request #42 from alexander255/master

Miscellaneous improvements

Author: mhchia

multiaddr/codecs/_util.py

@@ -2,7 +2,7 @@
     def packed_net_bytes_to_int(b):
         """Convert the given big-endian byte-string to an int."""
         return int.from_bytes(b, byteorder='big')
-else:  # PY2
+else:  # pragma: no cover (PY2)
     def packed_net_bytes_to_int(b):
         """Convert the given big-endian byte-string to an int."""
         return int(b.encode('hex'), 16)

multiaddr/codecs/fspath.py

@@ -13,16 +13,16 @@
 if hasattr(os, "fsencode") and hasattr(os, "fsdecode"):
     fsencode = os.fsencode
     fsdecode = os.fsdecode
-else:  # PY2
+else:  # pragma: no cover (PY2)
     import sys
 
     def fsencode(path):
-        if not isinstance(path, six.binary_type):  # pragma: no cover
+        if not isinstance(path, six.binary_type):
             path = path.encode(sys.getfilesystemencoding())
         return path
 
     def fsdecode(path):
-        if not isinstance(path, six.text_type):  # pragma: no cover
+        if not isinstance(path, six.text_type):
             path = path.decode(sys.getfilesystemencoding())
         return path
 

multiaddr/codecs/p2p.py

@@ -12,7 +12,7 @@
 
 def to_bytes(proto, string):
     # the address is a base58-encoded string
-    if six.PY2 and isinstance(string, unicode):
+    if six.PY2 and isinstance(string, unicode):  # pragma: no cover (PY2)
         string = string.encode("ascii")
     mm = base58.b58decode(string)
     if len(mm) < 5:

multiaddr/multiaddr.py

@@ -1,5 +1,9 @@
 # -*- coding: utf-8 -*-
-from copy import copy
+try:
+	import collections.abc
+except ImportError:  # pragma: no cover (PY2)
+	import collections
+	collections.abc = collections
 
 import six
 
@@ -10,7 +14,76 @@
 from .transforms import bytes_to_string
 
 
-class Multiaddr(object):
+__all__ = ("Multiaddr",)
+
+
+
+class MultiAddrKeys(collections.abc.KeysView, collections.abc.Sequence):
+    def __contains__(self, proto):
+        proto = protocols.protocol_with_any(proto)
+        return collections.abc.Sequence.__contains__(self, proto)
+
+    def __getitem__(self, idx):
+        if idx < 0:
+            idx = len(self)+idx
+        for idx2, proto in enumerate(self):
+            if idx2 == idx:
+                return proto
+        raise IndexError("Protocol list index out of range")
+
+    __hash__ = collections.abc.KeysView._hash
+
+    def __iter__(self):
+        for proto, _, _ in bytes_iter(self._mapping.to_bytes()):
+            yield proto
+
+
+class MultiAddrItems(collections.abc.ItemsView, collections.abc.Sequence):
+    def __contains__(self, item):
+        proto, value = item
+        proto = protocols.protocol_with_any(proto)
+        return collections.abc.Sequence.__contains__(self, (proto, value))
+
+    def __getitem__(self, idx):
+        if idx < 0:
+            idx = len(self)+idx
+        for idx2, item in enumerate(self):
+            if idx2 == idx:
+                return item
+        raise IndexError("Protocol item list index out of range")
+
+    def __iter__(self):
+        for proto, codec, part in bytes_iter(self._mapping.to_bytes()):
+            if codec.SIZE != 0:
+                try:
+                    # If we have an address, return it
+                    yield proto, codec.to_string(proto, part)
+                except Exception as exc:
+                    six.raise_from(exceptions.BinaryParseError(str(exc), self._mapping.to_bytes(), proto.name, exc), exc)
+            else:
+                # We were given something like '/utp', which doesn't have
+                # an address, so return None
+                yield proto, None
+
+
+class MultiAddrValues(collections.abc.ValuesView, collections.abc.Sequence):
+    __contains__ = collections.abc.Sequence.__contains__
+
+    def __getitem__(self, idx):
+        if idx < 0:
+            idx = len(self)+idx
+        for idx2, proto in enumerate(self):
+            if idx2 == idx:
+                return proto
+        raise IndexError("Protocol value list index out of range")
+
+    def __iter__(self):
+        for _, value in MultiAddrItems(self._mapping):
+            yield value
+
+
+
+class Multiaddr(collections.abc.Mapping):
     """Multiaddr is a representation of multiple nested internet addresses.
 
     Multiaddr is a cross-protocol, cross-platform format for representing
@@ -39,34 +112,42 @@ def __init__(self, addr):
         # On Python 2 text string will often be binary anyways so detect the
         # obvious case of a “binary-encoded” multiaddr starting with a slash
         # and decode it into text
-        if six.PY2 and isinstance(addr, str) and addr.startswith("/"):
+        if six.PY2 and isinstance(addr, str) and addr.startswith("/"):  # pragma: no cover (PY2)
             addr = addr.decode("utf-8")
 
         if isinstance(addr, six.text_type):
             self._bytes = string_to_bytes(addr)
         elif isinstance(addr, six.binary_type):
             self._bytes = addr
+        elif isinstance(addr, Multiaddr):
+            self._bytes = addr.to_bytes()
         else:
-            raise TypeError("MultiAddr must be bytes or str")
+            raise TypeError("MultiAddr must be bytes, str or another MultiAddr instance")
 
     def __eq__(self, other):
         """Checks if two Multiaddr objects are exactly equal."""
         return self._bytes == other._bytes
 
-    def __ne__(self, other):
-        return not (self == other)
-
     def __str__(self):
         """Return the string representation of this Multiaddr.
 
         May raise a :class:`~multiaddr.exceptions.BinaryParseError` if the
         stored MultiAddr binary representation is invalid."""
         return bytes_to_string(self._bytes)
 
+    def __contains__(self, proto):
+        return proto in MultiAddrKeys(self)
+
+    def __iter__(self):
+        return iter(MultiAddrKeys(self))
+
+    def __len__(self):
+        return sum((1 for _ in bytes_iter(self.to_bytes())))
+
     # On Python 2 __str__ needs to return binary text, so expose the original
     # function as __unicode__ and transparently encode its returned text based
     # on the current locale
-    if six.PY2:
+    if six.PY2:  # pragma: no cover (PY2)
         __unicode__ = __str__
 
         def __str__(self):
@@ -82,7 +163,15 @@ def to_bytes(self):
 
     def protocols(self):
         """Returns a list of Protocols this Multiaddr includes."""
-        return list(proto for proto, _, _ in bytes_iter(self.to_bytes()))
+        return MultiAddrKeys(self)
+
+    keys = protocols
+
+    def items(self):
+        return MultiAddrItems(self)
+
+    def values(self):
+        return MultiAddrValues(self)
 
     def encapsulate(self, other):
         """Wrap this Multiaddr around another.
@@ -91,7 +180,7 @@ def encapsulate(self, other):
             /ip4/1.2.3.4 encapsulate /tcp/80 = /ip4/1.2.3.4/tcp/80
         """
         mb = self.to_bytes()
-        ob = other.to_bytes()
+        ob = Multiaddr(other).to_bytes()
         return Multiaddr(b''.join([mb, ob]))
 
     def decapsulate(self, other):
@@ -100,35 +189,35 @@ def decapsulate(self, other):
         For example:
             /ip4/1.2.3.4/tcp/80 decapsulate /ip4/1.2.3.4 = /tcp/80
         """
-        s1 = str(self)
-        s2 = str(other)
+        s1 = self.to_bytes()
+        s2 = Multiaddr(other).to_bytes()
         try:
             idx = s1.rindex(s2)
         except ValueError:
             # if multiaddr not contained, returns a copy
-            return copy(self)
+            return Multiaddr(self)
         return Multiaddr(s1[:idx])
 
     def value_for_protocol(self, proto):
-        """Return the value (if any) following the specified protocol."""
-        if not isinstance(proto, protocols.Protocol):
-            if isinstance(proto, int):
-                proto = protocols.protocol_with_code(proto)
-            elif isinstance(proto, six.string_types):
-                proto = protocols.protocol_with_name(proto)
-            else:
-                raise TypeError("Protocol object, name or code expected, got {0!r}".format(proto))
-
-        for proto2, codec, part in bytes_iter(self.to_bytes()):
-            if proto2 == proto:
-                if codec.SIZE != 0:
-                    try:
-                        # If we have an address, return it
-                        return codec.to_string(proto2, part)
-                    except Exception as exc:
-                        six.raise_from(exceptions.BinaryParseError(str(exc), self.to_bytes(), proto2.name, exc), exc)
-                else:
-                    # We were given something like '/utp', which doesn't have
-                    # an address, so return ''
-                    return ''
+        """Return the value (if any) following the specified protocol
+        
+        Returns
+        -------
+        union[object, NoneType]
+            The parsed protocol value for the given protocol code or ``None``
+            if the given protocol does not require any value
+        
+        Raises
+        ------
+        ~multiaddr.exceptions.BinaryParseError
+            The stored MultiAddr binary representation is invalid
+        ~multiaddr.exceptions.ProtocolLookupError
+            MultiAddr does not contain any instance of this protocol
+        """
+        proto = protocols.protocol_with_any(proto)
+        for proto2, value in self.items():
+            if proto2 is proto or proto2 == proto:
+                return value
         raise exceptions.ProtocolLookupError(proto, str(self))
+
+    __getitem__ = value_for_protocol

multiaddr/protocols.py

@@ -100,13 +100,16 @@ def vcode(self):
         return varint.encode(self.code)
 
     def __eq__(self, other):
+        if not isinstance(other, Protocol):
+            return NotImplemented
+
         return all((self.code == other.code,
                     self.name == other.name,
                     self.codec == other.codec,
                     self.path == other.path))
 
-    def __ne__(self, other):
-        return not self == other
+    def __hash__(self):
+        return self.code
 
     def __repr__(self):
         return "Protocol(code={code!r}, name={name!r}, codec={codec!r})".format(
@@ -123,7 +126,7 @@ def _uvarint(buf):
     for i, b_str in enumerate(buf):
         if six.PY3:
             b = b_str
-        else:
+        else:  # pragma: no cover (PY2)
             b = int(binascii.b2a_hex(b_str), 16)
         if b < 0x80:
             if i > 9 or (i == 9 and b > 1):
@@ -200,6 +203,17 @@ def protocol_with_code(code):
     return _codes_to_protocols[code]
 
 
+def protocol_with_any(proto):
+    if isinstance(proto, Protocol):
+        return proto
+    elif isinstance(proto, int):
+        return protocol_with_code(proto)
+    elif isinstance(proto, six.string_types):
+        return protocol_with_name(proto)
+    else:
+        raise TypeError("Protocol object, name or code expected, got {0!r}".format(proto))
+
+
 def protocols_with_string(string):
     """Return a list of protocols matching given string."""
     # Normalize string

multiaddr/transforms.py

@@ -14,9 +14,6 @@
 
 
 def string_to_bytes(string):
-    if not string:
-        return b''
-
     bs = []
     for proto, codec, value in string_iter(string):
         bs.append(varint.encode(proto.code))
@@ -78,7 +75,7 @@ def string_iter(string):
                 value = "/" + "/".join(sp)
                 if not six.PY2:
                     sp.clear()
-                else:
+                else:  # pragma: no cover (PY2)
                     sp = []
             else:
                 value = sp.pop(0)

tests/test_multiaddr.py

@@ -190,17 +190,17 @@ def test_get_value():
         "p2p/QmbHVEEepCi7rn7VL7Exxpd2Ci9NNB6ifvqwhsrbRMgQFP")
 
     assert_value_for_proto(ma, P_IP4, "127.0.0.1")
-    assert_value_for_proto(ma, P_UTP, "")
+    assert_value_for_proto(ma, P_UTP, None)
     assert_value_for_proto(ma, P_TCP, "5555")
     assert_value_for_proto(ma, P_UDP, "1234")
     assert_value_for_proto(
         ma, P_P2P, "QmbHVEEepCi7rn7VL7Exxpd2Ci9NNB6ifvqwhsrbRMgQFP")
     assert_value_for_proto(ma, "ip4", "127.0.0.1")
-    assert_value_for_proto(ma, "utp", "")
+    assert_value_for_proto(ma, "utp", None)
     assert_value_for_proto(ma, "tcp", "5555")
     assert_value_for_proto(ma, "udp", "1234")
     assert_value_for_proto(ma, protocol_with_name("ip4"), "127.0.0.1")
-    assert_value_for_proto(ma, protocol_with_name("utp"), "")
+    assert_value_for_proto(ma, protocol_with_name("utp"), None)
     assert_value_for_proto(ma, protocol_with_name("tcp"), "5555")
     assert_value_for_proto(ma, protocol_with_name("udp"), "1234")
 
@@ -224,7 +224,7 @@ def test_get_value():
     a = Multiaddr("/ip4/0.0.0.0/udp/12345/utp")  # ending in a no-value one.
     assert_value_for_proto(a, P_IP4, "0.0.0.0")
     assert_value_for_proto(a, P_UDP, "12345")
-    assert_value_for_proto(a, P_UTP, "")
+    assert_value_for_proto(a, P_UTP, None)
 
     a = Multiaddr("/ip4/0.0.0.0/unix/a/b/c/d")  # ending in a path one.
     assert_value_for_proto(a, P_IP4, "0.0.0.0")
@@ -234,6 +234,34 @@ def test_get_value():
     assert_value_for_proto(a, P_UNIX, "/studio")  # only a path.
 
 
+def test_views():
+    ma = Multiaddr(
+        "/ip4/127.0.0.1/utp/tcp/5555/udp/1234/utp/"
+        "p2p/QmbHVEEepCi7rn7VL7Exxpd2Ci9NNB6ifvqwhsrbRMgQFP")
+
+    for idx, (proto1, proto2, item, value) in enumerate(zip(ma, ma.keys(), ma.items(), ma.values())):
+        assert (proto1, value) == (proto2, value) == item
+        assert proto1 in ma
+        assert proto2 in ma.keys()
+        assert item in ma.items()
+        assert value in ma.values()
+        assert ma.keys()[idx] == ma.keys()[idx-len(ma)] == proto1 == proto2
+        assert ma.items()[idx] == ma.items()[idx-len(ma)] == item
+        assert ma.values()[idx] == ma.values()[idx-len(ma)] == ma[proto1] == value
+
+    assert len(ma.keys()) == len(ma.items()) == len(ma.values()) == len(ma)
+    assert len(list(ma.keys())) == len(ma.keys())
+    assert len(list(ma.items())) == len(ma.items())
+    assert len(list(ma.values())) == len(ma.values())
+    
+    with pytest.raises(IndexError):
+        ma.keys()[len(ma)]
+    with pytest.raises(IndexError):
+        ma.items()[len(ma)]
+    with pytest.raises(IndexError):
+        ma.values()[len(ma)]
+
+
 def test_bad_initialization_no_params():
     with pytest.raises(TypeError):
         Multiaddr()