arif/zaza-openstack-tests
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Move useful helpers from Neutron tests to OpenStackBaseTest

Browse Source 
The dual-instance launch and retrieve is a common pattern in tests.

Let's share them.
This commit is contained in:
Frode Nordahl
2020-07-02 13:33:03 +02:00
parent 520830905b
commit 0f3d9bf7c4
2 changed files with 99 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

54
zaza/openstack/charm_tests/neutron/tests.py
View File

@@ -25,10 +25,7 @@ import logging
import tenacity
import unittest
import novaclient
import zaza
import zaza.openstack.charm_tests.glance.setup as glance_setup
import zaza.openstack.charm_tests.nova.utils as nova_utils
import zaza.openstack.charm_tests.test_utils as test_utils
import zaza.openstack.configure.guest as guest
@@ -608,7 +605,7 @@ class NeutronOpenvSwitchTest(NeutronPluginApiSharedTests):
logging.info('Testing pause resume')
class NeutronNetworkingBase(unittest.TestCase):
class NeutronNetworkingBase(test_utils.OpenStackBaseTest):
"""Base for checking openstack instances have valid networking."""
RESOURCE_PREFIX = 'zaza-neutrontests'
@@ -616,10 +613,7 @@ class NeutronNetworkingBase(unittest.TestCase):
@classmethod
def setUpClass(cls):
"""Run class setup for running Neutron API Networking tests."""
cls.keystone_session = (
openstack_utils.get_overcloud_keystone_session())
cls.nova_client = (
openstack_utils.get_nova_session_client(cls.keystone_session))
super(NeutronNetworkingBase, cls).setUpClass()
cls.neutron_client = (
openstack_utils.get_neutron_session_client(cls.keystone_session))
# NOTE(fnordahl): in the event of a test failure we do not want to run
@@ -737,44 +731,6 @@ class NeutronNetworkingBase(unittest.TestCase):
assert agent['admin_state_up']
assert agent['alive']
def launch_guests(self):
"""Launch two guests to use in tests."""
guest.launch_instance(
glance_setup.LTS_IMAGE_NAME,
vm_name='{}-ins-1'.format(self.RESOURCE_PREFIX))
guest.launch_instance(
glance_setup.LTS_IMAGE_NAME,
vm_name='{}-ins-2'.format(self.RESOURCE_PREFIX))
def retrieve_guest(self, nova_client, guest_name):
"""Return guest matching name.
:param nova_client: Nova client to use when checking status
:type nova_client: Nova client
:returns: the matching guest
:rtype: Union[novaclient.Server, None]
"""
try:
return nova_client.servers.find(name=guest_name)
except novaclient.exceptions.NotFound:
return None
def retrieve_guests(self, nova_client):
"""Return test guests.
:param nova_client: Nova client to use when checking status
:type nova_client: Nova client
:returns: the matching guest
:rtype: Union[novaclient.Server, None]
"""
instance_1 = self.retrieve_guest(
nova_client,
'{}-ins-1'.format(self.RESOURCE_PREFIX))
instance_2 = self.retrieve_guest(
nova_client,
'{}-ins-1'.format(self.RESOURCE_PREFIX))
return instance_1, instance_2
def check_connectivity(self, instance_1, instance_2):
"""Run North/South and East/West connectivity tests."""
def verify(stdin, stdout, stderr):
@@ -845,7 +801,7 @@ class NeutronNetworkingTest(NeutronNetworkingBase):
def test_instances_have_networking(self):
"""Validate North/South and East/West networking."""
self.launch_guests()
instance_1, instance_2 = self.retrieve_guests(self.nova_client)
instance_1, instance_2 = self.retrieve_guests()
self.check_connectivity(instance_1, instance_2)
self.run_tearDown = True
@@ -855,10 +811,10 @@ class NeutronNetworkingVRRPTests(NeutronNetworkingBase):
def test_gateway_failure(self):
"""Validate networking in the case of a gateway failure."""
instance_1, instance_2 = self.retrieve_guests(self.nova_client)
instance_1, instance_2 = self.retrieve_guests()
if not all([instance_1, instance_2]):
self.launch_guests()
instance_1, instance_2 = self.retrieve_guests(self.nova_client)
instance_1, instance_2 = self.retrieve_guests()
self.check_connectivity(instance_1, instance_2)
routers = self.neutron_client.list_routers(

94
zaza/openstack/charm_tests/test_utils.py
View File

@@ -16,12 +16,17 @@ import contextlib
import logging
import ipaddress
import subprocess
import tenacity
import unittest
import novaclient
import zaza.model as model
import zaza.charm_lifecycle.utils as lifecycle_utils
import zaza.openstack.configure.guest as configure_guest
import zaza.openstack.utilities.openstack as openstack_utils
import zaza.openstack.utilities.generic as generic_utils
import zaza.openstack.charm_tests.glance.setup as glance_setup
def skipIfNotHA(service_name):
@@ -432,6 +437,95 @@ class OpenStackBaseTest(BaseCharmTest):
cls.keystone_session = openstack_utils.get_overcloud_keystone_session(
model_name=cls.model_name)
cls.cacert = openstack_utils.get_cacert()
cls.nova_client = (
openstack_utils.get_nova_session_client(cls.keystone_session))
def launch_guest(self, guest_name, userdata=None):
"""Launch two guests to use in tests.
Note that it is up to the caller to have set the RESOURCE_PREFIX class
variable prior to calling this method.
Also note that this method will remove any already existing instance
with same name as what is requested.
:param guest_name: Name of instance
:type guest_name: str
:param userdata: Userdata to attach to instance
:type userdata: Optional[str]
:returns: Nova instance objects
:rtype: Server
"""
instance_name = '{}-{}'.format(self.RESOURCE_PREFIX, guest_name)
instance = self.retrieve_guest(instance_name)
if instance:
logging.info('Removing already existing instance ({}) with '
'requested name ({})'
.format(instance.id, instance_name))
openstack_utils.delete_resource(
self.nova_client.servers,
instance.id,
msg="server")
return configure_guest.launch_instance(
glance_setup.LTS_IMAGE_NAME,
vm_name=instance_name,
userdata=userdata)
def launch_guests(self, userdata=None):
"""Launch two guests to use in tests.
Note that it is up to the caller to have set the RESOURCE_PREFIX class
variable prior to calling this method.
:param userdata: Userdata to attach to instance
:type userdata: Optional[str]
:returns: List of launched Nova instance objects
:rtype: List[Server]
"""
launched_instances = []
for guest_number in range(1, 2+1):
for attempt in tenacity.Retrying(
stop=tenacity.stop_after_attempt(3),
wait=tenacity.wait_exponential(
multiplier=1, min=2, max=10)):
with attempt:
launched_instances.append(
self.launch_guest(
guest_name='ins-{}'.format(guest_number),
userdata=userdata))
return launched_instances
def retrieve_guest(self, guest_name):
"""Return guest matching name.
:param nova_client: Nova client to use when checking status
:type nova_client: Nova client
:returns: the matching guest
:rtype: Union[novaclient.Server, None]
"""
try:
return self.nova_client.servers.find(name=guest_name)
except novaclient.exceptions.NotFound:
return None
def retrieve_guests(self):
"""Return test guests.
Note that it is up to the caller to have set the RESOURCE_PREFIX class
variable prior to calling this method.
:param nova_client: Nova client to use when checking status
:type nova_client: Nova client
:returns: the matching guest
:rtype: Union[novaclient.Server, None]
"""
instance_1 = self.retrieve_guest(
'{}-ins-1'.format(self.RESOURCE_PREFIX))
instance_2 = self.retrieve_guest(
'{}-ins-1'.format(self.RESOURCE_PREFIX))
return instance_1, instance_2
def format_addr(addr):