replace 'werkzeug.urls.url_join' usage with 'urllib.parse.urljoin'

conftest.py

@@ -113,7 +113,7 @@ def pytest_generate_tests(metafunc):
 import logging
 import fs
 import secrets
-from werkzeug.urls import url_join
+from urllib.parse import urljoin
 
 ROOT_DIR = '{constants.ROOT_DIR}'
 # directory where mss output files are stored
@@ -147,7 +147,7 @@ def pytest_generate_tests(metafunc):
 # enable verification by Mail
 MAIL_ENABLED = False
 
-SQLALCHEMY_DB_URI = 'sqlite:///' + url_join(DATA_DIR, 'mscolab.db')
+SQLALCHEMY_DB_URI = 'sqlite:///' + urljoin(DATA_DIR, 'mscolab.db')
 
 # mscolab file upload settings
 UPLOAD_FOLDER = fs.path.join(DATA_DIR, 'uploads')

mslib/msui/mscolab.py

@@ -39,10 +39,10 @@
 import requests
 import re
 import urllib.request
+from urllib.parse import urljoin
 
 from fs import open_fs
 from PIL import Image
-from werkzeug.urls import url_join
 from keyring.errors import NoKeyringError, PasswordSetError, InitError
 
 from mslib.msui import flighttrack as ft
@@ -96,7 +96,7 @@ def unarchive_operation(self):
                 "token": self.mscolab.token,
                 "op_id": self.archived_op_id,
             }
-            url = url_join(self.mscolab.mscolab_server_url, 'set_last_used')
+            url = urljoin(self.mscolab.mscolab_server_url, 'set_last_used')
             try:
                 res = requests.post(url, data=data, timeout=(2, 10))
             except requests.exceptions.RequestException as e:
@@ -218,7 +218,7 @@ def connect_handler(self):
             s = requests.Session()
             s.auth = auth
             s.headers.update({'x-test': 'true'})
-            r = s.get(url_join(url, 'status'), timeout=(2, 10))
+            r = s.get(urljoin(url, 'status'), timeout=(2, 10))
             if r.status_code == 401:
                 self.set_status("Error", 'Server authentication data were incorrect.')
             elif r.status_code == 200:
@@ -1013,7 +1013,7 @@ def handle_delete_operation(self):
                         "token": self.token,
                         "op_id": self.active_op_id
                     }
-                    url = url_join(self.mscolab_server_url, 'delete_operation')
+                    url = urljoin(self.mscolab_server_url, 'delete_operation')
                     try:
                         res = requests.post(url, data=data, timeout=(2, 10))
                     except requests.exceptions.RequestException as e:
@@ -1046,7 +1046,7 @@ def handle_leave_operation(self):
                     "op_id": self.active_op_id,
                     "selected_userids": json.dumps([self.user["id"]])
                 }
-                url = url_join(self.mscolab_server_url, "delete_bulk_permissions")
+                url = urljoin(self.mscolab_server_url, "delete_bulk_permissions")
                 res = requests.post(url, data=data, timeout=(2, 10))
                 if res.text != "False":
                     res = res.json()
@@ -1093,7 +1093,7 @@ def change_category_handler(self):
                     "attribute": 'category',
                     "value": entered_operation_category
                 }
-                url = url_join(self.mscolab_server_url, 'update_operation')
+                url = urljoin(self.mscolab_server_url, 'update_operation')
                 r = requests.post(url, data=data, timeout=(2, 10))
                 if r.text == "True":
                     self.active_operation_category = entered_operation_category
@@ -1123,7 +1123,7 @@ def change_description_handler(self):
                     "attribute": 'description',
                     "value": entered_operation_desc
                 }
-                url = url_join(self.mscolab_server_url, 'update_operation')
+                url = urljoin(self.mscolab_server_url, 'update_operation')
                 r = requests.post(url, data=data, timeout=(2, 10))
                 if r.text == "True":
                     # Update active operation description label
@@ -1154,7 +1154,7 @@ def rename_operation_handler(self):
                     "attribute": 'path',
                     "value": entered_operation_name
                 }
-                url = url_join(self.mscolab_server_url, 'update_operation')
+                url = urljoin(self.mscolab_server_url, 'update_operation')
                 r = requests.post(url, data=data, timeout=(2, 10))
                 if r.text == "True":
                     # Update active operation name
@@ -1545,7 +1545,7 @@ def archive_operation(self):
                     "op_id": self.active_op_id,
                     "days": 31,
                 }
-                url = url_join(self.mscolab_server_url, 'set_last_used')
+                url = urljoin(self.mscolab_server_url, 'set_last_used')
                 try:
                     res = requests.post(url, data=data, timeout=(2, 10))
                 except requests.exceptions.RequestException as e:

mslib/msui/mscolab_admin_window.py

@@ -27,7 +27,7 @@
 import json
 
 import requests
-from werkzeug.urls import url_join
+from urllib.parse import urljoin
 
 from PyQt5 import QtCore, QtWidgets
 from mslib.utils.verify_user_token import verify_user_token
@@ -175,7 +175,7 @@ def set_label_text(self):
             "token": self.token,
             "op_id": self.op_id
         }
-        url = url_join(self.mscolab_server_url, "/creator_of_operation")
+        url = urljoin(self.mscolab_server_url, "/creator_of_operation")
         r = requests.get(url, data=data, timeout=(2, 10))
         if r.text != "False":
             _json = json.loads(r.text)
@@ -188,7 +188,7 @@ def load_import_operations(self):
             "token": self.token,
             "op_id": self.op_id
         }
-        url = url_join(self.mscolab_server_url, "operations")
+        url = urljoin(self.mscolab_server_url, "operations")
         r = requests.get(url, data=data, timeout=(2, 10))
         if r.text != "False":
             _json = json.loads(r.text)
@@ -202,7 +202,7 @@ def load_users_without_permission(self):
                 "token": self.token,
                 "op_id": self.op_id
             }
-            url = url_join(self.mscolab_server_url, "users_without_permission")
+            url = urljoin(self.mscolab_server_url, "users_without_permission")
             res = requests.get(url, data=data, timeout=(2, 10))
             if res.text != "False":
                 res = res.json()
@@ -227,7 +227,7 @@ def load_users_with_permission(self):
                 "token": self.token,
                 "op_id": self.op_id
             }
-            url = url_join(self.mscolab_server_url, "users_with_permission")
+            url = urljoin(self.mscolab_server_url, "users_with_permission")
             res = requests.get(url, data=data, timeout=(2, 10))
             if res.text != "False":
                 res = res.json()
@@ -259,7 +259,7 @@ def add_selected_users(self):
                 "selected_userids": json.dumps(selected_userids),
                 "selected_access_level": selected_access_level
             }
-            url = url_join(self.mscolab_server_url, "add_bulk_permissions")
+            url = urljoin(self.mscolab_server_url, "add_bulk_permissions")
             res = requests.post(url, data=data, timeout=(2, 10))
             if res.text != "False":
                 res = res.json()
@@ -290,7 +290,7 @@ def modify_selected_users(self):
                 "selected_userids": json.dumps(selected_userids),
                 "selected_access_level": selected_access_level
             }
-            url = url_join(self.mscolab_server_url, "modify_bulk_permissions")
+            url = urljoin(self.mscolab_server_url, "modify_bulk_permissions")
             res = requests.post(url, data=data, timeout=(2, 10))
             if res.text != "False":
                 res = res.json()
@@ -318,7 +318,7 @@ def delete_selected_users(self):
                 "op_id": self.op_id,
                 "selected_userids": json.dumps(selected_userids)
             }
-            url = url_join(self.mscolab_server_url, "delete_bulk_permissions")
+            url = urljoin(self.mscolab_server_url, "delete_bulk_permissions")
             res = requests.post(url, data=data, timeout=(2, 10))
             if res.text != "False":
                 res = res.json()
@@ -343,7 +343,7 @@ def import_permissions(self):
                 "current_op_id": self.op_id,
                 "import_op_id": import_op_id
             }
-            url = url_join(self.mscolab_server_url, 'import_permissions')
+            url = urljoin(self.mscolab_server_url, 'import_permissions')
             res = requests.post(url, data=data, timeout=(2, 10))
             if res.text != "False":
                 res = res.json()

mslib/msui/mscolab_chat.py

@@ -31,7 +31,7 @@
 import requests
 from markdown import Markdown
 from markdown.extensions import Extension
-from werkzeug.urls import url_join
+from urllib.parse import urljoin
 
 from mslib.mscolab.message_type import MessageType
 from PyQt5 import QtCore, QtGui, QtWidgets
@@ -275,7 +275,7 @@ def send_message(self):
                 "op_id": self.op_id,
                 "message_type": int(self.attachment_type)
             }
-            url = url_join(self.mscolab_server_url, 'message_attachment')
+            url = urljoin(self.mscolab_server_url, 'message_attachment')
             try:
                 requests.post(url, data=data, files=files, timeout=(2, 10))
             except requests.exceptions.ConnectionError:
@@ -332,7 +332,7 @@ def load_users(self):
             "token": self.token,
             "op_id": self.op_id
         }
-        url = url_join(self.mscolab_server_url, 'authorized_users')
+        url = urljoin(self.mscolab_server_url, 'authorized_users')
         r = requests.get(url, data=data, timeout=(2, 10))
         if r.text != "False":
             self.collaboratorsList.clear()
@@ -352,7 +352,7 @@ def load_all_messages(self):
             "timestamp": datetime.datetime(1970, 1, 1).strftime("%Y-%m-%d, %H:%M:%S")
         }
         # returns an array of messages
-        url = url_join(self.mscolab_server_url, "messages")
+        url = urljoin(self.mscolab_server_url, "messages")
 
         res = requests.get(url, data=data, timeout=(2, 10))
         if res.text != "False":
@@ -470,10 +470,10 @@ def setup_image_message_box(self):
         MAX_WIDTH = MAX_HEIGHT = 300
         self.messageBox = QtWidgets.QLabel()
         if '\\' in self.attachment_path:
-            img_url = url_join(self.chat_window.mscolab_server_url,
-                               self.attachment_path.replace('\\', '/').split('colabdata')[1])
+            img_url = urljoin(self.chat_window.mscolab_server_url,
+                              self.attachment_path.replace('\\', '/').split('colabdata')[1])
         else:
-            img_url = url_join(self.chat_window.mscolab_server_url, self.attachment_path)
+            img_url = urljoin(self.chat_window.mscolab_server_url, self.attachment_path)
         data = requests.get(img_url, timeout=(2, 10)).content
         image = QtGui.QImage()
         image.loadFromData(data)
@@ -505,7 +505,7 @@ def get_text_browser(self, text):
 
     def setup_text_message_box(self):
         if self.message_type == MessageType.DOCUMENT:
-            doc_url = url_join(self.chat_window.mscolab_server_url, self.attachment_path)
+            doc_url = urljoin(self.chat_window.mscolab_server_url, self.attachment_path)
             file_name = fs.path.basename(self.attachment_path)
             self.message_text = f"Document: [{file_name}]({doc_url})"
         self.messageBox = self.get_text_browser(self.message_text)
@@ -653,7 +653,7 @@ def handle_download_action(self):
         if self.message_type == MessageType.DOCUMENT:
             file_path = get_save_filename(self, "Save Document", default_filename, f"Document (*{file_ext})")
             if file_path is not None:
-                file_content = requests.get(url_join(self.chat_window.mscolab_server_url, self.attachment_path),
+                file_content = requests.get(urljoin(self.chat_window.mscolab_server_url, self.attachment_path),
                                             timeout=(2, 10)).content
                 with open(file_path, "wb") as f:
                     f.write(file_content)

mslib/msui/mscolab_version_history.py

@@ -29,7 +29,8 @@
 import json
 
 import requests
-from werkzeug.urls import url_encode, url_join
+from urllib.parse import urljoin
+from werkzeug.urls import url_encode
 
 from mslib.utils.verify_user_token import verify_user_token
 from mslib.msui.flighttrack import WaypointsTableModel
@@ -111,7 +112,7 @@ def load_current_waypoints(self):
                 "token": self.token,
                 "op_id": self.op_id
             }
-            url = url_join(self.mscolab_server_url, 'get_operation_by_id')
+            url = urljoin(self.mscolab_server_url, 'get_operation_by_id')
             res = requests.get(url, data=data, timeout=(2, 10))
             if res.text != "False":
                 xml_content = json.loads(res.text)["content"]
@@ -138,7 +139,7 @@ def load_all_changes(self):
                 named_version_only = True
             query_string = url_encode({"named_version": named_version_only})
             url_path = f'get_all_changes?{query_string}'
-            url = url_join(self.mscolab_server_url, url_path)
+            url = urljoin(self.mscolab_server_url, url_path)
             r = requests.get(url, data=data, timeout=(2, 10))
             if r.text != "False":
                 changes = json.loads(r.text)["changes"]
@@ -180,7 +181,7 @@ def preview_change(self, current_item, previous_item):
                 "token": self.token,
                 "ch_id": current_item.id
             }
-            url = url_join(self.mscolab_server_url, 'get_change_content')
+            url = urljoin(self.mscolab_server_url, 'get_change_content')
             res = requests.get(url, data=data, timeout=(2, 10))
             if res.text != "False":
                 res = res.json()
@@ -206,7 +207,7 @@ def request_set_version_name(self, version_name, ch_id):
                 "ch_id": ch_id,
                 "op_id": self.op_id
             }
-            url = url_join(self.mscolab_server_url, 'set_version_name')
+            url = urljoin(self.mscolab_server_url, 'set_version_name')
             res = requests.post(url, data=data, timeout=(2, 10))
             return res
         else:
@@ -273,7 +274,7 @@ def handle_undo(self):
                     "token": self.token,
                     "ch_id": self.changes.currentItem().id
                 }
-                url = url_join(self.mscolab_server_url, 'undo')
+                url = urljoin(self.mscolab_server_url, 'undo')
                 r = requests.post(url, data=data, timeout=(2, 10))
                 if r.text != "False":
                     # reload windows

tests/_test_mscolab/test_sockets_manager.py

@@ -29,8 +29,8 @@
 import socketio
 import datetime
 import requests
+from urllib.parse import urljoin
 
-from werkzeug.urls import url_join
 from mslib.msui.icons import icons
 from mslib.mscolab.conf import mscolab_settings
 from tests.utils import mscolab_check_free_port, LiveSocketTestCase
@@ -233,7 +233,7 @@ def test_get_messages_api(self):
             "timestamp": datetime.datetime(1970, 1, 1).strftime("%Y-%m-%d, %H:%M:%S")
         }
         # returns an array of messages
-        url = url_join(self.url, 'messages')
+        url = urljoin(self.url, 'messages')
         res = requests.get(url, data=data, timeout=(2, 10)).json()
         assert len(res["messages"]) == 2
 
@@ -269,7 +269,7 @@ def test_edit_message(self):
             "timestamp": datetime.datetime(1970, 1, 1).strftime("%Y-%m-%d, %H:%M:%S")
         }
         # returns an array of messages
-        url = url_join(self.url, 'messages')
+        url = urljoin(self.url, 'messages')
         res = requests.get(url, data=data, timeout=(2, 10)).json()
         assert len(res["messages"]) == 1
         messages = res["messages"][0]
@@ -307,7 +307,7 @@ def test_upload_file(self):
             "op_id": self.operation.id,
             "message_type": int(MessageType.IMAGE)
         }
-        url = url_join(self.url, 'message_attachment')
+        url = urljoin(self.url, 'message_attachment')
         requests.post(url, data=data, files=files, timeout=(2, 10))
         upload_dir = os.path.join(mscolab_settings.UPLOAD_FOLDER, str(self.user.id))
         assert os.path.exists(upload_dir)