openstack/deb-python-httpretty
Code Issues Proposed changes

Starting to refactor the core module

Browse Source
This commit is contained in:
Gabriel Falcao
2013-10-04 16:05:05 -04:00
parent 8bbbdfc143
commit 0443e321c2
4 changed files with 381 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
Makefile
View File

@@ -13,15 +13,15 @@ check_dependencies:
test: unit functional doctests
unit:
unit: prepare
@echo "Running unit tests ..."
@nosetests -s --verbosity=2 --with-coverage --cover-erase --cover-inclusive tests/unit --cover-package=httpretty
functional:
functional: prepare
@echo "Running functional tests ..."
@nosetests -s --verbosity=2 --with-coverage --cover-erase --cover-inclusive tests/functional --cover-package=httpretty
doctests:
doctests: prepare
@echo "Running documentation tests tests ..."
@steadymark README.md
@@ -45,3 +45,6 @@ docs: doctests
git commit -am 'documentation' && \
git push --force origin gh-pages && \
git checkout master
prepare:
@reset

112
httpretty/core.py
View File

@@ -81,21 +81,21 @@ old_ssl_wrap_socket = None
old_sslwrap_simple = None
old_sslsocket = None
if PY3:
if PY3: # pragma: no cover
basestring = (bytes, str)
try:
try: # pragma: no cover
import socks
old_socksocket = socks.socksocket
except ImportError:
socks = None
try:
try: # pragma: no cover
import ssl
old_ssl_wrap_socket = ssl.wrap_socket
if not PY3:
old_sslwrap_simple = ssl.sslwrap_simple
old_sslsocket = ssl.SSLSocket
except ImportError:
except ImportError: # pragma: no cover
ssl = None
@@ -104,48 +104,110 @@ DEFAULT_HTTP_PORTS = tuple(POTENTIAL_HTTP_PORTS)
class HTTPrettyRequest(BaseHTTPRequestHandler, BaseClass):
"""Represents a HTTP request. It takes a valid multi-line, `\r\n`
separated string with HTTP headers and parse them out using the
internal `parse_request` method.
It also replaces the `rfile` and `wfile` attributes with StringIO
instances so that we garantee that it won't make any I/O, neighter
for writing nor reading.
It has some convenience attributes:
`headers` -> a mimetype object that can be cast into a dictionary,
contains all the request headers
`method` -> the HTTP method used in this request
`querystring` -> a dictionary containing lists with the
attributes. Please notice that if you need a single value from a
query string you will need to get it manually like:
```python
>>> request.querystring
{'name': ['Gabriel Falcao']}
>>> print request.querystring['name'][0]
```
`parsed_body` -> a dictionary containing parsed request body or
None if HTTPrettyRequest doesn't know how to parse it. It
currently supports parsing body data that was sent under the
`content-type` headers values: 'application/json' or
'application/x-www-form-urlencoded'
"""
def __init__(self, headers, body=''):
# first of all, lets make sure that if headers or body are
# unicode strings, it must be converted into a utf-8 encoded
# byte string
self.raw_headers = utf8(headers.strip())
self.body = utf8(body)
self.raw_headers = utf8(headers)
self.rfile = StringIO(b'\r\n\r\n'.join([utf8(headers.strip()), self.body]))
self.wfile = StringIO()
# Now let's concatenate the headers with the body, and create
# `rfile` based on it
self.rfile = StringIO(b'\r\n\r\n'.join([self.raw_headers, self.body]))
self.wfile = StringIO() # Creating `wfile` as an empty
# StringIO, just to avoid any real
# I/O calls
# parsing the request line preemptively
self.raw_requestline = self.rfile.readline()
self.error_code = self.error_message = None
# initiating the error attributes with None
self.error_code = None
self.error_message = None
# Parse the request based on the attributes above
self.parse_request()
# making the HTTP method string available as the command
self.method = self.command
# Now 2 convenient attributes for the HTTPretty API:
# `querystring` holds a dictionary with the parsed query string
self.path = decode_utf8(self.path)
qstring = self.path.split("?", 1)[-1]
self.querystring = self.parse_querystring(qstring)
# And the body will be attempted to be parsed as
# `application/json` or `application/x-www-form-urlencoded`
self.parsed_body = self.parse_request_body(self.body)
def __str__(self):
return 'HTTPrettyRequest(headers={0}, body="{1}")'.format(
self.headers,
self.body,
return '<HTTPrettyRequest("{0}", total_headers={1}, body_length={2})>'.format(
self.headers.get('content-type', ''),
len(self.headers),
len(self.body),
)
def parse_querystring(self, qs):
unicode_qs = qs.encode('utf-8')
expanded = unquote(unicode_qs)
expanded = decode_utf8(unquote(utf8(qs)))
parsed = parse_qs(expanded)
result = {}
for k, v in parsed.iteritems():
result[k] = map(decode_utf8, v)
return result
def parse_request_body(self, body):
""" Attempt to parse the post based on the content-type passed. Return the regular body if not """
return_value = body.decode('utf-8')
PARSING_FUNCTIONS = {
'application/json': json.loads,
'text/json': json.loads,
'application/x-www-form-urlencoded': self.parse_querystring,
}
FALLBACK_FUNCTION = lambda x: x
content_type = self.headers.get('content-type', '')
do_parse = PARSING_FUNCTIONS.get(content_type, FALLBACK_FUNCTION)
try:
for header in self.headers.keys():
if header.lower() == 'content-type':
if self.headers['content-type'].lower() == 'application/json':
return_value = json.loads(return_value)
elif self.headers['content-type'].lower() == 'application/x-www-form-urlencoded':
return_value = self.parse_querystring(return_value)
finally:
return return_value
return do_parse(body)
except:
return body
class EmptyRequestHeaders(dict):
@@ -166,9 +228,6 @@ class FakeSSLSocket(object):
self._httpretty_sock = sock
def __getattr__(self, attr):
if attr == '_httpretty_sock':
return super(FakeSSLSocket, self).__getattribute__(attr)
return getattr(self._httpretty_sock, attr)
@@ -224,11 +283,12 @@ class fakesock(object):
self._address = (self._host, self._port) = address
self._closed = False
self.is_http = self._port in POTENTIAL_HTTP_PORTS
if not self.is_http:
self.truesock.connect(self._address)
def close(self):
if not self._closed:
if not (self.is_http and self._closed):
self.truesock.close()
self._closed = True

4
tests/__init__.py Normal file
View File

@@ -0,0 +1,4 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
from __future__ import unicode_literals
import sure

285
tests/unit/test_core.py Normal file
View File

@@ -0,0 +1,285 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
from __future__ import unicode_literals
import json
from datetime import datetime
from mock import Mock, patch, call
from sure import expect
from httpretty.core import HTTPrettyRequest, FakeSSLSocket, fakesock
def test_request_stubs_internals():
("HTTPrettyRequest is a BaseHTTPRequestHandler that replaces "
"real socket file descriptors with in-memory ones")
# Given a valid HTTP request header string
headers = "\r\n".join([
'POST /somewhere/?name=foo&age=bar HTTP/1.1',
'Accept-Encoding: identity',
'Host: github.com',
'Content-Type: application/json',
'Connection: close',
'User-Agent: Python-urllib/2.7',
])
# When I create a HTTPrettyRequest with an empty body
request = HTTPrettyRequest(headers, body='')
# Then it should have parsed the headers
dict(request.headers).should.equal({
'accept-encoding': 'identity',
'connection': 'close',
'content-type': 'application/json',
'host': 'github.com',
'user-agent': 'Python-urllib/2.7'
})
# And the `rfile` should be a StringIO
request.should.have.property('rfile').being.a('StringIO.StringIO')
# And the `wfile` should be a StringIO
request.should.have.property('wfile').being.a('StringIO.StringIO')
# And the `method` should be available
request.should.have.property('method').being.equal('POST')
def test_request_parse_querystring():
("HTTPrettyRequest#parse_querystring should parse unicode data")
# Given a request string containing a unicode encoded querystring
headers = "\r\n".join([
'POST /create?name=Gabriel+Falcão HTTP/1.1',
'Content-Type: multipart/form-data',
])
# When I create a HTTPrettyRequest with an empty body
request = HTTPrettyRequest(headers, body='')
# Then it should have a parsed querystring
request.querystring.should.equal({'name': ['Gabriel Falcão']})
def test_request_parse_body_when_it_is_application_json():
("HTTPrettyRequest#parse_request_body recognizes the "
"content-type `application/json` and parses it")
# Given a request string containing a unicode encoded querystring
headers = "\r\n".join([
'POST /create HTTP/1.1',
'Content-Type: application/json',
])
# And a valid json body
body = json.dumps({'name': 'Gabriel Falcão'})
# When I create a HTTPrettyRequest with that data
request = HTTPrettyRequest(headers, body)
# Then it should have a parsed body
request.parsed_body.should.equal({'name': 'Gabriel Falcão'})
def test_request_parse_body_when_it_is_text_json():
("HTTPrettyRequest#parse_request_body recognizes the "
"content-type `text/json` and parses it")
# Given a request string containing a unicode encoded querystring
headers = "\r\n".join([
'POST /create HTTP/1.1',
'Content-Type: text/json',
])
# And a valid json body
body = json.dumps({'name': 'Gabriel Falcão'})
# When I create a HTTPrettyRequest with that data
request = HTTPrettyRequest(headers, body)
# Then it should have a parsed body
request.parsed_body.should.equal({'name': 'Gabriel Falcão'})
def test_request_parse_body_when_it_is_urlencoded():
("HTTPrettyRequest#parse_request_body recognizes the "
"content-type `application/x-www-form-urlencoded` and parses it")
# Given a request string containing a unicode encoded querystring
headers = "\r\n".join([
'POST /create HTTP/1.1',
'Content-Type: application/x-www-form-urlencoded',
])
# And a valid urlencoded body
body = "name=Gabriel+Falcão&age=25&projects=httpretty&projects=sure&projects=lettuce"
# When I create a HTTPrettyRequest with that data
request = HTTPrettyRequest(headers, body)
# Then it should have a parsed body
request.parsed_body.should.equal({
'name': ['Gabriel Falcão'],
'age': ["25"],
'projects': ["httpretty", "sure", "lettuce"]
})
def test_request_parse_body_when_unrecognized():
("HTTPrettyRequest#parse_request_body returns the value as "
"is if the Content-Type is not recognized")
# Given a request string containing a unicode encoded querystring
headers = "\r\n".join([
'POST /create HTTP/1.1',
'Content-Type: whatever',
])
# And a valid urlencoded body
body = "foobar:\nlalala"
# When I create a HTTPrettyRequest with that data
request = HTTPrettyRequest(headers, body)
# Then it should have a parsed body
request.parsed_body.should.equal("foobar:\nlalala")
def test_request_string_representation():
("HTTPrettyRequest should have a debug-friendly "
"string representation")
# Given a request string containing a unicode encoded querystring
headers = "\r\n".join([
'POST /create HTTP/1.1',
'Content-Type: JPEG-baby',
])
# And a valid urlencoded body
body = "foobar:\nlalala"
# When I create a HTTPrettyRequest with that data
request = HTTPrettyRequest(headers, body)
# Then its string representation should show the headers and the body
str(request).should.equal('<HTTPrettyRequest("JPEG-baby", total_headers=1, body_length=14)>')
def test_fake_ssl_socket_proxies_its_ow_socket():
("FakeSSLSocket is a simpel wrapper around its own socket, "
"which was designed to be a HTTPretty fake socket")
# Given a sentinel mock object
socket = Mock()
# And a FakeSSLSocket wrapping it
ssl = FakeSSLSocket(socket)
# When I make a method call
ssl.send("FOO")
# Then it should bypass any method calls to its own socket
socket.send.assert_called_once_with("FOO")
@patch('httpretty.core.datetime')
def test_fakesock_socket_getpeercert(dt):
("fakesock.socket#getpeercert should return a hardcoded fake certificate")
# Background:
dt.now.return_value = datetime(2013, 10, 4, 4, 20, 0)
# Given a fake socket instance
socket = fakesock.socket()
# And that it's bound to some host and port
socket.connect(('somewhere.com', 80))
# When I retrieve the peer certificate
certificate = socket.getpeercert()
# Then it should return a hardcoded value
certificate.should.equal({
u'notAfter': 'Sep 29 04:20:00 GMT',
u'subject': (
((u'organizationName', u'*.somewhere.com'),),
((u'organizationalUnitName', u'Domain Control Validated'),),
((u'commonName', u'*.somewhere.com'),)),
u'subjectAltName': (
(u'DNS', u'*somewhere.com'),
(u'DNS', u'somewhere.com'),
(u'DNS', u'*')
)
})
def test_fakesock_socket_ssl():
("fakesock.socket#ssl should take a socket instance and return itself")
# Given a fake socket instance
socket = fakesock.socket()
# And a stubbed socket sentinel
sentinel = Mock()
# When I call `ssl` on that mock
result = socket.ssl(sentinel)
# Then it should have returned its first argument
result.should.equal(sentinel)
@patch('httpretty.core.old_socket')
@patch('httpretty.core.POTENTIAL_HTTP_PORTS')
def test_fakesock_socket_connect_fallback(POTENTIAL_HTTP_PORTS, old_socket):
("fakesock.socket#connect should open a real connection if the "
"given port is not a potential http port")
# Background: the potential http ports are 80 and 443
POTENTIAL_HTTP_PORTS.__contains__.side_effect = lambda other: int(other) in (80, 443)
# Given a fake socket instance
socket = fakesock.socket()
# When it is connected to a remote server in a port that isn't 80 nor 443
socket.connect(('somewhere.com', 42))
# Then it should have open a real connection in the background
old_socket.return_value.connect.assert_called_once_with(('somewhere.com', 42))
# And _closed is set to False
socket._closed.should.be.false
@patch('httpretty.core.old_socket')
def test_fakesock_socket_close(old_socket):
("fakesock.socket#close should close the actual socket in case "
"it's not http and _closed is False")
# Given a fake socket instance that is synthetically open
socket = fakesock.socket()
socket._closed = False
# When I close it
socket.close()
# Then its real socket should have been closed
old_socket.return_value.close.assert_called_once_with()
# And _closed is set to True
socket._closed.should.be.true
@patch('httpretty.core.old_socket')
def test_fakesock_socket_makefile(old_socket):
("fakesock.socket#makefile should set the mode, "
"bufsize and return its mocked file descriptor")
# Given a fake socket that has a mocked Entry associated with it
socket = fakesock.socket()
socket._entry = Mock()
# When I call makefile()
fd = socket.makefile(mode='rw', bufsize=512)
# Then it should have returned the socket's own filedescriptor
expect(fd).to.equal(socket.fd)
# And the mode should have been set in the socket instance
socket._mode.should.equal('rw')
# And the bufsize should have been set in the socket instance
socket._bufsize.should.equal(512)