1
0
mirror of synced 2024-11-22 09:14:23 +01:00

Bump coverage for torrent handling

This commit is contained in:
jvoisin 2018-07-08 15:13:03 +02:00
parent b5fcddd6a6
commit 3cd4f9111f
2 changed files with 25 additions and 10 deletions

View File

@ -60,8 +60,6 @@ class _BencodeHandler(object):
def __decode_int(s: bytes) -> Tuple[int, bytes]: def __decode_int(s: bytes) -> Tuple[int, bytes]:
s = s[1:] s = s[1:]
next_idx = s.index(b'e') next_idx = s.index(b'e')
if next_idx is None:
raise ValueError # missing suffix
if s.startswith(b'-0'): if s.startswith(b'-0'):
raise ValueError # negative zero doesn't exist raise ValueError # negative zero doesn't exist
elif s.startswith(b'0') and next_idx != 1: elif s.startswith(b'0') and next_idx != 1:
@ -70,16 +68,13 @@ class _BencodeHandler(object):
@staticmethod @staticmethod
def __decode_string(s: bytes) -> Tuple[bytes, bytes]: def __decode_string(s: bytes) -> Tuple[bytes, bytes]:
sep = s.index(b':') colon = s.index(b':')
if set is None: str_len = int(s[:colon])
raise ValueError # missing suffix print('S: %s' % s)
str_len = int(s[:sep]) if s[0] == '0' and colon != 1:
if str_len < 0:
raise ValueError
elif s[0] == b'0' and sep != 1:
raise ValueError raise ValueError
s = s[1:] s = s[1:]
return s[sep:sep+str_len], s[sep+str_len:] return s[colon:colon+str_len], s[colon+str_len:]
def __decode_list(self, s: bytes) -> Tuple[list, bytes]: def __decode_list(self, s: bytes) -> Tuple[list, bytes]:
r = list() r = list()

View File

@ -54,6 +54,26 @@ class TestCorruptedFiles(unittest.TestCase):
with self.assertRaises(ValueError): with self.assertRaises(ValueError):
torrent.TorrentParser('./tests/data/clean.torrent') torrent.TorrentParser('./tests/data/clean.torrent')
with open("./tests/data/clean.torrent", "w") as f:
f.write("i-0e")
with self.assertRaises(ValueError):
torrent.TorrentParser('./tests/data/clean.torrent')
with open("./tests/data/clean.torrent", "w") as f:
f.write("i00e")
with self.assertRaises(ValueError):
torrent.TorrentParser('./tests/data/clean.torrent')
with open("./tests/data/clean.torrent", "w") as f:
f.write("d01:AAAAAAAAA")
with self.assertRaises(ValueError):
torrent.TorrentParser('./tests/data/clean.torrent')
with open("./tests/data/clean.torrent", "w") as f:
f.write("1:aaa")
with self.assertRaises(ValueError):
torrent.TorrentParser('./tests/data/clean.torrent')
os.remove('./tests/data/clean.torrent') os.remove('./tests/data/clean.torrent')
def test_odg(self): def test_odg(self):