LBRYCommunity/lbry-sdk
1
0
Fork 1
Code Issues Pull requests Projects Releases 282 Packages Wiki Activity

Merge branch 'component-status'

Browse source
This commit is contained in:
Jack Robison 2018-08-03 12:51:49 -04:00
commit 01ee1e1468
No known key found for this signature in database
GPG key ID: DF25C68FE0239BB2
26 changed files with 1100 additions and 1705 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

12
CHANGELOG.md
View file

@ -24,15 +24,20 @@ at anytime.
### Changed
* api server class to use components, and for all JSONRPC API commands to be callable so long as the required components are available.
* return error messages when required conditions on components are not met for API calls
* `status` to no longer return a base58 encoded `lbry_id`, instead return this as the hex encoded `node_id` in a new `dht_node_status` field.
* `status` to no longer return a base58 encoded `lbry_id`, instead return this as the hex encoded `node_id` in a new `dht` field.
* `startup_status` field in the response to `status` to be a dict of component names to status booleans
* renamed the `blockchain_status` field in the response to `status` to `wallet`
* moved and renamed `wallet_is_encrypted` to `is_encrypted` in the `wallet` field in the response to `status`
* moved wallet, upnp and dht startup code from `Session` to `Components`
* attempt blob downloads from http mirror sources (by default) concurrently to p2p sources
* replace miniupnpc with [txupnp](https://github.com/lbryio/txupnp). Since txupnp is still under development, it will internally fall back to miniupnpc.
* simplified test_misc.py in the functional tests
### Added
* `skipped_components` list to the response from `status`
* `skipped_components` config setting, accemapts a list of names of components to not run
* `ComponentManager` for managing the lifecycles of dependencies
* component statuses (`blockchain_headers`, `dht`, `wallet`, `blob_manager` `hash_announcer`, and `file_manager`) to the response to `status`
* `skipped_components` config setting, accepts a list of names of components to not run
* `ComponentManager` for managing the life-cycles of dependencies
* `requires` decorator to register the components required by a `jsonrpc_` command, to facilitate commands registering asynchronously
* unittests for `ComponentManager`
* script to generate docs/api.json file (https://github.com/lbryio/lbry.tech/issues/42)
@ -42,6 +47,7 @@ at anytime.
*
### Removed
* `session_status` argument and response field from `status`
* most of the internal attributes from `Daemon`

12
lbrynet/core/BlobManager.py
View file

@ -1,8 +1,7 @@
import logging
import os
from sqlite3 import IntegrityError
from twisted.internet import threads, defer, task
from lbrynet import conf
from twisted.internet import threads, defer
from lbrynet.blob.blob_file import BlobFile
from lbrynet.blob.creator import BlobFileCreator
@ -26,23 +25,14 @@ class DiskBlobManager(object):
self.blobs = {}
self.blob_hashes_to_delete = {} # {blob_hash: being_deleted (True/False)}
self.check_should_announce_lc = None
# TODO: move this looping call to SQLiteStorage
if 'reflector' not in conf.settings['components_to_skip']:
self.check_should_announce_lc = task.LoopingCall(self.storage.verify_will_announce_all_head_and_sd_blobs)
@defer.inlineCallbacks
def setup(self):
if self.check_should_announce_lc and not self.check_should_announce_lc.running:
self.check_should_announce_lc.start(600)
if self._node_datastore is not None:
raw_blob_hashes = yield self.storage.get_all_finished_blobs()
self._node_datastore.completed_blobs.update(raw_blob_hashes)
defer.returnValue(True)
def stop(self):
if self.check_should_announce_lc and self.check_should_announce_lc.running:
self.check_should_announce_lc.stop()
return defer.succeed(True)
def get_blob(self, blob_hash, length=None):

150
lbrynet/core/Session.py
View file

@ -1,150 +0,0 @@
import logging
from twisted.internet import defer
from lbrynet.core.BlobManager import DiskBlobManager
from lbrynet.database.storage import SQLiteStorage
from lbrynet.core.RateLimiter import RateLimiter
from lbrynet.core.PaymentRateManager import BasePaymentRateManager, OnlyFreePaymentsManager
log = logging.getLogger(__name__)
class Session(object):
"""This class manages all important services common to any application that uses the network.
the hash announcer, which informs other peers that this peer is
associated with some hash. Usually, this means this peer has a
blob identified by the hash in question, but it can be used for
other purposes.
the peer finder, which finds peers that are associated with some
hash.
the blob manager, which keeps track of which blobs have been
downloaded and provides access to them,
the rate limiter, which attempts to ensure download and upload
rates stay below a set maximum
upnp, which opens holes in compatible firewalls so that remote
peers can connect to this peer.
"""
def __init__(self, blob_data_payment_rate, db_dir=None, node_id=None, dht_node_port=None,
known_dht_nodes=None, peer_finder=None, hash_announcer=None, blob_dir=None, blob_manager=None,
peer_port=None, rate_limiter=None, wallet=None, external_ip=None, storage=None,
dht_node=None, peer_manager=None, download_mirrors=None):
"""@param blob_data_payment_rate: The default payment rate for blob data
@param db_dir: The directory in which levelDB files should be stored
@param node_id: The unique ID of this node
@param peer_manager: An object which keeps track of all known
peers. If None, a PeerManager will be created
@param dht_node_port: The port on which the dht node should
listen for incoming connections
@param known_dht_nodes: A list of nodes which the dht node
should use to bootstrap into the dht
@param peer_finder: An object which is used to look up peers
that are associated with some hash. If None, a
DHTPeerFinder will be used, which looks for peers in the
distributed hash table.
@param hash_announcer: An object which announces to other
peers that this peer is associated with some hash. If
None, and peer_port is not None, a DHTHashAnnouncer will
be used. If None and peer_port is None, a
DummyHashAnnouncer will be used, which will not actually
announce anything.
@param blob_dir: The directory in which blobs will be
stored. If None and blob_manager is None, blobs will be
stored in memory only.
@param blob_manager: An object which keeps track of downloaded
blobs and provides access to them. If None, and blob_dir
is not None, a DiskBlobManager will be used, with the
given blob_dir. If None and blob_dir is None, a
TempBlobManager will be used, which stores blobs in memory
only.
@param peer_port: The port on which other peers should connect
to this peer
@param rate_limiter: An object which keeps track of the amount
of data transferred to and from this peer, and can limit
that rate if desired
@param wallet: An object which will be used to keep track of
expected payments and which will pay peers. If None, a
wallet which uses the Point Trader system will be used,
which is meant for testing only
"""
self.db_dir = db_dir
self.node_id = node_id
self.peer_manager = peer_manager
self.peer_finder = peer_finder
self.hash_announcer = hash_announcer
self.dht_node_port = dht_node_port
self.known_dht_nodes = known_dht_nodes
if self.known_dht_nodes is None:
self.known_dht_nodes = []
self.blob_dir = blob_dir
self.blob_manager = blob_manager
self.peer_port = peer_port
self.rate_limiter = rate_limiter
self.external_ip = external_ip
self.upnp_redirects = []
self.wallet = wallet
self.dht_node = dht_node
self.base_payment_rate_manager = BasePaymentRateManager(blob_data_payment_rate)
self.payment_rate_manager = OnlyFreePaymentsManager()
self.storage = storage or SQLiteStorage(self.db_dir)
self.download_mirrors = download_mirrors
def setup(self):
"""Create the blob directory and database if necessary, start all desired services"""
log.debug("Starting session.")
if self.dht_node is not None:
if self.peer_manager is None:
self.peer_manager = self.dht_node.peer_manager
if self.peer_finder is None:
self.peer_finder = self.dht_node.peer_finder
d = self.storage.setup()
d.addCallback(lambda _: self._setup_other_components())
return d
def shut_down(self):
"""Stop all services"""
log.info('Stopping session.')
ds = []
if self.rate_limiter is not None:
ds.append(defer.maybeDeferred(self.rate_limiter.stop))
if self.blob_manager is not None:
ds.append(defer.maybeDeferred(self.blob_manager.stop))
return defer.DeferredList(ds)
def _setup_other_components(self):
log.debug("Setting up the rest of the components")
if self.rate_limiter is None:
self.rate_limiter = RateLimiter()
if self.blob_manager is None:
if self.blob_dir is None:
raise Exception(
"TempBlobManager is no longer supported, specify BlobManager or db_dir")
else:
self.blob_manager = DiskBlobManager(self.blob_dir, self.storage, self.dht_node._dataStore)
self.rate_limiter.start()
d = self.blob_manager.setup()
return d

19
lbrynet/core/StreamDescriptor.py
View file

@ -425,7 +425,8 @@ class EncryptedFileStreamDescriptorValidator(object):
@defer.inlineCallbacks
def download_sd_blob(session, blob_hash, payment_rate_manager, timeout=None):
def download_sd_blob(blob_hash, blob_manager, peer_finder, rate_limiter, payment_rate_manager, wallet, timeout=None,
download_mirrors=None):
"""
Downloads a single blob from the network
@ -439,13 +440,13 @@ def download_sd_blob(session, blob_hash, payment_rate_manager, timeout=None):
"""
downloader = StandaloneBlobDownloader(blob_hash,
session.blob_manager,
session.peer_finder,
session.rate_limiter,
blob_manager,
peer_finder,
rate_limiter,
payment_rate_manager,
session.wallet,
wallet,
timeout)
mirror = HTTPBlobDownloader(session.blob_manager, [blob_hash], session.download_mirrors)
mirror = HTTPBlobDownloader(blob_manager, [blob_hash], download_mirrors or [])
mirror.start()
sd_blob = yield downloader.download()
mirror.stop()
@ -454,9 +455,9 @@ def download_sd_blob(session, blob_hash, payment_rate_manager, timeout=None):
try:
validate_descriptor(sd_info)
except InvalidStreamDescriptorError as err:
yield session.blob_manager.delete_blobs([blob_hash])
yield blob_manager.delete_blobs([blob_hash])
raise err
raw_sd = yield sd_reader._get_raw_data()
yield session.blob_manager.storage.add_known_blob(blob_hash, len(raw_sd))
yield save_sd_info(session.blob_manager, sd_blob.blob_hash, sd_info)
yield blob_manager.storage.add_known_blob(blob_hash, len(raw_sd))
yield save_sd_info(blob_manager, sd_blob.blob_hash, sd_info)
defer.returnValue(sd_blob)

104
lbrynet/core/Wallet.py
View file

@ -1,30 +1,25 @@
import os
from collections import defaultdict, deque
import datetime
import logging
from decimal import Decimal
import treq
from zope.interface import implements
from twisted.internet import threads, reactor, defer, task
from twisted.python.failure import Failure
from twisted.internet.error import ConnectionAborted
from hashlib import sha256
from lbryum import wallet as lbryum_wallet
from lbryum.network import Network
from lbryum.simple_config import SimpleConfig
from lbryum.constants import COIN
from lbryum.commands import Commands
from lbryum.errors import InvalidPassword
from lbryum.constants import HEADERS_URL, HEADER_SIZE
from lbryschema.uri import parse_lbry_uri
from lbryschema.claim import ClaimDict
from lbryschema.error import DecodeError
from lbryschema.decode import smart_decode
from lbrynet.txlbryum.factory import StratumClient
from lbrynet.interfaces import IRequestCreator, IQueryHandlerFactory, IQueryHandler, IWallet
from lbrynet.core.utils import DeferredDict
from lbrynet.core.client.ClientRequest import ClientRequest
@ -92,107 +87,8 @@ class Wallet(object):
self._batch_count = 20
self._pending_claim_checker = task.LoopingCall(self.fetch_and_save_heights_for_pending_claims)
@defer.inlineCallbacks
def fetch_headers_from_s3(self):
local_header_size = self.local_header_file_size()
resume_header = {"Range": "bytes={}-".format(local_header_size)}
response = yield treq.get(HEADERS_URL, headers=resume_header)
got_406 = response.code == 406 # our file is bigger
final_size_after_download = response.length + local_header_size
if got_406:
log.warning("s3 is more out of date than we are")
# should have something to download and a final length divisible by the header size
elif final_size_after_download and not final_size_after_download % HEADER_SIZE:
s3_height = (final_size_after_download / HEADER_SIZE) - 1
local_height = self.local_header_file_height()
if s3_height > local_height:
if local_header_size:
log.info("Resuming download of %i bytes from s3", response.length)
with open(os.path.join(self.config.path, "blockchain_headers"), "a+b") as headers_file:
yield treq.collect(response, headers_file.write)
else:
with open(os.path.join(self.config.path, "blockchain_headers"), "wb") as headers_file:
yield treq.collect(response, headers_file.write)
log.info("fetched headers from s3 (s3 height: %i), now verifying integrity after download.", s3_height)
self._check_header_file_integrity()
else:
log.warning("s3 is more out of date than we are")
else:
log.error("invalid size for headers from s3")
def local_header_file_height(self):
return max((self.local_header_file_size() / HEADER_SIZE) - 1, 0)
def local_header_file_size(self):
headers_path = os.path.join(self.config.path, "blockchain_headers")
if os.path.isfile(headers_path):
return os.stat(headers_path).st_size
return 0
@defer.inlineCallbacks
def get_remote_height(self, server, port):
connected = defer.Deferred()
connected.addTimeout(3, reactor, lambda *_: None)
client = StratumClient(connected)
reactor.connectTCP(server, port, client)
yield connected
remote_height = yield client.blockchain_block_get_server_height()
client.client.transport.loseConnection()
defer.returnValue(remote_height)
@defer.inlineCallbacks
def should_download_headers_from_s3(self):
from lbrynet import conf
if conf.settings['blockchain_name'] != "lbrycrd_main":
defer.returnValue(False)
self._check_header_file_integrity()
s3_headers_depth = conf.settings['s3_headers_depth']
if not s3_headers_depth:
defer.returnValue(False)
local_height = self.local_header_file_height()
for server_url in self.config.get('default_servers'):
port = int(self.config.get('default_servers')[server_url]['t'])
try:
remote_height = yield self.get_remote_height(server_url, port)
log.info("%s:%i height: %i, local height: %s", server_url, port, remote_height, local_height)
if remote_height > (local_height + s3_headers_depth):
defer.returnValue(True)
except Exception as err:
log.warning("error requesting remote height from %s:%i - %s", server_url, port, err)
defer.returnValue(False)
def _check_header_file_integrity(self):
# TODO: temporary workaround for usability. move to txlbryum and check headers instead of file integrity
from lbrynet import conf
if conf.settings['blockchain_name'] != "lbrycrd_main":
return
hashsum = sha256()
checksum_height, checksum = conf.settings['HEADERS_FILE_SHA256_CHECKSUM']
checksum_length_in_bytes = checksum_height * HEADER_SIZE
if self.local_header_file_size() < checksum_length_in_bytes:
return
headers_path = os.path.join(self.config.path, "blockchain_headers")
with open(headers_path, "rb") as headers_file:
hashsum.update(headers_file.read(checksum_length_in_bytes))
current_checksum = hashsum.hexdigest()
if current_checksum != checksum:
msg = "Expected checksum {}, got {}".format(checksum, current_checksum)
log.warning("Wallet file corrupted, checksum mismatch. " + msg)
log.warning("Deleting header file so it can be downloaded again.")
os.unlink(headers_path)
elif (self.local_header_file_size() % HEADER_SIZE) != 0:
log.warning("Header file is good up to checkpoint height, but incomplete. Truncating to checkpoint.")
with open(headers_path, "rb+") as headers_file:
headers_file.truncate(checksum_length_in_bytes)
@defer.inlineCallbacks
def start(self):
should_download_headers = yield self.should_download_headers_from_s3()
if should_download_headers:
try:
yield self.fetch_headers_from_s3()
except Exception as err:
log.error("failed to fetch headers from s3: %s", err)
log.info("Starting wallet.")
yield self._start()
self.stopped = False

1
lbrynet/core/log_support.py
View file

@ -89,6 +89,7 @@ def disable_third_party_loggers():
logging.getLogger('BitcoinRPC').setLevel(logging.INFO)
logging.getLogger('lbryum').setLevel(logging.WARNING)
logging.getLogger('twisted').setLevel(logging.CRITICAL)
logging.getLogger('txupnp').setLevel(logging.WARNING)
@_log_decorator

3
lbrynet/daemon/Component.py
View file

@ -37,6 +37,9 @@ class Component(object):
def running(self):
return self._running
def get_status(self):
return
def start(self):
raise NotImplementedError()

503
lbrynet/daemon/Components.py
View file

@ -1,10 +1,18 @@
import os
import logging
import miniupnpc
from hashlib import sha256
import treq
import math
import binascii
from twisted.internet import defer, threads, reactor, error
from txupnp.upnp import UPnP
from lbryum.simple_config import SimpleConfig
from lbryum.constants import HEADERS_URL, HEADER_SIZE
from lbrynet import conf
from lbrynet.core.Session import Session
from lbrynet.core.utils import DeferredDict
from lbrynet.core.PaymentRateManager import OnlyFreePaymentsManager
from lbrynet.core.RateLimiter import RateLimiter
from lbrynet.core.BlobManager import DiskBlobManager
from lbrynet.core.StreamDescriptor import StreamDescriptorIdentifier, EncryptedFileStreamType
from lbrynet.core.Wallet import LBRYumWallet
from lbrynet.core.server.BlobRequestHandler import BlobRequestHandlerFactory
@ -17,7 +25,7 @@ from lbrynet.file_manager.EncryptedFileManager import EncryptedFileManager
from lbrynet.lbry_file.client.EncryptedFileDownloader import EncryptedFileSaverFactory
from lbrynet.lbry_file.client.EncryptedFileOptions import add_lbry_file_to_sd_identifier
from lbrynet.reflector import ServerFactory as reflector_server_factory
from lbrynet.txlbryum.factory import StratumClient
from lbrynet.core.utils import generate_id
log = logging.getLogger(__name__)
@ -25,8 +33,9 @@ log = logging.getLogger(__name__)
# settings must be initialized before this file is imported
DATABASE_COMPONENT = "database"
BLOB_COMPONENT = "blob_manager"
HEADERS_COMPONENT = "blockchain_headers"
WALLET_COMPONENT = "wallet"
SESSION_COMPONENT = "session"
DHT_COMPONENT = "dht"
HASH_ANNOUNCER_COMPONENT = "hash_announcer"
STREAM_IDENTIFIER_COMPONENT = "stream_identifier"
@ -35,6 +44,28 @@ PEER_PROTOCOL_SERVER_COMPONENT = "peer_protocol_server"
REFLECTOR_COMPONENT = "reflector"
UPNP_COMPONENT = "upnp"
EXCHANGE_RATE_MANAGER_COMPONENT = "exchange_rate_manager"
RATE_LIMITER_COMPONENT = "rate_limiter"
PAYMENT_RATE_COMPONENT = "payment_rate_manager"
def get_wallet_config():
wallet_type = GCS('wallet')
if wallet_type == conf.LBRYCRD_WALLET:
raise ValueError('LBRYcrd Wallet is no longer supported')
elif wallet_type != conf.LBRYUM_WALLET:
raise ValueError('Wallet Type {} is not valid'.format(wallet_type))
lbryum_servers = {address: {'t': str(port)}
for address, port in GCS('lbryum_servers')}
config = {
'auto_connect': True,
'chain': GCS('blockchain_name'),
'default_servers': lbryum_servers
}
if 'use_keyring' in conf.settings:
config['use_keyring'] = GCS('use_keyring')
if conf.settings['lbryum_wallet_dir']:
config['lbryum_path'] = GCS('lbryum_wallet_dir')
return config
class ConfigSettings(object):
@ -138,9 +169,138 @@ class DatabaseComponent(Component):
self.storage = None
class HeadersComponent(Component):
component_name = HEADERS_COMPONENT
def __init__(self, component_manager):
Component.__init__(self, component_manager)
self.config = SimpleConfig(get_wallet_config())
self._downloading_headers = None
self._headers_progress_percent = None
@property
def component(self):
return self
def get_status(self):
return {} if not self._downloading_headers else {
'downloading_headers': self._downloading_headers,
'download_progress': self._headers_progress_percent
}
@defer.inlineCallbacks
def fetch_headers_from_s3(self):
def collector(data, h_file):
h_file.write(data)
local_size = float(h_file.tell())
final_size = float(final_size_after_download)
self._headers_progress_percent = math.ceil(local_size / final_size * 100)
local_header_size = self.local_header_file_size()
resume_header = {"Range": "bytes={}-".format(local_header_size)}
response = yield treq.get(HEADERS_URL, headers=resume_header)
got_406 = response.code == 406 # our file is bigger
final_size_after_download = response.length + local_header_size
if got_406:
log.warning("s3 is more out of date than we are")
# should have something to download and a final length divisible by the header size
elif final_size_after_download and not final_size_after_download % HEADER_SIZE:
s3_height = (final_size_after_download / HEADER_SIZE) - 1
local_height = self.local_header_file_height()
if s3_height > local_height:
if local_header_size:
log.info("Resuming download of %i bytes from s3", response.length)
with open(os.path.join(self.config.path, "blockchain_headers"), "a+b") as headers_file:
yield treq.collect(response, lambda d: collector(d, headers_file))
else:
with open(os.path.join(self.config.path, "blockchain_headers"), "wb") as headers_file:
yield treq.collect(response, lambda d: collector(d, headers_file))
log.info("fetched headers from s3 (s3 height: %i), now verifying integrity after download.", s3_height)
self._check_header_file_integrity()
else:
log.warning("s3 is more out of date than we are")
else:
log.error("invalid size for headers from s3")
def local_header_file_height(self):
return max((self.local_header_file_size() / HEADER_SIZE) - 1, 0)
def local_header_file_size(self):
headers_path = os.path.join(self.config.path, "blockchain_headers")
if os.path.isfile(headers_path):
return os.stat(headers_path).st_size
return 0
@defer.inlineCallbacks
def get_remote_height(self, server, port):
connected = defer.Deferred()
connected.addTimeout(3, reactor, lambda *_: None)
client = StratumClient(connected)
reactor.connectTCP(server, port, client)
yield connected
remote_height = yield client.blockchain_block_get_server_height()
client.client.transport.loseConnection()
defer.returnValue(remote_height)
@defer.inlineCallbacks
def should_download_headers_from_s3(self):
if conf.settings['blockchain_name'] != "lbrycrd_main":
defer.returnValue(False)
self._check_header_file_integrity()
s3_headers_depth = conf.settings['s3_headers_depth']
if not s3_headers_depth:
defer.returnValue(False)
local_height = self.local_header_file_height()
for server_url in self.config.get('default_servers'):
port = int(self.config.get('default_servers')[server_url]['t'])
try:
remote_height = yield self.get_remote_height(server_url, port)
log.info("%s:%i height: %i, local height: %s", server_url, port, remote_height, local_height)
if remote_height > (local_height + s3_headers_depth):
defer.returnValue(True)
except Exception as err:
log.warning("error requesting remote height from %s:%i - %s", server_url, port, err)
defer.returnValue(False)
def _check_header_file_integrity(self):
# TODO: temporary workaround for usability. move to txlbryum and check headers instead of file integrity
if conf.settings['blockchain_name'] != "lbrycrd_main":
return
hashsum = sha256()
checksum_height, checksum = conf.settings['HEADERS_FILE_SHA256_CHECKSUM']
checksum_length_in_bytes = checksum_height * HEADER_SIZE
if self.local_header_file_size() < checksum_length_in_bytes:
return
headers_path = os.path.join(self.config.path, "blockchain_headers")
with open(headers_path, "rb") as headers_file:
hashsum.update(headers_file.read(checksum_length_in_bytes))
current_checksum = hashsum.hexdigest()
if current_checksum != checksum:
msg = "Expected checksum {}, got {}".format(checksum, current_checksum)
log.warning("Wallet file corrupted, checksum mismatch. " + msg)
log.warning("Deleting header file so it can be downloaded again.")
os.unlink(headers_path)
elif (self.local_header_file_size() % HEADER_SIZE) != 0:
log.warning("Header file is good up to checkpoint height, but incomplete. Truncating to checkpoint.")
with open(headers_path, "rb+") as headers_file:
headers_file.truncate(checksum_length_in_bytes)
@defer.inlineCallbacks
def start(self):
self._downloading_headers = yield self.should_download_headers_from_s3()
if self._downloading_headers:
try:
yield self.fetch_headers_from_s3()
except Exception as err:
log.error("failed to fetch headers from s3: %s", err)
def stop(self):
return defer.succeed(None)
class WalletComponent(Component):
component_name = WALLET_COMPONENT
depends_on = [DATABASE_COMPONENT]
depends_on = [DATABASE_COMPONENT, HEADERS_COMPONENT]
def __init__(self, component_manager):
Component.__init__(self, component_manager)
@ -150,34 +310,25 @@ class WalletComponent(Component):
def component(self):
return self.wallet
@defer.inlineCallbacks
def get_status(self):
if self.wallet:
local_height = self.wallet.network.get_local_height()
remote_height = self.wallet.network.get_server_height()
best_hash = yield self.wallet.get_best_blockhash()
defer.returnValue({
'blocks': local_height,
'blocks_behind': remote_height - local_height,
'best_blockhash': best_hash,
'is_encrypted': self.wallet.wallet.use_encryption
})
@defer.inlineCallbacks
def start(self):
storage = self.component_manager.get_component(DATABASE_COMPONENT)
wallet_type = GCS('wallet')
if wallet_type == conf.LBRYCRD_WALLET:
raise ValueError('LBRYcrd Wallet is no longer supported')
elif wallet_type == conf.LBRYUM_WALLET:
log.info("Using lbryum wallet")
lbryum_servers = {address: {'t': str(port)}
for address, port in GCS('lbryum_servers')}
config = {
'auto_connect': True,
'chain': GCS('blockchain_name'),
'default_servers': lbryum_servers
}
if 'use_keyring' in conf.settings:
config['use_keyring'] = GCS('use_keyring')
if conf.settings['lbryum_wallet_dir']:
config['lbryum_path'] = GCS('lbryum_wallet_dir')
config = get_wallet_config()
self.wallet = LBRYumWallet(storage, config)
yield self.wallet.start()
else:
raise ValueError('Wallet Type {} is not valid'.format(wallet_type))
@defer.inlineCallbacks
def stop(self):
@ -185,40 +336,35 @@ class WalletComponent(Component):
self.wallet = None
class SessionComponent(Component):
component_name = SESSION_COMPONENT
depends_on = [DATABASE_COMPONENT, WALLET_COMPONENT, DHT_COMPONENT, HASH_ANNOUNCER_COMPONENT]
class BlobComponent(Component):
component_name = BLOB_COMPONENT
depends_on = [DATABASE_COMPONENT, DHT_COMPONENT]
def __init__(self, component_manager):
Component.__init__(self, component_manager)
self.session = None
self.blob_manager = None
@property
def component(self):
return self.session
return self.blob_manager
@defer.inlineCallbacks
def start(self):
self.session = Session(
GCS('data_rate'),
db_dir=GCS('data_dir'),
node_id=CS.get_node_id(),
blob_dir=CS.get_blobfiles_dir(),
dht_node=self.component_manager.get_component(DHT_COMPONENT),
hash_announcer=self.component_manager.get_component(HASH_ANNOUNCER_COMPONENT),
dht_node_port=GCS('dht_node_port'),
known_dht_nodes=GCS('known_dht_nodes'),
peer_port=GCS('peer_port'),
wallet=self.component_manager.get_component(WALLET_COMPONENT),
external_ip=CS.get_external_ip(),
storage=self.component_manager.get_component(DATABASE_COMPONENT),
download_mirrors=GCS('download_mirrors')
)
yield self.session.setup()
storage = self.component_manager.get_component(DATABASE_COMPONENT)
dht_node = self.component_manager.get_component(DHT_COMPONENT)
self.blob_manager = DiskBlobManager(CS.get_blobfiles_dir(), storage, dht_node._dataStore)
return self.blob_manager.setup()
def stop(self):
return self.blob_manager.stop()
@defer.inlineCallbacks
def stop(self):
yield self.session.shut_down()
def get_status(self):
count = 0
if self.blob_manager:
count = yield self.blob_manager.storage.count_finished_blobs()
defer.returnValue({
'finished_blobs': count
})
class DHTComponent(Component):
@ -229,12 +375,19 @@ class DHTComponent(Component):
Component.__init__(self, component_manager)
self.dht_node = None
self.upnp_component = None
self.udp_port, self.peer_port = None, None
self.udp_port = None
self.peer_port = None
@property
def component(self):
return self.dht_node
def get_status(self):
return {
'node_id': binascii.hexlify(CS.get_node_id()),
'peers_in_routing_table': 0 if not self.dht_node else len(self.dht_node.contacts)
}
@defer.inlineCallbacks
def start(self):
self.upnp_component = self.component_manager.get_component(UPNP_COMPONENT)
@ -285,10 +438,35 @@ class HashAnnouncerComponent(Component):
def stop(self):
yield self.hash_announcer.stop()
def get_status(self):
return {
'announce_queue_size': 0 if not self.hash_announcer else len(self.hash_announcer.hash_queue)
}
class RateLimiterComponent(Component):
component_name = RATE_LIMITER_COMPONENT
def __init__(self, component_manager):
Component.__init__(self, component_manager)
self.rate_limiter = RateLimiter()
@property
def component(self):
return self.rate_limiter
def start(self):
self.rate_limiter.start()
return defer.succeed(None)
def stop(self):
self.rate_limiter.stop()
return defer.succeed(None)
class StreamIdentifierComponent(Component):
component_name = STREAM_IDENTIFIER_COMPONENT
depends_on = [SESSION_COMPONENT]
depends_on = [DHT_COMPONENT, RATE_LIMITER_COMPONENT, BLOB_COMPONENT, DATABASE_COMPONENT, WALLET_COMPONENT]
def __init__(self, component_manager):
Component.__init__(self, component_manager)
@ -300,14 +478,19 @@ class StreamIdentifierComponent(Component):
@defer.inlineCallbacks
def start(self):
session = self.component_manager.get_component(SESSION_COMPONENT)
dht_node = self.component_manager.get_component(DHT_COMPONENT)
rate_limiter = self.component_manager.get_component(RATE_LIMITER_COMPONENT)
blob_manager = self.component_manager.get_component(BLOB_COMPONENT)
storage = self.component_manager.get_component(DATABASE_COMPONENT)
wallet = self.component_manager.get_component(WALLET_COMPONENT)
add_lbry_file_to_sd_identifier(self.sd_identifier)
file_saver_factory = EncryptedFileSaverFactory(
session.peer_finder,
session.rate_limiter,
session.blob_manager,
session.storage,
session.wallet,
dht_node.peer_finder,
rate_limiter,
blob_manager,
storage,
wallet,
GCS('download_directory')
)
yield self.sd_identifier.add_stream_downloader_factory(EncryptedFileStreamType, file_saver_factory)
@ -316,9 +499,28 @@ class StreamIdentifierComponent(Component):
pass
class PaymentRateComponent(Component):
component_name = PAYMENT_RATE_COMPONENT
def __init__(self, component_manager):
Component.__init__(self, component_manager)
self.payment_rate_manager = OnlyFreePaymentsManager()
@property
def component(self):
return self.payment_rate_manager
def start(self):
return defer.succeed(None)
def stop(self):
return defer.succeed(None)
class FileManagerComponent(Component):
component_name = FILE_MANAGER_COMPONENT
depends_on = [SESSION_COMPONENT, STREAM_IDENTIFIER_COMPONENT]
depends_on = [DHT_COMPONENT, RATE_LIMITER_COMPONENT, BLOB_COMPONENT, DATABASE_COMPONENT, WALLET_COMPONENT,
STREAM_IDENTIFIER_COMPONENT, PAYMENT_RATE_COMPONENT]
def __init__(self, component_manager):
Component.__init__(self, component_manager)
@ -328,12 +530,25 @@ class FileManagerComponent(Component):
def component(self):
return self.file_manager
def get_status(self):
if not self.file_manager:
return
return {
'managed_files': len(self.file_manager.lbry_files)
}
@defer.inlineCallbacks
def start(self):
session = self.component_manager.get_component(SESSION_COMPONENT)
dht_node = self.component_manager.get_component(DHT_COMPONENT)
rate_limiter = self.component_manager.get_component(RATE_LIMITER_COMPONENT)
blob_manager = self.component_manager.get_component(BLOB_COMPONENT)
storage = self.component_manager.get_component(DATABASE_COMPONENT)
wallet = self.component_manager.get_component(WALLET_COMPONENT)
sd_identifier = self.component_manager.get_component(STREAM_IDENTIFIER_COMPONENT)
payment_rate_manager = self.component_manager.get_component(PAYMENT_RATE_COMPONENT)
log.info('Starting the file manager')
self.file_manager = EncryptedFileManager(session, sd_identifier)
self.file_manager = EncryptedFileManager(dht_node.peer_finder, rate_limiter, blob_manager, wallet,
payment_rate_manager, storage, sd_identifier)
yield self.file_manager.setup()
log.info('Done setting up file manager')
@ -344,7 +559,8 @@ class FileManagerComponent(Component):
class PeerProtocolServerComponent(Component):
component_name = PEER_PROTOCOL_SERVER_COMPONENT
depends_on = [SESSION_COMPONENT, UPNP_COMPONENT]
depends_on = [UPNP_COMPONENT, DHT_COMPONENT, RATE_LIMITER_COMPONENT, BLOB_COMPONENT, WALLET_COMPONENT,
PAYMENT_RATE_COMPONENT]
def __init__(self, component_manager):
Component.__init__(self, component_manager)
@ -356,27 +572,23 @@ class PeerProtocolServerComponent(Component):
@defer.inlineCallbacks
def start(self):
query_handlers = {}
upnp_component = self.component_manager.get_component(UPNP_COMPONENT)
peer_port, udp_port = upnp_component.get_redirects()
session = self.component_manager.get_component(SESSION_COMPONENT)
handlers = [
wallet = self.component_manager.get_component(WALLET_COMPONENT)
peer_port = self.component_manager.get_component(UPNP_COMPONENT).upnp_redirects["TCP"]
query_handlers = {
handler.get_primary_query_identifier(): handler for handler in [
BlobRequestHandlerFactory(
session.blob_manager,
session.wallet,
session.payment_rate_manager,
self.component_manager.get_component(BLOB_COMPONENT),
wallet,
self.component_manager.get_component(PAYMENT_RATE_COMPONENT),
self.component_manager.analytics_manager
),
session.wallet.get_wallet_info_query_handler_factory(),
wallet.get_wallet_info_query_handler_factory(),
]
for handler in handlers:
query_id = handler.get_primary_query_identifier()
query_handlers[query_id] = handler
if peer_port is not None:
server_factory = ServerProtocolFactory(session.rate_limiter, query_handlers, session.peer_manager)
}
server_factory = ServerProtocolFactory(
self.component_manager.get_component(RATE_LIMITER_COMPONENT), query_handlers,
self.component_manager.get_component(DHT_COMPONENT).peer_manager
)
try:
log.info("Peer protocol listening on TCP %d", peer_port)
@ -398,7 +610,7 @@ class PeerProtocolServerComponent(Component):
class ReflectorComponent(Component):
component_name = REFLECTOR_COMPONENT
depends_on = [SESSION_COMPONENT, FILE_MANAGER_COMPONENT]
depends_on = [DHT_COMPONENT, BLOB_COMPONENT, FILE_MANAGER_COMPONENT]
def __init__(self, component_manager):
Component.__init__(self, component_manager)
@ -412,11 +624,10 @@ class ReflectorComponent(Component):
@defer.inlineCallbacks
def start(self):
log.info("Starting reflector server")
session = self.component_manager.get_component(SESSION_COMPONENT)
dht_node = self.component_manager.get_component(DHT_COMPONENT)
blob_manager = self.component_manager.get_component(BLOB_COMPONENT)
file_manager = self.component_manager.get_component(FILE_MANAGER_COMPONENT)
reflector_factory = reflector_server_factory(session.peer_manager, session.blob_manager, file_manager)
reflector_factory = reflector_server_factory(dht_node.peer_manager, blob_manager, file_manager)
try:
self.reflector_server = yield reactor.listenTCP(self.reflector_server_port, reflector_factory)
log.info('Started reflector on port %s', self.reflector_server_port)
@ -437,100 +648,54 @@ class UPnPComponent(Component):
def __init__(self, component_manager):
Component.__init__(self, component_manager)
self.peer_port = GCS('peer_port')
self.dht_node_port = GCS('dht_node_port')
self._default_peer_port = GCS('peer_port')
self._default_dht_node_port = GCS('dht_node_port')
self.use_upnp = GCS('use_upnp')
self.external_ip = CS.get_external_ip()
self.upnp_redirects = []
self.external_ip = None
self.upnp = UPnP(self.component_manager.reactor, try_miniupnpc_fallback=True)
self.upnp_redirects = {}
@property
def component(self):
return self
def get_redirects(self):
return self.peer_port, self.dht_node_port
if not self.use_upnp or not self.upnp_redirects:
return self._default_peer_port, self._default_dht_node_port
return self.upnp_redirects["TCP"], self.upnp_redirects["UDP"]
@defer.inlineCallbacks
def _setup_redirects(self):
self.external_ip = yield self.upnp.get_external_ip()
upnp_redirects = yield DeferredDict({
"UDP": self.upnp.get_next_mapping(self._default_dht_node_port, "UDP", "LBRY DHT port"),
"TCP": self.upnp.get_next_mapping(self._default_peer_port, "TCP", "LBRY peer port")
})
self.upnp_redirects.update(upnp_redirects)
@defer.inlineCallbacks
def start(self):
log.debug("In _try_upnp")
def get_free_port(upnp, port, protocol):
# returns an existing mapping if it exists
mapping = upnp.getspecificportmapping(port, protocol)
if not mapping:
return port
if upnp.lanaddr == mapping[0]:
return mapping[1]
return get_free_port(upnp, port + 1, protocol)
def get_port_mapping(upnp, port, protocol, description):
# try to map to the requested port, if there is already a mapping use the next external
# port available
if protocol not in ['UDP', 'TCP']:
raise Exception("invalid protocol")
port = get_free_port(upnp, port, protocol)
if isinstance(port, tuple):
log.info("Found existing UPnP redirect %s:%i (%s) to %s:%i, using it",
self.external_ip, port, protocol, upnp.lanaddr, port)
return port
upnp.addportmapping(port, protocol, upnp.lanaddr, port,
description, '')
log.info("Set UPnP redirect %s:%i (%s) to %s:%i", self.external_ip, port,
protocol, upnp.lanaddr, port)
return port
def threaded_try_upnp():
if self.use_upnp is False:
log.debug("Not using upnp")
return False
u = miniupnpc.UPnP()
num_devices_found = u.discover()
if num_devices_found > 0:
u.selectigd()
external_ip = u.externalipaddress()
if external_ip != '0.0.0.0' and not self.external_ip:
# best not to rely on this external ip, the router can be behind layers of NATs
self.external_ip = external_ip
if self.peer_port:
self.peer_port = get_port_mapping(u, self.peer_port, 'TCP', 'LBRY peer port')
self.upnp_redirects.append((self.peer_port, 'TCP'))
if self.dht_node_port:
self.dht_node_port = get_port_mapping(u, self.dht_node_port, 'UDP', 'LBRY DHT port')
self.upnp_redirects.append((self.dht_node_port, 'UDP'))
return True
return False
def upnp_failed(err):
log.warning("UPnP failed. Reason: %s", err.getErrorMessage())
return False
d = threads.deferToThread(threaded_try_upnp)
d.addErrback(upnp_failed)
return d
found = yield self.upnp.discover()
if found and not self.upnp.miniupnpc_runner:
log.info("set up redirects using txupnp")
elif found and self.upnp.miniupnpc_runner:
log.warning("failed to set up redirect with txupnp, miniupnpc fallback was successful")
if found:
try:
yield self._setup_redirects()
except Exception as err:
if not self.upnp.miniupnpc_runner:
started_fallback = yield self.upnp.start_miniupnpc_fallback()
if started_fallback:
yield self._setup_redirects()
else:
log.warning("failed to set up upnp redirects")
def stop(self):
log.info("Unsetting upnp for session")
def threaded_unset_upnp():
if self.use_upnp is False:
log.debug("Not using upnp")
return False
u = miniupnpc.UPnP()
num_devices_found = u.discover()
if num_devices_found > 0:
u.selectigd()
for port, protocol in self.upnp_redirects:
if u.getspecificportmapping(port, protocol) is None:
log.warning(
"UPnP redirect for %s %d was removed by something else.",
protocol, port)
else:
u.deleteportmapping(port, protocol)
log.info("Removed UPnP redirect for %s %d.", protocol, port)
self.upnp_redirects = []
d = threads.deferToThread(threaded_unset_upnp)
d.addErrback(lambda err: str(err))
return d
return defer.DeferredList(
[self.upnp.delete_port_mapping(port, protocol) for protocol, port in self.upnp_redirects.items()]
)
class ExchangeRateManagerComponent(Component):

191
lbrynet/daemon/Daemon.py
View file

@ -25,14 +25,13 @@ from lbryschema.decode import smart_decode
from lbrynet.core.system_info import get_lbrynet_version
from lbrynet import conf
from lbrynet.reflector import reupload
from lbrynet.daemon.Components import WALLET_COMPONENT, DATABASE_COMPONENT, SESSION_COMPONENT, DHT_COMPONENT
from lbrynet.daemon.Components import STREAM_IDENTIFIER_COMPONENT, FILE_MANAGER_COMPONENT
from lbrynet.daemon.Components import EXCHANGE_RATE_MANAGER_COMPONENT
from lbrynet.daemon.Components import WALLET_COMPONENT, DATABASE_COMPONENT, DHT_COMPONENT, BLOB_COMPONENT
from lbrynet.daemon.Components import STREAM_IDENTIFIER_COMPONENT, FILE_MANAGER_COMPONENT, RATE_LIMITER_COMPONENT
from lbrynet.daemon.Components import EXCHANGE_RATE_MANAGER_COMPONENT, PAYMENT_RATE_COMPONENT, UPNP_COMPONENT
from lbrynet.daemon.ComponentManager import RequiredCondition
from lbrynet.daemon.Downloader import GetStream
from lbrynet.daemon.Publisher import Publisher
from lbrynet.daemon.auth.server import AuthJSONRPCServer
from lbrynet.core.PaymentRateManager import OnlyFreePaymentsManager
from lbrynet.core import utils, system_info
from lbrynet.core.StreamDescriptor import download_sd_blob
from lbrynet.core.Error import InsufficientFundsError, UnknownNameError
@ -91,7 +90,7 @@ class IterableContainer(object):
class Checker(object):
"""The looping calls the daemon runs"""
INTERNET_CONNECTION = 'internet_connection_checker', 3600
INTERNET_CONNECTION = 'internet_connection_checker', 300
# CONNECTION_STATUS = 'connection_status_checker'
@ -186,13 +185,16 @@ class Daemon(AuthJSONRPCServer):
"""
component_attributes = {
EXCHANGE_RATE_MANAGER_COMPONENT: "exchange_rate_manager",
DATABASE_COMPONENT: "storage",
SESSION_COMPONENT: "session",
WALLET_COMPONENT: "wallet",
DHT_COMPONENT: "dht_node",
WALLET_COMPONENT: "wallet",
STREAM_IDENTIFIER_COMPONENT: "sd_identifier",
FILE_MANAGER_COMPONENT: "file_manager",
EXCHANGE_RATE_MANAGER_COMPONENT: "exchange_rate_manager",
PAYMENT_RATE_COMPONENT: "payment_rate_manager",
RATE_LIMITER_COMPONENT: "rate_limiter",
BLOB_COMPONENT: "blob_manager",
UPNP_COMPONENT: "upnp"
}
def __init__(self, analytics_manager=None, component_manager=None):
@ -218,9 +220,12 @@ class Daemon(AuthJSONRPCServer):
self.dht_node = None
self.wallet = None
self.sd_identifier = None
self.session = None
self.file_manager = None
self.exchange_rate_manager = None
self.payment_rate_manager = None
self.rate_limiter = None
self.blob_manager = None
self.upnp = None
# TODO: delete this
self.streams = {}
@ -254,10 +259,10 @@ class Daemon(AuthJSONRPCServer):
if not blob_hash:
raise Exception("Nothing to download")
rate_manager = rate_manager or self.session.payment_rate_manager
rate_manager = rate_manager or self.payment_rate_manager
timeout = timeout or 30
downloader = StandaloneBlobDownloader(
blob_hash, self.session.blob_manager, self.session.peer_finder, self.session.rate_limiter,
blob_hash, self.blob_manager, self.dht_node.peer_finder, self.rate_limiter,
rate_manager, self.wallet, timeout
)
return downloader.download()
@ -275,7 +280,7 @@ class Daemon(AuthJSONRPCServer):
}
blobs = {}
try:
sd_host = yield self.session.blob_manager.get_host_downloaded_from(sd_hash)
sd_host = yield self.blob_manager.get_host_downloaded_from(sd_hash)
except Exception:
sd_host = None
report["sd_blob"] = sd_host
@ -320,11 +325,12 @@ class Daemon(AuthJSONRPCServer):
else:
download_id = utils.random_string()
self.analytics_manager.send_download_started(download_id, name, claim_dict)
self.streams[sd_hash] = GetStream(self.sd_identifier, self.session,
self.exchange_rate_manager, conf.settings['max_key_fee'],
conf.settings['disable_max_key_fee'],
conf.settings['data_rate'], timeout)
self.streams[sd_hash] = GetStream(
self.sd_identifier, self.wallet, self.exchange_rate_manager, self.blob_manager,
self.dht_node.peer_finder, self.rate_limiter, self.payment_rate_manager, self.storage,
conf.settings['max_key_fee'], conf.settings['disable_max_key_fee'], conf.settings['data_rate'],
timeout
)
try:
lbry_file, finished_deferred = yield self.streams[sd_hash].start(
claim_dict, name, txid, nout, file_name
@ -350,9 +356,9 @@ class Daemon(AuthJSONRPCServer):
@defer.inlineCallbacks
def _publish_stream(self, name, bid, claim_dict, file_path=None, certificate_id=None,
claim_address=None, change_address=None):
publisher = Publisher(self.session, self.file_manager, self.wallet,
certificate_id)
publisher = Publisher(
self.blob_manager, self.payment_rate_manager, self.storage, self.file_manager, self.wallet, certificate_id
)
parse_lbry_uri(name)
if not file_path:
stream_hash = yield self.storage.get_stream_hash_for_sd_hash(
@ -388,16 +394,17 @@ class Daemon(AuthJSONRPCServer):
def _get_or_download_sd_blob(self, blob, sd_hash):
if blob:
return self.session.blob_manager.get_blob(blob[0])
return self.blob_manager.get_blob(blob[0])
return download_sd_blob(
self.session, sd_hash, self.session.payment_rate_manager, conf.settings['search_timeout']
sd_hash, self.blob_manager, self.dht_node.peer_finder, self.rate_limiter, self.payment_rate_manager,
self.wallet, timeout=conf.settings['search_timeout'], download_mirrors=conf.settings['download_mirrors']
)
def get_or_download_sd_blob(self, sd_hash):
"""Return previously downloaded sd blob if already in the blob
manager, otherwise download and return it
"""
d = self.session.blob_manager.completed_blobs([sd_hash])
d = self.blob_manager.completed_blobs([sd_hash])
d.addCallback(self._get_or_download_sd_blob, sd_hash)
return d
@ -416,7 +423,7 @@ class Daemon(AuthJSONRPCServer):
Calculate estimated LBC cost for a stream given its size in bytes
"""
if self.session.payment_rate_manager.generous:
if self.payment_rate_manager.generous:
return 0.0
return size / (10 ** 6) * conf.settings['data_rate']
@ -654,15 +661,12 @@ class Daemon(AuthJSONRPCServer):
############################################################################
@defer.inlineCallbacks
def jsonrpc_status(self, session_status=False):
def jsonrpc_status(self):
"""
Get daemon status
Usage:
status [--session_status]
Options:
--session_status : (bool) include session status in results
status
Returns:
(dict) lbrynet-daemon status
@ -679,6 +683,8 @@ class Daemon(AuthJSONRPCServer):
'hash_announcer': (bool),
'stream_identifier': (bool),
'file_manager': (bool),
'blob_manager': (bool),
'blockchain_headers': (bool),
'peer_protocol_server': (bool),
'reflector': (bool),
'upnp': (bool),
@ -688,35 +694,33 @@ class Daemon(AuthJSONRPCServer):
'code': (str) connection status code,
'message': (str) connection status message
},
'blockchain_status': {
'blockchain_headers': {
'downloading_headers': (bool),
'download_progress': (float) 0-100.0
},
'wallet': {
'blocks': (int) local blockchain height,
'blocks_behind': (int) remote_height - local_height,
'best_blockhash': (str) block hash of most recent block,
'is_encrypted': (bool)
},
'dht_node_status': {
'dht': {
'node_id': (str) lbry dht node id - hex encoded,
'peers_in_routing_table': (int) the number of peers in the routing table,
},
'wallet_is_encrypted': (bool),
If given the session status option:
'session_status': {
'managed_blobs': (int) count of blobs in the blob manager,
'managed_streams': (int) count of streams in the file manager,
'announce_queue_size': (int) number of blobs currently queued to be announced,
'should_announce_blobs': (int) number of blobs that should be announced,
'blob_manager': {
'finished_blobs': (int) number of finished blobs in the blob manager,
},
'hash_announcer': {
'announce_queue_size': (int) number of blobs currently queued to be announced
},
'file_manager': {
'managed_files': (int) count of files in the file manager,
}
}
"""
# on startup, the wallet or network won't be available but we still need this call to work
has_wallet = self.session and self.wallet and self.wallet.network
local_height = self.wallet.network.get_local_height() if has_wallet else 0
remote_height = self.wallet.network.get_server_height() if has_wallet else 0
best_hash = (yield self.wallet.get_best_blockhash()) if has_wallet else None
wallet_is_encrypted = has_wallet and self.wallet.wallet and \
self.wallet.wallet.use_encryption
connection_code = CONNECTION_STATUS_CONNECTED if utils.check_connection() else CONNECTION_STATUS_NETWORK
connection_code = CONNECTION_STATUS_CONNECTED if self.connected_to_internet else CONNECTION_STATUS_NETWORK
response = {
'installation_id': conf.settings.installation_id,
'is_running': all(self.component_manager.get_components_status().values()),
@ -727,29 +731,11 @@ class Daemon(AuthJSONRPCServer):
'code': connection_code,
'message': CONNECTION_MESSAGES[connection_code],
},
'wallet_is_encrypted': wallet_is_encrypted,
'blocks_behind': remote_height - local_height, # deprecated. remove from UI, then here
'blockchain_status': {
'blocks': local_height,
'blocks_behind': remote_height - local_height,
'best_blockhash': best_hash,
},
'dht_node_status': {
'node_id': conf.settings.node_id.encode('hex'),
'peers_in_routing_table': 0 if not self.component_manager.all_components_running(DHT_COMPONENT) else
len(self.dht_node.contacts)
}
}
if session_status:
blobs = yield self.session.blob_manager.get_all_verified_blobs()
announce_queue_size = self.session.hash_announcer.hash_queue_size()
should_announce_blobs = yield self.session.blob_manager.count_should_announce_blobs()
response['session_status'] = {
'managed_blobs': len(blobs),
'managed_streams': len(self.file_manager.lbry_files),
'announce_queue_size': announce_queue_size,
'should_announce_blobs': should_announce_blobs,
}
for component in self.component_manager.components:
status = yield defer.maybeDeferred(component.get_status)
if status:
response[component.component_name] = status
defer.returnValue(response)
def jsonrpc_version(self):
@ -1302,7 +1288,9 @@ class Daemon(AuthJSONRPCServer):
response = yield self._render_response(results)
defer.returnValue(response)
@requires(WALLET_COMPONENT, FILE_MANAGER_COMPONENT, SESSION_COMPONENT, conditions=[WALLET_IS_UNLOCKED])
@requires(STREAM_IDENTIFIER_COMPONENT, WALLET_COMPONENT, EXCHANGE_RATE_MANAGER_COMPONENT, BLOB_COMPONENT,
DHT_COMPONENT, RATE_LIMITER_COMPONENT, PAYMENT_RATE_COMPONENT, DATABASE_COMPONENT,
conditions=[WALLET_IS_UNLOCKED])
@defer.inlineCallbacks
def jsonrpc_get(self, uri, file_name=None, timeout=None):
"""
@ -1493,7 +1481,9 @@ class Daemon(AuthJSONRPCServer):
response = yield self._render_response(result)
defer.returnValue(response)
@requires(WALLET_COMPONENT, SESSION_COMPONENT, conditions=[WALLET_IS_UNLOCKED])
@requires(STREAM_IDENTIFIER_COMPONENT, WALLET_COMPONENT, EXCHANGE_RATE_MANAGER_COMPONENT, BLOB_COMPONENT,
DHT_COMPONENT, RATE_LIMITER_COMPONENT, PAYMENT_RATE_COMPONENT, DATABASE_COMPONENT,
conditions=[WALLET_IS_UNLOCKED])
@defer.inlineCallbacks
def jsonrpc_stream_cost_estimate(self, uri, size=None):
"""
@ -1648,7 +1638,8 @@ class Daemon(AuthJSONRPCServer):
result = yield self.wallet.import_certificate_info(serialized_certificate_info)
defer.returnValue(result)
@requires(WALLET_COMPONENT, FILE_MANAGER_COMPONENT, SESSION_COMPONENT, conditions=[WALLET_IS_UNLOCKED])
@requires(WALLET_COMPONENT, FILE_MANAGER_COMPONENT, BLOB_COMPONENT, PAYMENT_RATE_COMPONENT, DATABASE_COMPONENT,
conditions=[WALLET_IS_UNLOCKED])
@defer.inlineCallbacks
def jsonrpc_publish(self, name, bid, metadata=None, file_path=None, fee=None, title=None,
description=None, author=None, language=None, license=None,
@ -2531,7 +2522,8 @@ class Daemon(AuthJSONRPCServer):
d.addCallback(lambda r: self._render_response(r))
return d
@requires(WALLET_COMPONENT, SESSION_COMPONENT, conditions=[WALLET_IS_UNLOCKED])
@requires(WALLET_COMPONENT, DHT_COMPONENT, BLOB_COMPONENT, RATE_LIMITER_COMPONENT, PAYMENT_RATE_COMPONENT,
conditions=[WALLET_IS_UNLOCKED])
@defer.inlineCallbacks
def jsonrpc_blob_get(self, blob_hash, timeout=None, encoding=None, payment_rate_manager=None):
"""
@ -2562,9 +2554,7 @@ class Daemon(AuthJSONRPCServer):
}
timeout = timeout or 30
payment_rate_manager = get_blob_payment_rate_manager(self.session, payment_rate_manager)
blob = yield self._download_blob(blob_hash, rate_manager=payment_rate_manager,
timeout=timeout)
blob = yield self._download_blob(blob_hash, rate_manager=self.payment_rate_manager, timeout=timeout)
if encoding and encoding in decoders:
blob_file = blob.open_for_reading()
result = decoders[encoding](blob_file.read())
@ -2575,7 +2565,7 @@ class Daemon(AuthJSONRPCServer):
response = yield self._render_response(result)
defer.returnValue(response)
@requires(SESSION_COMPONENT)
@requires(BLOB_COMPONENT, DATABASE_COMPONENT)
@defer.inlineCallbacks
def jsonrpc_blob_delete(self, blob_hash):
"""
@ -2591,7 +2581,7 @@ class Daemon(AuthJSONRPCServer):
(str) Success/fail message
"""
if blob_hash not in self.session.blob_manager.blobs:
if blob_hash not in self.blob_manager.blobs:
response = yield self._render_response("Don't have that blob")
defer.returnValue(response)
try:
@ -2599,7 +2589,7 @@ class Daemon(AuthJSONRPCServer):
yield self.storage.delete_stream(stream_hash)
except Exception as err:
pass
yield self.session.blob_manager.delete_blobs([blob_hash])
yield self.blob_manager.delete_blobs([blob_hash])
response = yield self._render_response("Deleted %s" % blob_hash)
defer.returnValue(response)
@ -2629,7 +2619,7 @@ class Daemon(AuthJSONRPCServer):
err.trap(defer.TimeoutError)
return []
finished_deferred.addTimeout(timeout or conf.settings['peer_search_timeout'], self.session.dht_node.clock)
finished_deferred.addTimeout(timeout or conf.settings['peer_search_timeout'], self.dht_node.clock)
finished_deferred.addErrback(trap_timeout)
peers = yield finished_deferred
results = [
@ -2642,7 +2632,7 @@ class Daemon(AuthJSONRPCServer):
]
defer.returnValue(results)
@requires(SESSION_COMPONENT, DHT_COMPONENT, conditions=[DHT_HAS_CONTACTS])
@requires(DATABASE_COMPONENT)
@defer.inlineCallbacks
def jsonrpc_blob_announce(self, blob_hash=None, stream_hash=None, sd_hash=None):
"""
@ -2715,7 +2705,7 @@ class Daemon(AuthJSONRPCServer):
results = yield reupload.reflect_file(lbry_file, reflector_server=reflector_server)
defer.returnValue(results)
@requires(SESSION_COMPONENT, WALLET_COMPONENT)
@requires(BLOB_COMPONENT, WALLET_COMPONENT)
@defer.inlineCallbacks
def jsonrpc_blob_list(self, uri=None, stream_hash=None, sd_hash=None, needed=None,
finished=None, page_size=None, page=None):
@ -2754,16 +2744,16 @@ class Daemon(AuthJSONRPCServer):
if stream_hash:
crypt_blobs = yield self.storage.get_blobs_for_stream(stream_hash)
blobs = yield defer.gatherResults([
self.session.blob_manager.get_blob(crypt_blob.blob_hash, crypt_blob.length)
self.blob_manager.get_blob(crypt_blob.blob_hash, crypt_blob.length)
for crypt_blob in crypt_blobs if crypt_blob.blob_hash is not None
])
else:
blobs = []
# get_blobs_for_stream does not include the sd blob, so we'll add it manually
if sd_hash in self.session.blob_manager.blobs:
blobs = [self.session.blob_manager.blobs[sd_hash]] + blobs
if sd_hash in self.blob_manager.blobs:
blobs = [self.blob_manager.blobs[sd_hash]] + blobs
else:
blobs = self.session.blob_manager.blobs.itervalues()
blobs = self.blob_manager.blobs.itervalues()
if needed:
blobs = [blob for blob in blobs if not blob.get_is_verified()]
@ -2779,7 +2769,7 @@ class Daemon(AuthJSONRPCServer):
response = yield self._render_response(blob_hashes_for_return)
defer.returnValue(response)
@requires(SESSION_COMPONENT)
@requires(BLOB_COMPONENT)
def jsonrpc_blob_reflect(self, blob_hashes, reflector_server=None):
"""
Reflects specified blobs
@ -2794,11 +2784,11 @@ class Daemon(AuthJSONRPCServer):
(list) reflected blob hashes
"""
d = reupload.reflect_blob_hashes(blob_hashes, self.session.blob_manager, reflector_server)
d = reupload.reflect_blob_hashes(blob_hashes, self.blob_manager, reflector_server)
d.addCallback(lambda r: self._render_response(r))
return d
@requires(SESSION_COMPONENT)
@requires(BLOB_COMPONENT)
def jsonrpc_blob_reflect_all(self):
"""
Reflects all saved blobs
@ -2813,8 +2803,8 @@ class Daemon(AuthJSONRPCServer):
(bool) true if successful
"""
d = self.session.blob_manager.get_all_verified_blobs()
d.addCallback(reupload.reflect_blob_hashes, self.session.blob_manager)
d = self.blob_manager.get_all_verified_blobs()
d.addCallback(reupload.reflect_blob_hashes, self.blob_manager)
d.addCallback(lambda r: self._render_response(r))
return d
@ -2960,7 +2950,7 @@ class Daemon(AuthJSONRPCServer):
return self._blob_availability(blob_hash, search_timeout, blob_timeout)
@requires(SESSION_COMPONENT, WALLET_COMPONENT, DHT_COMPONENT, conditions=[WALLET_IS_UNLOCKED])
@requires(UPNP_COMPONENT, WALLET_COMPONENT, DHT_COMPONENT, conditions=[WALLET_IS_UNLOCKED])
@AuthJSONRPCServer.deprecated("stream_availability")
def jsonrpc_get_availability(self, uri, sd_timeout=None, peer_timeout=None):
"""
@ -2981,7 +2971,7 @@ class Daemon(AuthJSONRPCServer):
return self.jsonrpc_stream_availability(uri, peer_timeout, sd_timeout)
@requires(SESSION_COMPONENT, WALLET_COMPONENT, DHT_COMPONENT, conditions=[WALLET_IS_UNLOCKED])
@requires(UPNP_COMPONENT, WALLET_COMPONENT, DHT_COMPONENT, conditions=[WALLET_IS_UNLOCKED])
@defer.inlineCallbacks
def jsonrpc_stream_availability(self, uri, search_timeout=None, blob_timeout=None):
"""
@ -3029,7 +3019,7 @@ class Daemon(AuthJSONRPCServer):
'head_blob_hash': None,
'head_blob_availability': {},
'use_upnp': conf.settings['use_upnp'],
'upnp_redirect_is_set': len(self.session.upnp_redirects) > 0,
'upnp_redirect_is_set': len(self.upnp.get_redirects()) > 0,
'error': None
}
@ -3059,7 +3049,7 @@ class Daemon(AuthJSONRPCServer):
response['sd_hash'] = sd_hash
head_blob_hash = None
downloader = self._get_single_peer_downloader()
have_sd_blob = sd_hash in self.session.blob_manager.blobs
have_sd_blob = sd_hash in self.blob_manager.blobs
try:
sd_blob = yield self.jsonrpc_blob_get(sd_hash, timeout=blob_timeout,
encoding="json")
@ -3158,17 +3148,6 @@ def iter_lbry_file_search_values(search_fields):
yield searchtype, value
def get_blob_payment_rate_manager(session, payment_rate_manager=None):
if payment_rate_manager:
rate_managers = {
'only-free': OnlyFreePaymentsManager()
}
if payment_rate_manager in rate_managers:
payment_rate_manager = rate_managers[payment_rate_manager]
log.info("Downloading blob with rate manager: %s", payment_rate_manager)
return payment_rate_manager or session.payment_rate_manager
def create_key_getter(field):
search_path = field.split('.')
def key_getter(value):

21
lbrynet/daemon/Downloader.py
View file

@ -30,8 +30,8 @@ log = logging.getLogger(__name__)
class GetStream(object):
def __init__(self, sd_identifier, session, exchange_rate_manager,
max_key_fee, disable_max_key_fee, data_rate=None, timeout=None):
def __init__(self, sd_identifier, wallet, exchange_rate_manager, blob_manager, peer_finder, rate_limiter,
payment_rate_manager, storage, max_key_fee, disable_max_key_fee, data_rate=None, timeout=None):
self.timeout = timeout or conf.settings['download_timeout']
self.data_rate = data_rate or conf.settings['data_rate']
@ -41,11 +41,14 @@ class GetStream(object):
self.timeout_counter = 0
self.code = None
self.sd_hash = None
self.session = session
self.wallet = self.session.wallet
self.blob_manager = blob_manager
self.peer_finder = peer_finder
self.rate_limiter = rate_limiter
self.wallet = wallet
self.exchange_rate_manager = exchange_rate_manager
self.payment_rate_manager = self.session.payment_rate_manager
self.payment_rate_manager = payment_rate_manager
self.sd_identifier = sd_identifier
self.storage = storage
self.downloader = None
self.checker = LoopingCall(self.check_status)
@ -174,15 +177,17 @@ class GetStream(object):
@defer.inlineCallbacks
def _download_sd_blob(self):
sd_blob = yield download_sd_blob(self.session, self.sd_hash,
self.payment_rate_manager, self.timeout)
sd_blob = yield download_sd_blob(
self.sd_hash, self.blob_manager, self.peer_finder, self.rate_limiter, self.payment_rate_manager,
self.wallet, self.timeout, conf.settings['download_mirrors']
)
defer.returnValue(sd_blob)
@defer.inlineCallbacks
def _download(self, sd_blob, name, key_fee, txid, nout, file_name=None):
self.downloader = yield self._create_downloader(sd_blob, file_name=file_name)
yield self.pay_key_fee(key_fee, name)
yield self.session.storage.save_content_claim(self.downloader.stream_hash, "%s:%i" % (txid, nout))
yield self.storage.save_content_claim(self.downloader.stream_hash, "%s:%i" % (txid, nout))
log.info("Downloading lbry://%s (%s) --> %s", name, self.sd_hash[:6], self.download_path)
self.finished_deferred = self.downloader.start()
self.finished_deferred.addCallbacks(lambda result: self.finish(result, name), self.fail)

24
lbrynet/daemon/Publisher.py
View file

@ -11,8 +11,10 @@ log = logging.getLogger(__name__)
class Publisher(object):
def __init__(self, session, lbry_file_manager, wallet, certificate_id):
self.session = session
def __init__(self, blob_manager, payment_rate_manager, storage, lbry_file_manager, wallet, certificate_id):
self.blob_manager = blob_manager
self.payment_rate_manager = payment_rate_manager
self.storage = storage
self.lbry_file_manager = lbry_file_manager
self.wallet = wallet
self.certificate_id = certificate_id
@ -30,8 +32,10 @@ class Publisher(object):
file_name = os.path.basename(file_path)
with file_utils.get_read_handle(file_path) as read_handle:
self.lbry_file = yield create_lbry_file(self.session, self.lbry_file_manager, file_name,
read_handle)
self.lbry_file = yield create_lbry_file(
self.blob_manager, self.storage, self.payment_rate_manager, self.lbry_file_manager, file_name,
read_handle
)
if 'source' not in claim_dict['stream']:
claim_dict['stream']['source'] = {}
@ -42,15 +46,16 @@ class Publisher(object):
claim_out = yield self.make_claim(name, bid, claim_dict, claim_address, change_address)
# check if we have a file already for this claim (if this is a publish update with a new stream)
old_stream_hashes = yield self.session.storage.get_old_stream_hashes_for_claim_id(claim_out['claim_id'],
self.lbry_file.stream_hash)
old_stream_hashes = yield self.storage.get_old_stream_hashes_for_claim_id(
claim_out['claim_id'], self.lbry_file.stream_hash
)
if old_stream_hashes:
for lbry_file in filter(lambda l: l.stream_hash in old_stream_hashes,
list(self.lbry_file_manager.lbry_files)):
yield self.lbry_file_manager.delete_lbry_file(lbry_file, delete_file=False)
log.info("Removed old stream for claim update: %s", lbry_file.stream_hash)
yield self.session.storage.save_content_claim(
yield self.storage.save_content_claim(
self.lbry_file.stream_hash, "%s:%i" % (claim_out['txid'], claim_out['nout'])
)
defer.returnValue(claim_out)
@ -60,8 +65,9 @@ class Publisher(object):
"""Make a claim without creating a lbry file"""
claim_out = yield self.make_claim(name, bid, claim_dict, claim_address, change_address)
if stream_hash: # the stream_hash returned from the db will be None if this isn't a stream we have
yield self.session.storage.save_content_claim(stream_hash, "%s:%i" % (claim_out['txid'],
claim_out['nout']))
yield self.storage.save_content_claim(
stream_hash, "%s:%i" % (claim_out['txid'], claim_out['nout'])
)
self.lbry_file = [f for f in self.lbry_file_manager.lbry_files if f.stream_hash == stream_hash][0]
defer.returnValue(claim_out)

7
lbrynet/daemon/auth/server.py
View file

@ -196,11 +196,14 @@ class AuthJSONRPCServer(AuthorizedBase):
component_attributes = {}
def __init__(self, analytics_manager=None, component_manager=None, use_authentication=None, to_skip=None,
looping_calls=None):
looping_calls=None, reactor=None):
if not reactor:
from twisted.internet import reactor
self.analytics_manager = analytics_manager or analytics.Manager.new_instance()
self.component_manager = component_manager or ComponentManager(
analytics_manager=self.analytics_manager,
skip_components=to_skip or []
skip_components=to_skip or [],
reactor=reactor
)
self.looping_call_manager = LoopingCallManager({n: lc for n, (lc, t) in (looping_calls or {}).iteritems()})
self._looping_call_times = {n: t for n, (lc, t) in (looping_calls or {}).iteritems()}

16
lbrynet/database/storage.py
View file

@ -181,10 +181,17 @@ class SQLiteStorage(object):
# when it loads each file
self.content_claim_callbacks = {} # {<stream_hash>: <callable returning a deferred>}
if 'reflector' not in conf.settings['components_to_skip']:
self.check_should_announce_lc = task.LoopingCall(self.verify_will_announce_all_head_and_sd_blobs)
@defer.inlineCallbacks
def setup(self):
def _create_tables(transaction):
transaction.executescript(self.CREATE_TABLES_QUERY)
return self.db.runInteraction(_create_tables)
yield self.db.runInteraction(_create_tables)
if self.check_should_announce_lc and not self.check_should_announce_lc.running:
self.check_should_announce_lc.start(600)
defer.returnValue(None)
@defer.inlineCallbacks
def run_and_return_one_or_none(self, query, *args):
@ -203,6 +210,8 @@ class SQLiteStorage(object):
defer.returnValue([])
def stop(self):
if self.check_should_announce_lc and self.check_should_announce_lc.running:
self.check_should_announce_lc.stop()
self.db.close()
return defer.succeed(True)
@ -252,6 +261,11 @@ class SQLiteStorage(object):
)
defer.returnValue([blob_hash.decode('hex') for blob_hash in blob_hashes])
def count_finished_blobs(self):
return self.run_and_return_one_or_none(
"select count(*) from blob where status='finished'"
)
def update_last_announced_blob(self, blob_hash, last_announced):
return self.db.runOperation(
"update blob set next_announce_time=?, last_announced_time=?, single_announce=0 where blob_hash=?",

13
lbrynet/file_manager/EncryptedFileCreator.py
View file

@ -59,7 +59,8 @@ class EncryptedFileStreamCreator(CryptStreamCreator):
# we can simply read the file from the disk without needing to
# involve reactor.
@defer.inlineCallbacks
def create_lbry_file(session, lbry_file_manager, file_name, file_handle, key=None, iv_generator=None):
def create_lbry_file(blob_manager, storage, payment_rate_manager, lbry_file_manager, file_name, file_handle,
key=None, iv_generator=None):
"""Turn a plain file into an LBRY File.
An LBRY File is a collection of encrypted blobs of data and the metadata that binds them
@ -98,7 +99,7 @@ def create_lbry_file(session, lbry_file_manager, file_name, file_handle, key=Non
file_directory = os.path.dirname(file_handle.name)
lbry_file_creator = EncryptedFileStreamCreator(
session.blob_manager, lbry_file_manager, base_file_name, key, iv_generator
blob_manager, lbry_file_manager, base_file_name, key, iv_generator
)
yield lbry_file_creator.setup()
@ -114,18 +115,18 @@ def create_lbry_file(session, lbry_file_manager, file_name, file_handle, key=Non
log.debug("making the sd blob")
sd_info = lbry_file_creator.sd_info
descriptor_writer = BlobStreamDescriptorWriter(session.blob_manager)
descriptor_writer = BlobStreamDescriptorWriter(blob_manager)
sd_hash = yield descriptor_writer.create_descriptor(sd_info)
log.debug("saving the stream")
yield session.storage.store_stream(
yield storage.store_stream(
sd_info['stream_hash'], sd_hash, sd_info['stream_name'], sd_info['key'],
sd_info['suggested_file_name'], sd_info['blobs']
)
log.debug("adding to the file manager")
lbry_file = yield lbry_file_manager.add_published_file(
sd_info['stream_hash'], sd_hash, binascii.hexlify(file_directory), session.payment_rate_manager,
session.payment_rate_manager.min_blob_data_payment_rate
sd_info['stream_hash'], sd_hash, binascii.hexlify(file_directory), payment_rate_manager,
payment_rate_manager.min_blob_data_payment_rate
)
defer.returnValue(lbry_file)

18
lbrynet/file_manager/EncryptedFileDownloader.py
View file

@ -6,7 +6,7 @@ import binascii
from zope.interface import implements
from twisted.internet import defer
from lbrynet import conf
from lbrynet.core.client.StreamProgressManager import FullStreamProgressManager
from lbrynet.core.HTTPBlobDownloader import HTTPBlobDownloader
from lbrynet.core.utils import short_hash
@ -56,7 +56,11 @@ class ManagedEncryptedFileDownloader(EncryptedFileSaver):
self.channel_claim_id = None
self.channel_name = None
self.metadata = None
self.mirror = HTTPBlobDownloader(self.blob_manager, servers=download_mirrors) if download_mirrors else None
self.mirror = None
if download_mirrors:
self.mirror = HTTPBlobDownloader(
self.blob_manager, servers=download_mirrors or conf.settings['download_mirrors']
)
def set_claim_info(self, claim_info):
self.claim_id = claim_info['claim_id']
@ -163,23 +167,25 @@ class ManagedEncryptedFileDownloader(EncryptedFileSaver):
class ManagedEncryptedFileDownloaderFactory(object):
implements(IStreamDownloaderFactory)
def __init__(self, lbry_file_manager):
def __init__(self, lbry_file_manager, blob_manager):
self.lbry_file_manager = lbry_file_manager
self.blob_manager = blob_manager
def can_download(self, sd_validator):
# TODO: add a sd_validator for non live streams, use it
return True
@defer.inlineCallbacks
def make_downloader(self, metadata, data_rate, payment_rate_manager, download_directory, file_name=None):
stream_hash = yield save_sd_info(self.lbry_file_manager.session.blob_manager,
def make_downloader(self, metadata, data_rate, payment_rate_manager, download_directory, file_name=None,
download_mirrors=None):
stream_hash = yield save_sd_info(self.blob_manager,
metadata.source_blob_hash,
metadata.validator.raw_info)
if file_name:
file_name = binascii.hexlify(file_name)
lbry_file = yield self.lbry_file_manager.add_downloaded_file(
stream_hash, metadata.source_blob_hash, binascii.hexlify(download_directory), payment_rate_manager,
data_rate, file_name=file_name
data_rate, file_name=file_name, download_mirrors=download_mirrors
)
defer.returnValue(lbry_file)

60
lbrynet/file_manager/EncryptedFileManager.py
View file

@ -28,15 +28,17 @@ class EncryptedFileManager(object):
# when reflecting files, reflect up to this many files at a time
CONCURRENT_REFLECTS = 5
def __init__(self, session, sd_identifier):
def __init__(self, peer_finder, rate_limiter, blob_manager, wallet, payment_rate_manager, storage, sd_identifier):
self.auto_re_reflect = conf.settings['reflect_uploads'] and conf.settings['auto_re_reflect_interval'] > 0
self.auto_re_reflect_interval = conf.settings['auto_re_reflect_interval']
self.session = session
self.storage = session.storage
self.peer_finder = peer_finder
self.rate_limiter = rate_limiter
self.blob_manager = blob_manager
self.wallet = wallet
self.payment_rate_manager = payment_rate_manager
self.storage = storage
# TODO: why is sd_identifier part of the file manager?
self.sd_identifier = sd_identifier
assert sd_identifier
self.lbry_files = []
self.lbry_file_reflector = task.LoopingCall(self.reflect_lbry_files)
@ -47,14 +49,14 @@ class EncryptedFileManager(object):
log.info("Started file manager")
def get_lbry_file_status(self, lbry_file):
return self.session.storage.get_lbry_file_status(lbry_file.rowid)
return self.storage.get_lbry_file_status(lbry_file.rowid)
def set_lbry_file_data_payment_rate(self, lbry_file, new_rate):
return self.session.storage(lbry_file.rowid, new_rate)
return self.storage(lbry_file.rowid, new_rate)
def change_lbry_file_status(self, lbry_file, status):
log.debug("Changing status of %s to %s", lbry_file.stream_hash, status)
return self.session.storage.change_file_status(lbry_file.rowid, status)
return self.storage.change_file_status(lbry_file.rowid, status)
def get_lbry_file_status_reports(self):
ds = []
@ -71,36 +73,36 @@ class EncryptedFileManager(object):
return dl
def _add_to_sd_identifier(self):
downloader_factory = ManagedEncryptedFileDownloaderFactory(self)
downloader_factory = ManagedEncryptedFileDownloaderFactory(self, self.blob_manager)
self.sd_identifier.add_stream_downloader_factory(
EncryptedFileStreamType, downloader_factory)
def _get_lbry_file(self, rowid, stream_hash, payment_rate_manager, sd_hash, key,
stream_name, file_name, download_directory, suggested_file_name):
stream_name, file_name, download_directory, suggested_file_name, download_mirrors=None):
return ManagedEncryptedFileDownloader(
rowid,
stream_hash,
self.session.peer_finder,
self.session.rate_limiter,
self.session.blob_manager,
self.session.storage,
self.peer_finder,
self.rate_limiter,
self.blob_manager,
self.storage,
self,
payment_rate_manager,
self.session.wallet,
self.wallet,
download_directory,
file_name,
stream_name=stream_name,
sd_hash=sd_hash,
key=key,
suggested_file_name=suggested_file_name,
download_mirrors=self.session.download_mirrors
download_mirrors=download_mirrors
)
def _start_lbry_file(self, file_info, payment_rate_manager, claim_info):
def _start_lbry_file(self, file_info, payment_rate_manager, claim_info, download_mirrors=None):
lbry_file = self._get_lbry_file(
file_info['row_id'], file_info['stream_hash'], payment_rate_manager, file_info['sd_hash'],
file_info['key'], file_info['stream_name'], file_info['file_name'], file_info['download_directory'],
file_info['suggested_file_name']
file_info['suggested_file_name'], download_mirrors
)
if claim_info:
lbry_file.set_claim_info(claim_info)
@ -116,9 +118,9 @@ class EncryptedFileManager(object):
@defer.inlineCallbacks
def _start_lbry_files(self):
files = yield self.session.storage.get_all_lbry_files()
claim_infos = yield self.session.storage.get_claims_from_stream_hashes([file['stream_hash'] for file in files])
prm = self.session.payment_rate_manager
files = yield self.storage.get_all_lbry_files()
claim_infos = yield self.storage.get_claims_from_stream_hashes([file['stream_hash'] for file in files])
prm = self.payment_rate_manager
log.info("Starting %i files", len(files))
for file_info in files:
@ -154,7 +156,7 @@ class EncryptedFileManager(object):
@defer.inlineCallbacks
def add_published_file(self, stream_hash, sd_hash, download_directory, payment_rate_manager, blob_data_rate):
status = ManagedEncryptedFileDownloader.STATUS_FINISHED
stream_metadata = yield get_sd_info(self.session.storage, stream_hash, include_blobs=False)
stream_metadata = yield get_sd_info(self.storage, stream_hash, include_blobs=False)
key = stream_metadata['key']
stream_name = stream_metadata['stream_name']
file_name = stream_metadata['suggested_file_name']
@ -163,7 +165,7 @@ class EncryptedFileManager(object):
)
lbry_file = self._get_lbry_file(
rowid, stream_hash, payment_rate_manager, sd_hash, key, stream_name, file_name, download_directory,
stream_metadata['suggested_file_name']
stream_metadata['suggested_file_name'], download_mirrors=None
)
lbry_file.restore(status)
yield lbry_file.get_claim_info()
@ -173,11 +175,11 @@ class EncryptedFileManager(object):
@defer.inlineCallbacks
def add_downloaded_file(self, stream_hash, sd_hash, download_directory, payment_rate_manager=None,
blob_data_rate=None, status=None, file_name=None):
blob_data_rate=None, status=None, file_name=None, download_mirrors=None):
status = status or ManagedEncryptedFileDownloader.STATUS_STOPPED
payment_rate_manager = payment_rate_manager or self.session.payment_rate_manager
payment_rate_manager = payment_rate_manager or self.payment_rate_manager
blob_data_rate = blob_data_rate or payment_rate_manager.min_blob_data_payment_rate
stream_metadata = yield get_sd_info(self.session.storage, stream_hash, include_blobs=False)
stream_metadata = yield get_sd_info(self.storage, stream_hash, include_blobs=False)
key = stream_metadata['key']
stream_name = stream_metadata['stream_name']
file_name = file_name or stream_metadata['suggested_file_name']
@ -187,10 +189,10 @@ class EncryptedFileManager(object):
rowid = yield self.storage.save_downloaded_file(
stream_hash, os.path.basename(file_name.decode('hex')).encode('hex'), download_directory, blob_data_rate
)
file_name = yield self.session.storage.get_filename_for_rowid(rowid)
file_name = yield self.storage.get_filename_for_rowid(rowid)
lbry_file = self._get_lbry_file(
rowid, stream_hash, payment_rate_manager, sd_hash, key, stream_name, file_name, download_directory,
stream_metadata['suggested_file_name']
stream_metadata['suggested_file_name'], download_mirrors
)
lbry_file.restore(status)
yield lbry_file.get_claim_info(include_supports=False)
@ -222,7 +224,7 @@ class EncryptedFileManager(object):
del self.storage.content_claim_callbacks[lbry_file.stream_hash]
yield lbry_file.delete_data()
yield self.session.storage.delete_stream(lbry_file.stream_hash)
yield self.storage.delete_stream(lbry_file.stream_hash)
if delete_file and os.path.isfile(full_path):
os.remove(full_path)

1091
lbrynet/tests/functional/test_misc.py
View file

File diff suppressed because it is too large Load diff

174
lbrynet/tests/functional/test_reflector.py
View file

@ -1,33 +1,42 @@
from twisted.internet import defer, threads, error
import os
from twisted.internet import defer, error
from twisted.trial import unittest
from lbrynet import conf
from lbrynet.core.StreamDescriptor import get_sd_info
from lbrynet import reflector
from lbrynet.core import BlobManager, PeerManager
from lbrynet.core import Session
from lbrynet.core import StreamDescriptor
from lbrynet.lbry_file.client import EncryptedFileOptions
from lbrynet.file_manager import EncryptedFileCreator
from lbrynet.file_manager import EncryptedFileManager
from lbrynet.file_manager.EncryptedFileManager import EncryptedFileManager
from lbrynet.core.RateLimiter import DummyRateLimiter
from lbrynet.database.storage import SQLiteStorage
from lbrynet.core.PaymentRateManager import OnlyFreePaymentsManager
from lbrynet.tests import mocks
from lbrynet.tests.util import mk_db_and_blob_dir, rm_db_and_blob_dir
class TestReflector(unittest.TestCase):
def setUp(self):
mocks.mock_conf_settings(self)
self.session = None
self.lbry_file_manager = None
self.server_blob_manager = None
self.reflector_port = None
self.port = None
self.addCleanup(self.take_down_env)
mocks.mock_conf_settings(self)
self.server_db_dir, self.server_blob_dir = mk_db_and_blob_dir()
self.client_db_dir, self.client_blob_dir = mk_db_and_blob_dir()
prm = OnlyFreePaymentsManager()
wallet = mocks.Wallet()
peer_manager = PeerManager.PeerManager()
peer_finder = mocks.PeerFinder(5553, peer_manager, 2)
sd_identifier = StreamDescriptor.StreamDescriptorIdentifier()
self.server_storage = SQLiteStorage(self.server_db_dir)
self.server_blob_manager = BlobManager.DiskBlobManager(self.server_blob_dir, self.server_storage)
self.client_storage = SQLiteStorage(self.client_db_dir)
self.client_blob_manager = BlobManager.DiskBlobManager(self.client_blob_dir, self.client_storage)
self.server_lbry_file_manager = EncryptedFileManager(
peer_finder, DummyRateLimiter(), self.server_blob_manager, wallet, prm, self.server_storage,
StreamDescriptor.StreamDescriptorIdentifier()
)
self.client_lbry_file_manager = EncryptedFileManager(
peer_finder, DummyRateLimiter(), self.client_blob_manager, wallet, prm, self.client_storage,
StreamDescriptor.StreamDescriptorIdentifier()
)
self.expected_blobs = [
(
@ -46,60 +55,18 @@ class TestReflector(unittest.TestCase):
1015056
),
]
## Setup reflector client classes ##
self.db_dir, self.blob_dir = mk_db_and_blob_dir()
self.session = Session.Session(
conf.settings['data_rate'],
db_dir=self.db_dir,
node_id="abcd",
peer_finder=peer_finder,
peer_manager=peer_manager,
blob_dir=self.blob_dir,
peer_port=5553,
dht_node_port=4444,
wallet=wallet,
external_ip="127.0.0.1",
dht_node=mocks.Node(),
hash_announcer=mocks.Announcer(),
)
self.lbry_file_manager = EncryptedFileManager.EncryptedFileManager(self.session,
sd_identifier)
## Setup reflector server classes ##
self.server_db_dir, self.server_blob_dir = mk_db_and_blob_dir()
self.server_session = Session.Session(
conf.settings['data_rate'],
db_dir=self.server_db_dir,
node_id="abcd",
peer_finder=peer_finder,
peer_manager=peer_manager,
blob_dir=self.server_blob_dir,
peer_port=5554,
dht_node_port=4443,
wallet=wallet,
external_ip="127.0.0.1",
dht_node=mocks.Node(),
hash_announcer=mocks.Announcer(),
)
self.server_blob_manager = BlobManager.DiskBlobManager(self.server_blob_dir,
self.server_session.storage)
self.server_lbry_file_manager = EncryptedFileManager.EncryptedFileManager(
self.server_session, sd_identifier)
d = self.session.setup()
d.addCallback(lambda _: EncryptedFileOptions.add_lbry_file_to_sd_identifier(sd_identifier))
d.addCallback(lambda _: self.lbry_file_manager.setup())
d.addCallback(lambda _: self.server_session.setup())
d = self.server_storage.setup()
d.addCallback(lambda _: self.server_blob_manager.setup())
d.addCallback(lambda _: self.server_lbry_file_manager.setup())
d.addCallback(lambda _: self.client_storage.setup())
d.addCallback(lambda _: self.client_blob_manager.setup())
d.addCallback(lambda _: self.client_lbry_file_manager.setup())
@defer.inlineCallbacks
def verify_equal(sd_info, stream_hash):
self.assertDictEqual(mocks.create_stream_sd_file, sd_info)
sd_hash = yield self.session.storage.get_sd_blob_hash_for_stream(stream_hash)
sd_hash = yield self.client_storage.get_sd_blob_hash_for_stream(stream_hash)
defer.returnValue(sd_hash)
def save_sd_blob_hash(sd_hash):
@ -108,7 +75,7 @@ class TestReflector(unittest.TestCase):
def verify_stream_descriptor_file(stream_hash):
self.stream_hash = stream_hash
d = get_sd_info(self.lbry_file_manager.session.storage, stream_hash, True)
d = get_sd_info(self.client_storage, stream_hash, True)
d.addCallback(verify_equal, stream_hash)
d.addCallback(save_sd_blob_hash)
return d
@ -116,8 +83,7 @@ class TestReflector(unittest.TestCase):
def create_stream():
test_file = mocks.GenFile(5209343, b''.join([chr(i + 3) for i in xrange(0, 64, 6)]))
d = EncryptedFileCreator.create_lbry_file(
self.session,
self.lbry_file_manager,
self.client_blob_manager, self.client_storage, prm, self.client_lbry_file_manager,
"test_file",
test_file,
key="0123456701234567",
@ -127,8 +93,7 @@ class TestReflector(unittest.TestCase):
return d
def start_server():
server_factory = reflector.ServerFactory(
peer_manager, self.server_blob_manager,
server_factory = reflector.ServerFactory(peer_manager, self.server_blob_manager,
self.server_lbry_file_manager)
from twisted.internet import reactor
port = 8943
@ -144,29 +109,31 @@ class TestReflector(unittest.TestCase):
d.addCallback(lambda _: start_server())
return d
def take_down_env(self):
d = defer.succeed(True)
## Close client classes ##
d.addCallback(lambda _: self.lbry_file_manager.stop())
d.addCallback(lambda _: self.session.shut_down())
## Close server classes ##
d.addCallback(lambda _: self.server_blob_manager.stop())
d.addCallback(lambda _: self.server_lbry_file_manager.stop())
d.addCallback(lambda _: self.server_session.shut_down())
d.addCallback(lambda _: self.reflector_port.stopListening())
def delete_test_env():
@defer.inlineCallbacks
def tearDown(self):
lbry_files = self.client_lbry_file_manager.lbry_files
for lbry_file in lbry_files:
yield self.client_lbry_file_manager.delete_lbry_file(lbry_file)
yield self.client_lbry_file_manager.stop()
yield self.client_blob_manager.stop()
yield self.client_storage.stop()
self.reflector_port.stopListening()
lbry_files = self.server_lbry_file_manager.lbry_files
for lbry_file in lbry_files:
yield self.server_lbry_file_manager.delete_lbry_file(lbry_file)
yield self.server_lbry_file_manager.stop()
yield self.server_blob_manager.stop()
yield self.server_storage.stop()
try:
rm_db_and_blob_dir(self.db_dir, self.blob_dir)
rm_db_and_blob_dir(self.server_db_dir, self.server_blob_dir)
except:
rm_db_and_blob_dir(self.client_db_dir, self.client_blob_dir)
except Exception as err:
raise unittest.SkipTest("TODO: fix this for windows")
d.addCallback(lambda _: threads.deferToThread(delete_test_env))
d.addErrback(lambda err: str(err))
return d
try:
rm_db_and_blob_dir(self.server_db_dir, self.server_blob_dir)
except Exception as err:
raise unittest.SkipTest("TODO: fix this for windows")
if os.path.exists("test_file"):
os.remove("test_file")
def test_stream_reflector(self):
def verify_blob_on_reflector():
@ -178,16 +145,15 @@ class TestReflector(unittest.TestCase):
@defer.inlineCallbacks
def verify_stream_on_reflector():
# check stream_info_manager has all the right information
streams = yield self.server_session.storage.get_all_streams()
streams = yield self.server_storage.get_all_streams()
self.assertEqual(1, len(streams))
self.assertEqual(self.stream_hash, streams[0])
blobs = yield self.server_session.storage.get_blobs_for_stream(self.stream_hash)
blobs = yield self.server_storage.get_blobs_for_stream(self.stream_hash)
blob_hashes = [b.blob_hash for b in blobs if b.blob_hash is not None]
expected_blob_hashes = [b[0] for b in self.expected_blobs[:-1] if b[0] is not None]
self.assertEqual(expected_blob_hashes, blob_hashes)
sd_hash = yield self.server_session.storage.get_sd_blob_hash_for_stream(streams[0])
expected_sd_hash = self.expected_blobs[-1][0]
sd_hash = yield self.server_storage.get_sd_blob_hash_for_stream(streams[0])
self.assertEqual(self.sd_hash, sd_hash)
# check lbry file manager has the file
@ -195,14 +161,14 @@ class TestReflector(unittest.TestCase):
self.assertEqual(0, len(files))
streams = yield self.server_lbry_file_manager.storage.get_all_streams()
streams = yield self.server_storage.get_all_streams()
self.assertEqual(1, len(streams))
stream_info = yield self.server_lbry_file_manager.storage.get_stream_info(self.stream_hash)
stream_info = yield self.server_storage.get_stream_info(self.stream_hash)
self.assertEqual(self.sd_hash, stream_info[3])
self.assertEqual('test_file'.encode('hex'), stream_info[0])
# check should_announce blobs on blob_manager
blob_hashes = yield self.server_blob_manager.storage.get_all_should_announce_blobs()
blob_hashes = yield self.server_storage.get_all_should_announce_blobs()
self.assertSetEqual({self.sd_hash, expected_blob_hashes[0]}, set(blob_hashes))
def verify_have_blob(blob_hash, blob_size):
@ -211,7 +177,7 @@ class TestReflector(unittest.TestCase):
return d
def send_to_server():
factory = reflector.ClientFactory(self.session.blob_manager, self.stream_hash, self.sd_hash)
factory = reflector.ClientFactory(self.client_blob_manager, self.stream_hash, self.sd_hash)
from twisted.internet import reactor
reactor.connectTCP('localhost', self.port, factory)
@ -241,7 +207,7 @@ class TestReflector(unittest.TestCase):
def send_to_server(blob_hashes_to_send):
factory = reflector.BlobClientFactory(
self.session.blob_manager,
self.client_blob_manager,
blob_hashes_to_send
)
@ -261,10 +227,10 @@ class TestReflector(unittest.TestCase):
@defer.inlineCallbacks
def verify_stream_on_reflector():
# this protocol should not have any impact on stream info manager
streams = yield self.server_session.storage.get_all_streams()
streams = yield self.server_storage.get_all_streams()
self.assertEqual(0, len(streams))
# there should be no should announce blobs here
blob_hashes = yield self.server_blob_manager.storage.get_all_should_announce_blobs()
blob_hashes = yield self.server_storage.get_all_should_announce_blobs()
self.assertEqual(0, len(blob_hashes))
def verify_data_on_reflector():
@ -280,7 +246,7 @@ class TestReflector(unittest.TestCase):
def send_to_server(blob_hashes_to_send):
factory = reflector.BlobClientFactory(
self.session.blob_manager,
self.client_blob_manager,
blob_hashes_to_send
)
factory.protocol_version = 0
@ -311,20 +277,20 @@ class TestReflector(unittest.TestCase):
def verify_stream_on_reflector():
# check stream_info_manager has all the right information
streams = yield self.server_session.storage.get_all_streams()
streams = yield self.server_storage.get_all_streams()
self.assertEqual(1, len(streams))
self.assertEqual(self.stream_hash, streams[0])
blobs = yield self.server_session.storage.get_blobs_for_stream(self.stream_hash)
blobs = yield self.server_storage.get_blobs_for_stream(self.stream_hash)
blob_hashes = [b.blob_hash for b in blobs if b.blob_hash is not None]
expected_blob_hashes = [b[0] for b in self.expected_blobs[:-1] if b[0] is not None]
self.assertEqual(expected_blob_hashes, blob_hashes)
sd_hash = yield self.server_session.storage.get_sd_blob_hash_for_stream(
sd_hash = yield self.server_storage.get_sd_blob_hash_for_stream(
self.stream_hash)
self.assertEqual(self.sd_hash, sd_hash)
# check should_announce blobs on blob_manager
to_announce = yield self.server_blob_manager.storage.get_all_should_announce_blobs()
to_announce = yield self.server_storage.get_all_should_announce_blobs()
self.assertSetEqual(set(to_announce), {self.sd_hash, expected_blob_hashes[0]})
def verify_have_blob(blob_hash, blob_size):
@ -334,7 +300,7 @@ class TestReflector(unittest.TestCase):
def send_to_server_as_blobs(blob_hashes_to_send):
factory = reflector.BlobClientFactory(
self.session.blob_manager,
self.client_blob_manager,
blob_hashes_to_send
)
factory.protocol_version = 0
@ -344,7 +310,7 @@ class TestReflector(unittest.TestCase):
return factory.finished_deferred
def send_to_server_as_stream(result):
factory = reflector.ClientFactory(self.session.blob_manager, self.stream_hash, self.sd_hash)
factory = reflector.ClientFactory(self.client_blob_manager, self.stream_hash, self.sd_hash)
from twisted.internet import reactor
reactor.connectTCP('localhost', self.port, factory)

75
lbrynet/tests/functional/test_streamify.py
View file

@ -1,21 +1,18 @@
import os
import shutil
import tempfile
from hashlib import md5
from twisted.trial.unittest import TestCase
from twisted.internet import defer, threads
from lbrynet import conf
from lbrynet.file_manager.EncryptedFileManager import EncryptedFileManager
from lbrynet.core.Session import Session
from lbrynet.core.StreamDescriptor import StreamDescriptorIdentifier
from lbrynet.file_manager.EncryptedFileCreator import create_lbry_file
from lbrynet.lbry_file.client.EncryptedFileOptions import add_lbry_file_to_sd_identifier
from lbrynet.core.BlobManager import DiskBlobManager
from lbrynet.core.StreamDescriptor import get_sd_info
from lbrynet.core.PeerManager import PeerManager
from lbrynet.core.RateLimiter import DummyRateLimiter
from lbrynet.database.storage import SQLiteStorage
from lbrynet.core.PaymentRateManager import OnlyFreePaymentsManager
from lbrynet.file_manager.EncryptedFileCreator import create_lbry_file
from lbrynet.file_manager.EncryptedFileManager import EncryptedFileManager
from lbrynet.tests import mocks
@ -38,47 +35,42 @@ class TestStreamify(TestCase):
self.is_generous = True
self.db_dir = tempfile.mkdtemp()
self.blob_dir = os.path.join(self.db_dir, "blobfiles")
os.mkdir(self.blob_dir)
self.dht_node = FakeNode()
self.wallet = FakeWallet()
self.peer_manager = PeerManager()
self.peer_finder = FakePeerFinder(5553, self.peer_manager, 2)
self.rate_limiter = DummyRateLimiter()
self.sd_identifier = StreamDescriptorIdentifier()
os.mkdir(self.blob_dir)
self.storage = SQLiteStorage(self.db_dir)
self.blob_manager = DiskBlobManager(self.blob_dir, self.storage, self.dht_node._dataStore)
self.prm = OnlyFreePaymentsManager()
self.lbry_file_manager = EncryptedFileManager(
self.peer_finder, self.rate_limiter, self.blob_manager, self.wallet, self.prm, self.storage,
self.sd_identifier
)
d = self.storage.setup()
d.addCallback(lambda _: self.lbry_file_manager.setup())
return d
@defer.inlineCallbacks
def tearDown(self):
lbry_files = self.lbry_file_manager.lbry_files
for lbry_file in lbry_files:
yield self.lbry_file_manager.delete_lbry_file(lbry_file)
if self.lbry_file_manager is not None:
yield self.lbry_file_manager.stop()
if self.session is not None:
yield self.session.shut_down()
yield self.session.storage.stop()
yield self.storage.stop()
yield threads.deferToThread(shutil.rmtree, self.db_dir)
if os.path.exists("test_file"):
os.remove("test_file")
def test_create_stream(self):
self.session = Session(
conf.ADJUSTABLE_SETTINGS['data_rate'][1], db_dir=self.db_dir, node_id="abcd", peer_finder=self.peer_finder,
blob_dir=self.blob_dir, peer_port=5553, rate_limiter=self.rate_limiter, wallet=self.wallet,
external_ip="127.0.0.1", dht_node=self.dht_node
)
self.lbry_file_manager = EncryptedFileManager(self.session, self.sd_identifier)
d = self.session.setup()
d.addCallback(lambda _: add_lbry_file_to_sd_identifier(self.sd_identifier))
d.addCallback(lambda _: self.lbry_file_manager.setup())
def verify_equal(sd_info):
self.assertEqual(sd_info, test_create_stream_sd_file)
def verify_stream_descriptor_file(stream_hash):
d = get_sd_info(self.session.storage, stream_hash, True)
d = get_sd_info(self.storage, stream_hash, True)
d.addCallback(verify_equal)
return d
@ -90,39 +82,26 @@ class TestStreamify(TestCase):
def create_stream():
test_file = GenFile(5209343, b''.join([chr(i + 3) for i in xrange(0, 64, 6)]))
d = create_lbry_file(self.session, self.lbry_file_manager, "test_file", test_file,
key="0123456701234567", iv_generator=iv_generator())
d = create_lbry_file(
self.blob_manager, self.storage, self.prm, self.lbry_file_manager, "test_file", test_file,
key="0123456701234567", iv_generator=iv_generator()
)
d.addCallback(lambda lbry_file: lbry_file.stream_hash)
return d
d.addCallback(lambda _: create_stream())
d = create_stream()
d.addCallback(verify_stream_descriptor_file)
return d
def test_create_and_combine_stream(self):
self.session = Session(
conf.ADJUSTABLE_SETTINGS['data_rate'][1], db_dir=self.db_dir, node_id="abcd", peer_finder=self.peer_finder,
blob_dir=self.blob_dir, peer_port=5553, rate_limiter=self.rate_limiter, wallet=self.wallet,
external_ip="127.0.0.1", dht_node=self.dht_node
)
self.lbry_file_manager = EncryptedFileManager(self.session, self.sd_identifier)
@defer.inlineCallbacks
def create_stream():
def test_create_and_combine_stream(self):
test_file = GenFile(53209343, b''.join([chr(i + 5) for i in xrange(0, 64, 6)]))
lbry_file = yield create_lbry_file(self.session, self.lbry_file_manager, "test_file", test_file)
sd_hash = yield self.session.storage.get_sd_blob_hash_for_stream(lbry_file.stream_hash)
lbry_file = yield create_lbry_file(self.blob_manager, self.storage, self.prm, self.lbry_file_manager,
"test_file", test_file)
sd_hash = yield self.storage.get_sd_blob_hash_for_stream(lbry_file.stream_hash)
self.assertTrue(lbry_file.sd_hash, sd_hash)
yield lbry_file.start()
f = open('test_file')
hashsum = md5()
hashsum.update(f.read())
self.assertEqual(hashsum.hexdigest(), "68959747edc73df45e45db6379dd7b3b")
d = self.session.setup()
d.addCallback(lambda _: add_lbry_file_to_sd_identifier(self.sd_identifier))
d.addCallback(lambda _: self.lbry_file_manager.setup())
d.addCallback(lambda _: create_stream())
return d

11
lbrynet/tests/mocks.py
View file

@ -414,8 +414,8 @@ class FakeDelayedWallet(FakeComponent):
return d
class FakeDelayedSession(FakeComponent):
component_name = "session"
class FakeDelayedBlobManager(FakeComponent):
component_name = "blob_manager"
depends_on = [FakeDelayedWallet.component_name]
def start(self):
@ -431,7 +431,7 @@ class FakeDelayedSession(FakeComponent):
class FakeDelayedFileManager(FakeComponent):
component_name = "file_manager"
depends_on = [FakeDelayedSession.component_name]
depends_on = [FakeDelayedBlobManager.component_name]
def start(self):
d = defer.Deferred()
@ -441,6 +441,7 @@ class FakeDelayedFileManager(FakeComponent):
def stop(self):
return defer.succeed(True)
class FakeFileManager(FakeComponent):
component_name = "file_manager"
depends_on = []
@ -455,6 +456,10 @@ class FakeFileManager(FakeComponent):
def stop(self):
pass
def get_status(self):
return {}
create_stream_sd_file = {
'stream_name': '746573745f66696c65',
'blobs': [

53
lbrynet/tests/unit/components/test_Component_Manager.py
View file

@ -5,6 +5,7 @@ from lbrynet.daemon.ComponentManager import ComponentManager
from lbrynet.daemon.Components import DATABASE_COMPONENT, DHT_COMPONENT, STREAM_IDENTIFIER_COMPONENT
from lbrynet.daemon.Components import HASH_ANNOUNCER_COMPONENT, REFLECTOR_COMPONENT, UPNP_COMPONENT
from lbrynet.daemon.Components import PEER_PROTOCOL_SERVER_COMPONENT, EXCHANGE_RATE_MANAGER_COMPONENT
from lbrynet.daemon.Components import RATE_LIMITER_COMPONENT, HEADERS_COMPONENT, PAYMENT_RATE_COMPONENT
from lbrynet.daemon import Components
from lbrynet.tests import mocks
@ -13,17 +14,32 @@ class TestComponentManager(unittest.TestCase):
def setUp(self):
mocks.mock_conf_settings(self)
self.default_components_sort = [
[Components.DatabaseComponent,
[
Components.HeadersComponent,
Components.DatabaseComponent,
Components.ExchangeRateManagerComponent,
Components.UPnPComponent],
[Components.DHTComponent,
Components.WalletComponent],
[Components.HashAnnouncerComponent],
[Components.SessionComponent],
[Components.PeerProtocolServerComponent,
Components.StreamIdentifierComponent],
[Components.FileManagerComponent],
[Components.ReflectorComponent]
Components.PaymentRateComponent,
Components.RateLimiterComponent,
Components.UPnPComponent
],
[
Components.DHTComponent,
Components.WalletComponent
],
[
Components.BlobComponent,
Components.HashAnnouncerComponent
],
[
Components.PeerProtocolServerComponent,
Components.StreamIdentifierComponent
],
[
Components.FileManagerComponent
],
[
Components.ReflectorComponent
]
]
self.component_manager = ComponentManager()
@ -87,11 +103,12 @@ class TestComponentManagerProperStart(unittest.TestCase):
self.component_manager = ComponentManager(
skip_components=[DATABASE_COMPONENT, DHT_COMPONENT, HASH_ANNOUNCER_COMPONENT, STREAM_IDENTIFIER_COMPONENT,
PEER_PROTOCOL_SERVER_COMPONENT, REFLECTOR_COMPONENT, UPNP_COMPONENT,
HEADERS_COMPONENT, PAYMENT_RATE_COMPONENT, RATE_LIMITER_COMPONENT,
EXCHANGE_RATE_MANAGER_COMPONENT],
reactor=self.reactor,
wallet=mocks.FakeDelayedWallet,
session=mocks.FakeDelayedSession,
file_manager=mocks.FakeDelayedFileManager
file_manager=mocks.FakeDelayedFileManager,
blob_manager=mocks.FakeDelayedBlobManager
)
def tearDown(self):
@ -100,17 +117,17 @@ class TestComponentManagerProperStart(unittest.TestCase):
def test_proper_starting_of_components(self):
self.component_manager.setup()
self.assertTrue(self.component_manager.get_component('wallet').running)
self.assertFalse(self.component_manager.get_component('session').running)
self.assertFalse(self.component_manager.get_component('blob_manager').running)
self.assertFalse(self.component_manager.get_component('file_manager').running)
self.reactor.advance(1)
self.assertTrue(self.component_manager.get_component('wallet').running)
self.assertTrue(self.component_manager.get_component('session').running)
self.assertTrue(self.component_manager.get_component('blob_manager').running)
self.assertFalse(self.component_manager.get_component('file_manager').running)
self.reactor.advance(1)
self.assertTrue(self.component_manager.get_component('wallet').running)
self.assertTrue(self.component_manager.get_component('session').running)
self.assertTrue(self.component_manager.get_component('blob_manager').running)
self.assertTrue(self.component_manager.get_component('file_manager').running)
def test_proper_stopping_of_components(self):
@ -119,15 +136,15 @@ class TestComponentManagerProperStart(unittest.TestCase):
self.reactor.advance(1)
self.component_manager.stop()
self.assertFalse(self.component_manager.get_component('file_manager').running)
self.assertTrue(self.component_manager.get_component('session').running)
self.assertTrue(self.component_manager.get_component('blob_manager').running)
self.assertTrue(self.component_manager.get_component('wallet').running)
self.reactor.advance(1)
self.assertFalse(self.component_manager.get_component('file_manager').running)
self.assertFalse(self.component_manager.get_component('session').running)
self.assertFalse(self.component_manager.get_component('blob_manager').running)
self.assertTrue(self.component_manager.get_component('wallet').running)
self.reactor.advance(1)
self.assertFalse(self.component_manager.get_component('file_manager').running)
self.assertFalse(self.component_manager.get_component('session').running)
self.assertFalse(self.component_manager.get_component('blob_manager').running)
self.assertFalse(self.component_manager.get_component('wallet').running)

29
lbrynet/tests/unit/database/test_SQLiteStorage.py
View file

@ -7,9 +7,7 @@ from twisted.internet import defer
from twisted.trial import unittest
from lbrynet import conf
from lbrynet.database.storage import SQLiteStorage, open_file_for_writing
from lbrynet.core.StreamDescriptor import StreamDescriptorIdentifier
from lbrynet.file_manager.EncryptedFileDownloader import ManagedEncryptedFileDownloader
from lbrynet.file_manager.EncryptedFileManager import EncryptedFileManager
from lbrynet.tests.util import random_lbry_hash
log = logging.getLogger()
@ -67,7 +65,6 @@ fake_claim_info = {
}
class FakeAnnouncer(object):
def __init__(self):
self._queue_size = 0
@ -245,12 +242,8 @@ class FileStorageTests(StorageTest):
@defer.inlineCallbacks
def test_store_file(self):
session = MocSession(self.storage)
session.db_dir = self.db_dir
sd_identifier = StreamDescriptorIdentifier()
download_directory = self.db_dir
manager = EncryptedFileManager(session, sd_identifier)
out = yield manager.session.storage.get_all_lbry_files()
out = yield self.storage.get_all_lbry_files()
self.assertEqual(len(out), 0)
stream_hash = random_lbry_hash()
@ -268,33 +261,29 @@ class FileStorageTests(StorageTest):
blob_data_rate = 0
file_name = "test file"
out = yield manager.session.storage.save_published_file(
out = yield self.storage.save_published_file(
stream_hash, file_name, download_directory, blob_data_rate
)
rowid = yield manager.session.storage.get_rowid_for_stream_hash(stream_hash)
rowid = yield self.storage.get_rowid_for_stream_hash(stream_hash)
self.assertEqual(out, rowid)
files = yield manager.session.storage.get_all_lbry_files()
files = yield self.storage.get_all_lbry_files()
self.assertEqual(1, len(files))
status = yield manager.session.storage.get_lbry_file_status(rowid)
status = yield self.storage.get_lbry_file_status(rowid)
self.assertEqual(status, ManagedEncryptedFileDownloader.STATUS_STOPPED)
running = ManagedEncryptedFileDownloader.STATUS_RUNNING
yield manager.session.storage.change_file_status(rowid, running)
status = yield manager.session.storage.get_lbry_file_status(rowid)
yield self.storage.change_file_status(rowid, running)
status = yield self.storage.get_lbry_file_status(rowid)
self.assertEqual(status, ManagedEncryptedFileDownloader.STATUS_RUNNING)
class ContentClaimStorageTests(StorageTest):
@defer.inlineCallbacks
def test_store_content_claim(self):
session = MocSession(self.storage)
session.db_dir = self.db_dir
sd_identifier = StreamDescriptorIdentifier()
download_directory = self.db_dir
manager = EncryptedFileManager(session, sd_identifier)
out = yield manager.session.storage.get_all_lbry_files()
out = yield self.storage.get_all_lbry_files()
self.assertEqual(len(out), 0)
stream_hash = random_lbry_hash()
@ -307,7 +296,7 @@ class ContentClaimStorageTests(StorageTest):
yield self.make_and_store_fake_stream(blob_count=2, stream_hash=stream_hash, sd_hash=sd_hash)
blob_data_rate = 0
file_name = "test file"
yield manager.session.storage.save_published_file(
yield self.storage.save_published_file(
stream_hash, file_name, download_directory, blob_data_rate
)
yield self.storage.save_claims([fake_claim_info])

56
lbrynet/tests/unit/lbryfilemanager/test_EncryptedFileCreator.py
View file

@ -1,18 +1,29 @@
# -*- coding: utf-8 -*-
from cryptography.hazmat.primitives.ciphers.algorithms import AES
import mock
from twisted.trial import unittest
from twisted.internet import defer
from lbrynet.database.storage import SQLiteStorage
from lbrynet.core.StreamDescriptor import get_sd_info, BlobStreamDescriptorReader
from lbrynet.core import BlobManager
from lbrynet.core import Session
from lbrynet.core.StreamDescriptor import StreamDescriptorIdentifier
from lbrynet.core.BlobManager import DiskBlobManager
from lbrynet.core.PeerManager import PeerManager
from lbrynet.core.RateLimiter import DummyRateLimiter
from lbrynet.core.PaymentRateManager import OnlyFreePaymentsManager
from lbrynet.database.storage import SQLiteStorage
from lbrynet.file_manager import EncryptedFileCreator
from lbrynet.file_manager import EncryptedFileManager
from lbrynet.file_manager.EncryptedFileManager import EncryptedFileManager
from lbrynet.tests import mocks
from lbrynet.tests.util import mk_db_and_blob_dir, rm_db_and_blob_dir
FakeNode = mocks.Node
FakeWallet = mocks.Wallet
FakePeerFinder = mocks.PeerFinder
FakeAnnouncer = mocks.Announcer
GenFile = mocks.GenFile
test_create_stream_sd_file = mocks.create_stream_sd_file
DummyBlobAvailabilityTracker = mocks.BlobAvailabilityTracker
MB = 2**20
@ -24,32 +35,37 @@ def iv_generator():
class CreateEncryptedFileTest(unittest.TestCase):
timeout = 5
@defer.inlineCallbacks
def setUp(self):
mocks.mock_conf_settings(self)
self.tmp_db_dir, self.tmp_blob_dir = mk_db_and_blob_dir()
self.session = mock.Mock(spec=Session.Session)(None, None)
self.session.payment_rate_manager.min_blob_data_payment_rate = 0
self.blob_manager = BlobManager.DiskBlobManager(self.tmp_blob_dir, SQLiteStorage(self.tmp_db_dir))
self.session.blob_manager = self.blob_manager
self.session.storage = self.session.blob_manager.storage
self.file_manager = EncryptedFileManager.EncryptedFileManager(self.session, object())
yield self.session.blob_manager.storage.setup()
yield self.session.blob_manager.setup()
self.wallet = FakeWallet()
self.peer_manager = PeerManager()
self.peer_finder = FakePeerFinder(5553, self.peer_manager, 2)
self.rate_limiter = DummyRateLimiter()
self.sd_identifier = StreamDescriptorIdentifier()
self.storage = SQLiteStorage(self.tmp_db_dir)
self.blob_manager = DiskBlobManager(self.tmp_blob_dir, self.storage)
self.prm = OnlyFreePaymentsManager()
self.lbry_file_manager = EncryptedFileManager(self.peer_finder, self.rate_limiter, self.blob_manager,
self.wallet, self.prm, self.storage, self.sd_identifier)
d = self.storage.setup()
d.addCallback(lambda _: self.lbry_file_manager.setup())
return d
@defer.inlineCallbacks
def tearDown(self):
yield self.lbry_file_manager.stop()
yield self.blob_manager.stop()
yield self.session.storage.stop()
yield self.storage.stop()
rm_db_and_blob_dir(self.tmp_db_dir, self.tmp_blob_dir)
@defer.inlineCallbacks
def create_file(self, filename):
handle = mocks.GenFile(3*MB, '1')
key = '2' * (AES.block_size / 8)
out = yield EncryptedFileCreator.create_lbry_file(self.session, self.file_manager, filename, handle,
key, iv_generator())
out = yield EncryptedFileCreator.create_lbry_file(
self.blob_manager, self.storage, self.prm, self.lbry_file_manager, filename, handle, key, iv_generator()
)
defer.returnValue(out)
@defer.inlineCallbacks
@ -60,7 +76,7 @@ class CreateEncryptedFileTest(unittest.TestCase):
"c8728fe0534dd06fbcacae92b0891787ad9b68ffc8d20c1"
filename = 'test.file'
lbry_file = yield self.create_file(filename)
sd_hash = yield self.session.storage.get_sd_blob_hash_for_stream(lbry_file.stream_hash)
sd_hash = yield self.storage.get_sd_blob_hash_for_stream(lbry_file.stream_hash)
# read the sd blob file
sd_blob = self.blob_manager.blobs[sd_hash]
@ -68,7 +84,7 @@ class CreateEncryptedFileTest(unittest.TestCase):
sd_file_info = yield sd_reader.get_info()
# this comes from the database, the blobs returned are sorted
sd_info = yield get_sd_info(self.session.storage, lbry_file.stream_hash, include_blobs=True)
sd_info = yield get_sd_info(self.storage, lbry_file.stream_hash, include_blobs=True)
self.assertDictEqual(sd_info, sd_file_info)
self.assertListEqual(sd_info['blobs'], sd_file_info['blobs'])
self.assertEqual(sd_info['stream_hash'], expected_stream_hash)

65
lbrynet/tests/unit/lbrynet_daemon/test_Daemon.py
View file

@ -11,18 +11,18 @@ from faker import Faker
from lbryschema.decode import smart_decode
from lbryum.wallet import NewWallet
from lbrynet import conf
from lbrynet.core import Session, PaymentRateManager, Wallet
from lbrynet.core import Wallet
from lbrynet.database.storage import SQLiteStorage
from lbrynet.daemon.ComponentManager import ComponentManager
from lbrynet.daemon.Components import DATABASE_COMPONENT, DHT_COMPONENT, WALLET_COMPONENT, STREAM_IDENTIFIER_COMPONENT
from lbrynet.daemon.Components import HASH_ANNOUNCER_COMPONENT, REFLECTOR_COMPONENT, UPNP_COMPONENT, SESSION_COMPONENT
from lbrynet.daemon.Components import HASH_ANNOUNCER_COMPONENT, REFLECTOR_COMPONENT, UPNP_COMPONENT, BLOB_COMPONENT
from lbrynet.daemon.Components import PEER_PROTOCOL_SERVER_COMPONENT, EXCHANGE_RATE_MANAGER_COMPONENT
from lbrynet.daemon.Components import RATE_LIMITER_COMPONENT, HEADERS_COMPONENT, FILE_MANAGER_COMPONENT
from lbrynet.daemon.Daemon import Daemon as LBRYDaemon
from lbrynet.file_manager.EncryptedFileDownloader import ManagedEncryptedFileDownloader
from lbrynet.core.PaymentRateManager import OnlyFreePaymentsManager
from lbrynet.tests import util
from lbrynet.tests.mocks import mock_conf_settings, FakeNetwork, FakeFileManager
from lbrynet.tests.mocks import BlobAvailabilityTracker as DummyBlobAvailabilityTracker
from lbrynet.tests.mocks import ExchangeRateManager as DummyExchangeRateManager
from lbrynet.tests.mocks import BTCLBCFeed, USDBTCFeed
from lbrynet.tests.util import is_android
@ -40,19 +40,23 @@ def get_test_daemon(data_rate=None, generous=True, with_fee=False):
'BTCLBC': {'spot': 3.0, 'ts': util.DEFAULT_ISO_TIME + 1},
'USDBTC': {'spot': 2.0, 'ts': util.DEFAULT_ISO_TIME + 2}
}
daemon = LBRYDaemon(None)
daemon.session = mock.Mock(spec=Session.Session)
component_manager = ComponentManager(
skip_components=[DATABASE_COMPONENT, DHT_COMPONENT, WALLET_COMPONENT, UPNP_COMPONENT,
PEER_PROTOCOL_SERVER_COMPONENT, REFLECTOR_COMPONENT, HASH_ANNOUNCER_COMPONENT,
STREAM_IDENTIFIER_COMPONENT, EXCHANGE_RATE_MANAGER_COMPONENT, BLOB_COMPONENT,
HEADERS_COMPONENT, RATE_LIMITER_COMPONENT],
file_manager=FakeFileManager
)
daemon = LBRYDaemon(component_manager=component_manager)
daemon.payment_rate_manager = OnlyFreePaymentsManager()
daemon.wallet = mock.Mock(spec=Wallet.LBRYumWallet)
daemon.wallet.wallet = mock.Mock(spec=NewWallet)
daemon.wallet.wallet.use_encryption = False
daemon.wallet.network = FakeNetwork()
daemon.session.storage = mock.Mock(spec=SQLiteStorage)
daemon.storage = mock.Mock(spec=SQLiteStorage)
market_feeds = [BTCLBCFeed(), USDBTCFeed()]
daemon.exchange_rate_manager = DummyExchangeRateManager(market_feeds, rates)
base_prm = PaymentRateManager.BasePaymentRateManager(rate=data_rate)
prm = PaymentRateManager.NegotiatedPaymentRateManager(base_prm, DummyBlobAvailabilityTracker(),
generous=generous)
daemon.session.payment_rate_manager = prm
daemon.file_manager = component_manager.get_component(FILE_MANAGER_COMPONENT)
metadata = {
"author": "fake author",
@ -91,26 +95,26 @@ class TestCostEst(unittest.TestCase):
daemon = get_test_daemon(generous=True, with_fee=True)
self.assertEquals(daemon.get_est_cost("test", size).result, correct_result)
def test_fee_and_ungenerous_data(self):
size = 10000000
fake_fee_amount = 4.5
data_rate = conf.ADJUSTABLE_SETTINGS['data_rate'][1]
correct_result = size / 10 ** 6 * data_rate + fake_fee_amount
daemon = get_test_daemon(generous=False, with_fee=True)
self.assertEquals(daemon.get_est_cost("test", size).result, correct_result)
# def test_fee_and_ungenerous_data(self):
# size = 10000000
# fake_fee_amount = 4.5
# data_rate = conf.ADJUSTABLE_SETTINGS['data_rate'][1]
# correct_result = size / 10 ** 6 * data_rate + fake_fee_amount
# daemon = get_test_daemon(generous=False, with_fee=True)
# self.assertEquals(daemon.get_est_cost("test", size).result, correct_result)
def test_generous_data_and_no_fee(self):
size = 10000000
correct_result = 0.0
daemon = get_test_daemon(generous=True)
self.assertEquals(daemon.get_est_cost("test", size).result, correct_result)
def test_ungenerous_data_and_no_fee(self):
size = 10000000
data_rate = conf.ADJUSTABLE_SETTINGS['data_rate'][1]
correct_result = size / 10 ** 6 * data_rate
daemon = get_test_daemon(generous=False)
self.assertEquals(daemon.get_est_cost("test", size).result, correct_result)
#
# def test_ungenerous_data_and_no_fee(self):
# size = 10000000
# data_rate = conf.ADJUSTABLE_SETTINGS['data_rate'][1]
# correct_result = size / 10 ** 6 * data_rate
# daemon = get_test_daemon(generous=False)
# self.assertEquals(daemon.get_est_cost("test", size).result, correct_result)
class TestJsonRpc(unittest.TestCase):
@ -145,17 +149,7 @@ class TestFileListSorting(unittest.TestCase):
self.faker = Faker('en_US')
self.faker.seed(66410)
self.test_daemon = get_test_daemon()
component_manager = ComponentManager(
skip_components=[DATABASE_COMPONENT, DHT_COMPONENT, WALLET_COMPONENT, SESSION_COMPONENT, UPNP_COMPONENT,
PEER_PROTOCOL_SERVER_COMPONENT, REFLECTOR_COMPONENT, HASH_ANNOUNCER_COMPONENT,
STREAM_IDENTIFIER_COMPONENT, EXCHANGE_RATE_MANAGER_COMPONENT],
file_manager=FakeFileManager
)
component_manager.setup()
self.test_daemon.component_manager = component_manager
self.test_daemon.file_manager = component_manager.get_component("file_manager")
self.test_daemon.file_manager.lbry_files = self._get_fake_lbry_files()
# Pre-sorted lists of prices and file names in ascending order produced by
# faker with seed 66410. This seed was chosen becacuse it produces 3 results
# 'points_paid' at 6.0 and 2 results at 4.5 to test multiple sort criteria.
@ -166,6 +160,7 @@ class TestFileListSorting(unittest.TestCase):
self.test_authors = ['angela41', 'edward70', 'fhart', 'johnrosales',
'lucasfowler', 'peggytorres', 'qmitchell',
'trevoranderson', 'xmitchell', 'zhangsusan']
return self.test_daemon.component_manager.setup()
def test_sort_by_points_paid_no_direction_specified(self):
sort_options = ['points_paid']

28
lbrynet/tests/unit/lbrynet_daemon/test_Downloader.py
View file

@ -3,16 +3,18 @@ import mock
from twisted.trial import unittest
from twisted.internet import defer, task
from lbrynet.core import Session, PaymentRateManager, Wallet
from lbrynet.core import PaymentRateManager, Wallet
from lbrynet.core.Error import DownloadDataTimeout, DownloadSDTimeout
from lbrynet.daemon import Downloader
from lbrynet.core.StreamDescriptor import StreamDescriptorIdentifier
from lbrynet.database.storage import SQLiteStorage
from lbrynet.core.BlobManager import DiskBlobManager
from lbrynet.dht.peerfinder import DummyPeerFinder
from lbrynet.core.RateLimiter import DummyRateLimiter
from lbrynet.file_manager.EncryptedFileStatusReport import EncryptedFileStatusReport
from lbrynet.file_manager.EncryptedFileDownloader import ManagedEncryptedFileDownloader
from lbrynet.daemon.ExchangeRateManager import ExchangeRateManager
from lbrynet.tests.mocks import ExchangeRateManager as DummyExchangeRateManager
from lbrynet.tests.mocks import mock_conf_settings
@ -61,25 +63,22 @@ def moc_pay_key_fee(self, key_fee, name):
class GetStreamTests(unittest.TestCase):
def init_getstream_with_mocs(self):
mock_conf_settings(self)
sd_identifier = mock.Mock(spec=StreamDescriptorIdentifier)
session = mock.Mock(spec=Session.Session)
session.wallet = mock.Mock(spec=Wallet.LBRYumWallet)
wallet = mock.Mock(spec=Wallet.LBRYumWallet)
prm = mock.Mock(spec=PaymentRateManager.NegotiatedPaymentRateManager)
session.payment_rate_manager = prm
market_feeds = []
rates = {}
exchange_rate_manager = DummyExchangeRateManager(market_feeds, rates)
exchange_rate_manager = mock.Mock(spec=ExchangeRateManager)
storage = mock.Mock(spec=SQLiteStorage)
peer_finder = DummyPeerFinder()
blob_manager = mock.Mock(spec=DiskBlobManager)
max_key_fee = {'currency': "LBC", 'amount': 10, 'address': ''}
disable_max_key_fee = False
data_rate = {'currency': "LBC", 'amount': 0, 'address': ''}
getstream = Downloader.GetStream(sd_identifier, session,
exchange_rate_manager, max_key_fee, disable_max_key_fee, timeout=3, data_rate=data_rate)
getstream = Downloader.GetStream(
sd_identifier, wallet, exchange_rate_manager, blob_manager, peer_finder, DummyRateLimiter(), prm,
storage, max_key_fee, disable_max_key_fee, timeout=3, data_rate=data_rate
)
getstream.pay_key_fee_called = False
self.clock = task.Clock()
@ -100,7 +99,6 @@ class GetStreamTests(unittest.TestCase):
with self.assertRaises(AttributeError):
yield getstream.start(stream_info, name, "deadbeef" * 12, 0)
@defer.inlineCallbacks
def test_sd_blob_download_timeout(self):
"""

3
setup.py
View file

@ -24,13 +24,14 @@ requires = [
'lbryschema==0.0.16',
'lbryum==3.2.3',
'miniupnpc',
'txupnp==0.0.1a6',
'pyyaml',
'requests',
'txJSON-RPC',
'zope.interface',
'treq',
'docopt',
'six'
'six',
]
console_scripts = [