diff options
| author |   Cédric Ollivier <cedric.ollivier@orange.com> | 2017-12-19 19:58:15 +0100 |
|---|---|---|
| committer | Cédric Ollivier <cedric.ollivier@orange.com> | 2017-12-19 20:40:45 +0100 |
| commit | 00c01aeafab54ec7b9ea376099fb6aaff0da5f5a (patch) | |
| tree | 486d4d762305fdbf9099f2064a0e38998ec15e31 | |
| parent | afa5e6098c7fd24ee2a9259a5dbff64a3e2d0a68 (diff) | |
Remove openstack_[clean,snapshot]
It also removes the related unit tests and updates docs and cli.
Change-Id: Ie11f77402f2b5b7055a0c7c5d931c8ff21124482
Signed-off-by: Cédric Ollivier <cedric.ollivier@orange.com>
| -rw-r--r-- | docs/com/pres/framework/framework.md | 1 | ||||
| -rw-r--r-- | docs/testing/developer/devguide/index.rst | 4 | ||||
| -rw-r--r-- | docs/testing/user/configguide/configguide.rst | 1 | ||||
| -rw-r--r-- | functest/ci/config_functest.yaml | 1 | ||||
| -rw-r--r-- | functest/cli/cli_base.py | 17 | ||||
| -rw-r--r-- | functest/cli/commands/cli_os.py | 43 | ||||
| -rw-r--r-- | functest/tests/unit/cli/commands/test_cli_os.py | 77 | ||||
| -rw-r--r-- | functest/tests/unit/cli/test_cli_base.py | 20 | ||||
| -rw-r--r-- | functest/tests/unit/utils/test_openstack_clean.py | 744 | ||||
| -rw-r--r-- | functest/tests/unit/utils/test_openstack_snapshot.py | 238 | ||||
| -rw-r--r-- | functest/utils/openstack_clean.py | 432 | ||||
| -rw-r--r-- | functest/utils/openstack_snapshot.py | 165 | ||||
| -rw-r--r-- | setup.cfg | 2 |
13 files changed, 1 insertions, 1744 deletions
diff --git a/docs/com/pres/framework/framework.md b/docs/com/pres/framework/framework.md index 7979bede..109d8a03 100644 --- a/docs/com/pres/framework/framework.md +++ b/docs/com/pres/framework/framework.md @@ -65,7 +65,6 @@ base model for single test case | is_successful() | interpret the results | | get_duration() | return the duration | | push_to_db() | push the results to the DB | -| create_snapshot() | save the testing environment | | clean() | clean the resources | diff --git a/docs/testing/developer/devguide/index.rst b/docs/testing/developer/devguide/index.rst index 11d541ae..19a1bdcb 100644 --- a/docs/testing/developer/devguide/index.rst +++ b/docs/testing/developer/devguide/index.rst @@ -249,7 +249,7 @@ Functest util classes In order to simplify the creation of test cases, Functest develops also some functions that are used by internal test cases. Several features are supported such as logger, configuration management and -Openstack capabilities (snapshot, clean, tacker,..). +Openstack capabilities (tacker,..). These functions can be found under <repo>/functest/utils and can be described as follows:: @@ -259,8 +259,6 @@ follows:: |-- decorators.py |-- env.py |-- functest_utils.py - |-- openstack_clean.py - |-- openstack_snapshot.py |-- openstack_tacker.py `-- openstack_utils.py diff --git a/docs/testing/user/configguide/configguide.rst b/docs/testing/user/configguide/configguide.rst index 6bc1b5e3..a98772a6 100644 --- a/docs/testing/user/configguide/configguide.rst +++ b/docs/testing/user/configguide/configguide.rst @@ -462,7 +462,6 @@ The structure of functest repo can be described as follows:: | |-- functest_utils.py | |-- functest_vacation.py | |-- openstack_clean.py - | |-- openstack_snapshot.py | |-- openstack_tacker.py | `-- openstack_utils.py |-- requirements.txt diff --git a/functest/ci/config_functest.yaml b/functest/ci/config_functest.yaml index 315c6c9c..48322a6c 100644 --- a/functest/ci/config_functest.yaml +++ b/functest/ci/config_functest.yaml @@ -22,7 +22,6 @@ general: openstack: creds: /home/opnfv/functest/conf/openstack.creds - snapshot_file: /home/opnfv/functest/conf/openstack_snapshot.yaml image_name: Cirros-0.3.5 image_name_alt: Cirros-0.3.5-1 diff --git a/functest/cli/cli_base.py b/functest/cli/cli_base.py index aa8ab24b..5890e0a3 100644 --- a/functest/cli/cli_base.py +++ b/functest/cli/cli_base.py @@ -64,23 +64,6 @@ def os_check(): _openstack.check() -@openstack.command('snapshot-create', help="Generates a snapshot of the " - "current OpenStack resources.") -def os_snapshot_create(): - _openstack.snapshot_create() - - -@openstack.command('snapshot-show', help="Prints the OpenStack snapshot.") -def os_snapshot_show(): - _openstack.snapshot_show() - - -@openstack.command('clean', - help="Cleans the OpenStack resources except the snapshot.") -def os_clean(): - _openstack.clean() - - @openstack.command('show-credentials', help="Prints the OpenStack credentials.") def os_show_credentials(): diff --git a/functest/cli/commands/cli_os.py b/functest/cli/commands/cli_os.py index 1ec705a5..9057da84 100644 --- a/functest/cli/commands/cli_os.py +++ b/functest/cli/commands/cli_os.py @@ -11,13 +11,10 @@ import os import click -import six from six.moves.urllib.parse import urlparse from functest.ci import check_deployment from functest.utils.constants import CONST -import functest.utils.openstack_clean as os_clean -import functest.utils.openstack_snapshot as os_snapshot class OpenStack(object): @@ -27,7 +24,6 @@ class OpenStack(object): self.endpoint_ip = None self.endpoint_port = None self.openstack_creds = CONST.__getattribute__('openstack_creds') - self.snapshot_file = CONST.__getattribute__('openstack_snapshot_file') if self.os_auth_url: self.endpoint_ip = urlparse(self.os_auth_url).hostname self.endpoint_port = urlparse(self.os_auth_url).port @@ -56,45 +52,6 @@ class OpenStack(object): deployment = check_deployment.CheckDeployment() deployment.check_all() - def snapshot_create(self): - self.ping_endpoint() - if os.path.isfile(self.snapshot_file): - answer = six.moves.input( - "It seems there is already an OpenStack " - "snapshot. Do you want to overwrite it with " - "the current OpenStack status? [y|n]\n") - while True: - if answer.lower() in ["y", "yes"]: - break - elif answer.lower() in ["n", "no"]: - return - else: - answer = six.moves.input( - "Invalid answer. Please type [y|n]\n") - - click.echo("Generating Openstack snapshot...") - os_snapshot.main() - - def snapshot_show(self): - if not os.path.isfile(self.snapshot_file): - click.echo("There is no OpenStack snapshot created. To create " - "one run the command " - "'functest openstack snapshot-create'") - return - with open(self.snapshot_file, 'r') as yaml_file: - click.echo("\n%s" - % yaml_file.read()) - - def clean(self): - self.ping_endpoint() - if not os.path.isfile(self.snapshot_file): - click.echo("Not possible to clean OpenStack without a snapshot. " - "This could cause problems. " - "Run first the command " - "'functest openstack snapshot-create'") - return - os_clean.main() - class CliOpenStack(OpenStack): diff --git a/functest/tests/unit/cli/commands/test_cli_os.py b/functest/tests/unit/cli/commands/test_cli_os.py index 434370a5..b827e87c 100644 --- a/functest/tests/unit/cli/commands/test_cli_os.py +++ b/functest/tests/unit/cli/commands/test_cli_os.py @@ -66,83 +66,6 @@ class CliOpenStackTesting(unittest.TestCase): self.cli_os.check() self.assertTrue(mock_check_deployment.called) - @mock.patch('functest.cli.commands.cli_os.os.path.isfile', - return_value=False) - @mock.patch('functest.cli.commands.cli_os.click.echo') - def test_snapshot_create(self, mock_click_echo, mock_os_path): - with mock.patch.object(self.cli_os, 'ping_endpoint'), \ - mock.patch('functest.cli.commands.cli_os.os_snapshot.main') \ - as mock_os_snapshot: - self.cli_os.snapshot_create() - mock_click_echo.assert_called_once_with("Generating Openstack " - "snapshot...") - self.assertTrue(mock_os_snapshot.called) - - @mock.patch('functest.cli.commands.cli_os.os.path.isfile', - return_value=True) - @mock.patch('functest.cli.commands.cli_os.click.echo') - def test_snapshot_create_overwrite(self, mock_click_echo, mock_os_path): - with mock.patch('six.moves.input', return_value="y") \ - as mock_raw_input, \ - mock.patch.object(self.cli_os, 'ping_endpoint'), \ - mock.patch('functest.cli.commands.cli_os.os_snapshot.main') \ - as mock_os_snapshot: - self.cli_os.snapshot_create() - mock_click_echo.assert_called_once_with("Generating Openstack " - "snapshot...") - mock_raw_input.assert_any_call("It seems there is already an " - "OpenStack snapshot. Do you want " - "to overwrite it with the current " - "OpenStack status? [y|n]\n") - self.assertTrue(mock_os_snapshot.called) - - @mock.patch('functest.cli.commands.cli_os.os.path.isfile', - return_value=False) - @mock.patch('functest.cli.commands.cli_os.click.echo') - def test_snapshot_show_missing_snap(self, mock_click_echo, mock_os_path): - self.cli_os.snapshot_show() - mock_click_echo.assert_called_once_with("There is no OpenStack " - "snapshot created. To create " - "one run the command " - "'functest openstack " - "snapshot-create'") - - @mock.patch('functest.cli.commands.cli_os.os.path.isfile', - return_value=True) - @mock.patch('functest.cli.commands.cli_os.click.echo') - def test_snapshot_show_default(self, mock_click_echo, mock_os_path): - with mock.patch('six.moves.builtins.open', - mock.mock_open(read_data='0')) \ - as m: - self.cli_os.snapshot_file = self.snapshot_file - self.cli_os.snapshot_show() - m.assert_called_once_with(self.snapshot_file, 'r') - mock_click_echo.assert_called_once_with("\n0") - - @mock.patch('functest.cli.commands.cli_os.os.path.isfile', - return_value=True) - @mock.patch('functest.cli.commands.cli_os.click.echo') - def test_clean(self, mock_click_echo, mock_os_path): - with mock.patch.object(self.cli_os, 'ping_endpoint'), \ - mock.patch('functest.cli.commands.cli_os.os_clean.main') \ - as mock_os_clean: - self.cli_os.clean() - self.assertTrue(mock_os_clean.called) - - @mock.patch('functest.cli.commands.cli_os.os.path.isfile', - return_value=False) - @mock.patch('functest.cli.commands.cli_os.click.echo') - def test_clean_missing_file(self, mock_click_echo, mock_os_path): - with mock.patch.object(self.cli_os, 'ping_endpoint'): - self.cli_os.clean() - mock_click_echo.assert_called_once_with("Not possible to clean " - "OpenStack without a " - "snapshot. This could " - "cause problems. " - "Run first the command " - "'functest openstack " - "snapshot-create'") - @mock.patch('functest.cli.commands.cli_os.click.echo') def test_show_credentials(self, mock_click_echo): key = 'OS_KEY' diff --git a/functest/tests/unit/cli/test_cli_base.py b/functest/tests/unit/cli/test_cli_base.py index 08c9b736..bc2ca903 100644 --- a/functest/tests/unit/cli/test_cli_base.py +++ b/functest/tests/unit/cli/test_cli_base.py @@ -35,26 +35,6 @@ class CliBaseTesting(unittest.TestCase): self.assertEqual(result.exit_code, 0) self.assertTrue(mock_method.called) - def test_os_snapshot_create(self): - with mock.patch.object(self._openstack, 'snapshot_create') \ - as mock_method: - result = self.runner.invoke(cli_base.os_snapshot_create) - self.assertEqual(result.exit_code, 0) - self.assertTrue(mock_method.called) - - def test_os_snapshot_show(self): - with mock.patch.object(self._openstack, 'snapshot_show') \ - as mock_method: - result = self.runner.invoke(cli_base.os_snapshot_show) - self.assertEqual(result.exit_code, 0) - self.assertTrue(mock_method.called) - - def test_os_clean(self): - with mock.patch.object(self._openstack, 'clean') as mock_method: - result = self.runner.invoke(cli_base.os_clean) - self.assertEqual(result.exit_code, 0) - self.assertTrue(mock_method.called) - def test_os_show_credentials(self): with mock.patch.object(self._openstack, 'show_credentials') \ as mock_method: diff --git a/functest/tests/unit/utils/test_openstack_clean.py b/functest/tests/unit/utils/test_openstack_clean.py deleted file mode 100644 index afd9120a..00000000 --- a/functest/tests/unit/utils/test_openstack_clean.py +++ /dev/null @@ -1,744 +0,0 @@ -#!/usr/bin/env python - -# All rights reserved. This program and the accompanying materials -# are made available under the terms of the Apache License, Version 2.0 -# which accompanies this distribution, and is available at -# http://www.apache.org/licenses/LICENSE-2.0 - -import logging -import mock -import unittest - -from functest.utils import openstack_clean -from functest.tests.unit import test_utils - - -class OSCleanTesting(unittest.TestCase): - - def _get_instance(self, key): - mock_obj = mock.Mock() - attrs = {'id': 'id' + str(key), 'name': 'name' + str(key), - 'ip': 'ip' + str(key), 'status': 'ACTIVE', - 'OS-EXT-STS:task_state': '-'} - mock_obj.configure_mock(**attrs) - return mock_obj - - def _get_instance_deleted(self, key): - mock_obj = mock.Mock() - attrs = {'id': 'id' + str(key), 'name': 'name' + str(key), - 'ip': 'ip' + str(key), 'status': 'DELETED', - 'OS-EXT-STS:task_state': '-'} - mock_obj.configure_mock(**attrs) - return mock_obj - - def _get_instance_deleting(self, key): - mock_obj = mock.Mock() - attrs = {'id': 'id' + str(key), 'name': 'name' + str(key), - 'ip': 'ip' + str(key), 'status': 'BUILD', - 'OS-EXT-STS:task_state': 'deleting'} - mock_obj.configure_mock(**attrs) - return mock_obj - - def _get_instance_other(self, key): - mock_obj = mock.Mock() - attrs = {'id': 'id' + str(key), 'name': 'name' + str(key), - 'ip': 'ip' + str(key), 'status': 'BUILD', - 'OS-EXT-STS:task_state': 'networking'} - mock_obj.configure_mock(**attrs) - return mock_obj - - def setUp(self): - self.client = mock.Mock() - self.test_list = [self._get_instance(1), self._get_instance(2)] - self.deleted_list = [self._get_instance_deleted(5), - self._get_instance_deleting(6)] - self.other_list = [self._get_instance_other(7)] - self.update_list = {'id1': 'name1', 'id2': 'name2'} - self.remove_list = {'id3': 'name3', 'id4': 'name4'} - self.test_dict_list = [{'id': 'id1', 'name': 'name1', 'ip': 'ip1', - 'router:external': False, - 'external_gateway_info': None}, - {'id': 'id2', 'name': 'name2', 'ip': 'ip2', - 'router:external': False, - 'external_gateway_info': None}] - self.floatingips_list = [{'id': 'id1', 'floating_ip_address': 'ip1'}, - {'id': 'id2', 'floating_ip_address': 'ip2'}] - self.routers = [mock.Mock()] - self.ports = [mock.Mock()] - - @mock.patch('functest.utils.openstack_clean.logger.debug') - def test_separator(self, mock_logger_debug): - openstack_clean.separator() - mock_logger_debug.assert_called_once_with("-----------------" - "-----------------" - "---------") - - @mock.patch('time.sleep') - @mock.patch('functest.utils.openstack_clean.logger.debug') - def test_remove_instances(self, mock_logger_debug, *args): - with mock.patch('functest.utils.openstack_clean.os_utils' - '.get_instances', return_value=self.test_list): - openstack_clean.remove_instances(self.client, self.update_list) - mock_logger_debug.assert_any_call("Removing Nova instances...") - mock_logger_debug.assert_any_call(" > this is a default " - "instance and will " - "NOT be deleted.") - - @mock.patch('time.sleep') - @mock.patch('functest.utils.openstack_clean.logger.debug') - def test_remove_instances_missing_instances(self, mock_logger_debug, - *args): - with mock.patch('functest.utils.openstack_clean.os_utils' - '.get_instances', return_value=[]): - openstack_clean.remove_instances(self.client, self.update_list) - mock_logger_debug.assert_any_call("Removing Nova instances...") - mock_logger_debug.assert_any_call("No instances found.") - - @mock.patch('time.sleep') - @mock.patch('functest.utils.openstack_clean.logger.debug') - def test_remove_instances_delete_success(self, mock_logger_debug, *args): - with mock.patch('functest.utils.openstack_clean.os_utils' - '.get_instances', return_value=self.test_list), \ - mock.patch('functest.utils.openstack_clean.os_utils' - '.delete_instance', return_value=True): - openstack_clean.remove_instances(self.client, self.remove_list) - mock_logger_debug.assert_any_call("Removing Nova instances...") - mock_logger_debug.assert_any_call(" > Request sent.") - mock_logger_debug.assert_any_call(test_utils.RegexMatch("Removing" - " instance" - " '\s*\S+'" - " ...")) - - @mock.patch('time.sleep') - @mock.patch('functest.utils.openstack_clean.logger.debug') - def test_remove_instances_pending_delete_success(self, mock_logger_debug, - *args): - with mock.patch('functest.utils.openstack_clean.os_utils' - '.get_instances', return_value=self.deleted_list), \ - mock.patch('functest.utils.openstack_clean.os_utils' - '.delete_instance', return_value=True): - openstack_clean.remove_instances(self.client, self.remove_list) - mock_logger_debug.assert_any_call("Removing Nova instances...") - mock_logger_debug.test_utils.RegexMatch("Removing" - " instance" - " '\s*\S+'" - " ...").assert_not_called() - - @mock.patch('time.sleep') - @mock.patch('functest.utils.openstack_clean.logger.debug') - def test_remove_instances_other_delete_success(self, mock_logger_debug, - *args): - with mock.patch('functest.utils.openstack_clean.os_utils' - '.get_instances', return_value=self.other_list), \ - mock.patch('functest.utils.openstack_clean.os_utils' - '.delete_instance', return_value=True): - openstack_clean.remove_instances(self.client, self.remove_list) - mock_logger_debug.assert_any_call("Removing Nova instances...") - mock_logger_debug.assert_any_call(" > Request sent.") - mock_logger_debug.assert_any_call(test_utils.RegexMatch("Removing" - " instance" - " '\s*\S+'" - " ...")) - - @mock.patch('time.sleep') - @mock.patch('functest.utils.openstack_clean.logger.error') - @mock.patch('functest.utils.openstack_clean.logger.debug') - def test_remove_instances_delete_failed(self, mock_logger_debug, - mock_logger_error, *args): - with mock.patch('functest.utils.openstack_clean.os_utils' - '.get_instances', return_value=self.test_list), \ - mock.patch('functest.utils.openstack_clean.os_utils' - '.delete_instance', return_value=False): - openstack_clean.remove_instances(self.client, self.remove_list) - mock_logger_debug.assert_any_call("Removing Nova instances...") - mock_logger_error.assert_any_call(test_utils. - RegexMatch("There has been a " - "problem removing " - "the instance \s*\S+" - "...")) - mock_logger_debug.assert_any_call(test_utils.RegexMatch("Removing" - " instance" - " '\s*\S+'" - " ...")) - - @mock.patch('functest.utils.openstack_clean.logger.debug') - def test_remove_images(self, mock_logger_debug): - with mock.patch('functest.utils.openstack_clean.os_utils' - '.get_images', return_value=self.test_list): - openstack_clean.remove_images(self.client, self.update_list) - mock_logger_debug.assert_any_call("Removing Glance images...") - mock_logger_debug.assert_any_call(" > this is a default " - "image and will " - "NOT be deleted.") - - @mock.patch('functest.utils.openstack_clean.logger.debug') - def test_remove_images_missing_images(self, mock_logger_debug): - with mock.patch('functest.utils.openstack_clean.os_utils' - '.get_images', return_value=[]): - openstack_clean.remove_images(self.client, self.update_list) - mock_logger_debug.assert_any_call("Removing Glance images...") - mock_logger_debug.assert_any_call("No images found.") - - @mock.patch('functest.utils.openstack_clean.logger.debug') - def test_remove_images_delete_success(self, mock_logger_debug): - with mock.patch('functest.utils.openstack_clean.os_utils' - '.get_images', return_value=self.test_list), \ - mock.patch('functest.utils.openstack_clean.os_utils' - '.delete_glance_image', return_value=True): - openstack_clean.remove_images(self.client, self.remove_list) - mock_logger_debug.assert_any_call("Removing Glance images...") - mock_logger_debug.assert_any_call(" > Done!") - mock_logger_debug.assert_any_call(test_utils. - RegexMatch("Removing image " - "\s*\S+," - " ID=\s*\S+ ...")) - - @mock.patch('functest.utils.openstack_clean.logger.error') - @mock.patch('functest.utils.openstack_clean.logger.debug') - def test_remove_images_delete_failed(self, mock_logger_debug, - mock_logger_error): - with mock.patch('functest.utils.openstack_clean.os_utils' - '.get_images', return_value=self.test_list), \ - mock.patch('functest.utils.openstack_clean.os_utils' - '.delete_glance_image', return_value=False): - openstack_clean.remove_images(self.client, self.remove_list) - mock_logger_debug.assert_any_call("Removing Glance images...") - mock_logger_error.assert_any_call(test_utils. - RegexMatch("There has been a " - "problem removing the" - "image \s*\S+...")) - mock_logger_debug.assert_any_call(test_utils. - RegexMatch("Removing image " - "\s*\S+," - " ID=\s*\S+ ...")) - - @mock.patch('functest.utils.openstack_clean.logger.debug') - def test_remove_volumes(self, mock_logger_debug): - with mock.patch('functest.utils.openstack_clean.os_utils' - '.get_volumes', return_value=self.test_list): - openstack_clean.remove_volumes(self.client, self.update_list) - mock_logger_debug.assert_any_call("Removing Cinder volumes...") - mock_logger_debug.assert_any_call(" > this is a default " - "volume and will " - "NOT be deleted.") - - @mock.patch('functest.utils.openstack_clean.logger.debug') - def test_remove_volumes_missing_volumes(self, mock_logger_debug): - with mock.patch('functest.utils.openstack_clean.os_utils' - '.get_volumes', return_value=[]): - openstack_clean.remove_volumes(self.client, self.update_list) - mock_logger_debug.assert_any_call("Removing Cinder volumes...") - mock_logger_debug.assert_any_call("No volumes found.") - - @mock.patch('functest.utils.openstack_clean.logger.debug') - def test_remove_volumes_delete_success(self, mock_logger_debug): - with mock.patch('functest.utils.openstack_clean.os_utils' - '.get_volumes', return_value=self.test_list), \ - mock.patch('functest.utils.openstack_clean.os_utils' - '.delete_volume', return_value=True): - openstack_clean.remove_volumes(self.client, self.remove_list) - mock_logger_debug.assert_any_call("Removing Cinder volumes...") - mock_logger_debug.assert_any_call(" > Done!") - mock_logger_debug.assert_any_call(test_utils. - RegexMatch("Removing cinder " - "volume \s*\S+ ...")) - - @mock.patch('functest.utils.openstack_clean.logger.error') - @mock.patch('functest.utils.openstack_clean.logger.debug') - def test_remove_volumes_delete_failed(self, mock_logger_debug, - mock_logger_error): - with mock.patch('functest.utils.openstack_clean.os_utils' - '.get_volumes', return_value=self.test_list), \ - mock.patch('functest.utils.openstack_clean.os_utils' - '.delete_volume', return_value=False): - openstack_clean.remove_volumes(self.client, self.remove_list) - mock_logger_debug.assert_any_call("Removing Cinder volumes...") - mock_logger_error.assert_any_call(test_utils. - RegexMatch("There has been a " - "problem removing " - "the " - "volume \s*\S+...")) - mock_logger_debug.assert_any_call(test_utils. - RegexMatch("Removing cinder " - "volume \s*\S+ ...")) - - @mock.patch('functest.utils.openstack_clean.logger.debug') - def test_remove_floatingips(self, mock_logger_debug): - with mock.patch('functest.utils.openstack_clean.os_utils' - '.get_floating_ips', - return_value=self.floatingips_list): - openstack_clean.remove_floatingips(self.client, self.update_list) - mock_logger_debug.assert_any_call("Removing floating IPs...") - mock_logger_debug.assert_any_call(" > this is a default " - "floating IP and will " - "NOT be deleted.") - - @mock.patch('functest.utils.openstack_clean.logger.debug') - def test_remove_floatingips_missing_floatingips(self, mock_logger_debug): - with mock.patch('functest.utils.openstack_clean.os_utils' - '.get_floating_ips', return_value=[]): - openstack_clean.remove_floatingips(self.client, self.update_list) - mock_logger_debug.assert_any_call("Removing floating IPs...") - mock_logger_debug.assert_any_call("No floating IPs found.") - - @mock.patch('functest.utils.openstack_clean.logger.debug') - def test_remove_floatingips_delete_success(self, mock_logger_debug): - with mock.patch('functest.utils.openstack_clean.os_utils' - '.get_floating_ips', - side_effect=[self.floatingips_list, None]), \ - mock.patch('functest.utils.openstack_clean.os_utils' - '.delete_floating_ip', return_value=True): - openstack_clean.remove_floatingips(self.client, self.remove_list) - mock_logger_debug.assert_any_call("Removing floating IPs...") - mock_logger_debug.assert_any_call(" > Done!") - mock_logger_debug.assert_any_call(test_utils. - RegexMatch("Removing floating " - "IP \s*\S+ ...")) - - @mock.patch('functest.utils.openstack_clean.logger.error') - @mock.patch('functest.utils.openstack_clean.logger.debug') - def test_remove_floatingips_delete_failed(self, mock_logger_debug, - mock_logger_error): - with mock.patch('functest.utils.openstack_clean.os_utils' - '.get_floating_ips', - return_value=self.floatingips_list), \ - mock.patch('functest.utils.openstack_clean.os_utils' - '.delete_floating_ip', return_value=False): - openstack_clean.remove_floatingips(self.client, self.remove_list) - mock_logger_debug.assert_any_call("Removing floating IPs...") - mock_logger_error.assert_any_call(test_utils. - RegexMatch("There has been a " - "problem removing " - "the floating IP " - "\s*\S+...")) - mock_logger_debug.assert_any_call(test_utils. - RegexMatch("Removing floating " - "IP \s*\S+ ...")) - - @mock.patch('time.sleep') - @mock.patch('functest.utils.openstack_clean.remove_routers') - @mock.patch('functest.utils.openstack_clean.remove_ports') - @mock.patch('functest.utils.openstack_clean.logger.debug') - def test_remove_networks(self, mock_logger_debug, - mock_remove_ports, - mock_remove_routers, *args): - with mock.patch('functest.utils.openstack_clean.os_utils' - '.get_network_list', - return_value=self.test_dict_list), \ - mock.patch('functest.utils.openstack_clean.os_utils' - '.get_port_list', return_value=self.ports), \ - mock.patch('functest.utils.openstack_clean.os_utils' - '.get_router_list', return_value=self.routers): - openstack_clean.remove_networks(self.client, self.update_list, - self.update_list) - mock_logger_debug.assert_any_call("Removing Neutron objects") - mock_logger_debug.assert_any_call(" > this is a default " - "network and will " - "NOT be deleted.") - mock_remove_ports.assert_called_once_with(self.client, self.ports, - []) - mock_remove_routers.assert_called_once_with(self.client, - self.routers, - self.update_list) - - @mock.patch('time.sleep') - @mock.patch('functest.utils.openstack_clean.remove_routers') - @mock.patch('functest.utils.openstack_clean.remove_ports') - @mock.patch('functest.utils.openstack_clean.logger.debug') - def test_remove_networks_missing_networks(self, mock_logger_debug, - mock_remove_ports, - mock_remove_routers, *args): - with mock.patch('functest.utils.openstack_clean.os_utils' - '.get_network_list', return_value=None), \ - mock.patch('functest.utils.openstack_clean.os_utils' - '.get_port_list', return_value=self.ports), \ - mock.patch('functest.utils.openstack_clean.os_utils' - '.get_router_list', return_value=self.routers): - openstack_clean.remove_networks(self.client, self.update_list, - self.update_list) - mock_logger_debug.assert_any_call("Removing Neutron objects") - mock_logger_debug.assert_any_call("There are no networks in the" - " deployment. ") - mock_remove_ports.assert_called_once_with(self.client, self.ports, - []) - mock_remove_routers.assert_called_once_with(self.client, - self.routers, - self.update_list) - - @mock.patch('time.sleep') - @mock.patch('functest.utils.openstack_clean.remove_routers') - @mock.patch('functest.utils.openstack_clean.remove_ports') - @mock.patch('functest.utils.openstack_clean.logger.debug') - def test_remove_networks_delete_success(self, mock_logger_debug, - mock_remove_ports, - mock_remove_routers, *args): - - with mock.patch('functest.utils.openstack_clean.os_utils' - '.get_network_list', - return_value=self.test_dict_list), \ - mock.patch('functest.utils.openstack_clean.os_utils' - '.delete_neutron_net', return_value=True), \ - mock.patch('functest.utils.openstack_clean.os_utils' - '.get_port_list', return_value=self.ports), \ - mock.patch('functest.utils.openstack_clean.os_utils' - '.get_router_list', return_value=self.routers): - openstack_clean.remove_networks(self.client, self.remove_list, - self.remove_list) - mock_logger_debug.assert_any_call("Removing Neutron objects") - mock_logger_debug.assert_any_call(" > this network will be " - "deleted.") - mock_logger_debug.assert_any_call(" > Done!") - mock_logger_debug.assert_any_call(test_utils. - RegexMatch("Removing network " - "\s*\S+ ...")) - mock_remove_ports.assert_called_once_with(self.client, self.ports, - ['id1', 'id2']) - mock_remove_routers.assert_called_once_with(self.client, - self.routers, - self.remove_list) - - @mock.patch('time.sleep') - @mock.patch('functest.utils.openstack_clean.remove_routers') - @mock.patch('functest.utils.openstack_clean.remove_ports') - @mock.patch('functest.utils.openstack_clean.logger.error') - @mock.patch('functest.utils.openstack_clean.logger.debug') - def test_remove_networks_delete_failed(self, mock_logger_debug, - mock_logger_error, - mock_remove_ports, - mock_remove_routers, *args): - with mock.patch('functest.utils.openstack_clean.os_utils' - '.get_network_list', - return_value=self.test_dict_list), \ - mock.patch('functest.utils.openstack_clean.os_utils' - '.delete_neutron_net', return_value=False), \ - mock.patch('functest.utils.openstack_clean.os_utils' - '.get_port_list', return_value=self.ports), \ - mock.patch('functest.utils.openstack_clean.os_utils' - '.get_router_list', return_value=self.routers): - openstack_clean.remove_networks(self.client, self.remove_list, - self.remove_list) - mock_logger_debug.assert_any_call("Removing Neutron objects") - mock_logger_error.assert_any_call(test_utils. - RegexMatch("There has been a" - " problem removing" - " the network \s*\S+" - "...")) - mock_logger_debug.assert_any_call(test_utils. - RegexMatch("Removing network " - "\s*\S+ ...")) - mock_remove_ports.assert_called_once_with(self.client, self.ports, - ['id1', 'id2']) - mock_remove_routers.assert_called_once_with(self.client, - self.routers, - self.remove_list) - - # TODO: ports - @mock.patch('functest.utils.openstack_clean.os_utils.update_neutron_port') - @mock.patch('functest.utils.openstack_clean.logger.error') - @mock.patch('functest.utils.openstack_clean.logger.debug') - def test_force_remove_port(self, mock_logger_debug, - mock_logger_error, - mock_update_neutron_port): - with mock.patch('functest.utils.openstack_clean.os_utils' - '.delete_neutron_port', - return_value=True): - openstack_clean.force_remove_port(self.client, 'id') - mock_logger_debug.assert_any_call(" > Done!") - mock_logger_debug.assert_any_call(test_utils. - RegexMatch("Clearing device_" - "owner for port " - "\s*\S+ ...")) - - @mock.patch('functest.utils.openstack_clean.os_utils.update_neutron_port') - @mock.patch('functest.utils.openstack_clean.logger.error') - @mock.patch('functest.utils.openstack_clean.logger.debug') - def test_force_remove_port_failed(self, mock_logger_debug, - mock_logger_error, - mock_update_neutron_port): - with mock.patch('functest.utils.openstack_clean.os_utils' - '.delete_neutron_port', - return_value=False): - openstack_clean.force_remove_port(self.client, 'id') - mock_logger_error.assert_any_call("There has been a " - "problem removing " - "the port id...") - mock_logger_debug.assert_any_call(test_utils. - RegexMatch("Clearing device_" - "owner for port " - "\s*\S+ ...")) - - @mock.patch('functest.utils.openstack_clean.logger.debug') - def test_remove_routers_missing_routers(self, mock_logger_debug): - with mock.patch('functest.utils.openstack_clean.os_utils' - '.delete_neutron_router', - return_value=True): - openstack_clean.remove_routers(self.client, self.test_dict_list, - self.remove_list) - mock_logger_debug.assert_any_call("Router is not connected" - " to anything." - "Ready to remove...") - mock_logger_debug.assert_any_call(" > Done!") - mock_logger_debug.assert_any_call(test_utils. - RegexMatch("Removing router " - "\s*\S+(\s*\S+) ...")) - - @mock.patch('functest.utils.openstack_clean.logger.error') - @mock.patch('functest.utils.openstack_clean.logger.debug') - def test_remove_routers_failed(self, mock_logger_debug, - mock_logger_error): - with mock.patch('functest.utils.openstack_clean.os_utils' - '.delete_neutron_router', - return_value=False): - openstack_clean.remove_routers(self.client, self.test_dict_list, - self.remove_list) - mock_logger_debug.assert_any_call("Router is not connected" - " to anything." - "Ready to remove...") - mock_logger_debug.assert_any_call(test_utils. - RegexMatch("Removing router " - "\s*\S+(\s*\S+) ...")) - mock_logger_error.assert_any_call(test_utils. - RegexMatch("There has been " - "a problem" - " removing the " - "router \s*\S+(" - "\s*\S+)...")) - - @mock.patch('functest.utils.openstack_clean.logger.error') - @mock.patch('functest.utils.openstack_clean.logger.debug') - def test_remove_missing_external_gateway(self, mock_logger_debug, - mock_logger_error): - with mock.patch('functest.utils.openstack_clean.os_utils' - '.delete_neutron_router', - return_value=False), \ - mock.patch('functest.utils.openstack_clean.os_utils' - '.remove_gateway_router', - return_value=False): - self.test_dict_list[0]['external_gateway_info'] = mock.Mock() - openstack_clean.remove_routers(self.client, self.test_dict_list, - self.remove_list) - mock_logger_debug.assert_any_call("Router has gateway to external" - " network.Removing link...") - mock_logger_error.assert_any_call("There has been a problem " - "removing the gateway...") - mock_logger_debug.assert_any_call(test_utils. - RegexMatch("Removing router " - "\s*\S+(\s*\S+) ...")) - mock_logger_error.assert_any_call(test_utils. - RegexMatch("There has been " - "a problem" - " removing the " - "router \s*\S+(" - "\s*\S+)...")) - - @mock.patch('functest.utils.openstack_clean.logger.debug') - def remove_security_groups(self, mock_logger_debug): - with mock.patch('functest.utils.openstack_clean.os_utils' - '.get_security_groups', - return_value=self.test_dict_list): - openstack_clean.remove_security_groups(self.client, - self.update_list) - mock_logger_debug.assert_any_call("Removing Security groups...") - mock_logger_debug.assert_any_call(" > this is a default " - "security group and will NOT " - "be deleted.") - - @mock.patch('functest.utils.openstack_clean.logger.debug') - def test_remove_security_groups_missing_sec_group(self, mock_logger_debug): - with mock.patch('functest.utils.openstack_clean.os_utils' - '.get_security_groups', return_value=[]): - openstack_clean.remove_security_groups(self.client, - self.update_list) - mock_logger_debug.assert_any_call("Removing Security groups...") - mock_logger_debug.assert_any_call("No security groups found.") - - @mock.patch('functest.utils.openstack_clean.logger.debug') - def test_remove_security_groups_delete_success(self, mock_logger_debug): - with mock.patch('functest.utils.openstack_clean.os_utils' - '.get_security_groups', - return_value=self.test_dict_list), \ - mock.patch('functest.utils.openstack_clean.os_utils' - '.delete_security_group', return_value=True): - openstack_clean.remove_security_groups(self.client, - self.remove_list) - mock_logger_debug.assert_any_call("Removing Security groups...") - mock_logger_debug.assert_any_call(" > Done!") - mock_logger_debug.assert_any_call(test_utils. - RegexMatch(" Removing \s*\S+" - "...")) - - @mock.patch('functest.utils.openstack_clean.logger.error') - @mock.patch('functest.utils.openstack_clean.logger.debug') - def test_remove_security_groups_delete_failed(self, mock_logger_debug, - mock_logger_error): - with mock.patch('functest.utils.openstack_clean.os_utils' - '.get_security_groups', - return_value=self.test_dict_list), \ - mock.patch('functest.utils.openstack_clean.os_utils' - '.delete_security_group', return_value=False): - openstack_clean.remove_security_groups(self.client, - self.remove_list) - mock_logger_debug.assert_any_call("Removing Security groups...") - mock_logger_error.assert_any_call(test_utils. - RegexMatch("There has been a " - "problem removing " - "the security group" - " \s*\S+...")) - mock_logger_debug.assert_any_call(test_utils. - RegexMatch(" Removing \s*\S+" - "...")) - - @mock.patch('functest.utils.openstack_clean.logger.debug') - def test_remove_users(self, mock_logger_debug): - with mock.patch('functest.utils.openstack_clean.os_utils' - '.get_users', return_value=self.test_list): - openstack_clean.remove_users(self.client, self.update_list) - mock_logger_debug.assert_any_call("Removing Users...") - mock_logger_debug.assert_any_call(" > this is a default " - "user and will " - "NOT be deleted.") - - @mock.patch('functest.utils.openstack_clean.logger.debug') - def test_remove_users_missing_users(self, mock_logger_debug): - with mock.patch('functest.utils.openstack_clean.os_utils' - '.get_users', return_value=None): - openstack_clean.remove_users(self.client, self.update_list) - mock_logger_debug.assert_any_call("Removing Users...") - mock_logger_debug.assert_any_call("There are no users in" - " the deployment. ") - - @mock.patch('functest.utils.openstack_clean.logger.debug') - def test_remove_users_delete_success(self, mock_logger_debug): - with mock.patch('functest.utils.openstack_clean.os_utils' - '.get_users', return_value=self.test_list), \ - mock.patch('functest.utils.openstack_clean.os_utils' - '.delete_user', return_value=True): - openstack_clean.remove_users(self.client, self.remove_list) - mock_logger_debug.assert_any_call("Removing Users...") - mock_logger_debug.assert_any_call(" > Done!") - mock_logger_debug.assert_any_call(test_utils. - RegexMatch(" Removing " - "\s*\S+...")) - - @mock.patch('functest.utils.openstack_clean.logger.error') - @mock.patch('functest.utils.openstack_clean.logger.debug') - def test_remove_users_delete_failed(self, mock_logger_debug, - mock_logger_error): - with mock.patch('functest.utils.openstack_clean.os_utils' - '.get_users', return_value=self.test_list), \ - mock.patch('functest.utils.openstack_clean.os_utils' - '.delete_user', return_value=False): - openstack_clean.remove_users(self.client, self.remove_list) - mock_logger_debug.assert_any_call("Removing Users...") - mock_logger_error.assert_any_call(test_utils. - RegexMatch("There has been a " - "problem removing " - "the user \s*\S+" - "...")) - mock_logger_debug.assert_any_call(test_utils. - RegexMatch(" Removing " - "\s*\S+...")) - - @mock.patch('functest.utils.openstack_clean.logger.debug') - def test_remove_tenants(self, mock_logger_debug): - with mock.patch('functest.utils.openstack_clean.os_utils' - '.get_tenants', return_value=self.test_list): - openstack_clean.remove_tenants(self.client, self.update_list) - mock_logger_debug.assert_any_call("Removing Tenants...") - mock_logger_debug.assert_any_call(" > this is a default" - " tenant and will " - "NOT be deleted.") - - @mock.patch('functest.utils.openstack_clean.logger.debug') - def test_remove_tenants_missing_tenants(self, mock_logger_debug): - with mock.patch('functest.utils.openstack_clean.os_utils' - '.get_tenants', return_value=None): - openstack_clean.remove_tenants(self.client, self.update_list) - mock_logger_debug.assert_any_call("Removing Tenants...") - mock_logger_debug.assert_any_call("There are no tenants in" - " the deployment. ") - - @mock.patch('functest.utils.openstack_clean.logger.debug') - def test_remove_tenants_delete_success(self, mock_logger_debug): - with mock.patch('functest.utils.openstack_clean.os_utils' - '.get_tenants', return_value=self.test_list), \ - mock.patch('functest.utils.openstack_clean.os_utils' - '.delete_tenant', return_value=True): - openstack_clean.remove_tenants(self.client, self.remove_list) - mock_logger_debug.assert_any_call("Removing Tenants...") - mock_logger_debug.assert_any_call(" > Done!") - mock_logger_debug.assert_any_call(test_utils. - RegexMatch(" Removing " - "\s*\S+...")) - - @mock.patch('functest.utils.openstack_clean.logger.error') - @mock.patch('functest.utils.openstack_clean.logger.debug') - def test_remove_tenants_delete_failed(self, mock_logger_debug, - mock_logger_error): - with mock.patch('functest.utils.openstack_clean.os_utils' - '.get_tenants', return_value=self.test_list), \ - mock.patch('functest.utils.openstack_clean.os_utils' - '.delete_tenant', return_value=False): - openstack_clean.remove_tenants(self.client, self.remove_list) - mock_logger_debug.assert_any_call("Removing Tenants...") - mock_logger_error.assert_any_call(test_utils. - RegexMatch("There has been a " - "problem removing " - "the tenant \s*\S+" - "...")) - mock_logger_debug.assert_any_call(test_utils. - RegexMatch(" Removing " - "\s*\S+...")) - - @mock.patch('functest.utils.openstack_clean.os_utils.get_glance_client') - @mock.patch('functest.utils.openstack_clean.os_utils.get_cinder_client') - @mock.patch('functest.utils.openstack_clean.os_utils' - '.get_keystone_client') - @mock.patch('functest.utils.openstack_clean.os_utils' - '.get_neutron_client') - @mock.patch('functest.utils.openstack_clean.os_utils.get_nova_client') - @mock.patch('functest.utils.openstack_clean.os_utils.check_credentials', - return_value=True) - @mock.patch('functest.utils.openstack_clean.logger.info') - @mock.patch('functest.utils.openstack_clean.logger.debug') - def test_main_default(self, mock_logger_debug, mock_logger_info, - mock_creds, mock_nova, mock_neutron, - mock_keystone, mock_cinder, mock_glance): - - with mock.patch('functest.utils.openstack_clean.remove_instances') \ - as mock_remove_instances, \ - mock.patch('functest.utils.openstack_clean.remove_images') \ - as mock_remove_images, \ - mock.patch('functest.utils.openstack_clean.remove_volumes') \ - as mock_remove_volumes, \ - mock.patch('functest.utils.openstack_clean.remove_floatingips') \ - as mock_remove_floatingips, \ - mock.patch('functest.utils.openstack_clean.remove_networks') \ - as mock_remove_networks, \ - mock.patch('functest.utils.openstack_clean.' - 'remove_security_groups') \ - as mock_remove_security_groups, \ - mock.patch('functest.utils.openstack_clean.remove_users') \ - as mock_remove_users, \ - mock.patch('functest.utils.openstack_clean.remove_tenants') \ - as mock_remove_tenants, \ - mock.patch('functest.utils.openstack_clean.yaml.safe_load', - return_value=mock.Mock()), \ - mock.patch('six.moves.builtins.open', mock.mock_open()) as m: - openstack_clean.main() - self.assertTrue(mock_remove_instances) - self.assertTrue(mock_remove_images) - self.assertTrue(mock_remove_volumes) - self.assertTrue(mock_remove_floatingips) - self.assertTrue(mock_remove_networks) - self.assertTrue(mock_remove_security_groups) - self.assertTrue(mock_remove_users) - self.assertTrue(mock_remove_tenants) - m.assert_called_once_with(openstack_clean.OS_SNAPSHOT_FILE) - mock_logger_info.assert_called_once_with("Cleaning OpenStack " - "resources...") - - -if __name__ == "__main__": - logging.disable(logging.CRITICAL) - unittest.main(verbosity=2) diff --git a/functest/tests/unit/utils/test_openstack_snapshot.py b/functest/tests/unit/utils/test_openstack_snapshot.py deleted file mode 100644 index 919b28c6..00000000 --- a/functest/tests/unit/utils/test_openstack_snapshot.py +++ /dev/null @@ -1,238 +0,0 @@ -#!/usr/bin/env python - -# All rights reserved. This program and the accompanying materials -# are made available under the terms of the Apache License, Version 2.0 -# which accompanies this distribution, and is available at -# http://www.apache.org/licenses/LICENSE-2.0 - -import logging -import mock -import unittest - -from functest.utils import openstack_snapshot - - -class OSSnapshotTesting(unittest.TestCase): - - def _get_instance(self, key): - mock_obj = mock.Mock() - attrs = {'id': 'id' + str(key), 'name': 'name' + str(key), - 'ip': 'ip' + str(key)} - mock_obj.configure_mock(**attrs) - return mock_obj - - def setUp(self): - self.client = mock.Mock() - self.test_list = [self._get_instance(1), self._get_instance(2)] - self.update_list = {'id1': 'name1', 'id2': 'name2'} - self.update_floatingips = {'id1': 'ip1', 'id2': 'ip2'} - self.floatingips_list = [{'id': 'id1', 'floating_ip_address': 'ip1'}, - {'id': 'id2', 'floating_ip_address': 'ip2'}] - self.test_dict_list = [{'id': 'id1', 'name': 'name1', 'ip': 'ip1'}, - {'id': 'id2', 'name': 'name2', 'ip': 'ip2'}] - - @mock.patch('functest.utils.openstack_snapshot.logger.info') - def test_separator(self, mock_logger_info): - openstack_snapshot.separator() - mock_logger_info.assert_called_once_with("-----------------" - "-----------------" - "---------") - - @mock.patch('functest.utils.openstack_snapshot.logger.debug') - def test_get_instances(self, mock_logger_debug): - with mock.patch('functest.utils.openstack_snapshot.os_utils' - '.get_instances', return_value=self.test_list): - resp = openstack_snapshot.get_instances(self.client) - mock_logger_debug.assert_called_once_with("Getting instances...") - self.assertDictEqual(resp, {'instances': self.update_list}) - - @mock.patch('functest.utils.openstack_snapshot.logger.debug') - def test_get_instances_missing_instances(self, mock_logger_debug): - with mock.patch('functest.utils.openstack_snapshot.os_utils' - '.get_instances', return_value=[]): - resp = openstack_snapshot.get_instances(self.client) - mock_logger_debug.assert_called_once_with("Getting instances...") - self.assertDictEqual(resp, {'instances': {}}) - - @mock.patch('functest.utils.openstack_snapshot.logger.debug') - def test_get_images(self, mock_logger_debug): - with mock.patch('functest.utils.openstack_snapshot.os_utils' - '.get_images', return_value=self.test_list): - resp = openstack_snapshot.get_images(self.client) - mock_logger_debug.assert_called_once_with("Getting images...") - self.assertDictEqual(resp, {'images': self.update_list}) - - @mock.patch('functest.utils.openstack_snapshot.logger.debug') - def test_get_images_missing_images(self, mock_logger_debug): - with mock.patch('functest.utils.openstack_snapshot.os_utils' - '.get_images', return_value=[]): - resp = openstack_snapshot.get_images(self.client) - mock_logger_debug.assert_called_once_with("Getting images...") - self.assertDictEqual(resp, {'images': {}}) - - @mock.patch('functest.utils.openstack_snapshot.logger.debug') - def test_get_volumes(self, mock_logger_debug): - with mock.patch('functest.utils.openstack_snapshot.os_utils' - '.get_volumes', return_value=self.test_list): - resp = openstack_snapshot.get_volumes(self.client) - mock_logger_debug.assert_called_once_with("Getting volumes...") - self.assertDictEqual(resp, {'volumes': self.update_list}) - - @mock.patch('functest.utils.openstack_snapshot.logger.debug') - def test_get_volumes_missing_volumes(self, mock_logger_debug): - with mock.patch('functest.utils.openstack_snapshot.os_utils' - '.get_volumes', return_value=[]): - resp = openstack_snapshot.get_volumes(self.client) - mock_logger_debug.assert_called_once_with("Getting volumes...") - self.assertDictEqual(resp, {'volumes': {}}) - - @mock.patch('functest.utils.openstack_snapshot.logger.debug') - def test_get_networks(self, mock_logger_debug): - with mock.patch('functest.utils.openstack_snapshot.os_utils' - '.get_network_list', return_value=self.test_dict_list): - resp = openstack_snapshot.get_networks(self.client) - mock_logger_debug.assert_called_once_with("Getting networks") - self.assertDictEqual(resp, {'networks': self.update_list}) - - @mock.patch('functest.utils.openstack_snapshot.logger.debug') - def test_get_networks_missing_networks(self, mock_logger_debug): - with mock.patch('functest.utils.openstack_snapshot.os_utils' - '.get_network_list', return_value=[]): - resp = openstack_snapshot.get_networks(self.client) - mock_logger_debug.assert_called_once_with("Getting networks") - self.assertDictEqual(resp, {'networks': {}}) - - @mock.patch('functest.utils.openstack_snapshot.logger.debug') - def test_get_routers(self, mock_logger_debug): - with mock.patch('functest.utils.openstack_snapshot.os_utils' - '.get_router_list', return_value=self.test_dict_list): - resp = openstack_snapshot.get_routers(self.client) - mock_logger_debug.assert_called_once_with("Getting routers") - self.assertDictEqual(resp, {'routers': self.update_list}) - - @mock.patch('functest.utils.openstack_snapshot.logger.debug') - def test_get_routers_missing_routers(self, mock_logger_debug): - with mock.patch('functest.utils.openstack_snapshot.os_utils' - '.get_router_list', return_value=[]): - resp = openstack_snapshot.get_routers(self.client) - mock_logger_debug.assert_called_once_with("Getting routers") - self.assertDictEqual(resp, {'routers': {}}) - - @mock.patch('functest.utils.openstack_snapshot.logger.debug') - def test_get_secgroups(self, mock_logger_debug): - with mock.patch('functest.utils.openstack_snapshot.os_utils' - '.get_security_groups', - return_value=self.test_dict_list): - resp = openstack_snapshot.get_security_groups(self.client) - mock_logger_debug.assert_called_once_with("Getting Security " - "groups...") - self.assertDictEqual(resp, {'secgroups': self.update_list}) - - @mock.patch('functest.utils.openstack_snapshot.logger.debug') - def test_get_secgroups_missing_secgroups(self, mock_logger_debug): - with mock.patch('functest.utils.openstack_snapshot.os_utils' - '.get_security_groups', return_value=[]): - resp = openstack_snapshot.get_security_groups(self.client) - mock_logger_debug.assert_called_once_with("Getting Security " - "groups...") - self.assertDictEqual(resp, {'secgroups': {}}) - - @mock.patch('functest.utils.openstack_snapshot.logger.debug') - def test_get_floatingips(self, mock_logger_debug): - with mock.patch('functest.utils.openstack_snapshot.os_utils' - '.get_floating_ips', - return_value=self.floatingips_list): - resp = openstack_snapshot.get_floatingips(self.client) - mock_logger_debug.assert_called_once_with("Getting Floating " - "IPs...") - self.assertDictEqual(resp, {'floatingips': - self.update_floatingips}) - - @mock.patch('functest.utils.openstack_snapshot.logger.debug') - def test_get_floatingips_missing_floatingips(self, mock_logger_debug): - with mock.patch('functest.utils.openstack_snapshot.os_utils' - '.get_floating_ips', return_value=[]): - resp = openstack_snapshot.get_floatingips(self.client) - mock_logger_debug.assert_called_once_with("Getting Floating " - "IPs...") - self.assertDictEqual(resp, {'floatingips': {}}) - - @mock.patch('functest.utils.openstack_snapshot.logger.debug') - def test_get_users(self, mock_logger_debug): - with mock.patch('functest.utils.openstack_snapshot.os_utils' - '.get_users', return_value=self.test_list): - resp = openstack_snapshot.get_users(self.client) - mock_logger_debug.assert_called_once_with("Getting users...") - self.assertDictEqual(resp, {'users': self.update_list}) - - @mock.patch('functest.utils.openstack_snapshot.logger.debug') - def test_get_users_missing_users(self, mock_logger_debug): - with mock.patch('functest.utils.openstack_snapshot.os_utils' - '.get_users', return_value=[]): - resp = openstack_snapshot.get_users(self.client) - mock_logger_debug.assert_called_once_with("Getting users...") - self.assertDictEqual(resp, {'users': {}}) - - @mock.patch('functest.utils.openstack_snapshot.logger.debug') - def test_get_tenants(self, mock_logger_debug): - with mock.patch('functest.utils.openstack_snapshot.os_utils' - '.get_tenants', return_value=self.test_list): - resp = openstack_snapshot.get_tenants(self.client) - mock_logger_debug.assert_called_once_with("Getting tenants...") - self.assertDictEqual(resp, {'tenants': self.update_list}) - - @mock.patch('functest.utils.openstack_snapshot.logger.debug') - def test_get_tenants_missing_tenants(self, mock_logger_debug): - with mock.patch('functest.utils.openstack_snapshot.os_utils' - '.get_tenants', return_value=[]): - resp = openstack_snapshot.get_tenants(self.client) - mock_logger_debug.assert_called_once_with("Getting tenants...") - self.assertDictEqual(resp, {'tenants': {}}) - - @mock.patch('functest.utils.openstack_clean.os_utils.get_glance_client') - @mock.patch('functest.utils.openstack_snapshot.os_utils.get_cinder_client') - @mock.patch('functest.utils.openstack_snapshot.os_utils' - '.get_keystone_client') - @mock.patch('functest.utils.openstack_snapshot.os_utils' - '.get_neutron_client') - @mock.patch('functest.utils.openstack_snapshot.os_utils.get_nova_client') - @mock.patch('functest.utils.openstack_snapshot.os_utils.check_credentials') - @mock.patch('functest.utils.openstack_snapshot.logger.info') - @mock.patch('functest.utils.openstack_snapshot.logger.debug') - def test_main_default(self, mock_logger_debug, mock_logger_info, - mock_creds, mock_nova, mock_neutron, - mock_keystone, mock_cinder, mock_glance): - with mock.patch('functest.utils.openstack_snapshot.get_instances', - return_value=self.update_list), \ - mock.patch('functest.utils.openstack_snapshot.get_images', - return_value=self.update_list), \ - mock.patch('functest.utils.openstack_snapshot.get_images', - return_value=self.update_list), \ - mock.patch('functest.utils.openstack_snapshot.get_volumes', - return_value=self.update_list), \ - mock.patch('functest.utils.openstack_snapshot.get_networks', - return_value=self.update_list), \ - mock.patch('functest.utils.openstack_snapshot.get_routers', - return_value=self.update_list), \ - mock.patch('functest.utils.openstack_snapshot.get_security_groups', - return_value=self.update_list), \ - mock.patch('functest.utils.openstack_snapshot.get_floatingips', - return_value=self.update_floatingips), \ - mock.patch('functest.utils.openstack_snapshot.get_users', - return_value=self.update_list), \ - mock.patch('functest.utils.openstack_snapshot.get_tenants', - return_value=self.update_list), \ - mock.patch('six.moves.builtins.open', mock.mock_open()) as m: - openstack_snapshot.main() - mock_logger_info.assert_called_once_with("Generating OpenStack " - "snapshot...") - m.assert_called_once_with(openstack_snapshot.OS_SNAPSHOT_FILE, - 'w+') - mock_logger_debug.assert_any_call("NOTE: These objects will " - "NOT be deleted after " + - "running the test.") - - -if __name__ == "__main__": - logging.disable(logging.CRITICAL) - unittest.main(verbosity=2) diff --git a/functest/utils/openstack_clean.py b/functest/utils/openstack_clean.py deleted file mode 100644 index d7df8f84..00000000 --- a/functest/utils/openstack_clean.py +++ /dev/null @@ -1,432 +0,0 @@ -#!/usr/bin/env python -# -# Description: -# Cleans possible leftovers after running functest tests: -# - Nova instances -# - Glance images -# - Cinder volumes -# - Floating IPs -# - Neutron networks, subnets and ports -# - Routers -# - Users and tenants -# - Tacker VNFDs and VNFs -# - Tacker SFCs and SFC classifiers -# -# Author: -# jose.lausuch@ericsson.com -# -# -# All rights reserved. This program and the accompanying materials -# are made available under the terms of the Apache License, Version 2.0 -# which accompanies this distribution, and is available at -# http://www.apache.org/licenses/LICENSE-2.0 -# - -import logging -import time -import yaml - -import functest.utils.openstack_utils as os_utils -from functest.utils.constants import CONST - -logger = logging.getLogger(__name__) - -OS_SNAPSHOT_FILE = CONST.openstack_snapshot_file - - -def separator(): - logger.debug("-------------------------------------------") - - -def remove_instances(nova_client, default_instances): - logger.debug("Removing Nova instances...") - instances = os_utils.get_instances(nova_client) - if instances is None or len(instances) == 0: - logger.debug("No instances found.") - return - - for instance in instances: - instance_name = getattr(instance, 'name') - instance_id = getattr(instance, 'id') - instance_status = getattr(instance, 'status') - instance_state = getattr(instance, 'OS-EXT-STS:task_state') - - logger.debug("'%s', ID=%s " % (instance_name, instance_id)) - if (instance_id not in default_instances and - instance_name not in default_instances.values() and - instance_status != 'DELETED' and - (instance_status != 'BUILD' or instance_state != 'deleting')): - logger.debug("Removing instance '%s' ..." % instance_id) - if os_utils.delete_instance(nova_client, instance_id): - logger.debug(" > Request sent.") - else: - logger.error("There has been a problem removing the " - "instance %s..." % instance_id) - else: - logger.debug(" > this is a default instance and will " - "NOT be deleted.") - - timeout = 50 - while timeout > 0: - instances = os_utils.get_instances(nova_client) - for instance in instances: - instance_id = getattr(instance, 'id') - if instance_id not in default_instances: - logger.debug("Waiting for instances to be terminated...") - timeout -= 1 - time.sleep(1) - continue - break - - -def remove_images(glance_client, default_images): - logger.debug("Removing Glance images...") - images = os_utils.get_images(glance_client) - if images is None: - return -1 - images = {image.id: image.name for image in images} - if len(images) == 0: - logger.debug("No images found.") - return - - for image in images: - image_id = image - image_name = images.get(image_id) - logger.debug("'%s', ID=%s " % (image_name, image_id)) - if (image_id not in default_images and - image_name not in default_images.values()): - logger.debug("Removing image '%s', ID=%s ..." - % (image_name, image_id)) - if os_utils.delete_glance_image(glance_client, image_id): - logger.debug(" > Done!") - else: - logger.error("There has been a problem removing the" - "image %s..." % image_id) - else: - logger.debug(" > this is a default image and will " - "NOT be deleted.") - - -def remove_volumes(cinder_client, default_volumes): - logger.debug("Removing Cinder volumes...") - volumes = os_utils.get_volumes(cinder_client) - if volumes is None or len(volumes) == 0: - logger.debug("No volumes found.") - return - - for volume in volumes: - volume_id = getattr(volume, 'id') - volume_name = getattr(volume, 'name') - logger.debug("'%s', ID=%s " % (volume_name, volume_id)) - if (volume_id not in default_volumes and - volume_name not in default_volumes.values()): - logger.debug("Removing cinder volume %s ..." % volume_id) - if os_utils.delete_volume(cinder_client, volume_id): - logger.debug(" > Done!") - else: - logger.debug("Trying forced removal...") - if os_utils.delete_volume(cinder_client, - volume_id, - forced=True): - logger.debug(" > Done!") - else: - logger.error("There has been a problem removing the " - "volume %s..." % volume_id) - else: - logger.debug(" > this is a default volume and will " - "NOT be deleted.") - - -def remove_floatingips(neutron_client, default_floatingips): - logger.debug("Removing floating IPs...") - floatingips = os_utils.get_floating_ips(neutron_client) - if floatingips is None or len(floatingips) == 0: - logger.debug("No floating IPs found.") - return - - init_len = len(floatingips) - deleted = 0 - for fip in floatingips: - fip_id = fip['id'] - fip_ip = fip['floating_ip_address'] - logger.debug("'%s', ID=%s " % (fip_ip, fip_id)) - if (fip_id not in default_floatingips and - fip_ip not in default_floatingips.values()): - logger.debug("Removing floating IP %s ..." % fip_id) - if os_utils.delete_floating_ip(neutron_client, fip_id): - logger.debug(" > Done!") - deleted += 1 - else: - logger.error("There has been a problem removing the " - "floating IP %s..." % fip_id) - else: - logger.debug(" > this is a default floating IP and will " - "NOT be deleted.") - - timeout = 50 - while timeout > 0: - floatingips = os_utils.get_floating_ips(neutron_client) - if floatingips is None or len(floatingips) == (init_len - deleted): - break - else: - logger.debug("Waiting for floating ips to be released...") - timeout -= 1 - time.sleep(1) - - -def remove_networks(neutron_client, default_networks, default_routers): - logger.debug("Removing Neutron objects") - network_ids = [] - networks = os_utils.get_network_list(neutron_client) - if networks is None: - logger.debug("There are no networks in the deployment. ") - else: - logger.debug("Existing networks:") - for network in networks: - net_id = network['id'] - net_name = network['name'] - logger.debug(" '%s', ID=%s " % (net_name, net_id)) - if (net_id in default_networks and - net_name in default_networks.values()): - logger.debug(" > this is a default network and will " - "NOT be deleted.") - elif network['router:external'] is True: - logger.debug(" > this is an external network and will " - "NOT be deleted.") - else: - logger.debug(" > this network will be deleted.") - network_ids.append(net_id) - - # delete ports - ports = os_utils.get_port_list(neutron_client) - if ports is None: - logger.debug("There are no ports in the deployment. ") - else: - remove_ports(neutron_client, ports, network_ids) - - # remove routers - routers = os_utils.get_router_list(neutron_client) - if routers is None: - logger.debug("There are no routers in the deployment. ") - else: - remove_routers(neutron_client, routers, default_routers) - - # trozet: wait for Neutron to auto-cleanup HA networks when HA router is - # deleted - time.sleep(5) - - # remove networks - if network_ids is not None: - for net_id in network_ids: - networks = os_utils.get_network_list(neutron_client) - if networks is None: - logger.debug("No networks left to remove") - break - elif not any(network['id'] == net_id for network in networks): - logger.debug("Network %s has already been removed" % net_id) - continue - logger.debug("Removing network %s ..." % net_id) - if os_utils.delete_neutron_net(neutron_client, net_id): - logger.debug(" > Done!") - else: - logger.error("There has been a problem removing the " - "network %s..." % net_id) - - -def remove_ports(neutron_client, ports, network_ids): - for port in ports: - if port['network_id'] in network_ids: - port_id = port['id'] - try: - subnet_id = port['fixed_ips'][0]['subnet_id'] - except: - logger.debug(" > WARNING: Port %s does not contain fixed_ips" - % port_id) - logger.info(port) - router_id = port['device_id'] - if len(port['fixed_ips']) == 0 and router_id == '': - logger.debug("Removing port %s ..." % port_id) - if (os_utils.delete_neutron_port(neutron_client, port_id)): - logger.debug(" > Done!") - else: - logger.error("There has been a problem removing the " - "port %s ..." % port_id) - force_remove_port(neutron_client, port_id) - - elif port['device_owner'] == 'network:router_interface': - logger.debug("Detaching port %s (subnet %s) from router %s ..." - % (port_id, subnet_id, router_id)) - if os_utils.remove_interface_router( - neutron_client, router_id, subnet_id): - time.sleep(5) # leave 5 seconds to detach - logger.debug(" > Done!") - else: - logger.error("There has been a problem removing the " - "interface %s from router %s..." - % (subnet_id, router_id)) - force_remove_port(neutron_client, port_id) - else: - force_remove_port(neutron_client, port_id) - - -def force_remove_port(neutron_client, port_id): - logger.debug("Clearing device_owner for port %s ..." % port_id) - os_utils.update_neutron_port(neutron_client, port_id, - device_owner='clear') - logger.debug("Removing port %s ..." % port_id) - if os_utils.delete_neutron_port(neutron_client, port_id): - logger.debug(" > Done!") - else: - logger.error("There has been a problem removing the port %s..." - % port_id) - - -def remove_routers(neutron_client, routers, default_routers): - for router in routers: - router_id = router['id'] - router_name = router['name'] - if (router_id not in default_routers and - router_name not in default_routers.values()): - logger.debug("Checking '%s' with ID=(%s) ..." % (router_name, - router_id)) - if router['external_gateway_info'] is not None: - logger.debug("Router has gateway to external network." - "Removing link...") - if os_utils.remove_gateway_router(neutron_client, router_id): - logger.debug(" > Done!") - else: - logger.error("There has been a problem removing " - "the gateway...") - else: - logger.debug("Router is not connected to anything." - "Ready to remove...") - logger.debug("Removing router %s(%s) ..." - % (router_name, router_id)) - if os_utils.delete_neutron_router(neutron_client, router_id): - logger.debug(" > Done!") - else: - logger.error("There has been a problem removing the " - "router '%s'(%s)..." % (router_name, router_id)) - - -def remove_security_groups(neutron_client, default_security_groups): - logger.debug("Removing Security groups...") - secgroups = os_utils.get_security_groups(neutron_client) - if secgroups is None or len(secgroups) == 0: - logger.debug("No security groups found.") - return - - for secgroup in secgroups: - secgroup_name = secgroup['name'] - secgroup_id = secgroup['id'] - logger.debug("'%s', ID=%s " % (secgroup_name, secgroup_id)) - if secgroup_id not in default_security_groups: - logger.debug(" Removing '%s'..." % secgroup_name) - if os_utils.delete_security_group(neutron_client, secgroup_id): - logger.debug(" > Done!") - else: - logger.error("There has been a problem removing the " - "security group %s..." % secgroup_id) - else: - logger.debug(" > this is a default security group and will NOT " - "be deleted.") - - -def remove_users(keystone_client, default_users): - logger.debug("Removing Users...") - users = os_utils.get_users(keystone_client) - if users is None: - logger.debug("There are no users in the deployment. ") - return - - for user in users: - user_name = getattr(user, 'name') - user_id = getattr(user, 'id') - logger.debug("'%s', ID=%s " % (user_name, user_id)) - if (user_id not in default_users and - user_name not in default_users.values()): - logger.debug(" Removing '%s'..." % user_name) - if os_utils.delete_user(keystone_client, user_id): - logger.debug(" > Done!") - else: - logger.error("There has been a problem removing the " - "user '%s'(%s)..." % (user_name, user_id)) - else: - logger.debug(" > this is a default user and will " - "NOT be deleted.") - - -def remove_tenants(keystone_client, default_tenants): - logger.debug("Removing Tenants...") - tenants = os_utils.get_tenants(keystone_client) - if tenants is None: - logger.debug("There are no tenants in the deployment. ") - return - - for tenant in tenants: - tenant_name = getattr(tenant, 'name') - tenant_id = getattr(tenant, 'id') - logger.debug("'%s', ID=%s " % (tenant_name, tenant_id)) - if (tenant_id not in default_tenants and - tenant_name not in default_tenants.values()): - logger.debug(" Removing '%s'..." % tenant_name) - if os_utils.delete_tenant(keystone_client, tenant_id): - logger.debug(" > Done!") - else: - logger.error("There has been a problem removing the " - "tenant '%s'(%s)..." % (tenant_name, tenant_id)) - else: - logger.debug(" > this is a default tenant and will " - "NOT be deleted.") - - -def main(): - logging.basicConfig() - logger.info("Cleaning OpenStack resources...") - - nova_client = os_utils.get_nova_client() - neutron_client = os_utils.get_neutron_client() - keystone_client = os_utils.get_keystone_client() - cinder_client = os_utils.get_cinder_client() - glance_client = os_utils.get_glance_client() - - try: - with open(OS_SNAPSHOT_FILE) as f: - snapshot_yaml = yaml.safe_load(f) - except Exception: - logger.info("The file %s does not exist. The OpenStack snapshot must" - " be created first. Aborting cleanup." % OS_SNAPSHOT_FILE) - return 0 - - default_images = snapshot_yaml.get('images') - default_instances = snapshot_yaml.get('instances') - default_volumes = snapshot_yaml.get('volumes') - default_networks = snapshot_yaml.get('networks') - default_routers = snapshot_yaml.get('routers') - default_security_groups = snapshot_yaml.get('secgroups') - default_floatingips = snapshot_yaml.get('floatingips') - default_users = snapshot_yaml.get('users') - default_tenants = snapshot_yaml.get('tenants') - - if not os_utils.check_credentials(): - logger.error("Please source the openrc credentials and run " - "the script again.") - return -1 - - remove_instances(nova_client, default_instances) - separator() - remove_images(glance_client, default_images) - separator() - remove_volumes(cinder_client, default_volumes) - separator() - remove_floatingips(neutron_client, default_floatingips) - separator() - remove_networks(neutron_client, default_networks, default_routers) - separator() - remove_security_groups(neutron_client, default_security_groups) - separator() - remove_users(keystone_client, default_users) - separator() - remove_tenants(keystone_client, default_tenants) - separator() - return 0 diff --git a/functest/utils/openstack_snapshot.py b/functest/utils/openstack_snapshot.py deleted file mode 100644 index 3dc6f80c..00000000 --- a/functest/utils/openstack_snapshot.py +++ /dev/null @@ -1,165 +0,0 @@ -#!/usr/bin/env python -# -# Description: -# Generates a list of the current Openstack objects in the deployment: -# - Nova instances -# - Glance images -# - Cinder volumes -# - Floating IPs -# - Neutron networks, subnets and ports -# - Routers -# - Users and tenants -# -# Author: -# jose.lausuch@ericsson.com -# -# -# All rights reserved. This program and the accompanying materials -# are made available under the terms of the Apache License, Version 2.0 -# which accompanies this distribution, and is available at -# http://www.apache.org/licenses/LICENSE-2.0 -# - -import logging -import yaml - -import functest.utils.openstack_utils as os_utils -from functest.utils.constants import CONST - -logger = logging.getLogger(__name__) - - -OS_SNAPSHOT_FILE = CONST.openstack_snapshot_file - - -def separator(): - logger.info("-------------------------------------------") - - -def get_instances(nova_client): - logger.debug("Getting instances...") - dic_instances = {} - instances = os_utils.get_instances(nova_client) - if not (instances is None or len(instances) == 0): - for instance in instances: - dic_instances.update({getattr(instance, 'id'): getattr(instance, - 'name')}) - return {'instances': dic_instances} - - -def get_images(glance_client): - logger.debug("Getting images...") - dic_images = {} - images = os_utils.get_images(glance_client) - if images is None: - return -1 - dic_images.update({image.id: image.name for image in images}) - return {'images': dic_images} - - -def get_volumes(cinder_client): - logger.debug("Getting volumes...") - dic_volumes = {} - volumes = os_utils.get_volumes(cinder_client) - if volumes is not None: - for volume in volumes: - dic_volumes.update({volume.id: volume.name}) - return {'volumes': dic_volumes} - - -def get_networks(neutron_client): - logger.debug("Getting networks") - dic_networks = {} - networks = os_utils.get_network_list(neutron_client) - if networks is not None: - for network in networks: - dic_networks.update({network['id']: network['name']}) - return {'networks': dic_networks} - - -def get_routers(neutron_client): - logger.debug("Getting routers") - dic_routers = {} - routers = os_utils.get_router_list(neutron_client) - if routers is not None: - for router in routers: - dic_routers.update({router['id']: router['name']}) - return {'routers': dic_routers} - - -def get_security_groups(neutron_client): - logger.debug("Getting Security groups...") - dic_secgroups = {} - secgroups = os_utils.get_security_groups(neutron_client) - if not (secgroups is None or len(secgroups) == 0): - for secgroup in secgroups: - dic_secgroups.update({secgroup['id']: secgroup['name']}) - return {'secgroups': dic_secgroups} - - -def get_floatingips(neutron_client): - logger.debug("Getting Floating IPs...") - dic_floatingips = {} - floatingips = os_utils.get_floating_ips(neutron_client) - if not (floatingips is None or len(floatingips) == 0): - for floatingip in floatingips: - dic_floatingips.update({floatingip['id']: - floatingip['floating_ip_address']}) - return {'floatingips': dic_floatingips} - - -def get_users(keystone_client): - logger.debug("Getting users...") - dic_users = {} - users = os_utils.get_users(keystone_client) - if not (users is None or len(users) == 0): - for user in users: - dic_users.update({getattr(user, 'id'): getattr(user, 'name')}) - return {'users': dic_users} - - -def get_tenants(keystone_client): - logger.debug("Getting tenants...") - dic_tenants = {} - tenants = os_utils.get_tenants(keystone_client) - if not (tenants is None or len(tenants) == 0): - for tenant in tenants: - dic_tenants.update({getattr(tenant, 'id'): - getattr(tenant, 'name')}) - return {'tenants': dic_tenants} - - -def main(): - logging.basicConfig() - logger.info("Generating OpenStack snapshot...") - - nova_client = os_utils.get_nova_client() - neutron_client = os_utils.get_neutron_client() - keystone_client = os_utils.get_keystone_client() - cinder_client = os_utils.get_cinder_client() - glance_client = os_utils.get_glance_client() - - if not os_utils.check_credentials(): - logger.error("Please source the openrc credentials and run the" + - "script again.") - return -1 - - snapshot = {} - snapshot.update(get_instances(nova_client)) - snapshot.update(get_images(glance_client)) - snapshot.update(get_volumes(cinder_client)) - snapshot.update(get_networks(neutron_client)) - snapshot.update(get_routers(neutron_client)) - snapshot.update(get_security_groups(neutron_client)) - snapshot.update(get_floatingips(neutron_client)) - snapshot.update(get_users(keystone_client)) - snapshot.update(get_tenants(keystone_client)) - - with open(OS_SNAPSHOT_FILE, 'w+') as yaml_file: - yaml_file.write(yaml.safe_dump(snapshot, default_flow_style=False)) - yaml_file.seek(0) - logger.debug("Openstack Snapshot found in the deployment:\n%s" - % yaml_file.read()) - logger.debug("NOTE: These objects will NOT be deleted after " + - "running the test.") - return 0 @@ -14,8 +14,6 @@ console_scripts = functest_odl = functest.opnfv_tests.sdn.odl.odl:main functest_refstack_client = functest.opnfv_tests.openstack.refstack_client.refstack_client:main functest_tempest_conf = functest.opnfv_tests.openstack.refstack_client.tempest_conf:main - openstack_snapshot = functest.utils.openstack_snapshot:main - openstack_clean = functest.utils.openstack_clean:main run_tests = functest.ci.run_tests:main check_deployment = functest.ci.check_deployment:main functest_restapi = functest.api.server:main |