openstack
/
mistral-tempest-plugin
Code Issues Proposed changes

Merge "Remove all class methods in tempest tests"

This commit is contained in:
Zuul 2018-01-15 10:13:21 +00:00 committed by Gerrit Code Review
parent 63a0fe20f9 23b0fb12b7
commit 1b0854ee8d
5 changed files with 86 additions and 100 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
mistral_tempest_tests/tests/api/v2/test_action_executions.py
View File

@ -28,22 +28,20 @@ LOG = logging.getLogger(__name__)
class ActionExecutionTestsV2(base.TestCase):
_service = 'workflowv2'
@classmethod
def resource_setup(cls):
super(ActionExecutionTestsV2, cls).resource_setup()
def setUp(self):
super(ActionExecutionTestsV2, self).setUp()
cls.client.create_action_execution(
self.client.create_action_execution(
{
'name': 'std.echo',
'input': '{"output": "Hello, Mistral!"}'
}
)
@classmethod
def resource_cleanup(cls):
for action_ex in cls.client.action_executions:
def tearDown(self):
for action_ex in self.client.action_executions:
try:
cls.client.delete_obj('action_executions', action_ex)
self.client.delete_obj('action_executions', action_ex)
except Exception as e:
LOG.exception(
'Exception raised when deleting '
@ -51,9 +49,9 @@ class ActionExecutionTestsV2(base.TestCase):
action_ex, six.text_type(e)
)
cls.client.action_executions = []
self.client.action_executions = []
super(ActionExecutionTestsV2, cls).resource_cleanup()
super(ActionExecutionTestsV2, self).tearDown()
@decorators.attr(type='sanity')
@decorators.idempotent_id('a72603bd-5d49-4d92-9747-8da6322e867d')

41
mistral_tempest_tests/tests/base.py
View File

@ -34,31 +34,30 @@ class TestCase(test.BaseTestCase):
if not CONF.service_available.mistral:
raise cls.skipException("Mistral support is required.")
@classmethod
def resource_setup(cls):
def setUp(self):
"""Client authentication.
This method allows to initialize authentication before
each test case and define parameters of Mistral API Service.
"""
super(TestCase, cls).resource_setup()
super(TestCase, self).setUp()
if 'WITHOUT_AUTH' in os.environ:
cls.mgr = mock.MagicMock()
cls.mgr.auth_provider = service_base.AuthProv()
cls.admin_mgr = cls.alt_mgr = cls.mgr
self.mgr = mock.MagicMock()
self.mgr.auth_provider = service_base.AuthProv()
self.admin_mgr = self.alt_mgr = self.mgr
else:
cls.admin_mgr = cls.os_admin
cls.mgr = cls.os_primary
cls.alt_mgr = cls.os_alt
self.admin_mgr = self.os_admin
self.mgr = self.os_primary
self.alt_mgr = self.os_alt
if cls._service == 'workflowv2':
cls.admin_client = mistral_client.MistralClientV2(
cls.admin_mgr.auth_provider, cls._service)
cls.client = mistral_client.MistralClientV2(
cls.mgr.auth_provider, cls._service)
cls.alt_client = mistral_client.MistralClientV2(
cls.alt_mgr.auth_provider, cls._service)
if self._service == 'workflowv2':
self.admin_client = mistral_client.MistralClientV2(
self.admin_mgr.auth_provider, self._service)
self.client = mistral_client.MistralClientV2(
self.mgr.auth_provider, self._service)
self.alt_client = mistral_client.MistralClientV2(
self.alt_mgr.auth_provider, self._service)
def tearDown(self):
super(TestCase, self).tearDown()
@ -70,12 +69,12 @@ class TestCase(test.BaseTestCase):
class TestCaseAdvanced(TestCase):
@classmethod
def resource_setup(cls):
super(TestCaseAdvanced, cls).resource_setup()
cls.image_ref = CONF.compute.image_ref
cls.flavor_ref = CONF.compute.flavor_ref
def setUp(self):
super(TestCaseAdvanced, self).setUp()
self.image_ref = CONF.compute.image_ref
self.flavor_ref = CONF.compute.flavor_ref
def tearDown(self):
for wb in self.client.workbooks:

4
mistral_tempest_tests/tests/scenario/engine/actions/v2/test_multi_vim_authentication.py
View File

@ -25,10 +25,6 @@ from mistral_tempest_tests.tests import base
class MultiVimActionsTests(base.TestCase):
_service = 'workflowv2'
@classmethod
def resource_setup(cls):
super(MultiVimActionsTests, cls).resource_setup()
@decorators.attr(type='openstack')
@decorators.idempotent_id('dadc9960-9c03-41a9-9a9d-7e97d527e6dd')
def test_multi_vim_support_target_headers(self):

7
mistral_tempest_tests/tests/scenario/engine/actions/v2/test_openstack_actions.py
View File

@ -24,11 +24,10 @@ class OpenStackActionsTestsV2(base.TestCase):
# TODO(akuznetsova): add checks for task result after task_output
# TODO(akuznetsova): refactoring will be finished
@classmethod
def resource_setup(cls):
super(OpenStackActionsTestsV2, cls).resource_setup()
def setUp(self):
super(OpenStackActionsTestsV2, self).setUp()
_, cls.wb = cls.client.create_workbook(
_, self.wb = self.client.create_workbook(
'openstack/action_collection_wb.yaml')
@decorators.attr(type='openstack')

116
mistral_tempest_tests/tests/scenario/engine/actions/v2/test_ssh_actions.py
View File

@ -37,10 +37,9 @@ class SSHActionsTestsV2(base.TestCaseAdvanced):
_service = 'workflowv2'
@classmethod
def _create_security_group_rule_ssh(cls):
def _create_security_group_rule_ssh(self):
sec_groups = (
cls.mgr.compute_security_groups_client.
self.mgr.compute_security_groups_client.
list_security_groups()
)
sec_groups = sec_groups['security_groups']
@ -50,7 +49,7 @@ class SSHActionsTestsV2(base.TestCaseAdvanced):
)
rule = (
cls.mgr.compute_security_group_rules_client
self.mgr.compute_security_group_rules_client
.create_security_group_rule(
parent_group_id=default_group['id'],
ip_protocol="tcp",
@ -60,20 +59,18 @@ class SSHActionsTestsV2(base.TestCaseAdvanced):
)
)
cls.ssh_rule_id = rule['security_group_rule']['id']
self.ssh_rule_id = rule['security_group_rule']['id']
@classmethod
def _create_server(cls, server_name, **kwargs):
return cls.mgr.servers_client.create_server(
def _create_server(self, server_name, **kwargs):
return self.mgr.servers_client.create_server(
name=server_name,
imageRef=CONF.compute.image_ref,
flavorRef=CONF.compute.flavor_ref,
**kwargs
).get('server')
@classmethod
def _associate_floating_ip_to_server(cls, server_id):
fl_ip_client = cls.mgr.compute_floating_ips_client
def _associate_floating_ip_to_server(self, server_id):
fl_ip_client = self.mgr.compute_floating_ips_client
all_ips = fl_ip_client.list_floating_ips().get(
'floating_ips'
@ -96,8 +93,7 @@ class SSHActionsTestsV2(base.TestCaseAdvanced):
return ip
@classmethod
def _wait_until_server_up(cls, server_ip, timeout=120, delay=2):
def _wait_until_server_up(self, server_ip, timeout=120, delay=2):
seconds_remain = timeout
LOG.info("Waiting server SSH [IP=%s]...", server_ip)
@ -119,14 +115,13 @@ class SSHActionsTestsV2(base.TestCaseAdvanced):
"Failed waiting until server's '%s' SSH is up." % server_ip
)
@classmethod
def _wait_until_server_active(cls, server_id, timeout=60, delay=2):
def _wait_until_server_active(self, server_id, timeout=60, delay=2):
seconds_remain = timeout
LOG.info("Waiting server [id=%s]...", server_id)
while seconds_remain > 0:
server_info = cls.mgr.servers_client.show_server(server_id)
server_info = self.mgr.servers_client.show_server(server_id)
if server_info['server']['status'] == 'ACTIVE':
return
@ -137,15 +132,14 @@ class SSHActionsTestsV2(base.TestCaseAdvanced):
"Failed waiting until server %s is active." % server_id
)
@classmethod
def _wait_until_server_delete(cls, server_id, timeout=60, delay=2):
def _wait_until_server_delete(self, server_id, timeout=60, delay=2):
seconds_remain = timeout
LOG.info("Deleting server [id=%s]...", server_id)
while seconds_remain > 0:
try:
cls.mgr.servers_client.show_server(server_id)
self.mgr.servers_client.show_server(server_id)
seconds_remain -= delay
time.sleep(delay)
except exceptions.NotFound:
@ -153,95 +147,95 @@ class SSHActionsTestsV2(base.TestCaseAdvanced):
raise RuntimeError("Server delete timeout!")
@classmethod
def resource_setup(cls):
super(SSHActionsTestsV2, cls).resource_setup()
def setUp(self):
super(SSHActionsTestsV2, self).setUp()
# Modify security group for accessing VM via SSH.
cls._create_security_group_rule_ssh()
self._create_security_group_rule_ssh()
# Create keypair (public and private keys).
cls.private_key, cls.public_key = utils.generate_key_pair()
cls.key_name = 'mistral-functional-tests-key'
self.private_key, self.public_key = utils.generate_key_pair()
self.key_name = 'mistral-functional-tests-key'
# If ZUUL_PROJECT is specified, it means
# tests are running on Jenkins gate.
if os.environ.get('ZUUL_PROJECT'):
cls.key_dir = "/opt/stack/new/.ssh/"
self.key_dir = "/opt/stack/new/.ssh/"
if not path.exists(cls.key_dir):
os.mkdir(cls.key_dir)
if not path.exists(self.key_dir):
os.mkdir(self.key_dir)
else:
cls.key_dir = SSH_KEYS_DIRECTORY
self.key_dir = SSH_KEYS_DIRECTORY
utils.save_text_to(
cls.private_key,
cls.key_dir + cls.key_name,
self.private_key,
self.key_dir + self.key_name,
overwrite=True
)
LOG.info("Private key saved to %s", cls.key_dir + cls.key_name)
LOG.info("Private key saved to %s", self.key_dir + self.key_name)
# Create keypair in nova.
cls.mgr.keypairs_client.create_keypair(
name=cls.key_name,
public_key=cls.public_key
self.mgr.keypairs_client.create_keypair(
name=self.key_name,
public_key=self.public_key
)
# Start servers and provide key_name.
# Note: start public vm only after starting the guest one,
# so we can track public vm launching using ssh, but can't
# do the same with guest VM.
cls.guest_vm = cls._create_server(
self.guest_vm = self._create_server(
'mistral-guest-vm',
key_name=cls.key_name
key_name=self.key_name
)
cls.public_vm = cls._create_server(
self.public_vm = self._create_server(
'mistral-public-vm',
key_name=cls.key_name
key_name=self.key_name
)
cls._wait_until_server_active(cls.public_vm['id'])
self._wait_until_server_active(self.public_vm['id'])
cls.public_vm_ip = cls._associate_floating_ip_to_server(
cls.public_vm['id']
self.public_vm_ip = self._associate_floating_ip_to_server(
self.public_vm['id']
)
# Wait until server is up.
cls._wait_until_server_up(cls.public_vm_ip)
self._wait_until_server_up(self.public_vm_ip)
# Update servers info.
cls.public_vm = cls.mgr.servers_client.show_server(
cls.public_vm['id']
self.public_vm = self.mgr.servers_client.show_server(
self.public_vm['id']
).get('server')
cls.guest_vm = cls.mgr.servers_client.show_server(
cls.guest_vm['id']
self.guest_vm = self.mgr.servers_client.show_server(
self.guest_vm['id']
).get('server')
@classmethod
def resource_cleanup(cls):
fl_ip_client = cls.mgr.compute_floating_ips_client
def tearDown(self):
mgr = self.mgr
fl_ip_client = mgr.compute_floating_ips_client
fl_ip_client.disassociate_floating_ip_from_server(
cls.public_vm_ip,
cls.public_vm['id']
self.public_vm_ip,
self.public_vm['id']
)
cls.mgr.servers_client.delete_server(cls.public_vm['id'])
cls.mgr.servers_client.delete_server(cls.guest_vm['id'])
mgr.servers_client.delete_server(self.public_vm['id'])
mgr.servers_client.delete_server(self.guest_vm['id'])
cls._wait_until_server_delete(cls.public_vm['id'])
cls._wait_until_server_delete(cls.guest_vm['id'])
self._wait_until_server_delete(self.public_vm['id'])
self._wait_until_server_delete(self.guest_vm['id'])
cls.mgr.keypairs_client.delete_keypair(cls.key_name)
mgr.keypairs_client.delete_keypair(self.key_name)
cls.mgr.compute_security_group_rules_client.delete_security_group_rule(
cls.ssh_rule_id
mgr.compute_security_group_rules_client.delete_security_group_rule(
self.ssh_rule_id
)
os.remove(cls.key_dir + cls.key_name)
os.remove(self.key_dir + self.key_name)
super(SSHActionsTestsV2, cls).resource_cleanup()
super(SSHActionsTestsV2, self).tearDown()
@decorators.attr(type='sanity')
@decorators.idempotent_id('3e12a2ad-5b10-46b0-ae1f-ed34d3cc6ae2')