From ae11c5bb4b006027577f4bd93943164e6facc70c Mon Sep 17 00:00:00 2001 From: Jack Robison Date: Wed, 6 Feb 2019 12:18:24 -0500 Subject: [PATCH] fix setting finished event --- lbrynet/stream/assembler.py | 5 +++-- tests/unit/stream/test_downloader.py | 5 +++-- 2 files changed, 6 insertions(+), 4 deletions(-) diff --git a/lbrynet/stream/assembler.py b/lbrynet/stream/assembler.py index ae53dcc9a..37fc056b3 100644 --- a/lbrynet/stream/assembler.py +++ b/lbrynet/stream/assembler.py @@ -109,12 +109,13 @@ class StreamAssembler: self.descriptor.sd_hash) continue finally: - if written_blobs == len(self.descriptor.blobs) - 1: + if written_blobs == len(self.descriptor.blobs) - 2: log.debug("finished decrypting and assembling stream") self.stream_finished_event.set() await self.after_finished() else: - log.debug("stream decryption and assembly did not finish") + log.debug("stream decryption and assembly did not finish (%i/%i blobs are done)", written_blobs, + len(self.descriptor.blobs) - 2) async def get_blob(self, blob_hash: str, length: typing.Optional[int] = None) -> 'BlobFile': return self.blob_manager.get_blob(blob_hash, length) diff --git a/tests/unit/stream/test_downloader.py b/tests/unit/stream/test_downloader.py index bc9654e9b..33f86bf2f 100644 --- a/tests/unit/stream/test_downloader.py +++ b/tests/unit/stream/test_downloader.py @@ -37,15 +37,16 @@ class TestStreamDownloader(BlobExchangeTestBase): return q2, self.loop.create_task(_task()) mock_node.accumulate_peers = mock_accumulate_peers or _mock_accumulate_peers - self.downloader.download(mock_node) await self.downloader.stream_finished_event.wait() + self.assertTrue(self.downloader.stream_handle.closed) + self.assertTrue(os.path.isfile(self.downloader.output_path)) self.downloader.stop() + self.assertIs(self.downloader.stream_handle, None) self.assertTrue(os.path.isfile(self.downloader.output_path)) with open(self.downloader.output_path, 'rb') as f: self.assertEqual(f.read(), self.stream_bytes) await asyncio.sleep(0.01) - self.assertTrue(self.downloader.stream_handle.closed) async def test_transfer_stream(self): await self._test_transfer_stream(10)