HASH_MATCH_HEADER = 'X-BSN-BVS-HASH-MATCH'
# error messages
NXNETWORK = 'NXVNS'
+# NOTE(kevinbenton): This following is to give mock a target that doesn't
+# affect other users of httplib.HTTPConnection
+HTTPConnection = httplib.HTTPConnection
class RemoteRestError(exceptions.NeutronException):
return 0, None, None, None
self.currentconn.combined_cert = self.combined_cert
else:
- self.currentconn = httplib.HTTPConnection(
+ self.currentconn = HTTPConnection(
self.server, self.port, timeout=timeout)
if self.currentconn is None:
LOG.error(_('ServerProxy: Could not establish HTTP '
self.setup_config_files()
self.setup_patches()
super(BigSwitchDhcpAgentNotifierTestCase, self).setUp()
+ self.startHttpPatch()
CALLBACKS = 'neutron.plugins.bigswitch.plugin.RestProxyCallbacks'
CERTFETCH = 'neutron.plugins.bigswitch.servermanager.ServerPool._fetch_cert'
SERVER_MANAGER = 'neutron.plugins.bigswitch.servermanager'
-HTTPCON = 'httplib.HTTPConnection'
-SPAWN = 'eventlet.GreenPool.spawn_n'
+HTTPCON = 'neutron.plugins.bigswitch.servermanager.HTTPConnection'
+SPAWN = 'neutron.plugins.bigswitch.plugin.eventlet.GreenPool.spawn_n'
CWATCH = SERVER_MANAGER + '.ServerPool._consistency_watchdog'
cfg.CONF.set_override('cache_connections', False, 'RESTPROXY')
def setup_patches(self):
- self.httpPatch = mock.patch(HTTPCON, create=True,
- new=fake_server.HTTPConnectionMock)
self.plugin_notifier_p = mock.patch(NOTIFIER)
self.callbacks_p = mock.patch(CALLBACKS)
self.spawn_p = mock.patch(SPAWN)
self.addCleanup(db.clear_db)
self.callbacks_p.start()
self.plugin_notifier_p.start()
- self.httpPatch.start()
self.spawn_p.start()
self.watch_p.start()
+
+ def startHttpPatch(self):
+ self.httpPatch = mock.patch(HTTPCON,
+ new=fake_server.HTTPConnectionMock)
+ self.httpPatch.start()
SERVERMANAGER = PLUGIN + '.servermanager'
SERVERPOOL = SERVERMANAGER + '.ServerPool'
SERVERRESTCALL = SERVERMANAGER + '.ServerProxy.rest_call'
+HTTPCON = SERVERMANAGER + '.HTTPConnection'
-class CapabilitiesTests(test_router_db.RouterDBTestCase):
+class CapabilitiesTests(test_router_db.RouterDBTestBase):
def test_floating_ip_capability(self):
with nested(
all_floats = [f['floating_ip_address']
for floats in updates for f in floats]
self.assertIn(fip['floatingip']['floating_ip_address'], all_floats)
+
+ def test_keep_alive_capability(self):
+ with mock.patch(
+ SERVERRESTCALL, return_value=(200, None, None, '["keep-alive"]')
+ ):
+ # perform a task to cause capabilities to be retrieved
+ with self.floatingip_with_assoc():
+ pass
+ # now mock HTTP class instead of REST so we can see headers
+ conmock = mock.patch(HTTPCON).start()
+ instance = conmock.return_value
+ instance.getresponse.return_value.getheader.return_value = 'HASHHEADER'
+ with self.network():
+ callheaders = instance.request.mock_calls[0][1][3]
+ self.assertIn('Connection', callheaders)
+ self.assertEqual(callheaders['Connection'], 'keep-alive')
import neutron.tests.unit.test_extension_allowedaddresspairs as test_addr_pair
patch = mock.patch
+HTTPCON = 'neutron.plugins.bigswitch.servermanager.HTTPConnection'
class BigSwitchProxyPluginV2TestCase(test_base.BigSwitchTestBase,
super(BigSwitchProxyPluginV2TestCase,
self).setUp(self._plugin_name)
self.port_create_status = 'BUILD'
+ self.startHttpPatch()
class TestBigSwitchProxyBasicGet(test_plugin.TestBasicGet,
def test_rollback_for_port_create(self):
plugin = NeutronManager.get_plugin()
with self.subnet() as s:
- self.httpPatch = patch('httplib.HTTPConnection', create=True,
- new=fake_server.HTTPConnectionMock500)
- self.httpPatch.start()
- kwargs = {'device_id': 'somedevid'}
- # allow thread spawns for this patch
+ # stop normal patch
+ self.httpPatch.stop()
+ # allow thread spawns for this test
self.spawn_p.stop()
+ kwargs = {'device_id': 'somedevid'}
+ # put in a broken 'server'
+ httpPatch = patch(HTTPCON, new=fake_server.HTTPConnectionMock500)
+ httpPatch.start()
with self.port(subnet=s, **kwargs):
- self.spawn_p.start()
+ # wait for async port create request to finish
plugin.evpool.waitall()
- self.httpPatch.stop()
+ # put good 'server' back in
+ httpPatch.stop()
+ self.httpPatch.start()
ports = self._get_ports(s['subnet']['network_id'])
#failure to create should result in port in error state
self.assertEqual(ports[0]['status'], 'ERROR')
device_id='66') as port:
port = self._get_ports(n['network']['id'])[0]
data = {'port': {'name': 'aNewName', 'device_id': '99'}}
- self.httpPatch = patch('httplib.HTTPConnection', create=True,
- new=fake_server.HTTPConnectionMock500)
- self.httpPatch.start()
- self.new_update_request('ports',
- data,
- port['id']).get_response(self.api)
+ # stop normal patch
self.httpPatch.stop()
+ with patch(HTTPCON, new=fake_server.HTTPConnectionMock500):
+ self.new_update_request(
+ 'ports', data, port['id']).get_response(self.api)
+ self.httpPatch.start()
uport = self._get_ports(n['network']['id'])[0]
# name should have stayed the same
self.assertEqual(port['name'], uport['name'])
with self.network() as n:
with self.port(network_id=n['network']['id'],
device_id='somedevid') as port:
- self.httpPatch = patch('httplib.HTTPConnection', create=True,
- new=fake_server.HTTPConnectionMock500)
- self.httpPatch.start()
- self._delete('ports', port['port']['id'],
- expected_code=
- webob.exc.HTTPInternalServerError.code)
+ # stop normal patch
self.httpPatch.stop()
+ with patch(HTTPCON, new=fake_server.HTTPConnectionMock500):
+ self._delete('ports', port['port']['id'],
+ expected_code=
+ webob.exc.HTTPInternalServerError.code)
+ self.httpPatch.start()
port = self._get_ports(n['network']['id'])[0]
self.assertEqual('BUILD', port['status'])
self.spawn_p.stop()
with nested(
self.subnet(),
- patch('httplib.HTTPConnection', create=True,
+ patch(HTTPCON, create=True,
new=fake_server.HTTPConnectionMock404),
patch(test_base.RESTPROXY_PKG_PATH
+ '.NeutronRestProxyV2._send_all_data')
def test_rollback_on_network_create(self):
tid = test_api_v2._uuid()
kwargs = {'tenant_id': tid}
- self.httpPatch = patch('httplib.HTTPConnection', create=True,
- new=fake_server.HTTPConnectionMock500)
- self.httpPatch.start()
- self._create_network('json', 'netname', True, **kwargs)
self.httpPatch.stop()
+ with patch(HTTPCON, new=fake_server.HTTPConnectionMock500):
+ self._create_network('json', 'netname', True, **kwargs)
+ self.httpPatch.start()
self.assertFalse(self._get_networks(tid))
def test_rollback_on_network_update(self):
with self.network() as n:
data = {'network': {'name': 'aNewName'}}
- self.httpPatch = patch('httplib.HTTPConnection', create=True,
- new=fake_server.HTTPConnectionMock500)
- self.httpPatch.start()
- self.new_update_request('networks', data,
- n['network']['id']).get_response(self.api)
self.httpPatch.stop()
+ with patch(HTTPCON, new=fake_server.HTTPConnectionMock500):
+ self.new_update_request(
+ 'networks', data, n['network']['id']
+ ).get_response(self.api)
+ self.httpPatch.start()
updatedn = self._get_networks(n['network']['tenant_id'])[0]
# name should have stayed the same due to failure
self.assertEqual(n['network']['name'], updatedn['name'])
def test_rollback_on_network_delete(self):
with self.network() as n:
- self.httpPatch = patch('httplib.HTTPConnection', create=True,
- new=fake_server.HTTPConnectionMock500)
- self.httpPatch.start()
- self._delete('networks', n['network']['id'],
- expected_code=webob.exc.HTTPInternalServerError.code)
self.httpPatch.stop()
+ with patch(HTTPCON, new=fake_server.HTTPConnectionMock500):
+ self._delete(
+ 'networks', n['network']['id'],
+ expected_code=webob.exc.HTTPInternalServerError.code)
+ self.httpPatch.start()
# network should still exist in db
self.assertEqual(n['network']['id'],
self._get_networks(n['network']['tenant_id']
from neutron.tests.unit import test_l3_plugin
+HTTPCON = 'neutron.plugins.bigswitch.servermanager.httplib.HTTPConnection'
_uuid = uuidutils.generate_uuid
self.setup_config_files()
super(test_extradhcp.ExtraDhcpOptDBTestCase,
self).setUp(plugin=self._plugin_name)
+ self.startHttpPatch()
-class RouterDBTestCase(test_base.BigSwitchTestBase,
- test_l3_plugin.L3NatDBIntTestCase):
+class RouterDBTestBase(test_base.BigSwitchTestBase,
+ test_l3_plugin.L3BaseForIntTests,
+ test_l3_plugin.L3NatTestCaseMixin):
def setUp(self):
self.setup_patches()
self.setup_config_files()
ext_mgr = RouterRulesTestExtensionManager()
- super(RouterDBTestCase, self).setUp(plugin=self._plugin_name,
+ super(RouterDBTestBase, self).setUp(plugin=self._plugin_name,
ext_mgr=ext_mgr)
cfg.CONF.set_default('allow_overlapping_ips', False)
self.plugin_obj = NeutronManager.get_plugin()
+ self.startHttpPatch()
def tearDown(self):
- super(RouterDBTestCase, self).tearDown()
+ super(RouterDBTestBase, self).tearDown()
del test_config['config_files']
+
+class RouterDBTestCase(RouterDBTestBase,
+ test_l3_plugin.L3NatDBIntTestCase):
+
def test_router_remove_router_interface_wrong_subnet_returns_400(self):
with self.router() as r:
with self.subnet() as s:
port_id=p1['port']['id'],
tenant_id=tenant1_id)
multiFloatPatch = patch(
- 'httplib.HTTPConnection',
- create=True,
+ HTTPCON,
new=fake_server.VerifyMultiTenantFloatingIP)
multiFloatPatch.start()
fl2 = self._make_floatingip_for_tenant_port(
def test_rollback_on_router_create(self):
tid = test_api_v2._uuid()
- self.errhttpPatch = patch('httplib.HTTPConnection', create=True,
- new=fake_server.HTTPConnectionMock500)
- self.errhttpPatch.start()
- self._create_router('json', tid)
- self.errhttpPatch.stop()
+ self.httpPatch.stop()
+ with patch(HTTPCON, new=fake_server.HTTPConnectionMock500):
+ self._create_router('json', tid)
self.assertTrue(len(self._get_routers(tid)) == 0)
def test_rollback_on_router_update(self):
with self.router() as r:
data = {'router': {'name': 'aNewName'}}
- self.errhttpPatch = patch('httplib.HTTPConnection', create=True,
- new=fake_server.HTTPConnectionMock500)
- self.errhttpPatch.start()
- self.new_update_request('routers', data,
- r['router']['id']).get_response(self.api)
- self.errhttpPatch.stop()
+ self.httpPatch.stop()
+ with patch(HTTPCON, new=fake_server.HTTPConnectionMock500):
+ self.new_update_request(
+ 'routers', data, r['router']['id']).get_response(self.api)
+ self.httpPatch.start()
updatedr = self._get_routers(r['router']['tenant_id'])[0]
# name should have stayed the same due to failure
self.assertEqual(r['router']['name'], updatedr['name'])
def test_rollback_on_router_delete(self):
with self.router() as r:
- self.errhttpPatch = patch('httplib.HTTPConnection', create=True,
- new=fake_server.HTTPConnectionMock500)
- self.errhttpPatch.start()
- self._delete('routers', r['router']['id'],
- expected_code=exc.HTTPInternalServerError.code)
- self.errhttpPatch.stop()
+ self.httpPatch.stop()
+ with patch(HTTPCON, new=fake_server.HTTPConnectionMock500):
+ self._delete('routers', r['router']['id'],
+ expected_code=exc.HTTPInternalServerError.code)
+ self.httpPatch.start()
self.assertEqual(r['router']['id'],
self._get_routers(r['router']['tenant_id']
)[0]['id'])
plugin = manager.NeutronManager.get_plugin()
self.notifier = plugin.notifier
self.rpc = plugin.callbacks
+ self.startHttpPatch()
class TestSecServerRpcCallBack(test_sg_rpc.SGServerRpcCallBackMixinTestCase,
#
# @author: Kevin Benton, kevin.benton@bigswitch.com
#
+from contextlib import nested
import httplib
import socket
+import ssl
-from contextlib import nested
import mock
from oslo.config import cfg
from neutron.manager import NeutronManager
+from neutron.openstack.common import importutils
from neutron.plugins.bigswitch import servermanager
from neutron.tests.unit.bigswitch import test_restproxy_plugin as test_rp
-HTTPCON = 'httplib.HTTPConnection'
SERVERMANAGER = 'neutron.plugins.bigswitch.servermanager'
+HTTPCON = SERVERMANAGER + '.HTTPConnection'
+HTTPSCON = SERVERMANAGER + '.HTTPSConnectionWithValidation'
class ServerManagerTests(test_rp.BigSwitchProxyPluginV2TestCase):
+ def setUp(self):
+ self.socket_mock = mock.patch(
+ SERVERMANAGER + '.socket.create_connection').start()
+ self.wrap_mock = mock.patch(SERVERMANAGER + '.ssl.wrap_socket').start()
+ super(ServerManagerTests, self).setUp()
+ # http patch must not be running or it will mangle the servermanager
+ # import where the https connection classes are defined
+ self.httpPatch.stop()
+ self.sm = importutils.import_module(SERVERMANAGER)
+
def test_no_servers(self):
cfg.CONF.set_override('servers', [], 'RESTPROXY')
self.assertRaises(cfg.Error, servermanager.ServerPool)
def test_malformed_servers(self):
- cfg.CONF.set_override('servers', ['a:b:c'], 'RESTPROXY')
+ cfg.CONF.set_override('servers', ['1.2.3.4', '1.1.1.1:a'], 'RESTPROXY')
self.assertRaises(cfg.Error, servermanager.ServerPool)
+ def test_ipv6_server_address(self):
+ cfg.CONF.set_override(
+ 'servers', ['[ABCD:EF01:2345:6789:ABCD:EF01:2345:6789]:80'],
+ 'RESTPROXY')
+ s = servermanager.ServerPool()
+ self.assertEqual(s.servers[0].server,
+ '[ABCD:EF01:2345:6789:ABCD:EF01:2345:6789]')
+
def test_sticky_cert_fetch_fail(self):
pl = NeutronManager.get_plugin()
pl.servers.ssl = True
self.assertRaises(servermanager.RemoteRestError,
pl.servers._consistency_watchdog)
+ def test_consistency_hash_header(self):
+ # mock HTTP class instead of rest_call so we can see headers
+ with mock.patch(HTTPCON) as conmock:
+ rv = conmock.return_value
+ rv.getresponse.return_value.getheader.return_value = 'HASHHEADER'
+ with self.network():
+ callheaders = rv.request.mock_calls[0][1][3]
+ self.assertIn('X-BSN-BVS-HASH-MATCH', callheaders)
+ # first call will be False to indicate no previous state hash
+ self.assertEqual(callheaders['X-BSN-BVS-HASH-MATCH'], False)
+ # change the header that will be received on delete call
+ rv.getresponse.return_value.getheader.return_value = 'HASH2'
+
+ # net delete should have used header received on create
+ callheaders = rv.request.mock_calls[1][1][3]
+ self.assertEqual(callheaders['X-BSN-BVS-HASH-MATCH'], 'HASHHEADER')
+
+ # create again should now use header received from prev delete
+ with self.network():
+ callheaders = rv.request.mock_calls[2][1][3]
+ self.assertIn('X-BSN-BVS-HASH-MATCH', callheaders)
+ self.assertEqual(callheaders['X-BSN-BVS-HASH-MATCH'],
+ 'HASH2')
+
def test_file_put_contents(self):
pl = NeutronManager.get_plugin()
with mock.patch(SERVERMANAGER + '.open', create=True) as omock:
mock.call.write('certdata')
])
+ def test_auth_header(self):
+ cfg.CONF.set_override('server_auth', 'username:pass', 'RESTPROXY')
+ sp = servermanager.ServerPool()
+ with mock.patch(HTTPCON) as conmock:
+ rv = conmock.return_value
+ rv.getresponse.return_value.getheader.return_value = 'HASHHEADER'
+ sp.rest_create_network('tenant', 'network')
+ callheaders = rv.request.mock_calls[0][1][3]
+ self.assertIn('Authorization', callheaders)
+ self.assertEqual(callheaders['Authorization'],
+ 'Basic dXNlcm5hbWU6cGFzcw==')
+
+ def test_header_add(self):
+ sp = servermanager.ServerPool()
+ with mock.patch(HTTPCON) as conmock:
+ rv = conmock.return_value
+ rv.getresponse.return_value.getheader.return_value = 'HASHHEADER'
+ sp.servers[0].rest_call('GET', '/', headers={'EXTRA-HEADER': 'HI'})
+ callheaders = rv.request.mock_calls[0][1][3]
+ # verify normal headers weren't mangled
+ self.assertIn('Content-type', callheaders)
+ self.assertEqual(callheaders['Content-type'],
+ 'application/json')
+ # verify new header made it in
+ self.assertIn('EXTRA-HEADER', callheaders)
+ self.assertEqual(callheaders['EXTRA-HEADER'], 'HI')
+
+ def test_reconnect_on_timeout_change(self):
+ sp = servermanager.ServerPool()
+ with mock.patch(HTTPCON) as conmock:
+ rv = conmock.return_value
+ rv.getresponse.return_value.getheader.return_value = 'HASHHEADER'
+ sp.servers[0].capabilities = ['keep-alive']
+ sp.servers[0].rest_call('GET', '/', timeout=10)
+ # even with keep-alive enabled, a change in timeout will trigger
+ # a reconnect
+ sp.servers[0].rest_call('GET', '/', timeout=75)
+ conmock.assert_has_calls([
+ mock.call('localhost', 9000, timeout=10),
+ mock.call('localhost', 9000, timeout=75),
+ ], any_order=True)
+
+ def test_connect_failures(self):
+ sp = servermanager.ServerPool()
+ with mock.patch(HTTPCON, return_value=None):
+ resp = sp.servers[0].rest_call('GET', '/')
+ self.assertEqual(resp, (0, None, None, None))
+ # verify same behavior on ssl class
+ sp.servers[0].currentcon = False
+ sp.servers[0].ssl = True
+ with mock.patch(HTTPSCON, return_value=None):
+ resp = sp.servers[0].rest_call('GET', '/')
+ self.assertEqual(resp, (0, None, None, None))
+
def test_reconnect_cached_connection(self):
sp = servermanager.ServerPool()
with mock.patch(HTTPCON) as conmock:
conmock.return_value.request.side_effect = socket.timeout()
resp = sp.servers[0].rest_call('GET', '/')
self.assertEqual(resp, (0, None, None, None))
+
+ def test_cert_get_fail(self):
+ pl = NeutronManager.get_plugin()
+ pl.servers.ssl = True
+ with mock.patch('os.path.exists', return_value=False):
+ self.assertRaises(cfg.Error,
+ pl.servers._get_combined_cert_for_server,
+ *('example.org', 443))
+
+ def test_cert_make_dirs(self):
+ pl = NeutronManager.get_plugin()
+ pl.servers.ssl = True
+ cfg.CONF.set_override('ssl_sticky', False, 'RESTPROXY')
+ # pretend base dir exists, 3 children don't, and host cert does
+ with nested(
+ mock.patch('os.path.exists', side_effect=[True, False, False,
+ False, True]),
+ mock.patch('os.makedirs'),
+ mock.patch(SERVERMANAGER + '.ServerPool._combine_certs_to_file')
+ ) as (exmock, makemock, combmock):
+ # will raise error because no certs found
+ self.assertIn(
+ 'example.org',
+ pl.servers._get_combined_cert_for_server('example.org', 443)
+ )
+ base = cfg.CONF.RESTPROXY.ssl_cert_directory
+ hpath = base + '/host_certs/example.org.pem'
+ combpath = base + '/combined/example.org.pem'
+ combmock.assert_has_calls([mock.call([hpath], combpath)])
+ self.assertEqual(exmock.call_count, 5)
+ self.assertEqual(makemock.call_count, 3)
+
+ def test_no_cert_error(self):
+ pl = NeutronManager.get_plugin()
+ pl.servers.ssl = True
+ cfg.CONF.set_override('ssl_sticky', False, 'RESTPROXY')
+ # pretend base dir exists and 3 children do, but host cert doesn't
+ with mock.patch(
+ 'os.path.exists',
+ side_effect=[True, True, True, True, False]
+ ) as exmock:
+ # will raise error because no certs found
+ self.assertRaises(
+ cfg.Error,
+ pl.servers._get_combined_cert_for_server,
+ *('example.org', 443)
+ )
+ self.assertEqual(exmock.call_count, 5)
+
+ def test_action_success(self):
+ pl = NeutronManager.get_plugin()
+ self.assertTrue(pl.servers.action_success((200,)))
+
+ def test_server_failure(self):
+ pl = NeutronManager.get_plugin()
+ self.assertTrue(pl.servers.server_failure((404,)))
+ # server failure has an ignore codes option
+ self.assertFalse(pl.servers.server_failure((404,),
+ ignore_codes=[404]))
+
+ def test_conflict_triggers_sync(self):
+ pl = NeutronManager.get_plugin()
+ with mock.patch(
+ SERVERMANAGER + '.ServerProxy.rest_call',
+ return_value=(httplib.CONFLICT, 0, 0, 0)
+ ) as srestmock:
+ # making a call should trigger a conflict sync
+ pl.servers.rest_call('GET', '/', '', None, [])
+ srestmock.assert_has_calls([
+ mock.call('GET', '/', '', None, False, reconnect=True),
+ mock.call('PUT', '/topology',
+ {'routers': [], 'networks': []},
+ timeout=None)
+ ])
+
+ def test_conflict_sync_raises_error_without_topology(self):
+ pl = NeutronManager.get_plugin()
+ pl.servers.get_topo_function = None
+ with mock.patch(
+ SERVERMANAGER + '.ServerProxy.rest_call',
+ return_value=(httplib.CONFLICT, 0, 0, 0)
+ ):
+ # making a call should trigger a conflict sync that will
+ # error without the topology function set
+ self.assertRaises(
+ cfg.Error,
+ pl.servers.rest_call,
+ *('GET', '/', '', None, [])
+ )
+
+ def test_floating_calls(self):
+ pl = NeutronManager.get_plugin()
+ with mock.patch(SERVERMANAGER + '.ServerPool.rest_action') as ramock:
+ pl.servers.rest_create_floatingip('tenant', {'id': 'somefloat'})
+ pl.servers.rest_update_floatingip('tenant', {'name': 'myfl'}, 'id')
+ pl.servers.rest_delete_floatingip('tenant', 'oldid')
+ ramock.assert_has_calls([
+ mock.call('PUT', '/tenants/tenant/floatingips/somefloat',
+ errstr=u'Unable to create floating IP: %s'),
+ mock.call('PUT', '/tenants/tenant/floatingips/id',
+ errstr=u'Unable to update floating IP: %s'),
+ mock.call('DELETE', '/tenants/tenant/floatingips/oldid',
+ errstr=u'Unable to delete floating IP: %s')
+ ])
+
+ def test_HTTPSConnectionWithValidation_without_cert(self):
+ con = self.sm.HTTPSConnectionWithValidation(
+ 'www.example.org', 443, timeout=90)
+ con.source_address = '127.0.0.1'
+ con.request("GET", "/")
+ self.socket_mock.assert_has_calls([mock.call(
+ ('www.example.org', 443), 90, '127.0.0.1'
+ )])
+ self.wrap_mock.assert_has_calls([mock.call(
+ self.socket_mock(), None, None, cert_reqs=ssl.CERT_NONE
+ )])
+ self.assertEqual(con.sock, self.wrap_mock())
+
+ def test_HTTPSConnectionWithValidation_with_cert(self):
+ con = self.sm.HTTPSConnectionWithValidation(
+ 'www.example.org', 443, timeout=90)
+ con.combined_cert = 'SOMECERTS.pem'
+ con.source_address = '127.0.0.1'
+ con.request("GET", "/")
+ self.socket_mock.assert_has_calls([mock.call(
+ ('www.example.org', 443), 90, '127.0.0.1'
+ )])
+ self.wrap_mock.assert_has_calls([mock.call(
+ self.socket_mock(), None, None, ca_certs='SOMECERTS.pem',
+ cert_reqs=ssl.CERT_REQUIRED
+ )])
+ self.assertEqual(con.sock, self.wrap_mock())
+
+ def test_HTTPSConnectionWithValidation_tunnel(self):
+ tunnel_mock = mock.patch.object(
+ self.sm.HTTPSConnectionWithValidation,
+ '_tunnel').start()
+ con = self.sm.HTTPSConnectionWithValidation(
+ 'www.example.org', 443, timeout=90)
+ con.source_address = '127.0.0.1'
+ if not hasattr(con, 'set_tunnel'):
+ # no tunnel support in py26
+ return
+ con.set_tunnel('myproxy.local', 3128)
+ con.request("GET", "/")
+ self.socket_mock.assert_has_calls([mock.call(
+ ('www.example.org', 443), 90, '127.0.0.1'
+ )])
+ self.wrap_mock.assert_has_calls([mock.call(
+ self.socket_mock(), None, None, cert_reqs=ssl.CERT_NONE
+ )])
+ # _tunnel() doesn't take any args
+ tunnel_mock.assert_has_calls([mock.call()])
+ self.assertEqual(con._tunnel_host, 'myproxy.local')
+ self.assertEqual(con._tunnel_port, 3128)
+ self.assertEqual(con.sock, self.wrap_mock())
#
# @author: Kevin Benton, kevin.benton@bigswitch.com
#
+from contextlib import nested
import os
import mock
FILEPUT = SERVERMANAGER + '.ServerPool._file_put_contents'
GETCACERTS = SERVERMANAGER + '.ServerPool._get_ca_cert_paths'
GETHOSTCERT = SERVERMANAGER + '.ServerPool._get_host_cert_path'
+SSLGETCERT = SERVERMANAGER + '.ssl.get_server_certificate'
FAKECERTGET = 'neutron.tests.unit.bigswitch.fake_server.get_cert_contents'
-SSLGETCERT = 'ssl.get_server_certificate'
class test_ssl_certificate_base(test_plugin.NeutronDbPluginV2TestCase,
self.setup_config_files()
cfg.CONF.set_override('server_ssl', True, 'RESTPROXY')
cfg.CONF.set_override('ssl_sticky', True, 'RESTPROXY')
- self.httpsPatch = mock.patch(HTTPS, create=True,
- new=fake_server.HTTPSHostValidation)
- self.httpsPatch.start()
self._setUp()
# Set fake HTTPS connection's expectation
self.fake_certget_m.return_value = self.host_cert_val
def test_sticky_cert(self):
# SSL connection should be successful and cert should be cached
- with self.network():
+ with nested(
+ mock.patch(HTTPS, new=fake_server.HTTPSHostValidation),
+ self.network()
+ ):
# CA certs should have been checked for
self.getcacerts_m.assert_has_calls([mock.call(self.ca_certs_path)])
# cert should have been fetched via SSL lib
self.setup_config_files()
cfg.CONF.set_override('server_ssl', True, 'RESTPROXY')
cfg.CONF.set_override('ssl_sticky', True, 'RESTPROXY')
- self.httpsPatch = mock.patch(HTTPS, create=True,
- new=fake_server.HTTPSHostValidation)
- self.httpsPatch.start()
self._setUp()
# Set fake HTTPS connection's expectation to something wrong
data = {}
data['network'] = {'tenant_id': tid, 'name': 'name',
'admin_state_up': True}
- req = self.new_create_request('networks', data, 'json')
- res = req.get_response(self.api)
+ with mock.patch(HTTPS, new=fake_server.HTTPSHostValidation):
+ req = self.new_create_request('networks', data, 'json')
+ res = req.get_response(self.api)
self.assertEqual(res.status_int,
webob.exc.HTTPInternalServerError.code)
self.hcertpath_p.assert_has_calls([
cfg.CONF.set_override('server_ssl', True, 'RESTPROXY')
cfg.CONF.set_override('ssl_sticky', False, 'RESTPROXY')
cfg.CONF.set_override('no_ssl_validation', True, 'RESTPROXY')
- self.httpsPatch = mock.patch(HTTPS, create=True,
- new=fake_server.HTTPSNoValidation)
- self.httpsPatch.start()
self._setUp()
super(TestSslNoValidation, self).setUp()
def test_validation_disabled(self):
# SSL connection should be successful without any certificates
# If not, attempting to create a network will raise an exception
- with self.network():
+ with nested(
+ mock.patch(HTTPS, new=fake_server.HTTPSNoValidation),
+ self.network()
+ ):
# no sticky grabbing and no cert combining with no enforcement
self.assertFalse(self.sslgetcert_m.call_count)
self.assertFalse(self.certcomb_m.call_count)
Code Review / openstack-build / neutron-build.git / commitdiff