From: W. Trevor King Date: Fri, 24 Aug 2012 11:55:48 +0000 (-0400) Subject: util:http: pull HTTP helpers from libbe.storage.http into their own module. X-Git-Tag: 1.1.0~131 X-Git-Url: http://git.tremily.us/?a=commitdiff_plain;h=3e8b3e08cd9cbb99b44376a91fcfbef42ee3888e;p=be.git util:http: pull HTTP helpers from libbe.storage.http into their own module. This way they can be shared with the upcoming Command._run_remote. --- diff --git a/libbe/storage/http.py b/libbe/storage/http.py index c69598c..594fd0d 100644 --- a/libbe/storage/http.py +++ b/libbe/storage/http.py @@ -16,13 +16,6 @@ # You should have received a copy of the GNU General Public License along with # Bugs Everywhere. If not, see . -# For urllib2 information, see -# urllib2, from urllib2 - The Missing Manual -# http://www.voidspace.org.uk/python/articles/urllib2.shtml -# -# A dictionary of response codes is available in -# httplib.responses - """Define an HTTP-based :class:`~libbe.storage.base.VersionedStorage` implementation. @@ -34,11 +27,12 @@ See Also from __future__ import absolute_import import sys import urllib -import urllib2 import urlparse import libbe import libbe.version +import libbe.util.http +from libbe.util.http import HTTP_VALID, HTTP_USER_ERROR from . import base from libbe import TESTING @@ -53,79 +47,6 @@ if TESTING == True: import libbe.command.serve -USER_AGENT = 'BE-HTTP-Storage' -HTTP_OK = 200 -HTTP_FOUND = 302 -HTTP_TEMP_REDIRECT = 307 -HTTP_USER_ERROR = 418 -"""Status returned to indicate exceptions on the server side. - -A BE-specific extension to the HTTP/1.1 protocol (See `RFC 2616`_). - -.. _RFC 2616: http://www.w3.org/Protocols/rfc2616/rfc2616-sec6.html#sec6.1.1 -""" - -HTTP_VALID = [HTTP_OK, HTTP_FOUND, HTTP_TEMP_REDIRECT, HTTP_USER_ERROR] - -class InvalidURL (Exception): - def __init__(self, error=None, url=None, msg=None): - Exception.__init__(self, msg) - self.url = url - self.error = error - self.msg = msg - def __str__(self): - if self.msg == None: - if self.error == None: - return "Unknown URL error: %s" % self.url - return self.error.__str__() - return self.msg - -def get_post_url(url, get=True, data_dict=None, headers=[], agent=None): - """Execute a GET or POST transaction. - - Parameters - ---------- - url : str - The base URL (query portion added internally, if necessary). - get : bool - Use GET if True, otherwise use POST. - data_dict : dict - Data to send, either by URL query (if GET) or by POST (if POST). - headers : list - Extra HTTP headers to add to the request. - """ - if data_dict == None: - data_dict = {} - if agent is None: - agent = USER_AGENT - if get == True: - if data_dict != {}: - # encode get parameters in the url - param_string = urllib.urlencode(data_dict) - url = "%s?%s" % (url, param_string) - data = None - else: - data = urllib.urlencode(data_dict) - headers = dict(headers) - headers['User-Agent'] = agent - req = urllib2.Request(url, data=data, headers=headers) - try: - response = urllib2.urlopen(req) - except urllib2.HTTPError, e: - if hasattr(e, 'reason'): - msg = ('We failed to connect to the server.\nURL: {}\n' - 'Reason: {}').format(url, e.reason) - elif hasattr(e, 'code'): - msg = "The server couldn't fulfill the request.\nURL: %s\nError code: %s" \ - % (url, e.code) - raise InvalidURL(error=e, url=url, msg=msg) - page = response.read() - final_url = response.geturl() - info = response.info() - response.close() - return (page, final_url, info) - - class HTTP (base.VersionedStorage): """:class:`~libbe.storage.base.VersionedStorage` implementation over HTTP. @@ -133,6 +54,7 @@ class HTTP (base.VersionedStorage): Uses GET to retrieve information and POST to set information. """ name = 'HTTP' + user_agent = 'BE-HTTP-Storage' def __init__(self, repo, *args, **kwargs): repo,self.uname,self.password = self.parse_repo(repo) @@ -168,7 +90,8 @@ class HTTP (base.VersionedStorage): if self.uname != None and self.password != None: headers.append(('Authorization','Basic %s' % \ ('%s:%s' % (self.uname, self.password)).encode('base64'))) - return get_post_url(url, get, data_dict, headers) + return libbe.util.http.get_post_url( + url, get, data_dict, headers, agent=self.user_agent) def storage_version(self, revision=None): """Return the storage format for this backend.""" @@ -237,7 +160,7 @@ class HTTP (base.VersionedStorage): page,final_url,info = self.get_post_url( url, get=True, data_dict={'revision':revision}) - except InvalidURL, e: + except libbe.util.http.HTTPError, e: if not (hasattr(e.error, 'code') and e.error.code in HTTP_VALID): raise elif default == base.InvalidObject: @@ -255,7 +178,7 @@ class HTTP (base.VersionedStorage): page,final_url,info = self.get_post_url( url, get=False, data_dict={'value':value}) - except InvalidURL, e: + except libbe.util.http.HTTPError, e: if not (hasattr(e.error, 'code') and e.error.code in HTTP_VALID): raise if e.error.code == HTTP_USER_ERROR \ @@ -271,7 +194,7 @@ class HTTP (base.VersionedStorage): url, get=False, data_dict={'summary':summary, 'body':body, 'allow_empty':allow_empty}) - except InvalidURL, e: + except libbe.util.http.HTTPError, e: if not (hasattr(e.error, 'code') and e.error.code in HTTP_VALID): raise if e.error.code == HTTP_USER_ERROR: @@ -305,7 +228,7 @@ class HTTP (base.VersionedStorage): page,final_url,info = self.get_post_url( url, get=True, data_dict={'index':index}) - except InvalidURL, e: + except libbe.util.http.HTTPError, e: if not (hasattr(e.error, 'code') and e.error.code in HTTP_VALID): raise if e.error.code == HTTP_USER_ERROR: @@ -335,22 +258,6 @@ class HTTP (base.VersionedStorage): return page.rstrip('\n') if TESTING == True: - class GetPostUrlTestCase (unittest.TestCase): - """Test cases for get_post_url()""" - def test_get(self): - url = 'http://bugseverywhere.org/' - page,final_url,info = get_post_url(url=url) - self.failUnless(final_url == url, - 'Redirect?\n Expected: "%s"\n Got: "%s"' - % (url, final_url)) - def test_get_redirect(self): - url = 'http://physics.drexel.edu/~wking/code/be/redirect' - expected = 'http://physics.drexel.edu/~wking/' - page,final_url,info = get_post_url(url=url) - self.failUnless(final_url == expected, - 'Redirect?\n Expected: "%s"\n Got: "%s"' - % (expected, final_url)) - class TestingHTTP (HTTP): name = 'TestingHTTP' def __init__(self, repo, *args, **kwargs): @@ -420,7 +327,8 @@ if TESTING == True: def __str__(self): return self.string error = __estr(self.status) - raise InvalidURL(error=error, url=url, msg=output) + raise libbe.util.http.HTTPError( + error=error, url=url, msg=output) info = dict(self.response_headers) return (output, url, info) def _init(self): diff --git a/libbe/ui/command_line.py b/libbe/ui/command_line.py index d6e820b..164d2c7 100644 --- a/libbe/ui/command_line.py +++ b/libbe/ui/command_line.py @@ -29,10 +29,10 @@ import libbe.bugdir import libbe.command import libbe.command.util import libbe.storage -import libbe.storage.http import libbe.version import libbe.ui.util.pager import libbe.util.encoding +import libbe.util.http if libbe.TESTING == True: @@ -310,8 +310,8 @@ def dispatch(ui, command, args): except libbe.storage.ConnectionError, e: print >> ui.io.stdout, 'Connection Error:\n', e return 1 - except libbe.storage.http.InvalidURL, e: - print >> ui.io.stdout, 'Invalid URL:\n', e + except libbe.util.http.HTTPError, e: + print >> ui.io.stdout, 'HTTP Error:\n', e return 1 except (libbe.util.id.MultipleIDMatches, libbe.util.id.NoIDMatches, libbe.util.id.InvalidIDStructure), e: diff --git a/libbe/util/http.py b/libbe/util/http.py new file mode 100644 index 0000000..2f15b15 --- /dev/null +++ b/libbe/util/http.py @@ -0,0 +1,116 @@ +# Copyright + +# For urllib2 information, see +# urllib2, from urllib2 - The Missing Manual +# http://www.voidspace.org.uk/python/articles/urllib2.shtml +# +# A dictionary of response codes is available in +# httplib.responses +# but it is slow to load. + +import urllib +import urllib2 + +from libbe import TESTING + +if TESTING: + import unittest + + +HTTP_OK = 200 +HTTP_FOUND = 302 +HTTP_TEMP_REDIRECT = 307 +HTTP_USER_ERROR = 418 +"""Status returned to indicate exceptions on the server side. + +A BE-specific extension to the HTTP/1.1 protocol (See `RFC 2616`_). + +.. _RFC 2616: http://www.w3.org/Protocols/rfc2616/rfc2616-sec6.html#sec6.1.1 +""" + +HTTP_VALID = [HTTP_OK, HTTP_FOUND, HTTP_TEMP_REDIRECT, HTTP_USER_ERROR] + + +USER_AGENT = 'BE-agent' + + +class HTTPError (Exception): + def __init__(self, error=None, url=None, msg=None): + Exception.__init__(self, msg) + self.url = url + self.error = error + self.msg = msg + + def __str__(self): + if self.msg is None: + if self.error is None: + return 'Unknown HTTP error: {}'.format(self.url) + return str(self.error) + return self.msg + + +def get_post_url(url, get=True, data_dict=None, headers=[], agent=None): + """Execute a GET or POST transaction. + + Parameters + ---------- + url : str + The base URL (query portion added internally, if necessary). + get : bool + Use GET if True, otherwise use POST. + data_dict : dict + Data to send, either by URL query (if GET) or by POST (if POST). + headers : list + Extra HTTP headers to add to the request. + agent : str + User agent string overriding the BE default. + """ + if data_dict is None: + data_dict = {} + if agent is None: + agent = USER_AGENT + if get is True: + if data_dict != {}: + # encode get parameters in the url + param_string = urllib.urlencode(data_dict) + url = '{}?{}'.format(url, param_string) + data = None + else: + data = urllib.urlencode(data_dict) + headers = dict(headers) + headers['User-Agent'] = agent + req = urllib2.Request(url, data=data, headers=headers) + try: + response = urllib2.urlopen(req) + except urllib2.HTTPError, e: + if hasattr(e, 'reason'): + msg = ('We failed to connect to the server.\nURL: {}\n' + 'Reason: {}').format(url, e.reason) + elif hasattr(e, 'code'): + msg = ("The server couldn't fulfill the request.\nURL: {}\n" + 'Error code: {}').format(url, e.code) + raise HTTPError(error=e, url=url, msg=msg) + page = response.read() + final_url = response.geturl() + info = response.info() + response.close() + return (page, final_url, info) + + +if TESTING: + class GetPostUrlTestCase (unittest.TestCase): + """Test cases for get_post_url()""" + def test_get(self): + url = 'http://bugseverywhere.org/' + page,final_url,info = get_post_url(url=url) + self.failUnless(final_url == url, + 'Redirect?\n Expected: "{}"\n Got: "{}"'.format( + url, final_url)) + + def test_get_redirect(self): + url = 'http://physics.drexel.edu/~wking/code/be/redirect' + expected = 'http://physics.drexel.edu/~wking/' + page,final_url,info = get_post_url(url=url) + self.failUnless(final_url == expected, + 'Redirect?\n Expected: "{}"\n Got: "{}"'.format( + expected, final_url))