mirror of
https://github.com/espressif/openthread.git
synced 2026-07-29 23:27:46 +00:00
Add Python3 support to thread-cert. (#985)
This commit is contained in:
PFierek
@@ -61,12 +61,13 @@ class Cert_5_1_01_RouterAttach(unittest.TestCase):
|
||||
self.sniffer.start()
|
||||
|
||||
def tearDown(self):
|
||||
self.sniffer.stop()
|
||||
del self.sniffer
|
||||
|
||||
for node in list(self.nodes.values()):
|
||||
node.stop()
|
||||
del self.nodes
|
||||
|
||||
self.sniffer.stop()
|
||||
|
||||
def test(self):
|
||||
self.nodes[LEADER].start()
|
||||
self.nodes[LEADER].set_state('leader')
|
||||
@@ -170,6 +171,5 @@ class Cert_5_1_01_RouterAttach(unittest.TestCase):
|
||||
self.assertTrue(self.nodes[ROUTER].ping(addr))
|
||||
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
|
||||
@@ -120,7 +120,7 @@ class Cert_5_1_03_RouterAddressReallocation(unittest.TestCase):
|
||||
|
||||
# 5 - Router1
|
||||
# Router1 make two attempts to reconnect to its current Partition.
|
||||
for _ in xrange(2):
|
||||
for _ in range(2):
|
||||
msg = router1_messages.next_mle_message(mle.CommandType.PARENT_REQUEST)
|
||||
msg.assertSentWithHopLimit(255)
|
||||
msg.assertSentToDestinationAddress("ff02::2")
|
||||
|
||||
@@ -119,7 +119,7 @@ class Cert_5_1_04_RouterAddressReallocation(unittest.TestCase):
|
||||
|
||||
# 5 - Router1
|
||||
# Router1 make two attempts to reconnect to its current Partition.
|
||||
for _ in xrange(2):
|
||||
for _ in range(2):
|
||||
msg = router1_messages.next_mle_message(mle.CommandType.PARENT_REQUEST)
|
||||
msg.assertSentWithHopLimit(255)
|
||||
msg.assertSentToDestinationAddress("ff02::2")
|
||||
|
||||
@@ -27,16 +27,16 @@
|
||||
# POSSIBILITY OF SUCH DAMAGE.
|
||||
#
|
||||
|
||||
import struct
|
||||
from binascii import hexlify
|
||||
|
||||
import ipaddress
|
||||
|
||||
from binascii import hexlify
|
||||
import struct
|
||||
import sys
|
||||
|
||||
|
||||
def enum(*sequential, **named):
|
||||
enums = dict(zip(sequential, range(len(sequential))), **named)
|
||||
names = dict((value, key) for key, value in enums.iteritems())
|
||||
enums = dict(list(zip(sequential, list(range(len(sequential))))), **named)
|
||||
names = dict((value, key) for key, value in list(enums.items()))
|
||||
enums['name'] = names
|
||||
return type('Enum', (), enums)
|
||||
|
||||
@@ -60,12 +60,12 @@ class MessageInfo(object):
|
||||
self.payload_length = 0
|
||||
|
||||
def _convert_value_to_ip_address(self, value):
|
||||
if isinstance(value, str):
|
||||
value = unicode(value)
|
||||
|
||||
elif isinstance(value, bytearray):
|
||||
if isinstance(value, bytearray):
|
||||
value = bytes(value)
|
||||
|
||||
elif isinstance(value, str) and sys.version_info[0] == 2:
|
||||
value = value.decode("utf-8")
|
||||
|
||||
return ipaddress.ip_address(value)
|
||||
|
||||
@property
|
||||
@@ -135,7 +135,7 @@ class MacAddress(object):
|
||||
|
||||
@classmethod
|
||||
def from_rloc16(cls, rloc16, big_endian=True):
|
||||
if isinstance(rloc16, int) or isinstance(rloc16, long):
|
||||
if isinstance(rloc16, int):
|
||||
mac_address = struct.pack(">H", rloc16)
|
||||
elif isinstance(rloc16, bytearray):
|
||||
mac_address = rloc16[:2]
|
||||
|
||||
@@ -27,12 +27,18 @@
|
||||
# POSSIBILITY OF SUCH DAMAGE.
|
||||
#
|
||||
|
||||
try:
|
||||
from itertools import izip_longest as zip_longest
|
||||
except ImportError:
|
||||
from itertools import zip_longest
|
||||
|
||||
from ipaddress import ip_address
|
||||
|
||||
import abc
|
||||
import io
|
||||
import struct
|
||||
import sys
|
||||
|
||||
from itertools import izip_longest
|
||||
from ipaddress import ip_address
|
||||
|
||||
# Next headers for IPv6 protocols
|
||||
IPV6_NEXT_HEADER_HOP_BY_HOP = 0
|
||||
@@ -67,7 +73,7 @@ def calculate_checksum(data):
|
||||
int: calculated checksum
|
||||
"""
|
||||
# Create halfwords from data bytes. Example: data[0] = 0x01, data[1] = 0xb2 => 0x01b2
|
||||
halfwords = [((byte0 << 8) | byte1) for byte0, byte1 in izip_longest(data[::2], data[1::2], fillvalue=0x00)]
|
||||
halfwords = [((byte0 << 8) | byte1) for byte0, byte1 in zip_longest(data[::2], data[1::2], fillvalue=0x00)]
|
||||
|
||||
checksum = 0
|
||||
for halfword in halfwords:
|
||||
@@ -82,7 +88,7 @@ def calculate_checksum(data):
|
||||
return checksum
|
||||
|
||||
|
||||
class PacketFactory:
|
||||
class PacketFactory(object):
|
||||
|
||||
""" Interface for classes that produce objects from data. """
|
||||
|
||||
@@ -96,7 +102,7 @@ class PacketFactory:
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
class BuildableFromBytes:
|
||||
class BuildableFromBytes(object):
|
||||
|
||||
""" Interface for classes which can be built from bytes. """
|
||||
|
||||
@@ -111,7 +117,7 @@ class BuildableFromBytes:
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
class ConvertibleToBytes:
|
||||
class ConvertibleToBytes(object):
|
||||
|
||||
""" Interface for classes which can be converted to bytes. """
|
||||
|
||||
@@ -204,8 +210,8 @@ class IPv6PseudoHeader(ConvertibleToBytes):
|
||||
if isinstance(value, bytearray):
|
||||
value = bytes(value)
|
||||
|
||||
elif isinstance(value, unicode):
|
||||
value = str(value)
|
||||
elif isinstance(value, str) and sys.version_info[0] == 2:
|
||||
value = value.decode("utf-8")
|
||||
|
||||
return ip_address(value)
|
||||
|
||||
@@ -235,7 +241,7 @@ class IPv6PseudoHeader(ConvertibleToBytes):
|
||||
return data
|
||||
|
||||
|
||||
class IPv6Header(object, ConvertibleToBytes, BuildableFromBytes):
|
||||
class IPv6Header(ConvertibleToBytes, BuildableFromBytes):
|
||||
|
||||
""" Class representing IPv6 packet header. """
|
||||
|
||||
@@ -258,8 +264,8 @@ class IPv6Header(object, ConvertibleToBytes, BuildableFromBytes):
|
||||
if isinstance(value, bytearray):
|
||||
value = bytes(value)
|
||||
|
||||
elif isinstance(value, str):
|
||||
value = unicode(value)
|
||||
elif isinstance(value, str) and sys.version_info[0] == 2:
|
||||
value = value.decode("utf-8")
|
||||
|
||||
return ip_address(value)
|
||||
|
||||
@@ -439,7 +445,7 @@ class IPv6Packet(ConvertibleToBytes):
|
||||
return "IPv6Packet(\n\theader={})".format(self.ipv6_header)
|
||||
|
||||
|
||||
class UDPHeader(object, ConvertibleToBytes, BuildableFromBytes):
|
||||
class UDPHeader(ConvertibleToBytes, BuildableFromBytes):
|
||||
|
||||
""" Class representing UDP datagram header.
|
||||
|
||||
@@ -860,7 +866,7 @@ class IPv6PacketFactory(PacketFactory):
|
||||
return IPv6Packet(ipv6_header, upper_layer_protocol, extension_headers)
|
||||
|
||||
|
||||
class HopByHopOptionsFactory:
|
||||
class HopByHopOptionsFactory(object):
|
||||
|
||||
""" Factory that produces HopByHop options. """
|
||||
|
||||
|
||||
@@ -31,6 +31,7 @@ import collections
|
||||
import io
|
||||
import ipaddress
|
||||
import struct
|
||||
import sys
|
||||
|
||||
import common
|
||||
import config
|
||||
@@ -313,7 +314,8 @@ class Context():
|
||||
|
||||
def __init__(self, prefix, prefix_length=None):
|
||||
if isinstance(prefix, str):
|
||||
prefix = unicode(prefix)
|
||||
if sys.version_info[0] == 2:
|
||||
prefix = prefix.decode("utf-8")
|
||||
|
||||
prefix, prefix_length = prefix.split("/")
|
||||
prefix_length = int(prefix_length)
|
||||
@@ -453,7 +455,7 @@ class LowpanIpv6HeaderFactory:
|
||||
return (data_byte >> 6)
|
||||
|
||||
def _decompress_tf_4bytes(self, data):
|
||||
data_bytes = [ord(b) for b in data.read(4)]
|
||||
data_bytes = [b for b in bytearray(data.read(4))]
|
||||
|
||||
dscp = self._unpack_dscp(data_bytes[0])
|
||||
ecn = self._unpack_ecn(data_bytes[0])
|
||||
@@ -464,7 +466,7 @@ class LowpanIpv6HeaderFactory:
|
||||
return traffic_class, flow_label
|
||||
|
||||
def _decompress_tf_3bytes(self, data):
|
||||
data_bytes = [ord(b) for b in data.read(3)]
|
||||
data_bytes = [b for b in bytearray(data.read(3))]
|
||||
|
||||
ecn = self._unpack_ecn(data_bytes[0])
|
||||
|
||||
|
||||
@@ -153,7 +153,7 @@ class MacFrame:
|
||||
|
||||
# Check end of MAC frame
|
||||
if frame_type == MacHeader.FrameType.COMMAND:
|
||||
command_type = data.read(1)
|
||||
command_type = ord(data.read(1))
|
||||
else:
|
||||
command_type = None
|
||||
|
||||
|
||||
@@ -30,6 +30,7 @@
|
||||
import io
|
||||
import ipaddress
|
||||
import struct
|
||||
import sys
|
||||
|
||||
import common
|
||||
import ipv6
|
||||
@@ -193,7 +194,10 @@ class Message(object):
|
||||
assert sent_to_node == True
|
||||
|
||||
def assertSentToDestinationAddress(self, ipv6_address):
|
||||
assert self.ipv6_packet.ipv6_header.destination_address == ipaddress.ip_address(unicode(ipv6_address))
|
||||
if sys.version_info[0] == 2:
|
||||
ipv6_address = ipv6_address.decode("utf-8")
|
||||
|
||||
assert self.ipv6_packet.ipv6_header.destination_address == ipaddress.ip_address(ipv6_address)
|
||||
|
||||
def assertSentWithHopLimit(self, hop_limit):
|
||||
assert self.ipv6_packet.ipv6_header.hop_limit == hop_limit
|
||||
|
||||
@@ -511,7 +511,7 @@ class TlvRequest(object):
|
||||
class TlvRequestFactory:
|
||||
|
||||
def parse(self, data, message_info):
|
||||
tlvs = [ord(b) for b in data.read()]
|
||||
tlvs = [b for b in bytearray(data.read())]
|
||||
return TlvRequest(tlvs)
|
||||
|
||||
|
||||
|
||||
@@ -101,7 +101,7 @@ class Route(object):
|
||||
return "Route(border_router_16={}, prf={})".format(self.border_router_16, self.prf)
|
||||
|
||||
|
||||
class RouteFactory:
|
||||
class RouteFactory(object):
|
||||
|
||||
def parse(self, data, message_info):
|
||||
border_router_16 = struct.unpack(">H", data.read(2))[0]
|
||||
@@ -112,7 +112,7 @@ class RouteFactory:
|
||||
return Route(border_router_16, prf)
|
||||
|
||||
|
||||
class RoutesFactory:
|
||||
class RoutesFactory(object):
|
||||
|
||||
def __init__(self, route_factory):
|
||||
self._route_factory = route_factory
|
||||
@@ -149,7 +149,7 @@ class HasRoute(NetworkData):
|
||||
return "HasRoute(stable={}, routes=[{}])".format(self.stable, routes_str)
|
||||
|
||||
|
||||
class HasRouteFactory:
|
||||
class HasRouteFactory(object):
|
||||
|
||||
def __init__(self, routes_factory):
|
||||
self._routes_factory = routes_factory
|
||||
@@ -206,7 +206,7 @@ class PrefixSubTlvsFactory(SubTlvsFactory):
|
||||
super(PrefixSubTlvsFactory, self).__init__(sub_tlvs_factories)
|
||||
|
||||
|
||||
class PrefixFactory:
|
||||
class PrefixFactory(object):
|
||||
|
||||
def __init__(self, sub_tlvs_factory):
|
||||
self._sub_tlvs_factory = sub_tlvs_factory
|
||||
@@ -295,7 +295,7 @@ class BorderRouter(NetworkData):
|
||||
self.stable, self.border_router_16, self.prf, self.p, self.s, self.d, self.c, self.r, self.o, self.n)
|
||||
|
||||
|
||||
class BorderRouterFactory:
|
||||
class BorderRouterFactory(object):
|
||||
|
||||
def parse(self, data, message_info):
|
||||
border_router_16 = struct.unpack(">H", data.read(2))[0]
|
||||
@@ -348,7 +348,7 @@ class LowpanId(NetworkData):
|
||||
self.stable, self.c, self.cid, self.context_length)
|
||||
|
||||
|
||||
class LowpanIdFactory:
|
||||
class LowpanIdFactory(object):
|
||||
|
||||
def parse(self, data, message_info):
|
||||
data_byte = ord(data.read(1))
|
||||
@@ -361,14 +361,14 @@ class LowpanIdFactory:
|
||||
return LowpanId(c, cid, context_length, message_info.stable)
|
||||
|
||||
|
||||
class CommissioningData:
|
||||
class CommissioningData(object):
|
||||
|
||||
def __init__(self):
|
||||
# TODO: Not implemented yet
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
class CommissioningDataFactory:
|
||||
class CommissioningDataFactory(object):
|
||||
|
||||
def __init__(self):
|
||||
# TODO: Not implemented yet
|
||||
@@ -433,7 +433,7 @@ class ServiceSubTlvsFactory(SubTlvsFactory):
|
||||
super(ServiceSubTlvsFactory, self).__init__(sub_tlvs_factories)
|
||||
|
||||
|
||||
class ServiceFactory:
|
||||
class ServiceFactory(object):
|
||||
|
||||
def __init__(self, sub_tlvs_factory):
|
||||
self._sub_tlvs_factory = sub_tlvs_factory
|
||||
@@ -479,7 +479,7 @@ class Server(NetworkData):
|
||||
self.stable, self.server_16, hexlify(self.server_data))
|
||||
|
||||
|
||||
class ServerFactory:
|
||||
class ServerFactory(object):
|
||||
|
||||
def parse(self, data, message_info):
|
||||
server_16 = struct.unpack(">H", data.read(2))[0]
|
||||
|
||||
@@ -31,11 +31,15 @@ import collections
|
||||
import io
|
||||
import logging
|
||||
import os
|
||||
import Queue
|
||||
import select
|
||||
import socket
|
||||
import threading
|
||||
|
||||
try:
|
||||
import Queue
|
||||
except ImportError:
|
||||
import queue as Queue
|
||||
|
||||
import message
|
||||
|
||||
|
||||
@@ -78,6 +82,13 @@ class Sniffer:
|
||||
|
||||
self._buckets = collections.defaultdict(Queue.Queue)
|
||||
|
||||
def __del__(self):
|
||||
if self._socket is None:
|
||||
return
|
||||
|
||||
self._socket.close()
|
||||
del self._socket
|
||||
|
||||
def _nodeid_to_address(self, nodeid, ip_address="localhost"):
|
||||
return "", self.BASE_PORT + (self.PORT_OFFSET * self.WELLKNOWN_NODE_ID) + nodeid
|
||||
|
||||
@@ -88,7 +99,7 @@ class Sniffer:
|
||||
def _recv(self, fd):
|
||||
""" Receive data from socket with passed file descriptor. """
|
||||
|
||||
data, address = socket.fromfd(fd, socket.AF_INET, socket.SOCK_DGRAM).recvfrom(self.RECV_BUFFER_SIZE)
|
||||
data, address = self._socket.recvfrom(self.RECV_BUFFER_SIZE)
|
||||
|
||||
msg = self._message_factory.create(io.BytesIO(data))
|
||||
|
||||
|
||||
@@ -26,7 +26,6 @@
|
||||
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
|
||||
# POSSIBILITY OF SUCH DAMAGE.
|
||||
#
|
||||
|
||||
import random
|
||||
import struct
|
||||
import unittest
|
||||
@@ -36,7 +35,7 @@ import common
|
||||
|
||||
|
||||
def any_eui64():
|
||||
return bytearray([random.getrandbits(8) for _ in xrange(8)])
|
||||
return bytearray([random.getrandbits(8) for _ in range(8)])
|
||||
|
||||
|
||||
def any_rloc16_int():
|
||||
@@ -44,11 +43,11 @@ def any_rloc16_int():
|
||||
|
||||
|
||||
def any_rloc16_bytearray():
|
||||
return bytearray([random.getrandbits(8) for _ in xrange(2)])
|
||||
return bytearray([random.getrandbits(8) for _ in range(2)])
|
||||
|
||||
|
||||
def any_ipv6_address():
|
||||
return bytearray([random.getrandbits(8) for _ in xrange(16)])
|
||||
return bytearray([random.getrandbits(8) for _ in range(16)])
|
||||
|
||||
|
||||
class TestMessageInfo(unittest.TestCase):
|
||||
@@ -124,7 +123,7 @@ class TestMacAddress(unittest.TestCase):
|
||||
rloc16 = any_rloc16_int()
|
||||
|
||||
# WHEN
|
||||
mac_address = common.MacAddress.from_rloc16(rloc16)
|
||||
mac_address = common.MacAddress.from_rloc16(int(rloc16))
|
||||
|
||||
# THEN
|
||||
self.assertEqual(common.MacAddress.SHORT, mac_address.type)
|
||||
|
||||
@@ -49,7 +49,7 @@ def convert_aux_sec_hdr_to_bytearray(aux_sec_hdr):
|
||||
|
||||
|
||||
def any_eui64():
|
||||
return bytearray([random.getrandbits(8) for _ in xrange(8)])
|
||||
return bytearray([random.getrandbits(8) for _ in range(8)])
|
||||
|
||||
|
||||
def any_security_level():
|
||||
@@ -67,7 +67,7 @@ def any_key_id(key_id_mode):
|
||||
if key_id_mode == 2:
|
||||
length = 5
|
||||
|
||||
return bytearray([random.getrandbits(8) for _ in xrange(length)])
|
||||
return bytearray([random.getrandbits(8) for _ in range(length)])
|
||||
|
||||
|
||||
def any_auxiliary_security_header():
|
||||
@@ -82,17 +82,17 @@ def any_frame_counter():
|
||||
|
||||
|
||||
def any_ip_address():
|
||||
ip_address_bytes = bytearray([random.getrandbits(8) for _ in xrange(16)])
|
||||
ip_address_bytes = bytearray([random.getrandbits(8) for _ in range(16)])
|
||||
return ipaddress.ip_address(bytes(ip_address_bytes))
|
||||
|
||||
|
||||
def any_data(length=None):
|
||||
length = length if length is not None else random.randint(0, 128)
|
||||
return bytearray([random.getrandbits(8) for _ in xrange(length)])
|
||||
return bytearray([random.getrandbits(8) for _ in range(length)])
|
||||
|
||||
|
||||
def any_master_key():
|
||||
return bytearray([random.getrandbits(8) for _ in xrange(16)])
|
||||
return bytearray([random.getrandbits(8) for _ in range(16)])
|
||||
|
||||
|
||||
class TestCryptoEngine(unittest.TestCase):
|
||||
|
||||
@@ -34,7 +34,6 @@ import struct
|
||||
import unittest
|
||||
|
||||
from ipaddress import ip_address
|
||||
from mock import MagicMock, patch
|
||||
|
||||
from ipv6 import ICMPv6Header, UDPHeader, IPv6Header, IPv6PacketFactory, UDPDatagram, \
|
||||
UDPDatagramFactory, ICMPv6Factory, HopByHopFactory, MPLOptionFactory, ICMPv6, HopByHopOptionHeader, HopByHopOption, \
|
||||
@@ -154,12 +153,14 @@ def any_hop_by_hop_payload(next_header, hdr_ext_len, payload):
|
||||
|
||||
def any_body():
|
||||
length = any_uint(8)
|
||||
return bytearray("".join([random.choice(string.ascii_letters + string.digits + string.hexdigits) for _ in xrange(length)]))
|
||||
body = "".join([random.choice(string.ascii_letters + string.digits + string.hexdigits) for _ in range(length)])
|
||||
return bytearray(body.encode("utf-8"))
|
||||
|
||||
|
||||
def any_payload():
|
||||
length = any_uint(8)
|
||||
return bytearray("".join([random.choice(string.printable) for _ in range(length)]))
|
||||
payload = "".join([random.choice(string.printable) for _ in range(length)])
|
||||
return bytearray(payload.encode("utf-8"))
|
||||
|
||||
|
||||
def any_ip_address():
|
||||
@@ -196,7 +197,8 @@ def any_mpl_sequence():
|
||||
|
||||
def any_mpl_seed_id(S):
|
||||
length = MPLOption._seed_id_length[S]
|
||||
return bytearray("".join([random.choice(string.ascii_letters + string.digits + string.hexdigits) for _ in range(length)]))
|
||||
seed_id = "".join([random.choice(string.ascii_letters + string.digits + string.hexdigits) for _ in range(length)])
|
||||
return bytearray(seed_id.encode("utf-8"))
|
||||
|
||||
|
||||
def any_next_header():
|
||||
@@ -228,7 +230,8 @@ def any_length():
|
||||
|
||||
|
||||
def any_str(length=8):
|
||||
return "".join(random.choice(string.printable) for _ in range(length))
|
||||
s = "".join(random.choice(string.printable) for _ in range(length))
|
||||
return s.encode("utf-8")
|
||||
|
||||
|
||||
def any_bytes(length=4):
|
||||
|
||||
@@ -114,27 +114,27 @@ def any_hop_limit():
|
||||
|
||||
|
||||
def any_src_addr():
|
||||
return bytearray([random.getrandbits(8) for _ in xrange(16)])
|
||||
return bytearray([random.getrandbits(8) for _ in range(16)])
|
||||
|
||||
|
||||
def any_dst_addr():
|
||||
return bytearray([random.getrandbits(8) for _ in xrange(16)])
|
||||
return bytearray([random.getrandbits(8) for _ in range(16)])
|
||||
|
||||
|
||||
def any_eui64():
|
||||
return bytearray([random.getrandbits(8) for _ in xrange(8)])
|
||||
return bytearray([random.getrandbits(8) for _ in range(8)])
|
||||
|
||||
|
||||
def any_rloc16():
|
||||
return bytearray([random.getrandbits(8) for _ in xrange(2)])
|
||||
return bytearray([random.getrandbits(8) for _ in range(2)])
|
||||
|
||||
|
||||
def any_48bits_addr():
|
||||
return bytearray([random.getrandbits(8) for _ in xrange(6)])
|
||||
return bytearray([random.getrandbits(8) for _ in range(6)])
|
||||
|
||||
|
||||
def any_32bits_addr():
|
||||
return bytearray([random.getrandbits(8) for _ in xrange(4)])
|
||||
return bytearray([random.getrandbits(8) for _ in range(4)])
|
||||
|
||||
|
||||
def any_8bits_addr():
|
||||
@@ -190,15 +190,15 @@ def any_dci():
|
||||
|
||||
|
||||
def any_src_mac_addr():
|
||||
return bytearray([random.getrandbits(8) for _ in xrange(8)])
|
||||
return bytearray([random.getrandbits(8) for _ in range(8)])
|
||||
|
||||
|
||||
def any_dst_mac_addr():
|
||||
return bytearray([random.getrandbits(8) for _ in xrange(8)])
|
||||
return bytearray([random.getrandbits(8) for _ in range(8)])
|
||||
|
||||
|
||||
def any_context():
|
||||
prefix = bytearray([random.getrandbits(8) for _ in xrange(random.randint(2, 15))])
|
||||
prefix = bytearray([random.getrandbits(8) for _ in range(random.randint(2, 15))])
|
||||
prefix_length = len(prefix)
|
||||
return lowpan.Context(prefix, prefix_length * 8)
|
||||
|
||||
@@ -206,9 +206,9 @@ def any_context():
|
||||
def any_mac_address():
|
||||
length = random.choice([2, 8])
|
||||
if length == 2:
|
||||
return common.MacAddress.from_rloc16(bytearray([random.getrandbits(8) for _ in xrange(length)]))
|
||||
return common.MacAddress.from_rloc16(bytearray([random.getrandbits(8) for _ in range(length)]))
|
||||
elif length == 8:
|
||||
return common.MacAddress.from_eui64(bytearray([random.getrandbits(8) for _ in xrange(length)]))
|
||||
return common.MacAddress.from_eui64(bytearray([random.getrandbits(8) for _ in range(length)]))
|
||||
|
||||
|
||||
def any_hops_left():
|
||||
@@ -217,7 +217,7 @@ def any_hops_left():
|
||||
|
||||
def any_data(length=None):
|
||||
length = length if length is not None else random.randint(1, 64)
|
||||
return bytearray([random.getrandbits(8) for _ in xrange(length)])
|
||||
return bytearray([random.getrandbits(8) for _ in range(length)])
|
||||
|
||||
|
||||
def any_datagram_size():
|
||||
@@ -1171,7 +1171,7 @@ class TestLowpanUdpHeaderFactory(unittest.TestCase):
|
||||
src_port = any_src_port()
|
||||
dst_port = any_compressable_dst_port()
|
||||
|
||||
data_bytes = struct.pack(">H", src_port) + struct.pack(">H", dst_port)[1]
|
||||
data_bytes = struct.pack(">H", src_port) + bytearray([struct.pack(">H", dst_port)[1]])
|
||||
|
||||
# WHEN
|
||||
actual_src_port, actual_dst_port = factory._decompress_udp_ports(udphc, io.BytesIO(data_bytes))
|
||||
@@ -1192,7 +1192,7 @@ class TestLowpanUdpHeaderFactory(unittest.TestCase):
|
||||
src_port = any_compressable_src_port()
|
||||
dst_port = any_dst_port()
|
||||
|
||||
data_bytes = struct.pack(">H", src_port)[1] + struct.pack(">H", dst_port)
|
||||
data_bytes = bytearray([struct.pack(">H", src_port)[1]]) + struct.pack(">H", dst_port)
|
||||
|
||||
# WHEN
|
||||
actual_src_port, actual_dst_port = factory._decompress_udp_ports(udphc, io.BytesIO(data_bytes))
|
||||
@@ -1550,7 +1550,7 @@ class TestLowpanIpv6HeaderFactory(unittest.TestCase):
|
||||
def _merge_prefix_and_address(self, prefix, prefix_length, address):
|
||||
total_bytes = 16
|
||||
|
||||
prefix_length_in_bytes = prefix_length / 8
|
||||
prefix_length_in_bytes = int(prefix_length / 8)
|
||||
|
||||
if (prefix_length_in_bytes + len(address)) > total_bytes:
|
||||
total_bytes -= prefix_length_in_bytes
|
||||
|
||||
@@ -90,12 +90,12 @@ def any_timeout():
|
||||
|
||||
def any_challenge():
|
||||
length = random.randint(4, 8)
|
||||
return bytearray(random.getrandbits(8) for _ in xrange(length))
|
||||
return bytearray(random.getrandbits(8) for _ in range(length))
|
||||
|
||||
|
||||
def any_response():
|
||||
length = random.randint(4, 8)
|
||||
return bytearray(random.getrandbits(8) for _ in xrange(length))
|
||||
return bytearray(random.getrandbits(8) for _ in range(length))
|
||||
|
||||
|
||||
def any_link_layer_frame_counter():
|
||||
@@ -128,7 +128,7 @@ def any_router_id_mask():
|
||||
|
||||
def any_link_quality_and_route_data(length=None):
|
||||
length = length if length is not None else random.randint(0, 63)
|
||||
return [random.getrandbits(8) for _ in xrange(length)]
|
||||
return [random.getrandbits(8) for _ in range(length)]
|
||||
|
||||
|
||||
def any_partition_id():
|
||||
@@ -246,7 +246,7 @@ def any_tlvs(length=None):
|
||||
if length is None:
|
||||
length = random.randint(0, 16)
|
||||
|
||||
return [random.getrandbits(8) for _ in xrange(length)]
|
||||
return [random.getrandbits(8) for _ in range(length)]
|
||||
|
||||
|
||||
def any_cid():
|
||||
@@ -254,11 +254,11 @@ def any_cid():
|
||||
|
||||
|
||||
def any_iid():
|
||||
return bytearray([random.getrandbits(8) for _ in xrange(8)])
|
||||
return bytearray([random.getrandbits(8) for _ in range(8)])
|
||||
|
||||
|
||||
def any_ipv6_address():
|
||||
return bytearray([random.getrandbits(8) for _ in xrange(16)])
|
||||
return bytearray([random.getrandbits(8) for _ in range(16)])
|
||||
|
||||
|
||||
def any_addresses():
|
||||
@@ -292,11 +292,11 @@ def any_key_id(key_id_mode):
|
||||
elif key_id_mode == 3:
|
||||
length = 9
|
||||
|
||||
return bytearray([random.getrandbits(8) for _ in xrange(length)])
|
||||
return bytearray([random.getrandbits(8) for _ in range(length)])
|
||||
|
||||
|
||||
def any_eui64():
|
||||
return bytearray([random.getrandbits(8) for _ in xrange(8)])
|
||||
return bytearray([random.getrandbits(8) for _ in range(8)])
|
||||
|
||||
|
||||
class TestSourceAddress(unittest.TestCase):
|
||||
@@ -698,7 +698,7 @@ class TestRoute64Factory(unittest.TestCase):
|
||||
router_id_mask = any_router_id_mask()
|
||||
|
||||
router_count = 0
|
||||
for i in xrange(64):
|
||||
for i in range(64):
|
||||
router_count += (router_id_mask >> i) & 0x01
|
||||
|
||||
link_quality_and_route_data = any_link_quality_and_route_data(router_count)
|
||||
@@ -867,7 +867,7 @@ class TestNetworkDataFactory(unittest.TestCase):
|
||||
class DummyNetworkTlvsFactory:
|
||||
|
||||
def parse(self, data, context):
|
||||
return [ord(b) for b in data.read()]
|
||||
return [b for b in bytearray(data.read())]
|
||||
|
||||
tlvs = any_tlvs()
|
||||
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import io
|
||||
import math
|
||||
import random
|
||||
import struct
|
||||
import unittest
|
||||
@@ -110,7 +111,7 @@ def any_routes(count=None):
|
||||
if count is None:
|
||||
count = random.randint(0, 16)
|
||||
|
||||
return [any_route() for _ in xrange(6)]
|
||||
return [any_route() for _ in range(6)]
|
||||
|
||||
|
||||
def any_has_route():
|
||||
@@ -129,7 +130,7 @@ def any_prefix(prefix_length=None):
|
||||
if prefix_length is None:
|
||||
prefix_length = any_prefix_length()
|
||||
|
||||
return bytearray([random.getrandbits(8) for _ in xrange(prefix_length / 8)])
|
||||
return bytearray([random.getrandbits(8) for _ in range(int(math.ceil(prefix_length / 8)))])
|
||||
|
||||
|
||||
def any_p():
|
||||
@@ -185,7 +186,7 @@ def any_prefix_sub_tlvs():
|
||||
|
||||
sub_tlvs = []
|
||||
|
||||
for _id in xrange(random.randint(0, 16)):
|
||||
for _id in range(random.randint(1, 1)):
|
||||
c = random.choice(creator)
|
||||
sub_tlvs.append(c())
|
||||
|
||||
@@ -212,7 +213,7 @@ def any_service_data(data_length=None):
|
||||
if data_length is None:
|
||||
data_length = random.randint(0, 16)
|
||||
|
||||
return bytearray([random.getrandbits(8) for _ in xrange(data_length)])
|
||||
return bytearray([random.getrandbits(8) for _ in range(data_length)])
|
||||
|
||||
|
||||
def any_server_16():
|
||||
@@ -223,7 +224,7 @@ def any_server_data(data_length=None):
|
||||
if data_length is None:
|
||||
data_length = random.randint(0, 32)
|
||||
|
||||
return bytearray([random.getrandbits(8) for _ in xrange(data_length)])
|
||||
return bytearray([random.getrandbits(8) for _ in range(data_length)])
|
||||
|
||||
|
||||
def any_server():
|
||||
@@ -237,7 +238,7 @@ def any_service_sub_tlvs():
|
||||
|
||||
sub_tlvs = []
|
||||
|
||||
for _id in xrange(random.randint(0, 16)):
|
||||
for _id in range(random.randint(0, 16)):
|
||||
c = random.choice(creator)
|
||||
sub_tlvs.append(c())
|
||||
|
||||
|
||||
Reference in New Issue
Block a user