openstack
/
networking-l2gw
Code Issues Proposed changes

Python 3: add support

Browse Source 
The patche does the following:
1. remove nested with calls from unit tests
2. updates tox.ini to enable us to use the \ at the end
   of a line
3. All integers in python 3 are long.
4. Python 3 does not support __builtin__. Removed it from the tests.

Now the unit tests pass when running python 3.

This also adds support to tox.

Change-Id: Iacbf0c2c05248819af817109e7dd637f71c82404
changes/53/347253/2
Gary Kotton 7 years ago
parent 8d53fea503
commit 327f5d3af5
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]
self