summaryrefslogtreecommitdiff
path: root/tests/unit/test_httpretty.py
diff options
context:
space:
mode:
Diffstat (limited to 'tests/unit/test_httpretty.py')
-rw-r--r--tests/unit/test_httpretty.py320
1 files changed, 171 insertions, 149 deletions
diff --git a/tests/unit/test_httpretty.py b/tests/unit/test_httpretty.py
index 3ae4f4f..f072b31 100644
--- a/tests/unit/test_httpretty.py
+++ b/tests/unit/test_httpretty.py
@@ -50,38 +50,40 @@ Content-Type: %(content_type)s
def test_httpretty_should_raise_proper_exception_on_inconsistent_length():
- ("HTTPretty should raise proper exception on inconsistent Content-Length / "
- "registered response body")
+ (
+ "HTTPretty should raise proper exception on inconsistent Content-Length / "
+ "registered response body"
+ )
HTTPretty.register_uri.when.called_with(
HTTPretty.GET,
"http://github.com/gabrielfalcao",
body="that's me!",
- adding_headers={
- 'Content-Length': '999'
- }
+ adding_headers={"Content-Length": "999"},
).should.have.raised(
HTTPrettyError,
'HTTPretty got inconsistent parameters. The header Content-Length you registered expects size "999" '
- 'but the body you registered for that has actually length "10".'
+ 'but the body you registered for that has actually length "10".',
)
def test_httpretty_should_raise_on_socket_send_when_uri_registered():
- ("HTTPretty should raise a RuntimeError when the fakesocket is "
- "used in an invalid usage.")
+ (
+ "HTTPretty should raise a RuntimeError when the fakesocket is "
+ "used in an invalid usage."
+ )
import socket
+
HTTPretty.enable()
- HTTPretty.register_uri(HTTPretty.GET,
- 'http://127.0.0.1:5000')
+ HTTPretty.register_uri(HTTPretty.GET, "http://127.0.0.1:5000")
expect(core.POTENTIAL_HTTP_PORTS).to.be.equal(set([80, 5000]))
expect(core.POTENTIAL_HTTPS_PORTS).to.be.equal(set([443]))
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
- sock.connect(('127.0.0.1', 5000))
- expect(sock.send).when.called_with(b'whatever').to.throw(RuntimeError)
+ sock.connect(("127.0.0.1", 5000))
+ expect(sock.send).when.called_with(b"whatever").to.throw(RuntimeError)
sock.close()
# restore the previous value
@@ -91,17 +93,20 @@ def test_httpretty_should_raise_on_socket_send_when_uri_registered():
def test_httpretty_should_not_raise_on_socket_send_when_uri_not_registered():
- ("HTTPretty should not raise a RuntimeError when the fakesocket "
- "is used in an invalid usage.")
+ (
+ "HTTPretty should not raise a RuntimeError when the fakesocket "
+ "is used in an invalid usage."
+ )
import socket
+
HTTPretty.enable()
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, 0)
sock.setblocking(0)
- expect(sock.sendto).when.called_with(b'whatever',
- ('127.0.0.1', 53)
- ).should_not.throw(RuntimeError)
+ expect(sock.sendto).when.called_with(
+ b"whatever", ("127.0.0.1", 53)
+ ).should_not.throw(RuntimeError)
sock.close()
HTTPretty.reset()
@@ -109,7 +114,7 @@ def test_httpretty_should_not_raise_on_socket_send_when_uri_not_registered():
def test_does_not_have_last_request_by_default():
- 'HTTPretty.last_request is a dummy object by default'
+ "HTTPretty.last_request is a dummy object by default"
HTTPretty.reset()
expect(HTTPretty.last_request.headers).to.be.empty
@@ -119,93 +124,95 @@ def test_does_not_have_last_request_by_default():
def test_status_codes():
"HTTPretty supports N status codes"
- expect(STATUSES).to.equal({
- 100: "Continue",
- 101: "Switching Protocols",
- 102: "Processing",
- 200: "OK",
- 201: "Created",
- 202: "Accepted",
- 203: "Non-Authoritative Information",
- 204: "No Content",
- 205: "Reset Content",
- 206: "Partial Content",
- 207: "Multi-Status",
- 208: "Already Reported",
- 226: "IM Used",
- 300: "Multiple Choices",
- 301: "Moved Permanently",
- 302: "Found",
- 303: "See Other",
- 304: "Not Modified",
- 305: "Use Proxy",
- 306: "Switch Proxy",
- 307: "Temporary Redirect",
- 308: "Permanent Redirect",
- 400: "Bad Request",
- 401: "Unauthorized",
- 402: "Payment Required",
- 403: "Forbidden",
- 404: "Not Found",
- 405: "Method Not Allowed",
- 406: "Not Acceptable",
- 407: "Proxy Authentication Required",
- 408: "Request a Timeout",
- 409: "Conflict",
- 410: "Gone",
- 411: "Length Required",
- 412: "Precondition Failed",
- 413: "Request Entity Too Large",
- 414: "Request-URI Too Long",
- 415: "Unsupported Media Type",
- 416: "Requested Range Not Satisfiable",
- 417: "Expectation Failed",
- 418: "I'm a teapot",
- 420: "Enhance Your Calm",
- 422: "Unprocessable Entity",
- 423: "Locked",
- 424: "Failed Dependency",
- 425: "Unordered Collection",
- 426: "Upgrade Required",
- 428: "Precondition Required",
- 429: "Too Many Requests",
- 431: "Request Header Fields Too Large",
- 444: "No Response",
- 449: "Retry With",
- 450: "Blocked by Windows Parental Controls",
- 451: "Unavailable For Legal Reasons",
- 494: "Request Header Too Large",
- 495: "Cert Error",
- 496: "No Cert",
- 497: "HTTP to HTTPS",
- 499: "Client Closed Request",
- 500: "Internal Server Error",
- 501: "Not Implemented",
- 502: "Bad Gateway",
- 503: "Service Unavailable",
- 504: "Gateway Timeout",
- 505: "HTTP Version Not Supported",
- 506: "Variant Also Negotiates",
- 507: "Insufficient Storage",
- 508: "Loop Detected",
- 509: "Bandwidth Limit Exceeded",
- 510: "Not Extended",
- 511: "Network Authentication Required",
- 598: "Network read timeout error",
- 599: "Network connect timeout error",
- })
+ expect(STATUSES).to.equal(
+ {
+ 100: "Continue",
+ 101: "Switching Protocols",
+ 102: "Processing",
+ 200: "OK",
+ 201: "Created",
+ 202: "Accepted",
+ 203: "Non-Authoritative Information",
+ 204: "No Content",
+ 205: "Reset Content",
+ 206: "Partial Content",
+ 207: "Multi-Status",
+ 208: "Already Reported",
+ 226: "IM Used",
+ 300: "Multiple Choices",
+ 301: "Moved Permanently",
+ 302: "Found",
+ 303: "See Other",
+ 304: "Not Modified",
+ 305: "Use Proxy",
+ 306: "Switch Proxy",
+ 307: "Temporary Redirect",
+ 308: "Permanent Redirect",
+ 400: "Bad Request",
+ 401: "Unauthorized",
+ 402: "Payment Required",
+ 403: "Forbidden",
+ 404: "Not Found",
+ 405: "Method Not Allowed",
+ 406: "Not Acceptable",
+ 407: "Proxy Authentication Required",
+ 408: "Request a Timeout",
+ 409: "Conflict",
+ 410: "Gone",
+ 411: "Length Required",
+ 412: "Precondition Failed",
+ 413: "Request Entity Too Large",
+ 414: "Request-URI Too Long",
+ 415: "Unsupported Media Type",
+ 416: "Requested Range Not Satisfiable",
+ 417: "Expectation Failed",
+ 418: "I'm a teapot",
+ 420: "Enhance Your Calm",
+ 422: "Unprocessable Entity",
+ 423: "Locked",
+ 424: "Failed Dependency",
+ 425: "Unordered Collection",
+ 426: "Upgrade Required",
+ 428: "Precondition Required",
+ 429: "Too Many Requests",
+ 431: "Request Header Fields Too Large",
+ 444: "No Response",
+ 449: "Retry With",
+ 450: "Blocked by Windows Parental Controls",
+ 451: "Unavailable For Legal Reasons",
+ 494: "Request Header Too Large",
+ 495: "Cert Error",
+ 496: "No Cert",
+ 497: "HTTP to HTTPS",
+ 499: "Client Closed Request",
+ 500: "Internal Server Error",
+ 501: "Not Implemented",
+ 502: "Bad Gateway",
+ 503: "Service Unavailable",
+ 504: "Gateway Timeout",
+ 505: "HTTP Version Not Supported",
+ 506: "Variant Also Negotiates",
+ 507: "Insufficient Storage",
+ 508: "Loop Detected",
+ 509: "Bandwidth Limit Exceeded",
+ 510: "Not Extended",
+ 511: "Network Authentication Required",
+ 598: "Network read timeout error",
+ 599: "Network connect timeout error",
+ }
+ )
def test_uri_info_full_url():
uri_info = URIInfo(
- username='johhny',
- password='password',
- hostname=b'google.com',
+ username="johhny",
+ password="password",
+ hostname=b"google.com",
port=80,
- path=b'/',
- query=b'foo=bar&baz=test',
- fragment='',
- scheme='',
+ path=b"/",
+ query=b"foo=bar&baz=test",
+ fragment="",
+ scheme="",
)
expect(uri_info.full_url()).to.equal(
@@ -222,24 +229,24 @@ def test_uri_info_eq_ignores_case():
hostname matching.
"""
uri_info_uppercase = URIInfo(
- username='johhny',
- password='password',
- hostname=b'GOOGLE.COM',
+ username="johhny",
+ password="password",
+ hostname=b"GOOGLE.COM",
port=80,
- path=b'/',
- query=b'foo=bar&baz=test',
- fragment='',
- scheme='',
+ path=b"/",
+ query=b"foo=bar&baz=test",
+ fragment="",
+ scheme="",
)
uri_info_lowercase = URIInfo(
- username='johhny',
- password='password',
- hostname=b'google.com',
+ username="johhny",
+ password="password",
+ hostname=b"google.com",
port=80,
- path=b'/',
- query=b'foo=bar&baz=test',
- fragment='',
- scheme='',
+ path=b"/",
+ query=b"foo=bar&baz=test",
+ fragment="",
+ scheme="",
)
expect(uri_info_uppercase).to.equal(uri_info_lowercase)
@@ -256,43 +263,48 @@ def test_global_boolean_enabled():
def test_py3kobject_implements_valid__repr__based_on__str__():
class MyObject(BaseClass):
def __str__(self):
- return 'hi'
+ return "hi"
myobj = MyObject()
- expect(repr(myobj)).to.be.equal('hi')
+ expect(repr(myobj)).to.be.equal("hi")
def test_Entry_class_normalizes_headers():
- entry = Entry(HTTPretty.GET, 'http://example.com', 'example',
- host='example.com', cache_control='no-cache', x_forward_for='proxy')
+ entry = Entry(
+ HTTPretty.GET,
+ "http://example.com",
+ "example",
+ host="example.com",
+ cache_control="no-cache",
+ x_forward_for="proxy",
+ )
- entry.adding_headers.should.equal({
- 'Host': 'example.com',
- 'Cache-Control': 'no-cache',
- 'X-Forward-For': 'proxy'
- })
+ entry.adding_headers.should.equal(
+ {"Host": "example.com", "Cache-Control": "no-cache", "X-Forward-For": "proxy"}
+ )
def test_Entry_class_counts_multibyte_characters_in_bytes():
- entry = Entry(HTTPretty.GET, 'http://example.com', 'こんにちは')
+ entry = Entry(HTTPretty.GET, "http://example.com", "こんにちは")
buf = FakeSockFile()
entry.fill_filekind(buf)
response = buf.read()
- expect(b'content-length: 15\n').to.be.within(response)
+ expect(b"content-length: 15\n").to.be.within(response)
def test_Entry_class_counts_dynamic():
- result = (200, {}, 'こんにちは'.encode('utf-8'))
- entry = Entry(HTTPretty.GET, 'http://example.com', lambda *args: result)
- entry.info = URIInfo.from_uri('http://example.com', entry)
+ result = (200, {}, "こんにちは".encode("utf-8"))
+ entry = Entry(HTTPretty.GET, "http://example.com", lambda *args: result)
+ entry.info = URIInfo.from_uri("http://example.com", entry)
buf = FakeSockFile()
entry.fill_filekind(buf)
response = buf.getvalue()
- expect(b'content-length: 15\n').to.be.within(response)
+ expect(b"content-length: 15\n").to.be.within(response)
def test_fake_socket_passes_through_setblocking():
import socket
+
HTTPretty.enable()
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.truesock = MagicMock()
@@ -302,6 +314,7 @@ def test_fake_socket_passes_through_setblocking():
def test_fake_socket_passes_through_fileno():
import socket
+
with httpretty.enabled():
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.truesock = MagicMock()
@@ -311,15 +324,19 @@ def test_fake_socket_passes_through_fileno():
def test_fake_socket_passes_through_getsockopt():
import socket
+
HTTPretty.enable()
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.truesock = MagicMock()
- expect(s.getsockopt).called_with(socket.SOL_SOCKET, 1).should_not.throw(AttributeError)
+ expect(s.getsockopt).called_with(socket.SOL_SOCKET, 1).should_not.throw(
+ AttributeError
+ )
s.truesock.getsockopt.assert_called_with(socket.SOL_SOCKET, 1)
def test_fake_socket_passes_through_bind():
import socket
+
HTTPretty.enable()
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.truesock = MagicMock()
@@ -329,6 +346,7 @@ def test_fake_socket_passes_through_bind():
def test_fake_socket_passes_through_connect_ex():
import socket
+
HTTPretty.enable()
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.truesock = MagicMock()
@@ -338,6 +356,7 @@ def test_fake_socket_passes_through_connect_ex():
def test_fake_socket_passes_through_listen():
import socket
+
HTTPretty.enable()
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.truesock = MagicMock()
@@ -347,6 +366,7 @@ def test_fake_socket_passes_through_listen():
def test_fake_socket_passes_through_getpeername():
import socket
+
HTTPretty.enable()
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.truesock = MagicMock()
@@ -356,6 +376,7 @@ def test_fake_socket_passes_through_getpeername():
def test_fake_socket_passes_through_getsockname():
import socket
+
HTTPretty.enable()
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.truesock = MagicMock()
@@ -365,6 +386,7 @@ def test_fake_socket_passes_through_getsockname():
def test_fake_socket_passes_through_gettimeout():
import socket
+
HTTPretty.enable()
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.truesock = MagicMock()
@@ -374,6 +396,7 @@ def test_fake_socket_passes_through_gettimeout():
def test_fake_socket_passes_through_shutdown():
import socket
+
HTTPretty.enable()
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.truesock = MagicMock()
@@ -383,11 +406,12 @@ def test_fake_socket_passes_through_shutdown():
def test_unix_socket():
import socket
+
HTTPretty.enable()
# Create a UDS socket
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
- server_address = './not-exist-socket'
+ server_address = "./not-exist-socket"
try:
sock.connect(server_address)
except socket.error:
@@ -397,32 +421,32 @@ def test_unix_socket():
def test_HTTPrettyRequest_json_body():
""" A content-type of application/json should parse a valid json body """
- header = TEST_HEADER % {'content_type': 'application/json'}
- test_dict = {'hello': 'world'}
+ header = TEST_HEADER % {"content_type": "application/json"}
+ test_dict = {"hello": "world"}
request = HTTPrettyRequest(header, json.dumps(test_dict))
expect(request.parsed_body).to.equal(test_dict)
def test_HTTPrettyRequest_invalid_json_body():
""" A content-type of application/json with an invalid json body should return the content unaltered """
- header = TEST_HEADER % {'content_type': 'application/json'}
- invalid_json = u"{'hello', 'world','thisstringdoesntstops}"
+ header = TEST_HEADER % {"content_type": "application/json"}
+ invalid_json = "{'hello', 'world','thisstringdoesntstops}"
request = HTTPrettyRequest(header, invalid_json)
expect(request.parsed_body).to.equal(invalid_json)
def test_HTTPrettyRequest_queryparam():
""" A content-type of x-www-form-urlencoded with a valid queryparam body should return parsed content """
- header = TEST_HEADER % {'content_type': 'application/x-www-form-urlencoded'}
- valid_queryparam = u"hello=world&this=isavalidquerystring"
- valid_results = {'hello': ['world'], 'this': ['isavalidquerystring']}
+ header = TEST_HEADER % {"content_type": "application/x-www-form-urlencoded"}
+ valid_queryparam = "hello=world&this=isavalidquerystring"
+ valid_results = {"hello": ["world"], "this": ["isavalidquerystring"]}
request = HTTPrettyRequest(header, valid_queryparam)
expect(request.parsed_body).to.equal(valid_results)
def test_HTTPrettyRequest_arbitrarypost():
""" A non-handled content type request's post body should return the content unaltered """
- header = TEST_HEADER % {'content_type': 'thisis/notarealcontenttype'}
+ header = TEST_HEADER % {"content_type": "thisis/notarealcontenttype"}
gibberish_body = "1234567890!@#$%^&*()"
request = HTTPrettyRequest(header, gibberish_body)
expect(request.parsed_body).to.equal(gibberish_body)
@@ -435,8 +459,9 @@ def test_socktype_bad_python_version_regression():
https://bugs.python.org/issue20386
"""
import socket
+
someObject = object()
- with patch('socket.SocketType', someObject):
+ with patch("socket.SocketType", someObject):
HTTPretty.enable()
expect(socket.SocketType).to.equal(someObject)
HTTPretty.disable()
@@ -444,7 +469,8 @@ def test_socktype_bad_python_version_regression():
def test_socktype_good_python_version():
import socket
- with patch('socket.SocketType', socket.socket):
+
+ with patch("socket.SocketType", socket.socket):
HTTPretty.enable()
expect(socket.SocketType).to.equal(socket.socket)
HTTPretty.disable()
@@ -454,21 +480,17 @@ def test_httpretty_should_allow_registering_regex_hostnames():
"HTTPretty should allow registering regexes with requests"
HTTPretty.register_uri(
- HTTPretty.GET,
- re.compile(r'^http://\w+\.foo\.com/baz$'),
- body="yay",
+ HTTPretty.GET, re.compile(r"^http://\w+\.foo\.com/baz$"), body="yay"
)
- assert HTTPretty.match_http_address('www.foo.com', 80)
+ assert HTTPretty.match_http_address("www.foo.com", 80)
def test_httpretty_should_allow_registering_regex_hostnames_ssl():
"HTTPretty should allow registering regexes with requests (ssl version)"
HTTPretty.register_uri(
- HTTPretty.GET,
- re.compile(r'^https://\w+\.foo\.com/baz$'),
- body="yay",
+ HTTPretty.GET, re.compile(r"^https://\w+\.foo\.com/baz$"), body="yay"
)
- assert HTTPretty.match_https_hostname('www.foo.com')
+ assert HTTPretty.match_https_hostname("www.foo.com")