Switch IPDevice.exists() method to use pyroute2

Check if network device exists is now done by checking
interface index with pyroute2 interface instead of checking
if MAC address for device is set.

Change-Id: I2d5b95ec109fb19fc2a46c1017959f74011b9a22
Related-Bug: #1492714
This commit is contained in:
Sławek Kapłoński 2018-03-16 15:58:10 +01:00
parent 3ff872b913
commit aaf11f45ec
6 changed files with 37 additions and 35 deletions

View File

@ -294,15 +294,7 @@ class IPDevice(SubProcessBase):
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:
return bool(self.link.address)
except RuntimeError:
return False
finally:
self.set_log_fail_as_error(orig_log_fail_as_error)
return privileged.interface_exists(self.name, self.namespace)
def delete_addr_and_conntrack_state(self, cidr):
"""Delete an address along with its conntrack state

View File

@ -16,7 +16,7 @@ from neutron_lib.utils import helpers
from oslo_config import cfg
from oslo_log import log as logging
from neutron.common import config
from neutron.conf.agent import common as config
from neutron.plugins.ml2.drivers.linuxbridge.agent \
import linuxbridge_neutron_agent
@ -69,4 +69,5 @@ def main():
"""
cfg.CONF(sys.argv[1:])
config.setup_logging()
config.setup_privsep()
remove_empty_bridges()

View File

@ -194,6 +194,19 @@ def delete_interface(ifname, namespace, **kwargs):
_run_iproute_link("del", ifname, namespace, **kwargs)
@privileged.default.entrypoint
def interface_exists(ifname, namespace):
try:
idx = _get_link_id(ifname, namespace)
return bool(idx)
except NetworkInterfaceNotFound:
return False
except OSError as e:
if e.errno == errno.ENOENT:
return False
raise
@privileged.default.entrypoint
def add_neigh_entry(ip_version, ip_address, mac_address, device, namespace,
**kwargs):

View File

@ -112,6 +112,14 @@ def unstable_test(reason):
return decor
def get_rootwrap_cmd():
return os.environ.get('OS_ROOTWRAP_CMD', SUDO_CMD)
def get_rootwrap_daemon_cmd():
return os.environ.get('OS_ROOTWRAP_DAEMON_CMD')
class AttributeDict(dict):
"""
@ -424,10 +432,9 @@ class BaseTestCase(DietTestCase):
def setup_rootwrap(self):
agent_config.register_root_helper(cfg.CONF)
self.config(group='AGENT',
root_helper=os.environ.get('OS_ROOTWRAP_CMD', SUDO_CMD))
root_helper=get_rootwrap_cmd())
self.config(group='AGENT',
root_helper_daemon=os.environ.get(
'OS_ROOTWRAP_DAEMON_CMD'))
root_helper_daemon=get_rootwrap_daemon_cmd())
class PluginFixture(fixtures.Fixture):

View File

@ -20,6 +20,7 @@ from neutron_lib import constants
from neutron.agent.linux import ip_lib
from neutron.plugins.ml2.drivers.linuxbridge.agent import \
linuxbridge_neutron_agent as lb_agent
from neutron.tests import base as tests_base
from neutron.tests.common import config_fixtures
from neutron.tests.common import net_helpers
from neutron.tests.functional import base
@ -35,6 +36,16 @@ class LinuxbridgeCleanupTest(base.BaseSudoTestCase):
prefix=lb_agent.BRIDGE_NAME_PREFIX))).fixture
config = callback(br_fixture)
# NOTE(slaweq): use of oslo.privsep inside neutron-linuxbridge-cleanup
# script requires rootwrap helper to be configured in this script's
# config
config.update({
'AGENT': {
'root_helper': tests_base.get_rootwrap_cmd(),
'root_helper_daemon': tests_base.get_rootwrap_daemon_cmd()
}
})
config.update({'VXLAN': {'enable_vxlan': 'False'}})
temp_dir = self.useFixture(fixtures.TempDir()).path

View File

@ -1383,28 +1383,6 @@ class TestIpNetnsCommand(TestIPCmdBase):
class TestDeviceExists(base.BaseTestCase):
def test_device_exists(self):
with mock.patch('neutron.agent.common.utils.execute') as execute:
execute.return_value = LINK_SAMPLE[1]
self.assertTrue(ip_lib.device_exists('eth0'))
execute.assert_called_once_with(
['ip', '-o', 'link', 'show', 'eth0'],
run_as_root=False, 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 = ''
_execute.side_effect = RuntimeError
self.assertFalse(ip_lib.device_exists('eth0'))
def test_ensure_device_is_ready(self):
ip_lib_mock = mock.Mock()
with mock.patch.object(ip_lib, 'IPDevice', return_value=ip_lib_mock):