import unittest from unittest import mock import json from lbry.conf import Config from lbry.extras.daemon.storage import SQLiteStorage from lbry.extras.daemon.componentmanager import ComponentManager from lbry.extras.daemon.components import DATABASE_COMPONENT, DHT_COMPONENT, WALLET_COMPONENT from lbry.extras.daemon.components import HASH_ANNOUNCER_COMPONENT from lbry.extras.daemon.components import UPNP_COMPONENT, BLOB_COMPONENT from lbry.extras.daemon.components import PEER_PROTOCOL_SERVER_COMPONENT, EXCHANGE_RATE_MANAGER_COMPONENT from lbry.extras.daemon.daemon import Daemon as LBRYDaemon from lbry.wallet import WalletManager, Wallet from tests import test_utils # from tests.mocks import mock_conf_settings, FakeNetwork, FakeFileManager # from tests.mocks import ExchangeRateManager as DummyExchangeRateManager # from tests.mocks import BTCLBCFeed, USDBTCFeed from tests.test_utils import is_android def get_test_daemon(conf: Config, with_fee=False): conf.data_dir = '/tmp' rates = { 'BTCLBC': {'spot': 3.0, 'ts': test_utils.DEFAULT_ISO_TIME + 1}, 'USDBTC': {'spot': 2.0, 'ts': test_utils.DEFAULT_ISO_TIME + 2} } component_manager = ComponentManager( conf, skip_components=[ DATABASE_COMPONENT, DHT_COMPONENT, WALLET_COMPONENT, UPNP_COMPONENT, PEER_PROTOCOL_SERVER_COMPONENT, HASH_ANNOUNCER_COMPONENT, EXCHANGE_RATE_MANAGER_COMPONENT, BLOB_COMPONENT, RATE_LIMITER_COMPONENT], file_manager=FakeFileManager ) daemon = LBRYDaemon(conf, component_manager=component_manager) daemon.payment_rate_manager = OnlyFreePaymentsManager() daemon.wallet_manager = mock.Mock(spec=WalletManager) daemon.wallet_manager.wallet = mock.Mock(spec=Wallet) daemon.wallet_manager.use_encryption = False daemon.wallet_manager.network = FakeNetwork() daemon.storage = mock.Mock(spec=SQLiteStorage) market_feeds = [BTCLBCFeed(), USDBTCFeed()] daemon.exchange_rate_manager = DummyExchangeRateManager(market_feeds, rates) daemon.stream_manager = component_manager.get_component(FILE_MANAGER_COMPONENT) metadata = { "author": "fake author", "language": "en", "content_type": "fake/format", "description": "fake description", "license": "fake license", "license_url": "fake license url", "nsfw": False, "sources": { "lbry_sd_hash": 'd2b8b6e907dde95245fe6d144d16c2fdd60c4e0c6463ec98' 'b85642d06d8e9414e8fcfdcb7cb13532ec5454fb8fe7f280' }, "thumbnail": "fake thumbnail", "title": "fake title", "ver": "0.0.3" } if with_fee: metadata.update( {"fee": {"USD": {"address": "bQ6BGboPV2SpTMEP7wLNiAcnsZiH8ye6eA", "amount": 0.75}}}) migrated = smart_decode(json.dumps(metadata)) daemon._resolve = daemon.resolve = lambda *_: defer.succeed( {"test": {'claim': {'value': migrated.claim_dict}}}) return daemon @unittest.SkipTest class TestCostEst(unittest.TestCase): def setUp(self): test_utils.reset_time(self) def test_fee_and_generous_data(self): size = 10000000 correct_result = 4.5 daemon = get_test_daemon(Config(is_generous_host=True), with_fee=True) result = yield f2d(daemon.get_est_cost("test", size)) self.assertEqual(result, correct_result) def test_generous_data_and_no_fee(self): size = 10000000 correct_result = 0.0 daemon = get_test_daemon(Config(is_generous_host=True)) result = yield f2d(daemon.get_est_cost("test", size)) self.assertEqual(result, correct_result) @unittest.SkipTest class TestJsonRpc(unittest.TestCase): def setUp(self): def noop(): return None test_utils.reset_time(self) self.test_daemon = get_test_daemon(Config()) self.test_daemon.wallet_manager.get_best_blockhash = noop def test_status(self): status = yield f2d(self.test_daemon.jsonrpc_status()) self.assertDictContainsSubset({'is_running': False}, status) def test_help(self): result = self.test_daemon.jsonrpc_help(command='status') self.assertSubstring('daemon status', result['help']) if is_android(): test_help.skip = "Test cannot pass on Android because PYTHONOPTIMIZE removes the docstrings." @unittest.SkipTest class TestFileListSorting(unittest.TestCase): def setUp(self): test_utils.reset_time(self) self.test_daemon = get_test_daemon(Config()) self.test_daemon.file_manager.lbry_files = self._get_fake_lbry_files() self.test_points_paid = [ 2.5, 4.8, 5.9, 5.9, 5.9, 6.1, 7.1, 8.2, 8.4, 9.1 ] self.test_file_names = [ 'add.mp3', 'any.mov', 'day.tiff', 'decade.odt', 'different.json', 'hotel.bmp', 'might.bmp', 'physical.json', 'remember.mp3', 'than.ppt' ] self.test_authors = [ 'ashlee27', 'bfrederick', 'brittanyhicks', 'davidsonjeffrey', 'heidiherring', 'jlewis', 'kswanson', 'michelle50', 'richard64', 'xsteele' ] return f2d(self.test_daemon.component_manager.start()) def test_sort_by_points_paid_no_direction_specified(self): sort_options = ['points_paid'] file_list = yield f2d(self.test_daemon.jsonrpc_file_list(sort=sort_options)['items']) self.assertEqual(self.test_points_paid, [f['points_paid'] for f in file_list]) def test_sort_by_points_paid_ascending(self): sort_options = ['points_paid,asc'] file_list = yield f2d(self.test_daemon.jsonrpc_file_list(sort=sort_options)['items']) self.assertEqual(self.test_points_paid, [f['points_paid'] for f in file_list]) def test_sort_by_points_paid_descending(self): sort_options = ['points_paid, desc'] file_list = yield f2d(self.test_daemon.jsonrpc_file_list(sort=sort_options)['items']) self.assertEqual(list(reversed(self.test_points_paid)), [f['points_paid'] for f in file_list]) def test_sort_by_file_name_no_direction_specified(self): sort_options = ['file_name'] file_list = yield f2d(self.test_daemon.jsonrpc_file_list(sort=sort_options)['items']) self.assertEqual(self.test_file_names, [f['file_name'] for f in file_list]) def test_sort_by_file_name_ascending(self): sort_options = ['file_name,\nasc'] file_list = yield f2d(self.test_daemon.jsonrpc_file_list(sort=sort_options)['items']) self.assertEqual(self.test_file_names, [f['file_name'] for f in file_list]) def test_sort_by_file_name_descending(self): sort_options = ['\tfile_name,\n\tdesc'] file_list = yield f2d(self.test_daemon.jsonrpc_file_list(sort=sort_options)['items']) self.assertEqual(list(reversed(self.test_file_names)), [f['file_name'] for f in file_list]) def test_sort_by_multiple_criteria(self): expected = [ 'file_name=different.json, points_paid=9.1', 'file_name=physical.json, points_paid=8.4', 'file_name=any.mov, points_paid=8.2', 'file_name=hotel.bmp, points_paid=7.1', 'file_name=add.mp3, points_paid=6.1', 'file_name=decade.odt, points_paid=5.9', 'file_name=might.bmp, points_paid=5.9', 'file_name=than.ppt, points_paid=5.9', 'file_name=remember.mp3, points_paid=4.8', 'file_name=day.tiff, points_paid=2.5' ] format_result = lambda f: f"file_name={f['file_name']}, points_paid={f['points_paid']}" sort_options = ['file_name,asc', 'points_paid,desc'] file_list = yield f2d(self.test_daemon.jsonrpc_file_list(sort=sort_options)['items']) self.assertEqual(expected, [format_result(r) for r in file_list]) # Check that the list is not sorted as expected when sorted only by file_name. sort_options = ['file_name,asc'] file_list = yield f2d(self.test_daemon.jsonrpc_file_list(sort=sort_options)['items']) self.assertNotEqual(expected, [format_result(r) for r in file_list]) # Check that the list is not sorted as expected when sorted only by points_paid. sort_options = ['points_paid,desc'] file_list = yield f2d(self.test_daemon.jsonrpc_file_list(sort=sort_options)['items']) self.assertNotEqual(expected, [format_result(r) for r in file_list]) # Check that the list is not sorted as expected when not sorted at all. file_list = yield f2d(self.test_daemon.jsonrpc_file_list()['items']) self.assertNotEqual(expected, [format_result(r) for r in file_list]) def test_sort_by_nested_field(self): extract_authors = lambda file_list: [f['metadata']['author'] for f in file_list] sort_options = ['metadata.author'] file_list = yield f2d(self.test_daemon.jsonrpc_file_list(sort=sort_options)['items']) self.assertEqual(self.test_authors, extract_authors(file_list)) # Check that the list matches the expected in reverse when sorting in descending order. sort_options = ['metadata.author,desc'] file_list = yield f2d(self.test_daemon.jsonrpc_file_list(sort=sort_options)['items']) self.assertEqual(list(reversed(self.test_authors)), extract_authors(file_list)) # Check that the list is not sorted as expected when not sorted at all. file_list = yield f2d(self.test_daemon.jsonrpc_file_list()['items']) self.assertNotEqual(self.test_authors, extract_authors(file_list)) def test_invalid_sort_produces_meaningful_errors(self): sort_options = ['meta.author'] expected_message = "Failed to get 'meta.author', key 'meta' was not found." with self.assertRaisesRegex(Exception, expected_message): yield f2d(self.test_daemon.jsonrpc_file_list(sort=sort_options)['items']) sort_options = ['metadata.foo.bar'] expected_message = "Failed to get 'metadata.foo.bar', key 'foo' was not found." with self.assertRaisesRegex(Exception, expected_message): yield f2d(self.test_daemon.jsonrpc_file_list(sort=sort_options)['items']) @staticmethod def _get_fake_lbry_files(): faked_lbry_files = [] for metadata in FAKED_LBRY_FILES: lbry_file = mock.Mock(spec=ManagedEncryptedFileDownloader) for attribute in metadata: setattr(lbry_file, attribute, metadata[attribute]) async def get_total_bytes(): return 0 lbry_file.get_total_bytes = get_total_bytes async def status(): return EncryptedFileStatusReport( 'file_name', 1, 1, 'completed' ) lbry_file.status = status faked_lbry_files.append(lbry_file) return faked_lbry_files FAKED_LBRY_FILES = ( { 'channel_claim_id': '3aace03b007d108c668d201533b7b07ab2981d47', 'channel_name': '@ashlee27', 'claim_id': 'cb63e644b6629467c031d0097d52ab6e0f1a5bf8', 'claim_name': 'very-skill-place-growth', 'completed': True, 'download_directory': '/usually', 'download_path': '/usually/any.mov', 'file_name': 'any.mov', 'key': b'>a\x11}\xec\xc2j\x1c\xe9\xc5l]\xfc\x16s|', 'metadata': {'author': 'ashlee27', 'nsfw': True}, 'mime_type': 'multipart/signed', 'nout': 7197, 'outpoint': 'c5633a5932f9c8e3e5b9799c07251b236e3aec078b0546614f24a932b6b133f6', 'points_paid': 8.2, 'sd_hash': '3354ecf502870f6f6d59d21188755c1361c2cffaeb458764c179c136b26c4795083acfd93b3920218870b1a9c22535ef', 'stopped': True, 'stream_hash': 'c8f58a686726116c15a8de7f33b4f0d72504777c7dd0c48ba94d7bbea23d9c82b2f977081cd7c49d25d6a2841b232e1d', 'stream_name': 'down.txt', 'suggested_file_name': 'down.txt', 'txid': '1c02986bfdb77b1c338e60b60c4c9febc59130af2e225f51665067c3d3419a35', 'written_bytes': 6838, }, { 'channel_claim_id': 'dade35ea84001858d7cf10f50be3b5fea3e57fb7', 'channel_name': '@richard64', 'claim_id': '1c01096727da90140d333197fa8aaf88893f6ea8', 'claim_name': 'room-tonight-produce-good', 'completed': False, 'download_directory': '/ability', 'download_path': '/ability/day.tiff', 'file_name': 'day.tiff', 'key': b'`\x86j\xba\x97\x0c\xe4L\xad\x06nC\x8b]\xd6&', 'metadata': {'author': 'richard64', 'nsfw': False}, 'mime_type': 'multipart/related', 'nout': 9678, 'outpoint': '0138083012ce4ff5a6e4c0ec2fc08e11a52ebd2306d70a20f36424011f7c1330', 'points_paid': 2.5, 'sd_hash': '9e98f5e4bd4393b45a41839fe72a4df1f94b029e2339f8b14b9eaa9fec2be5245b160a59cfcc80fa86c1f91da67d5581', 'stopped': False, 'stream_hash': '7403ff9319292bdb022d66d0b88401775c6bb355fc0a75fe01452950fb19642ba58d492d34e24d8bf58375e6c2fca16f', 'stream_name': 'including.mp3', 'suggested_file_name': 'including.mp3', 'txid': '345bfc7b4a0c042b14474ac2cecf099236aec5a6730943630ffb176e7b421121', 'written_bytes': 8438, }, { 'channel_claim_id': '59766071ea2df38b4a751834a77246bf8ff8071d', 'channel_name': '@bfrederick', 'claim_id': '1c08967d515ac2a5fd3cb4e477fde18bddde22e2', 'claim_name': 'at-first-skill-agency', 'completed': True, 'download_directory': '/agree', 'download_path': '/agree/different.json', 'file_name': 'different.json', 'key': b'\xc06\xb9\x8e\x00S\xdbX\x1cC\xbd+\xfc\xea\xc96', 'metadata': {'author': 'bfrederick', 'nsfw': True}, 'mime_type': 'image/svg+xml', 'nout': 2975, 'outpoint': '62bb992e11c562e8064aec094e2b4eefb422e50b13ae49dc10f440a60f8e99bc', 'points_paid': 9.1, 'sd_hash': '3d809caf1266ec1ab78cc046a62f388434b6c59f85844500f59a1c75b4303b9ea27c532a231556e8b9776c544677bdf7', 'stopped': True, 'stream_hash': '9ecb8cf7dca7260f90666f05c88882017c786d31b572f3cba9447099ca9b49cdcb93f801db2249b7d32ff44ca6ffd69c', 'stream_name': 'north.doc', 'suggested_file_name': 'north.doc', 'txid': '98e12bce3a5db96e3513f1ff45afca8b69d61324117d6e839075ac512dd86251', 'written_bytes': 1929, }, { 'channel_claim_id': '046c8a762cd158a6e5b112d7d9c9e4b778f27388', 'channel_name': '@heidiherring', 'claim_id': '3d7573264601af7b5402cd54a66c13f2f93f296f', 'claim_name': 'drop-hot-military-drive', 'completed': False, 'download_directory': '/letter', 'download_path': '/letter/than.ppt', 'file_name': 'than.ppt', 'key': b'\n\xa7\xb3\x05\xab\x8e\xcc\n\xdcn\xd9\x81\xf3m\xf1t', 'metadata': {'author': 'heidiherring', 'nsfw': True}, 'mime_type': 'text/javascript', 'nout': 3452, 'outpoint': '5e8b55ffe59774366804c384f632f728f769bad566c784c6a23f1081724d00ea', 'points_paid': 5.9, 'sd_hash': 'c945b0acaf1c97dd7f262cf73cc3813ab6552f695ab00a445ca07a2a4da43bf3bb020cc7e338b484405d0865ef836480', 'stopped': False, 'stream_hash': '692ea08506d13422c875ce9d49fb9fe90b828d259d46c53adffa68a045e3852d4763b90ef94ef3864a1164bcab7eefa0', 'stream_name': 'few.css', 'suggested_file_name': 'few.css', 'txid': '9ef901facbf8bc133cfe67793fe4423c048753dab8370d987f6f552ed6483bbb', 'written_bytes': 9498, }, { 'channel_claim_id': 'adc87fa84d601aa1760d0f4585f02e60bc82c703', 'channel_name': '@xsteele', 'claim_id': '9022e1fd14646f6a4f6708566fbb6f6ac10ba3d5', 'claim_name': 'mean-television-miss-yourself', 'completed': True, 'download_directory': '/its', 'download_path': '/its/add.mp3', 'file_name': 'add.mp3', 'key': b'\xdc\xd1\xf1i!\x85\xc6\xc8\\\xe0\xd7\xc0\xceN