x/tobiko
Code Issues Proposed changes

Merge "Refactor test_floating_ip using the new required_fixture_setup function"

Browse Source
This commit is contained in:
Zuul 2019-04-17 16:30:02 +00:00 committed by Gerrit Code Review
commit 176b15bb7d
2 changed files with 66 additions and 58 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
tobiko/openstack/heat/_stack.py
View File

@ -53,7 +53,6 @@ class HeatStackFixture(tobiko.SharedFixture):
template_fixture = None
parameters = None
stack = None
outputs = None
def __init__(self, stack_name=None, template=None, parameters=None,
wait_interval=None, client=None):
@ -165,7 +164,7 @@ class HeatStackFixture(tobiko.SharedFixture):
stack = self.wait_for_stack_status(
expected_status={DELETE_COMPLETE})
self.stack = self.outputs = None
self.stack = self._outputs = None
try:
LOG.debug('Creating stack %r (re-tries left %d)...',
self.stack_name, retry)
@ -191,7 +190,7 @@ class HeatStackFixture(tobiko.SharedFixture):
"""Deletes stack."""
if not stack_id:
stack_id = self.stack_id
self.stack = self.outputs = None
self.stack = self._outputs = None
try:
self.client.stacks.delete(stack_id)
except exc.NotFound:
@ -217,7 +216,7 @@ class HeatStackFixture(tobiko.SharedFixture):
except exc.HTTPNotFound:
self.stack = stack = None
finally:
self.outputs = None
self._outputs = None
return stack
def wait_for_create_complete(self, check=True):
@ -247,20 +246,29 @@ class HeatStackFixture(tobiko.SharedFixture):
check_stack_status(stack, expected_status)
return stack
_outputs = None
@property
def outputs(self):
return self.wait_for_outputs()
def get_outputs(self):
stack = self.stack
if not hasattr(stack, 'outputs'):
stack = self.get_stack(resolve_outputs=True)
check_stack_status(stack, {CREATE_COMPLETE})
self.outputs = outputs = HeatStackOutputs(
self._outputs = outputs = HeatStackOutputs(
stack_name=self.stack_name,
outputs={output['output_key']: output['output_value']
for output in stack.outputs})
return outputs
def wait_for_outputs(self):
self.wait_for_create_complete()
return self.get_outputs()
if self._outputs:
return self._outputs
else:
self.wait_for_create_complete()
return self.get_outputs()
class HeatStackOutputs(object):

102
tobiko/tests/scenario/neutron/test_floating_ip.py
View File

@ -29,65 +29,70 @@ CONF = config.CONF
class FloatingIPFixture(heat.HeatStackFixture):
template = stacks.heat_template_file('floating_ip.yaml')
# template parameters
# --- template parameters ---
floating_network = CONF.tobiko.neutron.floating_network
image = CONF.tobiko.nova.image
flavor = CONF.tobiko.nova.flavor
internal_network_fixture = stacks.InternalNetworkFixture
@property
def internal_network(self):
return self.internal_network_stack.outputs.network_id
# --- required fixtures ---
internal_network_stack = tobiko.required_setup_fixture(
stacks.InternalNetworkFixture)
# --- class parameters ---
# derived attributes
internal_network = None
port_security_enabled = False
security_groups = None
def setup_parameters(self):
super(FloatingIPFixture, self).setup_parameters()
self.setup_internal_network()
if self.port_security_enabled:
self.setup_port_security()
def setup_internal_network(self):
self.internal_network = tobiko.setup_fixture(
self.internal_network_fixture).wait_for_outputs()
self.parameters['internal_network'] = self.internal_network.network_id
@neutron.skip_if_missing_networking_extensions('port-security')
def setup_port_security(self):
self.parameters.update(
port_security_enabled=self.port_security_enabled,
security_groups=self.security_groups or [])
# --- outputs ---
@property
def floating_ip_address(self):
return self.outputs.floating_ip_address
@property
def internal_network_mtu(self):
return self.internal_network_stack.outputs.mtu
class FloatingIPTest(base.TobikoTest):
"""Tests server connectivity"""
floating_ip_fixture = FloatingIPFixture
floating_ip_stack = tobiko.required_setup_fixture(FloatingIPFixture)
@classmethod
def setUpClass(cls):
super(FloatingIPTest, cls).setUpClass()
stack = tobiko.setup_fixture(cls.floating_ip_fixture)
outputs = stack.wait_for_outputs()
cls.floating_ip_address = outputs.floating_ip_address
cls.mtu = stack.internal_network.mtu
def setUp(self):
super(FloatingIPTest, self).setUp()
stack = self.floating_ip_stack
self.floating_ip_address = stack.floating_ip_address
self.internal_network_mtu = stack.internal_network_mtu
def test_ping(self):
ping.ping_until_received(self.floating_ip_address).assert_replied()
@neutron.skip_if_missing_networking_extensions('net-mtu')
def test_ping_with_mtu_packet(self):
def test_ping_with_mtu(self):
ping.ping_until_received(self.floating_ip_address,
packet_size=self.mtu,
packet_size=self.internal_network_mtu,
fragmentation=False).assert_replied()
@neutron.skip_if_missing_networking_extensions('net-mtu')
def test_ping_with_oversized_packet(self):
# Wait for VM to get ready
ping.ping_until_received(self.floating_ip_address)
# Send 5 over-sized packets
ping.ping(self.floating_ip_address, packet_size=self.mtu + 1,
ping.ping(self.floating_ip_address,
packet_size=self.internal_network_mtu + 1,
fragmentation=False, count=5,
check=False).assert_not_replied()
@ -98,22 +103,20 @@ class FloatingIPWithPortSecurityFixture(FloatingIPFixture):
class FloatingIPWithPortSecurityTest(base.TobikoTest):
floating_ip_fixture = FloatingIPFixture
floating_ip_with_securtity_fixture = FloatingIPWithPortSecurityFixture
floating_ip_stack = tobiko.required_setup_fixture(FloatingIPFixture)
floating_ip_with_securtity_stack = tobiko.required_setup_fixture(
FloatingIPWithPortSecurityFixture)
@classmethod
def setUpClass(cls):
super(FloatingIPWithPortSecurityTest, cls).setUpClass()
def setUp(self):
super(FloatingIPWithPortSecurityTest, self).setUp()
# Setup VM with port security
stack = tobiko.setup_fixture(cls.floating_ip_with_securtity_fixture)
outputs = stack.wait_for_outputs()
cls.floating_ip_address_with_security = outputs.floating_ip_address
self.floating_ip_address_with_security = (
self.floating_ip_with_securtity_stack.outputs.floating_ip_address)
# Setup VM without port security
stack = tobiko.setup_fixture(cls.floating_ip_fixture)
outputs = stack.wait_for_outputs()
cls.floating_ip_address = outputs.floating_ip_address
self.floating_ip_address = (
self.floating_ip_stack.outputs.floating_ip_address)
def test_ping(self):
ping.ping_until_received(self.floating_ip_address).assert_replied()
@ -122,27 +125,24 @@ class FloatingIPWithPortSecurityTest(base.TobikoTest):
class FloatingIPWithSecurityGroupFixture(FloatingIPWithPortSecurityFixture):
security_groups_fixture = stacks.SecurityGroupsFixture
security_groups = None
security_groups_stack = tobiko.required_setup_fixture(
stacks.SecurityGroupsFixture)
def setup_parameters(self):
super(FloatingIPWithSecurityGroupFixture, self).setup_parameters()
self.setup_security_groups()
self.parameters['security_groups'] = [
self.security_groups.icmp_security_group_id]
def setup_security_groups(self):
self.security_groups = tobiko.setup_fixture(
self.security_groups_fixture).wait_for_outputs()
@property
def security_groups(self):
return [self.security_groups_stack.outputs.icmp_security_group_id]
class FloatingIPWithSecurityGroupTest(FloatingIPTest):
floating_ip_fixture = FloatingIPWithSecurityGroupFixture
floating_ip_stack = tobiko.required_setup_fixture(
FloatingIPWithSecurityGroupFixture)
class FloatingIPWithNetMtuWritableFixture(FloatingIPFixture):
internal_network_fixture = stacks.InternalNetworkWithNetMtuWritableFixture
internal_network_stack = tobiko.required_setup_fixture(
stacks.InternalNetworkWithNetMtuWritableFixture)
class FlatingIpWithMtuWritableTest(FloatingIPTest):
floating_ip_fixture = FloatingIPWithNetMtuWritableFixture
floating_ip_stack = tobiko.required_setup_fixture(
FloatingIPWithNetMtuWritableFixture)