diff options
Diffstat (limited to 'functest')
| -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 |
9 files changed, 0 insertions, 1737 deletions
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 |