LBRYFoundation/LBRY-Vault
1
0
Fork 0
mirror of https://github.com/LBRYFoundation/LBRY-Vault.git synced 2025-08-31 17:31:36 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

transaction.BCDataStream: minor fixes

Browse source 
- fix read/write_boolean (though unused...)
- sanity check in read_bytes
This commit is contained in:
SomberNight 2020-02-25 20:05:46 +01:00
parent c0be0471f2
commit 6937b87a7c
No known key found for this signature in database
GPG key ID: B33B5F232C6271E9
2 changed files with 22 additions and 5 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

14
electrum/tests/test_transaction.py
View file

@ -55,6 +55,8 @@ class TestBCDataStream(ElectrumTestCase):
def test_bytes(self):
s = transaction.BCDataStream()
with self.assertRaises(transaction.SerializationError):
s.read_bytes(1)
s.write(b'foobar')
self.assertEqual(s.read_bytes(3), b'foo')
self.assertEqual(s.read_bytes(2), b'ba')
@ -64,6 +66,18 @@ class TestBCDataStream(ElectrumTestCase):
self.assertEqual(s.read_bytes(1), b'r')
self.assertEqual(s.read_bytes(0), b'')
def test_bool(self):
s = transaction.BCDataStream()
s.write(b'f\x00\x00b')
self.assertTrue(s.read_boolean())
self.assertFalse(s.read_boolean())
self.assertFalse(s.read_boolean())
self.assertTrue(s.read_boolean())
s.write_boolean(True)
s.write_boolean(False)
self.assertEqual(b'\x01\x00', s.read_bytes(2))
self.assertFalse(s.can_read_more())
class TestTransaction(ElectrumTestCase):

13
electrum/transaction.py
View file

@ -244,7 +244,8 @@ class BCDataStream(object):
self.input = None
self.read_cursor = 0
def write(self, _bytes): # Initialize with string of _bytes
def write(self, _bytes: Union[bytes, bytearray]): # Initialize with string of _bytes
assert isinstance(_bytes, (bytes, bytearray))
if self.input is None:
self.input = bytearray(_bytes)
else:
@ -271,12 +272,14 @@ class BCDataStream(object):
self.write_compact_size(len(string))
self.write(string)
def read_bytes(self, length) -> bytes:
def read_bytes(self, length: int) -> bytes:
if self.input is None:
raise SerializationError("call write(bytes) before trying to deserialize")
assert length >= 0
input_len = len(self.input)
read_begin = self.read_cursor
read_end = read_begin + length
if 0 <= read_begin <= input_len and read_end <= input_len:
if 0 <= read_begin <= read_end <= input_len:
result = self.input[read_begin:read_end] # type: bytearray
self.read_cursor += length
return bytes(result)
@ -288,7 +291,7 @@ class BCDataStream(object):
return False
return self.read_cursor < len(self.input)
def read_boolean(self): return self.read_bytes(1)[0] != chr(0)
def read_boolean(self) -> bool: return self.read_bytes(1) != b'\x00'
def read_int16(self): return self._read_num('<h')
def read_uint16(self): return self._read_num('<H')
def read_int32(self): return self._read_num('<i')
@ -296,7 +299,7 @@ class BCDataStream(object):
def read_int64(self): return self._read_num('<q')
def read_uint64(self): return self._read_num('<Q')
def write_boolean(self, val): return self.write(chr(1) if val else chr(0))
def write_boolean(self, val): return self.write(b'\x01' if val else b'\x00')
def write_int16(self, val): return self._write_num('<h', val)
def write_uint16(self, val): return self._write_num('<H', val)
def write_int32(self, val): return self._write_num('<i', val)