我们从Python开源项目中，提取了以下25个代码示例，用于说明如何使用twisted.web.client.HTTPDownloader()。
def test_downloadTimeout(self): """ If the timeout indicated by the C{timeout} parameter to L{client.HTTPDownloader.__init__} elapses without the complete response being received, the L{defer.Deferred} returned by L{client.downloadPage} fires with a L{Failure} wrapping a L{defer.TimeoutError}. """ self.cleanupServerConnections = 2 # Verify the behavior if no bytes are ever written. first = client.downloadPage( self.getURL("wait"), self.mktemp(), timeout=0.01) # Verify the behavior if some bytes are written but then the request # never completes. second = client.downloadPage( self.getURL("write-then-wait"), self.mktemp(), timeout=0.01) return defer.gatherResults([ self.assertFailure(first, defer.TimeoutError), self.assertFailure(second, defer.TimeoutError)])
def test_downloadTimeoutsWorkWithoutReading(self): """ If the timeout indicated by the C{timeout} parameter to L{client.HTTPDownloader.__init__} elapses without the complete response being received, the L{defer.Deferred} returned by L{client.downloadPage} fires with a L{Failure} wrapping a L{defer.TimeoutError}, even if the remote peer isn't reading data from the socket. """ self.cleanupServerConnections = 1 # The timeout here needs to be slightly longer to give the resource a # change to stop the reading. d = client.downloadPage( self.getURL("never-read"), self.mktemp(), timeout=0.05) return self.assertFailure(d, defer.TimeoutError)
def test_downloadHeaders(self): """ After L{client.HTTPDownloader.deferred} fires, the L{client.HTTPDownloader} instance's C{status} and C{response_headers} attributes are populated with the values from the response. """ def checkHeaders(factory): self.assertEqual(factory.status, b'200') self.assertEqual(factory.response_headers[b'content-type'][0], b'text/html') self.assertEqual(factory.response_headers[b'content-length'][0], b'10') os.unlink(factory.fileName) factory = client._makeGetterFactory( self.getURL('file'), client.HTTPDownloader, fileOrName=self.mktemp()) return factory.deferred.addCallback(lambda _: checkHeaders(factory))
def test_downloadCookies(self): """ The C{cookies} dict passed to the L{client.HTTPDownloader} initializer is used to populate the I{Cookie} header included in the request sent to the server. """ output = self.mktemp() factory = client._makeGetterFactory( self.getURL('cookiemirror'), client.HTTPDownloader, fileOrName=output, cookies={b'foo': b'bar'}) def cbFinished(ignored): self.assertEqual( FilePath(output).getContent(), b"[('foo', 'bar')]") factory.deferred.addCallback(cbFinished) return factory.deferred
def test_downloadRedirectLimit(self): """ When more than C{redirectLimit} HTTP redirects are encountered, the page request fails with L{InfiniteRedirection}. """ def checkRedirectCount(*a): self.assertEqual(f._redirectCount, 7) self.assertEqual(self.infiniteRedirectResource.count, 7) f = client._makeGetterFactory( self.getURL('infiniteRedirect'), client.HTTPDownloader, fileOrName=self.mktemp(), redirectLimit=7) d = self.assertFailure(f.deferred, error.InfiniteRedirection) d.addCallback(checkRedirectCount) return d
def __init__(self, url, outfile, headers=None): client.HTTPDownloader.__init__(self, url, outfile, headers=headers, agent="%s %s Enigma2 HbbTV/1.1.1 (+PVR+RTP+RTSP+RTMP+DL;OpenLD;;;;;)" % (getMachineBrand(), getMachineName())) self.status = self.progress_callback = self.error_callback = self.end_callback = None self.deferred = defer.Deferred() return
def noPage(self, reason): if self.status == "304": client.HTTPDownloader.page(self, "") else: client.HTTPDownloader.noPage(self, reason) if self.error_callback: self.error_callback(reason.getErrorMessage(), self.status)
def gotHeaders(self, headers): if self.status == "200": if "content-length" in headers: self.totalbytes = int(headers["content-length"][0]) else: self.totalbytes = 0 self.currentbytes = 0.0 return client.HTTPDownloader.gotHeaders(self, headers)
def pagePart(self, packet): if self.status == "200": self.currentbytes += len(packet) if self.totalbytes and self.progress_callback: self.progress_callback(self.currentbytes, self.totalbytes) return client.HTTPDownloader.pagePart(self, packet)
def pageEnd(self): ret = client.HTTPDownloader.pageEnd(self) if self.end_callback: self.end_callback() return ret
def __init__(self, url, outfile, headers=None): client.HTTPDownloader.__init__(self, url, outfile, headers=headers, agent="Enigma2 HbbTV/1.1.1 (+PVR+RTSP+DL;OpenPLi;;;)") self.status = None self.progress_callback = None self.deferred = defer.Deferred()
def noPage(self, reason): if self.status == "304": print reason.getErrorMessage() client.HTTPDownloader.page(self, "") else: client.HTTPDownloader.noPage(self, reason)
def gotHeaders(self, headers): if self.status == "200": if headers.has_key("content-length"): self.totalbytes = int(headers["content-length"][0]) else: self.totalbytes = 0 self.currentbytes = 0.0 return client.HTTPDownloader.gotHeaders(self, headers)
def pageEnd(self): return client.HTTPDownloader.pageEnd(self)
def __init__(self, url, outfile, headers=None): client.HTTPDownloader.__init__(self, url, outfile, headers=headers, agent="%s %s Enigma2 HbbTV/1.1.1 (+PVR+RTSP+DL;OpenBh;;;)" % (getMachineBrand(), getMachineName())) self.status = self.progress_callback = self.error_callback = self.end_callback = None self.deferred = defer.Deferred()
