Refactor rest utils

.flake8

@@ -1,6 +1,7 @@
 [flake8]
 max-line-length = 120
 extend-ignore =
+  I201, # Missing newline between import groups
   I202, # Additional newline in a group of imports
   E127, # continuation line over-indented for visual indent
   E128, # continuation line under-indented for visual indent
@@ -12,3 +13,4 @@ extend-ignore =
   E722, # Pylint-checked, do not use bare 'except'
   E501 # Pylint-checked, line too long
 import-order-style=pycharm
+application_import_names = tribler

src/tribler/core/components/libtorrent/download_manager/download_manager.py

@@ -32,8 +32,8 @@
     HTTPS_SCHEME,
     HTTP_SCHEME,
     MAGNET_SCHEME,
-    scheme_from_uri,
-    uri_to_path,
+    scheme_from_url,
+    url_to_path,
 )
 from tribler.core.utilities.simpledefs import DLSTATUS_SEEDING, MAX_LIBTORRENT_RATE_LIMIT, NTFY, STATEDIR_CHECKPOINT_DIR
 from tribler.core.utilities.unicode import hexlify
@@ -512,7 +512,7 @@ def _map_call_on_ltsessions(self, hops, funcname, *args, **kwargs):
             getattr(self.get_session(hops), funcname)(*args, **kwargs)
 
     async def start_download_from_uri(self, uri, config=None):
-        scheme = scheme_from_uri(uri)
+        scheme = scheme_from_url(uri)
 
         if scheme in (HTTP_SCHEME, HTTPS_SCHEME):
             tdef = await TorrentDef.load_from_url(uri)
@@ -527,7 +527,7 @@ async def start_download_from_uri(self, uri, config=None):
                 tdef = TorrentDefNoMetainfo(infohash, "Unknown name" if name is None else name, url=uri)
             return self.start_download(tdef=tdef, config=config)
         if scheme == FILE_SCHEME:
-            file = uri_to_path(uri)
+            file = url_to_path(uri)
             return self.start_download(torrent_file=file, config=config)
         raise Exception("invalid uri")
 

src/tribler/core/components/libtorrent/libtorrent_component.py

@@ -2,7 +2,7 @@
 from tribler.core.components.key.key_component import KeyComponent
 from tribler.core.components.libtorrent.download_manager.download_manager import DownloadManager
 from tribler.core.components.socks_servers.socks_servers_component import SocksServersComponent
-from tribler.core.utilities.rest_utils import path_to_uri
+from tribler.core.utilities.rest_utils import path_to_url
 
 
 class LibtorrentComponent(Component):
@@ -34,7 +34,7 @@ async def run(self):
 
         if config.gui_test_mode:
             from tribler.core.tests.tools.common import TORRENT_WITH_DIRS  # pylint: disable=import-outside-toplevel
-            uri = path_to_uri(TORRENT_WITH_DIRS)
+            uri = path_to_url(TORRENT_WITH_DIRS)
             await self.download_manager.start_download_from_uri(uri)
 
     async def shutdown(self):

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

@@ -2,18 +2,16 @@
 import os
 from unittest.mock import Mock
 
+import pytest
 from aiohttp.web_app import Application
-
 from ipv8.util import fail, succeed
 
-import pytest
-
 from tribler.core.components.libtorrent.download_manager.download_state import DownloadState
 from tribler.core.components.libtorrent.restapi.downloads_endpoint import DownloadsEndpoint, get_extended_status
 from tribler.core.components.restapi.rest.base_api_test import do_request
 from tribler.core.components.restapi.rest.rest_manager import error_middleware
 from tribler.core.tests.tools.common import TESTS_DATA_DIR
-from tribler.core.utilities.rest_utils import HTTP_SCHEME, path_to_uri
+from tribler.core.utilities.rest_utils import HTTP_SCHEME, path_to_url
 from tribler.core.utilities.simpledefs import (
     DLSTATUS_CIRCUITS,
     DLSTATUS_DOWNLOADING,
@@ -189,7 +187,7 @@ async def test_start_download_from_file(test_download, mock_dlmgr, rest_api):
     Testing whether we can start a download from a file
     """
     mock_dlmgr.start_download_from_uri = lambda *_, **__: succeed(test_download)
-    uri = path_to_uri(TESTS_DATA_DIR / 'video.avi.torrent')
+    uri = path_to_url(TESTS_DATA_DIR / 'video.avi.torrent')
     expected_json = {'started': True, 'infohash': 'c9a19e7fe5d9a6c106d6ea3c01746ac88ca3c7a5'}
     await do_request(rest_api, 'downloads', expected_code=200, request_type='PUT',
                      post_data={'uri': uri}, expected_json=expected_json)
@@ -204,7 +202,7 @@ def mocked_start_download(*_, config=None):
         return succeed(test_download)
 
     mock_dlmgr.start_download_from_uri = mocked_start_download
-    uri = path_to_uri(TESTS_DATA_DIR / 'video.avi.torrent')
+    uri = path_to_url(TESTS_DATA_DIR / 'video.avi.torrent')
     post_data = {'uri': uri, 'selected_files': [0]}
     expected_json = {'started': True, 'infohash': 'c9a19e7fe5d9a6c106d6ea3c01746ac88ca3c7a5'}
     await do_request(rest_api, 'downloads', expected_code=200, request_type='PUT',
@@ -466,7 +464,7 @@ async def test_stream_unknown_download(mock_dlmgr, rest_api):
     Testing whether the API returns error 404 if we stream a non-existent download
     """
     mock_dlmgr.get_download = lambda _: None
-    await do_request(rest_api, f'downloads/abcd/stream/0',
+    await do_request(rest_api, 'downloads/abcd/stream/0',
                      headers={'range': 'bytes=0-'}, expected_code=404, request_type='GET')
 
 

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

@@ -4,12 +4,10 @@
 from unittest.mock import MagicMock, patch
 from urllib.parse import quote_plus, unquote_plus
 
+import pytest
 from aiohttp.web_app import Application
-
 from ipv8.util import succeed
 
-import pytest
-
 from tribler.core import notifications
 from tribler.core.components.libtorrent.restapi.torrentinfo_endpoint import TorrentInfoEndpoint
 from tribler.core.components.libtorrent.settings import LibtorrentSettings
@@ -18,7 +16,7 @@
 from tribler.core.components.restapi.rest.base_api_test import do_request
 from tribler.core.components.restapi.rest.rest_manager import error_middleware
 from tribler.core.tests.tools.common import TESTS_DATA_DIR, TESTS_DIR, TORRENT_UBUNTU_FILE, UBUNTU_1504_INFOHASH
-from tribler.core.utilities.rest_utils import path_to_uri
+from tribler.core.utilities.rest_utils import path_to_url
 from tribler.core.utilities.unicode import hexlify
 
 SAMPLE_CHANNEL_FILES_DIR = TESTS_DIR / "data" / "sample_channel"
@@ -62,7 +60,7 @@ async def test_get_torrentinfo_escaped_characters(tmp_path, rest_api):
     source = TORRENT_UBUNTU_FILE
     destination = tmp_path / 'ubuntu%20%21 15.04.torrent'
     shutil.copyfile(source, destination)
-    uri = path_to_uri(destination)
+    uri = path_to_url(destination)
     response = await do_request(rest_api, url='torrentinfo', params={'uri': uri}, expected_code=200)
 
     assert 'metainfo' in response
@@ -74,7 +72,7 @@ async def test_get_torrentinfo(tmp_path, rest_api, endpoint: TorrentInfoEndpoint
     """
 
     def _path(file):
-        return path_to_uri(TESTS_DATA_DIR / file)
+        return path_to_url(TESTS_DATA_DIR / file)
 
     shutil.copyfile(TORRENT_UBUNTU_FILE, tmp_path / 'ubuntu.torrent')
 

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

@@ -26,8 +26,8 @@
     HTTPS_SCHEME,
     HTTP_SCHEME,
     MAGNET_SCHEME,
-    scheme_from_uri,
-    uri_to_path,
+    scheme_from_url,
+    url_to_path,
 )
 from tribler.core.utilities.unicode import hexlify, recursive_unicode
 from tribler.core.utilities.utilities import bdecode_compat, froze_it, parse_magnetlink
@@ -88,10 +88,10 @@ async def get_torrent_info(self, request):
             return RESTResponse({"error": "uri parameter missing"}, status=HTTP_BAD_REQUEST)
 
         metainfo = None
-        scheme = scheme_from_uri(uri)
+        scheme = scheme_from_url(uri)
 
         if scheme == FILE_SCHEME:
-            file = uri_to_path(uri)
+            file = url_to_path(uri)
             try:
                 tdef = TorrentDef.load(file)
                 metainfo = tdef.metainfo

src/tribler/core/components/libtorrent/tests/test_download_api.py

@@ -3,7 +3,7 @@
 import pytest
 
 from tribler.core.tests.tools.common import TORRENT_UBUNTU_FILE
-from tribler.core.utilities.rest_utils import path_to_uri
+from tribler.core.utilities.rest_utils import path_to_url
 from tribler.core.utilities.simpledefs import DLSTATUS_DOWNLOADING
 
 
@@ -19,7 +19,7 @@ async def test_download_torrent_from_url(tmp_path, file_server, download_manager
 @pytest.mark.asyncio
 @pytest.mark.timeout(10)
 async def test_download_torrent_from_file(download_manager):
-    uri = path_to_uri(TORRENT_UBUNTU_FILE)
+    uri = path_to_url(TORRENT_UBUNTU_FILE)
     d = await download_manager.start_download_from_uri(uri)
     await d.wait_for_status(DLSTATUS_DOWNLOADING)
 
@@ -29,6 +29,6 @@ async def test_download_torrent_from_file(download_manager):
 async def test_download_torrent_from_file_with_escaped_characters(download_manager, tmp_path):
     destination = tmp_path / 'ubuntu%20%21 15.04.torrent'
     shutil.copyfile(TORRENT_UBUNTU_FILE, destination)
-    uri = path_to_uri(destination)
+    uri = path_to_url(destination)
     d = await download_manager.start_download_from_uri(uri)
     await d.wait_for_status(DLSTATUS_DOWNLOADING)

src/tribler/core/utilities/rest_utils.py

@@ -10,44 +10,51 @@
 HTTPS_SCHEME = 'https'
 
 
-def path_to_uri(file_path: Union[str, Any]) -> str:
+def path_to_url(file_path: Union[str, Any], _path_cls=Path) -> str:
     """Convert path to url
 
     Example:
         '/path/to/file' -> 'file:///path/to/file'
     """
-    if not isinstance(file_path, str):
-        file_path = str(file_path)
-    return str(URL().build(scheme=FILE_SCHEME, path=file_path))
+    return _path_cls(file_path).as_uri()
 
 
-def uri_to_path(file_uri: str) -> str:
-    """Convert uri to path
+def url_to_path(file_url: str, _path_cls=Path) -> str:
+    """Convert url to path
 
     Example:
         'file:///path/to/file' -> '/path/to/file'
     """
-    path = URL(file_uri).path
+
+    def url_to_path_win():
+        if url.host:
+            # UNC file path, \\server\share\path...
+            # ref: https://docs.microsoft.com/en-us/dotnet/standard/io/file-path-formats
+            _, share, *segments = url.parts
+            return str(_path_cls(rf'\\{url.host}\{share}', *segments))
+        path = url.path.lstrip('/')
+        return str(_path_cls(path))
+
+    url = URL(file_url)
     if os.name == 'nt':
-        # Removes first slash for win OS
-        # see https://github.com/aio-libs/yarl/issues/674
-        return path.lstrip('/')
-    return path
+        return url_to_path_win()
+
+    return str(_path_cls(url.path))
 
 
-def scheme_from_uri(uri: str) -> str:
-    """Get scheme from URI
+def scheme_from_url(url: str) -> str:
+    """Get scheme from URL
 
     Examples:
         'file:///some/file' -> 'file'
         'magnet:link' -> 'magnet'
         'http://en.wikipedia.org' -> 'http'
     """
-    return URL(uri).scheme
+    return URL(url).scheme
 
 
-def uri_is_valid_file(file_uri: str) -> bool:
-    file_path = uri_to_path(file_uri)
+def url_is_valid_file(file_url: str) -> bool:
+    file_path = url_to_path(file_url)
     try:
         return Path(file_path).is_file()
     except OSError:

src/tribler/core/utilities/tests/test_rest_utils.py

@@ -1,61 +1,78 @@
+from pathlib import PurePosixPath, PureWindowsPath
 from unittest.mock import patch
 
 import pytest
 
-from tribler.core.utilities.rest_utils import path_to_uri, scheme_from_uri, uri_is_valid_file, uri_to_path
+from tribler.core.utilities.rest_utils import FILE_SCHEME, HTTP_SCHEME, MAGNET_SCHEME, path_to_url, scheme_from_url, \
+    url_is_valid_file, \
+    url_to_path
 
-NIX_PATHS = [
-    ('/path/to/file', 'file:///path/to/file'),
-    ('/path/to/file with space', 'file:///path/to/file%20with%20space'),
-    ('/path/to/%20%21file', 'file:///path/to/%2520%2521file'),  # See: https://github.com/Tribler/tribler/issues/6700
+POSIX_PATHS = [
+    ('//path/to/file'),
+    ('/path/to/file'),
+    ('/path/to/file with space'),
+    ('/path/to/%20%21file'),  # See: https://github.com/Tribler/tribler/issues/6700
+]
+
+POSIX_URL = [
+    ('file:/path', '/path'),
+    ('file:///path', '/path'),
+    ('file://part/path', '/path'),
 ]
 
 WIN_PATHS = [
-    ('C:\\path\\to\\file', 'file:///C:%5Cpath%5Cto%5Cfile'),
-    ('C:\\path\\to\\file with space', 'file:///C:%5Cpath%5Cto%5Cfile%20with%20space'),
-    ('C:\\path\\to\\%20%21file', 'file:///C:%5Cpath%5Cto%5C%2520%2521file'),
+    (r'C:\path\to\file'),
+    (r'C:\path\to\file with space'),
+    (r'C:\path\to\%20%21file'),
+]
+
+WIN_URL = [
+    ('file://server/share/path', r'\\server\share\path'),
 ]
 
 SCHEMES = [
-    ('file:///path/to/file', 'file'),
-    ('magnet:link', 'magnet'),
-    ('http://en.wikipedia.org', 'http'),
+    ('file:///path/to/file', FILE_SCHEME),
+    ('magnet:link', MAGNET_SCHEME),
+    ('http://en.wikipedia.org', HTTP_SCHEME),
 ]
 
+
 # posix
-@pytest.mark.parametrize('path, uri', NIX_PATHS)
+@pytest.mark.parametrize('path', POSIX_PATHS)
 @patch('os.name', 'posix')
-def test_path_to_uri(path, uri):
-    assert path_to_uri(path) == uri
+def test_round_trip_posix(path):
+    url = path_to_url(path, _path_cls=PurePosixPath)
+    assert url_to_path(url, _path_cls=PurePosixPath) == path
 
 
-@pytest.mark.parametrize('path, uri', NIX_PATHS)
+@pytest.mark.parametrize('url, path', POSIX_URL)
 @patch('os.name', 'posix')
-def test_uri_to_path(path, uri):
-    assert uri_to_path(uri) == path
+def test_posix_url_to_path(url, path):
+    assert url_to_path(url, _path_cls=PurePosixPath) == path
 
 
 # win
-@pytest.mark.parametrize('path, uri', WIN_PATHS)
+@pytest.mark.parametrize('path', WIN_PATHS)
 @patch('os.name', 'nt')
-def test_path_to_uri_win(path, uri):
-    assert path_to_uri(path) == uri
+def test_round_trip_win(path):
+    url = path_to_url(path, _path_cls=PureWindowsPath)
+    assert url_to_path(url, _path_cls=PureWindowsPath) == path
 
 
-@pytest.mark.parametrize('path, uri', WIN_PATHS)
+@pytest.mark.parametrize('url, path', WIN_URL)
 @patch('os.name', 'nt')
-def test_uri_to_path_win(path, uri):
-    assert uri_to_path(uri) == path
+def test_win_url_to_path(url, path):
+    assert url_to_path(url, _path_cls=PureWindowsPath) == path
 
 
 @pytest.mark.parametrize('path, scheme', SCHEMES)
 def test_scheme_from_uri(path, scheme):
-    assert scheme_from_uri(path) == scheme
+    assert scheme_from_url(path) == scheme
 
 
 def test_uri_is_valid_file(tmpdir):
     file_path = tmpdir / '1.txt'
     file_path.write('test')
-    file_uri = path_to_uri(file_path)
-    assert uri_is_valid_file(file_uri)
-    assert not uri_is_valid_file(file_uri + '/*')
+    file_uri = path_to_url(file_path)
+    assert url_is_valid_file(file_uri)
+    assert not url_is_valid_file(file_uri + '/*')

src/tribler/gui/dialogs/startdownloaddialog.py

@@ -8,7 +8,7 @@
 from PyQt5.QtCore import QTimer, pyqtSignal
 from PyQt5.QtWidgets import QFileDialog, QSizePolicy
 
-from tribler.core.utilities.rest_utils import FILE_SCHEME, MAGNET_SCHEME, scheme_from_uri, uri_to_path
+from tribler.core.utilities.rest_utils import FILE_SCHEME, MAGNET_SCHEME, scheme_from_url, url_to_path
 
 from tribler.gui.defs import METAINFO_MAX_RETRIES, METAINFO_TIMEOUT
 from tribler.gui.dialogs.confirmationdialog import ConfirmationDialog
@@ -34,10 +34,10 @@ def __init__(self, parent, download_uri):
         DialogContainer.__init__(self, parent)
 
         torrent_name = download_uri
-        scheme = scheme_from_uri(download_uri)
+        scheme = scheme_from_url(download_uri)
 
         if scheme == FILE_SCHEME:
-            torrent_name = uri_to_path(torrent_name)
+            torrent_name = url_to_path(torrent_name)
         elif scheme == MAGNET_SCHEME:
             torrent_name = unquote_plus(torrent_name)