Revert "refactor lbrynet-daemon into modular components" (#1286)

* Revert "fix reflector test"

This reverts commit 6a15b51ac3.

* Revert "refactor lbrynet-daemon into modular components (#1164)"

This reverts commit 75a6ff269e.

CHANGELOG.md

@@ -23,19 +23,14 @@ at anytime.
 ### Changed
   *
   *
-  * `publish` to accept bid as a decimal string
-  * all JSONRPC API commands are now registered asynchronously and are available to be called as soon as they are ready
 
 ### Added
-  * Component Manager for managing the dependencies, startup and stopping of components 
+  *
-  * `requires` decorator to register the components required by a `jsonrpc_` command, to facilitate commands registering asynchronously
-  * added unittests for Component Manager
   *
 
 ### Removed
-  * `STARTUP_STAGES` from `status` API and CLI call, it instead returns a dictionary of components along with their running status(this is a **potentially breaking change** if `STARTUP_STAGES` is relied upon)
+  *
-  * all component startup code from `Daemon.py`
+  *
-  * wallet, upnp and dht startup code from `session.py`, the code now resides in `Components.py`
 
 
 ## [0.20.3] - 2018-07-03

lbrynet/conf.py

@@ -168,11 +168,9 @@ def server_port(server_and_port):
 def server_list(servers):
     return [server_port(server) for server in servers]
 
-
 def server_list_reverse(servers):
     return ["%s:%s" % (server, port) for server, port in servers]
 
-
 class Env(envparse.Env):
     """An Env parser that automatically namespaces the variables with LBRY"""
 
@@ -301,8 +299,7 @@ ADJUSTABLE_SETTINGS = {
     'blockchain_name': (str, 'lbrycrd_main'),
     'lbryum_servers': (list, [('lbryumx1.lbry.io', 50001), ('lbryumx2.lbry.io',
         50001)], server_list, server_list_reverse),
-    's3_headers_depth': (int, 96 * 10),   # download headers from s3 when the local height is more than 10 chunks behind
+    's3_headers_depth': (int, 96 * 10)   # download headers from s3 when the local height is more than 10 chunks behind
-    'components_to_skip': (list, [])  # components which will be skipped during start-up of daemon
 }
 
 

lbrynet/core/Error.py

@@ -155,23 +155,13 @@ class InvalidAuthenticationToken(Exception):
 class NegotiationError(Exception):
     pass
 
-
 class InvalidCurrencyError(Exception):
     def __init__(self, currency):
         self.currency = currency
         Exception.__init__(
             self, 'Invalid currency: {} is not a supported currency.'.format(currency))
 
-
 class NoSuchDirectoryError(Exception):
     def __init__(self, directory):
         self.directory = directory
         Exception.__init__(self, 'No such directory {}'.format(directory))
-
-
-class ComponentStartConditionNotMet(Exception):
-    pass
-
-
-class ComponentsNotStarted(Exception):
-    pass

lbrynet/core/Session.py

@@ -1,8 +1,11 @@
 import logging
-from twisted.internet import defer
+import miniupnpc
+from twisted.internet import threads, defer
 from lbrynet.core.BlobManager import DiskBlobManager
+from lbrynet.dht import node, hashannouncer
 from lbrynet.database.storage import SQLiteStorage
 from lbrynet.core.RateLimiter import RateLimiter
+from lbrynet.core.utils import generate_id
 from lbrynet.core.PaymentRateManager import BasePaymentRateManager, OnlyFreePaymentsManager
 
 log = logging.getLogger(__name__)
@@ -29,11 +32,11 @@ class Session(object):
     peers can connect to this peer.
     """
 
-    def __init__(self, blob_data_payment_rate, db_dir=None, node_id=None, dht_node_port=None,
+    def __init__(self, blob_data_payment_rate, db_dir=None, node_id=None, peer_manager=None, dht_node_port=None,
                  known_dht_nodes=None, peer_finder=None, hash_announcer=None, blob_dir=None, blob_manager=None,
-                 peer_port=None, use_upnp=True, rate_limiter=None, wallet=None, blob_tracker_class=None,
+                 peer_port=None, use_upnp=True, rate_limiter=None, wallet=None, dht_node_class=node.Node,
-                 payment_rate_manager_class=None, is_generous=True, external_ip=None, storage=None,
+                 blob_tracker_class=None, payment_rate_manager_class=None, is_generous=True, external_ip=None,
-                 dht_node=None, peer_manager=None):
+                 storage=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
@@ -108,7 +111,8 @@ class Session(object):
         self.external_ip = external_ip
         self.upnp_redirects = []
         self.wallet = wallet
-        self.dht_node = dht_node
+        self.dht_node_class = dht_node_class
+        self.dht_node = None
         self.base_payment_rate_manager = BasePaymentRateManager(blob_data_payment_rate)
         self.payment_rate_manager = OnlyFreePaymentsManager()
         # self.payment_rate_manager_class = payment_rate_manager_class or NegotiatedPaymentRateManager
@@ -120,14 +124,15 @@ class Session(object):
 
         log.debug("Starting session.")
 
-        if self.dht_node is not None:
+        if self.node_id is None:
-            if self.peer_manager is None:
+            self.node_id = generate_id()
-                self.peer_manager = self.dht_node.peer_manager
 
-            if self.peer_finder is None:
+        if self.use_upnp is True:
-                self.peer_finder = self.dht_node.peer_finder
+            d = self._try_upnp()
-
+        else:
-        d = self.storage.setup()
+            d = defer.succeed(True)
+        d.addCallback(lambda _: self.storage.setup())
+        d.addCallback(lambda _: self._setup_dht())
         d.addCallback(lambda _: self._setup_other_components())
         return d
 
@@ -135,16 +140,97 @@ class Session(object):
         """Stop all services"""
         log.info('Stopping session.')
         ds = []
+        if self.hash_announcer:
+            self.hash_announcer.stop()
         # if self.blob_tracker is not None:
         #     ds.append(defer.maybeDeferred(self.blob_tracker.stop))
+        if self.dht_node is not None:
+            ds.append(defer.maybeDeferred(self.dht_node.stop))
         if self.rate_limiter is not None:
             ds.append(defer.maybeDeferred(self.rate_limiter.stop))
+        if self.wallet is not None:
+            ds.append(defer.maybeDeferred(self.wallet.stop))
         if self.blob_manager is not None:
             ds.append(defer.maybeDeferred(self.blob_manager.stop))
-        # if self.use_upnp is True:
+        if self.use_upnp is True:
-        #     ds.append(defer.maybeDeferred(self._unset_upnp))
+            ds.append(defer.maybeDeferred(self._unset_upnp))
         return defer.DeferredList(ds)
 
+    def _try_upnp(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
+
+    def _setup_dht(self):  # does not block startup, the dht will re-attempt if necessary
+        self.dht_node = self.dht_node_class(
+            node_id=self.node_id,
+            udpPort=self.dht_node_port,
+            externalIP=self.external_ip,
+            peerPort=self.peer_port,
+            peer_manager=self.peer_manager,
+            peer_finder=self.peer_finder,
+        )
+        if not self.hash_announcer:
+            self.hash_announcer = hashannouncer.DHTHashAnnouncer(self.dht_node, self.storage)
+        self.peer_manager = self.dht_node.peer_manager
+        self.peer_finder = self.dht_node.peer_finder
+        d = self.dht_node.start(self.known_dht_nodes)
+        d.addCallback(lambda _: log.info("Joined the dht"))
+        d.addCallback(lambda _: self.hash_announcer.start())
+
     def _setup_other_components(self):
         log.debug("Setting up the rest of the components")
 
@@ -169,4 +255,28 @@ class Session(object):
 
         self.rate_limiter.start()
         d = self.blob_manager.setup()
+        d.addCallback(lambda _: self.wallet.start())
+        # d.addCallback(lambda _: self.blob_tracker.start())
+        return d
+
+    def _unset_upnp(self):
+        log.info("Unsetting upnp for session")
+
+        def threaded_unset_upnp():
+            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

lbrynet/core/Wallet.py

@@ -938,7 +938,9 @@ class LBRYumWallet(Wallet):
         self._lag_counter = 0
         self.blocks_behind = 0
         self.catchup_progress = 0
-        self.is_wallet_unlocked = None
+
+        # fired when the wallet actually unlocks (wallet_unlocked_d can be called multiple times)
+        self.wallet_unlock_success = defer.Deferred()
 
     def _is_first_run(self):
         return (not self.printed_retrieving_headers and
@@ -951,23 +953,21 @@ class LBRYumWallet(Wallet):
         return self._cmd_runner
 
     def check_locked(self):
-        """
+        if not self.wallet.use_encryption:
-        Checks if the wallet is encrypted(locked) or not
+            log.info("Wallet is not encrypted")
-
+            self.wallet_unlock_success.callback(True)
-        :return: (boolean) indicating whether the wallet is locked or not
+        elif not self._cmd_runner:
-        """
-        if not self._cmd_runner:
             raise Exception("Command runner hasn't been initialized yet")
         elif self._cmd_runner.locked:
             log.info("Waiting for wallet password")
             self.wallet_unlocked_d.addCallback(self.unlock)
-        return self.is_wallet_unlocked
+        return self.wallet_unlock_success
 
     def unlock(self, password):
         if self._cmd_runner and self._cmd_runner.locked:
             try:
                 self._cmd_runner.unlock_wallet(password)
-                self.is_wallet_unlocked = True
+                self.wallet_unlock_success.callback(True)
                 log.info("Unlocked the wallet!")
             except InvalidPassword:
                 log.warning("Incorrect password, try again")
@@ -1054,7 +1054,6 @@ class LBRYumWallet(Wallet):
             wallet.create_main_account()
             wallet.synchronize()
         self.wallet = wallet
-        self.is_wallet_unlocked = not self.wallet.use_encryption
         self._check_large_wallet()
         return defer.succeed(True)
 

lbrynet/daemon/Component.py

@@ -1,63 +0,0 @@
-import logging
-from twisted.internet import defer
-from ComponentManager import ComponentManager
-
-log = logging.getLogger(__name__)
-
-
-class ComponentType(type):
-    def __new__(mcs, name, bases, newattrs):
-        klass = type.__new__(mcs, name, bases, newattrs)
-        if name != "Component":
-            ComponentManager.default_component_classes[klass.component_name] = klass
-        return klass
-
-
-class Component(object):
-    """
-    lbrynet-daemon component helper
-
-    Inheriting classes will be automatically registered with the ComponentManager and must implement setup and stop
-    methods
-    """
-
-    __metaclass__ = ComponentType
-    depends_on = []
-    component_name = None
-
-    def __init__(self, component_manager):
-        self.component_manager = component_manager
-        self._running = False
-
-    @property
-    def running(self):
-        return self._running
-
-    def start(self):
-        raise NotImplementedError()
-
-    def stop(self):
-        raise NotImplementedError()
-
-    def component(self):
-        raise NotImplementedError()
-
-    @defer.inlineCallbacks
-    def _setup(self):
-        try:
-            result = yield defer.maybeDeferred(self.start)
-            self._running = True
-            defer.returnValue(result)
-        except Exception as err:
-            log.exception("Error setting up %s", self.component_name or self.__class__.__name__)
-            raise err
-
-    @defer.inlineCallbacks
-    def _stop(self):
-        try:
-            result = yield defer.maybeDeferred(self.stop)
-            self._running = False
-            defer.returnValue(result)
-        except Exception as err:
-            log.exception("Error stopping %s", self.__class__.__name__)
-            raise err

lbrynet/daemon/ComponentManager.py

@@ -1,139 +0,0 @@
-import logging
-from twisted.internet import defer
-
-from lbrynet import conf
-from lbrynet.core.Error import ComponentStartConditionNotMet
-
-log = logging.getLogger(__name__)
-
-
-class ComponentManager(object):
-    default_component_classes = {}
-
-    def __init__(self, reactor=None, analytics_manager=None, skip_components=None, **override_components):
-        self.skip_components = skip_components or []
-        self.skip_components.extend(conf.settings['components_to_skip'])
-
-        self.reactor = reactor
-        self.component_classes = {}
-        self.components = set()
-        self.analytics_manager = analytics_manager
-
-        for component_name, component_class in self.default_component_classes.iteritems():
-            if component_name in override_components:
-                component_class = override_components.pop(component_name)
-            if component_name not in self.skip_components:
-                self.component_classes[component_name] = component_class
-
-        if override_components:
-            raise SyntaxError("unexpected components: %s" % override_components)
-
-        for component_class in self.component_classes.itervalues():
-            self.components.add(component_class(self))
-
-    def sort_components(self, reverse=False):
-        """
-        Sort components by requirements
-        """
-        steps = []
-        staged = set()
-        components = set(self.components)
-
-        # components with no requirements
-        step = []
-        for component in set(components):
-            if not component.depends_on:
-                step.append(component)
-                staged.add(component.component_name)
-                components.remove(component)
-
-        if step:
-            steps.append(step)
-
-        while components:
-            step = []
-            to_stage = set()
-            for component in set(components):
-                reqs_met = 0
-                for needed in component.depends_on:
-                    if needed in staged:
-                        reqs_met += 1
-                if reqs_met == len(component.depends_on):
-                    step.append(component)
-                    to_stage.add(component.component_name)
-                    components.remove(component)
-            if step:
-                staged.update(to_stage)
-                steps.append(step)
-            elif components:
-                raise ComponentStartConditionNotMet("Unresolved dependencies for: %s" % components)
-        if reverse:
-            steps.reverse()
-        return steps
-
-    @defer.inlineCallbacks
-    def setup(self, **callbacks):
-        """
-        Start Components in sequence sorted by requirements
-
-        :return: (defer.Deferred)
-        """
-
-        for component_name, cb in callbacks.iteritems():
-            if component_name not in self.component_classes:
-                raise NameError("unknown component: %s" % component_name)
-            if not callable(cb):
-                raise ValueError("%s is not callable" % cb)
-
-        def _setup(component):
-            if component.component_name in callbacks:
-                d = component._setup()
-                d.addCallback(callbacks[component.component_name])
-                return d
-            return component._setup()
-
-        stages = self.sort_components()
-        for stage in stages:
-            yield defer.DeferredList([_setup(component) for component in stage])
-
-    @defer.inlineCallbacks
-    def stop(self):
-        """
-        Stop Components in reversed startup order
-
-        :return: (defer.Deferred)
-        """
-        stages = self.sort_components(reverse=True)
-        for stage in stages:
-            yield defer.DeferredList([component._stop() for component in stage])
-
-    def all_components_running(self, *component_names):
-        """
-        Check if components are running
-
-        :return: (bool) True if all specified components are running
-        """
-        components = {component.component_name: component for component in self.components}
-        for component in component_names:
-            if component not in components:
-                raise NameError("%s is not a known Component" % component)
-            if not components[component].running:
-                return False
-        return True
-
-    def get_components_status(self):
-        """
-        List status of all the components, whether they are running or not
-
-        :return: (dict) {(str) component_name: (bool) True is running else False}
-        """
-        return {
-            component.component_name: component.running
-            for component in self.components
-        }
-
-    def get_component(self, component_name):
-        for component in self.components:
-            if component.component_name == component_name:
-                return component.component
-        raise NameError(component_name)

lbrynet/daemon/Components.py

@@ -1,523 +0,0 @@
-import os
-import logging
-import miniupnpc
-from twisted.internet import defer, threads, reactor, error
-
-from lbrynet import conf
-from lbrynet.core.Session import Session
-from lbrynet.core.StreamDescriptor import StreamDescriptorIdentifier, EncryptedFileStreamType
-from lbrynet.core.Wallet import LBRYumWallet
-from lbrynet.core.server.BlobRequestHandler import BlobRequestHandlerFactory
-from lbrynet.core.server.ServerProtocol import ServerProtocolFactory
-from lbrynet.daemon.Component import Component
-from lbrynet.database.storage import SQLiteStorage
-from lbrynet.dht import node, hashannouncer
-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.core.utils import generate_id
-
-log = logging.getLogger(__name__)
-
-# settings must be initialized before this file is imported
-
-DATABASE_COMPONENT = "database"
-WALLET_COMPONENT = "wallet"
-SESSION_COMPONENT = "session"
-DHT_COMPONENT = "dht"
-HASH_ANNOUNCER_COMPONENT = "hashAnnouncer"
-STREAM_IDENTIFIER_COMPONENT = "streamIdentifier"
-FILE_MANAGER_COMPONENT = "fileManager"
-PEER_PROTOCOL_SERVER_COMPONENT = "peerProtocolServer"
-REFLECTOR_COMPONENT = "reflector"
-UPNP_COMPONENT = "upnp"
-
-
-class ConfigSettings(object):
-    @staticmethod
-    def get_conf_setting(setting_name):
-        return conf.settings[setting_name]
-
-    @staticmethod
-    def get_blobfiles_dir():
-        if conf.settings['BLOBFILES_DIR'] == "blobfiles":
-            return os.path.join(GCS("data_dir"), "blobfiles")
-        else:
-            log.info("Using non-default blobfiles directory: %s", conf.settings['BLOBFILES_DIR'])
-            return conf.settings['BLOBFILES_DIR']
-
-    @staticmethod
-    def get_node_id():
-        return conf.settings.node_id
-
-    @staticmethod
-    def get_external_ip():
-        from lbrynet.core.system_info import get_platform
-        platform = get_platform(get_ip=True)
-        return platform['ip']
-
-
-# Shorthand for common ConfigSettings methods
-CS = ConfigSettings
-GCS = ConfigSettings.get_conf_setting
-
-
-class DatabaseComponent(Component):
-    component_name = DATABASE_COMPONENT
-
-    def __init__(self, component_manager):
-        Component.__init__(self, component_manager)
-        self.storage = None
-
-    @property
-    def component(self):
-        return self.storage
-
-    @staticmethod
-    def get_current_db_revision():
-        return 9
-
-    @staticmethod
-    def get_revision_filename():
-        return conf.settings.get_db_revision_filename()
-
-    @staticmethod
-    def _write_db_revision_file(version_num):
-        with open(conf.settings.get_db_revision_filename(), mode='w') as db_revision:
-            db_revision.write(str(version_num))
-
-    @defer.inlineCallbacks
-    def start(self):
-        # check directories exist, create them if they don't
-        log.info("Loading databases")
-
-        if not os.path.exists(GCS('download_directory')):
-            os.mkdir(GCS('download_directory'))
-
-        if not os.path.exists(GCS('data_dir')):
-            os.mkdir(GCS('data_dir'))
-            self._write_db_revision_file(self.get_current_db_revision())
-            log.debug("Created the db revision file: %s", self.get_revision_filename())
-
-        if not os.path.exists(CS.get_blobfiles_dir()):
-            os.mkdir(CS.get_blobfiles_dir())
-            log.debug("Created the blobfile directory: %s", str(CS.get_blobfiles_dir()))
-
-        if not os.path.exists(self.get_revision_filename()):
-            log.warning("db_revision file not found. Creating it")
-            self._write_db_revision_file(self.get_current_db_revision())
-
-        # check the db migration and run any needed migrations
-        with open(self.get_revision_filename(), "r") as revision_read_handle:
-            old_revision = int(revision_read_handle.read().strip())
-
-        if old_revision > self.get_current_db_revision():
-            raise Exception('This version of lbrynet is not compatible with the database\n'
-                            'Your database is revision %i, expected %i' %
-                            (old_revision, self.get_current_db_revision()))
-        if old_revision < self.get_current_db_revision():
-            from lbrynet.database.migrator import dbmigrator
-            log.info("Upgrading your databases (revision %i to %i)", old_revision, self.get_current_db_revision())
-            yield threads.deferToThread(
-                dbmigrator.migrate_db, GCS('data_dir'), old_revision, self.get_current_db_revision()
-            )
-            self._write_db_revision_file(self.get_current_db_revision())
-            log.info("Finished upgrading the databases.")
-
-        # start SQLiteStorage
-        self.storage = SQLiteStorage(GCS('data_dir'))
-        yield self.storage.setup()
-
-    @defer.inlineCallbacks
-    def stop(self):
-        yield self.storage.stop()
-        self.storage = None
-
-
-class WalletComponent(Component):
-    component_name = WALLET_COMPONENT
-    depends_on = [DATABASE_COMPONENT]
-
-    def __init__(self, component_manager):
-        Component.__init__(self, component_manager)
-        self.wallet = None
-
-    @property
-    def component(self):
-        return self.wallet
-
-    @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')
-            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):
-        yield self.wallet.stop()
-        self.wallet = None
-
-
-class SessionComponent(Component):
-    component_name = SESSION_COMPONENT
-    depends_on = [DATABASE_COMPONENT, WALLET_COMPONENT, DHT_COMPONENT, HASH_ANNOUNCER_COMPONENT]
-
-    def __init__(self, component_manager):
-        Component.__init__(self, component_manager)
-        self.session = None
-
-    @property
-    def component(self):
-        return self.session
-
-    @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'),
-            use_upnp=GCS('use_upnp'),
-            wallet=self.component_manager.get_component(WALLET_COMPONENT),
-            is_generous=GCS('is_generous_host'),
-            external_ip=CS.get_external_ip(),
-            storage=self.component_manager.get_component(DATABASE_COMPONENT)
-        )
-        yield self.session.setup()
-
-    @defer.inlineCallbacks
-    def stop(self):
-        yield self.session.shut_down()
-
-
-class DHTComponent(Component):
-    component_name = DHT_COMPONENT
-    depends_on = [UPNP_COMPONENT]
-
-    def __init__(self, component_manager):
-        Component.__init__(self, component_manager)
-        self.dht_node = None
-        self.upnp_component = None
-        self.udp_port, self.peer_port = None, None
-
-    @property
-    def component(self):
-        return self.dht_node
-
-    @defer.inlineCallbacks
-    def start(self):
-        self.upnp_component = self.component_manager.get_component(UPNP_COMPONENT)
-        self.udp_port, self.peer_port = self.upnp_component.get_redirects()
-        node_id = CS.get_node_id()
-        if node_id is None:
-            node_id = generate_id()
-
-        self.dht_node = node.Node(
-            node_id=node_id,
-            udpPort=self.udp_port,
-            externalIP=CS.get_external_ip(),
-            peerPort=self.peer_port
-        )
-        yield self.dht_node.start(GCS('known_dht_nodes'))
-        log.info("Joined the dht")
-
-    @defer.inlineCallbacks
-    def stop(self):
-        yield self.dht_node.stop()
-
-
-class HashAnnouncer(Component):
-    component_name = HASH_ANNOUNCER_COMPONENT
-    depends_on = [DHT_COMPONENT, DATABASE_COMPONENT]
-
-    def __init__(self, component_manager):
-        Component.__init__(self, component_manager)
-        self.hash_announcer = None
-
-    @property
-    def component(self):
-        return self.hash_announcer
-
-    @defer.inlineCallbacks
-    def start(self):
-        storage = self.component_manager.get_component(DATABASE_COMPONENT)
-        dht_node = self.component_manager.get_component(DHT_COMPONENT)
-        self.hash_announcer = hashannouncer.DHTHashAnnouncer(dht_node, storage)
-        yield self.hash_announcer.start()
-
-    @defer.inlineCallbacks
-    def stop(self):
-        yield self.hash_announcer.stop()
-
-
-class StreamIdentifier(Component):
-    component_name = STREAM_IDENTIFIER_COMPONENT
-    depends_on = [SESSION_COMPONENT]
-
-    def __init__(self, component_manager):
-        Component.__init__(self, component_manager)
-        self.sd_identifier = StreamDescriptorIdentifier()
-
-    @property
-    def component(self):
-        return self.sd_identifier
-
-    @defer.inlineCallbacks
-    def start(self):
-        session = self.component_manager.get_component(SESSION_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,
-            GCS('download_directory')
-        )
-        yield self.sd_identifier.add_stream_downloader_factory(EncryptedFileStreamType, file_saver_factory)
-
-    def stop(self):
-        pass
-
-
-class FileManager(Component):
-    component_name = FILE_MANAGER_COMPONENT
-    depends_on = [SESSION_COMPONENT, STREAM_IDENTIFIER_COMPONENT]
-
-    def __init__(self, component_manager):
-        Component.__init__(self, component_manager)
-        self.file_manager = None
-
-    @property
-    def component(self):
-        return self.file_manager
-
-    @defer.inlineCallbacks
-    def start(self):
-        session = self.component_manager.get_component(SESSION_COMPONENT)
-        sd_identifier = self.component_manager.get_component(STREAM_IDENTIFIER_COMPONENT)
-        log.info('Starting the file manager')
-        self.file_manager = EncryptedFileManager(session, sd_identifier)
-        yield self.file_manager.setup()
-        log.info('Done setting up file manager')
-
-    @defer.inlineCallbacks
-    def stop(self):
-        yield self.file_manager.stop()
-
-
-class PeerProtocolServer(Component):
-    component_name = PEER_PROTOCOL_SERVER_COMPONENT
-    depends_on = [SESSION_COMPONENT]
-
-    def __init__(self, component_manager):
-        Component.__init__(self, component_manager)
-        self.lbry_server_port = None
-
-    @property
-    def component(self):
-        return self.lbry_server_port
-
-    @defer.inlineCallbacks
-    def start(self):
-        query_handlers = {}
-        peer_port = GCS('peer_port')
-        session = self.component_manager.get_component(SESSION_COMPONENT)
-
-        handlers = [
-            BlobRequestHandlerFactory(
-                session.blob_manager,
-                session.wallet,
-                session.payment_rate_manager,
-                self.component_manager.analytics_manager
-            ),
-            session.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)
-
-            try:
-                log.info("Peer protocol listening on TCP %d", peer_port)
-                self.lbry_server_port = yield reactor.listenTCP(peer_port, server_factory)
-            except error.CannotListenError as e:
-                import traceback
-                log.error("Couldn't bind to port %d. Visit lbry.io/faq/how-to-change-port for"
-                          " more details.", peer_port)
-                log.error("%s", traceback.format_exc())
-                raise ValueError("%s lbrynet may already be running on your computer." % str(e))
-
-    @defer.inlineCallbacks
-    def stop(self):
-        if self.lbry_server_port is not None:
-            self.lbry_server_port, old_port = None, self.lbry_server_port
-            log.info('Stop listening on port %s', old_port.port)
-            yield old_port.stopListening()
-
-
-class ReflectorComponent(Component):
-    component_name = REFLECTOR_COMPONENT
-    depends_on = [SESSION_COMPONENT, FILE_MANAGER_COMPONENT]
-
-    def __init__(self, component_manager):
-        Component.__init__(self, component_manager)
-        self.reflector_server_port = GCS('reflector_port')
-        self.run_reflector_server = GCS('run_reflector_server')
-
-    @property
-    def component(self):
-        return self
-
-    @defer.inlineCallbacks
-    def start(self):
-        session = self.component_manager.get_component(SESSION_COMPONENT)
-        file_manager = self.component_manager.get_component(FILE_MANAGER_COMPONENT)
-
-        if self.run_reflector_server and self.reflector_server_port is not None:
-            log.info("Starting reflector server")
-            reflector_factory = reflector_server_factory(session.peer_manager, session.blob_manager, file_manager)
-            try:
-                self.reflector_server_port = yield reactor.listenTCP(self.reflector_server_port, reflector_factory)
-                log.info('Started reflector on port %s', self.reflector_server_port)
-            except error.CannotListenError as e:
-                log.exception("Couldn't bind reflector to port %d", self.reflector_server_port)
-                raise ValueError("{} lbrynet may already be running on your computer.".format(e))
-
-    def stop(self):
-        if self.run_reflector_server and self.reflector_server_port is not None:
-            log.info("Stopping reflector server")
-            if self.reflector_server_port is not None:
-                self.reflector_server_port, p = None, self.reflector_server_port
-                yield p.stopListening
-
-
-class UPnPComponent(Component):
-    component_name = UPNP_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.use_upnp = GCS('use_upnp')
-        self.external_ip = CS.get_external_ip()
-        self.upnp_redirects = []
-
-    @property
-    def component(self):
-        return self
-
-    def get_redirects(self):
-        return self.peer_port, self.dht_node_port
-
-    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
-
-    def stop(self):
-        log.info("Unsetting upnp for session")
-
-        def threaded_unset_upnp():
-            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

lbrynet/daemon/DaemonCLI.py

@@ -7,7 +7,7 @@ from collections import OrderedDict
 from lbrynet import conf
 from lbrynet.core import utils
 from lbrynet.daemon.auth.client import JSONRPCException, LBRYAPIClient, AuthAPIClient
-from lbrynet.daemon.Daemon import Daemon
+from lbrynet.daemon.Daemon import LOADING_WALLET_CODE, Daemon
 from lbrynet.core.system_info import get_platform
 from jsonrpc.common import RPCError
 from requests.exceptions import ConnectionError
@@ -21,13 +21,17 @@ def remove_brackets(key):
     return key
 
 
-def set_kwargs(parsed_args):
+def set_flag_vals(flag_names, parsed_args):
     kwargs = OrderedDict()
     for key, arg in parsed_args.iteritems():
         if arg is None:
             continue
-        elif key.startswith("--") and remove_brackets(key[2:]) not in kwargs:
+        elif key.startswith("--"):
+            if remove_brackets(key[2:]) not in kwargs:
                 k = remove_brackets(key[2:])
+        elif key in flag_names:
+            if remove_brackets(flag_names[key]) not in kwargs:
+                k = remove_brackets(flag_names[key])
         elif remove_brackets(key) not in kwargs:
             k = remove_brackets(key)
         kwargs[k] = guess_type(arg, k)
@@ -75,22 +79,26 @@ def main():
         method = new_method
 
     fn = Daemon.callable_methods[method]
+    if hasattr(fn, "_flags"):
+        flag_names = fn._flags
+    else:
+        flag_names = {}
 
     parsed = docopt(fn.__doc__, args)
-    kwargs = set_kwargs(parsed)
+    kwargs = set_flag_vals(flag_names, parsed)
     colorama.init()
     conf.initialize_settings()
 
     try:
         api = LBRYAPIClient.get_client()
-        api.status()
+        status = api.status()
     except (URLError, ConnectionError) as err:
         if isinstance(err, HTTPError) and err.code == UNAUTHORIZED:
             api = AuthAPIClient.config()
             # this can happen if the daemon is using auth with the --http-auth flag
             # when the config setting is to not use it
             try:
-                api.status()
+                status = api.status()
             except:
                 print_error("Daemon requires authentication, but none was provided.",
                             suggest_help=False)
@@ -100,6 +108,20 @@ def main():
                         suggest_help=False)
             return 1
 
+    status_code = status['startup_status']['code']
+
+    if status_code != "started" and method not in Daemon.allowed_during_startup:
+        print "Daemon is in the process of starting. Please try again in a bit."
+        message = status['startup_status']['message']
+        if message:
+            if (
+                status['startup_status']['code'] == LOADING_WALLET_CODE
+                and status['blockchain_status']['blocks_behind'] > 0
+            ):
+                message += '. Blocks left: ' + str(status['blockchain_status']['blocks_behind'])
+            print "  Status: " + message
+        return 1
+
     # TODO: check if port is bound. Error if its not
 
     try:

lbrynet/daemon/__init__.py

@@ -1,3 +1,3 @@
-import Components  # register Component classes
 from lbrynet.daemon.auth.client import LBRYAPIClient
+
 get_client = LBRYAPIClient.get_client

lbrynet/daemon/auth/server.py

@@ -4,7 +4,6 @@ import json
 import inspect
 
 from decimal import Decimal
-from functools import wraps
 from zope.interface import implements
 from twisted.web import server, resource
 from twisted.internet import defer
@@ -16,7 +15,6 @@ from traceback import format_exc
 from lbrynet import conf
 from lbrynet.core.Error import InvalidAuthenticationToken
 from lbrynet.core import utils
-from lbrynet.core.Error import ComponentsNotStarted, ComponentStartConditionNotMet
 from lbrynet.daemon.auth.util import APIKey, get_auth_message
 from lbrynet.daemon.auth.client import LBRY_SECRET
 from lbrynet.undecorated import undecorated
@@ -143,27 +141,6 @@ class AuthorizedBase(object):
             return f
         return _deprecated_wrapper
 
-    @staticmethod
-    def requires(*components, **component_conditionals):
-        def _wrap(fn):
-            @defer.inlineCallbacks
-            @wraps(fn)
-            def _inner(*args, **kwargs):
-                if component_conditionals:
-                    for component_name, condition in component_conditionals.iteritems():
-                        if not callable(condition):
-                            raise SyntaxError("The specified condition is invalid/not callable")
-                        if not (yield condition(args[0].component_manager.get_component(component_name))):
-                            raise ComponentStartConditionNotMet(
-                                "Not all conditions required to start component are met")
-                if args[0].component_manager.all_components_running(*components):
-                    result = yield fn(*args, **kwargs)
-                    defer.returnValue(result)
-                else:
-                    raise ComponentsNotStarted("Not all required components are set up:", components)
-            return _inner
-        return _wrap
-
 
 class AuthJSONRPCServer(AuthorizedBase):
     """
@@ -172,6 +149,7 @@ class AuthJSONRPCServer(AuthorizedBase):
     API methods are named with a leading "jsonrpc_"
 
     Attributes:
+        allowed_during_startup (list): list of api methods that are callable before the server has finished startup
         sessions (dict): (dict): {<session id>: <lbrynet.daemon.auth.util.APIKey>}
         callable_methods (dict): {<api method name>: <api method>}
 
@@ -438,6 +416,9 @@ class AuthJSONRPCServer(AuthorizedBase):
     def _verify_method_is_callable(self, function_path):
         if function_path not in self.callable_methods:
             raise UnknownAPIMethodError(function_path)
+        if not self.announced_startup:
+            if function_path not in self.allowed_during_startup:
+                raise NotAllowedDuringStartupError(function_path)
 
     def _get_jsonrpc_method(self, function_path):
         if function_path in self.deprecated_methods:

lbrynet/tests/functional/test_misc.py

@@ -39,7 +39,6 @@ DummyBlobAvailabilityTracker = mocks.BlobAvailabilityTracker
 log_format = "%(funcName)s(): %(message)s"
 logging.basicConfig(level=logging.CRITICAL, format=log_format)
 
-TEST_SKIP_STRING_ANDROID = "Test cannot pass on Android because multiprocessing is not supported at the OS level."
 
 def require_system(system):
     def wrapper(fn):
@@ -104,15 +103,13 @@ class LbryUploader(object):
         rate_limiter = RateLimiter()
         self.sd_identifier = StreamDescriptorIdentifier()
         self.db_dir, self.blob_dir = mk_db_and_blob_dir()
-        dht_node = FakeNode(peer_finder=peer_finder, peer_manager=peer_manager, udpPort=4445, peerPort=5553,
-                        node_id="abcd", externalIP="127.0.0.1")
 
         self.session = Session(
             conf.ADJUSTABLE_SETTINGS['data_rate'][1], db_dir=self.db_dir, blob_dir=self.blob_dir,
             node_id="abcd", peer_finder=peer_finder, hash_announcer=hash_announcer,
             peer_port=5553, dht_node_port=4445, use_upnp=False, rate_limiter=rate_limiter, wallet=wallet,
             blob_tracker_class=DummyBlobAvailabilityTracker,
-            dht_node=dht_node, is_generous=self.is_generous, external_ip="127.0.0.1")
+            dht_node_class=FakeNode, is_generous=self.is_generous, external_ip="127.0.0.1")
         self.lbry_file_manager = EncryptedFileManager(self.session, self.sd_identifier)
         if self.ul_rate_limit is not None:
             self.session.rate_limiter.set_ul_limit(self.ul_rate_limit)
@@ -200,7 +197,7 @@ def start_lbry_reuploader(sd_hash, kill_event, dead_event,
 
     db_dir, blob_dir = mk_db_and_blob_dir()
     session = Session(conf.ADJUSTABLE_SETTINGS['data_rate'][1], db_dir=db_dir,
-                      node_id="abcd" + str(n), dht_node_port=4446,
+                      node_id="abcd" + str(n), dht_node_port=4446, dht_node_class=FakeNode,
                       peer_finder=peer_finder, hash_announcer=hash_announcer,
                       blob_dir=blob_dir, peer_port=peer_port,
                       use_upnp=False, rate_limiter=rate_limiter, wallet=wallet,
@@ -306,16 +303,13 @@ def start_blob_uploader(blob_hash_queue, kill_event, dead_event, slow, is_genero
 
     db_dir, blob_dir = mk_db_and_blob_dir()
 
-    dht_node = FakeNode(peer_finder=peer_finder, peer_manager=peer_manager, udpPort=4445, peerPort=5553,
-                    node_id="abcd", externalIP="127.0.0.1")
-
     session = Session(conf.ADJUSTABLE_SETTINGS['data_rate'][1], db_dir=db_dir, node_id="efgh",
-                      peer_finder=peer_finder, hash_announcer=hash_announcer,
+                      peer_finder=peer_finder, hash_announcer=hash_announcer, dht_node_class=FakeNode,
                       blob_dir=blob_dir, peer_port=peer_port, dht_node_port=4446,
                       use_upnp=False, rate_limiter=rate_limiter, wallet=wallet,
                       blob_tracker_class=DummyBlobAvailabilityTracker,
                       is_generous=conf.ADJUSTABLE_SETTINGS['is_generous_host'][1],
-                      external_ip="127.0.0.1", dht_node=dht_node)
+                      external_ip="127.0.0.1")
 
     if slow is True:
         session.rate_limiter.set_ul_limit(2 ** 11)
@@ -484,8 +478,6 @@ class TestTransfer(TestCase):
         hash_announcer = FakeAnnouncer()
         rate_limiter = DummyRateLimiter()
         sd_identifier = StreamDescriptorIdentifier()
-        dht_node = FakeNode(peer_finder=peer_finder, peer_manager=peer_manager, udpPort=4445, peerPort=5553,
-                        node_id="abcd", externalIP="127.0.0.1")
 
         db_dir, blob_dir = mk_db_and_blob_dir()
         self.session = Session(
@@ -494,7 +486,7 @@ class TestTransfer(TestCase):
             blob_dir=blob_dir, peer_port=5553, dht_node_port=4445,
             use_upnp=False, rate_limiter=rate_limiter, wallet=wallet,
             blob_tracker_class=DummyBlobAvailabilityTracker,
-            dht_node=dht_node, is_generous=self.is_generous, external_ip="127.0.0.1")
+            dht_node_class=FakeNode, is_generous=self.is_generous, external_ip="127.0.0.1")
 
         self.lbry_file_manager = EncryptedFileManager(
             self.session, sd_identifier)
@@ -574,16 +566,14 @@ class TestTransfer(TestCase):
         peer_finder = FakePeerFinder(5553, peer_manager, 2)
         hash_announcer = FakeAnnouncer()
         rate_limiter = DummyRateLimiter()
-        dht_node = FakeNode(peer_finder=peer_finder, peer_manager=peer_manager, udpPort=4445, peerPort=5553,
-                        node_id="abcd", externalIP="127.0.0.1")
 
         db_dir, blob_dir = mk_db_and_blob_dir()
         self.session = Session(
             conf.ADJUSTABLE_SETTINGS['data_rate'][1], db_dir=db_dir, node_id="abcd",
             peer_finder=peer_finder, hash_announcer=hash_announcer,
-            blob_dir=blob_dir, peer_port=5553, dht_node_port=4445,
+            blob_dir=blob_dir, peer_port=5553, dht_node_port=4445, dht_node_class=FakeNode,
             use_upnp=False, rate_limiter=rate_limiter, wallet=wallet,
-            blob_tracker_class=DummyBlobAvailabilityTracker, dht_node=dht_node,
+            blob_tracker_class=DummyBlobAvailabilityTracker,
             is_generous=conf.ADJUSTABLE_SETTINGS['is_generous_host'][1], external_ip="127.0.0.1")
 
         d1 = self.wait_for_hash_from_queue(blob_hash_queue_1)
@@ -656,19 +646,17 @@ class TestTransfer(TestCase):
         hash_announcer = FakeAnnouncer()
         rate_limiter = DummyRateLimiter()
         sd_identifier = StreamDescriptorIdentifier()
-        dht_node = FakeNode(peer_finder=peer_finder, peer_manager=peer_manager, udpPort=4445, peerPort=5553,
-                        node_id="abcd", externalIP="127.0.0.1")
 
         downloaders = []
 
         db_dir, blob_dir = mk_db_and_blob_dir()
         self.session = Session(conf.ADJUSTABLE_SETTINGS['data_rate'][1], db_dir=db_dir,
-                               node_id="abcd", peer_finder=peer_finder, dht_node_port=4445,
+                               node_id="abcd", peer_finder=peer_finder, dht_node_port=4445, dht_node_class=FakeNode,
                                hash_announcer=hash_announcer, blob_dir=blob_dir, peer_port=5553,
                                use_upnp=False, rate_limiter=rate_limiter, wallet=wallet,
                                blob_tracker_class=DummyBlobAvailabilityTracker,
                                is_generous=conf.ADJUSTABLE_SETTINGS['is_generous_host'][1],
-                               external_ip="127.0.0.1", dht_node=dht_node)
+                               external_ip="127.0.0.1")
 
         self.lbry_file_manager = EncryptedFileManager(self.session, sd_identifier)
 
@@ -770,7 +758,7 @@ class TestTransfer(TestCase):
         sd_identifier = StreamDescriptorIdentifier()
 
         db_dir, blob_dir = mk_db_and_blob_dir()
-        self.session = Session(conf.ADJUSTABLE_SETTINGS['data_rate'][1], db_dir=db_dir,
+        self.session = Session(conf.ADJUSTABLE_SETTINGS['data_rate'][1], db_dir=db_dir, dht_node_class=FakeNode,
                                node_id="abcd", peer_finder=peer_finder, dht_node_port=4445,
                                hash_announcer=hash_announcer, blob_dir=blob_dir,
                                peer_port=5553, use_upnp=False, rate_limiter=rate_limiter,
@@ -854,10 +842,3 @@ class TestTransfer(TestCase):
         d.addBoth(stop)
 
         return d
-
-    if is_android():
-        test_lbry_transfer.skip = TEST_SKIP_STRING_ANDROID
-        test_last_blob_retrieval.skip = TEST_SKIP_STRING_ANDROID
-        test_double_download.skip = TEST_SKIP_STRING_ANDROID
-        test_multiple_uploaders.skip = TEST_SKIP_STRING_ANDROID
-

lbrynet/tests/functional/test_reflector.py

@@ -53,7 +53,6 @@ class TestReflector(unittest.TestCase):
             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,
@@ -75,7 +74,6 @@ class TestReflector(unittest.TestCase):
             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,

lbrynet/tests/functional/test_streamify.py

@@ -30,7 +30,6 @@ DummyBlobAvailabilityTracker = mocks.BlobAvailabilityTracker
 
 class TestStreamify(TestCase):
     maxDiff = 5000
-
     def setUp(self):
         mocks.mock_conf_settings(self)
         self.session = None
@@ -38,12 +37,6 @@ class TestStreamify(TestCase):
         self.is_generous = True
         self.db_dir = tempfile.mkdtemp()
         self.blob_dir = os.path.join(self.db_dir, "blobfiles")
-        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)
 
     @defer.inlineCallbacks
@@ -61,17 +54,26 @@ class TestStreamify(TestCase):
             os.remove("test_file")
 
     def test_create_stream(self):
+        wallet = FakeWallet()
+        peer_manager = PeerManager()
+        peer_finder = FakePeerFinder(5553, peer_manager, 2)
+        hash_announcer = FakeAnnouncer()
+        rate_limiter = DummyRateLimiter()
+        sd_identifier = StreamDescriptorIdentifier()
 
         self.session = Session(
-            conf.ADJUSTABLE_SETTINGS['data_rate'][1], db_dir=self.db_dir, node_id="abcd", peer_finder=self.peer_finder,
+            conf.ADJUSTABLE_SETTINGS['data_rate'][1], db_dir=self.db_dir, node_id="abcd",
-            blob_dir=self.blob_dir, peer_port=5553, use_upnp=False, rate_limiter=self.rate_limiter, wallet=self.wallet,
+            peer_finder=peer_finder, hash_announcer=hash_announcer,
-            blob_tracker_class=DummyBlobAvailabilityTracker, external_ip="127.0.0.1", dht_node=self.dht_node
+            blob_dir=self.blob_dir, peer_port=5553,
+            use_upnp=False, rate_limiter=rate_limiter, wallet=wallet,
+            blob_tracker_class=DummyBlobAvailabilityTracker,
+            is_generous=self.is_generous, external_ip="127.0.0.1", dht_node_class=mocks.Node
         )
 
-        self.lbry_file_manager = EncryptedFileManager(self.session, self.sd_identifier)
+        self.lbry_file_manager = EncryptedFileManager(self.session, sd_identifier)
 
         d = self.session.setup()
-        d.addCallback(lambda _: add_lbry_file_to_sd_identifier(self.sd_identifier))
+        d.addCallback(lambda _: add_lbry_file_to_sd_identifier(sd_identifier))
         d.addCallback(lambda _: self.lbry_file_manager.setup())
 
         def verify_equal(sd_info):
@@ -100,14 +102,22 @@ class TestStreamify(TestCase):
         return d
 
     def test_create_and_combine_stream(self):
+        wallet = FakeWallet()
+        peer_manager = PeerManager()
+        peer_finder = FakePeerFinder(5553, peer_manager, 2)
+        hash_announcer = FakeAnnouncer()
+        rate_limiter = DummyRateLimiter()
+        sd_identifier = StreamDescriptorIdentifier()
 
         self.session = Session(
-            conf.ADJUSTABLE_SETTINGS['data_rate'][1], db_dir=self.db_dir, node_id="abcd", peer_finder=self.peer_finder,
+            conf.ADJUSTABLE_SETTINGS['data_rate'][1], db_dir=self.db_dir, node_id="abcd",
-            blob_dir=self.blob_dir, peer_port=5553, use_upnp=False, rate_limiter=self.rate_limiter, wallet=self.wallet,
+            peer_finder=peer_finder, hash_announcer=hash_announcer,
-            blob_tracker_class=DummyBlobAvailabilityTracker, external_ip="127.0.0.1", dht_node=self.dht_node
+            blob_dir=self.blob_dir, peer_port=5553, dht_node_class=mocks.Node,
+            use_upnp=False, rate_limiter=rate_limiter, wallet=wallet,
+            blob_tracker_class=DummyBlobAvailabilityTracker, external_ip="127.0.0.1"
         )
 
-        self.lbry_file_manager = EncryptedFileManager(self.session, self.sd_identifier)
+        self.lbry_file_manager = EncryptedFileManager(self.session, sd_identifier)
 
         @defer.inlineCallbacks
         def create_stream():
@@ -122,7 +132,7 @@ class TestStreamify(TestCase):
             self.assertEqual(hashsum.hexdigest(), "68959747edc73df45e45db6379dd7b3b")
 
         d = self.session.setup()
-        d.addCallback(lambda _: add_lbry_file_to_sd_identifier(self.sd_identifier))
+        d.addCallback(lambda _: add_lbry_file_to_sd_identifier(sd_identifier))
         d.addCallback(lambda _: self.lbry_file_manager.setup())
         d.addCallback(lambda _: create_stream())
         return d

lbrynet/tests/mocks.py

@@ -1,6 +1,5 @@
 import base64
 import io
-import mock
 
 from cryptography.hazmat.backends import default_backend
 from cryptography.hazmat.primitives.asymmetric import rsa
@@ -11,7 +10,6 @@ from twisted.python.failure import Failure
 from lbrynet.core.client.ClientRequest import ClientRequest
 from lbrynet.core.Error import RequestCanceledError
 from lbrynet.core import BlobAvailability
-from lbrynet.file_manager.EncryptedFileManager import EncryptedFileManager
 from lbrynet.dht.node import Node as RealNode
 from lbrynet.daemon import ExchangeRateManager as ERM
 from lbrynet import conf
@@ -65,7 +63,6 @@ class BTCLBCFeed(ERM.MarketFeed):
             0.0
         )
 
-
 class USDBTCFeed(ERM.MarketFeed):
     def __init__(self):
         ERM.MarketFeed.__init__(
@@ -77,7 +74,6 @@ class USDBTCFeed(ERM.MarketFeed):
             0.0
         )
 
-
 class ExchangeRateManager(ERM.ExchangeRateManager):
     def __init__(self, market_feeds, rates):
         self.market_feeds = market_feeds
@@ -364,96 +360,6 @@ class BlobAvailabilityTracker(BlobAvailability.BlobAvailabilityTracker):
         pass
 
 
-# The components below viz. FakeWallet, FakeSession, FakeFileManager are just for testing Component Manager's
-# startup and stop
-class FakeComponent(object):
-    depends_on = []
-    component_name = None
-
-    def __init__(self, component_manager):
-        self.component_manager = component_manager
-        self._running = False
-
-    @property
-    def running(self):
-        return self._running
-
-    def start(self):
-        raise NotImplementedError  # Override
-
-    def stop(self):
-        return defer.succeed(None)
-
-    @property
-    def component(self):
-        return self
-
-    @defer.inlineCallbacks
-    def _setup(self):
-        result = yield defer.maybeDeferred(self.start)
-        self._running = True
-        defer.returnValue(result)
-
-    @defer.inlineCallbacks
-    def _stop(self):
-        result = yield defer.maybeDeferred(self.stop)
-        self._running = False
-        defer.returnValue(result)
-
-
-class FakeDelayedWallet(FakeComponent):
-    component_name = "wallet"
-    depends_on = []
-
-    def start(self):
-        return defer.succeed(True)
-
-    def stop(self):
-        d = defer.Deferred()
-        self.component_manager.reactor.callLater(1, d.callback, True)
-        return d
-
-
-class FakeDelayedSession(FakeComponent):
-    component_name = "session"
-    depends_on = [FakeDelayedWallet.component_name]
-
-    def start(self):
-        d = defer.Deferred()
-        self.component_manager.reactor.callLater(1, d.callback, True)
-        return d
-
-    def stop(self):
-        d = defer.Deferred()
-        self.component_manager.reactor.callLater(1, d.callback, True)
-        return d
-
-
-class FakeDelayedFileManager(FakeComponent):
-    component_name = "file_manager"
-    depends_on = [FakeDelayedSession.component_name]
-
-    def start(self):
-        d = defer.Deferred()
-        self.component_manager.reactor.callLater(1, d.callback, True)
-        return d
-
-    def stop(self):
-        return defer.succeed(True)
-
-class FakeFileManager(FakeComponent):
-    component_name = "fileManager"
-    depends_on = []
-
-    @property
-    def component(self):
-        return mock.Mock(spec=EncryptedFileManager)
-
-    def start(self):
-        return defer.succeed(True)
-
-    def stop(self):
-        pass
 
 create_stream_sd_file = {
     'stream_name': '746573745f66696c65',

lbrynet/tests/unit/components/test_Component_Manager.py

@@ -1,131 +0,0 @@
-from twisted.internet.task import Clock
-from twisted.trial import unittest
-
-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
-from lbrynet.daemon import Components
-from lbrynet.tests import mocks
-
-
-class TestComponentManager(unittest.TestCase):
-    def setUp(self):
-        mocks.mock_conf_settings(self)
-        self.default_components_sort = [
-            [Components.DatabaseComponent,
-             Components.UPnPComponent],
-            [Components.DHTComponent,
-             Components.WalletComponent],
-            [Components.HashAnnouncer],
-            [Components.SessionComponent],
-            [Components.PeerProtocolServer,
-             Components.StreamIdentifier],
-            [Components.FileManager],
-            [Components.ReflectorComponent]
-        ]
-        self.component_manager = ComponentManager()
-
-    def tearDown(self):
-        pass
-
-    def test_sort_components(self):
-        stages = self.component_manager.sort_components()
-
-        for stage_list, sorted_stage_list in zip(stages, self.default_components_sort):
-            self.assertSetEqual(set([type(stage) for stage in stage_list]), set(sorted_stage_list))
-
-    def test_sort_components_reverse(self):
-        rev_stages = self.component_manager.sort_components(reverse=True)
-        reverse_default_components_sort = reversed(self.default_components_sort)
-
-        for stage_list, sorted_stage_list in zip(rev_stages, reverse_default_components_sort):
-            self.assertSetEqual(set([type(stage) for stage in stage_list]), set(sorted_stage_list))
-
-    def test_get_component_not_exists(self):
-
-        with self.assertRaises(NameError):
-            self.component_manager.get_component("random_component")
-
-
-class TestComponentManagerOverrides(unittest.TestCase):
-    def setUp(self):
-        mocks.mock_conf_settings(self)
-
-    def test_init_with_overrides(self):
-        class FakeWallet(object):
-            component_name = "wallet"
-            depends_on = []
-
-            def __init__(self, component_manager):
-                self.component_manager = component_manager
-
-            @property
-            def component(self):
-                return self
-
-        new_component_manager = ComponentManager(wallet=FakeWallet)
-        fake_wallet = new_component_manager.get_component("wallet")
-        # wallet should be an instance of FakeWallet and not WalletComponent from Components.py
-        self.assertEquals(type(fake_wallet), FakeWallet)
-        self.assertNotEquals(type(fake_wallet), Components.WalletComponent)
-
-    def test_init_with_wrong_overrides(self):
-        class FakeRandomComponent(object):
-            component_name = "someComponent"
-            depends_on = []
-
-        with self.assertRaises(SyntaxError):
-            ComponentManager(randomComponent=FakeRandomComponent)
-
-
-class TestComponentManagerProperStart(unittest.TestCase):
-    def setUp(self):
-        self.reactor = Clock()
-        mocks.mock_conf_settings(self)
-        self.component_manager = ComponentManager(
-            skip_components=[DATABASE_COMPONENT, DHT_COMPONENT, HASH_ANNOUNCER_COMPONENT, STREAM_IDENTIFIER_COMPONENT,
-                             PEER_PROTOCOL_SERVER_COMPONENT, REFLECTOR_COMPONENT, UPNP_COMPONENT],
-            reactor=self.reactor,
-            wallet=mocks.FakeDelayedWallet,
-            session=mocks.FakeDelayedSession,
-            fileManager=mocks.FakeDelayedFileManager
-        )
-
-    def tearDown(self):
-        pass
-
-    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('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.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('file_manager').running)
-
-    def test_proper_stopping_of_components(self):
-        self.component_manager.setup()
-        self.reactor.advance(1)
-        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('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.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('wallet').running)

lbrynet/tests/unit/lbrynet_daemon/test_Daemon.py

@@ -1,10 +1,11 @@
 import mock
 import json
+import unittest
 import random
 from os import path
 
 from twisted.internet import defer
-from twisted.trial import unittest
+from twisted import trial
 
 from faker import Faker
 
@@ -13,15 +14,12 @@ from lbryum.wallet import NewWallet
 from lbrynet import conf
 from lbrynet.core import Session, PaymentRateManager, 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 PEER_PROTOCOL_SERVER_COMPONENT
 from lbrynet.daemon.Daemon import Daemon as LBRYDaemon
+from lbrynet.file_manager.EncryptedFileManager import EncryptedFileManager
 from lbrynet.file_manager.EncryptedFileDownloader import ManagedEncryptedFileDownloader
 
 from lbrynet.tests import util
-from lbrynet.tests.mocks import mock_conf_settings, FakeNetwork, FakeFileManager
+from lbrynet.tests.mocks import mock_conf_settings, FakeNetwork
 from lbrynet.tests.mocks import BlobAvailabilityTracker as DummyBlobAvailabilityTracker
 from lbrynet.tests.mocks import ExchangeRateManager as DummyExchangeRateManager
 from lbrynet.tests.mocks import BTCLBCFeed, USDBTCFeed
@@ -42,10 +40,10 @@ def get_test_daemon(data_rate=None, generous=True, with_fee=False):
     }
     daemon = LBRYDaemon(None)
     daemon.session = mock.Mock(spec=Session.Session)
-    daemon.wallet = mock.Mock(spec=Wallet.LBRYumWallet)
+    daemon.session.wallet = mock.Mock(spec=Wallet.LBRYumWallet)
-    daemon.wallet.wallet = mock.Mock(spec=NewWallet)
+    daemon.session.wallet.wallet = mock.Mock(spec=NewWallet)
-    daemon.wallet.wallet.use_encryption = False
+    daemon.session.wallet.wallet.use_encryption = False
-    daemon.wallet.network = FakeNetwork()
+    daemon.session.wallet.network = FakeNetwork()
     daemon.session.storage = mock.Mock(spec=SQLiteStorage)
     market_feeds = [BTCLBCFeed(), USDBTCFeed()]
     daemon.exchange_rate_manager = DummyExchangeRateManager(market_feeds, rates)
@@ -75,12 +73,12 @@ def get_test_daemon(data_rate=None, generous=True, with_fee=False):
             {"fee": {"USD": {"address": "bQ6BGboPV2SpTMEP7wLNiAcnsZiH8ye6eA", "amount": 0.75}}})
     daemon._resolve_name = lambda _: defer.succeed(metadata)
     migrated = smart_decode(json.dumps(metadata))
-    daemon.wallet.resolve = lambda *_: defer.succeed(
+    daemon.session.wallet.resolve = lambda *_: defer.succeed(
         {"test": {'claim': {'value': migrated.claim_dict}}})
     return daemon
 
 
-class TestCostEst(unittest.TestCase):
+class TestCostEst(trial.unittest.TestCase):
     def setUp(self):
         mock_conf_settings(self)
         util.resetTime(self)
@@ -113,8 +111,7 @@ class TestCostEst(unittest.TestCase):
         self.assertEquals(daemon.get_est_cost("test", size).result, correct_result)
 
 
-class TestJsonRpc(unittest.TestCase):
+class TestJsonRpc(trial.unittest.TestCase):
-
     def setUp(self):
         def noop():
             return None
@@ -122,39 +119,30 @@ class TestJsonRpc(unittest.TestCase):
         mock_conf_settings(self)
         util.resetTime(self)
         self.test_daemon = get_test_daemon()
-        self.test_daemon.wallet.is_first_run = False
+        self.test_daemon.session.wallet.is_first_run = False
-        self.test_daemon.wallet.get_best_blockhash = noop
+        self.test_daemon.session.wallet.get_best_blockhash = noop
 
     def test_status(self):
         d = defer.maybeDeferred(self.test_daemon.jsonrpc_status)
         d.addCallback(lambda status: self.assertDictContainsSubset({'is_running': False}, status))
 
+    @unittest.skipIf(is_android(),
+                     'Test cannot pass on Android because PYTHONOPTIMIZE removes the docstrings.')
     def test_help(self):
         d = defer.maybeDeferred(self.test_daemon.jsonrpc_help, command='status')
         d.addCallback(lambda result: self.assertSubstring('daemon status', result['help']))
         # self.assertSubstring('daemon status', d.result)
 
-    if is_android():
-        test_help.skip = "Test cannot pass on Android because PYTHONOPTIMIZE removes the docstrings."
 
-
+class TestFileListSorting(trial.unittest.TestCase):
-class TestFileListSorting(unittest.TestCase):
     def setUp(self):
         mock_conf_settings(self)
         util.resetTime(self)
         self.faker = Faker('en_US')
         self.faker.seed(66410)
         self.test_daemon = get_test_daemon()
-        component_manager = ComponentManager(
+        self.test_daemon.lbry_file_manager = mock.Mock(spec=EncryptedFileManager)
-            skip_components=[DATABASE_COMPONENT, DHT_COMPONENT, WALLET_COMPONENT, SESSION_COMPONENT, UPNP_COMPONENT,
+        self.test_daemon.lbry_file_manager.lbry_files = self._get_fake_lbry_files()
-                             PEER_PROTOCOL_SERVER_COMPONENT, REFLECTOR_COMPONENT, HASH_ANNOUNCER_COMPONENT,
-                             STREAM_IDENTIFIER_COMPONENT],
-            fileManager=FakeFileManager
-        )
-        component_manager.setup()
-        self.test_daemon.component_manager = component_manager
-        self.test_daemon.file_manager = component_manager.get_component("fileManager")
-        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