Browse Source

Share stack across tests

Less time is wasted on creating stacks for each separate
test.

Also, updated requirements.

Change-Id: I99c998dd7706300de5f96ea84698098655785a22
changes/32/608432/4
abregman 3 years ago
parent
commit
0cf6dfb5c3
  1. 10
      README.rst
  2. 14
      requirements.txt
  3. 3
      tobiko/common/clients.py
  4. 3
      tobiko/config.py
  5. 1
      tobiko/plugin.py
  6. 25
      tobiko/tests/scenario/base.py
  7. 85
      tobiko/tests/scenario/templates/scenario.yaml
  8. 23
      tobiko/tests/scenario/test_floatingip.py

10
README.rst

@ -2,10 +2,8 @@
Tobiko
======
To run pre-upgrade tests:
Tempest plugin for testing upgrades
tempest run --regex pre
To run post-upgrade tests:
tempest run --regex post
* Free software: Apache license
* Source: https://git.openstack.org/cgit/openstack/tobiko
* Bugs: https://bugs.launchpad.net/tobiko

14
requirements.txt

@ -1,7 +1,7 @@
keystoneauth1
oslo.config
oslo.log
python-heatclient
testscenarios
testtools
tempest
keystoneauth1>=3.4.0
oslo.config>=5.2.0
oslo.log>=3.36.0
python-heatclient>=1.10.0
testscenarios>=0.4
testtools>=2.2.0
tempest>=17.1.0

3
tobiko/common/clients.py

@ -105,9 +105,6 @@ class ClientManager(object):
{'user_domain_name': self.get_user_domain_name(),
'project_domain_name': self.get_project_domain_name()})
with open('/tmp/stam', 'w+') as f:
f.write(str(kwargs))
loader = loading.get_plugin_loader('password')
auth = loader.load_from_options(**kwargs)
return session.Session(auth=auth, verify=False)

3
tobiko/config.py

@ -25,6 +25,9 @@ TobikoGroup = [
help="Floating network name "),
cfg.StrOpt('admin_username',
help="Username to use for admin API requests."),
cfg.StrOpt('user_domain_name',
default="Default",
help="User domain name")
]
tobiko_group = cfg.OptGroup(name="tobiko_plugin",

1
tobiko/plugin.py

@ -11,7 +11,6 @@
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import os
from tempest import config

25
tobiko/tests/scenario/base.py

@ -12,10 +12,12 @@
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import os
import subprocess
from heatclient import exc
from tempest.common.utils import net_utils
from tempest.lib.common.utils import test_utils
@ -35,6 +37,27 @@ class ScenarioTestsBase(base.TobikoTest):
self.stackManager = stack.StackManager(self.clientManager,
templates_dir)
try:
self.stackManager.get_stack("scenario")
except exc.HTTPNotFound:
self.create_stack()
def create_stack(self):
"""Creates stack to be used by all scenario tests."""
# Defines parameters required by heat template
parameters = {'public_net': self.conf.network.floating_network_name,
'image': self.conf.compute.image_ref,
'flavor': "m1.micro"}
# creates stack and stores its ID
st = self.stackManager.create_stack(stack_name="scenario",
template_name="scenario.yaml",
parameters=parameters)
sid = st['stack']['id']
self.stackManager.wait_for_status_complete(sid, 'floating_ip')
def ping_ip_address(self, ip_address, should_succeed=True,
ping_timeout=None, mtu=None):

85
tobiko/tests/scenario/templates/scenario.yaml

@ -0,0 +1,85 @@
heat_template_version: 2013-05-23
description: |
Template to create an instance and check connectivity to it
parameters:
flavor:
type: string
image:
type: string
subnet_cidr:
type: string
default: 190.40.2.0/24
public_net:
type: string
default: public
private_net:
type: string
default: heat-net
dns_servers:
type: comma_delimited_list
default: ["8.8.8.8", "8.8.4.4"]
resources:
sg:
type: OS::Neutron::SecurityGroup
properties:
name: sg
description: Security group to allow ICMP and SSH
rules:
- protocol: icmp
- protocol: tcp
port_range_min: 22
port_range_max: 22
floating_ip:
type: OS::Neutron::FloatingIP
properties:
floating_network: {get_param: public_net}
network:
type: OS::Neutron::Net
subnet:
type: OS::Neutron::Subnet
properties:
network: {get_resource: network}
ip_version: 4
cidr: {get_param: subnet_cidr}
dns_nameservers: {get_param: dns_servers}
router:
type: OS::Neutron::Router
properties:
external_gateway_info:
network: {get_param: public_net}
router_interface:
type: OS::Neutron::RouterInterface
properties:
router: {get_resource: router}
subnet: {get_resource: subnet}
wait_handle:
type: OS::Heat::WaitConditionHandle
server:
type: OS::Nova::Server
properties:
image: {get_param: image}
flavor: {get_param: flavor}
networks:
- subnet: {get_resource: subnet}
security_groups:
- {get_resource: sg}
server_floating_ip_assoc:
type: OS::Neutron::FloatingIPAssociation
properties:
floatingip_id: {get_resource: floating_ip}
port_id: {get_attr: [server, addresses, {get_resource: network}, 0, port]}
outputs:
server_ip:
value: {get_attr: [floating_ip, floating_ip_address]}

23
tobiko/tests/scenario/test_floatingip.py

@ -19,24 +19,10 @@ class FloatingIPTest(base.ScenarioTestsBase):
"""Tests server connectivity"""
def test_pre_fip(self):
"""Creates a server and checks it can reach it."""
# Defines parameters required by heat template
parameters = {'public_net': self.conf.network.floating_network_name,
'image': self.conf.compute.image_ref,
'flavor': "m1.micro"}
# creates stack and stores its ID
st = self.stackManager.create_stack(stack_name="fip",
template_name="fip.yaml",
parameters=parameters)
sid = st['stack']['id']
# Before pinging the floating IP, ensure resource is ready
self.stackManager.wait_for_status_complete(sid, 'floating_ip')
"""Validates connectivity to a server created by another test."""
# Get floating IP address
stack = self.stackManager.client.stacks.get(sid)
stack = self.stackManager.get_stack(stack_name="scenario")
server_fip = stack.outputs[0]['output_value']
# Check if instance is reachable
@ -44,9 +30,10 @@ class FloatingIPTest(base.ScenarioTestsBase):
self.fail("IP address is not reachable: %s" % server_fip)
def test_post_fip(self):
"""Validates connectivity to a server created by another test."""
"""Validates connectivity to a server post upgrade."""
stack = self.stackManager.get_stack(stack_name="fip")
# Get floating IP address
stack = self.stackManager.get_stack(stack_name="scenario")
server_fip = stack.outputs[0]['output_value']
# Check if instance is reachable

Loading…
Cancel
Save