From 71c2a82b3d5a24cbf4a81e7017b81edb39e536fb Mon Sep 17 00:00:00 2001 From: Lonami Exo Date: Thu, 28 Sep 2017 11:49:45 +0200 Subject: [PATCH] Completely remove BinaryWriter from the project --- telethon/extensions/__init__.py | 4 +- telethon/extensions/binary_writer.py | 152 --------------------------- telethon_tests/utils_test.py | 77 +++++--------- 3 files changed, 25 insertions(+), 208 deletions(-) delete mode 100644 telethon/extensions/binary_writer.py diff --git a/telethon/extensions/__init__.py b/telethon/extensions/__init__.py index 06e2f087..0e932eac 100644 --- a/telethon/extensions/__init__.py +++ b/telethon/extensions/__init__.py @@ -1,9 +1,7 @@ """ Several extensions Python is missing, such as a proper class to handle a TCP communication with support for cancelling the operation, and an utility class -to work with arbitrary binary data in a more comfortable way (writing ints, -strings, bytes, etc.) +to read arbitrary binary data in a more comfortable way, with int/strings/etc. """ -from .binary_writer import BinaryWriter from .binary_reader import BinaryReader from .tcp_client import TcpClient diff --git a/telethon/extensions/binary_writer.py b/telethon/extensions/binary_writer.py deleted file mode 100644 index 8147e1fd..00000000 --- a/telethon/extensions/binary_writer.py +++ /dev/null @@ -1,152 +0,0 @@ -from io import BufferedWriter, BytesIO, DEFAULT_BUFFER_SIZE -from struct import pack - - -class BinaryWriter: - """ - Small utility class to write binary data. - Also creates a "Memory Stream" if necessary - """ - - def __init__(self, stream=None, known_length=None): - if not stream: - stream = BytesIO() - - if known_length is None: - # On some systems, DEFAULT_BUFFER_SIZE defaults to 8192 - # That's over 16 times as big as necessary for most messages - known_length = max(DEFAULT_BUFFER_SIZE, 1024) - - self.writer = BufferedWriter(stream, buffer_size=known_length) - self.written_count = 0 - - # region Writing - - # "All numbers are written as little endian." - # https://core.telegram.org/mtproto - def write_byte(self, value): - """Writes a single byte value""" - self.writer.write(pack('B', value)) - self.written_count += 1 - - def write_int(self, value, signed=True): - """Writes an integer value (4 bytes), optionally signed""" - self.writer.write( - int.to_bytes( - value, length=4, byteorder='little', signed=signed)) - self.written_count += 4 - - def write_long(self, value, signed=True): - """Writes a long integer value (8 bytes), optionally signed""" - self.writer.write( - int.to_bytes( - value, length=8, byteorder='little', signed=signed)) - self.written_count += 8 - - def write_float(self, value): - """Writes a floating point value (4 bytes)""" - self.writer.write(pack('> 8) % 256])) - self.write(bytes([(len(data) >> 16) % 256])) - self.write(data) - - self.write(bytes(padding)) - - def tgwrite_string(self, string): - """Write a string by using Telegram guidelines""" - self.tgwrite_bytes(string.encode('utf-8')) - - def tgwrite_bool(self, boolean): - """Write a boolean value by using Telegram guidelines""" - # boolTrue boolFalse - self.write_int(0x997275b5 if boolean else 0xbc799737, signed=False) - - def tgwrite_date(self, datetime): - """Converts a Python datetime object into Unix time - (used by Telegram) and writes it - """ - value = 0 if datetime is None else int(datetime.timestamp()) - self.write_int(value) - - def tgwrite_object(self, tlobject): - """Writes a Telegram object""" - self.write(tlobject.to_bytes()) - - def tgwrite_vector(self, vector): - """Writes a vector of Telegram objects""" - self.write_int(0x1cb5c415, signed=False) # Vector's constructor ID - self.write_int(len(vector)) - for item in vector: - self.tgwrite_object(item) - - # endregion - - def flush(self): - """Flush the current stream to "update" changes""" - self.writer.flush() - - def close(self): - """Close the current stream""" - self.writer.close() - - def get_bytes(self, flush=True): - """Get the current bytes array content from the buffer, - optionally flushing first - """ - if flush: - self.writer.flush() - return self.writer.raw.getvalue() - - def get_written_bytes_count(self): - """Gets the count of bytes written in the buffer. - This may NOT be equal to the stream length if one - was provided when initializing the writer - """ - return self.written_count - - # with block - def __enter__(self): - return self - - def __exit__(self, exc_type, exc_val, exc_tb): - self.close() diff --git a/telethon_tests/utils_test.py b/telethon_tests/utils_test.py index 5c53740c..790f3f4d 100644 --- a/telethon_tests/utils_test.py +++ b/telethon_tests/utils_test.py @@ -1,90 +1,61 @@ import os import unittest -from telethon.extensions import BinaryReader, BinaryWriter +from telethon.tl import TLObject +from telethon.extensions import BinaryReader class UtilsTests(unittest.TestCase): @staticmethod def test_binary_writer_reader(): - # Test that we can write and read properly - with BinaryWriter() as writer: - writer.write_byte(1) - writer.write_int(5) - writer.write_long(13) - writer.write_float(17.0) - writer.write_double(25.0) - writer.write(bytes([26, 27, 28, 29, 30, 31, 32])) - writer.write_large_int(2**127, 128, signed=False) - - data = writer.get_bytes() - expected = b'\x01\x05\x00\x00\x00\r\x00\x00\x00\x00\x00\x00\x00\x00\x00\x88A\x00\x00\x00\x00\x00\x00' \ - b'9@\x1a\x1b\x1c\x1d\x1e\x1f \x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x80' - - assert data == expected, 'Retrieved data does not match the expected value' + # Test that we can read properly + data = b'\x01\x05\x00\x00\x00\r\x00\x00\x00\x00\x00\x00\x00\x00\x00' \ + b'\x88A\x00\x00\x00\x00\x00\x009@\x1a\x1b\x1c\x1d\x1e\x1f ' \ + b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00' \ + b'\x00\x80' with BinaryReader(data) as reader: value = reader.read_byte() - assert value == 1, 'Example byte should be 1 but is {}'.format( - value) + assert value == 1, 'Example byte should be 1 but is {}'.format(value) value = reader.read_int() - assert value == 5, 'Example integer should be 5 but is {}'.format( - value) + assert value == 5, 'Example integer should be 5 but is {}'.format(value) value = reader.read_long() - assert value == 13, 'Example long integer should be 13 but is {}'.format( - value) + assert value == 13, 'Example long integer should be 13 but is {}'.format(value) value = reader.read_float() - assert value == 17.0, 'Example float should be 17.0 but is {}'.format( - value) + assert value == 17.0, 'Example float should be 17.0 but is {}'.format(value) value = reader.read_double() - assert value == 25.0, 'Example double should be 25.0 but is {}'.format( - value) + assert value == 25.0, 'Example double should be 25.0 but is {}'.format(value) value = reader.read(7) assert value == bytes([26, 27, 28, 29, 30, 31, 32]), 'Example bytes should be {} but is {}' \ .format(bytes([26, 27, 28, 29, 30, 31, 32]), value) value = reader.read_large_int(128, signed=False) - assert value == 2**127, 'Example large integer should be {} but is {}'.format( - 2**127, value) - - # Test Telegram that types are written right - with BinaryWriter() as writer: - writer.write_int(0x60469778) - buffer = writer.get_bytes() - valid = b'\x78\x97\x46\x60' # Tested written bytes using C#'s MemoryStream - - assert buffer == valid, 'Written type should be {} but is {}'.format( - list(valid), list(buffer)) + assert value == 2**127, 'Example large integer should be {} but is {}'.format(2**127, value) @staticmethod def test_binary_tgwriter_tgreader(): small_data = os.urandom(33) - small_data_padded = os.urandom( - 19) # +1 byte for length = 20 (evenly divisible by 4) + small_data_padded = os.urandom(19) # +1 byte for length = 20 (%4 = 0) large_data = os.urandom(999) large_data_padded = os.urandom(1024) data = (small_data, small_data_padded, large_data, large_data_padded) string = 'Testing Telegram strings, this should work properly!' + serialized = b''.join(TLObject.serialize_bytes(d) for d in data) + \ + TLObject.serialize_bytes(string) - with BinaryWriter() as writer: - # First write the data + with BinaryReader(serialized) as reader: + # And then try reading it without errors (it should be unharmed!) for datum in data: - writer.tgwrite_bytes(datum) - writer.tgwrite_string(string) + value = reader.tgread_bytes() + assert value == datum, 'Example bytes should be {} but is {}'.format( + datum, value) - with BinaryReader(writer.get_bytes()) as reader: - # And then try reading it without errors (it should be unharmed!) - for datum in data: - value = reader.tgread_bytes() - assert value == datum, 'Example bytes should be {} but is {}'.format( - datum, value) - - value = reader.tgread_string() - assert value == string, 'Example string should be {} but is {}'.format( - string, value) + value = reader.tgread_string() + assert value == string, 'Example string should be {} but is {}'.format( + string, value)