x/tobiko
Code Issues Proposed changes

Fix tests to check network network stack subnet gateway ips.

Browse Source 
Change-Id: I757e2252db53e874cd8fb27f0dffa37922cc1f28
This commit is contained in:
Federico Ressi 2019-06-18 17:37:03 +02:00
parent ceb5420cf4
commit 6446c8e03b
5 changed files with 73 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
tobiko/openstack/_find.py
View File

@ -17,20 +17,23 @@ from __future__ import absolute_import
import tobiko
def find_resource(obj, resources, resource_type, properties=None):
resources = list(find_resources(obj, resources, properties=properties))
def find_resource(obj, resources, resource_type, properties=None, **params):
if obj:
resources = list(find_resources(obj, resources, properties=properties))
count = len(resources)
if count == 0:
raise ResourceNotFound(obj=obj,
resource_type=resource_type,
properties=properties)
properties=properties,
params=params)
if count > 1:
resource_ids = [r['id'] for r in resources]
raise MultipleResourcesFound(obj=obj,
resource_type=resource_type,
properties=properties,
count=len(resources),
resource_ids=resource_ids)
resource_ids=resource_ids,
params=params)
return resources[0]
@ -38,16 +41,18 @@ def find_resources(obj, resources, properties=None):
properties = properties or ('id', 'name')
for resource in resources:
for property_name in properties:
if obj == resource[property_name]:
value = resource[property_name]
if obj == value:
yield resource
break
class ResourceNotFound(tobiko.TobikoException):
message = ("No such {resource_type} found for {obj!r} in "
"properties {properties!r}")
message = ("No such {resource_type} found for obj={obj!r}, "
"properties={properties!r} and params={params!r}")
class MultipleResourcesFound(tobiko.TobikoException):
message = ("{count} {resource_type}d found for {obj!r} in "
"properties {properties!r}: {resource_ids}")
message = ("{count} {resource_type}s found for obj={obj!r}, "
"properties={properties!r} and params={params!r}: "
"{resource_ids}")

3
tobiko/openstack/neutron/__init__.py
View File

@ -24,10 +24,13 @@ NeutronClientFixture = _client.NeutronClientFixture
find_network = _client.find_network
list_networks = _client.list_networks
find_subnet = _client.find_subnet
find_port = _client.find_port
list_ports = _client.list_ports
list_subnets = _client.list_subnets
list_subnet_cidrs = _client.list_subnet_cidrs
show_network = _client.show_network
show_router = _client.show_router
show_port = _client.show_port
show_subnet = _client.show_subnet
new_ipv4_cidr = _cidr.new_ipv4_cidr

26
tobiko/openstack/neutron/_client.py
View File

@ -62,18 +62,25 @@ def get_neutron_client(session=None, shared=True, init_client=None,
return client.client
def find_network(obj, properties=None, client=None, **params):
def find_network(obj=None, properties=None, client=None, **params):
"""Look for the unique network matching some property values"""
return _find.find_resource(
obj=obj, resource_type='network', properties=properties,
resources=list_networks(client=client, **params))
resources=list_networks(client=client, **params), **params)
def find_subnet(obj, properties=None, client=None, **params):
def find_port(obj=None, properties=None, client=None, **params):
"""Look for the unique network matching some property values"""
return _find.find_resource(
obj=obj, resource_type='port', properties=properties,
resources=list_ports(client=client, **params), **params)
def find_subnet(obj=None, properties=None, client=None, **params):
"""Look for the unique subnet matching some property values"""
return _find.find_resource(
obj=obj, resource_type='subnet', properties=properties,
resources=list_subnets(client=client, **params))
resources=list_subnets(client=client, **params), **params)
def list_networks(show=False, client=None, **params):
@ -83,6 +90,13 @@ def list_networks(show=False, client=None, **params):
return networks
def list_ports(show=False, client=None, **params):
ports = neutron_client(client).list_ports(**params)['ports']
if show:
ports = [show_port(p['id'], client=client) for p in ports]
return ports
def list_subnets(show=False, client=None, **params):
subnets = neutron_client(client).list_subnets(**params)
if isinstance(subnets, collections.Mapping):
@ -101,6 +115,10 @@ def show_network(network, client=None, **params):
return neutron_client(client).show_network(network, **params)['network']
def show_port(port, client=None, **params):
return neutron_client(client).show_port(port, **params)['port']
def show_router(router, client=None, **params):
return neutron_client(client).show_router(router, **params)['router']

18
tobiko/openstack/stacks/_neutron.py
View File

@ -102,6 +102,24 @@ class NetworkStackFixture(heat.HeatStackFixture):
def gateway_details(self):
return neutron.show_router(self.gateway_id)
@property
def ipv4_gateway_port_details(self):
return neutron.find_port(
[{'subnet_id': self.ipv4_subnet_id,
'ip_address': self.ipv4_subnet_details['gateway_ip']}],
properties=['fixed_ips'],
device_id=self.gateway_id,
network_id=self.network_id)
@property
def ipv6_gateway_port_details(self):
return neutron.find_port(
[{'subnet_id': self.ipv6_subnet_id,
'ip_address': self.ipv6_subnet_details['gateway_ip']}],
properties=['fixed_ips'],
device_id=self.gateway_id,
network_id=self.network_id)
@property
def gateway_network_details(self):
return neutron.show_network(self.gateway_network_id)

30
tobiko/tests/functional/openstack/test_stacks.py
View File

@ -59,20 +59,6 @@ class NetworkTestCase(testtools.TestCase):
self.assertEqual(neutron.show_subnet(self.stack.ipv6_subnet_id),
subnet)
def test_ipv4_subnet_gateway_ip(self):
if not self.stack.has_ipv4 or self.stack.has_gateway:
tobiko.skip('Stack {!s} has no IPv4 gateway',
self.stack.stack_name)
self.assertEqual(str(self.stack.ipv4_cidr.ip + 1),
self.stack.ipv4_subnet_details['gateway_ip'])
def test_ipv6_subnet_gateway_ip(self):
if not self.stack.has_ipv6 or self.stack.has_gateway:
tobiko.skip('Stack {!s} has no IPv6 gateway',
self.stack.stack_name)
self.assertEqual(str(self.stack.ipv6_cidr.ip + 1),
self.stack.ipv6_subnet_details['gateway_ip'])
def test_gateway_network(self):
if not self.stack.has_gateway:
tobiko.skip('Stack {!s} has no gateway',
@ -81,6 +67,22 @@ class NetworkTestCase(testtools.TestCase):
self.stack.gateway_network_id,
self.stack.gateway_details['external_gateway_info']['network_id'])
def test_ipv4_gateway_ip(self):
if not self.stack.has_ipv4 or not self.stack.has_gateway:
tobiko.skip('Stack {!s} has no IPv4 gateway',
self.stack.stack_name)
self.assertEqual(
self.stack.ipv4_gateway_port_details['fixed_ips'][0]['ip_address'],
self.stack.ipv4_subnet_details['gateway_ip'])
def test_ipv6_gateway_ip(self):
if not self.stack.has_ipv6 or not self.stack.has_gateway:
tobiko.skip('Stack {!s} has no IPv6 gateway',
self.stack.stack_name)
self.assertEqual(
self.stack.ipv6_gateway_port_details['fixed_ips'][0]['ip_address'],
self.stack.ipv6_subnet_details['gateway_ip'])
@neutron.skip_if_missing_networking_extensions('net-mtu-write')
class NetworkWithNetMtuWriteTestCase(NetworkTestCase):