From 275409f770f16bf579bf63cd960a6d35b1448bf5 Mon Sep 17 00:00:00 2001 From: drew2a Date: Tue, 23 Jan 2024 00:08:25 +0700 Subject: [PATCH] Refactor get_metainfo function and add test cases - Refactored the `get_metainfo` function in `DownloadManager` class to change the type of the `url` parameter from `Optional[Union[str, bytes]]` to `Optional[str]`. - Added a new test case `test_get_torrentinfo_get_metainfo_from_downloaded_magnet` in the file `test_torrentinfo_endpoint.py`. This test case checks if the correct arguments are passed to the `get_metainfo` function when retrieving metainfo from a downloaded magnet. --- .../download_manager/download_manager.py | 2 +- .../tests/test_torrentinfo_endpoint.py | 26 ++++++++++++++++--- .../restapi/torrentinfo_endpoint.py | 12 +++++---- 3 files changed, 31 insertions(+), 9 deletions(-) diff --git a/src/tribler/core/components/libtorrent/download_manager/download_manager.py b/src/tribler/core/components/libtorrent/download_manager/download_manager.py index 7ea1543500a..04308d6b71a 100644 --- a/src/tribler/core/components/libtorrent/download_manager/download_manager.py +++ b/src/tribler/core/components/libtorrent/download_manager/download_manager.py @@ -489,7 +489,7 @@ def update_ip_filter(self, lt_session, ip_addresses): lt_session.set_ip_filter(ip_filter) async def get_metainfo(self, infohash: bytes, timeout: float = 30, hops: Optional[int] = None, - url: Optional[Union[str, bytes]] = None, raise_errors: bool = False) -> Optional[Dict]: + url: Optional[str] = None, raise_errors: bool = False) -> Optional[Dict]: """ Lookup metainfo for a given infohash. The mechanism works by joining the swarm for the infohash connecting to a few peers, and downloading the metadata for the torrent. diff --git a/src/tribler/core/components/libtorrent/restapi/tests/test_torrentinfo_endpoint.py b/src/tribler/core/components/libtorrent/restapi/tests/test_torrentinfo_endpoint.py index 35b2a43d1fd..f53bb181dd6 100644 --- a/src/tribler/core/components/libtorrent/restapi/tests/test_torrentinfo_endpoint.py +++ b/src/tribler/core/components/libtorrent/restapi/tests/test_torrentinfo_endpoint.py @@ -22,6 +22,8 @@ from tribler.core.utilities.rest_utils import path_to_url from tribler.core.utilities.unicode import hexlify +TARGET = 'tribler.core.components.libtorrent.restapi.torrentinfo_endpoint' + SAMPLE_CHANNEL_FILES_DIR = TESTS_DIR / "data" / "sample_channel" @@ -96,7 +98,7 @@ async def mock_http_query(*_): with open(tmp_path / "ubuntu.torrent", 'rb') as f: return f.read() - with patch("tribler.core.components.libtorrent.restapi.torrentinfo_endpoint.query_http_uri", new=mock_http_query): + with patch(f"{TARGET}.query_http_uri", new=mock_http_query): verify_valid_dict(await do_request(rest_api, url, params={'uri': path}, expected_code=200)) path = quote_plus(f'magnet:?xt=urn:btih:{hexlify(UBUNTU_1504_INFOHASH)}' @@ -168,12 +170,30 @@ async def test_get_torrentinfo_invalid_magnet(rest_api): mocked_query_http_uri = AsyncMock(return_value=b'magnet:?xt=urn:ed2k:' + b"any hash") params = {'uri': 'http://any.uri'} - with patch('tribler.core.components.libtorrent.restapi.torrentinfo_endpoint.query_http_uri', mocked_query_http_uri): + with patch(f'{TARGET}.query_http_uri', mocked_query_http_uri): result = await do_request(rest_api, 'torrentinfo', params=params, expected_code=HTTP_INTERNAL_SERVER_ERROR) assert 'error' in result +@patch(f'{TARGET}.unshorten', new=AsyncMock(return_value='http://any.uri')) +@patch(f'{TARGET}.tdef_to_metadata_dict', new=MagicMock()) +@patch.object(TorrentDef, 'load_from_dict', new=MagicMock()) +async def test_get_torrentinfo_get_metainfo_from_downloaded_magnet(rest_api, download_manager: DownloadManager): + # Test that the `get_metainfo` function passes the correct arguments. + magnet = b'magnet:?xt=urn:btih:' + b'0' * 40 + mocked_query_http_uri = AsyncMock(return_value=magnet) + params = {'uri': 'any non empty uri'} + + download_manager.get_metainfo = AsyncMock(return_value={b'info': {}}) + + with patch(f'{TARGET}.query_http_uri', mocked_query_http_uri): + await do_request(rest_api, 'torrentinfo', params=params) + + expected_url = magnet.decode('utf-8') + download_manager.get_metainfo.assert_called_with(b'\x00' * 20, timeout=60, hops=None, url=expected_url) + + async def test_on_got_invalid_metainfo(rest_api): """ Test whether the right operations happen when we receive an invalid metainfo object @@ -194,7 +214,7 @@ async def test_on_got_invalid_metainfo(rest_api): ] -@patch("tribler.core.components.libtorrent.restapi.torrentinfo_endpoint.query_http_uri") +@patch(f"{TARGET}.query_http_uri") @pytest.mark.parametrize("exception", caught_exceptions) async def test_torrentinfo_endpoint_timeout_error(mocked_query_http_uri: AsyncMock, exception: Exception): # Test that in the case of exceptions related to querying HTTP URI specified in this tests, diff --git a/src/tribler/core/components/libtorrent/restapi/torrentinfo_endpoint.py b/src/tribler/core/components/libtorrent/restapi/torrentinfo_endpoint.py index 81ff8d93033..7fdadb140db 100644 --- a/src/tribler/core/components/libtorrent/restapi/torrentinfo_endpoint.py +++ b/src/tribler/core/components/libtorrent/restapi/torrentinfo_endpoint.py @@ -113,8 +113,8 @@ async def get_torrent_info(self, request): {"error": f'Error while getting an infohash from magnet: {e.__class__.__name__}: {e}'}, status=HTTP_INTERNAL_SERVER_ERROR ) - - metainfo = await self.download_manager.get_metainfo(infohash, timeout=60, hops=hops, url=response) + url = response.decode("utf-8") + metainfo = await self.download_manager.get_metainfo(infohash, timeout=60, hops=hops, url=url) else: metainfo = bdecode_compat(response) elif scheme == MAGNET_SCHEME: @@ -140,7 +140,8 @@ async def get_torrent_info(self, request): return RESTResponse({"error": "invalid response"}, status=HTTP_INTERNAL_SERVER_ERROR) # Add the torrent to metadata.db - metadata_dict = tdef_to_metadata_dict(TorrentDef.load_from_dict(metainfo)) + torrent_def = TorrentDef.load_from_dict(metainfo) + metadata_dict = tdef_to_metadata_dict(torrent_def) self.download_manager.notifier[notifications.torrent_metadata_added](metadata_dict) infohash = metadata_dict['infohash'] @@ -151,7 +152,8 @@ async def get_torrent_info(self, request): # Check if the torrent is already in the downloads encoded_metainfo = deepcopy(metainfo) - encoded_metainfo = hexlify(json.dumps(recursive_unicode( - encoded_metainfo, ignore_errors=True), ensure_ascii=False).encode('utf-8')) + ready_for_unicode = recursive_unicode(encoded_metainfo, ignore_errors=True) + json_dump = json.dumps(ready_for_unicode, ensure_ascii=False) + encoded_metainfo = hexlify(json_dump.encode('utf-8')) return RESTResponse({"metainfo": encoded_metainfo, "download_exists": download and not download_is_metainfo_request})