1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
|
# vim: ft=python fileencoding=utf-8 sts=4 sw=4 et:
# Copyright 2014-2021 Florian Bruhin (The Compiler) <mail@qutebrowser.org>
#
# Based on the Eric5 helpviewer,
# Copyright (c) 2009 - 2014 Detlev Offenbach <detlev@die-offenbachs.de>
#
# This file is part of qutebrowser.
#
# qutebrowser is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# qutebrowser is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with qutebrowser. If not, see <https://www.gnu.org/licenses/>.
#
# For some reason, a segfault will be triggered if the unnecessary lambdas in
# this file aren't there.
# pylint: disable=unnecessary-lambda
"""Special network replies.."""
from PyQt5.QtNetwork import QNetworkReply, QNetworkRequest
from PyQt5.QtCore import pyqtSlot, QIODevice, QByteArray, QTimer
class FixedDataNetworkReply(QNetworkReply):
"""QNetworkReply subclass for fixed data."""
def __init__(self, request, fileData, mimeType, parent=None): # noqa: N803
"""Constructor.
Args:
request: reference to the request object (QNetworkRequest)
fileData: reference to the data buffer (QByteArray)
mimeType: for the reply (string)
parent: reference to the parent object (QObject)
"""
super().__init__(parent)
self._data = fileData
self.setRequest(request)
self.setUrl(request.url())
self.setOpenMode(QIODevice.ReadOnly)
self.setHeader(QNetworkRequest.ContentTypeHeader, mimeType)
self.setHeader(QNetworkRequest.ContentLengthHeader,
QByteArray.number(len(fileData)))
self.setAttribute(QNetworkRequest.HttpStatusCodeAttribute, 200)
self.setAttribute(QNetworkRequest.HttpReasonPhraseAttribute, 'OK')
# For some reason, a segfault will be triggered if these lambdas aren't
# there.
# pylint: disable=unnecessary-lambda
QTimer.singleShot(0, lambda: self.metaDataChanged.emit())
QTimer.singleShot(0, lambda: self.readyRead.emit())
QTimer.singleShot(0, lambda: self.finished.emit())
@pyqtSlot()
def abort(self):
"""Abort the operation."""
def bytesAvailable(self):
"""Determine the bytes available for being read.
Return:
bytes available (int)
"""
return len(self._data) + super().bytesAvailable()
def readData(self, maxlen):
"""Retrieve data from the reply object.
Args:
maxlen maximum number of bytes to read (int)
Return:
bytestring containing the data
"""
len_ = min(maxlen, len(self._data))
buf = bytes(self._data[:len_])
self._data = self._data[len_:]
return buf
def isFinished(self):
return True
def isRunning(self):
return False
class ErrorNetworkReply(QNetworkReply):
"""QNetworkReply which always returns an error."""
def __init__(self, req, errorstring, error, parent=None):
"""Constructor.
Args:
req: The QNetworkRequest associated with this reply.
errorstring: The error string to print.
error: The numerical error value.
parent: The parent to pass to QNetworkReply.
"""
super().__init__(parent)
self.setRequest(req)
self.setUrl(req.url())
# We don't actually want to read anything, but we still need to open
# the device to avoid getting a warning.
self.setOpenMode(QIODevice.ReadOnly)
self.setError(error, errorstring)
QTimer.singleShot(0, lambda: self.error.emit(error))
QTimer.singleShot(0, lambda: self.finished.emit())
def abort(self):
"""Do nothing since it's a fake reply."""
def bytesAvailable(self):
"""We always have 0 bytes available."""
return 0
def readData(self, _maxlen):
"""No data available."""
return bytes()
def isFinished(self):
return True
def isRunning(self):
return False
class RedirectNetworkReply(QNetworkReply):
"""A reply which redirects to the given URL."""
def __init__(self, new_url, parent=None):
super().__init__(parent)
self.setAttribute(QNetworkRequest.RedirectionTargetAttribute, new_url)
QTimer.singleShot(0, lambda: self.finished.emit())
def abort(self):
"""Called when there's e.g. a redirection limit."""
def readData(self, _maxlen):
return bytes()
|
