def _analysis_404(self):
"""
??404?????????URL????
"""
try:
conn1 = httplib.HTTPConnection(self.site_parse[1], timeout=self.timeout)
conn1.request('GET', self.site_parse[2]+'/../g0ogle/go0g1e/l.php', headers=self.headers)
response = conn1.getresponse()
self.not_found_page_length = response.getheader('Content-Length')
except (httplib.HTTPException, socket.timeout, socket.gaierror, Exception), e:
logger.error('url %s is unreachable, Exception %s %s' % (self.site, e.__class__.__name__, e))
print 'url %s is unreachable, Exception %s %s' % (self.site, e.__class__.__name__, e)
sys.exit(1)
python类HTTPException()的实例源码
def open_url(connection, url, return_redirect_url=False):
'''Tries to open url and return page's html'''
if 'goodreads.com' in url:
url = url[url.find('goodreads.com') + len('goodreads.com'):]
try:
connection.request('GET', url, headers=HEADERS)
response = connection.getresponse()
if response.status == 301 or response.status == 302:
if return_redirect_url:
return response.msg['location']
response = open_url(connection, response.msg['location'])
else:
response = response.read()
except (HTTPException, socket.error):
time.sleep(1)
connection.close()
connection.connect()
connection.request('GET', url, headers=HEADERS)
response = connection.getresponse()
if response.status == 301 or response.status == 302:
if return_redirect_url:
return response.msg['location']
response = open_url(connection, response.msg['location'])
else:
response = response.read()
if 'Page Not Found' in response:
raise PageDoesNotExist('Page not found.')
return response
def _TestServerConnection(self):
# Wait for server to start
server_msg = ''
for timeout in xrange(1, 5):
client_error = None
try:
with contextlib.closing(httplib.HTTPConnection(
'127.0.0.1', self.port, timeout=timeout)) as http:
http.set_debuglevel(timeout > 3)
http.request('HEAD', '/')
r = http.getresponse()
r.read()
if (r.status == 200 and r.reason == 'OK' and
r.getheader('Server') == self.server_tag):
return (None, server_msg)
client_error = ('Bad response: %s %s version %s\n ' %
(r.status, r.reason, r.version) +
'\n '.join([': '.join(h) for h in r.getheaders()]))
except (httplib.HTTPException, socket.error) as client_error:
pass # Probably too quick connecting: try again
# Check for server startup error messages
ix = self.process.expect([pexpect.TIMEOUT, pexpect.EOF, '.+'],
timeout=timeout)
if ix == 2: # stdout spew from the server
server_msg += self.process.match.group(0) # pylint: disable=no-member
elif ix == 1: # EOF -- server has quit so giveup.
client_error = client_error or 'Server exited'
break
return (client_error or 'Timeout', server_msg)
def _TestServerConnection(self):
# Wait for server to start
server_msg = ''
for timeout in xrange(1, 5):
client_error = None
try:
with contextlib.closing(httplib.HTTPConnection(
'127.0.0.1', self.port, timeout=timeout)) as http:
http.set_debuglevel(timeout > 3)
http.request('HEAD', '/')
r = http.getresponse()
r.read()
if (r.status == 200 and r.reason == 'OK' and
r.getheader('Server') == self.server_tag):
return (None, server_msg)
client_error = ('Bad response: %s %s version %s\n ' %
(r.status, r.reason, r.version) +
'\n '.join([': '.join(h) for h in r.getheaders()]))
except (httplib.HTTPException, socket.error) as client_error:
pass # Probably too quick connecting: try again
# Check for server startup error messages
ix = self.process.expect([pexpect.TIMEOUT, pexpect.EOF, '.+'],
timeout=timeout)
if ix == 2: # stdout spew from the server
server_msg += self.process.match.group(0) # pylint: disable=no-member
elif ix == 1: # EOF -- server has quit so giveup.
client_error = client_error or 'Server exited'
break
return (client_error or 'Timeout', server_msg)
def __init__(self, *args, **kw):
unittest.TestCase.__init__(self, *args, **kw)
try:
self.open_mapping_file().close() # test it to report the error early
except (IOError, HTTPException):
self.skipTest("Could not retrieve "+self.mapfileurl)
def test_too_many_headers(self):
headers = '\r\n'.join('Header%d: foo' % i for i in xrange(200)) + '\r\n'
text = ('HTTP/1.1 200 OK\r\n' + headers)
s = FakeSocket(text)
r = httplib.HTTPResponse(s)
self.assertRaises(httplib.HTTPException, r.begin)
def retry_if_network_error(exception):
pick_da_useragent()
do_retry = isinstance(exception, (requests.exceptions.RequestException, httplib.HTTPException, socket.error))
if do_retry:
echo(Fore.YELLOW + 'Retrying... ' + repr(exception))
return do_retry
test_multibytecodec_support.py 文件源码
项目：python2-tracer
作者: extremecoders-re
项目源码
文件源码
阅读 31
收藏 0
点赞 0
评论 0
def __init__(self, *args, **kw):
unittest.TestCase.__init__(self, *args, **kw)
try:
self.open_mapping_file().close() # test it to report the error early
except (IOError, HTTPException):
self.skipTest("Could not retrieve "+self.mapfileurl)
def test_too_many_headers(self):
headers = '\r\n'.join('Header%d: foo' % i for i in xrange(200)) + '\r\n'
text = ('HTTP/1.1 200 OK\r\n' + headers)
s = FakeSocket(text)
r = httplib.HTTPResponse(s)
self.assertRaises(httplib.HTTPException, r.begin)
def _start_transaction(self, h, req):
try:
if req.has_data():
data = req.data
if hasattr(req, 'selector'):
h.putrequest(req.get_method() or 'POST', req.selector, skip_host=req.has_header("Host"), skip_accept_encoding=req.has_header("Accept-encoding"))
else:
h.putrequest(req.get_method() or 'POST', req.get_selector(), skip_host=req.has_header("Host"), skip_accept_encoding=req.has_header("Accept-encoding"))
if not req.headers.has_key('Content-type'):
h.putheader('Content-type',
'application/x-www-form-urlencoded')
if not req.headers.has_key('Content-length'):
h.putheader('Content-length', '%d' % len(data))
else:
if hasattr(req, 'selector'):
h.putrequest(req.get_method() or 'GET', req.selector, skip_host=req.has_header("Host"), skip_accept_encoding=req.has_header("Accept-encoding"))
else:
h.putrequest(req.get_method() or 'GET', req.get_selector(), skip_host=req.has_header("Host"), skip_accept_encoding=req.has_header("Accept-encoding"))
except (socket.error, httplib.HTTPException), err:
raise urllib2.URLError(err)
if not req.headers.has_key('Connection'):
req.headers['Connection'] = 'keep-alive'
for args in self.parent.addheaders:
if not req.headers.has_key(args[0]):
h.putheader(*args)
for k, v in req.headers.items():
h.putheader(k, v)
h.endheaders()
if req.has_data():
h.send(data)
def selenium_login(self):
try:
self.pydriver.get(self.config_dict[TABLEAU_SERVER_URL])
time.sleep(2)
self.pydriver.page_source.encode('utf-8')
logins = self.pydriver.find_elements(By.XPATH, '//input')
if logins is None or len(logins) == 0:
ACLogger().get_logger().error('selenium_login error 1')
return False
the_login = logins[0]
password = logins[1]
the_login.send_keys(self.config_dict[TABLEAU_SERVER_USERNAME])
password.send_keys(self.config_dict[TABLEAU_SERVER_PASSWORD])
buttons = self.pydriver.find_elements(By.XPATH, '//button')
button = None
for b in buttons:
if str(b.text) == 'Sign In':
button = b
if button is None:
ACLogger().get_logger().error('selenium_login error 2')
return False
button.click()
time.sleep(2)
if 'Workbooks' not in self.pydriver.title:
ACLogger().get_logger().error('selenium_login error 3')
return False
except (WebDriverException, StaleElementReferenceException, HTTPException, ResponseNotReady) , e:
ACLogger().get_logger().error('selenium_login error: %s' % str(e))
return False
ACLogger().get_logger().info(
'Logged in %s successfully, next page title: %s' % (self.config_dict[TABLEAU_SERVER_URL], self.pydriver.title))
return True
def connect(self):
"""Opens a HTTP connection to the RPC server."""
logger.debug("Opening connection to %s:%s", self.url.hostname, self.url.port)
try:
self.connection = httplib.HTTPConnection(self.url.hostname, self.url.port)
self.connection.connect()
except (httplib.HTTPException, socket.error) as e:
raise errors.InterfaceError('Unable to connect to the specified service', e)
def close(self):
"""Closes the HTTP connection to the RPC server."""
if self.connection is not None:
logger.debug("Closing connection to %s:%s", self.url.hostname, self.url.port)
try:
self.connection.close()
except httplib.HTTPException:
logger.warning("Error while closing connection", exc_info=True)
self.connection = None
def _post_request(self, body, headers):
retry_count = self.max_retries
while True:
logger.debug("POST %s %r %r", self.url.path, body, headers)
try:
self.connection.request('POST', self.url.path, body=body, headers=headers)
response = self.connection.getresponse()
except httplib.HTTPException as e:
if retry_count > 0:
delay = math.exp(-retry_count)
logger.debug("HTTP protocol error, will retry in %s seconds...", delay, exc_info=True)
self.close()
self.connect()
time.sleep(delay)
retry_count -= 1
continue
raise errors.InterfaceError('RPC request failed', cause=e)
else:
if response.status == httplib.SERVICE_UNAVAILABLE:
if retry_count > 0:
delay = math.exp(-retry_count)
logger.debug("Service unavailable, will retry in %s seconds...", delay, exc_info=True)
time.sleep(delay)
retry_count -= 1
continue
return response
def get_status(host, path, file):
try:
conn = HTTPConnection(host)
conn.request('HEAD', '/{0}/{1}'.format(path, file))
res = conn.getresponse()
except (HTTPException, socket.timeout, socket.error):
return 0
else:
return res.status
def getresponse(self):
if self.response:
return self.response
else:
raise httplib.HTTPException()
def __init__(self, response):
raw_headers = response.getheaders()
headers = dict()
for k, v in raw_headers:
headers[k.lower()] = v
self.request_id = headers.get('x-cas-requestid')
self.status = response.status
sys.stdout.write('====== debug: error: receive status: %s\n' % response.status)
sys.stdout.write('====== debug: error: receive headers: %s\n' % headers)
content = ''
try:
content = response.read()
body = json.loads(content)
sys.stdout.write('====== debug: error: receive body: %s\n' % body)
self.code = body.get('code')
self.type = body.get('type')
self.message = body.get('message')
msg = 'Expected 2xx, got '
msg += '(%d, code=%s, message=%s, type=%s, request_id=%s)' % \
(self.status, self.code,
self.message, self.type, self.request_id)
except (HTTPException, ValueError):
msg = 'Expected 2xx, got (%d, content=%s, request_id=%s)' % \
(self.status, content, self.request_id)
super(CASServerError, self).__init__(msg)
def __init__(self, *args, **kw):
unittest.TestCase.__init__(self, *args, **kw)
try:
self.open_mapping_file().close() # test it to report the error early
except (IOError, HTTPException):
self.skipTest("Could not retrieve "+self.mapfileurl)
def test_too_many_headers(self):
headers = '\r\n'.join('Header%d: foo' % i for i in xrange(200)) + '\r\n'
text = ('HTTP/1.1 200 OK\r\n' + headers)
s = FakeSocket(text)
r = httplib.HTTPResponse(s)
self.assertRaises(httplib.HTTPException, r.begin)
def __init__(self, *args, **kw):
unittest.TestCase.__init__(self, *args, **kw)
try:
self.open_mapping_file().close() # test it to report the error early
except (IOError, HTTPException):
self.skipTest("Could not retrieve "+self.mapfileurl)
