Remove get_infohash_from_magnet

src/tribler/core/components/libtorrent/restapi/tests/test_torrentinfo_endpoint.py

@@ -207,18 +207,3 @@ async def test_torrentinfo_endpoint_timeout_error(mocked_query_http_uri: AsyncMo
     info = await endpoint.get_torrent_info(request)
 
     assert info.status == HTTP_INTERNAL_SERVER_ERROR
-
-
-def test_get_infohash_from_magnet(endpoint: TorrentInfoEndpoint):
-    # Test that we can get infohash from magnet link
-    magnet = f'magnet:?xt=urn:btih:{"A" * 40}'
-    expected = unhexlify("A" * 40)
-    assert endpoint.get_infohash_from_magnet(magnet) == expected
-
-
-def test_get_infohash_from_wrong_magnet(endpoint: TorrentInfoEndpoint):
-    # Test that the `get_infohash_from_magnet` method returns None if the magnet link is wrong (for exampple it
-    # contains `ed2k` hash)
-    magnet = f'magnet:?xt=urn:ed2k:{"A" * 40}'
-    expected = None
-    assert endpoint.get_infohash_from_magnet(magnet) == expected

src/tribler/core/components/libtorrent/restapi/torrentinfo_endpoint.py

@@ -3,7 +3,6 @@
 from asyncio.exceptions import TimeoutError as AsyncTimeoutError
 from copy import deepcopy
 from ssl import SSLError
-from typing import Optional, Union
 
 from aiohttp import ClientConnectorError, ClientResponseError, ClientSession, ServerConnectionError, web
 from aiohttp_apispec import docs
@@ -52,7 +51,6 @@ class TorrentInfoEndpoint(RESTEndpoint):
     def __init__(self, download_manager: DownloadManager):
         super().__init__()
         self.download_manager = download_manager
-        self.last_error: Optional[Exception] = None
 
     def setup_routes(self):
         self.app.add_routes([web.get('', self.get_torrent_info)])
@@ -108,18 +106,28 @@ async def get_torrent_info(self, request):
                 return RESTResponse({"error": str(e)}, status=HTTP_INTERNAL_SERVER_ERROR)
 
             if response.startswith(b'magnet'):
-                infohash = self.get_infohash_from_magnet(response)
-                if not infohash:
-                    return RESTResponse({"error": str(self.last_error)}, status=HTTP_INTERNAL_SERVER_ERROR)
+                try:
+                    _, infohash, _ = parse_magnetlink(response)
+                except RuntimeError as e:
+                    return RESTResponse(
+                        {"error": f'Error while getting an ingo hash 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)
             else:
                 metainfo = bdecode_compat(response)
         elif scheme == MAGNET_SCHEME:
             self._logger.info(f'{MAGNET_SCHEME} scheme detected')
 
-            infohash = self.get_infohash_from_magnet(uri)
-            if not infohash:
-                return RESTResponse({"error": str(self.last_error)}, status=HTTP_BAD_REQUEST)
+            try:
+                _, infohash, _ = parse_magnetlink(uri)
+            except RuntimeError as e:
+                return RESTResponse(
+                    {"error": f'Error while getting an ingo hash from magnet: {e.__class__.__name__}: {e}'},
+                    status=HTTP_BAD_REQUEST
+                )
+
             metainfo = await self.download_manager.get_metainfo(infohash, timeout=60, hops=hops, url=uri)
         else:
             return RESTResponse({"error": "invalid uri"}, status=HTTP_BAD_REQUEST)
@@ -153,19 +161,3 @@ async def get_torrent_info(self, request):
             encoded_metainfo, ignore_errors=True), ensure_ascii=False).encode('utf-8'))
         return RESTResponse({"metainfo": encoded_metainfo,
                              "download_exists": download and not download_is_metainfo_request})
-
-    def get_infohash_from_magnet(self, magnet: Union[str, bytes]) -> Optional[bytes]:
-        """ Get the infohash from a magnet link.
-        Args:
-            magnet: A string representing the magnet link.
-        Returns:
-            bytes: The infohash of the magnet link. If the magnet link is invalid, None is returned.
-        """
-        infohash = None
-        try:
-            _, infohash, _ = parse_magnetlink(magnet)
-        except RuntimeError as e:
-            self._logger.warning(f'Error while getting an ingo hash from magnet: {e.__class__.__name__}: {e}',
-                                 exc_info=e)
-            self.last_error = e
-        return infohash