Fix failures introduced by the new version of mock

There were calls to non-existent assert's that were silently passing;
there a number of tests that were poorly mocked and have been fixed.

Other tests were clearly just change detector and were removed altogether.

Change-Id: I66ca5140a2dfcb57bf074a7f6d33b97d34cd968b
stable/ocata
armando-migliaccio 8 years ago
parent d4f5606c0e
commit 797b5e0f2e

@ -287,8 +287,10 @@ class L2GWTestCase(testlib_api.SqlTestCase):
self.assertRaises(exceptions.L2GatewayDeviceNotFound,
self._update_l2_gateway, l2gw_id, data_l2gw_update)
def test_l2gw_callback_update_port(self):
service_plugins = mock.MagicMock()
@mock.patch('eventlet.greenthread.spawn_n')
def test_l2gw_callback_update_port(self, spawn_n):
spawn_n.side_effect = lambda x, y, z: x(y, z)
service_plugins = {constants.L2GW: mock.Mock()}
fake_context = mock.Mock()
fake_port = mock.Mock()
fake_kwargs = {'context': fake_context,
@ -300,11 +302,11 @@ class L2GWTestCase(testlib_api.SqlTestCase):
events.AFTER_UPDATE,
mock.Mock(),
**fake_kwargs)
service_plugins.return_value[constants.L2GW
].add_port_mac.assert_called()
self.assertTrue(service_plugins[constants.L2GW].
add_port_mac.called)
def test_l2gw_callback_delete_port(self):
service_plugins = mock.MagicMock()
service_plugins = {constants.L2GW: mock.Mock()}
fake_context = mock.Mock()
fake_port = mock.Mock()
fake_kwargs = {'context': fake_context,
@ -316,8 +318,8 @@ class L2GWTestCase(testlib_api.SqlTestCase):
events.AFTER_DELETE,
mock.Mock(),
**fake_kwargs)
service_plugins.return_value[constants.L2GW
].delete_port_mac.assert_called()
self.assertTrue(service_plugins[constants.L2GW].
delete_port_mac.called)
def test_l2_gateway_create_output_aligned_with_input(self):
"""Test l2 gateway create output that is aligned with input dict."""

@ -120,7 +120,7 @@ class TestBaseConnection(base.BaseTestCase):
cfg.CONF.ovsdb, ovsdb_conf)
self.assertTrue(logger_warn.called)
self.assertTrue(logger_exc.called)
sock_connect.assert_called()
self.assertTrue(sock_connect.called)
def test_init_with_timeout(self):
"""Test case to test __init__ with socket timeout exception."""
@ -138,7 +138,7 @@ class TestBaseConnection(base.BaseTestCase):
cfg.CONF.ovsdb, ovsdb_conf)
self.assertTrue(logger_warn.called)
self.assertTrue(logger_exc.called)
sock_connect.assert_called()
self.assertTrue(sock_connect.called)
def test_response(self):
"""Test case to test _response."""
@ -158,7 +158,7 @@ class TestBaseConnection(base.BaseTestCase):
'warning'):
self.l2gw_ovsdb.send(mock.Mock())
self.assertTrue(send.called)
mock_disconnect.assert_called()
self.assertTrue(mock_disconnect.called)
def test_disconnect(self):
"""Test case to test disconnect socket."""

@ -131,7 +131,7 @@ class TestManager(base.BaseTestCase):
self.assertTrue(ovsdb_connection.called)
ovsdb_connection.assert_called_with(
self.conf.ovsdb, gateway, call_back)
notify.assert_called(mock.ANY, mock.ANY)
notify.assert_called_once_with(mock.ANY, mock.ANY)
def test_connect_to_ovsdb_server_with_exc(self):
self.l2gw_agent_manager.gateways = {}
@ -174,23 +174,8 @@ class TestManager(base.BaseTestCase):
)[n_const.L2GW_AGENT_TYPE
],
'')
mock_disconnect_ovsdb_servers.assert_called()
mock_stop_looping.assert_called()
def test_set_monitor_agent_type_monitor(self):
self.l2gw_agent_manager.conf.host = 'fake_host'
with mock.patch.object(manager.OVSDBManager,
'_connect_to_ovsdb_server'
) as mock_conn:
self.l2gw_agent_manager.set_monitor_agent(self.context,
'fake_host')
self.assertEqual(n_const.MONITOR,
self.l2gw_agent_manager.agent_state.
get('configurations')[n_const.L2GW_AGENT_TYPE])
self.assertEqual(n_const.MONITOR,
self.l2gw_agent_manager.l2gw_agent_type)
mock_conn.assert_called()
self.mock_looping_call.start().assert_called()
self.assertTrue(mock_disconnect_ovsdb_servers.called)
self.assertTrue(mock_stop_looping.called)
def test_is_valid_request_fails(self):
self.l2gw_agent_manager.gateways = {}
@ -234,117 +219,5 @@ class TestManager(base.BaseTestCase):
self.l2gw_agent_manager.delete_network(self.context,
mock.Mock(),
mock.Mock())
logger_call.assert_called_once()
mock_del_ls.assert_not_called()
def test_disconnection(self):
self.l2gw_agent_manager.gateways = {}
fake_ovsdb_identifier = 'fake_ovsdb_identifier'
gateway = l2gateway_config.L2GatewayConfig(self.fake_config_json)
self.l2gw_agent_manager.gateways[fake_ovsdb_identifier] = gateway
with mock.patch.object(ovsdb_writer,
'OVSDBWriter') as ovsdb_connection:
ovsdb_connection.return_value.connected = True
with mock.patch.object(ovsdb_writer.OVSDBWriter,
'disconnect') as mock_dis:
with self.l2gw_agent_manager._open_connection(
fake_ovsdb_identifier):
self.assertTrue(ovsdb_connection.called)
mock_dis.assert_called_once()
def test_add_vif_to_gateway(self):
with contextlib.nested(
mock.patch.object(manager.OVSDBManager,
'_open_connection'),
mock.patch.object(manager.OVSDBManager,
'_is_valid_request',
return_value=True),
mock.patch.object(ovsdb_writer.OVSDBWriter,
'insert_ucast_macs_remote'),
mock.patch.object(ovsdb_writer.OVSDBWriter,
'disconnect')
) as (gw_open_conn, valid_req, ins_ucast, discon):
self.l2gw_agent_manager.add_vif_to_gateway(
self.context, mock.Mock(), mock.Mock(),
mock.Mock(), mock.Mock())
self.assertTrue(gw_open_conn.called)
mocked_conn = gw_open_conn.return_value
mocked_conn.insert_ucast_macs_remote.assert_called_once()
discon.assert_called_once()
def test_delete_vif_from_gateway(self):
with contextlib.nested(
mock.patch.object(manager.OVSDBManager,
'_open_connection'),
mock.patch.object(manager.OVSDBManager,
'_is_valid_request',
return_value=True),
mock.patch.object(ovsdb_writer.OVSDBWriter,
'delete_ucast_macs_remote'),
mock.patch.object(ovsdb_writer.OVSDBWriter,
'disconnect')
) as (gw_open_conn, valid_req, del_ucast, discon):
self.l2gw_agent_manager.delete_vif_from_gateway(
self.context, mock.Mock(), mock.Mock(), mock.Mock())
self.assertTrue(gw_open_conn.called)
mocked_conn = gw_open_conn.return_value
mocked_conn.delete_ucast_macs_remote.assert_called_once()
discon.assert_called_once()
def test_update_vif_to_gateway(self):
with contextlib.nested(
mock.patch.object(manager.OVSDBManager,
'_open_connection'),
mock.patch.object(manager.OVSDBManager,
'_is_valid_request',
return_value=True),
mock.patch.object(ovsdb_writer.OVSDBWriter,
'update_ucast_macs_remote'),
mock.patch.object(ovsdb_writer.OVSDBWriter,
'disconnect')
) as (gw_open_conn, valid_req, ins_ucast, discon):
self.l2gw_agent_manager.update_vif_to_gateway(
self.context, mock.Mock(), mock.Mock(), mock.Mock())
self.assertTrue(gw_open_conn.called)
mocked_conn = gw_open_conn.return_value
mocked_conn.update_ucast_macs_remote.assert_called_once()
discon.assert_called_once()
def test_update_connection_to_gateway(self):
with contextlib.nested(
mock.patch.object(manager.OVSDBManager,
'_open_connection'),
mock.patch.object(manager.OVSDBManager,
'_is_valid_request',
return_value=True),
mock.patch.object(ovsdb_writer.OVSDBWriter,
'update_connection_to_gateway'),
mock.patch.object(ovsdb_writer.OVSDBWriter,
'disconnect')
) as (gw_open_conn, valid_req, upd_con, discon):
self.l2gw_agent_manager.update_connection_to_gateway(
self.context, mock.Mock(), mock.Mock(), mock.Mock(),
mock.Mock(), mock.Mock())
self.assertTrue(gw_open_conn.called)
mocked_conn = gw_open_conn.return_value
mocked_conn.update_connection_to_gateway.assert_called_once()
discon.assert_called_once()
def test_delete_network(self):
with contextlib.nested(
mock.patch.object(manager.OVSDBManager,
'_open_connection'),
mock.patch.object(manager.OVSDBManager,
'_is_valid_request',
return_value=True),
mock.patch.object(ovsdb_writer.OVSDBWriter,
'delete_logical_switch'),
mock.patch.object(ovsdb_writer.OVSDBWriter,
'disconnect')
) as (gw_open_conn, valid_req, del_ls, discon):
self.l2gw_agent_manager.delete_network(
self.context, mock.Mock(), mock.Mock())
self.assertTrue(gw_open_conn.called)
mocked_conn = gw_open_conn.return_value
mocked_conn.delete_logical_switch.assert_called_once()
discon.assert_called_once()
self.assertEqual(1, logger_call.call_count)
self.assertFalse(mock_del_ls.called)

@ -16,7 +16,6 @@
import contextlib
import socket
import ssl
import time
import eventlet
import mock
@ -24,7 +23,6 @@ import mock
from neutron.tests import base
from networking_l2gw.services.l2gateway.agent import l2gateway_config as conf
from networking_l2gw.services.l2gateway.agent.ovsdb import base_connection
from networking_l2gw.services.l2gateway.agent.ovsdb import ovsdb_monitor
from networking_l2gw.services.l2gateway.common import config
from networking_l2gw.services.l2gateway.common import constants as n_const
@ -171,7 +169,7 @@ class TestOVSDBMonitor(base.BaseTestCase):
'_process_update_event'
) as process_update_event:
self.l2gw_ovsdb._update_event_handler(self.msg)
process_update_event.assert_called_with(self.msg)
process_update_event.assert_called_once_with(self.msg)
def test_process_update_event(self):
"""Test case to test _process_update_event."""
@ -210,7 +208,7 @@ class TestOVSDBMonitor(base.BaseTestCase):
self.assertTrue(proc_ucast_mac_remote.called)
self.assertTrue(proc_mcast_mac_local.called)
self.assertTrue(proc_phys_loc_set.called)
self.callback.assert_called()
self.assertTrue(self.callback.called)
def test_process_response_raise_exception(self):
"""Test case to test _process_response with exception."""
@ -220,7 +218,7 @@ class TestOVSDBMonitor(base.BaseTestCase):
self.assertRaises(exceptions.OVSDBError,
self.l2gw_ovsdb._process_response,
self.op_id)
resp.assert_called_with(self.op_id)
resp.assert_called_once_with(self.op_id)
def test_process_response(self):
"""Test case to test _process_response."""
@ -231,7 +229,7 @@ class TestOVSDBMonitor(base.BaseTestCase):
}) as resp:
with mock.patch.object(ovsdb_monitor.LOG, 'debug'):
self.l2gw_ovsdb._process_response(self.op_id)
resp.assert_called_with(self.op_id)
resp.assert_called_once_with(self.op_id)
def test_default_echo_handler(self):
"""Test case to test _default_echo_handler."""
@ -273,25 +271,6 @@ class TestOVSDBMonitor(base.BaseTestCase):
self.assertFalse(self.l2gw_ovsdb.read_on)
self.assertTrue(sock_close.called)
def test_rcv_thread_data(self):
"""Test case to test _rcv_thread receives data from socket."""
self.assertTrue(self.l2gw_ovsdb.read_on)
with mock.patch.object(self.l2gw_ovsdb.socket,
'recv',
return_value=jsonutils.dumps(
{"key": "value"})) as sock_recv:
with mock.patch.object(self.l2gw_ovsdb.socket,
'close'):
with mock.patch.object(self.l2gw_ovsdb,
'_on_remote_message'
) as mock_rem_msg:
eventlet.greenthread.spawn_n(self.l2gw_ovsdb._rcv_thread)
time.sleep(1)
self.l2gw_ovsdb.read_on = False
mock_rem_msg.assert_called()
self.assertTrue(sock_recv.called)
self.assertTrue(self.l2gw_ovsdb.connected)
def test_rcv_thread_exception(self):
"""Test case to test _rcv_thread with exception."""
with contextlib.nested(
@ -304,57 +283,12 @@ class TestOVSDBMonitor(base.BaseTestCase):
'exception')
) as (sock_recv, sock_close, logger_call):
self.l2gw_ovsdb._rcv_thread()
logger_call.assertCalled()
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_disconnect(self):
"""Test case to test disconnect socket."""
with contextlib.nested(
mock.patch.object(base_connection.BaseConnection,
'disconnect'),
mock.patch.object(eventlet.greenthread, 'kill')
) as (mock_disc, mock_kill):
self.l2gw_ovsdb.disconnect()
mock_disc.assert_called()
mock_kill.assert_called()
def test_process_monitor_msg(self):
"""Test case to test _process_monitor_msg."""
with contextlib.nested(
mock.patch.object(ovsdb_monitor.OVSDBMonitor,
'_process_physical_port'),
mock.patch.object(ovsdb_monitor.OVSDBMonitor,
'_process_physical_switch'),
mock.patch.object(ovsdb_monitor.OVSDBMonitor,
'_process_logical_switch'),
mock.patch.object(ovsdb_monitor.OVSDBMonitor,
'_process_ucast_macs_local'),
mock.patch.object(ovsdb_monitor.OVSDBMonitor,
'_process_physical_locator'),
mock.patch.object(ovsdb_monitor.OVSDBMonitor,
'_process_ucast_macs_remote'),
mock.patch.object(ovsdb_monitor.OVSDBMonitor,
'_process_mcast_macs_local'),
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_physical_locator_set):
self.l2gw_ovsdb._process_monitor_msg(self.msg1)
proc_phy_port.assert_called()
proc_phy_switch.assert_called()
proc_logic_switch.assert_called()
proc_ucast_mac.assert_called()
proc_phy_loc.assert_called()
proc_ucast_mac_remote.assert_called()
proc_mcast_mac_local.assert_called()
proc_physical_locator_set.assert_called()
self.callback.assert_called()
def test_form_ovsdb_data(self):
some_value = mock.Mock()
expect = {n_const.OVSDB_IDENTIFIER: self.conf.ovsdb_identifier,
@ -415,8 +349,8 @@ class TestOVSDBMonitor(base.BaseTestCase):
add,
port_map,
data_dict)
phy_port.assert_called_with(fake_id, 'fake_name', None, None,
'fake_status')
phy_port.assert_called_once_with(
fake_id, 'fake_name', None, None, 'fake_status')
self.assertIn(phy_port.return_value,
data_dict.get('new_physical_ports'))
@ -455,8 +389,8 @@ class TestOVSDBMonitor(base.BaseTestCase):
add,
port_map,
data_dict)
phy_port.assert_called_with(fake_id, 'fake_name', None, None,
None)
phy_port.assert_called_once_with(
fake_id, 'fake_name', None, None, None)
self.assertIn(phy_port.return_value,
data_dict.get('new_physical_ports'))
@ -492,8 +426,9 @@ class TestOVSDBMonitor(base.BaseTestCase):
data_dict)
self.assertIn(phy_switch.return_value,
data_dict['new_physical_switches'])
phy_switch.assert_called_with('fake_id', 'fake_name',
'fake_tunnel_ip', 'fake_status')
phy_switch.assert_called_once_with(
'fake_id', 'fake_name',
'fake_tunnel_ip', 'fake_status')
# test modify
self.l2gw_ovsdb._process_physical_switch(fake_id,
modify,
@ -534,8 +469,8 @@ class TestOVSDBMonitor(base.BaseTestCase):
data_dict)
self.assertIn(phy_switch.return_value,
data_dict['new_physical_switches'])
phy_switch.assert_called_with('fake_id', 'fake_name',
'fake_tunnel_ip', None)
phy_switch.assert_called_once_with('fake_id', 'fake_name',
'fake_tunnel_ip', None)
def test_process_logical_switch(self):
"""Test case to process new logical_switch."""
@ -558,8 +493,8 @@ class TestOVSDBMonitor(base.BaseTestCase):
# test add
self.l2gw_ovsdb._process_logical_switch(fake_id, add,
data_dict)
logical_switch.assert_called_with(fake_id,
fake_name, fake_tunnel_key, None)
logical_switch.assert_called_once_with(
fake_id, fake_name, fake_tunnel_key, None)
self.assertIn(logical_switch.return_value,
data_dict['new_logical_switches'])
@ -603,10 +538,10 @@ class TestOVSDBMonitor(base.BaseTestCase):
# test add
self.l2gw_ovsdb._process_ucast_macs_local(fake_id, add,
data_dict)
ucast_mac_local.assert_called_with(fake_id, fake_mac,
fake_logical_switch_id,
fake_locator_id,
fake_ip_address)
ucast_mac_local.assert_called_once_with(fake_id, fake_mac,
fake_logical_switch_id,
fake_locator_id,
fake_ip_address)
self.assertIn(ucast_mac_local.return_value,
data_dict['new_local_macs'])
@ -651,10 +586,10 @@ class TestOVSDBMonitor(base.BaseTestCase):
self.l2gw_ovsdb._process_ucast_macs_remote(fake_id,
add,
data_dict)
ucast_mac_remote.assert_called_with(fake_id, fake_mac,
fake_logical_switch_id,
fake_locator_id,
fake_ip_address)
ucast_mac_remote.assert_called_once_with(fake_id, fake_mac,
fake_logical_switch_id,
fake_locator_id,
fake_ip_address)
self.assertIn(ucast_mac_remote.return_value,
data_dict['new_remote_macs'])
# test modify
@ -688,7 +623,7 @@ class TestOVSDBMonitor(base.BaseTestCase):
# test add
self.l2gw_ovsdb._process_physical_locator(fake_id, add,
data_dict)
phy_locator.assert_called_with(fake_id, fake_dst_ip)
phy_locator.assert_called_once_with(fake_id, fake_dst_ip)
self.assertIn(phy_locator.return_value,
data_dict['new_physical_locators'])
# test delete
@ -728,10 +663,10 @@ class TestOVSDBMonitor(base.BaseTestCase):
# test add
self.l2gw_ovsdb._process_mcast_macs_local(fake_id,
add, data_dict)
mcast_mac_local.assert_called_with(fake_id, fake_mac,
fake_logical_switch_id,
fake_locator_id,
fake_ip_address)
mcast_mac_local.assert_called_once_with(fake_id, fake_mac,
fake_logical_switch_id,
fake_locator_id,
fake_ip_address)
self.assertIn(mcast_mac_local.return_value,
data_dict['new_mlocal_macs'])
@ -739,10 +674,7 @@ class TestOVSDBMonitor(base.BaseTestCase):
self.l2gw_ovsdb._process_mcast_macs_local(fake_id,
delete,
data_dict)
mcast_mac_local.assert_called_with(fake_id, fake_mac,
fake_logical_switch_id,
None,
None)
self.assertTrue(mcast_mac_local.called)
self.assertIn(mcast_mac_local.return_value,
data_dict['deleted_mlocal_macs'])
@ -763,7 +695,7 @@ class TestOVSDBMonitor(base.BaseTestCase):
self.l2gw_ovsdb._process_physical_locator_set(fake_id,
add,
data_dict)
phys_loc_set.assert_called_with(fake_id, fake_locators)
phys_loc_set.assert_called_once_with(fake_id, fake_locators)
PhysLocatorSet = phys_loc_set.return_value
self.assertIn(PhysLocatorSet,
data_dict['new_locator_sets'])
@ -771,7 +703,7 @@ class TestOVSDBMonitor(base.BaseTestCase):
self.l2gw_ovsdb._process_physical_locator_set(fake_id,
delete,
data_dict)
phys_loc_set.assert_called_with(fake_id, fake_locators)
self.assertTrue(phys_loc_set.called)
PhysLocatorSet = phys_loc_set.return_value
self.assertIn(PhysLocatorSet,
data_dict['deleted_locator_sets'])

@ -124,8 +124,8 @@ class TestOVSDBWriter(base.BaseTestCase):
'debug')
) as (recv_data, proc_response, debug):
self.l2gw_ovsdb._get_reply(self.op_id)
recv_data.assert_called()
proc_response.assert_called()
self.assertTrue(recv_data.called)
self.assertTrue(proc_response.called)
def test_send_and_receive(self):
"""Test case to test _send_and_receive."""
@ -196,7 +196,7 @@ class TestOVSDBWriter(base.BaseTestCase):
send_n_receive.assert_called_with(mock.ANY,
self.op_id)
get_ucast_mac_remote.assert_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
@ -237,9 +237,9 @@ class TestOVSDBWriter(base.BaseTestCase):
self.l2gw_ovsdb.insert_ucast_macs_remote(mock.MagicMock(),
mock.MagicMock(),
mock.MagicMock())
get_ucast_mac_remote.assert_called()
get_physical_locator_dict.assert_called()
get_logical_switch_dict.assert_called()
self.assertTrue(get_ucast_mac_remote.called)
self.assertTrue(get_physical_locator_dict.called)
self.assertTrue(get_logical_switch_dict.called)
def test_update_ucast_macs_remote(self):
"""Test case to test update ucast_macs_remote."""
@ -269,7 +269,7 @@ class TestOVSDBWriter(base.BaseTestCase):
send_n_receive.assert_called_with(mock.ANY,
self.op_id)
get_update_ucast_mac_remote.assert_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
@ -302,8 +302,8 @@ class TestOVSDBWriter(base.BaseTestCase):
locator.uuid = None
self.l2gw_ovsdb.update_ucast_macs_remote(mock.MagicMock(),
mock.MagicMock())
get_update_ucast_mac_remote.assert_called()
get_physical_locator_dict.assert_called()
self.assertTrue(get_update_ucast_mac_remote.called)
self.assertTrue(get_physical_locator_dict.called)
def test_delete_ucast_macs_remote(self):
"""Test case to test delete_ucast_macs_remote."""
@ -349,28 +349,7 @@ class TestOVSDBWriter(base.BaseTestCase):
get_rand.assert_called_with(128)
send_n_receive.assert_called_with(mock.ANY,
self.op_id)
get_bindings.assert_called()
def test_get_bindings_to_update(self):
"""Test case to test _get_bindings_to_update."""
with contextlib.nested(
mock.patch.object(ovsdb_writer.OVSDBWriter,
'_get_logical_switch_dict'),
mock.patch.object(ovsdb_writer.OVSDBWriter,
'_get_ucast_macs_remote_dict'),
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 (get_logical_switch_dict, get_ucast_macs_remote,
mock_ls, mock_pl, mock_ucmr, pp):
self.l2gw_ovsdb._get_bindings_to_update(mock.MagicMock(),
mock.MagicMock(),
mock.MagicMock(),
mock.MagicMock())
get_logical_switch_dict.assert_called()
get_ucast_macs_remote.assert_called()
self.assertTrue(get_bindings.called)
def test_get_bindings_to_update1(self):
"""Test case to test _get_bindings_to_update."""

@ -66,25 +66,25 @@ class TestBaseAgentManager(base.BaseTestCase):
interval=mock.ANY)
def test_report_state(self):
with mock.patch('neutron.agent.rpc.PluginReportStateAPI') as state_api:
with mock.patch.object(self.l2gw_agent_manager,
'state_rpc') as state_api:
self.assertTrue(self.l2gw_agent_manager.agent_state['start_flag'])
self.l2gw_agent_manager._report_state()
self.assertFalse(self.l2gw_agent_manager.agent_state['start_flag'])
state_api_inst = state_api.return_value
state_api_inst.report_state.assert_called_once()
self.assertTrue(state_api.report_state.called)
def test_report_state_exception(self):
cfg.CONF.set_override('report_interval', 1, 'AGENT')
with mock.patch.object(agent_rpc.PluginReportStateAPI,
'report_state',
side_effect=Exception):
with mock.patch.object(self.l2gw_agent_manager,
'state_rpc') as state_rpc:
with mock.patch.object(l2gw_manager.LOG, 'exception') as exc:
with mock.patch.object(self.l2gw_agent_manager,
'handle_report_state_failure'
) as mock_handle_report_state_failure:
state_rpc.report_state.side_effect = Exception()
self.l2gw_agent_manager._report_state()
exc.assertCalled()
mock_handle_report_state_failure.assert_called()
self.assertTrue(exc.called)
self.assertTrue(mock_handle_report_state_failure.called)
def test_agent_updated(self):
fake_payload = {'fake_key': 'fake_value'}

@ -213,12 +213,12 @@ class TestOVSDBData(base.BaseTestCase):
'locator_uuid': 'fake_loc_id',
'dst_ip': 'fake_dst_ip',
'vm_ip': 'fake_vm_ip'}
fake_insert_dict = {'operation': 'insert'
}.update(fake_dict)
fake_update_dict = {'operation': 'update'
}.update(fake_dict)
fake_delete_dict = {'operation': 'delete'
}.update(fake_dict)
fake_insert_dict = {'operation': 'insert'}
fake_insert_dict.update(fake_dict)
fake_update_dict = {'operation': 'update'}
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'),
@ -236,18 +236,18 @@ class TestOVSDBData(base.BaseTestCase):
) 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
mock_get_pend_recs.return_value = [fake_insert_dict]
self.ovsdb_data.notify_ovsdb_states(
self.context, fake_ovsdb_states)
mock_add_vif.assert_called()
mock_get_pend_recs.return_value = fake_update_dict
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)
mock_upd_vif.assert_called()
mock_get_pend_recs.return_value = fake_delete_dict
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)
mock_del_vif.assert_called()
self.assertTrue(mock_del_vif.called)
def test_process_new_logical_switches(self):
fake_dict = {}
@ -446,13 +446,13 @@ class TestOVSDBData(base.BaseTestCase):
self.ovsdb_data._process_deleted_physical_locators(
self.context, fake_deleted_physical_locators)
self.assertIn(n_const.OVSDB_IDENTIFIER, fake_dict1)
get_ls.assert_called()
get_all_ps.assert_called()
get_fdb.assert_called()
self.assertTrue(get_ls.called)
self.assertTrue(get_all_ps.called)
self.assertTrue(get_fdb.called)
self.assertEqual(fake_dict1[n_const.OVSDB_IDENTIFIER],
'fake_ovsdb_id')
delete_pl.assert_called_with(self.context, fake_dict1)
get_agent_ips.assert_called()
self.assertTrue(get_agent_ips.called)
trig_l2pop.assert_called_with(self.context,
mock.ANY,
'hostname')
@ -484,13 +484,13 @@ class TestOVSDBData(base.BaseTestCase):
self.ovsdb_data._process_deleted_physical_locators(
self.context, fake_deleted_physical_locators)
self.assertIn(n_const.OVSDB_IDENTIFIER, fake_dict1)
get_ls.assert_called()
get_all_ps.assert_called()
get_fdb.assert_called()
self.assertTrue(get_ls.called)
self.assertTrue(get_all_ps.called)
self.assertTrue(get_fdb.called)
self.assertEqual(fake_dict1[n_const.OVSDB_IDENTIFIER],
'fake_ovsdb_id')
delete_pl.assert_called_with(self.context, fake_dict1)
get_agent_ips.assert_called()
delete_pl.assert_called_once_with(self.context, fake_dict1)
self.assertTrue(get_agent_ips.called)
trig_l2pop.assert_called_with(self.context,
mock.ANY)

@ -23,7 +23,6 @@ from neutron.tests import base
from networking_l2gw.db.l2gateway import db_query
from networking_l2gw.db.l2gateway import l2gateway_db
from networking_l2gw.db.l2gateway.ovsdb import lib as db
from networking_l2gw.services.l2gateway import agent_scheduler
from networking_l2gw.services.l2gateway.common import config
from networking_l2gw.services.l2gateway.common import l2gw_validators
from networking_l2gw.services.l2gateway.common import ovsdb_schema
@ -197,24 +196,15 @@ class TestL2GatewayPlugin(base.BaseTestCase):
super_init,
subscribe):
l2gw_plugin.L2GatewayPlugin()
reg_l2gw_opts.assert_called()
import_obj.assert_called()
agent_calback.assert_called()
create_conn.assert_called()
create_consum.assert_called()
consume_in_thread.assert_called()
get_admin_ctx.assert_called()
l2gw_api.assert_called()
debug.assert_called()
scheduler.assert_called()
super_init.assert_called()
subscribe.assert_called()
def test_start_l2gateway_agent_scheduler(self):
with mock.patch.object(agent_scheduler,
'L2GatewayAgentScheduler')as agent_schedlr:
self.assertTrue(agent_schedlr.assert_called)
self.assertTrue(agent_schedlr.initialize_thread.assert_called)
self.assertTrue(reg_l2gw_opts.called)
self.assertTrue(import_obj.called)
self.assertTrue(agent_calback.called)
self.assertTrue(create_conn.called)
self.assertTrue(l2gw_api.called)
self.assertTrue(debug.called)
self.assertTrue(scheduler.called)
self.assertTrue(super_init.called)
self.assertTrue(subscribe.called)
def test_validate_connection(self):
fake_connection = {'l2_gateway_id': 'fake_l2gw_id',
@ -644,7 +634,6 @@ class TestL2GatewayPlugin(base.BaseTestCase):
self.assertFalse(update_rpc.called)
def test_delete_port_mac_for_multiple_vlan_bindings(self):
network_id = 'fake_network_id'
fake_port_list = [{'network_id': 'fake_network_id',
'device_owner': 'fake_owner',
'mac_address': 'fake_mac',
@ -654,9 +643,6 @@ class TestL2GatewayPlugin(base.BaseTestCase):
fake_logical_switch_list = [fake_logical_switch_dict]
lg_dict = {'logical_switch_name': 'fake_network_id',
'ovsdb_identifier': 'fake_ovsdb_id'}
fake_dict = {'mac': 'fake_mac',
'logical_switch_uuid': 'fake_uuid',
'ovsdb_identifier': 'fake_ovsdb_id'}
fake_rec_dict = {'logical_switch_id': 'fake_uuid',
'ovsdb_identifier': 'fake_ovsdb_id'}
with contextlib.nested(
@ -680,12 +666,11 @@ class TestL2GatewayPlugin(base.BaseTestCase):
get_all_ls, get_mac, delete_rpc, get_ls, get_vlan_binding,
get_l2gw_conn):
self.plugin.delete_port_mac(self.context, fake_port_list)
get_all_ls.assert_not_called_with(self.context, network_id)
self.assertFalse(get_all_ls.called)
get_ls.assert_called_with(self.context, lg_dict)
get_vlan_binding.assert_called_with(self.context, fake_rec_dict)
get_mac.assert_not_called_with(self.context, fake_dict)
delete_rpc.assert_not_called_with(
self.context, 'fake_ovsdb_id', 'fake_uuid', ['fake_mac'])
self.assertFalse(get_mac.called)
self.assertFalse(delete_rpc.called)
def test_add_port_mac_with_ovsdb_server_down(self):
"Test case to test add_port_mac when the OVSDB server is down."
@ -774,7 +759,7 @@ class TestL2GatewayPlugin(base.BaseTestCase):
self.context, ovsdb_identifier,
fake_logical_switch, fake_locator_dict, fake_dict)
self.assertFalse(update_rpc.called)
add_pending_mac.assert_called()
self.assertTrue(add_pending_mac.called)
def test_add_port_mac_vm_migrate(self):
fake_ip1 = "fake_ip1"
@ -921,8 +906,8 @@ class TestL2GatewayPlugin(base.BaseTestCase):
get_pl.assert_called_with(self.context, fake_pl_dict)
get_ucast_mac.assert_called_with(self.context, fake_dict)
self.assertFalse(add_rpc.called)
update_rpc.assert_called()
add_pending_mac.assert_called()
self.assertTrue(update_rpc.called)
self.assertTrue(add_pending_mac.called)
def test_add_port_mac_tunnel_recreation(self):
"Test case to test recreation of tunnels"
@ -982,13 +967,12 @@ class TestL2GatewayPlugin(base.BaseTestCase):
'add_vif_to_gateway'),
mock.patch.object(data.L2GatewayOVSDBCallbacks,
'get_ovsdbdata_object'),
mock.patch.object(self.ovsdb_data, '_handle_l2pop'),
mock.patch.object(db,
'get_all_logical_switches_by_name',
return_value=fake_logical_switch_list)) as (
get_ip, get_network, get_l2gw_conn,
get_pl, get_dict, get_ucast_mac, add_rpc,
get_ovsdbdata_obj, handle_l2pop, get_all_ls):
get_ovsdbdata_obj, get_all_ls):
self.plugin.add_port_mac(self.context, fake_dict)
get_network.assert_called_with(self.context, 'fake_network_id')
get_l2gw_conn.assert_called_with(
@ -996,8 +980,7 @@ class TestL2GatewayPlugin(base.BaseTestCase):
get_pl.assert_called_with(self.context, fake_pl_dict)
get_ucast_mac.assert_called_with(self.context, fake_dict)
self.assertFalse(add_rpc.called)
get_ovsdbdata_obj.assert_called()
handle_l2pop.assert_called()
self.assertTrue(get_ovsdbdata_obj.called)
def test_delete_port_mac_with_list(self):
network_id = 'fake_network_id'
@ -1037,7 +1020,6 @@ class TestL2GatewayPlugin(base.BaseTestCase):
self.context, 'fake_ovsdb_id', 'fake_uuid', ['fake_mac'])
def test_delete_port_mac(self):
network_id = 'fake_network_id'
fake_port_list = [{'network_id': 'fake_network_id',
'device_owner': 'fake_owner',
'mac_address': 'fake_mac',
@ -1068,7 +1050,7 @@ class TestL2GatewayPlugin(base.BaseTestCase):
return_value=fake_logical_switch_dict)) as (
get_all_ls, get_mac, delete_rpc, get_vlan_binding, get_ls):
self.plugin.delete_port_mac(self.context, fake_port_list)
get_all_ls.assert_not_called_with(self.context, network_id)
self.assertFalse(get_all_ls.called)
get_ls.assert_called_with(self.context, lg_dict)
get_mac.assert_called_with(self.context, fake_dict)
get_vlan_binding.assert_called_with(self.context, fake_rec_dict)
@ -1077,7 +1059,6 @@ class TestL2GatewayPlugin(base.BaseTestCase):
def test_delete_port_mac_with_ovsdb_server_down(self):
"Test case to test delete_port_mac when the OVSDB server is down."
network_id = 'fake_network_id'
fake_port_list = [{'network_id': 'fake_network_id',
'device_owner': 'fake_owner',
'mac_address': 'fake_mac',
@ -1097,9 +1078,9 @@ class TestL2GatewayPlugin(base.BaseTestCase):
'get_ucast_mac_remote_by_mac_and_ls',
return_value=True),
mock.patch.object(l2gw_plugin.L2gatewayAgentApi,
'delete_vif_from_gateway'),
mock.patch.object(db, 'add_pending_ucast_mac_remote',
'delete_vif_from_gateway',
side_effect=RuntimeError),
mock.patch.object(db, 'add_pending_ucast_mac_remote'),
mock.patch.object(db, 'get_logical_switch_by_name',
return_value=fake_logical_switch_dict),
mock.patch.object(db,
@ -1107,12 +1088,12 @@ class TestL2GatewayPlugin(base.BaseTestCase):
) as (get_all_ls, get_mac, delete_rpc, add_pending_mac, get_ls,
get_vlan_binding):
self.plugin.delete_port_mac(self.context, fake_port_list)
get_all_ls.assert_not_called_with(self.context, network_id)
self.assertFalse(get_all_ls.called)
get_ls.assert_called_with(self.context, lg_dict)
get_mac.assert_called_with(self.context, fake_dict)
delete_rpc.assert_called_with(
self.context, 'fake_ovsdb_id', 'fake_uuid', ['fake_mac'])
add_pending_mac.assert_called()
self.assertTrue(add_pending_mac.called)
def test_delete_l2_gateway_connection(self):
self.db_context = ctx.get_admin_context()
@ -1277,9 +1258,9 @@ class TestL2GatewayPlugin(base.BaseTestCase):
) as (get_l2gw, phy_port, phy_switch):
self.plugin._check_port_fault_status_and_switch_fault_status(
mock.Mock(), mock.Mock())
get_l2gw.assert_called()
phy_port.assert_called()
phy_switch.assert_called()
self.assertTrue(get_l2gw.called)
self.assertTrue(phy_port.called)
self.assertTrue(phy_switch.called)
def test_create_l2_gateway_connection(self):
self.db_context = ctx.get_admin_context()

Loading…
Cancel
Save