kindred/create
1
1
Fork 0
Code Issues 53 Pull Requests Actions Packages Projects 9 Releases 2 Wiki Activity

Addon Manager: pylint cleanup of NetworkManager

Browse Source
This commit is contained in:
Chris Hennes
2022-07-28 15:29:38 -05:00
parent ea06981ccc
commit 9bde00a104
1 changed files with 123 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files

205
src/Mod/AddonManager/NetworkManager.py
View File

@@ -22,7 +22,7 @@
# * *
# ***************************************************************************
"""
#############################################################################
#
# ABOUT NETWORK MANAGER
@@ -52,7 +52,15 @@
# accesses: the blocking_get() function blocks until the network transmission
# is complete, directly returning a QByteArray object with the received data.
# Do not run on the main GUI thread!
"""
import threading
import os
import queue
import itertools
import tempfile
import sys
from typing import Dict, List, Optional
try:
import FreeCAD
@@ -62,18 +70,14 @@ try:
HAVE_FREECAD = True
translate = FreeCAD.Qt.translate
except Exception:
except ImportError:
# For standalone testing support working without the FreeCAD import
HAVE_FREECAD = False
import threading
from PySide2 import QtCore
if FreeCAD.GuiUp:
from PySide2 import QtWidgets
import os
import queue
import itertools
import tempfile
from typing import Dict, List, Optional
# This is the global instance of the NetworkManager that outside code
# should access
@@ -82,7 +86,7 @@ AM_NETWORK_MANAGER = None
HAVE_QTNETWORK = True
try:
from PySide2 import QtNetwork
except Exception:
except ImportError:
if HAVE_FREECAD:
FreeCAD.Console.PrintError(
translate(
@@ -95,12 +99,14 @@ except Exception:
print(
"Could not import QtNetwork, unable to test this file. Try installing the python3-pyside2.qtnetwork package."
)
exit(1)
sys.exit(1)
HAVE_QTNETWORK = False
if HAVE_QTNETWORK:
class QueueItem:
"""A container for information about an item in the network queue."""
def __init__(
self, index: int, request: QtNetwork.QNetworkRequest, track_progress: bool
):
@@ -164,68 +170,22 @@ if HAVE_QTNETWORK:
self.QNAM.setCache(self.diskCache)
self.monitored_connections: List[int] = []
self._setup_proxy()
# A helper connection for our blocking interface
self.completed.connect(self.__synchronous_process_completion)
# Set up our worker connection
self.__request_queued.connect(self.__setup_network_request)
def _setup_proxy(self):
""" Set up the proxy based on user preferences or prompts on command line """
# Set up the proxy, if necesssary:
noProxyCheck = True
systemProxyCheck = False
userProxyCheck = False
proxy_string = ""
if HAVE_FREECAD:
pref = FreeCAD.ParamGet("User parameter:BaseApp/Preferences/Addons")
noProxyCheck = pref.GetBool("NoProxyCheck", noProxyCheck)
systemProxyCheck = pref.GetBool("SystemProxyCheck", systemProxyCheck)
userProxyCheck = pref.GetBool("UserProxyCheck", userProxyCheck)
proxy_string = pref.GetString("ProxyUrl", "")
# Add some error checking to the proxy setup, since for historical reasons they
# are independent booleans, rather than an enumeration:
count = [noProxyCheck, systemProxyCheck, userProxyCheck].count(True)
if count != 1:
FreeCAD.Console.PrintWarning(
translate(
"AddonsInstaller",
"Parameter error: mutually exclusive proxy options set. Resetting to default.",
)
+ "\n"
)
noProxyCheck = True
systemProxyCheck = False
userProxyCheck = False
pref.SetBool("NoProxyCheck", noProxyCheck)
pref.SetBool("SystemProxyCheck", systemProxyCheck)
pref.SetBool("UserProxyCheck", userProxyCheck)
if userProxyCheck and not proxy_string:
FreeCAD.Console.PrintWarning(
translate(
"AddonsInstaller",
"Parameter error: user proxy indicated, but no proxy provided. Resetting to default.",
)
+ "\n"
)
noProxyCheck = True
userProxyCheck = False
pref.SetBool("NoProxyCheck", noProxyCheck)
pref.SetBool("UserProxyCheck", userProxyCheck)
noProxyCheck, systemProxyCheck, userProxyCheck, proxy_string = self._setup_proxy_freecad()
else:
print("Please select a proxy type:")
print("1) No proxy")
print("2) Use system proxy settings")
print("3) Custom proxy settings")
result = input("Choice: ")
if result == "1":
pass
elif result == "2":
noProxyCheck = False
systemProxyCheck = True
elif result == "3":
noProxyCheck = False
userProxyCheck = True
proxy_string = input("Enter your proxy server (host:port): ")
else:
print(f"Got {result}, expected 1, 2, or 3.")
app.quit()
noProxyCheck, systemProxyCheck, userProxyCheck, proxy_string = self._setup_proxy_standalone()
if noProxyCheck:
pass
@@ -247,16 +207,80 @@ if HAVE_QTNETWORK:
)
self.QNAM.setProxy(proxy)
# A helper connection for our blocking interface
self.completed.connect(self.__synchronous_process_completion)
def _setup_proxy_freecad(self):
""" If we are running within FreeCAD, this uses the config data to set up the proxy """
noProxyCheck = True
systemProxyCheck = False
userProxyCheck = False
proxy_string = ""
pref = FreeCAD.ParamGet("User parameter:BaseApp/Preferences/Addons")
noProxyCheck = pref.GetBool("NoProxyCheck", noProxyCheck)
systemProxyCheck = pref.GetBool("SystemProxyCheck", systemProxyCheck)
userProxyCheck = pref.GetBool("UserProxyCheck", userProxyCheck)
proxy_string = pref.GetString("ProxyUrl", "")
# Set up our worker connection
self.__request_queued.connect(self.__setup_network_request)
# Add some error checking to the proxy setup, since for historical reasons they
# are independent booleans, rather than an enumeration:
option_count = [noProxyCheck, systemProxyCheck, userProxyCheck].count(True)
if option_count != 1:
FreeCAD.Console.PrintWarning(
translate(
"AddonsInstaller",
"Parameter error: mutually exclusive proxy options set. Resetting to default.",
)
+ "\n"
)
noProxyCheck = True
systemProxyCheck = False
userProxyCheck = False
pref.SetBool("NoProxyCheck", noProxyCheck)
pref.SetBool("SystemProxyCheck", systemProxyCheck)
pref.SetBool("UserProxyCheck", userProxyCheck)
if userProxyCheck and not proxy_string:
FreeCAD.Console.PrintWarning(
translate(
"AddonsInstaller",
"Parameter error: user proxy indicated, but no proxy provided. Resetting to default.",
)
+ "\n"
)
noProxyCheck = True
userProxyCheck = False
pref.SetBool("NoProxyCheck", noProxyCheck)
pref.SetBool("UserProxyCheck", userProxyCheck)
return noProxyCheck, systemProxyCheck, userProxyCheck, proxy_string
def _setup_proxy_standalone(self):
""" If we are NOT running inside FreeCAD, prompt the user for proxy information """
noProxyCheck = True
systemProxyCheck = False
userProxyCheck = False
proxy_string = ""
print("Please select a proxy type:")
print("1) No proxy")
print("2) Use system proxy settings")
print("3) Custom proxy settings")
result = input("Choice: ")
if result == "1":
pass
elif result == "2":
noProxyCheck = False
systemProxyCheck = True
elif result == "3":
noProxyCheck = False
userProxyCheck = True
proxy_string = input("Enter your proxy server (host:port): ")
else:
print(f"Got {result}, expected 1, 2, or 3.")
app.quit()
return noProxyCheck, systemProxyCheck, userProxyCheck, proxy_string
def __aboutToQuit(self):
pass
"""Called when the application is about to quit. Not currently used."""
def __setup_network_request(self):
"""Get the next request off the queue and launch it."""
try:
item = self.queue.get_nowait()
if item:
@@ -272,6 +296,7 @@ if HAVE_QTNETWORK:
def __launch_request(
self, index: int, request: QtNetwork.QNetworkRequest
) -> None:
"""Given a network request, ask the QNetworkAccessManager to begin processing it."""
reply = self.QNAM.get(request)
self.replies[index] = reply
@@ -342,12 +367,12 @@ if HAVE_QTNETWORK:
self.synchronous_complete.pop(current_index)
if current_index in self.synchronous_result_data:
return self.synchronous_result_data.pop(current_index)
else:
return None
return None
def __synchronous_process_completion(
self, index: int, code: int, data: QtCore.QByteArray
) -> None:
"""Check the return status of a completed process, and handle its returned data (if any)."""
with self.synchronous_lock:
if index in self.synchronous_complete:
if code == 200:
@@ -363,6 +388,7 @@ if HAVE_QTNETWORK:
self.synchronous_complete[index] = True
def __create_get_request(self, url: str) -> QtNetwork.QNetworkRequest:
"""Construct a network request to a given URL"""
request = QtNetwork.QNetworkRequest(QtCore.QUrl(url))
request.setAttribute(
QtNetwork.QNetworkRequest.RedirectPolicyAttribute,
@@ -390,6 +416,7 @@ if HAVE_QTNETWORK:
break
def abort(self, index: int):
"""Abort a specific request"""
if index in self.replies and self.replies[index].isRunning():
self.replies[index].abort()
elif index < self.__last_started_index:
@@ -401,6 +428,8 @@ if HAVE_QTNETWORK:
reply: QtNetwork.QNetworkProxy,
authenticator: QtNetwork.QAuthenticator,
):
"""If proxy authentication is required, attempt to authenticate. If the GUI is running this displays
a window asking for credentials. If the GUI is not running, it prompts on the command line."""
if HAVE_FREECAD and FreeCAD.GuiUp:
proxy_authentication = FreeCADGui.PySideUic.loadUi(
os.path.join(os.path.dirname(__file__), "proxy_authentication.ui")
@@ -434,9 +463,10 @@ if HAVE_QTNETWORK:
_reply: QtNetwork.QNetworkReply,
_authenticator: QtNetwork.QAuthenticator,
):
pass
"""Unused."""
def __follow_redirect(self, url):
"""Used with the QNetworkAccessManager to follow redirects."""
sender = self.sender()
if sender:
for index, reply in self.replies.items():
@@ -448,6 +478,7 @@ if HAVE_QTNETWORK:
self.__launch_request(current_index, self.__create_get_request(url))
def __on_ssl_error(self, reply: str, errors: List[str]):
"""Called when an SSL error occurs: prints the error information."""
if HAVE_FREECAD:
FreeCAD.Console.PrintWarning(
translate("AddonsInstaller", "Error with encrypted connection")
@@ -462,6 +493,7 @@ if HAVE_QTNETWORK:
print(error)
def __download_progress(self, bytesReceived: int, bytesTotal: int) -> None:
"""Monitors download progress and emits a progress_made signal"""
sender = self.sender()
if not sender:
return
@@ -471,6 +503,7 @@ if HAVE_QTNETWORK:
return
def __ready_to_read(self) -> None:
"""Called when data is available, this reads that data."""
sender = self.sender()
if not sender:
return
@@ -481,6 +514,7 @@ if HAVE_QTNETWORK:
return
def __data_incoming(self, index: int, reply: QtNetwork.QNetworkReply) -> None:
"""Read incoming data and attach it to a data object"""
if not index in self.replies:
# We already finished this reply, this is a vestigial signal
return
@@ -492,7 +526,7 @@ if HAVE_QTNETWORK:
f = self.file_buffers[index]
try:
f.write(buffer.data())
except Exception as e:
except IOError as e:
if HAVE_FREECAD:
FreeCAD.Console.PrintError(
f"Network Manager internal error: {str(e)}"
@@ -501,6 +535,8 @@ if HAVE_QTNETWORK:
print(f"Network Manager internal error: {str(e)}")
def __reply_finished(self) -> None:
"""Called when a reply has been completed: this makes sure the data has been read and
any notifications have been called."""
reply = self.sender()
if not reply:
print(
@@ -567,28 +603,32 @@ else: # HAVE_QTNETWORK is false:
self.unmonitored_queue = queue.Queue()
def submit_unmonitored_request(self, _) -> int:
"""Returns a fake index that can be used for testing -- nothing is actually queued"""
current_index = next(itertools.count())
self.unmonitored_queue.put(current_index)
return current_index
def submit_monitored_request(self, _) -> int:
"""Returns a fake index that can be used for testing -- nothing is actually queued"""
current_index = next(itertools.count())
self.monitored_queue.put(current_index)
return current_index
def blocking_get(self, _: str) -> QtCore.QByteArray:
"""No operation - returns None immediately"""
return None
def abort_all(
self,
):
pass # Nothing to do
"""There is nothing to abort in this case"""
def abort(self, _):
pass # Nothing to do
"""There is nothing to abort in this case"""
def InitializeNetworkManager():
"""Called once at the beginning of program execution to create the appropriate manager object"""
global AM_NETWORK_MANAGER
if AM_NETWORK_MANAGER is None:
AM_NETWORK_MANAGER = NetworkManager()
@@ -611,6 +651,7 @@ if __name__ == "__main__":
]
def handle_completion(index: int, code: int, data):
"""Attached to the completion signal, prints diagnostic information about the network access"""
global count
if code == 200:
print(
@@ -624,14 +665,14 @@ if __name__ == "__main__":
count += 1
if count >= len(urls):
print(f"Shutting down...", flush=True)
print("Shutting down...", flush=True)
AM_NETWORK_MANAGER.requestInterruption()
AM_NETWORK_MANAGER.wait(5000)
app.quit()
AM_NETWORK_MANAGER.completed.connect(handle_completion)
for url in urls:
AM_NETWORK_MANAGER.submit_unmonitored_get(url)
for test_url in urls:
AM_NETWORK_MANAGER.submit_unmonitored_get(test_url)
app.exec_()