Skip to content
Snippets Groups Projects
Commit 353292354985 authored by Luke Bakken's avatar Luke Bakken
Browse files

Remove ssl_options tests as in the master branch

parent 0e70d49d33ed
No related branches found
No related tags found
No related merge requests found
Show whitespace changes
Inline Side-by-side
tests/unit/base_connection_tests.py
+ 0
− 117
View file @ 35329235
......@@ -83,120 +83,3 @@
keepalive_call = mock.call.setsockopt(socket.SOL_SOCKET,
socket.SO_KEEPALIVE, 1)
self.assertNotIn(keepalive_call, sock_mock.method_calls)
def test_ssl_wrap_socket_with_none_ssl_options(self):
params = pika.ConnectionParameters(ssl_options=None)
self.assertIsNone(params.ssl_options)
with mock.patch('pika.connection.Connection.connect'):
conn = base_connection.BaseConnection(parameters=params)
with mock.patch('pika.adapters.base_connection'
'.ssl.wrap_socket') as wrap_socket_mock:
sock_mock = mock.Mock()
conn._wrap_socket(sock_mock)
wrap_socket_mock.assert_called_once_with(
sock_mock, do_handshake_on_connect=conn.DO_HANDSHAKE)
def test_ssl_wrap_socket_with_dict_ssl_options(self):
ssl_options = dict(ssl='options', handshake=False)
params = pika.ConnectionParameters(ssl_options=ssl_options)
self.assertEqual(params.ssl_options, ssl_options)
with mock.patch('pika.connection.Connection.connect'):
conn = base_connection.BaseConnection(parameters=params)
with mock.patch('pika.adapters.base_connection'
'.ssl.wrap_socket') as wrap_socket_mock:
sock_mock = mock.Mock()
conn._wrap_socket(sock_mock)
wrap_socket_mock.assert_called_once_with(
sock_mock,
do_handshake_on_connect=conn.DO_HANDSHAKE,
ssl='options',
handshake=False)
@mock.patch('ssl.SSLContext.load_cert_chain')
@mock.patch('ssl.SSLContext.load_verify_locations')
@mock.patch('ssl.SSLContext.set_ciphers')
@mock.patch('ssl.SSLContext.wrap_socket')
@unittest.skipIf(sys.version_info < (2,7,0), 'Unavailable ssl features')
def test_ssl_wrap_socket_with_default_ssl_options_obj(self,
wrap_socket_mock,
set_ciphers_mock,
load_verify_mock,
load_certs_mock):
ssl_options = pika.SSLOptions()
params = pika.ConnectionParameters(ssl_options=ssl_options)
#self.assertEqual(params.ssl_options, ssl_options)
with mock.patch('pika.connection.Connection.connect'):
conn = base_connection.BaseConnection(parameters=params)
sock_mock = mock.Mock()
conn._wrap_socket(sock_mock)
load_certs_mock.assert_not_called()
load_verify_mock.assert_not_called()
# the __init__ of SSLContext calls set_ciphers,
# hence the 'called once'
set_ciphers_mock.assert_called_once()
wrap_socket_mock.assert_called_once_with(
sock_mock,
server_side=False,
do_handshake_on_connect=conn.DO_HANDSHAKE,
suppress_ragged_eofs=True,
server_hostname=None
)
@mock.patch('ssl.SSLContext.load_cert_chain')
@mock.patch('ssl.SSLContext.load_verify_locations')
@mock.patch('ssl.SSLContext.set_ciphers')
@mock.patch('ssl.SSLContext.wrap_socket')
@unittest.skipIf(sys.version_info < (2,7,0), 'Unavailable ssl features')
def test_ssl_wrap_socket_with_ssl_options_obj(self,
wrap_socket_mock,
set_ciphers_mock,
load_verify_mock,
load_certs_mock):
ssl_options = pika.SSLOptions(certfile='/some/cert/file.crt',
keyfile='/some/key/file.crt',
key_password='pa55w0rd',
cafile='/some/ca/file.crt',
capath='/some/ca/path',
cadata='/some/data/or/something',
ciphers='ciphers',
server_hostname='some.virtual.host',
do_handshake_on_connect=False,
suppress_ragged_eofs=False,
server_side=True)
params = pika.ConnectionParameters(ssl_options=ssl_options)
#self.assertEqual(params.ssl_options, ssl_options)
with mock.patch('pika.connection.Connection.connect'):
conn = base_connection.BaseConnection(parameters=params)
sock_mock = mock.Mock()
conn._wrap_socket(sock_mock)
load_certs_mock.assert_called_once_with(certfile='/some/cert/file.crt',
keyfile='/some/key/file.crt',
password='pa55w0rd')
load_verify_mock.assert_called_once_with(cafile='/some/ca/file.crt',
capath='/some/ca/path',
cadata='/some/data/or/something')
# the constructor of SSLContext calls set_ciphers as well
set_ciphers_mock.assert_called_with('ciphers')
wrap_socket_mock.assert_called_once_with(
sock_mock,
server_side=True,
do_handshake_on_connect=False,
suppress_ragged_eofs=False,
server_hostname='some.virtual.host'
)
tests/unit/connection_parameters_tests.py
+ 6
− 20
View file @ 35329235
......@@ -605,14 +605,7 @@
'locale': 'en_UK',
'retry_delay': 3,
'socket_timeout': 100.5,
'ssl_options': {
'ca_certs': '/etc/ssl',
'certfile': '/etc/certs/cert.pem',
'keyfile': '/etc/certs/key.pem',
'password': 'test123',
'ciphers': None,
'server_hostname': 'blah.blah.com'
},
'ssl_options': None,
'tcp_options': {
'TCP_USER_TIMEOUT': 1000,
'TCP_KEEPIDLE': 60
......@@ -637,10 +630,6 @@
expected_value = query_args[t_param]
actual_value = getattr(params, t_param)
if t_param == 'ssl_options':
self.assertEqual(actual_value.server_hostname,
expected_value['server_hostname'])
else:
self.assertEqual(
actual_value,
expected_value,
......@@ -651,9 +640,6 @@
backpressure == 't')
# check all values from base URL
self.assertIsNotNone(params.ssl_options)
self.assertIsNotNone(params.ssl_options.context)
self.assertIsInstance(params.ssl_options.context, ssl.SSLContext)
self.assertEqual(params.credentials.username, 'myuser')
self.assertEqual(params.credentials.password, 'mypass')
self.assertEqual(params.host, 'www.test.com')
......
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment