From fba3198c9e70162fcd437db4986cbb8d58e1be91 Mon Sep 17 00:00:00 2001 From: PFierek Date: Thu, 17 Nov 2016 14:16:59 +0100 Subject: [PATCH] Add Python3 support to thread-cert. (#985) --- .../thread-cert/Cert_5_1_01_RouterAttach.py | 6 ++-- .../Cert_5_1_03_RouterAddressReallocation.py | 2 +- .../Cert_5_1_04_RouterAddressReallocation.py | 2 +- tests/scripts/thread-cert/common.py | 20 ++++++------ tests/scripts/thread-cert/ipv6.py | 32 +++++++++++-------- tests/scripts/thread-cert/lowpan.py | 8 +++-- tests/scripts/thread-cert/mac802154.py | 2 +- tests/scripts/thread-cert/message.py | 6 +++- tests/scripts/thread-cert/mle.py | 2 +- tests/scripts/thread-cert/network_data.py | 20 ++++++------ tests/scripts/thread-cert/sniffer.py | 15 +++++++-- tests/scripts/thread-cert/test_common.py | 9 +++--- tests/scripts/thread-cert/test_crypto.py | 10 +++--- tests/scripts/thread-cert/test_ipv6.py | 13 +++++--- tests/scripts/thread-cert/test_lowpan.py | 30 ++++++++--------- tests/scripts/thread-cert/test_mle.py | 20 ++++++------ .../scripts/thread-cert/test_network_data.py | 13 ++++---- 17 files changed, 118 insertions(+), 92 deletions(-) diff --git a/tests/scripts/thread-cert/Cert_5_1_01_RouterAttach.py b/tests/scripts/thread-cert/Cert_5_1_01_RouterAttach.py index aa6fdd55d..515cdc7b9 100755 --- a/tests/scripts/thread-cert/Cert_5_1_01_RouterAttach.py +++ b/tests/scripts/thread-cert/Cert_5_1_01_RouterAttach.py @@ -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() diff --git a/tests/scripts/thread-cert/Cert_5_1_03_RouterAddressReallocation.py b/tests/scripts/thread-cert/Cert_5_1_03_RouterAddressReallocation.py index e8495ee18..3820937bc 100755 --- a/tests/scripts/thread-cert/Cert_5_1_03_RouterAddressReallocation.py +++ b/tests/scripts/thread-cert/Cert_5_1_03_RouterAddressReallocation.py @@ -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") diff --git a/tests/scripts/thread-cert/Cert_5_1_04_RouterAddressReallocation.py b/tests/scripts/thread-cert/Cert_5_1_04_RouterAddressReallocation.py index ef0b2956e..13b793b49 100755 --- a/tests/scripts/thread-cert/Cert_5_1_04_RouterAddressReallocation.py +++ b/tests/scripts/thread-cert/Cert_5_1_04_RouterAddressReallocation.py @@ -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") diff --git a/tests/scripts/thread-cert/common.py b/tests/scripts/thread-cert/common.py index baefaeefa..3e5f6f388 100644 --- a/tests/scripts/thread-cert/common.py +++ b/tests/scripts/thread-cert/common.py @@ -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] diff --git a/tests/scripts/thread-cert/ipv6.py b/tests/scripts/thread-cert/ipv6.py index 287c09777..505f078b5 100644 --- a/tests/scripts/thread-cert/ipv6.py +++ b/tests/scripts/thread-cert/ipv6.py @@ -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. """ diff --git a/tests/scripts/thread-cert/lowpan.py b/tests/scripts/thread-cert/lowpan.py index db3ef2bb2..b5827ac2b 100644 --- a/tests/scripts/thread-cert/lowpan.py +++ b/tests/scripts/thread-cert/lowpan.py @@ -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]) diff --git a/tests/scripts/thread-cert/mac802154.py b/tests/scripts/thread-cert/mac802154.py index 6969fef5e..6a0d4c4ef 100644 --- a/tests/scripts/thread-cert/mac802154.py +++ b/tests/scripts/thread-cert/mac802154.py @@ -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 diff --git a/tests/scripts/thread-cert/message.py b/tests/scripts/thread-cert/message.py index 616c99b1d..b771c3ad5 100644 --- a/tests/scripts/thread-cert/message.py +++ b/tests/scripts/thread-cert/message.py @@ -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 diff --git a/tests/scripts/thread-cert/mle.py b/tests/scripts/thread-cert/mle.py index 26064432b..4ad06f7df 100644 --- a/tests/scripts/thread-cert/mle.py +++ b/tests/scripts/thread-cert/mle.py @@ -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) diff --git a/tests/scripts/thread-cert/network_data.py b/tests/scripts/thread-cert/network_data.py index 5d4094b5f..6258f6d70 100644 --- a/tests/scripts/thread-cert/network_data.py +++ b/tests/scripts/thread-cert/network_data.py @@ -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] diff --git a/tests/scripts/thread-cert/sniffer.py b/tests/scripts/thread-cert/sniffer.py index 571bdb833..77c667801 100644 --- a/tests/scripts/thread-cert/sniffer.py +++ b/tests/scripts/thread-cert/sniffer.py @@ -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)) diff --git a/tests/scripts/thread-cert/test_common.py b/tests/scripts/thread-cert/test_common.py index 4114f230e..4fcd969a4 100644 --- a/tests/scripts/thread-cert/test_common.py +++ b/tests/scripts/thread-cert/test_common.py @@ -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) diff --git a/tests/scripts/thread-cert/test_crypto.py b/tests/scripts/thread-cert/test_crypto.py index e4bf38188..dcb73e8e5 100644 --- a/tests/scripts/thread-cert/test_crypto.py +++ b/tests/scripts/thread-cert/test_crypto.py @@ -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): diff --git a/tests/scripts/thread-cert/test_ipv6.py b/tests/scripts/thread-cert/test_ipv6.py index 284528983..b6832e0f6 100644 --- a/tests/scripts/thread-cert/test_ipv6.py +++ b/tests/scripts/thread-cert/test_ipv6.py @@ -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): diff --git a/tests/scripts/thread-cert/test_lowpan.py b/tests/scripts/thread-cert/test_lowpan.py index a68c26133..ec9270ba0 100644 --- a/tests/scripts/thread-cert/test_lowpan.py +++ b/tests/scripts/thread-cert/test_lowpan.py @@ -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 diff --git a/tests/scripts/thread-cert/test_mle.py b/tests/scripts/thread-cert/test_mle.py index cd428657f..7bad514c2 100644 --- a/tests/scripts/thread-cert/test_mle.py +++ b/tests/scripts/thread-cert/test_mle.py @@ -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() diff --git a/tests/scripts/thread-cert/test_network_data.py b/tests/scripts/thread-cert/test_network_data.py index b031620e2..895376237 100644 --- a/tests/scripts/thread-cert/test_network_data.py +++ b/tests/scripts/thread-cert/test_network_data.py @@ -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())