# scan system for any existing fip ports
ri.dist_fip_count = 0
rtr_2_fip_interface = self.get_rtr_ext_device_name(ri.router_id)
- if ip_lib.device_exists(rtr_2_fip_interface, namespace=ri.ns_name):
- device = ip_lib.IPDevice(rtr_2_fip_interface, namespace=ri.ns_name)
+ device = ip_lib.IPDevice(rtr_2_fip_interface, namespace=ri.ns_name)
+ if device.exists():
existing_cidrs = [addr['cidr'] for addr in device.addr.list()]
fip_cidrs = [c for c in existing_cidrs if
common_utils.is_cidr_host(c)]
def set_log_fail_as_error(self, fail_with_error):
self.log_fail_as_error = fail_with_error
+ def get_log_fail_as_error(self):
+ return self.log_fail_as_error
+
class IPWrapper(SubProcessBase):
def __init__(self, namespace=None):
def __str__(self):
return self.name
+ def exists(self):
+ """Return True if the device exists in the namespace."""
+ # we must save and restore this before returning
+ orig_log_fail_as_error = self.get_log_fail_as_error()
+ self.set_log_fail_as_error(False)
+ try:
+ address = self.link.address
+ except RuntimeError:
+ exists = False
+ else:
+ exists = bool(address)
+ finally:
+ self.set_log_fail_as_error(orig_log_fail_as_error)
+
+ return exists
+
def delete_addr_and_conntrack_state(self, cidr):
"""Delete an address along with its conntrack state
def device_exists(device_name, namespace=None):
"""Return True if the device exists in the namespace."""
- try:
- dev = IPDevice(device_name, namespace=namespace)
- dev.set_log_fail_as_error(False)
- address = dev.link.address
- except RuntimeError:
- return False
- return bool(address)
+ return IPDevice(device_name, namespace=namespace).exists()
def device_exists_with_ips_and_mac(device_name, ip_cidrs, mac, namespace=None):
Non-internal OVS ports need to be removed manually.
"""
for port in ports:
- if ip_lib.device_exists(port):
- device = ip_lib.IPDevice(port)
+ device = ip_lib.IPDevice(port)
+ if device.exists():
device.link.delete()
LOG.info(_LI("Deleting port: %s"), port)
tap_device_name)
def delete_bridge(self, bridge_name):
- if ip_lib.device_exists(bridge_name):
+ bridge_device = bridge_lib.BridgeDevice(bridge_name)
+ if bridge_device.exists():
physical_interfaces = set(self.interface_mappings.values())
interfaces_on_bridge = self.get_interfaces_on_bridge(bridge_name)
for interface in interfaces_on_bridge:
self.delete_interface(interface)
LOG.debug("Deleting bridge %s", bridge_name)
- bridge_device = bridge_lib.BridgeDevice(bridge_name)
if bridge_device.link.set_down():
return
if bridge_device.delbr():
del self.network_map[network_id]
def remove_interface(self, bridge_name, interface_name):
- if ip_lib.device_exists(bridge_name):
+ bridge_device = bridge_lib.BridgeDevice(bridge_name)
+ if bridge_device.exists():
if not self.is_device_on_bridge(interface_name):
return True
LOG.debug("Removing device %(interface_name)s from bridge "
"%(bridge_name)s",
{'interface_name': interface_name,
'bridge_name': bridge_name})
- if bridge_lib.BridgeDevice(bridge_name).delif(interface_name):
+ if bridge_device.delif(interface_name):
return False
LOG.debug("Done removing device %(interface_name)s from bridge "
"%(bridge_name)s",
return False
def delete_interface(self, interface):
- if ip_lib.device_exists(interface):
+ device = self.ip.device(interface)
+ if device.exists():
LOG.debug("Deleting interface %s",
interface)
- device = self.ip.device(interface)
device.link.set_down()
device.link.delete()
LOG.debug("Done deleting interface %s", interface)
if int_type == 'patch':
self.int_br.delete_port(int_if_name)
br.delete_port(phys_if_name)
- if ip_lib.device_exists(int_if_name):
- ip_lib.IPDevice(int_if_name).link.delete()
+ device = ip_lib.IPDevice(int_if_name)
+ if device.exists():
+ device.link.delete()
# Give udev a chance to process its rules here, to avoid
# race conditions between commands launched by udev rules
# and the subsequent call to ip_wrapper.add_veth
self.assertFalse(
ip_lib.device_exists(attr.name, namespace=attr.namespace))
+ def test_ipdevice_exists(self):
+ attr = self.generate_device_details()
+ device = self.manage_device(attr)
+ self.assertTrue(device.exists())
+ device.link.delete()
+ self.assertFalse(device.exists())
+
def test_vxlan_exists(self):
attr = self.generate_device_details()
ip = ip_lib.IPWrapper(namespace=attr.namespace)
_execute.assert_called_once_with(['o'], 'link', ('show', 'eth0'),
log_fail_as_error=False)
+ def test_device_exists_reset_fail(self):
+ device = ip_lib.IPDevice('eth0')
+ device.set_log_fail_as_error(True)
+ with mock.patch.object(ip_lib.IPDevice, '_execute') as _execute:
+ _execute.return_value = LINK_SAMPLE[1]
+ self.assertTrue(device.exists())
+ self.assertTrue(device.get_log_fail_as_error())
+
def test_device_does_not_exist(self):
with mock.patch.object(ip_lib.IPDevice, '_execute') as _execute:
_execute.return_value = ''
ports = ['tap1234', 'tap5678', 'tap09ab']
port_found = [True, False, True]
- with mock.patch.object(
- ip_lib, 'device_exists',
- side_effect=port_found) as device_exists:
- util.delete_neutron_ports(ports)
- device_exists.assert_has_calls([mock.call(p) for p in ports])
- mock_ip.assert_has_calls(
- [mock.call('tap1234'),
+ mock_ip.return_value.exists.side_effect = port_found
+ util.delete_neutron_ports(ports)
+ mock_ip.assert_has_calls(
+ [mock.call('tap1234'),
+ mock.call().exists(),
mock.call().link.delete(),
+ mock.call('tap5678'),
+ mock.call().exists(),
mock.call('tap09ab'),
+ mock.call().exists(),
mock.call().link.delete()])
"1", "tap234")
def test_delete_bridge(self):
- bridge_device = mock.Mock()
- with mock.patch.object(ip_lib, "device_exists") as de_fn,\
+ with mock.patch.object(ip_lib.IPDevice, "exists") as de_fn,\
+ mock.patch.object(ip_lib, "IpLinkCommand") as link_cmd,\
mock.patch.object(self.lbm,
"get_interfaces_on_bridge") as getif_fn,\
mock.patch.object(self.lbm, "remove_interface"),\
"get_interface_details") as if_det_fn,\
mock.patch.object(self.lbm,
"update_interface_ip_details") as updif_fn,\
- mock.patch.object(self.lbm, "delete_interface") as del_interface,\
- mock.patch.object(bridge_lib, "BridgeDevice",
- return_value=bridge_device):
+ mock.patch.object(self.lbm, "delete_interface") as delif_fn:
de_fn.return_value = False
self.lbm.delete_bridge("br0")
self.assertFalse(getif_fn.called)
de_fn.return_value = True
getif_fn.return_value = ["eth0", "eth1", "vxlan-1002"]
if_det_fn.return_value = ("ips", "gateway")
- bridge_device.link.set_down.return_value = False
+ link_cmd.set_down.return_value = False
self.lbm.delete_bridge("br0")
updif_fn.assert_called_with("eth1", "br0", "ips", "gateway")
- del_interface.assert_called_with("vxlan-1002")
+ delif_fn.assert_called_with("vxlan-1002")
def test_delete_bridge_with_ip(self):
bridge_device = mock.Mock()
lbm = linuxbridge_neutron_agent.LinuxBridgeManager(
bridge_mappings, interface_mappings)
- bridge_device = mock.Mock()
- with mock.patch.object(ip_lib, "device_exists") as de_fn,\
+ with mock.patch.object(ip_lib.IPDevice, "exists") as de_fn,\
+ mock.patch.object(ip_lib, "IpLinkCommand") as link_cmd,\
mock.patch.object(lbm,
"get_interfaces_on_bridge") as getif_fn,\
mock.patch.object(lbm, "remove_interface"),\
- mock.patch.object(lbm, "delete_interface") as del_interface,\
- mock.patch.object(bridge_lib, "BridgeDevice",
- return_value=bridge_device):
+ mock.patch.object(lbm, "delete_interface") as del_interface:
de_fn.return_value = False
lbm.delete_bridge("br0")
self.assertFalse(getif_fn.called)
de_fn.return_value = True
getif_fn.return_value = ["vxlan-1002"]
- bridge_device.link.set_down.return_value = False
+ link_cmd.set_down.return_value = False
lbm.delete_bridge("br0")
del_interface.assert_called_with("vxlan-1002")
self.assertEqual(2, del_br_fn.call_count)
def test_remove_interface(self):
- bridge_device = mock.Mock()
- with mock.patch.object(ip_lib, "device_exists") as de_fn,\
+ with mock.patch.object(ip_lib.IPDevice, "exists") as de_fn,\
mock.patch.object(self.lbm,
"is_device_on_bridge") as isdev_fn,\
- mock.patch.object(bridge_lib, "BridgeDevice",
- return_value=bridge_device):
+ mock.patch.object(bridge_lib.BridgeDevice,
+ "delif") as delif_fn:
de_fn.return_value = False
self.assertFalse(self.lbm.remove_interface("br0", "eth0"))
self.assertFalse(isdev_fn.called)
self.assertTrue(self.lbm.remove_interface("br0", "eth0"))
isdev_fn.return_value = True
- bridge_device.delif.return_value = True
+ delif_fn.return_value = True
self.assertFalse(self.lbm.remove_interface("br0", "eth0"))
- bridge_device.delif.return_value = False
+ delif_fn.return_value = False
self.assertTrue(self.lbm.remove_interface("br0", "eth0"))
def test_delete_interface(self):
- with mock.patch.object(ip_lib, "device_exists") as de_fn,\
+ with mock.patch.object(ip_lib.IPDevice, "exists") as de_fn,\
mock.patch.object(ip_lib.IpLinkCommand, "set_down") as down_fn,\
mock.patch.object(ip_lib.IpLinkCommand, "delete") as delete_fn:
de_fn.return_value = False
self.assertFalse(int_br.drop_port.called)
def test_setup_physical_bridges(self):
- with mock.patch.object(ip_lib, "device_exists") as devex_fn,\
+ with mock.patch.object(ip_lib.IPDevice, "exists") as devex_fn,\
mock.patch.object(sys, "exit"),\
mock.patch.object(utils, "execute"),\
mock.patch.object(self.agent, 'br_phys_cls') as phys_br_cls,\
def test_setup_physical_bridges_using_veth_interconnection(self):
self.agent.use_veth_interconnection = True
- with mock.patch.object(ip_lib, "device_exists") as devex_fn,\
+ with mock.patch.object(ip_lib.IPDevice, "exists") as devex_fn,\
mock.patch.object(sys, "exit"),\
mock.patch.object(utils, "execute") as utilsexec_fn,\
mock.patch.object(self.agent, 'br_phys_cls') as phys_br_cls,\
self.mock_tun_bridge.add_port.return_value = self.INT_OFPORT
self.mock_tun_bridge.add_patch_port.return_value = self.INT_OFPORT
- self.device_exists = mock.patch.object(ip_lib, 'device_exists').start()
- self.device_exists.return_value = True
-
self.ipdevice = mock.patch.object(ip_lib, 'IPDevice').start()
self.ipwrapper = mock.patch.object(ip_lib, 'IPWrapper').start()
mock.call.setup_default_table(self.INT_OFPORT, arp_responder),
]
- self.device_exists_expected = []
-
self.ipdevice_expected = []
self.ipwrapper_expected = [mock.call()]
self.mock_map_tun_bridge_expected)
self._verify_mock_call(self.mock_tun_bridge,
self.mock_tun_bridge_expected)
- self._verify_mock_call(self.device_exists, self.device_exists_expected)
self._verify_mock_call(self.ipdevice, self.ipdevice_expected)
self._verify_mock_call(self.ipwrapper, self.ipwrapper_expected)
self._verify_mock_call(self.get_bridges, self.get_bridges_expected)
mock.call.setup_default_table(self.INT_OFPORT, arp_responder),
]
- self.device_exists_expected = [
- mock.call('int-%s' % self.MAP_TUN_BRIDGE),
- ]
-
self.ipdevice_expected = [
mock.call('int-%s' % self.MAP_TUN_BRIDGE),
+ mock.call().exists(),
+ nonzero(mock.call().exists()),
mock.call().link.delete()
]
self.ipwrapper_expected = [
Code Review / openstack-build / neutron-build.git / commitdiff