x/tobiko
Code Issues Proposed changes

Allow to select IP address by scope

Browse Source 
Change-Id: I1d4cdd2564c35fe9c16df6232dcbd689ad7feb7a
This commit is contained in:
Federico Ressi 2019-10-15 08:10:02 +02:00
parent 2b6b2ffe77
commit cd301aacd2
3 changed files with 57 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
tobiko/openstack/topology/_topology.py
View File

@ -39,10 +39,6 @@ DEFAULT_TOPOLOGY_CLASS = (
'tobiko.openstack.topology._topology.OpenStackTopology')
LOCAL_IPS = {netaddr.IPAddress('127.0.0.1'),
netaddr.IPAddress('::1')}
def get_openstack_topology(topology_class=None):
topology_class = topology_class or get_default_openstack_topology_class()
return tobiko.setup_fixture(topology_class)
@ -129,7 +125,6 @@ class OpenStackTopology(tobiko.SharedFixture):
self._nodes_by_group = collections.OrderedDict()
def setup_fixture(self):
self._unreachable_ips.update(LOCAL_IPS)
self.discover_nodes()
def cleanup_fixture(self):
@ -337,7 +332,8 @@ class OpenStackTopology(tobiko.SharedFixture):
return ip_version and int(ip_version) or None
def _ips_from_host(self, **kwargs):
return ip.list_ip_addresses(ip_version=self.ip_version, **kwargs)
return ip.list_ip_addresses(ip_version=self.ip_version,
scope='global', **kwargs)
def _ips(self, obj):
if isinstance(obj, tobiko.Selection):
@ -357,8 +353,7 @@ class OpenStackTopology(tobiko.SharedFixture):
else:
ips = tobiko.select([
netaddr.IPAddress(sockaddr[0])
for _, _, _, _, sockaddr in addrinfo
if netaddr.IPAddress(sockaddr[0]) not in LOCAL_IPS])
for _, _, _, _, sockaddr in addrinfo])
else:
for item in iter(obj):
tobiko.check_valid_type(item, netaddr.IPAddress)

23
tobiko/shell/ip.py
View File

@ -33,7 +33,7 @@ INETS = {
}
def list_ip_addresses(ip_version=None, **execute_params):
def list_ip_addresses(ip_version=None, scope=None, **execute_params):
inets = INETS.get(ip_version)
if inets is None:
error = "invalid IP version: {!r}".format(ip_version)
@ -46,11 +46,22 @@ def list_ip_addresses(ip_version=None, **execute_params):
for line in output.splitlines():
fields = line.strip().split()
inet = fields[2]
if inet in inets:
address = fields[3]
if '/' in address:
address, _ = address.split('/', 1)
ips.append(netaddr.IPAddress(address))
if inet not in inets:
continue # List only address of selected IP version
if scope:
try:
scope_index = fields.index('scope')
if fields[scope_index + 1] != scope:
continue
except (IndexError, ValueError):
continue
address = fields[3]
if '/' in address:
# Remove netmask prefix length
address, _ = address.split('/', 1)
ips.append(netaddr.IPAddress(address))
return ips

39
tobiko/tests/functional/shell/test_ip.py
View File

@ -36,13 +36,36 @@ class IpTest(testtools.TestCase):
ubuntu_stack = tobiko.required_setup_fixture(
stacks.UbuntuServerStackFixture)
def test_list_ip_addresses(self, ip_version=None, **execute_params):
ips = ip.list_ip_addresses(ip_version=ip_version, **execute_params)
def test_list_ip_addresses(self, ip_version=None, scope=None,
**execute_params):
ips = ip.list_ip_addresses(ip_version=ip_version, scope=scope,
**execute_params)
self.assertIsInstance(ips, tobiko.Selection)
for ip_address in ips:
self.assertIsInstance(ip_address, netaddr.IPAddress)
if ip_version:
self.assertEqual(ips.with_attributes(version=ip_version), ips)
if scope:
if scope == 'link':
self.assertEqual(ips.with_attributes(version=4), [])
self.assertEqual(ips.with_attributes(version=6), ips)
elif scope == 'host':
self.assertEqual(ips.with_attributes(version=4),
[netaddr.IPAddress('127.0.0.1')])
self.assertEqual(ips.with_attributes(version=6),
[netaddr.IPAddress('::1')])
elif scope == 'global':
self.assertNotIn(netaddr.IPAddress('127.0.0.1'), ips)
self.assertNotIn(netaddr.IPAddress('::1'), ips)
def test_list_ip_addresses_with_host_scope(self, **execute_params):
self.test_list_ip_addresses(scope='host', **execute_params)
def test_list_ip_addresses_with_link_scope(self, **execute_params):
self.test_list_ip_addresses(scope='link', **execute_params)
def test_list_ip_addresses_with_global_scope(self, **execute_params):
self.test_list_ip_addresses(scope='global', **execute_params)
def test_list_ip_addresses_with_ipv4(self):
self.test_list_ip_addresses(ip_version=4)
@ -65,6 +88,18 @@ class IpTest(testtools.TestCase):
self.skip('SSH proxy server not configured')
self.test_list_ip_addresses(ssh_client=ssh_client)
def test_list_ip_addresses_with_proxy_ssh_client_and_host_scope(
self, **execute_params):
self.test_list_ip_addresses(scope='host', **execute_params)
def test_list_ip_addresses_with_proxy_ssh_client_and_link_scope(
self, **execute_params):
self.test_list_ip_addresses(scope='link', **execute_params)
def test_list_ip_addresses_with_proxy_ssh_client_and_global_scope(
self, **execute_params):
self.test_list_ip_addresses(scope='global', **execute_params)
def test_list_namespaces(self, **execute_params):
namespaces = ip.list_network_namespaces(**execute_params)
self.assertIsInstance(namespaces, list)