openstack
/
networking-l2gw
Code Issues Proposed changes

Merge "Python 3: add support"

Browse Source
changes/21/348121/1
Jenkins 7 years ago committed by Gerrit Code Review
parent 9ffe362082 327f5d3af5
commit a97dbb8066
11 changed files with 1046 additions and 1283 deletions
Show Stats Download Patch File Download Diff File

93
networking_l2gw/tests/unit/services/l2gateway/agent/ovsdb/test_base_connection.py
Unescape View File

@ -15,7 +15,6 @@
import eventlet
import contextlib
import os.path
import socket
import ssl
@ -104,10 +103,9 @@ class TestBaseConnection(base.BaseTestCase):
"""Test case to test __init__."""
fakesocket = SocketClass()
with contextlib.nested(
mock.patch.object(base_connection.LOG, 'debug'),
mock.patch.object(socket, 'socket', return_value=fakesocket)
) as(logger_call, mock_sock):
with mock.patch.object(base_connection.LOG, 'debug') as logger_call, \
mock.patch.object(socket, 'socket',
return_value=fakesocket):
self.l2gw_ovsdb.__init__(mock.Mock(), self.conf)
self.assertTrue(self.l2gw_ovsdb.connected)
self.assertTrue(logger_call.called)
@ -117,13 +115,10 @@ class TestBaseConnection(base.BaseTestCase):
"""Test case to test __init__ with socket error exception."""
fakesocket = SocketClass(socket.error)
with contextlib.nested(
mock.patch.object(base_connection.LOG, 'exception'),
mock.patch.object(base_connection.LOG, 'warning'),
mock.patch.object(socket, 'socket', return_value=fakesocket),
mock.patch.object(time, 'sleep')
) as(logger_exc, logger_warn,
sock_connect, mock_sleep):
with mock.patch.object(base_connection.LOG, 'exception') as logger_exc, \
mock.patch.object(base_connection.LOG, 'warning') as logger_warn, \
mock.patch.object(socket, 'socket', return_value=fakesocket) as sock_connect, \
mock.patch.object(time, 'sleep'):
ovsdb_conf = FakeConf()
self.assertRaises(socket.error, base_connection.BaseConnection,
cfg.CONF.ovsdb, ovsdb_conf)
@ -135,13 +130,10 @@ class TestBaseConnection(base.BaseTestCase):
"""Test case to test __init__ with socket timeout exception."""
fakesocket = SocketClass(socket.timeout)
with contextlib.nested(
mock.patch.object(base_connection.LOG, 'exception'),
mock.patch.object(base_connection.LOG, 'warning'),
mock.patch.object(socket, 'socket', return_value=fakesocket),
mock.patch.object(time, 'sleep')
) as(logger_exc, logger_warn,
sock_connect, mock_sleep):
with mock.patch.object(base_connection.LOG, 'exception') as logger_exc, \
mock.patch.object(base_connection.LOG, 'warning') as logger_warn, \
mock.patch.object(socket, 'socket', return_value=fakesocket) as sock_connect, \
mock.patch.object(time, 'sleep'):
ovsdb_conf = FakeConf()
self.assertRaises(socket.timeout, base_connection.BaseConnection,
cfg.CONF.ovsdb, ovsdb_conf)
@ -225,12 +217,11 @@ class TestBaseConnection_with_enable_manager(base.BaseTestCase):
self.l2gw_ovsdb_conn.ovsdb_fd_states = {fake_ip: 'fake_status'}
self.l2gw_ovsdb_conn.mgr.l2gw_agent_type = n_const.MONITOR
self.l2gw_ovsdb_conn.ovsdb_conn_list = [fake_ip]
with contextlib.nested(
mock.patch.object(eventlet.greenthread, 'spawn_n',
side_effect=Exception),
mock.patch.object(base_connection.LOG, 'warning'),
mock.patch.object(self.l2gw_ovsdb_conn, 'disconnect')) as (
mock_thread, mock_warning, mock_disconnect):
with mock.patch.object(eventlet.greenthread, 'spawn_n',
side_effect=Exception) as mock_thread, \
mock.patch.object(base_connection.LOG, 'warning') as mock_warning, \
mock.patch.object(self.l2gw_ovsdb_conn,
'disconnect') as mock_disconnect:
self.l2gw_ovsdb_conn._send_monitor_msg_to_ovsdb_connection(
fake_ip)
self.assertTrue(mock_thread.called)
@ -242,18 +233,13 @@ class TestBaseConnection_with_enable_manager(base.BaseTestCase):
"params": "fake_params",
"id": "fake_id",
}
with contextlib.nested(
mock.patch.object(eventlet.greenthread, 'sleep'),
mock.patch.object(jsonutils, 'loads', return_value=fake_resp),
mock.patch.object(self.fakesocket,
'recv',
return_value=fake_resp),
mock.patch.object(self.fakesocket,
'send')
) as (
fake_thread, mock_loads,
mock_sock_rcv,
mock_sock_send):
with mock.patch.object(eventlet.greenthread, 'sleep') as fake_thread, \
mock.patch.object(jsonutils, 'loads', return_value=fake_resp) as mock_loads, \
mock.patch.object(self.fakesocket,
'recv',
return_value=fake_resp) as mock_sock_rcv, \
mock.patch.object(self.fakesocket,
'send') as mock_sock_send:
self.l2gw_ovsdb_conn._echo_response(self.fake_ip)
self.assertTrue(fake_thread.called)
self.assertTrue(mock_sock_rcv.called)
@ -262,16 +248,13 @@ class TestBaseConnection_with_enable_manager(base.BaseTestCase):
self.assertTrue(mock_sock_send.called)
def test_common_sock_rcv_thread_none(self):
with contextlib.nested(
mock.patch.object(base_connection.BaseConnection,
'_echo_response'),
mock.patch.object(eventlet.greenthread, 'sleep'),
mock.patch.object(self.fakesocket,
'recv', return_value=None),
mock.patch.object(base_connection.BaseConnection,
'disconnect')) as (
mock_resp, green_thrd_sleep,
mock_rcv, mock_disconnect):
with mock.patch.object(base_connection.BaseConnection,
'_echo_response') as mock_resp, \
mock.patch.object(eventlet.greenthread, 'sleep') as green_thrd_sleep, \
mock.patch.object(self.fakesocket,
'recv', return_value=None) as mock_rcv, \
mock.patch.object(base_connection.BaseConnection,
'disconnect') as mock_disconnect:
self.l2gw_ovsdb_conn.check_c_sock = True
self.l2gw_ovsdb_conn.read_on = True
self.l2gw_ovsdb_conn._common_sock_rcv_thread(self.fake_ip)
@ -312,11 +295,9 @@ class TestBaseConnection_with_enable_manager(base.BaseTestCase):
cfg.CONF.set_override('l2_gw_agent_ca_cert_base_path',
'/home',
'ovsdb')
with contextlib.nested(
mock.patch.object(os.path, 'isfile', return_value=True),
mock.patch.object(base_connection.LOG, 'error'),
mock.patch.object(ssl, 'wrap_socket')
) as (mock_isfile, mock_error, mock_wrap_sock):
with mock.patch.object(os.path, 'isfile', return_value=True) as mock_isfile, \
mock.patch.object(base_connection.LOG, 'error') as mock_error, \
mock.patch.object(ssl, 'wrap_socket') as mock_wrap_sock:
self.l2gw_ovsdb_conn._is_ssl_configured('10.10.10.10',
self.mock_sock)
self.assertTrue(mock_isfile.called)
@ -341,11 +322,9 @@ class TestBaseConnection_with_enable_manager(base.BaseTestCase):
cfg.CONF.set_override('l2_gw_agent_ca_cert_base_path',
'/home/',
'ovsdb')
with contextlib.nested(
mock.patch.object(os.path, 'isfile', return_value=False),
mock.patch.object(base_connection.LOG, 'error'),
mock.patch.object(ssl, 'wrap_socket')
) as (mock_isfile, mock_error, mock_wrap_sock):
with mock.patch.object(os.path, 'isfile', return_value=False) as mock_isfile, \
mock.patch.object(base_connection.LOG, 'error') as mock_error, \
mock.patch.object(ssl, 'wrap_socket') as mock_wrap_sock:
self.l2gw_ovsdb_conn._is_ssl_configured('10.10.10.10',
self.mock_sock)
self.assertTrue(mock_isfile.called)

194
networking_l2gw/tests/unit/services/l2gateway/agent/ovsdb/test_manager.py
Unescape View File

@ -13,7 +13,6 @@
# License for the specific language governing permissions and limitations
# under the License.
import contextlib
import socket
import eventlet
@ -117,16 +116,14 @@ class TestManager(base.BaseTestCase):
gateway = l2gateway_config.L2GatewayConfig(self.fake_config_json)
ovsdb_ident = self.fake_config_json.get(n_const.OVSDB_IDENTIFIER)
self.l2gw_agent_manager.gateways[ovsdb_ident] = gateway
with contextlib.nested(
mock.patch.object(ovsdb_monitor,
'OVSDBMonitor'),
mock.patch.object(eventlet.greenthread,
'spawn_n'),
mock.patch.object(manager.OVSDBManager,
'agent_to_plugin_rpc'),
mock.patch.object(self.plugin_rpc,
'notify_ovsdb_states')
) as (ovsdb_connection, event_spawn, call_back, notify):
with mock.patch.object(ovsdb_monitor,
'OVSDBMonitor') as ovsdb_connection, \
mock.patch.object(eventlet.greenthread,
'spawn_n') as event_spawn, \
mock.patch.object(manager.OVSDBManager,
'agent_to_plugin_rpc') as call_back, \
mock.patch.object(self.plugin_rpc,
'notify_ovsdb_states') as notify:
self.l2gw_agent_manager._connect_to_ovsdb_server()
self.assertTrue(event_spawn.called)
self.assertTrue(ovsdb_connection.called)
@ -140,33 +137,28 @@ class TestManager(base.BaseTestCase):
gateway = l2gateway_config.L2GatewayConfig(self.fake_config_json)
ovsdb_ident = self.fake_config_json.get(n_const.OVSDB_IDENTIFIER)
self.l2gw_agent_manager.gateways[ovsdb_ident] = gateway
with contextlib.nested(
mock.patch.object(ovsdb_monitor.OVSDBMonitor,
'__init__',
side_effect=socket.error
),
mock.patch.object(eventlet.greenthread,
'spawn_n'),
mock.patch.object(manager.LOG, 'error')
) as (ovsdb_connection, event_spawn, mock_warn):
self.l2gw_agent_manager._connect_to_ovsdb_server()
event_spawn.assert_not_called()
with mock.patch.object(ovsdb_monitor.OVSDBMonitor,
'__init__',
side_effect=socket.error
), \
mock.patch.object(eventlet.greenthread,
'spawn_n') as event_spawn, \
mock.patch.object(manager.LOG, 'error'):
self.l2gw_agent_manager._connect_to_ovsdb_server()
event_spawn.assert_not_called()
def test_handle_report_state_failure(self):
self.l2gw_agent_manager.l2gw_agent_type = n_const.MONITOR
with contextlib.nested(
mock.patch.object(self.l2gw_agent_manager,
'_disconnect_all_ovsdb_servers'
),
mock.patch.object(agent_rpc.PluginReportStateAPI,
'report_state',
side_effect=Exception),
mock.patch.object(base_agent_manager.LOG,
'exception'),
mock.patch.object(self.l2gw_agent_manager,
'_stop_looping_task')
) as (mock_disconnect_ovsdb_servers, mock_report_state,
mock_log, mock_stop_looping):
with mock.patch.object(self.l2gw_agent_manager,
'_disconnect_all_ovsdb_servers'
) as mock_disconnect_ovsdb_servers, \
mock.patch.object(agent_rpc.PluginReportStateAPI,
'report_state',
side_effect=Exception), \
mock.patch.object(base_agent_manager.LOG,
'exception'), \
mock.patch.object(self.l2gw_agent_manager,
'_stop_looping_task') as mock_stop_looping:
self.l2gw_agent_manager._report_state()
self.assertEqual(self.l2gw_agent_manager.l2gw_agent_type,
'')
@ -210,26 +202,23 @@ class TestManager(base.BaseTestCase):
self.l2gw_agent_manager.gateways = {}
gateway = l2gateway_config.L2GatewayConfig(self.fake_config_json)
self.l2gw_agent_manager.gateways['fake_ovsdb_identifier'] = gateway
with contextlib.nested(
mock.patch.object(manager.LOG, 'warning'),
mock.patch.object(socket.socket, 'connect'),
mock.patch.object(ovsdb_writer.OVSDBWriter,
'delete_logical_switch')
) as (logger_call, mock_connect, mock_del_ls):
mock_connect.side_effect = socket.error
self.l2gw_agent_manager.delete_network(self.context,
mock.Mock(),
mock.Mock())
self.assertEqual(1, logger_call.call_count)
self.assertFalse(mock_del_ls.called)
with mock.patch.object(manager.LOG, 'warning') as logger_call, \
mock.patch.object(socket.socket, 'connect') as mock_connect, \
mock.patch.object(ovsdb_writer.OVSDBWriter,
'delete_logical_switch') as mock_del_ls:
mock_connect.side_effect = socket.error
self.l2gw_agent_manager.delete_network(self.context,
mock.Mock(),
mock.Mock())
self.assertEqual(1, logger_call.call_count)
self.assertFalse(mock_del_ls.called)
def test_init_with_enable_manager(self):
cfg.CONF.set_override('enable_manager', True, 'ovsdb')
with contextlib.nested(
mock.patch.object(manager.OVSDBManager,
'_sock_open_connection'),
mock.patch.object(loopingcall, 'FixedIntervalLoopingCall')) as (
mock_sock_open_conn, mock_loop):
with mock.patch.object(manager.OVSDBManager,
'_sock_open_connection') as mock_sock_open_conn, \
mock.patch.object(loopingcall,
'FixedIntervalLoopingCall') as mock_loop:
self.l2gw_agent_manager.__init__()
self.assertTrue(mock_sock_open_conn.called)
self.assertTrue(mock_loop.called)
@ -259,14 +248,13 @@ class TestManager(base.BaseTestCase):
self.l2gw_agent_manager.conf.host = 'fake_host'
self.l2gw_agent_manager.__init__()
self.l2gw_agent_manager.l2gw_agent_type = n_const.MONITOR
with contextlib.nested(
mock.patch.object(ovsdb_common_class,
'OVSDB_commom_class'),
mock.patch.object(eventlet.greenthread,
'spawn_n'),
mock.patch.object(self.l2gw_agent_manager,
'_start_looping_task_ovsdb_states')) as (
mock_ovsdb_common, mock_thread, mock_looping):
with mock.patch.object(ovsdb_common_class,
'OVSDB_commom_class') as mock_ovsdb_common, \
mock.patch.object(eventlet.greenthread,
'spawn_n') as mock_thread, \
mock.patch.object(
self.l2gw_agent_manager,
'_start_looping_task_ovsdb_states') as mock_looping:
self.l2gw_agent_manager.ovsdb_fd = mock_ovsdb_common.return_value
self.l2gw_agent_manager.ovsdb_fd.check_monitor_table_thread = False
self.l2gw_agent_manager.ovsdb_fd.check_sock_rcv = True
@ -306,11 +294,9 @@ class TestManager(base.BaseTestCase):
self.l2gw_agent_manager.__init__()
self.l2gw_agent_manager.l2gw_agent_type = ''
fake_op_method = 'CREATE'
with contextlib.nested(
mock.patch.object(ovsdb_common_class, 'OVSDB_commom_class'),
mock.patch.object(manager.OVSDBManager,
'_sock_open_connection')) as (
mock_ovsdb_common, mock_open_conn):
with mock.patch.object(ovsdb_common_class, 'OVSDB_commom_class') as mock_ovsdb_common, \
mock.patch.object(manager.OVSDBManager,
'_sock_open_connection') as mock_open_conn:
self.l2gw_agent_manager.ovsdb_fd = mock_ovsdb_common.return_value
self.l2gw_agent_manager.ovsdb_fd.ovsdb_conn_list = ['fake_ip']
self.l2gw_agent_manager.update_connection_to_gateway(
@ -329,11 +315,10 @@ class TestManager(base.BaseTestCase):
cfg.CONF.set_override('enable_manager', False, 'ovsdb')
self.l2gw_agent_manager.__init__()
fake_op_method = 'CREATE'
with contextlib.nested(
mock.patch.object(self.l2gw_agent_manager,
'_is_valid_request', return_value=True),
mock.patch.object(ovsdb_writer, 'OVSDBWriter'
)) as (mock_valid_req, mock_ovsdb_fd):
with mock.patch.object(self.l2gw_agent_manager,
'_is_valid_request', return_value=True) as mock_valid_req, \
mock.patch.object(ovsdb_writer, 'OVSDBWriter'
) as mock_ovsdb_fd:
self.l2gw_agent_manager.update_connection_to_gateway(
self.context, 'fake_ovsdb_id', "fake_logical_switch_dict",
"fake_locator_dicts", "fake_mac_dicts", "fake_port_dicts",
@ -364,11 +349,10 @@ class TestManager(base.BaseTestCase):
"""Test case to test delete_network with enable_manager=False."""
cfg.CONF.set_override('enable_manager', False, 'ovsdb')
self.l2gw_agent_manager.__init__()
with contextlib.nested(
mock.patch.object(self.l2gw_agent_manager,
'_is_valid_request', return_value=True),
mock.patch.object(ovsdb_writer, 'OVSDBWriter'
)) as (mock_valid_req, mock_ovsdb_fd):
with mock.patch.object(self.l2gw_agent_manager,
'_is_valid_request', return_value=True) as mock_valid_req, \
mock.patch.object(ovsdb_writer, 'OVSDBWriter'
) as mock_ovsdb_fd:
self.l2gw_agent_manager.delete_network(
self.context, 'fake_ovsdb_id', "fake_logical_switch_uuid")
ovsdb_sock_fd = mock_ovsdb_fd.return_value
@ -381,11 +365,9 @@ class TestManager(base.BaseTestCase):
cfg.CONF.set_override('enable_manager', True, 'ovsdb')
self.l2gw_agent_manager.__init__()
self.l2gw_agent_manager.l2gw_agent_type = ''
with contextlib.nested(
mock.patch.object(ovsdb_common_class, 'OVSDB_commom_class'),
mock.patch.object(manager.OVSDBManager,
'_sock_open_connection')) as (
mock_ovsdb_common, mock_open_conn):
with mock.patch.object(ovsdb_common_class, 'OVSDB_commom_class') as mock_ovsdb_common, \
mock.patch.object(manager.OVSDBManager,
'_sock_open_connection') as mock_open_conn:
self.l2gw_agent_manager.ovsdb_fd = mock_ovsdb_common.return_value
self.l2gw_agent_manager.ovsdb_fd.ovsdb_conn_list = ['fake_ip']
self.l2gw_agent_manager.delete_network(
@ -415,11 +397,9 @@ class TestManager(base.BaseTestCase):
cfg.CONF.set_override('enable_manager', True, 'ovsdb')
self.l2gw_agent_manager.__init__()
self.l2gw_agent_manager.l2gw_agent_type = ''
with contextlib.nested(
mock.patch.object(ovsdb_common_class, 'OVSDB_commom_class'),
mock.patch.object(manager.OVSDBManager,
'_sock_open_connection')) as (
mock_ovsdb_common, mock_open_conn):
with mock.patch.object(ovsdb_common_class, 'OVSDB_commom_class') as mock_ovsdb_common, \
mock.patch.object(manager.OVSDBManager,
'_sock_open_connection') as mock_open_conn:
self.l2gw_agent_manager.ovsdb_fd = mock_ovsdb_common.return_value
self.l2gw_agent_manager.ovsdb_fd.check_c_sock = True
self.l2gw_agent_manager.ovsdb_fd.ovsdb_conn_list = ['fake_ip']
@ -439,11 +419,11 @@ class TestManager(base.BaseTestCase):
"""
cfg.CONF.set_override('enable_manager', False, 'ovsdb')
self.l2gw_agent_manager.__init__()
with contextlib.nested(
mock.patch.object(self.l2gw_agent_manager,
'_is_valid_request', return_value=True),
mock.patch.object(ovsdb_writer, 'OVSDBWriter'
)) as (mock_valid_req, mock_ovsdb_fd):
with mock.patch.object(self.l2gw_agent_manager,
'_is_valid_request',
return_value=True) as mock_valid_req, \
mock.patch.object(ovsdb_writer, 'OVSDBWriter'
) as mock_ovsdb_fd:
self.l2gw_agent_manager.add_vif_to_gateway(
self.context, 'fake_ovsdb_id', "fake_logical_switch_dict",
"fake_locator_dict", "fake_mac_dict")
@ -475,11 +455,9 @@ class TestManager(base.BaseTestCase):
cfg.CONF.set_override('enable_manager', True, 'ovsdb')
self.l2gw_agent_manager.__init__()
self.l2gw_agent_manager.l2gw_agent_type = ''
with contextlib.nested(
mock.patch.object(ovsdb_common_class, 'OVSDB_commom_class'),
mock.patch.object(manager.OVSDBManager,
'_sock_open_connection')) as (
mock_ovsdb_common, mock_open_conn):
with mock.patch.object(ovsdb_common_class, 'OVSDB_commom_class') as mock_ovsdb_common, \
mock.patch.object(manager.OVSDBManager,
'_sock_open_connection') as mock_open_conn:
self.l2gw_agent_manager.ovsdb_fd = mock_ovsdb_common.return_value
self.l2gw_agent_manager.ovsdb_fd.ovsdb_conn_list = ['fake_ip']
self.l2gw_agent_manager.delete_vif_from_gateway(
@ -497,11 +475,10 @@ class TestManager(base.BaseTestCase):
"""
cfg.CONF.set_override('enable_manager', False, 'ovsdb')
self.l2gw_agent_manager.__init__()
with contextlib.nested(
mock.patch.object(self.l2gw_agent_manager,
'_is_valid_request', return_value=True),
mock.patch.object(ovsdb_writer, 'OVSDBWriter'
)) as (mock_valid_req, mock_ovsdb_fd):
with mock.patch.object(self.l2gw_agent_manager,
'_is_valid_request', return_value=True) as mock_valid_req, \
mock.patch.object(ovsdb_writer, 'OVSDBWriter'
) as mock_ovsdb_fd:
self.l2gw_agent_manager.delete_vif_from_gateway(
self.context, 'fake_ovsdb_id', "fake_logical_switch_uuid",
"fake_mac")
@ -535,11 +512,9 @@ class TestManager(base.BaseTestCase):
cfg.CONF.set_override('enable_manager', True, 'ovsdb')
self.l2gw_agent_manager.__init__()
self.l2gw_agent_manager.l2gw_agent_type = ''
with contextlib.nested(
mock.patch.object(ovsdb_common_class, 'OVSDB_commom_class'),
mock.patch.object(manager.OVSDBManager,
'_sock_open_connection')) as (
mock_ovsdb_common, mock_open_conn):
with mock.patch.object(ovsdb_common_class, 'OVSDB_commom_class') as mock_ovsdb_common, \
mock.patch.object(manager.OVSDBManager,
'_sock_open_connection') as mock_open_conn:
self.l2gw_agent_manager.ovsdb_fd = mock_ovsdb_common.return_value
self.l2gw_agent_manager.ovsdb_fd.ovsdb_conn_list = ['fake_ip']
self.l2gw_agent_manager.update_vif_to_gateway(
@ -557,11 +532,10 @@ class TestManager(base.BaseTestCase):
"""
cfg.CONF.set_override('enable_manager', False, 'ovsdb')
self.l2gw_agent_manager.__init__()
with contextlib.nested(
mock.patch.object(self.l2gw_agent_manager,
'_is_valid_request', return_value=True),
mock.patch.object(ovsdb_writer, 'OVSDBWriter'
)) as (mock_valid_req, mock_ovsdb_fd):
with mock.patch.object(self.l2gw_agent_manager,
'_is_valid_request', return_value=True) as mock_valid_req, \
mock.patch.object(ovsdb_writer, 'OVSDBWriter'
) as mock_ovsdb_fd:
self.l2gw_agent_manager.update_vif_to_gateway(
self.context, 'fake_ovsdb_id', "fake_locator_dict",
"fake_mac_dict")

137
networking_l2gw/tests/unit/services/l2gateway/agent/ovsdb/test_ovsdb_monitor.py
Unescape View File

@ -13,7 +13,6 @@
# License for the specific language governing permissions and limitations
# under the License.
import contextlib
import socket
import ssl
@ -91,11 +90,10 @@ class TestOVSDBMonitor(base.BaseTestCase):
"""Test case to test __init__."""
fakesocket = base_test.SocketClass()
with contextlib.nested(
mock.patch.object(ovsdb_monitor.LOG, 'debug'),
mock.patch.object(eventlet.greenthread, 'spawn'),
mock.patch.object(socket, 'socket', return_value=fakesocket)
) as(logger_call, gt, mock_sock):
with mock.patch.object(ovsdb_monitor.LOG, 'debug'), \
mock.patch.object(eventlet.greenthread, 'spawn') as gt, \
mock.patch.object(socket, 'socket',
return_value=fakesocket):
self.l2gw_ovsdb.__init__(mock.Mock(), self.conf,
self.callback)
self.assertTrue(self.l2gw_ovsdb.connected)
@ -122,19 +120,16 @@ class TestOVSDBMonitor(base.BaseTestCase):
def test_set_monitor_response_handler(self):
"""Test case to test _set_monitor_response_handler with error_msg."""
self.l2gw_ovsdb.connected = True
with contextlib.nested(
mock.patch.object(ovsdb_monitor.OVSDBMonitor,
'_set_handler'),
mock.patch.object(ovsdb_monitor.OVSDBMonitor,
'send', return_value=True),
mock.patch.object(ovsdb_monitor.OVSDBMonitor,
'_process_response',
return_value=(mock.ANY, False)),
mock.patch.object(ovsdb_monitor.OVSDBMonitor,
'_process_monitor_msg'),
mock.patch.object(ovsdb_monitor.LOG, 'warning')) as (
set_handler, send, process_resp,
process_monitor_msg, logger_call):
with mock.patch.object(ovsdb_monitor.OVSDBMonitor,
'_set_handler') as set_handler, \
mock.patch.object(ovsdb_monitor.OVSDBMonitor,
'send', return_value=True) as send, \
mock.patch.object(ovsdb_monitor.OVSDBMonitor,
'_process_response',
return_value=(mock.ANY, False)) as process_resp, \
mock.patch.object(ovsdb_monitor.OVSDBMonitor,
'_process_monitor_msg') as process_monitor_msg, \
mock.patch.object(ovsdb_monitor.LOG, 'warning'):
self.l2gw_ovsdb.set_monitor_response_handler()
self.assertTrue(set_handler.called)
self.assertTrue(send.called)
@ -144,19 +139,17 @@ class TestOVSDBMonitor(base.BaseTestCase):
def test_set_monitor_response_handler_with_error_in_send(self):
"""Test case to test _set_monitor_response_handler."""
self.l2gw_ovsdb.connected = True
with contextlib.nested(
mock.patch.object(ovsdb_monitor.OVSDBMonitor,
'_set_handler'),
mock.patch.object(ovsdb_monitor.OVSDBMonitor,
'send', return_value=False),
mock.patch.object(ovsdb_monitor.OVSDBMonitor,
'_process_response',
return_value=(mock.ANY, True)),
mock.patch.object(ovsdb_monitor.OVSDBMonitor,
'_process_monitor_msg'),
mock.patch.object(ovsdb_monitor.LOG, 'warning')) as (
set_handler, send,
process_resp, process_monitor_msg, logger_call):
with mock.patch.object(ovsdb_monitor.OVSDBMonitor,
'_set_handler') as set_handler, \
mock.patch.object(ovsdb_monitor.OVSDBMonitor,
'send', return_value=False) as send, \
mock.patch.object(ovsdb_monitor.OVSDBMonitor,
'_process_response',
return_value=(mock.ANY, True)) as process_resp, \
mock.patch.object(ovsdb_monitor.OVSDBMonitor,
'_process_monitor_msg') as process_monitor_msg, \
mock.patch.object(ovsdb_monitor.LOG,
'warning'):
self.l2gw_ovsdb.set_monitor_response_handler()
self.assertTrue(set_handler.called)
self.assertTrue(send.called)
@ -173,42 +166,34 @@ class TestOVSDBMonitor(base.BaseTestCase):
def test_process_update_event(self):
"""Test case to test _process_update_event."""
with contextlib.nested(
mock.patch.object(ovsdb_monitor.OVSDBMonitor,
'_process_physical_port'),
mock.patch.object(ovsdb_monitor.OVSDBMonitor,
'_process_physical_switch'),
with mock.patch.object(ovsdb_monitor.OVSDBMonitor,
'_process_physical_port') as proc_phy_port, \
mock.patch.object(ovsdb_monitor.OVSDBMonitor,
'_process_logical_switch'),
'_process_physical_switch') as proc_phy_switch, \
mock.patch.object(ovsdb_monitor.OVSDBMonitor,
'_process_ucast_macs_local'),
'_process_logical_switch') as proc_logic_switch, \
mock.patch.object(ovsdb_monitor.OVSDBMonitor,
'_process_physical_locator'),
'_process_ucast_macs_local') as proc_ucast_mac, \
mock.patch.object(ovsdb_monitor.OVSDBMonitor,
'_process_ucast_macs_remote'),
'_process_physical_locator') as proc_phy_loc, \
mock.patch.object(ovsdb_monitor.OVSDBMonitor,
'_process_mcast_macs_local'),
'_process_ucast_macs_remote') as proc_ucast_mac_remote, \
mock.patch.object(ovsdb_monitor.OVSDBMonitor,
'_process_physical_locator_set')
) as (proc_phy_port,
proc_phy_switch,
proc_logic_switch,
proc_ucast_mac,
proc_phy_loc,
proc_ucast_mac_remote,
proc_mcast_mac_local,
proc_phys_loc_set):
self.l2gw_ovsdb._setup_dispatch_table()
self.l2gw_ovsdb._process_update_event(self.msg2, mock.ANY)
self.assertTrue(proc_phy_port.called)
self.assertTrue(proc_phy_switch.called)
self.assertTrue(proc_logic_switch.called)
self.assertTrue(proc_ucast_mac.called)
self.assertTrue(proc_phy_loc.called)
self.assertTrue(proc_ucast_mac_remote.called)
self.assertTrue(proc_mcast_mac_local.called)
self.assertTrue(proc_phys_loc_set.called)
self.assertTrue(self.callback.called)
'_process_mcast_macs_local') as proc_mcast_mac_local, \
mock.patch.object(
ovsdb_monitor.OVSDBMonitor,
'_process_physical_locator_set') as proc_phys_loc_set:
self.l2gw_ovsdb._setup_dispatch_table()
self.l2gw_ovsdb._process_update_event(self.msg2, mock.ANY)
self.assertTrue(proc_phy_port.called)
self.assertTrue(proc_phy_switch.called)
self.assertTrue(proc_logic_switch.called)
self.assertTrue(proc_ucast_mac.called)
self.assertTrue(proc_phy_loc.called)
self.assertTrue(proc_ucast_mac_remote.called)
self.assertTrue(proc_mcast_mac_local.called)
self.assertTrue(proc_phys_loc_set.called)
self.assertTrue(self.callback.called)
def test_process_response_raise_exception(self):
"""Test case to test _process_response with exception."""
@ -273,21 +258,19 @@ class TestOVSDBMonitor(base.BaseTestCase):
def test_rcv_thread_exception(self):
"""Test case to test _rcv_thread with exception."""
with contextlib.nested(
mock.patch.object(self.l2gw_ovsdb.socket, 'recv',
side_effect=Exception,
return_value=None),
mock.patch.object(self.l2gw_ovsdb.socket,
'close'),
mock.patch.object(ovsdb_monitor.LOG,
'exception')
) as (sock_recv, sock_close, logger_call):
self.l2gw_ovsdb._rcv_thread()
self.assertTrue(logger_call.called)
self.assertTrue(sock_recv.called)
self.assertFalse(self.l2gw_ovsdb.connected)
self.assertFalse(self.l2gw_ovsdb.read_on)
self.assertTrue(sock_close.called)
with mock.patch.object(self.l2gw_ovsdb.socket, 'recv',
side_effect=Exception,
return_value=None) as sock_recv, \
mock.patch.object(self.l2gw_ovsdb.socket,
'close') as sock_close, \
mock.patch.object(ovsdb_monitor.LOG,
'exception') as logger_call:
self.l2gw_ovsdb._rcv_thread()
self.assertTrue(logger_call.called)
self.assertTrue(sock_recv.called)
self.assertFalse(self.l2gw_ovsdb.connected)
self.assertFalse(self.l2gw_ovsdb.read_on)
self.assertTrue(sock_close.called)
def test_form_ovsdb_data(self):
some_value = mock.Mock()

310
networking_l2gw/tests/unit/services/l2gateway/agent/ovsdb/test_ovsdb_writer.py
Unescape View File

@ -13,7 +13,6 @@
# License for the specific language governing permissions and limitations
# under the License.
import contextlib
import random
import socket
import ssl
@ -112,17 +111,16 @@ class TestOVSDBWriter(base.BaseTestCase):
"""Test case to test _get_reply."""
ret_value = jsonutils.dumps({self.op_id:
'foo_value'})
with contextlib.nested(
mock.patch.object(ovsdb_writer.OVSDBWriter,
'_recv_data',
return_value=jsonutils.dumps({self.op_id:
'foo_value'})),
with mock.patch.object(
ovsdb_writer.OVSDBWriter,
'_recv_data',
return_value=jsonutils.dumps({self.op_id:
'foo_value'})) as recv_data, \
mock.patch.object(ovsdb_writer.OVSDBWriter,
'_process_response',
return_value=(ret_value, None)),
return_value=(ret_value, None)) as proc_response, \
mock.patch.object(ovsdb_writer.LOG,
'debug')
) as (recv_data, proc_response, debug):
'debug'):
self.l2gw_ovsdb._get_reply(self.op_id, mock.ANY)
self.assertTrue(recv_data.called)
self.assertTrue(proc_response.called)
@ -180,69 +178,53 @@ class TestOVSDBWriter(base.BaseTestCase):
def test_insert_ucast_macs_remote(self):
"""Test case to test insert ucast_macs_remote."""
with contextlib.nested(
mock.patch.object(ovsdb_writer.OVSDBWriter,
'_get_ucast_macs_remote_dict'),
mock.patch.object(random,
'getrandbits',
return_value=self.op_id
),
mock.patch.object(ovsdb_writer.OVSDBWriter,
'_send_and_receive'
),
mock.patch.object(ovsdb_writer.LOG,
'debug'),
mock.patch.object(ovsdb_schema, 'LogicalSwitch'),
mock.patch.object(ovsdb_schema, 'PhysicalLocator'),
mock.patch.object(ovsdb_schema, 'UcastMacsRemote'),
) as (get_ucast_mac_remote,
get_rand,
send_n_receive,
mock_log,
mock_ls,
mock_pl,
mock_ucmr):
self.l2gw_ovsdb.insert_ucast_macs_remote(mock.MagicMock(),
mock.MagicMock(),
mock.MagicMock(),
mock.ANY)
get_rand.assert_called_with(128)
send_n_receive.assert_called_with(mock.ANY,
self.op_id, mock.ANY, True)
with mock.patch.object(ovsdb_writer.OVSDBWriter,
'_get_ucast_macs_remote_dict') as get_ucast_mac_remote, \
mock.patch.object(random,
'getrandbits',
return_value=self.op_id
) as get_rand, \
mock.patch.object(ovsdb_writer.OVSDBWriter,
'_send_and_receive'
) as send_n_receive, \
mock.patch.object(ovsdb_writer.LOG,
'debug'), \
mock.patch.object(ovsdb_schema, 'LogicalSwitch'), \
mock.patch.object(ovsdb_schema, 'PhysicalLocator'), \
mock.patch.object(ovsdb_schema, 'UcastMacsRemote'):
self.l2gw_ovsdb.insert_ucast_macs_remote(mock.MagicMock(),
mock.MagicMock(),
mock.MagicMock(),
mock.ANY)
get_rand.assert_called_with(128)
send_n_receive.assert_called_with(mock.ANY,
self.op_id, mock.ANY, True)
self.assertTrue(get_ucast_mac_remote.called)
self.assertTrue(get_ucast_mac_remote.called)
def test_insert_ucast_macs_remote_with_no_locator_id(self):
"""Test case to test insert ucast_macs_remote
without locator_id and logical_switch_id.
"""
with contextlib.nested(
mock.patch.object(ovsdb_writer.OVSDBWriter,
'_get_ucast_macs_remote_dict'),
mock.patch.object(ovsdb_writer.OVSDBWriter,
'_get_physical_locator_dict'),
mock.patch.object(ovsdb_writer.OVSDBWriter,
'_get_logical_switch_dict'),
mock.patch.object(random,
'getrandbits',
return_value=self.op_id
),
mock.patch.object(ovsdb_writer.OVSDBWriter,
'_send_and_receive'
),
mock.patch.object(ovsdb_writer.LOG,
'debug'),
mock.patch.object(ovsdb_schema, 'LogicalSwitch'),
mock.patch.object(ovsdb_schema, 'PhysicalLocator'),
mock.patch.object(ovsdb_schema, 'UcastMacsRemote'),
) as (get_ucast_mac_remote,
get_physical_locator_dict,
get_logical_switch_dict,
get_rand,
send_n_receive,
mock_log,
mock_ls, mock_pl, mock_ucmr):
with mock.patch.object(ovsdb_writer.OVSDBWriter,
'_get_ucast_macs_remote_dict') as get_ucast_mac_remote, \
mock.patch.object(ovsdb_writer.OVSDBWriter,
'_get_physical_locator_dict') as get_physical_locator_dict, \
mock.patch.object(ovsdb_writer.OVSDBWriter,
'_get_logical_switch_dict') as get_logical_switch_dict, \
mock.patch.object(random,
'getrandbits',
return_value=self.op_id
), \
mock.patch.object(ovsdb_writer.OVSDBWriter,
'_send_and_receive'
), \
mock.patch.object(ovsdb_writer.LOG,
'debug'), \
mock.patch.object(ovsdb_schema, 'LogicalSwitch') as mock_ls, \
mock.patch.object(ovsdb_schema, 'PhysicalLocator') as mock_pl, \
mock.patch.object(ovsdb_schema, 'UcastMacsRemote'):
locator = mock_pl.return_value
locator.uuid = None
ls = mock_ls.return_value
@ -258,62 +240,48 @@ class TestOVSDBWriter(base.BaseTestCase):
def test_update_ucast_macs_remote(self):
"""Test case to test update ucast_macs_remote."""
with contextlib.nested(
mock.patch.object(ovsdb_writer.OVSDBWriter,
'_get_dict_for_update_ucast_mac_remote'),
mock.patch.object(random,
'getrandbits',
return_value=self.op_id
),
mock.patch.object(ovsdb_writer.OVSDBWriter,
'_send_and_receive'
),
mock.patch.object(ovsdb_writer.LOG,
'debug'),
mock.patch.object(ovsdb_schema, 'PhysicalLocator'),
mock.patch.object(ovsdb_schema, 'UcastMacsRemote'),
) as (get_update_ucast_mac_remote,
get_rand,
send_n_receive,
mock_log,
mock_pl,
mock_ucmr):
self.l2gw_ovsdb.update_ucast_macs_remote(mock.MagicMock(),
mock.MagicMock(),
mock.ANY)
get_rand.assert_called_with(128)
send_n_receive.assert_called_with(mock.ANY,
self.op_id, mock.ANY, True)
with mock.patch.object(ovsdb_writer.OVSDBWriter,
'_get_dict_for_update_ucast_mac_remote') as get_update_ucast_mac_remote, \
mock.patch.object(random,
'getrandbits',
return_value=self.op_id
) as get_rand, \
mock.patch.object(ovsdb_writer.OVSDBWriter,
'_send_and_receive'
) as send_n_receive, \
mock.patch.object(ovsdb_writer.LOG,
'debug'), \
mock.patch.object(ovsdb_schema, 'PhysicalLocator'), \
mock.patch.object(ovsdb_schema, 'UcastMacsRemote'):
self.l2gw_ovsdb.update_ucast_macs_remote(mock.MagicMock(),
mock.MagicMock(),
mock.ANY)
get_rand.assert_called_with(128)
send_n_receive.assert_called_with(mock.ANY,
self.op_id, mock.ANY, True)
self.assertTrue(get_update_ucast_mac_remote.called)
self.assertTrue(get_update_ucast_mac_remote.called)
def test_update_ucast_macs_remote_with_no_locator_id(self):
"""Test case to test update ucast_macs_remote
without locator_id and logical_switch_id.
"""
with contextlib.nested(
mock.patch.object(ovsdb_writer.OVSDBWriter,
'_get_dict_for_update_ucast_mac_remote'),
mock.patch.object(ovsdb_writer.OVSDBWriter,
'_get_physical_locator_dict'),
mock.patch.object(random,
'getrandbits',
return_value=self.op_id
),
mock.patch.object(ovsdb_writer.OVSDBWriter,
'_send_and_receive'
),
mock.patch.object(ovsdb_writer.LOG,
'debug'),
mock.patch.object(ovsdb_schema, 'PhysicalLocator'),
mock.patch.object(ovsdb_schema, 'UcastMacsRemote'),
) as (get_update_ucast_mac_remote,
get_physical_locator_dict,
get_rand,
send_n_receive,
mock_log,
mock_pl, mock_ucmr):
with mock.patch.object(ovsdb_writer.OVSDBWriter,
'_get_dict_for_update_ucast_mac_remote') as get_update_ucast_mac_remote, \
mock.patch.object(ovsdb_writer.OVSDBWriter,
'_get_physical_locator_dict') as get_physical_locator_dict, \
mock.patch.object(random,
'getrandbits',
return_value=self.op_id
), \
mock.patch.object(ovsdb_writer.OVSDBWriter,
'_send_and_receive'
), \
mock.patch.object(ovsdb_writer.LOG,
'debug'), \
mock.patch.object(ovsdb_schema, 'PhysicalLocator') as mock_pl, \
mock.patch.object(ovsdb_schema, 'UcastMacsRemote'):
locator = mock_pl.return_value
locator.uuid = None
self.l2gw_ovsdb.update_ucast_macs_remote(mock.MagicMock(),
@ -324,68 +292,56 @@ class TestOVSDBWriter(base.BaseTestCase):
def test_delete_ucast_macs_remote(self):
"""Test case to test delete_ucast_macs_remote."""
with contextlib.nested(
mock.patch.object(random,
'getrandbits',
return_value=self.op_id
),
mock.patch.object(ovsdb_writer.OVSDBWriter,
'_send_and_receive'
),
mock.patch.object(ovsdb_writer.LOG,
'debug')
) as (get_rand,
send_n_receive,
mock_log):
self.l2gw_ovsdb.delete_ucast_macs_remote(mock.Mock(),
mock.MagicMock(),
mock.ANY)
get_rand.assert_called_with(128)
send_n_receive.assert_called_with(mock.ANY,
self.op_id, mock.ANY, True)
with mock.patch.object(random,
'getrandbits',
return_value=self.op_id
) as get_rand, \
mock.patch.object(ovsdb_writer.OVSDBWriter,
'_send_and_receive'
) as send_n_receive, \
mock.patch.object(ovsdb_writer.LOG,
'debug'):
self.l2gw_ovsdb.delete_ucast_macs_remote(mock.Mock(),
mock.MagicMock(),
mock.ANY)
get_rand.assert_called_with(128)
send_n_receive.assert_called_with(mock.ANY,
self.op_id, mock.ANY, True)
def test_update_connection_to_gateway(self):
"""Test case to test update_connection_to_gateway."""
with contextlib.nested(
mock.patch.object(ovsdb_writer.OVSDBWriter,
'_get_bindings_to_update'),
mock.patch.object(random,
'getrandbits',
return_value=self.op_id
),
mock.patch.object(ovsdb_writer.OVSDBWriter,
'_send_and_receive'
),
mock.patch.object(ovsdb_writer.LOG,
'debug')
) as (get_bindings,
get_rand,
send_n_receive,
mock_log):
self.l2gw_ovsdb.update_connection_to_gateway(
mock.Mock(), mock.Mock(), mock.Mock(), mock.Mock(),
mock.ANY, mock.ANY)
get_rand.assert_called_with(128)
send_n_receive.assert_called_with(mock.ANY,
self.op_id, mock.ANY, True)
self.assertTrue(get_bindings.called)
with mock.patch.object(ovsdb_writer.OVSDBWriter,
'_get_bindings_to_update') as get_bindings, \
mock.patch.object(random,
'getrandbits',
return_value=self.op_id
) as get_rand, \
mock.patch.object(ovsdb_writer.OVSDBWriter,
'_send_and_receive'
) as send_n_receive, \
mock.patch.object(ovsdb_writer.LOG,
'debug'):
self.l2gw_ovsdb.update_connection_to_gateway(
mock.Mock(), mock.Mock(), mock.Mock(), mock.Mock(),
mock.ANY, mock.ANY)
get_rand.assert_called_with(128)
send_n_receive.assert_called_with(mock.ANY,
self.op_id, mock.ANY, True)
self.assertTrue(get_bindings.called)
def test_get_bindings_to_update1(self):
"""Test case to test _get_bindings_to_update."""
fake_op_method = 'CREATE'
with contextlib.nested(
mock.patch.object(ovsdb_writer.OVSDBWriter,
'_form_logical_switch'),
mock.patch.object(ovsdb_writer.OVSDBWriter,
'_form_physical_locators'),
mock.patch.object(ovsdb_writer.OVSDBWriter,
'_form_ports'),
mock.patch.object(ovsdb_schema, 'LogicalSwitch'),
mock.patch.object(ovsdb_schema, 'PhysicalLocator'),
mock.patch.object(ovsdb_schema, 'UcastMacsRemote'),
mock.patch.object(ovsdb_schema, 'PhysicalPort')
) as (form_ls, form_pl, form_pp,
mock_ls, mock_pl, mock_ucmr, mock_pp):
with mock.patch.object(ovsdb_writer.OVSDBWriter,
'_form_logical_switch') as form_ls, \
mock.patch.object(ovsdb_writer.OVSDBWriter,
'_form_physical_locators') as form_pl, \
mock.patch.object(ovsdb_writer.OVSDBWriter,
'_form_ports') as form_pp, \
mock.patch.object(ovsdb_schema, 'LogicalSwitch') as mock_ls, \
mock.patch.object(ovsdb_schema, 'PhysicalLocator') as mock_pl, \
mock.patch.object(ovsdb_schema, 'UcastMacsRemote') as mock_ucmr, \
mock.patch.object(ovsdb_schema, 'PhysicalPort') as mock_pp:
ls = mock_ls.return_value = ovsdb_schema.LogicalSwitch(
'ls_uuid', 'ls_name', 'ls_key', 'ls_desc')
pl = mock_pl.return_value = ovsdb_schema.PhysicalLocator(
@ -492,11 +448,9 @@ class TestOVSDBWriter(base.BaseTestCase):
None,
None,
'')
with contextlib.nested(
mock.patch.object(socket, 'socket', return_value=fake_socket
),
mock.patch.object(ovsdb_writer.LOG, 'warning')
) as (fake_sock, fake_warn):
with mock.patch.object(socket, 'socket',
return_value=fake_socket):
with mock.patch.object(ovsdb_writer.LOG, 'warning'):
ovsdb_conf = base_test.FakeConf()
l2gw_obj = ovsdb_writer.OVSDBWriter(
cfg.CONF.ovsdb, ovsdb_conf)
@ -509,11 +463,9 @@ class TestOVSDBWriter(base.BaseTestCase):
fake_socket = base_test.SocketClass(None,
None,
socket.error)
with contextlib.nested(
mock.patch.object(socket, 'socket', return_value=fake_socket
),
mock.patch.object(ovsdb_writer.LOG, 'warning')
) as (fake_sock, fake_warn):
with mock.patch.object(socket, 'socket', return_value=fake_socket):
with mock.patch.object(ovsdb_writer.LOG,
'warning') as fake_warn:
ovsdb_conf = base_test.FakeConf()
l2gw_obj = ovsdb_writer.OVSDBWriter(
cfg.CONF.ovsdb, ovsdb_conf)

17
networking_l2gw/tests/unit/services/l2gateway/agent/test_l2gw_agent.py
Unescape View File

@ -13,8 +13,6 @@
# License for the specific language governing permissions and limitations
# under the License.
import contextlib
import mock
from neutron.common import config as common_config
@ -45,14 +43,11 @@ class TestL2gwAgent(base.BaseTestCase):
def test_main_l2gw_agent(self):
logging_str = 'neutron.agent.common.config.setup_logging'
common_config_str = mock.patch.object(common_config, 'init').start()
with contextlib.nested(
mock.patch.object(common_config_str, 'init'),
mock.patch(logging_str),
mock.patch.object(agent.service, 'launch'),
mock.patch('sys.argv'),
mock.patch.object(agent.manager, 'OVSDBManager'),
mock.patch.object(cfg.CONF, 'register_opts')
) as (mock_config, mock_logging, mock_launch, sys_argv,
mgr_cls, ro):
with mock.patch.object(common_config_str, 'init'), \
mock.patch(logging_str), \
mock.patch.object(agent.service, 'launch') as mock_launch, \
mock.patch('sys.argv'), \
mock.patch.object(agent.manager, 'OVSDBManager'), \
mock.patch.object(cfg.CONF, 'register_opts'):
agent.main()
mock_launch.assert_called_once_with(cfg.CONF, mock.ANY)

14
networking_l2gw/tests/unit/services/l2gateway/common/test_tunnel_calls.py
Unescape View File

@ -15,7 +15,6 @@
import mock
import contextlib
from neutron.plugins.ml2.drivers.l2pop import rpc as l2pop_rpc
from neutron.plugins.ml2.drivers import type_tunnel
from neutron.plugins.ml2 import managers
@ -34,13 +33,12 @@ class TestTunnelCalls(base.BaseTestCase):
self.context = mock.MagicMock()
def test_trigger_tunnel_sync(self):
with contextlib.nested(
mock.patch.object(rpc, 'RpcCallbacks'),
mock.patch.object(type_tunnel.TunnelRpcCallbackMixin,
'tunnel_sync')) as (mock_rpc, mock_tunnel_sync):
self.tunnel_call.trigger_tunnel_sync(self.context, 'fake_ip')
mock_tunnel_sync.assert_called_with(
self.context, tunnel_ip='fake_ip', tunnel_type='vxlan')
with mock.patch.object(rpc, 'RpcCallbacks'):
with mock.patch.object(type_tunnel.TunnelRpcCallbackMixin,
'tunnel_sync') as mock_tunnel_sync:
self.tunnel_call.trigger_tunnel_sync(self.context, 'fake_ip')
mock_tunnel_sync.assert_called_with(
self.context, tunnel_ip='fake_ip', tunnel_type='vxlan')
def test_trigger_l2pop_sync(self):
fake_fdb_entry = "fake_fdb_entry"

451
networking_l2gw/tests/unit/services/l2gateway/ovsdb/test_data.py
Unescape View File

@ -15,7 +15,6 @@
import mock
import contextlib
from neutron import context
from neutron import manager
from neutron.plugins.ml2 import managers
@ -123,52 +122,36 @@ class TestOVSDBData(base.BaseTestCase):
'deleted_physical_locators': fake_deleted_physical_locators,
'deleted_local_macs': fake_deleted_local_macs,
'deleted_remote_macs': fake_deleted_remote_macs}
with contextlib.nested(
with mock.patch.object(self.ovsdb_data,
'_process_new_logical_switches') as process_new_logical_switches, \
mock.patch.object(self.ovsdb_data,
'_process_new_logical_switches'),
'_process_new_physical_ports') as process_new_physical_ports, \
mock.patch.object(self.ovsdb_data,
'_process_new_physical_ports'),
'_process_new_physical_switches') as process_new_physical_switches, \
mock.patch.object(self.ovsdb_data,
'_process_new_physical_switches'),
'_process_new_physical_locators') as process_new_physical_locators, \
mock.patch.object(self.ovsdb_data,
'_process_new_physical_locators'),
'_process_new_local_macs') as process_new_local_macs, \
mock.patch.object(self.ovsdb_data,
'_process_new_local_macs'),
'_process_new_remote_macs') as process_new_remote_macs, \
mock.patch.object(self.ovsdb_data,
'_process_new_remote_macs'),
'_process_modified_remote_macs') as process_modified_remote_macs, \
mock.patch.object(self.ovsdb_data,
'_process_modified_remote_macs'),
'_process_modified_physical_ports') as process_modified_physical_ports, \
mock.patch.object(self.ovsdb_data,
'_process_modified_physical_ports'),
'_process_deleted_logical_switches') as process_deleted_logical_switches, \
mock.patch.object(self.ovsdb_data,
'_process_deleted_logical_switches'),
'_process_deleted_physical_switches') as process_deleted_physical_switches, \
mock.patch.object(self.ovsdb_data,
'_process_deleted_physical_switches'),
'_process_deleted_physical_ports') as process_deleted_physical_ports, \
mock.patch.object(self.ovsdb_data,
'_process_deleted_physical_ports'),
'_process_deleted_physical_locators') as process_deleted_physical_locators, \
mock.patch.object(self.ovsdb_data,
'_process_deleted_physical_locators'),
'_process_deleted_local_macs') as process_deleted_local_macs, \
mock.patch.object(self.ovsdb_data,
'_process_deleted_local_macs'),
'_process_deleted_remote_macs') as process_deleted_remote_macs, \
mock.patch.object(self.ovsdb_data,
'_process_deleted_remote_macs'),
mock.patch.object(self.ovsdb_data,
'_handle_l2pop')
) as (process_new_logical_switches,
process_new_physical_ports,
process_new_physical_switches,
process_new_physical_locators,
process_new_local_macs,
process_new_remote_macs,
process_modified_remote_macs,
process_modified_physical_ports,
process_deleted_logical_switches,
process_deleted_physical_switches,
process_deleted_physical_ports,
process_deleted_physical_locators,
process_deleted_local_macs,
process_deleted_remote_macs,
mock_handle_l2pop):
'_handle_l2pop') as mock_handle_l2pop:
self.ovsdb_data.entry_table = {
'new_logical_switches': process_new_logical_switches,
'new_physical_ports': process_new_physical_ports,
@ -217,7 +200,18 @@ class TestOVSDBData(base.BaseTestCase):
self.context, fake_deleted_remote_macs)
self.assertTrue(mock_handle_l2pop.called)
def test_notify_ovsdb_states(self):
@mock.patch.object(lib, 'get_all_pending_remote_macs_in_asc_order')
@mock.patch.object(lib, 'delete_pending_ucast_mac_remote')
@mock.patch.object(ovsdb_schema, 'LogicalSwitch')
@mock.patch.object(ovsdb_schema, 'PhysicalLocator')
@mock.patch.object(ovsdb_schema, 'UcastMacsRemote')
@mock.patch.object(agent_api.L2gatewayAgentApi, 'add_vif_to_gateway')
@mock.patch.object(agent_api.L2gatewayAgentApi, 'update_vif_to_gateway')
@mock.patch.object(agent_api.L2gatewayAgentApi, 'delete_vif_from_gateway')
def test_notify_ovsdb_states(self, mock_del_vif, mock_upd_vif,
mock_add_vif, mock_ucmr, mock_pl,
mock_ls, mock_del_pend_recs,
mock_get_pend_recs):
fake_ovsdb_states = {'ovsdb1': 'connected'}
fake_dict = {'logical_switch_uuid': 'fake_ls_id',
'mac': 'fake_mac',
@ -230,35 +224,18 @@ class TestOVSDBData(base.BaseTestCase):
fake_update_dict.update(fake_dict)
fake_delete_dict = {'operation': 'delete'}
fake_delete_dict.update(fake_dict)
with contextlib.nested(
mock.patch.object(
lib, 'get_all_pending_remote_macs_in_asc_order'),
mock.patch.object(lib,
'delete_pending_ucast_mac_remote'),
mock.patch.object(ovsdb_schema, 'LogicalSwitch'),
mock.patch.object(ovsdb_schema, 'PhysicalLocator'),
mock.patch.object(ovsdb_schema, 'UcastMacsRemote'),
mock.patch.object(agent_api.L2gatewayAgentApi,
'add_vif_to_gateway'),
mock.patch.object(agent_api.L2gatewayAgentApi,
'update_vif_to_gateway'),
mock.patch.object(agent_api.L2gatewayAgentApi,
'delete_vif_from_gateway')
) as (mock_get_pend_recs, mock_del_pend_recs,
mock_ls, mock_pl, mock_ucmr, mock_add_vif,
mock_upd_vif, mock_del_vif):
mock_get_pend_recs.return_value = [fake_insert_dict]
self.ovsdb_data.notify_ovsdb_states(
self.context, fake_ovsdb_states)
self.assertTrue(mock_add_vif.called)
mock_get_pend_recs.return_value = [fake_update_dict]
self.ovsdb_data.notify_ovsdb_states(
self.context, fake_ovsdb_states)
self.assertTrue(mock_upd_vif.called)
mock_get_pend_recs.return_value = [fake_delete_dict]
self.ovsdb_data.notify_ovsdb_states(
self.context, fake_ovsdb_states)
self.assertTrue(mock_del_vif.called)
mock_get_pend_recs.return_value = [fake_insert_dict]
self.ovsdb_data.notify_ovsdb_states(
self.context, fake_ovsdb_states)
self.assertTrue(mock_add_vif.called)
mock_get_pend_recs.return_value = [fake_update_dict]