Madiator2011/lbry-sdk
1
0
Fork 0
forked from LBRYCommunity/lbry-sdk
Code Pull requests Activity

test that creating a stream marks sd and head to announce

Browse source
This commit is contained in:
Victor Shyba 2019-02-05 01:07:37 -03:00
parent 6ead932ccb
commit 30846f932b
1 changed files with 22 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

22
tests/unit/stream/test_assembler.py
View file

@ -2,6 +2,7 @@ import os
import asyncio import asyncio
import tempfile import tempfile
import shutil import shutil
from torba.testcase import AsyncioTestCase from torba.testcase import AsyncioTestCase
from lbrynet.conf import Config from lbrynet.conf import Config
from lbrynet.blob.blob_file import MAX_BLOB_SIZE from lbrynet.blob.blob_file import MAX_BLOB_SIZE
@ -9,6 +10,7 @@ from lbrynet.extras.daemon.storage import SQLiteStorage
from lbrynet.blob.blob_manager import BlobFileManager from lbrynet.blob.blob_manager import BlobFileManager
from lbrynet.stream.assembler import StreamAssembler from lbrynet.stream.assembler import StreamAssembler
from lbrynet.stream.descriptor import StreamDescriptor from lbrynet.stream.descriptor import StreamDescriptor
from lbrynet.stream.stream_manager import StreamManager
class TestStreamAssembler(AsyncioTestCase): class TestStreamAssembler(AsyncioTestCase):
@ -80,3 +82,23 @@ class TestStreamAssembler(AsyncioTestCase):
async def test_create_and_decrypt_random(self): async def test_create_and_decrypt_random(self):
self.cleartext = os.urandom(20000000) self.cleartext = os.urandom(20000000)
await self.test_create_and_decrypt_one_blob_stream() await self.test_create_and_decrypt_one_blob_stream()
async def test_create_managed_stream_announces(self):
# setup a blob manager
storage = SQLiteStorage(Config(), ":memory:")
await storage.open()
tmp_dir = tempfile.mkdtemp()
self.addCleanup(lambda: shutil.rmtree(tmp_dir))
blob_manager = BlobFileManager(self.loop, tmp_dir, storage)
stream_manager = StreamManager(self.loop, Config(), blob_manager, None, storage, None)
# create the stream
download_dir = tempfile.mkdtemp()
self.addCleanup(lambda: shutil.rmtree(download_dir))
file_path = os.path.join(download_dir, "test_file")
with open(file_path, 'wb') as f:
f.write(b'testtest')
stream = await stream_manager.create_stream(file_path)
self.assertEqual(
[stream.sd_hash, stream.descriptor.blobs[0].blob_hash],
await storage.get_blobs_to_announce())