kill.assert_has_calls(mock.call(respawning=True))
sleep.assert_has_calls(mock.call(self.proc.respawn_interval))
- spawn.assert_called_once()
+ spawn.assert_called_once_with()
def _test__watch_process(self, callback, kill_event):
self.proc._kill_event = kill_event
self.proc._watch_process(callback, kill_event)
if not kill_event.ready():
- func.assert_called_once()
+ func.assert_called_once_with()
def test__watch_process_exits_on_callback_failure(self):
self._test__watch_process(lambda: False, eventlet.event.Event())
with mock.patch.object(self.proc, '_spawn') as mock_start:
self.proc.start()
- mock_start.assert_called_once()
+ mock_start.assert_called_once_with()
def test__iter_queue_returns_empty_list_for_empty_queue(self):
result = list(self.proc._iter_queue(eventlet.queue.LightQueue()))
else:
self.assertIsNone(self.proc._kill_event)
- mock_kill_event.send.assert_called_once()
+ mock_kill_event.send.assert_called_once_with()
if pid:
- mock_kill_process.assert_called_once(pid)
+ mock_kill_process.assert_called_once_with(pid)
def test__kill_when_respawning_does_not_clear_kill_event(self):
self._test__kill(True)
# Create a provider network
new_net_dict = self._cisco_nexus_plugin.create_network(
self.providernet, self.attachment1)
- mock_db.assert_called_once()
+ self.assertEqual(mock_db.call_count, 1)
for attr in NET_ATTRS:
self.assertEqual(new_net_dict[attr], self.providernet[attr])
# Delete the provider network
kwargs = {}
kwargs['info'] = {'new_controller': '10.10.10.1'}
self.agent.info_update('dummy', **kwargs)
- run_vsctl_func.assert_called_one_with(['set-controller',
- 'br_int',
- 'tcp:10.10.10.1'])
+ run_vsctl_func.assert_called_once_with(['set-controller',
+ 'br_int',
+ 'tcp:10.10.10.1'])
def test_get_info(self):
with mock.patch.object(self.agent.int_br,
with mock.patch.object(utils, 'execute') as exec_fn:
exec_fn.return_value = False
self.assertEqual(self.lbm.ensure_vlan("eth0", "1"), "eth0.1")
- exec_fn.assert_called_twice()
+ # FIXME(kevinbenton): validate the params to the exec_fn calls
+ self.assertEqual(exec_fn.call_count, 2)
exec_fn.return_value = True
self.assertIsNone(self.lbm.ensure_vlan("eth0", "1"))
- exec_fn.assert_called_once()
+ self.assertEqual(exec_fn.call_count, 3)
def test_ensure_vxlan(self):
seg_id = "12345678"
dhcp_call = mock.call.add_bridge_dhcp(bridge, gateway_ip, cidr,
host_rts=host_rts,
- dns_servers=dns_servers)
+ dns_nservers=dns_servers)
self.client.create_dhcp(bridge, gateway_ip, cidr, host_rts=host_rts,
dns_servers=dns_servers)
-
- bridge.assert_has_call(dhcp_call)
+ self.mock_api.assert_has_calls([dhcp_call])
def test_delete_dhcp(self):
routers,
ri.router['tenant_id'])
mock_get_firewalls_for_tenant.assert_called_once_with(ctx)
- mock_driver_update_firewall.assert_called_once(
+ mock_driver_update_firewall.assert_called_once_with(
routers,
fake_firewall_list[0])
routers,
ri.router['tenant_id'])
mock_get_firewalls_for_tenant.assert_called_once_with(ctx)
- mock_driver_delete_firewall.assert_called_once(
+ mock_driver_delete_firewall.assert_called_once_with(
routers,
fake_firewall_list[0])
with mock.patch.object(dhcp, 'LOG') as log:
self.assertRaises(SystemExit, dhcp.DeviceManager,
cfg.CONF, 'sudo', None)
- log.error.assert_called_once()
+ self.assertEqual(log.error.call_count, 1)
class TestLogArgs(base.BaseTestCase):
network = FakeV4Network()
dh._set_default_route(network, 'tap-name')
- device.route.get_gateway.assert_called_once()
+ self.assertEqual(device.route.get_gateway.call_count, 1)
self.assertFalse(device.route.delete_gateway.called)
device.route.add_gateway.assert_called_once_with('192.168.0.1')
network.namespace = 'qdhcp-1234'
dh._set_default_route(network, 'tap-name')
- device.route.get_gateway.assert_called_once()
+ self.assertEqual(device.route.get_gateway.call_count, 1)
self.assertFalse(device.route.delete_gateway.called)
self.assertFalse(device.route.add_gateway.called)
network.namespace = 'qdhcp-1234'
dh._set_default_route(network, 'tap-name')
- device.route.get_gateway.assert_called_once()
+ self.assertEqual(device.route.get_gateway.call_count, 1)
device.route.delete_gateway.assert_called_once_with('192.168.0.1')
self.assertFalse(device.route.add_gateway.called)
network.namespace = 'qdhcp-1234'
dh._set_default_route(network, 'tap-name')
- device.route.get_gateway.assert_called_once()
+ self.assertEqual(device.route.get_gateway.call_count, 1)
device.route.delete_gateway.assert_called_once_with('192.168.0.1')
self.assertFalse(device.route.add_gateway.called)
network = FakeV4Network()
dh._set_default_route(network, 'tap-name')
- device.route.get_gateway.assert_called_once()
+ self.assertEqual(device.route.get_gateway.call_count, 1)
self.assertFalse(device.route.delete_gateway.called)
self.assertFalse(device.route.add_gateway.called)
network = FakeV4Network()
dh._set_default_route(network, 'tap-name')
- device.route.get_gateway.assert_called_once()
+ self.assertEqual(device.route.get_gateway.call_count, 1)
self.assertFalse(device.route.delete_gateway.called)
device.route.add_gateway.assert_called_once_with('192.168.0.1')
network.subnets = [subnet2, FakeV4Subnet()]
dh._set_default_route(network, 'tap-name')
- device.route.get_gateway.assert_called_once()
+ self.assertEqual(device.route.get_gateway.call_count, 1)
self.assertFalse(device.route.delete_gateway.called)
device.route.add_gateway.assert_called_once_with('192.168.1.1')
del router[l3_constants.INTERFACE_KEY]
del router['gw_port']
agent.process_router(ri)
- self.send_arp.assert_called_once()
+ self.assertEqual(self.send_arp.call_count, 1)
self.assertFalse(agent.process_router_floating_ip_addresses.called)
self.assertFalse(agent.process_router_floating_ip_nat_rules.called)
if r not in ri.iptables_manager.ipv4['nat'].rules]
self.assertEqual(len(nat_rules_delta), 2)
self._verify_snat_rules(nat_rules_delta, router)
- self.send_arp.assert_called_once()
+ self.assertEqual(self.send_arp.call_count, 1)
def test_process_router_snat_enabled(self):
agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)
if r not in orig_nat_rules]
self.assertEqual(len(nat_rules_delta), 2)
self._verify_snat_rules(nat_rules_delta, router)
- self.send_arp.assert_called_once()
+ self.assertEqual(self.send_arp.call_count, 1)
def test_process_router_interface_added(self):
agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)
if r not in orig_nat_rules]
self.assertEqual(len(nat_rules_delta), 1)
self._verify_snat_rules(nat_rules_delta, router)
- self.send_arp.assert_called_once()
+ # send_arp is called both times process_router is called
+ self.assertEqual(self.send_arp.call_count, 2)
def test_process_router_interface_removed(self):
agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)
if r not in ri.iptables_manager.ipv4['nat'].rules]
self.assertEqual(len(nat_rules_delta), 1)
self._verify_snat_rules(nat_rules_delta, router, negate=True)
- self.send_arp.assert_called_once()
+ # send_arp is called both times process_router is called
+ self.assertEqual(self.send_arp.call_count, 2)
def test_process_router_internal_network_added_unexpected_error(self):
agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)
with mock.patch('neutron.agent.linux.interface.LOG.debug') as log:
br = interface.BridgeInterfaceDriver(self.conf)
br.unplug('tap0')
- log.assert_called_once()
+ self.assertEqual(log.call_count, 1)
self.ip_dev.assert_has_calls([mock.call('tap0', 'sudo', None),
mock.call().link.delete()])