openstack/deb-python-httpretty
Code Issues Proposed changes

dirty and nasty

Browse Source
This commit is contained in:
Gabriel Falcao
2011-06-28 13:21:10 -04:00
parent ba8ed55662
commit c182adeb0f
3 changed files with 122 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

104
httpretty/__init__.py
View File

@@ -27,6 +27,7 @@ version = '0.2'
import re
import socket
import functools
import warnings
import logging
import traceback
@@ -94,6 +95,7 @@ class fakesock(object):
_entry = None
debuglevel = 0
_sent_data = []
def __init__(self, family, type, protocol):
self.family = family
self.type = type
@@ -247,7 +249,12 @@ STATUSES = {
class Entry(object):
def __init__(self, method, uri, body, adding_headers=None, forcing_headers=None, status=200, **headers):
def __init__(self, method, uri, body,
adding_headers=None,
forcing_headers=None,
status=200,
**headers):
self.method = method
self.uri = uri
self.body = body
@@ -265,27 +272,34 @@ class Entry(object):
def validate(self):
content_length_keys = 'Content-Length', 'content-cength'
for key in content_length_keys:
got = self.adding_headers.get(key, self.forcing_headers.get(key, None))
got = self.adding_headers.get(
key, self.forcing_headers.get(key, None))
if got is None:
continue
try:
igot = int(got)
except ValueError:
warnings.warn('HTTPretty got to register the Content-Length header with "%r" which is not a number' % got)
warnings.warn(
'HTTPretty got to register the Content-Length header ' \
'with "%r" which is not a number' % got,
)
if igot > self.body_length:
raise HTTPrettyError(
'HTTPretty got inconsistent parameters. The header Content-Length you registered expects size "%d" '
'but the body you registered for that has actually length "%d".\n'
'Fix that, or if you really want that, call register_uri with "fill_with" callback.' % (
igot, self.body_length
'HTTPretty got inconsistent parameters. The header ' \
'Content-Length you registered expects size "%d" but ' \
'the body you registered for that has actually length ' \
'"%d".\nFix that, or if you really want that, call ' \
'register_uri with "fill_with" callback.' % (
igot, self.body_length,
)
)
def __repr__(self):
return r'<Entry %s %s getting %d>' % (self.method, self.uri, self.status)
return r'<Entry %s %s getting %d>' % (
self.method, self.uri, self.status)
def fill_filekind(self, fk):
now = datetime.utcnow()
@@ -325,7 +339,7 @@ class Entry(object):
for k, v in headers.items():
string_list.append(
'%s: %s' % (k, unicode(v))
'%s: %s' % (k, unicode(v)),
)
fk.write("\n".join(string_list))
@@ -345,6 +359,7 @@ class URIInfo(object):
fragment='',
entries=None,
last_request=None):
self.username = username or ''
self.password = password or ''
self.hostname = hostname or ''
@@ -360,7 +375,6 @@ class URIInfo(object):
if self.current_entry >= len(self.entries):
self.current_entry = -1
if not self.entries:
raise ValueError('I have no entries: %s' % self)
@@ -370,8 +384,17 @@ class URIInfo(object):
return entry
def __unicode__(self):
attrs = 'username', 'password', 'hostname', 'port', 'path', 'query', 'fragment'
return ur'<httpretty.URIInfo(%s)>' % ", ".join(['%s="%s"' % (k, getattr(self, k, '')) for k in attrs])
attrs = (
'username',
'password',
'hostname',
'port',
'path',
'query',
'fragment',
)
fmt = ", ".join(['%s="%s"' % (k, getattr(self, k, '')) for k in attrs])
return ur'<httpretty.URIInfo(%s)>' % fmt
def __repr__(self):
return unicode(self)
@@ -414,20 +437,27 @@ class HTTPretty(object):
return request
@classmethod
def register_uri(cls, method, uri, body='HTTPretty :)', adding_headers=None, forcing_headers=None, status=200, responses=None, **headers):
def register_uri(cls, method, uri, body='HTTPretty :)',
adding_headers=None,
forcing_headers=None,
status=200,
responses=None, **headers):
if isinstance(responses, list) and len(responses) > 0:
entries_for_this_uri = responses
else:
headers['body'] = body
headers['adding_headers'] = adding_headers
headers['forcing_headers'] = forcing_headers
headers['status'] = status
headers['responses'] = responses
entries_for_this_uri = [
cls.Response(
body=body,
adding_headers=adding_headers,
forcing_headers=forcing_headers,
status=status, **headers
)
cls.Response(**headers),
]
map(lambda e: setattr(e, 'uri', uri) or setattr(e, 'method', method), entries_for_this_uri)
map(lambda e: setattr(e, 'uri', uri) or setattr(e, 'method', method),
entries_for_this_uri)
info = URIInfo.from_uri(uri, entries_for_this_uri)
if cls._entries.has_key(info):
@@ -439,16 +469,14 @@ class HTTPretty(object):
return u'<HTTPretty with %d URI entries>' % len(self._entries)
@classmethod
def Response(cls, body, adding_headers=None, forcing_headers=None, status=200, **headers):
return Entry(
method=None,
uri=None,
body=body,
adding_headers=adding_headers,
forcing_headers=forcing_headers,
status=status,
**headers
)
def Response(cls, body, adding_headers=None, forcing_headers=None,
status=200, **headers):
headers['body'] = body
headers['adding_headers'] = adding_headers
headers['forcing_headers'] = forcing_headers
headers['status'] = status
return Entry(method=None, uri=None, **headers)
@classmethod
def disable(cls):
@@ -490,4 +518,16 @@ class HTTPretty(object):
socks.socksocket = fakesock.socket
socks.__dict__['socksocket'] = fakesock.socket
HTTPretty.enable()
def httprettified(test):
"A decorator tests that use HTTPretty"
@functools.wraps
def wrapper(*args, **kw):
HTTPretty.enable()
try:
r = test(*args, **kw)
finally:
HTTPretty.disable()
return r
return wrapper

20
tests/functional/test_bypass.py
View File

@@ -27,23 +27,28 @@
import urllib2
from testserver import Server
from sure import that, that_with_context
from httpretty import HTTPretty
from httpretty import HTTPretty, httprettified
def start_server(context):
context.server = Server(9999)
context.server.start()
HTTPretty.enable()
def stop_server(context):
context.server.stop()
HTTPretty.enable()
@httprettified
@that_with_context(start_server, stop_server)
def test_httpretty_bypasses_when_disabled():
u"HTTPretty should bypass all requests by disabling it"
HTTPretty.register_uri(HTTPretty.GET, "http://localhost:9999/go-for-bubbles/",
body="glub glub")
HTTPretty.register_uri(
HTTPretty.GET, "http://localhost:9999/go-for-bubbles/",
body="glub glub")
HTTPretty.disable()
@@ -51,7 +56,8 @@ def test_httpretty_bypasses_when_disabled():
got1 = fd.read()
fd.close()
assert that(got1).equals('. o O 0 O o . o O 0 O o . o O 0 O o . o O 0 O o . o O 0 O o .')
assert that(got1).equals(
'. o O 0 O o . o O 0 O o . o O 0 O o . o O 0 O o . o O 0 O o .')
fd = urllib2.urlopen('http://localhost:9999/come-again/')
got2 = fd.read()
@@ -68,12 +74,14 @@ def test_httpretty_bypasses_when_disabled():
assert that(got3).equals('glub glub')
@httprettified
@that_with_context(start_server, stop_server)
def test_httpretty_bypasses_a_unregistered_request():
u"HTTPretty should bypass a unregistered request by disabling it"
HTTPretty.register_uri(HTTPretty.GET, "http://localhost:9999/go-for-bubbles/",
body="glub glub")
HTTPretty.register_uri(
HTTPretty.GET, "http://localhost:9999/go-for-bubbles/",
body="glub glub")
fd = urllib2.urlopen('http://localhost:9999/go-for-bubbles/')
got1 = fd.read()

55
tests/functional/test_httplib2.py
View File

@@ -26,17 +26,20 @@
# OTHER DEALINGS IN THE SOFTWARE.
import httplib2
from sure import *
from httpretty import HTTPretty
from sure import that, that_with_context, within, microseconds
from httpretty import HTTPretty, httprettified
def prepare(context, now):
HTTPretty.enable()
context.http = httplib2.Http()
def and_clear(context, now):
HTTPretty.enable()
context.http = httplib2.Http()
@within(two=microseconds)
@that_with_context(prepare, and_clear)
def test_httpretty_should_mock_a_simple_get_with_httplib2_read(context, now):
@@ -50,6 +53,8 @@ def test_httpretty_should_mock_a_simple_get_with_httplib2_read(context, now):
assert that(HTTPretty.last_request.method).equals('GET')
assert that(HTTPretty.last_request.path).equals('/')
@httprettified
@within(two=microseconds)
@that_with_context(prepare, and_clear)
def test_httpretty_should_mock_headers_httplib2(context, now):
@@ -67,7 +72,7 @@ def test_httpretty_should_mock_headers_httplib2(context, now):
'content-length': '35',
'status': '201',
'server': 'Python/HTTPretty',
'date': now.strftime('%a, %d %b %Y %H:%M:%SGMT')
'date': now.strftime('%a, %d %b %Y %H:%M:%SGMT'),
})
@@ -93,9 +98,10 @@ def test_httpretty_should_allow_adding_and_overwritting_httplib2(context, now):
'content-length': '27',
'status': '200',
'server': 'Apache',
'date': now.strftime('%a, %d %b %Y %H:%M:%S GMT')
'date': now.strftime('%a, %d %b %Y %H:%M:%S GMT'),
})
@within(two=microseconds)
@that_with_context(prepare, and_clear)
def test_httpretty_should_allow_forcing_headers_httplib2(context, now):
@@ -110,9 +116,13 @@ def test_httpretty_should_allow_forcing_headers_httplib2(context, now):
headers, _ = context.http.request('http://github.com', 'GET')
assert that(headers).equals({
'content-location': 'http://github.com/', # httplib2 FORCES content-location even if the server does not provide it
'content-location': 'http://github.com/', # httplib2 FORCES
# content-location
# even if the
# server does not
# provide it
'content-type': 'application/xml',
'status': '200', # httplib2 also ALWAYS put status on headers
'status': '200', # httplib2 also ALWAYS put status on headers
})
@@ -132,38 +142,45 @@ def test_httpretty_should_allow_adding_and_overwritting_by_kwargs_u2(context, no
assert that(headers).equals({
'content-type': 'application/json',
'content-location': 'http://github.com/', # httplib2 FORCES content-location even if the server does not provide it
'content-location': 'http://github.com/', # httplib2 FORCES
# content-location
# even if the
# server does not
# provide it
'connection': 'close',
'content-length': '27',
'status': '200',
'server': 'Apache',
'date': now.strftime('%a, %d %b %Y %H:%M:%S GMT')
'date': now.strftime('%a, %d %b %Y %H:%M:%S GMT'),
})
@within(two=microseconds)
@that_with_context(prepare, and_clear)
def test_httpretty_should_support_a_list_of_successive_responses_httplib2(context, now):
u"HTTPretty should support adding a list of successive responses with httplib2"
def test_rotating_responses_with_httplib2(context, now):
u"HTTPretty should support rotating responses with httplib2"
HTTPretty.register_uri(HTTPretty.GET, "http://github.com/gabrielfalcao/httpretty",
responses=[
HTTPretty.Response(body="first response", status=201),
HTTPretty.Response(body='second and last response', status=202),
])
HTTPretty.register_uri(
HTTPretty.GET, "http://github.com/gabrielfalcao/httpretty",
responses=[
HTTPretty.Response(body="first response", status=201),
HTTPretty.Response(body='second and last response', status=202),
])
headers1, body1 = context.http.request('http://github.com/gabrielfalcao/httpretty', 'GET')
headers1, body1 = context.http.request(
'http://github.com/gabrielfalcao/httpretty', 'GET')
assert that(headers1['status']).equals('201')
assert that(body1).equals('first response')
headers2, body2 = context.http.request('http://github.com/gabrielfalcao/httpretty', 'GET')
headers2, body2 = context.http.request(
'http://github.com/gabrielfalcao/httpretty', 'GET')
assert that(headers2['status']).equals('202')
assert that(body2).equals('second and last response')
headers3, body3 = context.http.request('http://github.com/gabrielfalcao/httpretty', 'GET')
headers3, body3 = context.http.request(
'http://github.com/gabrielfalcao/httpretty', 'GET')
assert that(headers3['status']).equals('202')
assert that(body3).equals('second and last response')