python类getPage()的实例源码

def testBrokenDownload(self):
        # test what happens when download gets disconnected in the middle
        d = client.getPage(self.getURL("broken"))
        d = self.assertFailure(d, client.PartialDownloadError)
        d.addCallback(lambda exc: self.assertEquals(exc.response, "abc"))
        return d

def testHostHeader(self):
        # if we pass Host header explicitly, it should be used, otherwise
        # it should extract from url
        return defer.gatherResults([
            client.getPage(self.getURL("host")).addCallback(self.assertEquals, "127.0.0.1"),
            client.getPage(self.getURL("host"), headers={"Host": "www.example.com"}).addCallback(self.assertEquals, "www.example.com")])

def testGetPage(self):
        return client.getPage(self.getURL("file")).addCallback(self.assertEquals, "0123456789")

def testTimeoutNotTriggering(self):
        # Test that when the timeout doesn't trigger, things work as expected.
        d = client.getPage(self.getURL("wait"), timeout=100)
        d.addCallback(self.assertEquals, "hello!!!")
        return d

def testServerError(self):
        return self._downloadTest(lambda url: client.getPage(self.getURL(url)))

def testRedirect(self):
        return client.getPage(self.getURL("redirect")).addCallback(self._cbRedirect)

def _cbRedirect(self, pageData):
        self.assertEquals(pageData, "0123456789")
        d = self.assertFailure(
            client.getPage(self.getURL("redirect"), followRedirect=0),
            error.PageRedirect)
        d.addCallback(self._cbCheckLocation)
        return d

def testHoppingAround(self):
        return client.getPage(self.getHTTP("one")
            ).addCallback(self.assertEquals, "FOUND IT!"
            )

def testNoCookies(self):
        return client.getPage(self.getHTTP("cookiemirror")
            ).addCallback(self.assertEquals, "[]"
            )

def testRawNoCookies(self):
        return client.getPage(self.getHTTP("rawcookiemirror")
            ).addCallback(self.assertEquals, "None"
            )

def testRawSomeCookies(self):
        cookies = {'foo': 'bar', 'baz': 'quux'}
        return client.getPage(self.getHTTP("rawcookiemirror"), cookies=cookies
            ).addCallback(self.assertEquals, "'foo=bar; baz=quux'"
            )

def _getPage(self):
        d = client.getPage(self.url)
        d.addErrback(self.noPage)
        d.addCallback(self.page)
        self.call = None

def getList(self, callback, errback):
        self.urlbase = "%s/%s/%s/images/" % (self.feed_base, self.OE_vers, self.box)
        print "[getList]", self.urlbase
        self.callback = callback
        self.errback = errback
        client.getPage(self.urlbase).addCallback(self.feed_finished).addErrback(self.feed_failed)

def getMD5(self):
        url = "http://www.dreamboxupdate.com/download/opendreambox/dreambox-nfiflasher-%s-md5sums" % self.box
        client.getPage(url).addCallback(self.md5sums_finished).addErrback(self.feed_failed)

def nfo_download(self, branch, idx):
        nfourl = (self.feedlists[branch][idx][1])[:-4]+".nfo"
        self.nfofilename = (self.feedlists[branch][idx][0])[:-4]+".nfo"
        print "[check_for_NFO]", nfourl
        client.getPage(nfourl).addCallback(self.nfo_finished).addErrback(self.nfo_failed)

def download(self, url):
        return getPage(url, timeout=20, headers={'Accept': 'application/json'})

def call(self, method, **kwargs):
        """
        Call the given remote method with the given arguments, as keywords.

        Returns a deferred, called with SOAPpy structure representing
        the soap response.

        @param method: The method name to call, eg. 'GetExternalIP'
        @param kwargs: The parameters of the call, as keywords
        @return: A deferred called with the external ip address of this host
        @rtype: L{twisted.internet.defer.Deferred}
        """
        payload = SOAPpy.buildSOAP(method=method, config=Config, namespace=self._prefix, kw=kwargs)
        # Here begins the nasty hack
        payload = payload.replace(
            # Upnp wants s: instead of SOAP-ENV
            'SOAP-ENV','s').replace(
            # Doesn't seem to like these encoding stuff
            'xmlns:SOAP-ENC="http://schemas.xmlsoap.org/soap/encoding/"', '').replace(
            'SOAP-ENC:root="1"', '').replace(
            # And it wants u: instead of ns1 namespace for arguments..
            'ns1','u')

        logging.debug("SOAP Payload:\n%s", payload)

        return client.getPage(self._url, postdata=payload, method="POST",
            headers={'content-type': 'text/xml',        'SOAPACTION': '%s#%s' % (self._prefix, method)}
    ).addCallbacks(self._got_page, self._got_error)

def _http_do(url, headers, timeout, method, params):
    id_ = 0

    try:
        data = yield client.getPage(
            url=url,
            method='POST',
            headers=dict(headers, **{'Content-Type': 'application/json'}),
            postdata=json.dumps({
                'jsonrpc': '2.0',
                'method': method,
                'params': params,
                'id': id_,
            }),
            timeout=timeout,
        )
    except error.Error, e:
        try:
            resp = json.loads(e.response)
        except:
            raise e
    else:
        resp = json.loads(data)

    if resp['id'] != id_:
        raise ValueError('invalid id')
    if 'error' in resp and resp['error'] is not None:
        raise Error_for_code(resp['error']['code'])(resp['error']['message'], resp['error'].get('data', None))
    defer.returnValue(resp['result'])

def async_url_read(url, timeout=0, **kwargs):
    if url.startswith('http'):
        content = getPage(encode(url), timeout=timeout)
    else:
        content = async_read_file(url, StringTransport(), **kwargs)

    return content

def makeAPICall(self, path, method="POST", params=None):
        if params:
            postdata = json.dumps(params)
        else:
            postdata = None
        uri = networkString("http://127.0.0.1:%d/API/%s" % (
            self.plainPortno, path))
        return client.getPage(
            uri, method=method, timeout=1, postdata=postdata,
            headers={'X-Bitmask-Auth': 'aaa'})