<commit_before># Copyright (c) 2015 <NAME>
#
# See the file license.txt for copying permission.
import unittest
import asyncio
from hbmqtt.codecs.header import MQTTHeaderCodec, MQTTHeaderException
from hbmqtt.message import MessageType, MQTTHeader
from hbmqtt.codecs.utils import bytes_to_hex_str


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

    def test_decode_ok(self):
        stream = asyncio.StreamReader(loop=self.loop)
        stream.feed_data(b'\x10\x7f')
        header = self.loop.run_until_complete(MQTTHeaderCodec.decode(stream))
        self.assertEqual(header.message_type, MessageType.CONNECT)
        self.assertFalse(header.flags & 0x08)
        self.assertEqual((header.flags & 0x06) >> 1, 0)
        self.assertFalse(header.flags & 0x01)
        self.assertEqual(header.remaining_length, 127)

    def test_decode_ok_with_length(self):
        stream = asyncio.StreamReader(loop=self.loop)
        stream.feed_data(b'\x10\xff\xff\xff\x7f')
        header = self.loop.run_until_complete(MQTTHeaderCodec.decode(stream))
        self.assertEqual(header.message_type, MessageType.CONNECT)
        self.assertFalse(header.flags & 0x08)
        self.assertEqual((header.flags & 0x06) >> 1, 0)
        self.assertFalse(header.flags & 0x01)
        self.assertEqual(header.remaining_length, 268435455)

    def test_decode_reserved(self):
        stream = asyncio.StreamReader(loop=self.loop)
        stream.feed_data(b'\x0f\x7f')
        with self.assertRaises(MQTTHeaderException):
            self.loop.run_until_complete(MQTTHeaderCodec.decode(stream))

    def test_decode_ko_with_length(self):
        stream = asyncio.StreamReader(loop=self.loop)
        stream.feed_data(b'\x10\xff\xff\xff\xff\x7f')
        with self.assertRaises(MQTTHeaderException):
            self.loop.run_until_complete(MQTTHeaderCodec.decode(stream))

    def test_encode(self):
        header = MQTTHeader(MessageType.CONNECT, 0x00, 0)
        data = self.loop.run_until_complete(MQTTHeaderCodec.encode(header))
        self.assertEqual(data, b'\x10\x00')

    def test_encode_2(self):
        header = MQTTHeader(MessageType.CONNECT, 0x00, 268435455)
        data = self.loop.run_until_complete(MQTTHeaderCodec.encode(header))
        self.assertEqual(data, b'\x10\xff\xff\xff\x7f')<commit_msg>Fix unit testing<commit_after># Copyright (c) 2015 <NAME>
#
# See the file license.txt for copying permission.
import unittest
import asyncio
from hbmqtt.codecs.header import MQTTHeaderCodec, MQTTHeaderException
from hbmqtt.message import MessageType, MQTTHeader
from hbmqtt.codecs.utils import bytes_to_hex_str


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

    def test_decode_ok(self):
        stream = asyncio.StreamReader(loop=self.loop)
        stream.feed_data(b'\x10\x7f')
        header = self.loop.run_until_complete(MQTTHeaderCodec.decode(stream))
        self.assertEqual(header.message_type, MessageType.CONNECT)
        self.assertFalse(header.flags & 0x08)
        self.assertEqual((header.flags & 0x06) >> 1, 0)
        self.assertFalse(header.flags & 0x01)
        self.assertEqual(header.remaining_length, 127)

    def test_decode_ok_with_length(self):
        stream = asyncio.StreamReader(loop=self.loop)
        stream.feed_data(b'\x10\xff\xff\xff\x7f')
        header = self.loop.run_until_complete(MQTTHeaderCodec.decode(stream))
        self.assertEqual(header.message_type, MessageType.CONNECT)
        self.assertFalse(header.flags & 0x08)
        self.assertEqual((header.flags & 0x06) >> 1, 0)
        self.assertFalse(header.flags & 0x01)
        self.assertEqual(header.remaining_length, 268435455)

    def test_decode_reserved(self):
        stream = asyncio.StreamReader(loop=self.loop)
        stream.feed_data(b'\x0f\x7f')
        with self.assertRaises(MQTTHeaderException):
            self.loop.run_until_complete(MQTTHeaderCodec.decode(stream))

    def test_decode_ko_with_length(self):
        stream = asyncio.StreamReader(loop=self.loop)
        stream.feed_data(b'\x10\xff\xff\xff\xff\x7f')
        with self.assertRaises(MQTTHeaderException):
            self.loop.run_until_complete(MQTTHeaderCodec.decode(stream))

    def test_encode(self):
        header = MQTTHeader(MessageType.CONNECT, 0x00, 0)
        data = MQTTHeaderCodec.encode(header)
        self.assertEqual(data, b'\x10\x00')

    def test_encode_2(self):
        header = MQTTHeader(MessageType.CONNECT, 0x00, 268435455)
        data = MQTTHeaderCodec.encode(header)
        self.assertEqual(data, b'\x10\xff\xff\xff\x7f')
{'fraction_non_alphanumeric': 0.10765601544070341, 'fraction_numerical': 0.03131031524769462, 'mean_word_length': 3.5591397849462365, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1333431', 'n_tokens_mistral': 1691, 'n_tokens_neox': 1532, 'n_words': 221}