Fix win path

src/tribler-common/tribler_common/rest_utils.py

@@ -1,4 +1,4 @@
-import re
+import os
 from typing import Any, Union
 
 from yarl import URL
@@ -7,16 +7,17 @@
 HTTP_SCHEME = 'http'
 FILE_SCHEME = 'file'
 
-re_fix_win_path = re.compile(r'^(?:/)(\w:)', re.UNICODE)
 
-
-def to_file_uri(file_path: Union[str, Any]) -> str:
+def path_to_uri(file_path: Union[str, Any]) -> str:
     if not isinstance(file_path, str):
         file_path = str(file_path)
     return str(URL().build(scheme=FILE_SCHEME, path=file_path))
 
 
-def from_file_uri(file_uri: str) -> str:
+def uri_to_path(file_uri: str) -> str:
     path = URL(file_uri).path
-    corrected_path = re_fix_win_path.sub(r'\1', path)
-    return corrected_path
+    if os.name == 'nt':
+        # Removes first slash for win OS
+        # see https://github.com/aio-libs/yarl/issues/674
+        return path.lstrip('/')
+    return path

src/tribler-common/tribler_common/tests/test_rest_utils.py

@@ -1,35 +1,43 @@
+from unittest.mock import patch
+
 import pytest
 
-from tribler_common.rest_utils import from_file_uri, re_fix_win_path, to_file_uri
+from tribler_common.rest_utils import path_to_uri, uri_to_path
 
-PATHS = [
-    # nix
+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
-    # win
-    ('E:\\path\\to\\file', 'file:///E:%5Cpath%5Cto%5Cfile'),
 ]
 
-PATHS_TO_FIX = [
-    # correct
-    ('c:\\', 'c:\\'),
-    ('/path', '/path'),
-    # corrupted:
-    ('/c:\\', 'c:\\'),
-    ('/C:\\', 'C:\\'),
+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'),
 ]
 
 
-@pytest.mark.parametrize('path,uri', PATHS)
-def test_to_file_uri(path, uri):
-    assert to_file_uri(path) == uri
+# posix
+@pytest.mark.parametrize('path,uri', NIX_PATHS)
+@patch('os.name', 'posix')
+def test_path_to_uri(path, uri):
+    assert path_to_uri(path) == uri
+
+
+@pytest.mark.parametrize('path,uri', NIX_PATHS)
+@patch('os.name', 'posix')
+def test_uri_to_path(path, uri):
+    assert uri_to_path(uri) == path
 
 
-@pytest.mark.parametrize('path,uri', PATHS)
-def test_from_file_uri(path, uri):
-    assert from_file_uri(uri) == path
+# win
+@pytest.mark.parametrize('path,uri', WIN_PATHS)
+@patch('os.name', 'nt')
+def test_path_to_uri_win(path, uri):
+    assert path_to_uri(path) == uri
 
 
-@pytest.mark.parametrize('original,expected', PATHS_TO_FIX)
-def test_re_fix_win_path(original, expected):
-    assert re_fix_win_path.sub(r'\1', original) == expected
+@pytest.mark.parametrize('path,uri', WIN_PATHS)
+@patch('os.name', 'nt')
+def test_uri_to_path_win(path, uri):
+    assert uri_to_path(uri) == path

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

@@ -16,7 +16,7 @@
 from ipv8.taskmanager import TaskManager, task
 
 from tribler_common.network_utils import default_network_utils
-from tribler_common.rest_utils import FILE_SCHEME, HTTP_SCHEME, MAGNET_SCHEME, from_file_uri
+from tribler_common.rest_utils import FILE_SCHEME, HTTP_SCHEME, MAGNET_SCHEME, uri_to_path
 from tribler_common.simpledefs import DLSTATUS_SEEDING, MAX_LIBTORRENT_RATE_LIMIT, NTFY, STATEDIR_CHECKPOINT_DIR
 
 from tribler_core.components.libtorrent.download_manager.dht_health_manager import DHTHealthManager
@@ -518,7 +518,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 uri.startswith(FILE_SCHEME):
-            file = from_file_uri(uri)
+            file = uri_to_path(uri)
             return self.start_download(torrent_file=file, config=config)
         raise Exception("invalid uri")
 

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

@@ -1,4 +1,4 @@
-from tribler_common.rest_utils import to_file_uri
+from tribler_common.rest_utils import path_to_uri
 
 from tribler_core.components.base import Component
 from tribler_core.components.key.key_component import KeyComponent
@@ -35,7 +35,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 = to_file_uri(TORRENT_WITH_DIRS)
+            uri = path_to_uri(TORRENT_WITH_DIRS)
             await self.download_manager.start_download_from_uri(uri)
 
     async def shutdown(self):

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

@@ -8,7 +8,7 @@
 
 import pytest
 
-from tribler_common.rest_utils import HTTP_SCHEME, to_file_uri
+from tribler_common.rest_utils import HTTP_SCHEME, path_to_uri
 from tribler_common.simpledefs import DLSTATUS_CIRCUITS, DLSTATUS_DOWNLOADING, DLSTATUS_EXIT_NODES, DLSTATUS_STOPPED
 
 from tribler_core.components.libtorrent.download_manager.download_state import DownloadState
@@ -185,7 +185,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 = to_file_uri(TESTS_DATA_DIR / 'video.avi.torrent')
+    uri = path_to_uri(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)
@@ -200,7 +200,7 @@ def mocked_start_download(*_, config=None):
         return succeed(test_download)
 
     mock_dlmgr.start_download_from_uri = mocked_start_download
-    uri = to_file_uri(TESTS_DATA_DIR / 'video.avi.torrent')
+    uri = path_to_uri(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',

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

@@ -10,7 +10,7 @@
 
 import pytest
 
-from tribler_common.rest_utils import to_file_uri
+from tribler_common.rest_utils import path_to_uri
 from tribler_common.simpledefs import NTFY
 
 from tribler_core.components.libtorrent.restapi.torrentinfo_endpoint import TorrentInfoEndpoint
@@ -63,7 +63,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 = to_file_uri(destination)
+    uri = path_to_uri(destination)
     response = await do_request(rest_api, url='torrentinfo', params={'uri': uri}, expected_code=200)
 
     assert 'metainfo' in response
@@ -75,7 +75,7 @@ async def test_get_torrentinfo(tmp_path, rest_api, endpoint: TorrentInfoEndpoint
     """
 
     def _path(file):
-        return to_file_uri(TESTS_DATA_DIR / file)
+        return path_to_uri(TESTS_DATA_DIR / file)
 
     shutil.copyfile(TORRENT_UBUNTU_FILE, tmp_path / 'ubuntu.torrent')
 

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

@@ -10,7 +10,7 @@
 
 from marshmallow.fields import String
 
-from tribler_common.rest_utils import FILE_SCHEME, HTTP_SCHEME, MAGNET_SCHEME, from_file_uri
+from tribler_common.rest_utils import FILE_SCHEME, HTTP_SCHEME, MAGNET_SCHEME, uri_to_path
 from tribler_common.simpledefs import NTFY
 
 from tribler_core.components.libtorrent.download_manager.download_manager import DownloadManager
@@ -83,7 +83,7 @@ async def get_torrent_info(self, request):
 
         metainfo = None
         if uri.startswith(FILE_SCHEME):
-            file = from_file_uri(uri)
+            file = uri_to_path(uri)
             try:
                 tdef = TorrentDef.load(file)
                 metainfo = tdef.metainfo

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

@@ -2,7 +2,7 @@
 
 import pytest
 
-from tribler_common.rest_utils import to_file_uri
+from tribler_common.rest_utils import path_to_uri
 from tribler_common.simpledefs import DLSTATUS_DOWNLOADING
 
 from tribler_core.tests.tools.common import TORRENT_UBUNTU_FILE
@@ -20,7 +20,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 = to_file_uri(TORRENT_UBUNTU_FILE)
+    uri = path_to_uri(TORRENT_UBUNTU_FILE)
     d = await download_manager.start_download_from_uri(uri)
     await d.wait_for_status(DLSTATUS_DOWNLOADING)
 
@@ -30,6 +30,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 = to_file_uri(destination)
+    uri = path_to_uri(destination)
     d = await download_manager.start_download_from_uri(uri)
     await d.wait_for_status(DLSTATUS_DOWNLOADING)

src/tribler-gui/tribler_gui/dialogs/startdownloaddialog.py

@@ -8,7 +8,7 @@
 from PyQt5.QtCore import QTimer, pyqtSignal
 from PyQt5.QtWidgets import QFileDialog, QSizePolicy
 
-from tribler_common.rest_utils import FILE_SCHEME, MAGNET_SCHEME, from_file_uri
+from tribler_common.rest_utils import FILE_SCHEME, MAGNET_SCHEME, uri_to_path
 
 from tribler_gui.defs import METAINFO_MAX_RETRIES, METAINFO_TIMEOUT
 from tribler_gui.dialogs.confirmationdialog import ConfirmationDialog
@@ -35,7 +35,7 @@ def __init__(self, parent, download_uri):
 
         torrent_name = download_uri
         if torrent_name.startswith(FILE_SCHEME):
-            torrent_name = from_file_uri(torrent_name)
+            torrent_name = uri_to_path(torrent_name)
         elif torrent_name.startswith(MAGNET_SCHEME):
             torrent_name = unquote_plus(torrent_name)
 

src/tribler-gui/tribler_gui/start_gui.py

@@ -5,7 +5,7 @@
 from PyQt5.QtCore import QSettings
 
 from tribler_common.logger import load_logger_config
-from tribler_common.rest_utils import to_file_uri
+from tribler_common.rest_utils import path_to_uri
 from tribler_common.sentry_reporter.sentry_reporter import SentryStrategy
 
 from tribler_core.check_os import (
@@ -59,7 +59,7 @@ def run_gui(api_port, api_key, root_state_dir, parsed_args):
             logger.info('GUI Application is already running. Passing a torrent file path to it.')
             for arg in sys.argv[1:]:
                 if os.path.exists(arg) and arg.endswith(".torrent"):
-                    app.send_message(to_file_uri(arg))
+                    app.send_message(path_to_uri(arg))
                 elif arg.startswith('magnet'):
                     app.send_message(arg)
             logger.info('Close the current application.')

src/tribler-gui/tribler_gui/tests/test_gui.py

@@ -12,7 +12,7 @@
 
 import tribler_common
 from tribler_common.reported_error import ReportedError
-from tribler_common.rest_utils import to_file_uri
+from tribler_common.rest_utils import path_to_uri
 from tribler_common.sentry_reporter.sentry_reporter import SentryReporter
 from tribler_common.tag_constants import MIN_TAG_LENGTH
 
@@ -394,7 +394,7 @@ def test_add_download_url(window):
     go_to_and_wait_for_downloads(window)
     window.on_add_torrent_from_url()
     screenshot(window, name="add_torrent_url_dialog")
-    uri = to_file_uri(TORRENT_WITH_DIRS)
+    uri = path_to_uri(TORRENT_WITH_DIRS)
     window.dialog.dialog_widget.dialog_input.setText(uri)
     QTest.mouseClick(window.dialog.buttons[0], Qt.LeftButton)
     QTest.qWait(200)

src/tribler-gui/tribler_gui/tribler_app.py

@@ -4,7 +4,7 @@
 
 from PyQt5.QtCore import QCoreApplication, QEvent, Qt
 
-from tribler_common.rest_utils import to_file_uri
+from tribler_common.rest_utils import path_to_uri
 
 from tribler_core.utilities.unicode import ensure_unicode
 
@@ -41,7 +41,7 @@ def parse_sys_args(self, args):
         for arg in args[1:]:
             if os.path.exists(arg):
                 file_path = ensure_unicode(arg, 'utf8')
-                uri = to_file_uri(file_path)
+                uri = path_to_uri(file_path)
                 self.handle_uri(uri)
             elif arg.startswith('magnet'):
                 self.handle_uri(arg)
@@ -64,6 +64,6 @@ def parse_sys_args(self, args):
 
     def event(self, event):
         if event.type() == QEvent.FileOpen and event.file().endswith(".torrent"):
-            uri = to_file_uri(event.file())
+            uri = path_to_uri(event.file())
             self.handle_uri(uri)
         return QtSingleApplication.event(self, event)

src/tribler-gui/tribler_gui/tribler_window.py

@@ -40,7 +40,7 @@
 
 from tribler_common.network_utils import default_network_utils
 from tribler_common.process_checker import ProcessChecker
-from tribler_common.rest_utils import FILE_SCHEME, MAGNET_SCHEME, from_file_uri
+from tribler_common.rest_utils import FILE_SCHEME, MAGNET_SCHEME, uri_to_path
 from tribler_common.utilities import parse_query
 from tribler_common.version_manager import VersionHistory
 
@@ -634,7 +634,7 @@ def show_add_torrent_to_channel_dialog_from_uri(self, uri):
         def on_add_button_pressed(channel_id):
             post_data = {}
             if uri.startswith(FILE_SCHEME):
-                file_path = from_file_uri(uri)
+                file_path = uri_to_path(uri)
                 with open(file_path) as torrent_file:
                     post_data['torrent'] = b64encode(torrent_file.read()).decode('utf8')
             elif uri.startswith(MAGNET_SCHEME):
@@ -1141,7 +1141,7 @@ def get_urls_from_dragndrop_list(cls, e):
 
     def dragEnterEvent(self, e):
         file_urls = self.get_urls_from_dragndrop_list(e)
-        if any(Path(from_file_uri(fu)).is_file() for fu in file_urls):
+        if any(Path(uri_to_path(fu)).is_file() for fu in file_urls):
             e.accept()
         else:
             e.ignore()
@@ -1150,7 +1150,7 @@ def dropEvent(self, e):
         file_urls = self.get_urls_from_dragndrop_list(e)
 
         for fu in file_urls:
-            path = Path(from_file_uri(fu))
+            path = Path(uri_to_path(fu))
             if path.is_file():
                 self.start_download_from_uri(fu)