Added some convenience functions to NetworkedPrinterOutputdevice

This also moves the getUser from legacy to networked printer

CL-541
This commit is contained in:
Jaime van Kessel 2017-11-28 12:43:49 +01:00
parent c1c59925de
commit cfc6a3ad48
2 changed files with 37 additions and 18 deletions

View file

@ -12,7 +12,9 @@ from PyQt5.QtCore import pyqtProperty, pyqtSignal, pyqtSlot, QObject, QTimer, py
from time import time from time import time
from typing import Callable, Any, Optional from typing import Callable, Any, Optional
from enum import IntEnum from enum import IntEnum
from typing import List
import os
class AuthState(IntEnum): class AuthState(IntEnum):
NotAuthenticated = 1 NotAuthenticated = 1
@ -121,6 +123,28 @@ class NetworkedPrinterOutputDevice(PrinterOutputDevice):
request.setHeader(QNetworkRequest.UserAgentHeader, self._user_agent) request.setHeader(QNetworkRequest.UserAgentHeader, self._user_agent)
return request return request
def _createFormPart(self, content_header, data, content_type = None):
part = QHttpPart()
if not content_header.startswith("form-data;"):
content_header = "form_data; " + content_header
part.setHeader(QNetworkRequest.ContentDispositionHeader, content_header)
if content_type is not None:
part.setHeader(QNetworkRequest.ContentTypeHeader, content_type)
part.setBody(data)
return part
## Convenience function to get the username from the OS.
# The code was copied from the getpass module, as we try to use as little dependencies as possible.
def _getUserName(self):
for name in ("LOGNAME", "USER", "LNAME", "USERNAME"):
user = os.environ.get(name)
if user:
return user
return "Unknown User" # Couldn't find out username.
def _clearCachedMultiPart(self, reply): def _clearCachedMultiPart(self, reply):
if id(reply) in self._cached_multiparts: if id(reply) in self._cached_multiparts:
del self._cached_multiparts[id(reply)] del self._cached_multiparts[id(reply)]
@ -160,29 +184,32 @@ class NetworkedPrinterOutputDevice(PrinterOutputDevice):
if onFinished is not None: if onFinished is not None:
self._onFinishedCallbacks[reply.url().toString() + str(reply.operation())] = onFinished self._onFinishedCallbacks[reply.url().toString() + str(reply.operation())] = onFinished
def postForm(self, target: str, header_data: str, body_data: bytes, onFinished: Optional[Callable[[Any, QNetworkReply], None]], onProgress: Callable = None): def postFormWithParts(self, target:str, parts: List[QHttpPart], onFinished: Optional[Callable[[Any, QNetworkReply], None]], onProgress: Callable = None):
if self._manager is None: if self._manager is None:
self._createNetworkManager() self._createNetworkManager()
request = self._createEmptyRequest(target, content_type=None) request = self._createEmptyRequest(target, content_type=None)
multi_post_part = QHttpMultiPart(QHttpMultiPart.FormDataType) multi_post_part = QHttpMultiPart(QHttpMultiPart.FormDataType)
post_part = QHttpPart() for part in parts:
post_part.setHeader(QNetworkRequest.ContentDispositionHeader, header_data) multi_post_part.append(part)
post_part.setBody(body_data)
multi_post_part.append(post_part)
self._last_request_time = time() self._last_request_time = time()
reply = self._manager.post(request, multi_post_part) reply = self._manager.post(request, multi_post_part)
# Due to garbage collection on python doing some weird stuff, we need to keep hold of a reference self._cached_multiparts[id(reply)] = (multi_post_part, reply)
self._cached_multiparts[id(reply)] = (post_part, multi_post_part, reply)
if onProgress is not None: if onProgress is not None:
reply.uploadProgress.connect(onProgress) reply.uploadProgress.connect(onProgress)
if onFinished is not None: if onFinished is not None:
self._onFinishedCallbacks[reply.url().toString() + str(reply.operation())] = onFinished self._onFinishedCallbacks[reply.url().toString() + str(reply.operation())] = onFinished
def postForm(self, target: str, header_data: str, body_data: bytes, onFinished: Optional[Callable[[Any, QNetworkReply], None]], onProgress: Callable = None):
post_part = QHttpPart()
post_part.setHeader(QNetworkRequest.ContentDispositionHeader, header_data)
post_part.setBody(body_data)
self.postFormWithParts(target, [post_part], onFinished, onProgress)
def _onAuthenticationRequired(self, reply, authenticator): def _onAuthenticationRequired(self, reply, authenticator):
Logger.log("w", "Request to {url} required authentication, which was not implemented".format(url = reply.url().toString())) Logger.log("w", "Request to {url} required authentication, which was not implemented".format(url = reply.url().toString()))

View file

@ -143,6 +143,7 @@ class LegacyUM3OutputDevice(NetworkedPrinterOutputDevice):
continue # If it's not readonly, it's created by user, so skip it. continue # If it's not readonly, it's created by user, so skip it.
file_name = "none.xml" file_name = "none.xml"
self.postForm("materials", "form-data; name=\"file\";filename=\"%s\"" % file_name, xml_data.encode(), onFinished=None) self.postForm("materials", "form-data; name=\"file\";filename=\"%s\"" % file_name, xml_data.encode(), onFinished=None)
except NotImplementedError: except NotImplementedError:
@ -596,13 +597,4 @@ class LegacyUM3OutputDevice(NetworkedPrinterOutputDevice):
result = "********" + result result = "********" + result
return result return result
return self._authentication_key return self._authentication_key
## Convenience function to get the username from the OS.
# The code was copied from the getpass module, as we try to use as little dependencies as possible.
def _getUserName(self):
for name in ("LOGNAME", "USER", "LNAME", "USERNAME"):
user = os.environ.get(name)
if user:
return user
return "Unknown User" # Couldn't find out username.