Rewrite test_utils.py to be pytest-style
MODIFY
test/test_util.py
test/test_util.py
@@ -2,12 +2,12 @@ import hashlib
import warnings
import logging
import io
-import unittest
import ssl
import socket
from itertools import chain
from mock import patch, Mock
+import pytest
from urllib3 import add_stderr_logger, disable_warnings
from urllib3.util.request import make_headers, rewind_body, _FAILEDTELL
@@ -48,8 +48,8 @@ from . import clear_warnings
TIMEOUT_EPOCH = 1000
-class TestUtil(unittest.TestCase):
- def test_get_host(self):
+class TestUtil(object):
+
url_host_map = {
# Hosts
'http://google.com/mail': ('http', 'google.com', None),
@@ -92,30 +92,7 @@ class TestUtil(unittest.TestCase):
'http://[::192.9.5.5]/ipng': ('http', '[::192.9.5.5]', None),
'http://[::ffff:129.144.52.38]:42/index.html': ('http', '[::ffff:129.144.52.38]', 42),
'http://[2010:836b:4179::836b:4179]': ('http', '[2010:836b:4179::836b:4179]', None),
- }
- for url, expected_host in url_host_map.items():
- returned_host = get_host(url)
- self.assertEqual(returned_host, expected_host)
- def test_invalid_host(self):
- # TODO: Add more tests
- invalid_host = [
- 'http://google.com:foo',
- 'http://::1/',
- 'http://::1:80/',
- 'http://google.com:-80',
- six.u('http://google.com:\xb2\xb2'), # \xb2 = ^2
- ]
-
- for location in invalid_host:
- self.assertRaises(LocationParseError, get_host, location)
-
- def test_host_normalization(self):
- """
- Asserts the scheme and hosts with a normalizable scheme are
- converted to lower-case.
- """
- url_host_map = {
# Hosts
'HTTP://GOOGLE.COM/mail/': ('http', 'google.com', None),
'GOogle.COM/mail': ('http', 'google.com', None),
@@ -132,22 +109,35 @@ class TestUtil(unittest.TestCase):
'http+UNIX://%2fvar%2frun%2fSOCKET/path': (
'http+unix', '%2fvar%2frun%2fSOCKET', None),
}
- for url, expected_host in url_host_map.items():
+
+ @pytest.mark.parametrize('url, expected_host', url_host_map.items())
+ def test_get_host(self, url, expected_host):
returned_host = get_host(url)
- self.assertEqual(returned_host, expected_host)
+ assert returned_host == expected_host
- def test_parse_url_normalization(self):
- """Assert parse_url normalizes the scheme/host, and only the scheme/host"""
- test_urls = [
+ # TODO: Add more tests
+ @pytest.mark.parametrize('location', [
+ 'http://google.com:foo',
+ 'http://::1/',
+ 'http://::1:80/',
+ 'http://google.com:-80',
+ six.u('http://google.com:\xb2\xb2'), # \xb2 = ^2
+ ])
+ def test_invalid_host(self, location):
+ with pytest.raises(LocationParseError):
+ get_host(location)
+
+ @pytest.mark.parametrize('url, expected_normalized_url', [
('HTTP://GOOGLE.COM/MAIL/', 'http://google.com/MAIL/'),
('HTTP://JeremyCline:Hunter2@Example.com:8080/',
'http://JeremyCline:Hunter2@example.com:8080/'),
('HTTPS://Example.Com/?Key=Value', 'https://example.com/?Key=Value'),
('Https://Example.Com/#Fragment', 'https://example.com/#Fragment'),
- ]
- for url, expected_normalized_url in test_urls:
+ ])
+ def test_parse_url_normalization(self, url, expected_normalized_url):
+ """Assert parse_url normalizes the scheme/host, and only the scheme/host"""
actual_normalized_url = parse_url(url).url
- self.assertEqual(actual_normalized_url, expected_normalized_url)
+ assert actual_normalized_url == expected_normalized_url
parse_url_host_map = [
('http://google.com/mail', Url('http', host='google.com', path='/mail')),
@@ -191,28 +181,29 @@ class TestUtil(unittest.TestCase):
# Empty Port
'http://google.com:': Url('http', host='google.com'),
'http://google.com:/': Url('http', host='google.com', path='/'),
-
}
- def test_parse_url(self):
- for url, expected_Url in chain(self.parse_url_host_map,
- self.non_round_tripping_parse_url_host_map.items()):
+ @pytest.mark.parametrize(
+ 'url, expected_Url',
+ chain(parse_url_host_map, non_round_tripping_parse_url_host_map.items())
+ )
+ def test_parse_url(self, url, expected_Url):
returned_Url = parse_url(url)
- self.assertEqual(returned_Url, expected_Url)
+ assert returned_Url == expected_Url
- def test_unparse_url(self):
- for url, expected_Url in self.parse_url_host_map:
- self.assertEqual(url, expected_Url.url)
+ @pytest.mark.parametrize('url, expected_Url', parse_url_host_map)
+ def test_unparse_url(self, url, expected_Url):
+ assert url == expected_Url.url
def test_parse_url_invalid_IPv6(self):
- self.assertRaises(ValueError, parse_url, '[::1')
+ with pytest.raises(ValueError):
+ parse_url('[::1')
def test_Url_str(self):
U = Url('http', host='google.com')
- self.assertEqual(str(U), U.url)
+ assert str(U) == U.url
- def test_request_uri(self):
- url_host_map = {
+ request_uri_map = {
'http://google.com/mail': '/mail',
'http://google.com/mail/': '/mail/',
'http://google.com/': '/',
@@ -223,11 +214,12 @@ class TestUtil(unittest.TestCase):
'#': '/',
'/foo?bar=baz': '/foo?bar=baz',
}
- for url, expected_request_uri in url_host_map.items():
+
+ @pytest.mark.parametrize('url, expected_request_uri', request_uri_map.items())
+ def test_request_uri(self, url, expected_request_uri):
returned_url = parse_url(url)
- self.assertEqual(returned_url.request_uri, expected_request_uri)
+ assert returned_url.request_uri == expected_request_uri
- def test_netloc(self):
url_netloc_map = {
'http://google.com/mail': 'google.com',
'http://google.com:80/mail': 'google.com:80',
@@ -235,56 +227,43 @@ class TestUtil(unittest.TestCase):
'google.com:12345': 'google.com:12345',
}
- for url, expected_netloc in url_netloc_map.items():
- self.assertEqual(parse_url(url).netloc, expected_netloc)
-
- def test_make_headers(self):
- self.assertEqual(
- make_headers(accept_encoding=True),
- {'accept-encoding': 'gzip,deflate'})
-
- self.assertEqual(
- make_headers(accept_encoding='foo,bar'),
- {'accept-encoding': 'foo,bar'})
-
- self.assertEqual(
- make_headers(accept_encoding=['foo', 'bar']),
- {'accept-encoding': 'foo,bar'})
-
- self.assertEqual(
- make_headers(accept_encoding=True, user_agent='banana'),
- {'accept-encoding': 'gzip,deflate', 'user-agent': 'banana'})
-
- self.assertEqual(
- make_headers(user_agent='banana'),
- {'user-agent': 'banana'})
-
- self.assertEqual(
- make_headers(keep_alive=True),
- {'connection': 'keep-alive'})
-
- self.assertEqual(
- make_headers(basic_auth='foo:bar'),
- {'authorization': 'Basic Zm9vOmJhcg=='})
-
- self.assertEqual(
- make_headers(proxy_basic_auth='foo:bar'),
- {'proxy-authorization': 'Basic Zm9vOmJhcg=='})
-
- self.assertEqual(
- make_headers(disable_cache=True),
- {'cache-control': 'no-cache'})
+ @pytest.mark.parametrize('url, expected_netloc', url_netloc_map.items())
+ def test_netloc(self, url, expected_netloc):
+ assert parse_url(url).netloc == expected_netloc
+
+ @pytest.mark.parametrize('kwargs, expected', [
+ ({'accept_encoding': True},
+ {'accept-encoding': 'gzip,deflate'}),
+ ({'accept_encoding': 'foo,bar'},
+ {'accept-encoding': 'foo,bar'}),
+ ({'accept_encoding': ['foo', 'bar']},
+ {'accept-encoding': 'foo,bar'}),
+ ({'accept_encoding': True, 'user_agent': 'banana'},
+ {'accept-encoding': 'gzip,deflate', 'user-agent': 'banana'}),
+ ({'user_agent': 'banana'},
+ {'user-agent': 'banana'}),
+ ({'keep_alive': True},
+ {'connection': 'keep-alive'}),
+ ({'basic_auth': 'foo:bar'},
+ {'authorization': 'Basic Zm9vOmJhcg=='}),
+ ({'proxy_basic_auth': 'foo:bar'},
+ {'proxy-authorization': 'Basic Zm9vOmJhcg=='}),
+ ({'disable_cache': True},
+ {'cache-control': 'no-cache'}),
+ ])
+ def test_make_headers(self, kwargs, expected):
+ assert make_headers(**kwargs) == expected
def test_rewind_body(self):
body = io.BytesIO(b'test data')
- self.assertEqual(body.read(), b'test data')
+ assert body.read() == b'test data'
# Assert the file object has been consumed
- self.assertEqual(body.read(), b'')
+ assert body.read() == b''
# Rewind it back to just be b'data'
rewind_body(body, 5)
- self.assertEqual(body.read(), b'data')
+ assert body.read() == b'data'
def test_rewind_body_failed_tell(self):
body = io.BytesIO(b'test data')
@@ -292,15 +271,18 @@ class TestUtil(unittest.TestCase):
# Simulate failed tell()
body_pos = _FAILEDTELL
- self.assertRaises(UnrewindableBodyError, rewind_body, body, body_pos)
+ with pytest.raises(UnrewindableBodyError):
+ rewind_body(body, body_pos)
def test_rewind_body_bad_position(self):
body = io.BytesIO(b'test data')
body.read() # Consume body
# Pass non-integer position
- self.assertRaises(ValueError, rewind_body, body, None)
- self.assertRaises(ValueError, rewind_body, body, object())
+ with pytest.raises(ValueError):
+ rewind_body(body, body_pos=None)
+ with pytest.raises(ValueError):
+ rewind_body(body, body_pos=object())
def test_rewind_body_failed_seek(self):
class BadSeek():
@@ -308,24 +290,24 @@ class TestUtil(unittest.TestCase):
def seek(self, pos, offset=0):
raise IOError
- self.assertRaises(UnrewindableBodyError, rewind_body, BadSeek(), 2)
+ with pytest.raises(UnrewindableBodyError):
+ rewind_body(BadSeek(), body_pos=2)
- def test_split_first(self):
- test_cases = {
+ @pytest.mark.parametrize('input, expected', {
('abcd', 'b'): ('a', 'cd', 'b'),
('abcd', 'cb'): ('a', 'cd', 'b'),
('abcd', ''): ('abcd', '', None),
('abcd', 'a'): ('', 'bcd', 'a'),
('abcd', 'ab'): ('', 'bcd', 'a'),
- }
- for input, expected in test_cases.items():
+ }.items())
+ def test_split_first(self, input, expected):
output = split_first(*input)
- self.assertEqual(output, expected)
+ assert output == expected
def test_add_stderr_logger(self):
handler = add_stderr_logger(level=logging.INFO) # Don't actually print debug
logger = logging.getLogger('urllib3')
- self.assertTrue(handler in logger.handlers)
+ assert handler in logger.handlers
logger.debug('Testing add_stderr_logger')
logger.removeHandler(handler)
@@ -334,10 +316,10 @@ class TestUtil(unittest.TestCase):
with warnings.catch_warnings(record=True) as w:
clear_warnings()
warnings.warn('This is a test.', InsecureRequestWarning)
- self.assertEqual(len(w), 1)
+ assert len(w) == 1
disable_warnings()
warnings.warn('This is a test.', InsecureRequestWarning)
- self.assertEqual(len(w), 1)
+ assert len(w) == 1
def _make_time_pass(self, seconds, timeout, time_mock):
""" Make some time pass for the timeout object """
@@ -346,47 +328,19 @@ class TestUtil(unittest.TestCase):
time_mock.return_value = TIMEOUT_EPOCH + seconds
return timeout
- def test_invalid_timeouts(self):
- try:
- Timeout(total=-1)
- self.fail("negative value should throw exception")
- except ValueError as e:
- self.assertTrue('less than' in str(e))
- try:
- Timeout(connect=2, total=-1)
- self.fail("negative value should throw exception")
- except ValueError as e:
- self.assertTrue('less than' in str(e))
-
- try:
- Timeout(read=-1)
- self.fail("negative value should throw exception")
- except ValueError as e:
- self.assertTrue('less than' in str(e))
-
- try:
- Timeout(connect=False)
- self.fail("boolean values should throw exception")
- except ValueError as e:
- self.assertTrue('cannot be a boolean' in str(e))
-
- try:
- Timeout(read=True)
- self.fail("boolean values should throw exception")
- except ValueError as e:
- self.assertTrue('cannot be a boolean' in str(e))
-
- try:
- Timeout(connect=0)
- self.fail("value <= 0 should throw exception")
- except ValueError as e:
- self.assertTrue('less than or equal' in str(e))
-
- try:
- Timeout(read="foo")
- self.fail("string value should not be allowed")
- except ValueError as e:
- self.assertTrue('int, float or None' in str(e))
+ @pytest.mark.parametrize('kwargs, message', [
+ ({'total': -1}, 'less than'),
+ ({'connect': 2, 'total': -1}, 'less than'),
+ ({'read': -1}, 'less than'),
+ ({'connect': False}, 'cannot be a boolean'),
+ ({'read': True}, 'cannot be a boolean'),
+ ({'connect': 0}, 'less than or equal'),
+ ({'read': 'foo'}, 'int, float or None')
+ ])
+ def test_invalid_timeouts(self, kwargs, message):
+ with pytest.raises(ValueError) as e:
+ Timeout(**kwargs)
+ assert message in str(e.value)
@patch('urllib3.util.timeout.current_time')
def test_timeout(self, current_time):
@@ -395,70 +349,78 @@ class TestUtil(unittest.TestCase):
# make 'no time' elapse
timeout = self._make_time_pass(seconds=0, timeout=timeout,
time_mock=current_time)
- self.assertEqual(timeout.read_timeout, 3)
- self.assertEqual(timeout.connect_timeout, 3)
+ assert timeout.read_timeout == 3
+ assert timeout.connect_timeout == 3
timeout = Timeout(total=3, connect=2)
- self.assertEqual(timeout.connect_timeout, 2)
+ assert timeout.connect_timeout == 2
timeout = Timeout()
- self.assertEqual(timeout.connect_timeout, Timeout.DEFAULT_TIMEOUT)
+ assert timeout.connect_timeout == Timeout.DEFAULT_TIMEOUT
# Connect takes 5 seconds, leaving 5 seconds for read
timeout = Timeout(total=10, read=7)
timeout = self._make_time_pass(seconds=5, timeout=timeout,
time_mock=current_time)
- self.assertEqual(timeout.read_timeout, 5)
+ assert timeout.read_timeout == 5
# Connect takes 2 seconds, read timeout still 7 seconds
timeout = Timeout(total=10, read=7)
timeout = self._make_time_pass(seconds=2, timeout=timeout,
time_mock=current_time)
- self.assertEqual(timeout.read_timeout, 7)
+ assert timeout.read_timeout == 7
timeout = Timeout(total=10, read=7)
- self.assertEqual(timeout.read_timeout, 7)
+ assert timeout.read_timeout == 7
timeout = Timeout(total=None, read=None, connect=None)
- self.assertEqual(timeout.connect_timeout, None)
- self.assertEqual(timeout.read_timeout, None)
- self.assertEqual(timeout.total, None)
+ assert timeout.connect_timeout is None
+ assert timeout.read_timeout is None
+ assert timeout.total is None
timeout = Timeout(5)
- self.assertEqual(timeout.total, 5)
+ assert timeout.total == 5
def test_timeout_str(self):
timeout = Timeout(connect=1, read=2, total=3)
- self.assertEqual(str(timeout), "Timeout(connect=1, read=2, total=3)")
+ assert str(timeout) == "Timeout(connect=1, read=2, total=3)"
timeout = Timeout(connect=1, read=None, total=3)
- self.assertEqual(str(timeout), "Timeout(connect=1, read=None, total=3)")
+ assert str(timeout) == "Timeout(connect=1, read=None, total=3)"
@patch('urllib3.util.timeout.current_time')
def test_timeout_elapsed(self, current_time):
current_time.return_value = TIMEOUT_EPOCH
timeout = Timeout(total=3)
- self.assertRaises(TimeoutStateError, timeout.get_connect_duration)
+ with pytest.raises(TimeoutStateError):
+ timeout.get_connect_duration()
timeout.start_connect()
- self.assertRaises(TimeoutStateError, timeout.start_connect)
+ with pytest.raises(TimeoutStateError):
+ timeout.start_connect()
current_time.return_value = TIMEOUT_EPOCH + 2
- self.assertEqual(timeout.get_connect_duration(), 2)
+ assert timeout.get_connect_duration() == 2
current_time.return_value = TIMEOUT_EPOCH + 37
- self.assertEqual(timeout.get_connect_duration(), 37)
-
- def test_resolve_cert_reqs(self):
- self.assertEqual(resolve_cert_reqs(None), ssl.CERT_NONE)
- self.assertEqual(resolve_cert_reqs(ssl.CERT_NONE), ssl.CERT_NONE)
- self.assertEqual(resolve_cert_reqs(ssl.CERT_REQUIRED), ssl.CERT_REQUIRED)
- self.assertEqual(resolve_cert_reqs('REQUIRED'), ssl.CERT_REQUIRED)
- self.assertEqual(resolve_cert_reqs('CERT_REQUIRED'), ssl.CERT_REQUIRED)
-
- def test_resolve_ssl_version(self):
- self.assertEqual(resolve_ssl_version(ssl.PROTOCOL_TLSv1), ssl.PROTOCOL_TLSv1)
- self.assertEqual(resolve_ssl_version("PROTOCOL_TLSv1"), ssl.PROTOCOL_TLSv1)
- self.assertEqual(resolve_ssl_version("TLSv1"), ssl.PROTOCOL_TLSv1)
- self.assertEqual(resolve_ssl_version(ssl.PROTOCOL_SSLv23), ssl.PROTOCOL_SSLv23)
+ assert timeout.get_connect_duration() == 37
+
+ @pytest.mark.parametrize('candidate, requirements', [
+ (None, ssl.CERT_NONE),
+ (ssl.CERT_NONE, ssl.CERT_NONE),
+ (ssl.CERT_REQUIRED, ssl.CERT_REQUIRED),
+ ('REQUIRED', ssl.CERT_REQUIRED),
+ ('CERT_REQUIRED', ssl.CERT_REQUIRED),
+ ])
+ def test_resolve_cert_reqs(self, candidate, requirements):
+ assert resolve_cert_reqs(candidate) == requirements
+
+ @pytest.mark.parametrize('candidate, version', [
+ (ssl.PROTOCOL_TLSv1, ssl.PROTOCOL_TLSv1),
+ ("PROTOCOL_TLSv1", ssl.PROTOCOL_TLSv1),
+ ("TLSv1", ssl.PROTOCOL_TLSv1),
+ (ssl.PROTOCOL_SSLv23, ssl.PROTOCOL_SSLv23),
+ ])
+ def test_resolve_ssl_version(self, candidate, version):
+ assert resolve_ssl_version(candidate) == version
def test_is_fp_closed_object_supports_closed(self):
class ClosedFile(object):
@@ -466,7 +428,7 @@ class TestUtil(unittest.TestCase):
def closed(self):
return True
- self.assertTrue(is_fp_closed(ClosedFile()))
+ assert is_fp_closed(ClosedFile())
def test_is_fp_closed_object_has_none_fp(self):
class NoneFpFile(object):
@@ -474,7 +436,7 @@ class TestUtil(unittest.TestCase):
def fp(self):
return None
- self.assertTrue(is_fp_closed(NoneFpFile()))
+ assert is_fp_closed(NoneFpFile())
def test_is_fp_closed_object_has_fp(self):
class FpFile(object):
@@ -482,13 +444,14 @@ class TestUtil(unittest.TestCase):
def fp(self):
return True
- self.assertTrue(not is_fp_closed(FpFile()))
+ assert not is_fp_closed(FpFile())
def test_is_fp_closed_object_has_neither_fp_nor_closed(self):
class NotReallyAFile(object):
pass
- self.assertRaises(ValueError, is_fp_closed, NotReallyAFile())
+ with pytest.raises(ValueError):
+ is_fp_closed(NotReallyAFile())
def test_ssl_wrap_socket_loads_the_cert_chain(self):
socket = object()
@@ -497,7 +460,8 @@ class TestUtil(unittest.TestCase):
certfile='/path/to/certfile')
mock_context.load_cert_chain.assert_called_once_with(
- '/path/to/certfile', None)
+ '/path/to/certfile', None
+ )
@patch('urllib3.util.ssl_.create_urllib3_context')
def test_ssl_wrap_socket_creates_new_context(self,
@@ -515,7 +479,8 @@ class TestUtil(unittest.TestCase):
ssl_wrap_socket(ssl_context=mock_context, ca_certs='/path/to/pem',
sock=socket)
mock_context.load_verify_locations.assert_called_once_with(
- '/path/to/pem', None)
+ '/path/to/pem', None
+ )
def test_ssl_wrap_socket_loads_certificate_directories(self):
socket = object()
@@ -523,7 +488,8 @@ class TestUtil(unittest.TestCase):
ssl_wrap_socket(ssl_context=mock_context, ca_cert_dir='/path/to/pems',
sock=socket)
mock_context.load_verify_locations.assert_called_once_with(
- None, '/path/to/pems')
+ None, '/path/to/pems'
+ )
def test_ssl_wrap_socket_with_no_sni_warns(self):
socket = object()
@@ -543,55 +509,59 @@ class TestUtil(unittest.TestCase):
def test_const_compare_digest_fallback(self):
target = hashlib.sha256(b'abcdef').digest()
- self.assertTrue(_const_compare_digest_backport(target, target))
+ assert _const_compare_digest_backport(target, target)
prefix = target[:-1]
- self.assertFalse(_const_compare_digest_backport(target, prefix))
+ assert not _const_compare_digest_backport(target, prefix)
suffix = target + b'0'
- self.assertFalse(_const_compare_digest_backport(target, suffix))
+ assert not _const_compare_digest_backport(target, suffix)
incorrect = hashlib.sha256(b'xyz').digest()
- self.assertFalse(_const_compare_digest_backport(target, incorrect))
+ assert not _const_compare_digest_backport(target, incorrect)
def test_has_ipv6_disabled_on_compile(self):
with patch('socket.has_ipv6', False):
- self.assertFalse(_has_ipv6('::1'))
+ assert not _has_ipv6('::1')
def test_has_ipv6_enabled_but_fails(self):
with patch('socket.has_ipv6', True):
with patch('socket.socket') as mock:
instance = mock.return_value
instance.bind = Mock(side_effect=Exception('No IPv6 here!'))
- self.assertFalse(_has_ipv6('::1'))
+ assert not _has_ipv6('::1')
def test_has_ipv6_enabled_and_working(self):
with patch('socket.has_ipv6', True):
with patch('socket.socket') as mock:
instance = mock.return_value
instance.bind.return_value = True
- self.assertTrue(_has_ipv6('::1'))
+ assert _has_ipv6('::1')
def test_ip_family_ipv6_enabled(self):
with patch('urllib3.util.connection.HAS_IPV6', True):
- self.assertEqual(allowed_gai_family(), socket.AF_UNSPEC)
+ assert allowed_gai_family() == socket.AF_UNSPEC
def test_ip_family_ipv6_disabled(self):
with patch('urllib3.util.connection.HAS_IPV6', False):
- self.assertEqual(allowed_gai_family(), socket.AF_INET)
+ assert allowed_gai_family() == socket.AF_INET
- def test_parse_retry_after(self):
- invalid = [
+ @pytest.mark.parametrize('value', [
"-1",
"+1",
"1.0",
six.u("\xb2"), # \xb2 = ^2
- ]
+ ])
+ def test_parse_retry_after_invalid(self, value):
retry = Retry()
-
- for value in invalid:
- self.assertRaises(InvalidHeader, retry.parse_retry_after, value)
-
- self.assertEqual(retry.parse_retry_after("0"), 0)
- self.assertEqual(retry.parse_retry_after("1000"), 1000)
- self.assertEqual(retry.parse_retry_after("\t42 "), 42)
+ with pytest.raises(InvalidHeader):
+ retry.parse_retry_after(value)
+
+ @pytest.mark.parametrize('value, expected', [
+ ("0", 0),
+ ("1000", 1000),
+ ("\t42 ", 42),
+ ])
+ def test_parse_retry_after(self, value, expected):
+ retry = Retry()
+ assert retry.parse_retry_after(value) == expected

Rewrite test_utils.py to be pytest-style