diff --git a/multiaddr/util.py b/multiaddr/util.py index 7d9de93..b049655 100644 --- a/multiaddr/util.py +++ b/multiaddr/util.py @@ -1,4 +1,5 @@ import binascii +import six import struct from .codec import bytes_split @@ -27,7 +28,11 @@ def int_to_hex(i, size): Note that size is the size of the resulting hex string. So, for a 32Byte int, size should be 64 (two hex characters per byte".""" f_str = "{0:0%sx}" % size - return f_str.format(i).lower() + buf = f_str.format(i).lower() + if six.PY3: + buf = bytes(buf, 'utf-8') + + return buf def encode_big_endian_32(i): diff --git a/tests/test_codec.py b/tests/test_codec.py new file mode 100644 index 0000000..bd3c334 --- /dev/null +++ b/tests/test_codec.py @@ -0,0 +1,119 @@ +import pytest + +from multiaddr.codec import address_bytes_to_string +from multiaddr.codec import address_string_to_bytes +from multiaddr.codec import bytes_split +from multiaddr.codec import bytes_to_string +from multiaddr.codec import size_for_addr +from multiaddr.codec import string_to_bytes + +from multiaddr.protocols import _names_to_protocols +from multiaddr.protocols import Protocol + +# These test values were generated by running them +# through the go implementation of multiaddr. +# https://github.com/jbenet/multiaddr +ADDR_BYTES_MAP_STR_TEST_DATA = [ + (_names_to_protocols['ip4'], b'0a0b0c0d', '10.11.12.13'), + (_names_to_protocols['ip6'], b'1aa12bb23cc34dd45ee56ff67ab78ac8', + '1aa1:2bb2:3cc3:4dd4:5ee5:6ff6:7ab7:8ac8'), + (_names_to_protocols['tcp'], b'abcd', '43981'), + (_names_to_protocols['onion'], b'9a18087306369043091f04d2', + 'timaq4ygg2iegci7:1234'), + (_names_to_protocols['ipfs'], + b'221220d52ebb89d85b02a284948203a62ff28389c57c9f42beec4ec20db76a68911c0b', + 'QmcgpsyWgH8Y8ajJz1Cu72KnS5uo2Aa2LpzU7kinSupNKC'), +] + +BYTES_MAP_STR_TEST_DATA = [ + ("/ip4/127.0.0.1/udp/1234", b'047f0000011104d2'), + ("/ip4/127.0.0.1/tcp/4321", b'047f0000010610e1'), + ("/ip4/127.0.0.1/udp/1234/ip4/127.0.0.1/tcp/4321", + b'047f0000011104d2047f0000010610e1') +] + + +@pytest.mark.parametrize("proto, buf, expected", [ + (_names_to_protocols['https'], b'\x01\x02\x03', 0), + (_names_to_protocols['ip4'], b'\x01\x02\x03', 4), + (_names_to_protocols['ipfs'], b'\x40\x50\x60\x51', 65), +]) +def test_size_for_addr(proto, buf, expected): + assert size_for_addr(proto, buf) == expected + + +@pytest.mark.parametrize("buf, expected", [ + # "/ip4/127.0.0.1/udp/1234/ip4/127.0.0.1/tcp/4321" + (b'047f0000011104d2047f0000010610e1', + [b'\x04\x7f\x00\x00\x01', + b'\x11\x04\xd2', + b'\x04\x7f\x00\x00\x01', + b'\x06\x10\xe1']), +]) +def test_bytes_split(buf, expected): + assert bytes_split(buf) == expected + + +@pytest.mark.parametrize("proto, buf, expected", ADDR_BYTES_MAP_STR_TEST_DATA) +def test_address_bytes_to_string(proto, buf, expected): + assert address_bytes_to_string(proto, buf) == expected + + +@pytest.mark.parametrize("proto, expected, string", + ADDR_BYTES_MAP_STR_TEST_DATA) +def test_address_string_to_bytes(proto, string, expected): + assert address_string_to_bytes(proto, string) == expected + + +@pytest.mark.parametrize("string, buf", BYTES_MAP_STR_TEST_DATA) +def test_string_to_bytes(string, buf): + assert string_to_bytes(string) == buf + + +@pytest.mark.parametrize("string, buf", BYTES_MAP_STR_TEST_DATA) +def test_bytes_to_string(string, buf): + assert bytes_to_string(buf) == string + + +@pytest.mark.parametrize("string", [ + 'test', + '/ip4/' +]) +def test_string_to_bytes_value_error(string): + with pytest.raises(ValueError): + string_to_bytes(string) + + +class DummyProtocol(Protocol): + def __init__(self, code, size, name, vcode): + self.code = code + self.size = size + self.name = name + self.vcode = vcode + + +@pytest.mark.parametrize("proto, address", [ + (DummyProtocol(234, 32, 'test', b'123'), '1.2.3.4'), + (_names_to_protocols['ip4'], '1124.2.3'), + (_names_to_protocols['ip6'], '123:31224444'), + (_names_to_protocols['tcp'], 'a'), + (_names_to_protocols['tcp'], '100000'), + (_names_to_protocols['onion'], '100000'), + (_names_to_protocols['onion'], '1234567890123456:0'), + (_names_to_protocols['onion'], 'timaq4ygg2iegci7:a'), + (_names_to_protocols['onion'], 'timaq4ygg2iegci7:0'), + (_names_to_protocols['onion'], 'timaq4ygg2iegci7:71234'), + (_names_to_protocols['ipfs'], '15230d52ebb89d85b02a284948203a'), +]) +def test_address_string_to_bytes_value_error(proto, address): + with pytest.raises(ValueError): + address_string_to_bytes(proto, address) + + +@pytest.mark.parametrize("proto, buf", [ + (DummyProtocol(234, 32, 'test', b'123'), b'0a0b0c0d'), + (_names_to_protocols['ipfs'], b'15230d52ebb89d85b02a284948203a') +]) +def test_address_bytes_to_string_value_error(proto, buf): + with pytest.raises(ValueError): + address_bytes_to_string(proto, buf) diff --git a/tests/test_multiaddr.py b/tests/test_multiaddr.py index 9abf4f7..dc3e439 100755 --- a/tests/test_multiaddr.py +++ b/tests/test_multiaddr.py @@ -2,8 +2,6 @@ # -*- coding: utf-8 -*- import pytest -from multiaddr.codec import string_to_bytes -from multiaddr.codec import bytes_to_string from multiaddr.multiaddr import Multiaddr from multiaddr.multiaddr import ProtocolNotFoundException from multiaddr.protocols import protocol_with_name @@ -103,21 +101,6 @@ def test_eq(): assert m4 == m3 -def test_string_to_bytes(): - assert string_to_bytes("/ip4/127.0.0.1/udp/1234") == b'047f0000011104d2' - assert string_to_bytes("/ip4/127.0.0.1/tcp/4321") == b'047f0000010610e1' - assert ( - string_to_bytes("/ip4/127.0.0.1/udp/1234/ip4/127.0.0.1/tcp/4321") == - b'047f0000011104d2047f0000010610e1') - - -def test_bytes_to_string(): - assert bytes_to_string(b"047f0000011104d2") == "/ip4/127.0.0.1/udp/1234" - assert bytes_to_string(b"047f0000010610e1") == "/ip4/127.0.0.1/tcp/4321" - assert (bytes_to_string(b"047f0000011104d2047f0000010610e1") == - "/ip4/127.0.0.1/udp/1234/ip4/127.0.0.1/tcp/4321") - - @pytest.mark.parametrize( 'test_vals', [("/ip4/1.2.3.4/udp/1234", ["/ip4/1.2.3.4", "/udp/1234"]), @@ -232,7 +215,7 @@ def test_bad_initialization_no_params(): def test_bad_initialization_too_many_params(): with pytest.raises(ValueError): - Multiaddr("/ip4/0.0.0.0", string_to_bytes("/ip4/0.0.0.0")) + Multiaddr("/ip4/0.0.0.0", "") def test_get_value_too_many_fields_protocol(monkeypatch):