_BUF_SIZE = 16 * 1024 ** 2 # in bytes
-def _random_file(size: int) -> tempfile.NamedTemporaryFile:
- r = Random(0)
+def _random_file(size: int, rand_src: Random) -> tempfile.NamedTemporaryFile:
f = tempfile.NamedTemporaryFile(buffering = _BUF_SIZE)
try:
while f.tell() < size:
write_size = min(size - f.tell(), _BUF_SIZE)
- f.write(bytes(r.getrandbits(8) for _ in range(write_size)))
+ f.write(bytes(rand_src.getrandbits(8) for _ in range(write_size)))
f.seek(0)
return f
except Exception:
class Tests(unittest.TestCase):
_SCHEMAS = {"public", "diskjumble", "bittorrent"}
- _BUF_SIZE = 16 * 1024 ** 2 # in bytes
def _basic_fresh_verify_helper(self, read_size):
sector_size = 32
torrent_len = 3 * piece_size
disk = self._write_disk(sector_size, torrent_len // sector_size)
- with _random_file(torrent_len) as torrent_file:
+ with _random_file(torrent_len, Random(0)) as torrent_file:
torrent = _Torrent(torrent_file, piece_size)
torrent_file.seek(0)
self._write_torrent(torrent)