<s> import shutil <EOL> from multiprocessing import Process , Event , Queue <EOL> import logging <EOL> import sys <EOL> import random <EOL> import io <EOL> from Crypto . PublicKey import RSA <EOL> from Crypto import Random <EOL> from Crypto . Hash import MD5 <EOL> from lbrynet . conf import MIN_BLOB_DATA_PAYMENT_RATE <EOL> from lbrynet . conf import MIN_BLOB_INFO_PAYMENT_RATE <EOL> from lbrynet . lbrylive . LiveStreamCreator import FileLiveStreamCreator <EOL> from lbrynet . lbrylive . PaymentRateManager import BaseLiveStreamPaymentRateManager <EOL> from lbrynet . lbrylive . PaymentRateManager import LiveStreamPaymentRateManager <EOL> from lbrynet . lbrylive . LiveStreamMetadataManager import DBLiveStreamMetadataManager <EOL> from lbrynet . lbrylive . LiveStreamMetadataManager import TempLiveStreamMetadataManager <EOL> from lbrynet . lbryfile . LBRYFileMetadataManager import TempLBRYFileMetadataManager , DBLBRYFileMetadataManager <EOL> from lbrynet . lbryfilemanager . LBRYFileManager import LBRYFileManager <EOL> from lbrynet . core . PaymentRateManager import PaymentRateManager <EOL> from lbrynet . core . PTCWallet import PointTraderKeyQueryHandlerFactory , PointTraderKeyExchanger <EOL> from lbrynet . core . Session import LBRYSession <EOL> from lbrynet . core . client . StandaloneBlobDownloader import StandaloneBlobDownloader <EOL> from lbrynet . core . StreamDescriptor import BlobStreamDescriptorWriter <EOL> from lbrynet . core . StreamDescriptor import StreamDescriptorIdentifier <EOL> from lbrynet . core . StreamDescriptor import download_sd_blob <EOL> from lbrynet . lbryfilemanager . LBRYFileCreator import create_lbry_file <EOL> from lbrynet . lbryfile . client . LBRYFileOptions import add_lbry_file_to_sd_identifier <EOL> from lbrynet . lbryfile . StreamDescriptor import get_sd_info <EOL> from twisted . internet import defer , threads , task <EOL> from twisted . trial . unittest import TestCase <EOL> from twisted . python . failure import Failure <EOL> import os <EOL> from lbrynet . core . PeerManager import PeerManager <EOL> from lbrynet . core . RateLimiter import DummyRateLimiter , RateLimiter <EOL> from lbrynet . core . server . BlobAvailabilityHandler import BlobAvailabilityHandlerFactory <EOL> from lbrynet . core . server . BlobRequestHandler import BlobRequestHandlerFactory <EOL> from lbrynet . core . server . ServerProtocol import ServerProtocolFactory <EOL> from lbrynet . lbrylive . server . LiveBlobInfoQueryHandler import CryptBlobInfoQueryHandlerFactory <EOL> from lbrynet . lbrylive . client . LiveStreamOptions import add_live_stream_to_sd_identifier <EOL> from lbrynet . lbrylive . client . LiveStreamDownloader import add_full_live_stream_downloader_to_sd_identifier <EOL> log_format = "<STR_LIT>" <EOL> logging . basicConfig ( level = logging . WARNING , format = log_format ) <EOL> class FakeNode ( object ) : <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> pass <EOL> def joinNetwork ( self , * args ) : <EOL> pass <EOL> def stop ( self ) : <EOL> pass <EOL> class FakeWallet ( object ) : <EOL> def __init__ ( self ) : <EOL> self . private_key = RSA . generate ( <NUM_LIT> ) <EOL> self . encoded_public_key = self . private_key . publickey ( ) . exportKey ( ) <EOL> def start ( self ) : <EOL> return defer . succeed ( True ) <EOL> def stop ( self ) : <EOL> return defer . succeed ( True ) <EOL> def get_info_exchanger ( self ) : <EOL> return PointTraderKeyExchanger ( self ) <EOL> def get_wallet_info_query_handler_factory ( self ) : <EOL> return PointTraderKeyQueryHandlerFactory ( self ) <EOL> def reserve_points ( self , * args ) : <EOL> return True <EOL> def cancel_point_reservation ( self , * args ) : <EOL> pass <EOL> def send_points ( self , * args ) : <EOL> return defer . succeed ( True ) <EOL> def add_expected_payment ( self , * args ) : <EOL> pass <EOL> def get_balance ( self ) : <EOL> return defer . succeed ( <NUM_LIT:1000> ) <EOL> def set_public_key_for_peer ( self , peer , public_key ) : <EOL> pass <EOL> class FakePeerFinder ( object ) : <EOL> def __init__ ( self , start_port , peer_manager , num_peers ) : <EOL> self . start_port = start_port <EOL> self . peer_manager = peer_manager <EOL> self . num_peers = num_peers <EOL> self . count = <NUM_LIT:0> <EOL> def find_peers_for_blob ( self , * args ) : <EOL> peer_port = self . start_port + self . count <EOL> self . count += <NUM_LIT:1> <EOL> if self . count >= self . num_peers : <EOL> self . count = <NUM_LIT:0> <EOL> return defer . succeed ( [ self . peer_manager . get_peer ( "<STR_LIT:127.0.0.1>" , peer_port ) ] ) <EOL> def run_manage_loop ( self ) : <EOL> pass <EOL> def stop ( self ) : <EOL> pass <EOL> class FakeAnnouncer ( object ) : <EOL> def __init__ ( self , * args ) : <EOL> pass <EOL> def add_supplier ( self , supplier ) : <EOL> pass <EOL> def immediate_announce ( self , * args ) : <EOL> pass <EOL> def run_manage_loop ( self ) : <EOL> pass <EOL> def stop ( self ) : <EOL> pass <EOL> class GenFile ( io . RawIOBase ) : <EOL> def __init__ ( self , size , pattern ) : <EOL> io . RawIOBase . __init__ ( self ) <EOL> self . size = size <EOL> self . pattern = pattern <EOL> self . read_so_far = <NUM_LIT:0> <EOL> self . buff = b'<STR_LIT>' <EOL> self . last_offset = <NUM_LIT:0> <EOL> def readable ( self ) : <EOL> return True <EOL> def writable ( self ) : <EOL> return False <EOL> def read ( self , n = - <NUM_LIT:1> ) : <EOL> if n > - <NUM_LIT:1> : <EOL> bytes_to_read = min ( n , self . size - self . read_so_far ) <EOL> else : <EOL> bytes_to_read = self . size - self . read_so_far <EOL> output , self . buff = self . buff [ : bytes_to_read ] , self . buff [ bytes_to_read : ] <EOL> bytes_to_read -= len ( output ) <EOL> while bytes_to_read > <NUM_LIT:0> : <EOL> self . buff = self . _generate_chunk ( ) <EOL> new_output , self . buff = self . buff [ : bytes_to_read ] , self . buff [ bytes_to_read : ] <EOL> bytes_to_read -= len ( new_output ) <EOL> output += new_output <EOL> self . read_so_far += len ( output ) <EOL> return output <EOL> def readall ( self ) : <EOL> return self . read ( ) <EOL> def _generate_chunk ( self , n = <NUM_LIT:2> ** <NUM_LIT:10> ) : <EOL> output = self . pattern [ self . last_offset : self . last_offset + n ] <EOL> n_left = n - len ( output ) <EOL> whole_patterns = n_left / len ( self . pattern ) <EOL> output += self . pattern * whole_patterns <EOL> self . last_offset = n - len ( output ) <EOL> output += self . pattern [ : self . last_offset ] <EOL> return output <EOL> test_create_stream_sd_file = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ <EOL> { '<STR_LIT>' : <NUM_LIT> , '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : <NUM_LIT> , '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : <NUM_LIT> , '<STR_LIT>' : <NUM_LIT:2> , <EOL> '<STR_LIT>' : <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : <NUM_LIT:0> , '<STR_LIT>' : <NUM_LIT:3> , '<STR_LIT>' : '<STR_LIT>' } ] , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:key>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> def start_lbry_uploader ( sd_hash_queue , kill_event , dead_event , file_size , ul_rate_limit = None ) : <EOL> sys . modules = sys . modules . copy ( ) <EOL> del sys . modules [ '<STR_LIT>' ] <EOL> import twisted . internet <EOL> twisted . internet . reactor = twisted . internet . epollreactor . EPollReactor ( ) <EOL> sys . modules [ '<STR_LIT>' ] = twisted . internet . reactor <EOL> from twisted . internet import reactor <EOL> logging . debug ( "<STR_LIT>" ) <EOL> Random . atfork ( ) <EOL> r = random . Random ( ) <EOL> r . seed ( "<STR_LIT>" ) <EOL> wallet = FakeWallet ( ) <EOL> peer_manager = PeerManager ( ) <EOL> peer_finder = FakePeerFinder ( <NUM_LIT> , peer_manager , <NUM_LIT:1> ) <EOL> hash_announcer = FakeAnnouncer ( ) <EOL> rate_limiter = RateLimiter ( ) <EOL> sd_identifier = StreamDescriptorIdentifier ( ) <EOL> db_dir = "<STR_LIT>" <EOL> os . mkdir ( db_dir ) <EOL> session = LBRYSession ( MIN_BLOB_DATA_PAYMENT_RATE , db_dir = db_dir , lbryid = "<STR_LIT>" , <EOL> peer_finder = peer_finder , hash_announcer = hash_announcer , peer_port = <NUM_LIT> , <EOL> use_upnp = False , rate_limiter = rate_limiter , wallet = wallet ) <EOL> stream_info_manager = TempLBRYFileMetadataManager ( ) <EOL> lbry_file_manager = LBRYFileManager ( session , stream_info_manager , sd_identifier ) <EOL> if ul_rate_limit is not None : <EOL> session . rate_limiter . set_ul_limit ( ul_rate_limit ) <EOL> def start_all ( ) : <EOL> d = session . setup ( ) <EOL> d . addCallback ( lambda _ : add_lbry_file_to_sd_identifier ( sd_identifier ) ) <EOL> d . addCallback ( lambda _ : lbry_file_manager . setup ( ) ) <EOL> d . addCallback ( lambda _ : start_server ( ) ) <EOL> d . addCallback ( lambda _ : create_stream ( ) ) <EOL> d . addCallback ( create_stream_descriptor ) <EOL> d . addCallback ( put_sd_hash_on_queue ) <EOL> def print_error ( err ) : <EOL> logging . critical ( "<STR_LIT>" , err . getErrorMessage ( ) ) <EOL> d . addErrback ( print_error ) <EOL> return d <EOL> def start_server ( ) : <EOL> server_port = None <EOL> query_handler_factories = { <EOL> BlobAvailabilityHandlerFactory ( session . blob_manager ) : True , <EOL> BlobRequestHandlerFactory ( session . blob_manager , session . wallet , <EOL> PaymentRateManager ( session . base_payment_rate_manager ) ) : True , <EOL> session . wallet . get_wallet_info_query_handler_factory ( ) : True , <EOL> } <EOL> server_factory = ServerProtocolFactory ( session . rate_limiter , <EOL> query_handler_factories , <EOL> session . peer_manager ) <EOL> server_port = reactor . listenTCP ( <NUM_LIT> , server_factory ) <EOL> logging . debug ( "<STR_LIT>" ) <EOL> def kill_server ( ) : <EOL> ds = [ ] <EOL> ds . append ( session . shut_down ( ) ) <EOL> ds . append ( lbry_file_manager . stop ( ) ) <EOL> if server_port : <EOL> ds . append ( server_port . stopListening ( ) ) <EOL> kill_check . stop ( ) <EOL> dead_event . set ( ) <EOL> dl = defer . DeferredList ( ds ) <EOL> dl . addCallback ( lambda _ : reactor . stop ( ) ) <EOL> return dl <EOL> def check_for_kill ( ) : <EOL> if kill_event . is_set ( ) : <EOL> kill_server ( ) <EOL> kill_check = task . LoopingCall ( check_for_kill ) <EOL> kill_check . start ( <NUM_LIT:1.0> ) <EOL> return True <EOL> def create_stream ( ) : <EOL> test_file = GenFile ( file_size , b'<STR_LIT>' . join ( [ chr ( i ) for i in xrange ( <NUM_LIT:0> , <NUM_LIT:64> , <NUM_LIT:6> ) ] ) ) <EOL> d = create_lbry_file ( session , lbry_file_manager , "<STR_LIT>" , test_file ) <EOL> return d <EOL> def create_stream_descriptor ( stream_hash ) : <EOL> descriptor_writer = BlobStreamDescriptorWriter ( session . blob_manager ) <EOL> d = get_sd_info ( lbry_file_manager . stream_info_manager , stream_hash , True ) <EOL> d . addCallback ( descriptor_writer . create_descriptor ) <EOL> return d <EOL> def put_sd_hash_on_queue ( sd_hash ) : <EOL> sd_hash_queue . put ( sd_hash ) <EOL> reactor . callLater ( <NUM_LIT:1> , start_all ) <EOL> reactor . run ( ) <EOL> def start_lbry_reuploader ( sd_hash , kill_event , dead_event , ready_event , n , ul_rate_limit = None ) : <EOL> sys . modules = sys . modules . copy ( ) <EOL> del sys . modules [ '<STR_LIT>' ] <EOL> import twisted . internet <EOL> twisted . internet . reactor = twisted . internet . epollreactor . EPollReactor ( ) <EOL> sys . modules [ '<STR_LIT>' ] = twisted . internet . reactor <EOL> from twisted . internet import reactor <EOL> logging . debug ( "<STR_LIT>" ) <EOL> Random . atfork ( ) <EOL> r = random . Random ( ) <EOL> r . seed ( "<STR_LIT>" ) <EOL> wallet = FakeWallet ( ) <EOL> peer_port = <NUM_LIT> + n <EOL> peer_manager = PeerManager ( ) <EOL> peer_finder = FakePeerFinder ( <NUM_LIT> , peer_manager , <NUM_LIT:1> ) <EOL> hash_announcer = FakeAnnouncer ( ) <EOL> rate_limiter = RateLimiter ( ) <EOL> sd_identifier = StreamDescriptorIdentifier ( ) <EOL> db_dir = "<STR_LIT>" + str ( n ) <EOL> blob_dir = os . path . join ( db_dir , "<STR_LIT>" ) <EOL> os . mkdir ( db_dir ) <EOL> os . mkdir ( blob_dir ) <EOL> session = LBRYSession ( MIN_BLOB_DATA_PAYMENT_RATE , db_dir = db_dir , lbryid = "<STR_LIT>" + str ( n ) , <EOL> peer_finder = peer_finder , hash_announcer = hash_announcer , <EOL> blob_dir = None , peer_port = peer_port , <EOL> use_upnp = False , rate_limiter = rate_limiter , wallet = wallet ) <EOL> stream_info_manager = TempLBRYFileMetadataManager ( ) <EOL> lbry_file_manager = LBRYFileManager ( session , stream_info_manager , sd_identifier ) <EOL> if ul_rate_limit is not None : <EOL> session . rate_limiter . set_ul_limit ( ul_rate_limit ) <EOL> def make_downloader ( metadata , prm ) : <EOL> info_validator = metadata . validator <EOL> options = metadata . options <EOL> factories = metadata . factories <EOL> chosen_options = [ o . default_value for o in options . get_downloader_options ( info_validator , prm ) ] <EOL> return factories [ <NUM_LIT:0> ] . make_downloader ( metadata , chosen_options , prm ) <EOL> def download_file ( ) : <EOL> prm = PaymentRateManager ( session . base_payment_rate_manager ) <EOL> d = download_sd_blob ( session , sd_hash , prm ) <EOL> d . addCallback ( sd_identifier . get_metadata_for_sd_blob ) <EOL> d . addCallback ( make_downloader , prm ) <EOL> d . addCallback ( lambda downloader : downloader . start ( ) ) <EOL> return d <EOL> def start_transfer ( ) : <EOL> logging . debug ( "<STR_LIT>" ) <EOL> d = session . setup ( ) <EOL> d . addCallback ( lambda _ : add_lbry_file_to_sd_identifier ( sd_identifier ) ) <EOL> d . addCallback ( lambda _ : lbry_file_manager . setup ( ) ) <EOL> d . addCallback ( lambda _ : download_file ( ) ) <EOL> return d <EOL> def start_server ( ) : <EOL> server_port = None <EOL> query_handler_factories = { <EOL> BlobAvailabilityHandlerFactory ( session . blob_manager ) : True , <EOL> BlobRequestHandlerFactory ( session . blob_manager , session . wallet , <EOL> PaymentRateManager ( session . base_payment_rate_manager ) ) : True , <EOL> session . wallet . get_wallet_info_query_handler_factory ( ) : True , <EOL> } <EOL> server_factory = ServerProtocolFactory ( session . rate_limiter , <EOL> query_handler_factories , <EOL> session . peer_manager ) <EOL> server_port = reactor . listenTCP ( peer_port , server_factory ) <EOL> logging . debug ( "<STR_LIT>" ) <EOL> def kill_server ( ) : <EOL> ds = [ ] <EOL> ds . append ( session . shut_down ( ) ) <EOL> ds . append ( lbry_file_manager . stop ( ) ) <EOL> if server_port : <EOL> ds . append ( server_port . stopListening ( ) ) <EOL> kill_check . stop ( ) <EOL> dead_event . set ( ) <EOL> dl = defer . DeferredList ( ds ) <EOL> dl . addCallback ( lambda _ : reactor . stop ( ) ) <EOL> return dl <EOL> def check_for_kill ( ) : <EOL> if kill_event . is_set ( ) : <EOL> kill_server ( ) <EOL> kill_check = task . LoopingCall ( check_for_kill ) <EOL> kill_check . start ( <NUM_LIT:1.0> ) <EOL> ready_event . set ( ) <EOL> logging . debug ( "<STR_LIT>" ) <EOL> d = task . deferLater ( reactor , <NUM_LIT:1.0> , start_transfer ) <EOL> d . addCallback ( lambda _ : start_server ( ) ) <EOL> reactor . run ( ) <EOL> def start_live_server ( sd_hash_queue , kill_event , dead_event ) : <EOL> sys . modules = sys . modules . copy ( ) <EOL> del sys . modules [ '<STR_LIT>' ] <EOL> import twisted . internet <EOL> twisted . internet . reactor = twisted . internet . epollreactor . EPollReactor ( ) <EOL> sys . modules [ '<STR_LIT>' ] = twisted . internet . reactor <EOL> from twisted . internet import reactor <EOL> logging . debug ( "<STR_LIT>" ) <EOL> Random . atfork ( ) <EOL> r = random . Random ( ) <EOL> r . seed ( "<STR_LIT>" ) <EOL> wallet = FakeWallet ( ) <EOL> peer_manager = PeerManager ( ) <EOL> peer_finder = FakePeerFinder ( <NUM_LIT> , peer_manager , <NUM_LIT:1> ) <EOL> hash_announcer = FakeAnnouncer ( ) <EOL> rate_limiter = DummyRateLimiter ( ) <EOL> sd_identifier = StreamDescriptorIdentifier ( ) <EOL> db_dir = "<STR_LIT>" <EOL> os . mkdir ( db_dir ) <EOL> session = LBRYSession ( MIN_BLOB_DATA_PAYMENT_RATE , db_dir = db_dir , lbryid = "<STR_LIT>" , <EOL> peer_finder = peer_finder , hash_announcer = hash_announcer , peer_port = <NUM_LIT> , <EOL> use_upnp = False , rate_limiter = rate_limiter , wallet = wallet ) <EOL> base_payment_rate_manager = BaseLiveStreamPaymentRateManager ( MIN_BLOB_INFO_PAYMENT_RATE ) <EOL> data_payment_rate_manager = PaymentRateManager ( session . base_payment_rate_manager ) <EOL> payment_rate_manager = LiveStreamPaymentRateManager ( base_payment_rate_manager , <EOL> data_payment_rate_manager ) <EOL> stream_info_manager = DBLiveStreamMetadataManager ( session . db_dir , hash_announcer ) <EOL> logging . debug ( "<STR_LIT>" ) <EOL> server_port = [ ] <EOL> def start_listening ( ) : <EOL> logging . debug ( "<STR_LIT>" ) <EOL> query_handler_factories = { <EOL> CryptBlobInfoQueryHandlerFactory ( stream_info_manager , session . wallet , <EOL> payment_rate_manager ) : True , <EOL> BlobAvailabilityHandlerFactory ( session . blob_manager ) : True , <EOL> BlobRequestHandlerFactory ( session . blob_manager , session . wallet , <EOL> payment_rate_manager ) : True , <EOL> session . wallet . get_wallet_info_query_handler_factory ( ) : True , <EOL> } <EOL> server_factory = ServerProtocolFactory ( session . rate_limiter , <EOL> query_handler_factories , <EOL> session . peer_manager ) <EOL> server_port . append ( reactor . listenTCP ( <NUM_LIT> , server_factory ) ) <EOL> logging . debug ( "<STR_LIT>" ) <EOL> def create_stream ( ) : <EOL> logging . debug ( "<STR_LIT>" ) <EOL> test_file = GenFile ( <NUM_LIT> , b'<STR_LIT>' . join ( [ chr ( i + <NUM_LIT:2> ) for i in xrange ( <NUM_LIT:0> , <NUM_LIT:64> , <NUM_LIT:6> ) ] ) ) <EOL> stream_creator_helper = FileLiveStreamCreator ( session . blob_manager , stream_info_manager , <EOL> "<STR_LIT>" , test_file ) <EOL> d = stream_creator_helper . setup ( ) <EOL> d . addCallback ( lambda _ : stream_creator_helper . publish_stream_descriptor ( ) ) <EOL> d . addCallback ( put_sd_hash_on_queue ) <EOL> d . addCallback ( lambda _ : stream_creator_helper . start_streaming ( ) ) <EOL> return d <EOL> def put_sd_hash_on_queue ( sd_hash ) : <EOL> logging . debug ( "<STR_LIT>" , str ( sd_hash ) ) <EOL> sd_hash_queue . put ( sd_hash ) <EOL> logging . debug ( "<STR_LIT>" ) <EOL> def set_dead_event ( ) : <EOL> logging . debug ( "<STR_LIT>" ) <EOL> dead_event . set ( ) <EOL> def print_error ( err ) : <EOL> logging . debug ( "<STR_LIT>" , err . getTraceback ( ) ) <EOL> def stop_reactor ( ) : <EOL> logging . debug ( "<STR_LIT>" ) <EOL> reactor . stop ( ) <EOL> def shut_down ( arg ) : <EOL> logging . debug ( "<STR_LIT>" ) <EOL> if isinstance ( arg , Failure ) : <EOL> logging . error ( "<STR_LIT>" , arg . getTraceback ( ) ) <EOL> d = defer . maybeDeferred ( server_port [ <NUM_LIT:0> ] . stopListening ) <EOL> d . addErrback ( print_error ) <EOL> d . addCallback ( lambda _ : session . shut_down ( ) ) <EOL> d . addCallback ( lambda _ : stream_info_manager . stop ( ) ) <EOL> d . addErrback ( print_error ) <EOL> d . addCallback ( lambda _ : set_dead_event ( ) ) <EOL> d . addErrback ( print_error ) <EOL> d . addCallback ( lambda _ : reactor . callLater ( <NUM_LIT:0> , stop_reactor ) ) <EOL> d . addErrback ( print_error ) <EOL> return d <EOL> def wait_for_kill_event ( ) : <EOL> d = defer . Deferred ( ) <EOL> def check_for_kill ( ) : <EOL> if kill_event . is_set ( ) : <EOL> logging . debug ( "<STR_LIT>" ) <EOL> kill_check . stop ( ) <EOL> d . callback ( True ) <EOL> kill_check = task . LoopingCall ( check_for_kill ) <EOL> kill_check . start ( <NUM_LIT:1.0> ) <EOL> return d <EOL> def enable_live_stream ( ) : <EOL> base_live_stream_payment_rate_manager = BaseLiveStreamPaymentRateManager ( <EOL> MIN_BLOB_INFO_PAYMENT_RATE <EOL> ) <EOL> add_live_stream_to_sd_identifier ( sd_identifier , base_live_stream_payment_rate_manager ) <EOL> add_full_live_stream_downloader_to_sd_identifier ( session , stream_info_manager , sd_identifier , <EOL> base_live_stream_payment_rate_manager ) <EOL> def run_server ( ) : <EOL> d = session . setup ( ) <EOL> d . addCallback ( lambda _ : stream_info_manager . setup ( ) ) <EOL> d . addCallback ( lambda _ : enable_live_stream ( ) ) <EOL> d . addCallback ( lambda _ : start_listening ( ) ) <EOL> d . addCallback ( lambda _ : create_stream ( ) ) <EOL> d . addCallback ( lambda _ : wait_for_kill_event ( ) ) <EOL> d . addBoth ( shut_down ) <EOL> return d <EOL> reactor . callLater ( <NUM_LIT:1> , run_server ) <EOL> reactor . run ( ) <EOL> def start_blob_uploader ( blob_hash_queue , kill_event , dead_event , slow ) : <EOL> sys . modules = sys . modules . copy ( ) <EOL> del sys . modules [ '<STR_LIT>' ] <EOL> import twisted . internet <EOL> twisted . internet . reactor = twisted . internet . epollreactor . EPollReactor ( ) <EOL> sys . modules [ '<STR_LIT>' ] = twisted . internet . reactor <EOL> from twisted . internet import reactor <EOL> logging . debug ( "<STR_LIT>" ) <EOL> Random . atfork ( ) <EOL> wallet = FakeWallet ( ) <EOL> peer_manager = PeerManager ( ) <EOL> peer_finder = FakePeerFinder ( <NUM_LIT> , peer_manager , <NUM_LIT:1> ) <EOL> hash_announcer = FakeAnnouncer ( ) <EOL> rate_limiter = RateLimiter ( ) <EOL> if slow is True : <EOL> peer_port = <NUM_LIT> <EOL> db_dir = "<STR_LIT>" <EOL> else : <EOL> peer_port = <NUM_LIT> <EOL> db_dir = "<STR_LIT>" <EOL> blob_dir = os . path . join ( db_dir , "<STR_LIT>" ) <EOL> os . mkdir ( db_dir ) <EOL> os . mkdir ( blob_dir ) <EOL> session = LBRYSession ( MIN_BLOB_DATA_PAYMENT_RATE , db_dir = db_dir , lbryid = "<STR_LIT>" , <EOL> peer_finder = peer_finder , hash_announcer = hash_announcer , <EOL> blob_dir = blob_dir , peer_port = peer_port , <EOL> use_upnp = False , rate_limiter = rate_limiter , wallet = wallet ) <EOL> if slow is True : <EOL> session . rate_limiter . set_ul_limit ( <NUM_LIT:2> ** <NUM_LIT:11> ) <EOL> def start_all ( ) : <EOL> d = session . setup ( ) <EOL> d . addCallback ( lambda _ : start_server ( ) ) <EOL> d . addCallback ( lambda _ : create_single_blob ( ) ) <EOL> d . addCallback ( put_blob_hash_on_queue ) <EOL> def print_error ( err ) : <EOL> logging . critical ( "<STR_LIT>" , err . getErrorMessage ( ) ) <EOL> d . addErrback ( print_error ) <EOL> return d <EOL> def start_server ( ) : <EOL> server_port = None <EOL> query_handler_factories = { <EOL> BlobAvailabilityHandlerFactory ( session . blob_manager ) : True , <EOL> BlobRequestHandlerFactory ( session . blob_manager , session . wallet , <EOL> PaymentRateManager ( session . base_payment_rate_manager ) ) : True , <EOL> session . wallet . get_wallet_info_query_handler_factory ( ) : True , <EOL> } <EOL> server_factory = ServerProtocolFactory ( session . rate_limiter , <EOL> query_handler_factories , <EOL> session . peer_manager ) <EOL> server_port = reactor . listenTCP ( peer_port , server_factory ) <EOL> logging . debug ( "<STR_LIT>" ) <EOL> def kill_server ( ) : <EOL> ds = [ ] <EOL> ds . append ( session . shut_down ( ) ) <EOL> if server_port : <EOL> ds . append ( server_port . stopListening ( ) ) <EOL> kill_check . stop ( ) <EOL> dead_event . set ( ) <EOL> dl = defer . DeferredList ( ds ) <EOL> dl . addCallback ( lambda _ : reactor . stop ( ) ) <EOL> return dl <EOL> def check_for_kill ( ) : <EOL> if kill_event . is_set ( ) : <EOL> kill_server ( ) <EOL> kill_check = task . LoopingCall ( check_for_kill ) <EOL> kill_check . start ( <NUM_LIT:1.0> ) <EOL> return True <EOL> def create_single_blob ( ) : <EOL> blob_creator = session . blob_manager . get_blob_creator ( ) <EOL> blob_creator . write ( "<STR_LIT:0>" * <NUM_LIT:2> ** <NUM_LIT> ) <EOL> return blob_creator . close ( ) <EOL> def put_blob_hash_on_queue ( blob_hash ) : <EOL> logging . debug ( "<STR_LIT>" , str ( blob_hash ) ) <EOL> blob_hash_queue . put ( blob_hash ) <EOL> logging . debug ( "<STR_LIT>" ) <EOL> reactor . callLater ( <NUM_LIT:1> , start_all ) <EOL> reactor . run ( ) <EOL> class TestTransfer ( TestCase ) : <EOL> def setUp ( self ) : <EOL> self . server_processes = [ ] <EOL> self . session = None <EOL> self . stream_info_manager = None <EOL> self . lbry_file_manager = None <EOL> self . addCleanup ( self . take_down_env ) <EOL> def take_down_env ( self ) : <EOL> d = defer . succeed ( True ) <EOL> if self . lbry_file_manager is not None : <EOL> d . addCallback ( lambda _ : self . lbry_file_manager . stop ( ) ) <EOL> if self . session is not None : <EOL> d . addCallback ( lambda _ : self . session . shut_down ( ) ) <EOL> if self . stream_info_manager is not None : <EOL> d . addCallback ( lambda _ : self . stream_info_manager . stop ( ) ) <EOL> def delete_test_env ( ) : <EOL> dirs = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> files = [ '<STR_LIT>' ] <EOL> for di in dirs : <EOL> if os . path . exists ( di ) : <EOL> shutil . rmtree ( di ) <EOL> for f in files : <EOL> if os . path . exists ( f ) : <EOL> os . remove ( f ) <EOL> for p in self . server_processes : <EOL> p . terminate ( ) <EOL> return True <EOL> d . addCallback ( lambda _ : threads . deferToThread ( delete_test_env ) ) <EOL> return d <EOL> @ staticmethod <EOL> def wait_for_event ( event , timeout ) : <EOL> from twisted . internet import reactor <EOL> d = defer . Deferred ( ) <EOL> def stop ( ) : <EOL> set_check . stop ( ) <EOL> if stop_call . active ( ) : <EOL> stop_call . cancel ( ) <EOL> d . callback ( True ) <EOL> def check_if_event_set ( ) : <EOL> if event . is_set ( ) : <EOL> logging . debug ( "<STR_LIT>" ) <EOL> stop ( ) <EOL> def done_waiting ( ) : <EOL> logging . warning ( "<STR_LIT>" ) <EOL> stop ( ) <EOL> set_check = task . LoopingCall ( check_if_event_set ) <EOL> set_check . start ( <NUM_LIT> ) <EOL> stop_call = reactor . callLater ( timeout , done_waiting ) <EOL> return d <EOL> @ staticmethod <EOL> def wait_for_hash_from_queue ( hash_queue ) : <EOL> logging . debug ( "<STR_LIT>" ) <EOL> d = defer . Deferred ( ) <EOL> def check_for_start ( ) : <EOL> if hash_queue . empty ( ) is False : <EOL> logging . debug ( "<STR_LIT>" ) <EOL> start_check . stop ( ) <EOL> d . callback ( hash_queue . get ( False ) ) <EOL> else : <EOL> logging . debug ( "<STR_LIT>" ) <EOL> start_check = task . LoopingCall ( check_for_start ) <EOL> start_check . start ( <NUM_LIT:1.0> ) <EOL> return d <EOL> def test_lbry_transfer ( self ) : <EOL> sd_hash_queue = Queue ( ) <EOL> kill_event = Event ( ) <EOL> dead_event = Event ( ) <EOL> uploader = Process ( target = start_lbry_uploader , args = ( sd_hash_queue , kill_event , dead_event , <NUM_LIT> ) ) <EOL> uploader . start ( ) <EOL> self . server_processes . append ( uploader ) <EOL> logging . debug ( "<STR_LIT>" ) <EOL> wallet = FakeWallet ( ) <EOL> peer_manager = PeerManager ( ) <EOL> peer_finder = FakePeerFinder ( <NUM_LIT> , peer_manager , <NUM_LIT:1> ) <EOL> hash_announcer = FakeAnnouncer ( ) <EOL> rate_limiter = DummyRateLimiter ( ) <EOL> sd_identifier = StreamDescriptorIdentifier ( ) <EOL> db_dir = "<STR_LIT>" <EOL> blob_dir = os . path . join ( db_dir , "<STR_LIT>" ) <EOL> os . mkdir ( db_dir ) <EOL> os . mkdir ( blob_dir ) <EOL> self . session = LBRYSession ( MIN_BLOB_DATA_PAYMENT_RATE , db_dir = db_dir , lbryid = "<STR_LIT>" , <EOL> peer_finder = peer_finder , hash_announcer = hash_announcer , <EOL> blob_dir = blob_dir , peer_port = <NUM_LIT> , <EOL> use_upnp = False , rate_limiter = rate_limiter , wallet = wallet ) <EOL> self . stream_info_manager = TempLBRYFileMetadataManager ( ) <EOL> self . lbry_file_manager = LBRYFileManager ( self . session , self . stream_info_manager , sd_identifier ) <EOL> def make_downloader ( metadata , prm ) : <EOL> info_validator = metadata . validator <EOL> options = metadata . options <EOL> factories = metadata . factories <EOL> chosen_options = [ o . default_value for o in options . get_downloader_options ( info_validator , prm ) ] <EOL> return factories [ <NUM_LIT:0> ] . make_downloader ( metadata , chosen_options , prm ) <EOL> def download_file ( sd_hash ) : <EOL> prm = PaymentRateManager ( self . session . base_payment_rate_manager ) <EOL> d = download_sd_blob ( self . session , sd_hash , prm ) <EOL> d . addCallback ( sd_identifier . get_metadata_for_sd_blob ) <EOL> d . addCallback ( make_downloader , prm ) <EOL> d . addCallback ( lambda downloader : downloader . start ( ) ) <EOL> return d <EOL> def check_md5_sum ( ) : <EOL> f = open ( '<STR_LIT>' ) <EOL> hashsum = MD5 . new ( ) <EOL> hashsum . update ( f . read ( ) ) <EOL> self . assertEqual ( hashsum . hexdigest ( ) , "<STR_LIT>" ) <EOL> def start_transfer ( sd_hash ) : <EOL> logging . debug ( "<STR_LIT>" ) <EOL> d = self . session . setup ( ) <EOL> d . addCallback ( lambda _ : add_lbry_file_to_sd_identifier ( sd_identifier ) ) <EOL> d . addCallback ( lambda _ : self . lbry_file_manager . setup ( ) ) <EOL> d . addCallback ( lambda _ : download_file ( sd_hash ) ) <EOL> d . addCallback ( lambda _ : check_md5_sum ( ) ) <EOL> return d <EOL> def stop ( arg ) : <EOL> if isinstance ( arg , Failure ) : <EOL> logging . debug ( "<STR_LIT>" , arg . getTraceback ( ) ) <EOL> else : <EOL> logging . debug ( "<STR_LIT>" ) <EOL> kill_event . set ( ) <EOL> logging . debug ( "<STR_LIT>" ) <EOL> d = self . wait_for_event ( dead_event , <NUM_LIT:15> ) <EOL> def print_shutting_down ( ) : <EOL> logging . info ( "<STR_LIT>" ) <EOL> d . addCallback ( lambda _ : print_shutting_down ( ) ) <EOL> d . addCallback ( lambda _ : arg ) <EOL> return d <EOL> d = self . wait_for_hash_from_queue ( sd_hash_queue ) <EOL> d . addCallback ( start_transfer ) <EOL> d . addBoth ( stop ) <EOL> return d <EOL> def test_live_transfer ( self ) : <EOL> sd_hash_queue = Queue ( ) <EOL> kill_event = Event ( ) <EOL> dead_event = Event ( ) <EOL> server_args = ( sd_hash_queue , kill_event , dead_event ) <EOL> server = Process ( target = start_live_server , args = server_args ) <EOL> server . start ( ) <EOL> self . server_processes . append ( server ) <EOL> wallet = FakeWallet ( ) <EOL> peer_manager = PeerManager ( ) <EOL> peer_finder = FakePeerFinder ( <NUM_LIT> , peer_manager , <NUM_LIT:1> ) <EOL> hash_announcer = FakeAnnouncer ( ) <EOL> rate_limiter = DummyRateLimiter ( ) <EOL> sd_identifier = StreamDescriptorIdentifier ( ) <EOL> db_dir = "<STR_LIT>" <EOL> os . mkdir ( db_dir ) <EOL> self . session = LBRYSession ( MIN_BLOB_DATA_PAYMENT_RATE , db_dir = db_dir , lbryid = "<STR_LIT>" , <EOL> peer_finder = peer_finder , hash_announcer = hash_announcer , blob_dir = None , <EOL> peer_port = <NUM_LIT> , use_upnp = False , rate_limiter = rate_limiter , wallet = wallet ) <EOL> self . stream_info_manager = TempLiveStreamMetadataManager ( hash_announcer ) <EOL> d = self . wait_for_hash_from_queue ( sd_hash_queue ) <EOL> def create_downloader ( metadata , prm ) : <EOL> info_validator = metadata . validator <EOL> options = metadata . options <EOL> factories = metadata . factories <EOL> chosen_options = [ o . default_value for o in options . get_downloader_options ( info_validator , prm ) ] <EOL> return factories [ <NUM_LIT:0> ] . make_downloader ( metadata , chosen_options , prm ) <EOL> def start_lbry_file ( lbry_file ) : <EOL> lbry_file = lbry_file <EOL> logging . debug ( "<STR_LIT>" ) <EOL> return lbry_file . start ( ) <EOL> def download_stream ( sd_blob_hash ) : <EOL> logging . debug ( "<STR_LIT>" ) <EOL> prm = PaymentRateManager ( self . session . base_payment_rate_manager ) <EOL> d = download_sd_blob ( self . session , sd_blob_hash , prm ) <EOL> d . addCallback ( sd_identifier . get_metadata_for_sd_blob ) <EOL> d . addCallback ( create_downloader , prm ) <EOL> d . addCallback ( start_lbry_file ) <EOL> return d <EOL> def do_download ( sd_blob_hash ) : <EOL> logging . debug ( "<STR_LIT>" ) <EOL> d = self . session . setup ( ) <EOL> d . addCallback ( lambda _ : enable_live_stream ( ) ) <EOL> d . addCallback ( lambda _ : download_stream ( sd_blob_hash ) ) <EOL> return d <EOL> def enable_live_stream ( ) : <EOL> base_live_stream_payment_rate_manager = BaseLiveStreamPaymentRateManager ( <EOL> MIN_BLOB_INFO_PAYMENT_RATE <EOL> ) <EOL> add_live_stream_to_sd_identifier ( sd_identifier , <EOL> base_live_stream_payment_rate_manager ) <EOL> add_full_live_stream_downloader_to_sd_identifier ( self . session , self . stream_info_manager , <EOL> sd_identifier , <EOL> base_live_stream_payment_rate_manager ) <EOL> d . addCallback ( do_download ) <EOL> def check_md5_sum ( ) : <EOL> f = open ( '<STR_LIT>' ) <EOL> hashsum = MD5 . new ( ) <EOL> hashsum . update ( f . read ( ) ) <EOL> self . assertEqual ( hashsum . hexdigest ( ) , "<STR_LIT>" ) <EOL> d . addCallback ( lambda _ : check_md5_sum ( ) ) <EOL> def stop ( arg ) : <EOL> if isinstance ( arg , Failure ) : <EOL> logging . debug ( "<STR_LIT>" , arg . getTraceback ( ) ) <EOL> else : <EOL> logging . debug ( "<STR_LIT>" ) <EOL> kill_event . set ( ) <EOL> logging . debug ( "<STR_LIT>" ) <EOL> d = self . wait_for_event ( dead_event , <NUM_LIT:15> ) <EOL> def print_shutting_down ( ) : <EOL> logging . info ( "<STR_LIT>" ) <EOL> d . addCallback ( lambda _ : print_shutting_down ( ) ) <EOL> d . addCallback ( lambda _ : arg ) <EOL> return d <EOL> d . addBoth ( stop ) <EOL> return d <EOL> def test_last_blob_retrieval ( self ) : <EOL> kill_event = Event ( ) <EOL> dead_event_1 = Event ( ) <EOL> blob_hash_queue_1 = Queue ( ) <EOL> blob_hash_queue_2 = Queue ( ) <EOL> fast_uploader = Process ( target = start_blob_uploader , <EOL> args = ( blob_hash_queue_1 , kill_event , dead_event_1 , False ) ) <EOL> fast_uploader . start ( ) <EOL> self . server_processes . append ( fast_uploader ) <EOL> dead_event_2 = Event ( ) <EOL> slow_uploader = Process ( target = start_blob_uploader , <EOL> args = ( blob_hash_queue_2 , kill_event , dead_event_2 , True ) ) <EOL> slow_uploader . start ( ) <EOL> self . server_processes . append ( slow_uploader ) <EOL> logging . debug ( "<STR_LIT>" ) <EOL> wallet = FakeWallet ( ) <EOL> peer_manager = PeerManager ( ) <EOL> peer_finder = FakePeerFinder ( <NUM_LIT> , peer_manager , <NUM_LIT:2> ) <EOL> hash_announcer = FakeAnnouncer ( ) <EOL> rate_limiter = DummyRateLimiter ( ) <EOL> db_dir = "<STR_LIT>" <EOL> blob_dir = os . path . join ( db_dir , "<STR_LIT>" ) <EOL> os . mkdir ( db_dir ) <EOL> os . mkdir ( blob_dir ) <EOL> self . session = LBRYSession ( MIN_BLOB_DATA_PAYMENT_RATE , db_dir = db_dir , lbryid = "<STR_LIT>" , <EOL> peer_finder = peer_finder , hash_announcer = hash_announcer , <EOL> blob_dir = blob_dir , peer_port = <NUM_LIT> , <EOL> use_upnp = False , rate_limiter = rate_limiter , wallet = wallet ) <EOL> d1 = self . wait_for_hash_from_queue ( blob_hash_queue_1 ) <EOL> d2 = self . wait_for_hash_from_queue ( blob_hash_queue_2 ) <EOL> d = defer . DeferredList ( [ d1 , d2 ] , fireOnOneErrback = True ) <EOL> def get_blob_hash ( results ) : <EOL> self . assertEqual ( results [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] , results [ <NUM_LIT:1> ] [ <NUM_LIT:1> ] ) <EOL> return results [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] <EOL> d . addCallback ( get_blob_hash ) <EOL> def download_blob ( blob_hash ) : <EOL> prm = PaymentRateManager ( self . session . base_payment_rate_manager ) <EOL> downloader = StandaloneBlobDownloader ( blob_hash , self . session . blob_manager , peer_finder , <EOL> rate_limiter , prm , wallet ) <EOL> d = downloader . download ( ) <EOL> return d <EOL> def start_transfer ( blob_hash ) : <EOL> logging . debug ( "<STR_LIT>" ) <EOL> d = self . session . setup ( ) <EOL> d . addCallback ( lambda _ : download_blob ( blob_hash ) ) <EOL> return d <EOL> d . addCallback ( start_transfer ) <EOL> def stop ( arg ) : <EOL> if isinstance ( arg , Failure ) : <EOL> logging . debug ( "<STR_LIT>" , arg . getTraceback ( ) ) <EOL> else : <EOL> logging . debug ( "<STR_LIT>" ) <EOL> kill_event . set ( ) <EOL> logging . debug ( "<STR_LIT>" ) <EOL> d1 = self . wait_for_event ( dead_event_1 , <NUM_LIT:15> ) <EOL> d2 = self . wait_for_event ( dead_event_2 , <NUM_LIT:15> ) <EOL> dl = defer . DeferredList ( [ d1 , d2 ] ) <EOL> def print_shutting_down ( ) : <EOL> logging . info ( "<STR_LIT>" ) <EOL> dl . addCallback ( lambda _ : print_shutting_down ( ) ) <EOL> dl . addCallback ( lambda _ : arg ) <EOL> return dl <EOL> d . addBoth ( stop ) <EOL> return d <EOL> def test_double_download ( self ) : <EOL> sd_hash_queue = Queue ( ) <EOL> kill_event = Event ( ) <EOL> dead_event = Event ( ) <EOL> uploader = Process ( target = start_lbry_uploader , args = ( sd_hash_queue , kill_event , dead_event , <NUM_LIT> ) ) <EOL> uploader . start ( ) <EOL> self . server_processes . append ( uploader ) <EOL> logging . debug ( "<STR_LIT>" ) <EOL> wallet = FakeWallet ( ) <EOL> peer_manager = PeerManager ( ) <EOL> peer_finder = FakePeerFinder ( <NUM_LIT> , peer_manager , <NUM_LIT:1> ) <EOL> hash_announcer = FakeAnnouncer ( ) <EOL> rate_limiter = DummyRateLimiter ( ) <EOL> sd_identifier = StreamDescriptorIdentifier ( ) <EOL> downloaders = [ ] <EOL> db_dir = "<STR_LIT>" <EOL> blob_dir = os . path . join ( db_dir , "<STR_LIT>" ) <EOL> os . mkdir ( db_dir ) <EOL> os . mkdir ( blob_dir ) <EOL> self . session = LBRYSession ( MIN_BLOB_DATA_PAYMENT_RATE , db_dir = db_dir , lbryid = "<STR_LIT>" , <EOL> peer_finder = peer_finder , hash_announcer = hash_announcer , <EOL> blob_dir = blob_dir , peer_port = <NUM_LIT> , use_upnp = False , <EOL> rate_limiter = rate_limiter , wallet = wallet ) <EOL> self . stream_info_manager = DBLBRYFileMetadataManager ( self . session . db_dir ) <EOL> self . lbry_file_manager = LBRYFileManager ( self . session , self . stream_info_manager , sd_identifier ) <EOL> def make_downloader ( metadata , prm ) : <EOL> info_validator = metadata . validator <EOL> options = metadata . options <EOL> factories = metadata . factories <EOL> chosen_options = [ o . default_value for o in options . get_downloader_options ( info_validator , prm ) ] <EOL> return factories [ <NUM_LIT:0> ] . make_downloader ( metadata , chosen_options , prm ) <EOL> def append_downloader ( downloader ) : <EOL> downloaders . append ( downloader ) <EOL> return downloader <EOL> def download_file ( sd_hash ) : <EOL> prm = PaymentRateManager ( self . session . base_payment_rate_manager ) <EOL> d = download_sd_blob ( self . session , sd_hash , prm ) <EOL> d . addCallback ( sd_identifier . get_metadata_for_sd_blob ) <EOL> d . addCallback ( make_downloader , prm ) <EOL> d . addCallback ( append_downloader ) <EOL> d . addCallback ( lambda downloader : downloader . start ( ) ) <EOL> return d <EOL> def check_md5_sum ( ) : <EOL> f = open ( '<STR_LIT>' ) <EOL> hashsum = MD5 . new ( ) <EOL> hashsum . update ( f . read ( ) ) <EOL> self . assertEqual ( hashsum . hexdigest ( ) , "<STR_LIT>" ) <EOL> def delete_lbry_file ( ) : <EOL> logging . debug ( "<STR_LIT>" ) <EOL> d = self . lbry_file_manager . delete_lbry_file ( downloaders [ <NUM_LIT:0> ] ) <EOL> d . addCallback ( lambda _ : self . lbry_file_manager . get_count_for_stream_hash ( downloaders [ <NUM_LIT:0> ] . stream_hash ) ) <EOL> d . addCallback ( lambda c : self . stream_info_manager . delete_stream ( downloaders [ <NUM_LIT:1> ] . stream_hash ) if c == <NUM_LIT:0> else True ) <EOL> return d <EOL> def check_lbry_file ( ) : <EOL> d = downloaders [ <NUM_LIT:1> ] . status ( ) <EOL> d . addCallback ( lambda _ : downloaders [ <NUM_LIT:1> ] . status ( ) ) <EOL> def check_status_report ( status_report ) : <EOL> self . assertEqual ( status_report . num_known , status_report . num_completed ) <EOL> self . assertEqual ( status_report . num_known , <NUM_LIT:3> ) <EOL> d . addCallback ( check_status_report ) <EOL> return d <EOL> def start_transfer ( sd_hash ) : <EOL> logging . debug ( "<STR_LIT>" ) <EOL> d = self . session . setup ( ) <EOL> d . addCallback ( lambda _ : self . stream_info_manager . setup ( ) ) <EOL> d . addCallback ( lambda _ : add_lbry_file_to_sd_identifier ( sd_identifier ) ) <EOL> d . addCallback ( lambda _ : self . lbry_file_manager . setup ( ) ) <EOL> d . addCallback ( lambda _ : download_file ( sd_hash ) ) <EOL> d . addCallback ( lambda _ : check_md5_sum ( ) ) <EOL> d . addCallback ( lambda _ : download_file ( sd_hash ) ) <EOL> d . addCallback ( lambda _ : delete_lbry_file ( ) ) <EOL> d . addCallback ( lambda _ : check_lbry_file ( ) ) <EOL> return d <EOL> def stop ( arg ) : <EOL> if isinstance ( arg , Failure ) : <EOL> logging . debug ( "<STR_LIT>" , arg . getTraceback ( ) ) <EOL> else : <EOL> logging . debug ( "<STR_LIT>" ) <EOL> kill_event . set ( ) <EOL> logging . debug ( "<STR_LIT>" ) <EOL> d = self . wait_for_event ( dead_event , <NUM_LIT:15> ) <EOL> def print_shutting_down ( ) : <EOL> logging . info ( "<STR_LIT>" ) <EOL> d . addCallback ( lambda _ : print_shutting_down ( ) ) <EOL> d . addCallback ( lambda _ : arg ) <EOL> return d <EOL> d = self . wait_for_hash_from_queue ( sd_hash_queue ) <EOL> d . addCallback ( start_transfer ) <EOL> d . addBoth ( stop ) <EOL> return d <EOL> def test_multiple_uploaders ( self ) : <EOL> sd_hash_queue = Queue ( ) <EOL> num_uploaders = <NUM_LIT:3> <EOL> kill_event = Event ( ) <EOL> dead_events = [ Event ( ) for _ in range ( num_uploaders ) ] <EOL> ready_events = [ Event ( ) for _ in range ( <NUM_LIT:1> , num_uploaders ) ] <EOL> uploader = Process ( target = start_lbry_uploader , args = ( sd_hash_queue , kill_event , dead_events [ <NUM_LIT:0> ] , <EOL> <NUM_LIT> , <NUM_LIT:2> ** <NUM_LIT> ) ) <EOL> uploader . start ( ) <EOL> self . server_processes . append ( uploader ) <EOL> logging . debug ( "<STR_LIT>" ) <EOL> wallet = FakeWallet ( ) <EOL> peer_manager = PeerManager ( ) <EOL> peer_finder = FakePeerFinder ( <NUM_LIT> , peer_manager , num_uploaders ) <EOL> hash_announcer = FakeAnnouncer ( ) <EOL> rate_limiter = DummyRateLimiter ( ) <EOL> sd_identifier = StreamDescriptorIdentifier ( ) <EOL> db_dir = "<STR_LIT>" <EOL> blob_dir = os . path . join ( db_dir , "<STR_LIT>" ) <EOL> os . mkdir ( db_dir ) <EOL> os . mkdir ( blob_dir ) <EOL> self . session = LBRYSession ( MIN_BLOB_DATA_PAYMENT_RATE , db_dir = db_dir , lbryid = "<STR_LIT>" , <EOL> peer_finder = peer_finder , hash_announcer = hash_announcer , <EOL> blob_dir = None , peer_port = <NUM_LIT> , <EOL> use_upnp = False , rate_limiter = rate_limiter , wallet = wallet ) <EOL> self . stream_info_manager = TempLBRYFileMetadataManager ( ) <EOL> self . lbry_file_manager = LBRYFileManager ( self . session , self . stream_info_manager , sd_identifier ) <EOL> def start_additional_uploaders ( sd_hash ) : <EOL> for i in range ( <NUM_LIT:1> , num_uploaders ) : <EOL> uploader = Process ( target = start_lbry_reuploader , <EOL> args = ( sd_hash , kill_event , dead_events [ i ] , ready_events [ i - <NUM_LIT:1> ] , i , <NUM_LIT:2> ** <NUM_LIT:10> ) ) <EOL> uploader . start ( ) <EOL> self . server_processes . append ( uploader ) <EOL> return defer . succeed ( True ) <EOL> def wait_for_ready_events ( ) : <EOL> return defer . DeferredList ( [ self . wait_for_event ( ready_event , <NUM_LIT> ) for ready_event in ready_events ] ) <EOL> def make_downloader ( metadata , prm ) : <EOL> info_validator = metadata . validator <EOL> options = metadata . options <EOL> factories = metadata . factories <EOL> chosen_options = [ o . default_value for o in options . get_downloader_options ( info_validator , prm ) ] <EOL> return factories [ <NUM_LIT:0> ] . make_downloader ( metadata , chosen_options , prm ) <EOL> def download_file ( sd_hash ) : <EOL> prm = PaymentRateManager ( self . session . base_payment_rate_manager ) <EOL> d = download_sd_blob ( self . session , sd_hash , prm ) <EOL> d . addCallback ( sd_identifier . get_metadata_for_sd_blob ) <EOL> d . addCallback ( make_downloader , prm ) <EOL> d . addCallback ( lambda downloader : downloader . start ( ) ) <EOL> return d <EOL> def check_md5_sum ( ) : <EOL> f = open ( '<STR_LIT>' ) <EOL> hashsum = MD5 . new ( ) <EOL> hashsum . update ( f . read ( ) ) <EOL> self . assertEqual ( hashsum . hexdigest ( ) , "<STR_LIT>" ) <EOL> def start_transfer ( sd_hash ) : <EOL> logging . debug ( "<STR_LIT>" ) <EOL> d = start_additional_uploaders ( sd_hash ) <EOL> d . addCallback ( lambda _ : wait_for_ready_events ( ) ) <EOL> d . addCallback ( lambda _ : self . session . setup ( ) ) <EOL> d . addCallback ( lambda _ : add_lbry_file_to_sd_identifier ( sd_identifier ) ) <EOL> d . addCallback ( lambda _ : self . lbry_file_manager . setup ( ) ) <EOL> d . addCallback ( lambda _ : download_file ( sd_hash ) ) <EOL> d . addCallback ( lambda _ : check_md5_sum ( ) ) <EOL> return d <EOL> def stop ( arg ) : <EOL> if isinstance ( arg , Failure ) : <EOL> logging . debug ( "<STR_LIT>" , arg . getTraceback ( ) ) <EOL> else : <EOL> logging . debug ( "<STR_LIT>" ) <EOL> kill_event . set ( ) <EOL> logging . debug ( "<STR_LIT>" ) <EOL> d = defer . DeferredList ( [ self . wait_for_event ( dead_event , <NUM_LIT:15> ) for dead_event in dead_events ] ) <EOL> def print_shutting_down ( ) : <EOL> logging . info ( "<STR_LIT>" ) <EOL> d . addCallback ( lambda _ : print_shutting_down ( ) ) <EOL> d . addCallback ( lambda _ : arg ) <EOL> return d <EOL> d = self . wait_for_hash_from_queue ( sd_hash_queue ) <EOL> d . addCallback ( start_transfer ) <EOL> d . addBoth ( stop ) <EOL> return d <EOL> class TestStreamify ( TestCase ) : <EOL> def setUp ( self ) : <EOL> self . session = None <EOL> self . stream_info_manager = None <EOL> self . lbry_file_manager = None <EOL> self . addCleanup ( self . take_down_env ) <EOL> def take_down_env ( self ) : <EOL> d = defer . succeed ( True ) <EOL> if self . lbry_file_manager is not None : <EOL> d . addCallback ( lambda _ : self . lbry_file_manager . stop ( ) ) <EOL> if self . session is not None : <EOL> d . addCallback ( lambda _ : self . session . shut_down ( ) ) <EOL> if self . stream_info_manager is not None : <EOL> d . addCallback ( lambda _ : self . stream_info_manager . stop ( ) ) <EOL> def delete_test_env ( ) : <EOL> shutil . rmtree ( '<STR_LIT>' ) <EOL> if os . path . exists ( "<STR_LIT>" ) : <EOL> os . remove ( "<STR_LIT>" ) <EOL> d . addCallback ( lambda _ : threads . deferToThread ( delete_test_env ) ) <EOL> return d <EOL> def test_create_stream ( self ) : <EOL> wallet = FakeWallet ( ) <EOL> peer_manager = PeerManager ( ) <EOL> peer_finder = FakePeerFinder ( <NUM_LIT> , peer_manager , <NUM_LIT:2> ) <EOL> hash_announcer = FakeAnnouncer ( ) <EOL> rate_limiter = DummyRateLimiter ( ) <EOL> sd_identifier = StreamDescriptorIdentifier ( ) <EOL> db_dir = "<STR_LIT>" <EOL> blob_dir = os . path . join ( db_dir , "<STR_LIT>" ) <EOL> os . mkdir ( db_dir ) <EOL> os . mkdir ( blob_dir ) <EOL> self . session = LBRYSession ( MIN_BLOB_DATA_PAYMENT_RATE , db_dir = db_dir , lbryid = "<STR_LIT>" , <EOL> peer_finder = peer_finder , hash_announcer = hash_announcer , <EOL> blob_dir = blob_dir , peer_port = <NUM_LIT> , <EOL> use_upnp = False , rate_limiter = rate_limiter , wallet = wallet ) <EOL> self . stream_info_manager = TempLBRYFileMetadataManager ( ) <EOL> self . lbry_file_manager = LBRYFileManager ( self . session , self . stream_info_manager , sd_identifier ) <EOL> d = self . session . setup ( ) <EOL> d . addCallback ( lambda _ : self . stream_info_manager . setup ( ) ) <EOL> d . addCallback ( lambda _ : add_lbry_file_to_sd_identifier ( sd_identifier ) ) <EOL> d . addCallback ( lambda _ : self . lbry_file_manager . setup ( ) ) <EOL> def verify_equal ( sd_info ) : <EOL> self . assertEqual ( sd_info , test_create_stream_sd_file ) <EOL> def verify_stream_descriptor_file ( stream_hash ) : <EOL> d = get_sd_info ( self . lbry_file_manager . stream_info_manager , stream_hash , True ) <EOL> d . addCallback ( verify_equal ) <EOL> return d <EOL> def iv_generator ( ) : <EOL> iv = <NUM_LIT:0> <EOL> while <NUM_LIT:1> : <EOL> iv += <NUM_LIT:1> <EOL> yield "<STR_LIT>" % iv <EOL> def create_stream ( ) : <EOL> test_file = GenFile ( <NUM_LIT> , b'<STR_LIT>' . join ( [ chr ( i + <NUM_LIT:3> ) for i in xrange ( <NUM_LIT:0> , <NUM_LIT:64> , <NUM_LIT:6> ) ] ) ) <EOL> d = create_lbry_file ( self . session , self . lbry_file_manager , "<STR_LIT>" , test_file , <EOL> key = "<STR_LIT>" , iv_generator = iv_generator ( ) ) <EOL> return d <EOL> d . addCallback ( lambda _ : create_stream ( ) ) <EOL> d . addCallback ( verify_stream_descriptor_file ) <EOL> return d <EOL> def test_create_and_combine_stream ( self ) : <EOL> wallet = FakeWallet ( ) <EOL> peer_manager = PeerManager ( ) <EOL> peer_finder = FakePeerFinder ( <NUM_LIT> , peer_manager , <NUM_LIT:2> ) <EOL> hash_announcer = FakeAnnouncer ( ) <EOL> rate_limiter = DummyRateLimiter ( ) <EOL> sd_identifier = StreamDescriptorIdentifier ( ) <EOL> db_dir = "<STR_LIT>" <EOL> blob_dir = os . path . join ( db_dir , "<STR_LIT>" ) <EOL> os . mkdir ( db_dir ) <EOL> os . mkdir ( blob_dir ) <EOL> self . session = LBRYSession ( MIN_BLOB_DATA_PAYMENT_RATE , db_dir = db_dir , lbryid = "<STR_LIT>" , <EOL> peer_finder = peer_finder , hash_announcer = hash_announcer , <EOL> blob_dir = blob_dir , peer_port = <NUM_LIT> , <EOL> use_upnp = False , rate_limiter = rate_limiter , wallet = wallet ) <EOL> self . stream_info_manager = DBLBRYFileMetadataManager ( self . session . db_dir ) <EOL> self . lbry_file_manager = LBRYFileManager ( self . session , self . stream_info_manager , sd_identifier ) <EOL> def start_lbry_file ( lbry_file ) : <EOL> logging . debug ( "<STR_LIT>" ) <EOL> d = lbry_file . start ( ) <EOL> return d <EOL> def combine_stream ( stream_hash ) : <EOL> prm = PaymentRateManager ( self . session . base_payment_rate_manager ) <EOL> d = self . lbry_file_manager . add_lbry_file ( stream_hash , prm ) <EOL> d . addCallback ( start_lbry_file ) <EOL> def check_md5_sum ( ) : <EOL> f = open ( '<STR_LIT>' ) <EOL> hashsum = MD5 . new ( ) <EOL> hashsum . update ( f . read ( ) ) <EOL> self . assertEqual ( hashsum . hexdigest ( ) , "<STR_LIT>" ) <EOL> d . addCallback ( lambda _ : check_md5_sum ( ) ) <EOL> return d <EOL> def create_stream ( ) : <EOL> test_file = GenFile ( <NUM_LIT> , b'<STR_LIT>' . join ( [ chr ( i + <NUM_LIT:5> ) for i in xrange ( <NUM_LIT:0> , <NUM_LIT:64> , <NUM_LIT:6> ) ] ) ) <EOL> return create_lbry_file ( self . session , self . lbry_file_manager , "<STR_LIT>" , test_file , <EOL> suggested_file_name = "<STR_LIT>" ) <EOL> d = self . session . setup ( ) <EOL> d . addCallback ( lambda _ : self . stream_info_manager . setup ( ) ) <EOL> d . addCallback ( lambda _ : add_lbry_file_to_sd_identifier ( sd_identifier ) ) <EOL> d . addCallback ( lambda _ : self . lbry_file_manager . setup ( ) ) <EOL> d . addCallback ( lambda _ : create_stream ( ) ) <EOL> d . addCallback ( combine_stream ) <EOL> return d </s>
