# Path: hbmqtt/codecs.py
# def bytes_to_hex_str(data):
#     """
#     converts a sequence of bytes into its displayable hex representation, ie: 0x??????
#     :param data: byte sequence
#     :return: Hexadecimal displayable representation
#     """
#     return '0x' + ''.join(format(b, '02x') for b in data)
# 
# def bytes_to_int(data):
#     """
#     convert a sequence of bytes to an integer using big endian byte ordering
#     :param data: byte sequence
#     :return: integer value
#     """
#     try:
#         return int.from_bytes(data, byteorder='big')
#     except:
#         return data
# 
# @asyncio.coroutine
# def decode_string(reader) -> bytes:
#     """
#     Read a string from a reader and decode it according to MQTT string specification
#     :param reader: Stream reader
#     :return: UTF-8 string read from stream
#     """
#     length_bytes = yield from read_or_raise(reader, 2)
#     str_length = unpack("!H", length_bytes)
#     if str_length[0]:
#         byte_str = yield from read_or_raise(reader, str_length[0])
#         try:
#             return byte_str.decode(encoding='utf-8')
#         except:
#             return str(byte_str)
#     else:
#         return ''
# 
# def encode_string(string: str) -> bytes:
#     data = string.encode(encoding='utf-8')
#     data_length = len(data)
#     return int_to_bytes(data_length, 2) + data


import unittest
import asyncio
from hbmqtt.codecs import (
    bytes_to_hex_str,
    bytes_to_int,
    decode_string,
    encode_string,
)
# Copyright (c) 2015 Nicolas JOUANIN
#
# See the file license.txt for copying permission.



class TestCodecs(unittest.TestCase):
    def setUp(self):
        self.loop = asyncio.new_event_loop()

    def test_bytes_to_hex_str(self):
        ret = bytes_to_hex_str(b'\x7f')
        self.assertEqual(ret, '0x7f')

    def test_bytes_to_int(self):
        ret = bytes_to_int(b'\x7f')
        self.assertEqual(ret, 127)

# Path: hbmqtt/codecs.py
# def bytes_to_hex_str(data):
#     """
#     converts a sequence of bytes into its displayable hex representation, ie: 0x??????
#     :param data: byte sequence
#     :return: Hexadecimal displayable representation
#     """
#     return '0x' + ''.join(format(b, '02x') for b in data)
# 
# def bytes_to_int(data):
#     """
#     convert a sequence of bytes to an integer using big endian byte ordering
#     :param data: byte sequence
#     :return: integer value
#     """
#     try:
#         return int.from_bytes(data, byteorder='big')
#     except:
#         return data
# 
# @asyncio.coroutine
# def decode_string(reader) -> bytes:
#     """
#     Read a string from a reader and decode it according to MQTT string specification
#     :param reader: Stream reader
#     :return: UTF-8 string read from stream
#     """
#     length_bytes = yield from read_or_raise(reader, 2)
#     str_length = unpack("!H", length_bytes)
#     if str_length[0]:
#         byte_str = yield from read_or_raise(reader, str_length[0])
#         try:
#             return byte_str.decode(encoding='utf-8')
#         except:
#             return str(byte_str)
#     else:
#         return ''
# 
# def encode_string(string: str) -> bytes:
#     data = string.encode(encoding='utf-8')
#     data_length = len(data)
#     return int_to_bytes(data_length, 2) + data


# Path: tests/test_codecs.py
import unittest
import asyncio
from hbmqtt.codecs import (
    bytes_to_hex_str,
    bytes_to_int,
    decode_string,
    encode_string,
)

# Copyright (c) 2015 Nicolas JOUANIN
#
# See the file license.txt for copying permission.



class TestCodecs(unittest.TestCase):
    def setUp(self):
        self.loop = asyncio.new_event_loop()

    def test_bytes_to_hex_str(self):
        ret = bytes_to_hex_str(b'\x7f')
        self.assertEqual(ret, '0x7f')

    def test_bytes_to_int(self):
        ret = bytes_to_int(b'\x7f')
        self.assertEqual(ret, 127)

        ret = bytes_to_int(b'\xff\xff')