openstack
/
neutron
Code Issues Proposed changes

Merge "Add broadcast to IpAddrCommand.get_devices_with_ip() return value"

This commit is contained in:
Zuul 2018-03-12 16:45:14 +00:00 committed by Gerrit Code Review
parent c4d4336e9f a573711ce8
commit c542ec801a
3 changed files with 64 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
neutron/agent/linux/ip_lib.py
View File

@ -625,10 +625,12 @@ class IpAddrCommand(IpDeviceCommandBase):
continue
parts = line.split(" ")
broadcast = None
if parts[0] == 'inet6':
scope = parts[3]
else:
if parts[2] == 'brd':
broadcast = parts[3]
scope = parts[5]
else:
scope = parts[3]
@ -636,6 +638,7 @@ class IpAddrCommand(IpDeviceCommandBase):
retval.append(dict(name=device_name,
cidr=parts[1],
scope=scope,
broadcast=broadcast,
dynamic=('dynamic' == parts[-1]),
tentative=('tentative' in line),
dadfailed=('dadfailed' == parts[-1])))

32
neutron/tests/functional/agent/linux/test_ip_lib.py
View File

@ -383,11 +383,21 @@ class IpLibTestCase(IpLibTestFramework):
self.assertEqual(alias, device.link.alias)
def _add_and_check_ips(self, device, ip_addresses):
for cidr, scope in ip_addresses:
device.addr.add(str(cidr), scope)
for cidr, scope, expected_broadcast in ip_addresses:
# For IPv4 address add_broadcast flag will be set to True only
# if expected_broadcast is given.
# For IPv6 add_broadcast flag can be set to True always but
# broadcast address will not be set, so expected_broadcast for
# IPv6 should be always given as None.
add_broadcast = True
if cidr.version == constants.IP_VERSION_4:
add_broadcast = bool(expected_broadcast)
device.addr.add(str(cidr), scope, add_broadcast)
device_ips_info = [
(ip_info['cidr'], ip_info['scope']) for
(netaddr.IPNetwork(ip_info['cidr']),
ip_info['scope'],
ip_info['broadcast']) for
ip_info in device.addr.list()]
self.assertItemsEqual(ip_addresses, device_ips_info)
@ -399,10 +409,10 @@ class IpLibTestCase(IpLibTestFramework):
def test_add_ip_address(self):
ip_addresses = [
("10.10.10.10/30", "global"),
("11.11.11.11/28", "link"),
("2801::1/120", "global"),
("fe80::/64", "link")]
(netaddr.IPNetwork("10.10.10.10/30"), "global", '10.10.10.11'),
(netaddr.IPNetwork("11.11.11.11/28"), "link", None),
(netaddr.IPNetwork("2801::1/120"), "global", None),
(netaddr.IPNetwork("fe80::/64"), "link", None)]
attr = self.generate_device_details(ip_cidrs=[])
device = self.manage_device(attr)
self._add_and_check_ips(device, ip_addresses)
@ -421,10 +431,10 @@ class IpLibTestCase(IpLibTestFramework):
def test_flush_ip_addresses(self):
ip_addresses = [
("10.10.10.10/30", "global"),
("11.11.11.11/28", "link"),
("2801::1/120", "global"),
("fe80::/64", "link")]
(netaddr.IPNetwork("10.10.10.10/30"), "global", '10.10.10.11'),
(netaddr.IPNetwork("11.11.11.11/28"), "link", None),
(netaddr.IPNetwork("2801::1/120"), "global", None),
(netaddr.IPNetwork("fe80::/64"), "link", None)]
attr = self.generate_device_details(ip_cidrs=[])
device = self.manage_device(attr)

58
neutron/tests/unit/agent/linux/test_ip_lib.py
View File

@ -895,27 +895,41 @@ class TestIpAddrCommand(TestIPCmdBase):
6, self.parent.name, self.addr_cmd._parent.namespace)
def test_list(self):
expected = [
expected_brd = [
dict(name='eth0', scope='global', tentative=False, dadfailed=False,
dynamic=False, cidr='172.16.77.240/24',
broadcast='172.16.77.255')]
expected_no_brd = [
dict(name='eth0', scope='global', tentative=False, dadfailed=False,
dynamic=False, cidr='172.16.77.240/24', broadcast=None)]
expected_ipv6 = [
dict(name='eth0', scope='global', dadfailed=False, tentative=False,
dynamic=False, cidr='172.16.77.240/24'),
dict(name='eth0', scope='global', dadfailed=False, tentative=False,
dynamic=True, cidr='2001:470:9:1224:5595:dd51:6ba2:e788/64'),
dynamic=True, cidr='2001:470:9:1224:5595:dd51:6ba2:e788/64',
broadcast=None),
dict(name='eth0', scope='link', dadfailed=False, tentative=True,
dynamic=False, cidr='fe80::3023:39ff:febc:22ae/64'),
dynamic=False, cidr='fe80::3023:39ff:febc:22ae/64',
broadcast=None),
dict(name='eth0', scope='link', dadfailed=True, tentative=True,
dynamic=False, cidr='fe80::3023:39ff:febc:22af/64'),
dynamic=False, cidr='fe80::3023:39ff:febc:22af/64',
broadcast=None),
dict(name='eth0', scope='global', dadfailed=False, tentative=False,
dynamic=True, cidr='2001:470:9:1224:fd91:272:581e:3a32/64'),
dynamic=True, cidr='2001:470:9:1224:fd91:272:581e:3a32/64',
broadcast=None),
dict(name='eth0', scope='global', dadfailed=False, tentative=False,
dynamic=True, cidr='2001:470:9:1224:4508:b885:5fb:740b/64'),
dynamic=True, cidr='2001:470:9:1224:4508:b885:5fb:740b/64',
broadcast=None),
dict(name='eth0', scope='global', dadfailed=False, tentative=False,
dynamic=True, cidr='2001:470:9:1224:dfcc:aaff:feb9:76ce/64'),
dynamic=True, cidr='2001:470:9:1224:dfcc:aaff:feb9:76ce/64',
broadcast=None),
dict(name='eth0', scope='link', dadfailed=False, tentative=False,
dynamic=False, cidr='fe80::dfcc:aaff:feb9:76ce/64')]
dynamic=False, cidr='fe80::dfcc:aaff:feb9:76ce/64',
broadcast=None)]
test_cases = [ADDR_SAMPLE, ADDR_SAMPLE2]
cases = [
(ADDR_SAMPLE, expected_brd + expected_ipv6),
(ADDR_SAMPLE2, expected_no_brd + expected_ipv6)]
for test_case in test_cases:
for test_case, expected in cases:
self.parent._run = mock.Mock(return_value=test_case)
self.assertEqual(expected, self.addr_cmd.list())
self._assert_call([], ('show', 'tap0'))
@ -941,17 +955,24 @@ class TestIpAddrCommand(TestIPCmdBase):
wait_time=1)
def test_list_filtered(self):
expected = [
expected_brd = [
dict(name='eth0', scope='global', tentative=False, dadfailed=False,
dynamic=False, cidr='172.16.77.240/24')]
dynamic=False, cidr='172.16.77.240/24',
broadcast='172.16.77.255')]
expected_no_brd = [
dict(name='eth0', scope='global', tentative=False, dadfailed=False,
dynamic=False, cidr='172.16.77.240/24', broadcast=None)]
test_cases = [ADDR_SAMPLE, ADDR_SAMPLE2]
cases = [
(ADDR_SAMPLE, expected_brd), (ADDR_SAMPLE2, expected_no_brd)]
for test_case in test_cases:
for test_case, expected in cases:
output = '\n'.join(test_case.split('\n')[0:4])
self.parent._run.return_value = output
self.assertEqual(self.addr_cmd.list('global',
filters=['permanent']), expected)
self.assertEqual(
expected,
self.addr_cmd.list(
'global', filters=['permanent']))
self._assert_call([], ('show', 'tap0', 'permanent', 'scope',
'global'))
@ -964,6 +985,7 @@ class TestIpAddrCommand(TestIPCmdBase):
devices = self.addr_cmd.get_devices_with_ip(to='172.16.77.240/24')
self.assertEqual(1, len(devices))
expected = {'cidr': '172.16.77.240/24',
'broadcast': '172.16.77.255',
'dadfailed': False,
'dynamic': False,
'name': 'eth0',