from unittest import mock import json from twisted.internet import defer from twisted.trial import unittest from lbrynet.extras.compat import f2d from lbrynet.schema.decode import smart_decode from lbrynet.extras.daemon.storage import SQLiteStorage from lbrynet.extras.daemon.ComponentManager import ComponentManager from lbrynet.extras.daemon.Components import DATABASE_COMPONENT, DHT_COMPONENT, WALLET_COMPONENT from lbrynet.extras.daemon.Components import HASH_ANNOUNCER_COMPONENT, REFLECTOR_COMPONENT from lbrynet.extras.daemon.Components import UPNP_COMPONENT, BLOB_COMPONENT from lbrynet.extras.daemon.Components import PEER_PROTOCOL_SERVER_COMPONENT, EXCHANGE_RATE_MANAGER_COMPONENT from lbrynet.extras.daemon.Components import RATE_LIMITER_COMPONENT, HEADERS_COMPONENT, FILE_MANAGER_COMPONENT from lbrynet.extras.daemon.Daemon import Daemon as LBRYDaemon from lbrynet.blob.EncryptedFileDownloader import ManagedEncryptedFileDownloader from lbrynet.blob.EncryptedFileStatusReport import EncryptedFileStatusReport from lbrynet.extras.wallet import LbryWalletManager from torba.client.wallet import Wallet from lbrynet.conf import Config from lbrynet.p2p.PaymentRateManager import OnlyFreePaymentsManager from tests import test_utils from tests.mocks import 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, REFLECTOR_COMPONENT, HASH_ANNOUNCER_COMPONENT, EXCHANGE_RATE_MANAGER_COMPONENT, BLOB_COMPONENT, HEADERS_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=LbryWalletManager) daemon.wallet_manager.wallet = mock.Mock(spec=Wallet) daemon.wallet_manager.wallet.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.file_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.wallet_manager.resolve = lambda *_: defer.succeed( {"test": {'claim': {'value': migrated.claim_dict}}}) return daemon class TestCostEst(unittest.TestCase): def setUp(self): test_utils.reset_time(self) @defer.inlineCallbacks 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) @defer.inlineCallbacks def test_fee_and_ungenerous_data(self): conf = Config(is_generous_host=False) size = 10000000 fake_fee_amount = 4.5 correct_result = size / 10 ** 6 * conf.data_rate + fake_fee_amount daemon = get_test_daemon(conf, with_fee=True) result = yield f2d(daemon.get_est_cost("test", size)) self.assertEqual(result, round(correct_result, 1)) @defer.inlineCallbacks 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) @defer.inlineCallbacks def test_ungenerous_data_and_no_fee(self): conf = Config(is_generous_host=False) size = 10000000 correct_result = size / 10 ** 6 * conf.data_rate daemon = get_test_daemon(conf) result = yield f2d(daemon.get_est_cost("test", size)) self.assertEqual(result, round(correct_result, 1)) 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 @defer.inlineCallbacks 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." 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.setup()) @defer.inlineCallbacks 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)) self.assertEqual(self.test_points_paid, [f['points_paid'] for f in file_list]) @defer.inlineCallbacks 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)) self.assertEqual(self.test_points_paid, [f['points_paid'] for f in file_list]) @defer.inlineCallbacks 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)) self.assertEqual(list(reversed(self.test_points_paid)), [f['points_paid'] for f in file_list]) @defer.inlineCallbacks 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)) self.assertEqual(self.test_file_names, [f['file_name'] for f in file_list]) @defer.inlineCallbacks 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)) self.assertEqual(self.test_file_names, [f['file_name'] for f in file_list]) @defer.inlineCallbacks 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)) self.assertEqual(list(reversed(self.test_file_names)), [f['file_name'] for f in file_list]) @defer.inlineCallbacks 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: 'file_name={}, points_paid={}'.format(f['file_name'], f['points_paid']) sort_options = ['file_name,asc', 'points_paid,desc'] file_list = yield f2d(self.test_daemon.jsonrpc_file_list(sort=sort_options)) 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)) 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)) 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()) self.assertNotEqual(expected, [format_result(r) for r in file_list]) @defer.inlineCallbacks 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)) 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)) 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()) self.assertNotEqual(self.test_authors, extract_authors(file_list)) @defer.inlineCallbacks 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)) 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)) @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