| # Copyright (c) 2017 IBM Corp. |
| # Licensed under the Apache License, Version 2.0 (the "License"); |
| # you may not use this file except in compliance with the License. |
| # You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, software |
| # distributed under the License is distributed on an "AS IS" BASIS, |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| # See the License for the specific language governing permissions and |
| # limitations under the License. |
| |
| from unittest import mock |
| |
| import fixtures |
| import testtools |
| |
| from tempest.lib.common import validation_resources as vr |
| from tempest.lib import exceptions as lib_exc |
| from tempest.lib.services import clients |
| from tempest.tests import base |
| from tempest.tests.lib import fake_credentials |
| from tempest.tests.lib.services import registry_fixture |
| |
| FAKE_SECURITY_GROUP = {'security_group': {'id': 'sg_id'}} |
| FAKE_KEYPAIR = {'keypair': {'name': 'keypair_name'}} |
| FAKE_FIP_NOVA_NET = {'floating_ip': {'ip': '1.2.3.4', 'id': '1234'}} |
| FAKE_FIP_NEUTRON = {'floatingip': {'floating_ip_address': '1.2.3.4', |
| 'id': '1234'}} |
| |
| SERVICES = 'tempest.lib.services' |
| SG_CLIENT = (SERVICES + '.%s.security_groups_client.SecurityGroupsClient.%s') |
| SGR_CLIENT = (SERVICES + '.%s.security_group_rules_client.' |
| 'SecurityGroupRulesClient.create_security_group_rule') |
| KP_CLIENT = (SERVICES + '.compute.keypairs_client.KeyPairsClient.%s') |
| FIP_CLIENT = (SERVICES + '.%s.floating_ips_client.FloatingIPsClient.%s') |
| |
| |
| class TestValidationResources(base.TestCase): |
| |
| def setUp(self): |
| super(TestValidationResources, self).setUp() |
| self.useFixture(registry_fixture.RegistryFixture()) |
| self.mock_sg_compute = self.useFixture(fixtures.MockPatch( |
| SG_CLIENT % ('compute', 'create_security_group'), autospec=True, |
| return_value=FAKE_SECURITY_GROUP)) |
| self.mock_sg_network = self.useFixture(fixtures.MockPatch( |
| SG_CLIENT % ('network', 'create_security_group'), autospec=True, |
| return_value=FAKE_SECURITY_GROUP)) |
| self.mock_sgr_compute = self.useFixture(fixtures.MockPatch( |
| SGR_CLIENT % 'compute', autospec=True)) |
| self.mock_sgr_network = self.useFixture(fixtures.MockPatch( |
| SGR_CLIENT % 'network', autospec=True)) |
| self.mock_kp = self.useFixture(fixtures.MockPatch( |
| KP_CLIENT % 'create_keypair', autospec=True, |
| return_value=FAKE_KEYPAIR)) |
| self.mock_fip_compute = self.useFixture(fixtures.MockPatch( |
| FIP_CLIENT % ('compute', 'create_floating_ip'), autospec=True, |
| return_value=FAKE_FIP_NOVA_NET)) |
| self.mock_fip_network = self.useFixture(fixtures.MockPatch( |
| FIP_CLIENT % ('network', 'create_floatingip'), autospec=True, |
| return_value=FAKE_FIP_NEUTRON)) |
| self.os = clients.ServiceClients( |
| fake_credentials.FakeKeystoneV3Credentials(), 'fake_uri') |
| |
| def test_create_ssh_security_group_nova_net(self): |
| expected_sg_id = FAKE_SECURITY_GROUP['security_group']['id'] |
| sg = vr.create_ssh_security_group(self.os, add_rule=True, |
| use_neutron=False) |
| self.assertEqual(FAKE_SECURITY_GROUP['security_group'], sg) |
| # Neutron clients have not been used |
| self.assertEqual(self.mock_sg_network.mock.call_count, 0) |
| self.assertEqual(self.mock_sgr_network.mock.call_count, 0) |
| # Nova-net clients assertions |
| self.assertGreater(self.mock_sg_compute.mock.call_count, 0) |
| self.assertGreater(self.mock_sgr_compute.mock.call_count, 0) |
| for call in self.mock_sgr_compute.mock.call_args_list[1:]: |
| self.assertIn(expected_sg_id, call[1].values()) |
| |
| def test_create_ssh_security_group_neutron(self): |
| expected_sg_id = FAKE_SECURITY_GROUP['security_group']['id'] |
| expected_ethertype = 'fake_ethertype' |
| sg = vr.create_ssh_security_group(self.os, add_rule=True, |
| use_neutron=True, |
| ethertype=expected_ethertype) |
| self.assertEqual(FAKE_SECURITY_GROUP['security_group'], sg) |
| # Nova-net clients have not been used |
| self.assertEqual(self.mock_sg_compute.mock.call_count, 0) |
| self.assertEqual(self.mock_sgr_compute.mock.call_count, 0) |
| # Nova-net clients assertions |
| self.assertGreater(self.mock_sg_network.mock.call_count, 0) |
| self.assertGreater(self.mock_sgr_network.mock.call_count, 0) |
| # Check SG ID and ethertype are passed down to rules |
| for call in self.mock_sgr_network.mock.call_args_list[1:]: |
| self.assertIn(expected_sg_id, call[1].values()) |
| self.assertIn(expected_ethertype, call[1].values()) |
| |
| def test_create_ssh_security_no_rules(self): |
| sg = vr.create_ssh_security_group(self.os, add_rule=False) |
| self.assertEqual(FAKE_SECURITY_GROUP['security_group'], sg) |
| # SG Rules clients have not been used |
| self.assertEqual(self.mock_sgr_compute.mock.call_count, 0) |
| self.assertEqual(self.mock_sgr_network.mock.call_count, 0) |
| |
| @mock.patch.object(vr, 'create_ssh_security_group', |
| return_value=FAKE_SECURITY_GROUP['security_group']) |
| def test_create_validation_resources_nova_net(self, mock_create_sg): |
| expected_floating_network_id = 'my_fni' |
| expected_floating_network_name = 'my_fnn' |
| resources = vr.create_validation_resources( |
| self.os, keypair=True, floating_ip=True, security_group=True, |
| security_group_rules=True, ethertype='IPv6', use_neutron=False, |
| floating_network_id=expected_floating_network_id, |
| floating_network_name=expected_floating_network_name) |
| # Keypair calls |
| self.assertGreater(self.mock_kp.mock.call_count, 0) |
| # Floating IP calls |
| self.assertGreater(self.mock_fip_compute.mock.call_count, 0) |
| for call in self.mock_fip_compute.mock.call_args_list[1:]: |
| self.assertIn(expected_floating_network_name, call[1].values()) |
| self.assertNotIn(expected_floating_network_id, call[1].values()) |
| self.assertEqual(self.mock_fip_network.mock.call_count, 0) |
| # SG calls |
| mock_create_sg.assert_called_once() |
| # Resources |
| for resource in ['keypair', 'floating_ip', 'security_group']: |
| self.assertIn(resource, resources) |
| self.assertEqual(FAKE_KEYPAIR['keypair'], resources['keypair']) |
| self.assertEqual(FAKE_SECURITY_GROUP['security_group'], |
| resources['security_group']) |
| self.assertEqual(FAKE_FIP_NOVA_NET['floating_ip'], |
| resources['floating_ip']) |
| |
| @mock.patch.object(vr, 'create_ssh_security_group', |
| return_value=FAKE_SECURITY_GROUP['security_group']) |
| def test_create_validation_resources_neutron(self, mock_create_sg): |
| expected_floating_network_id = 'my_fni' |
| expected_floating_network_name = 'my_fnn' |
| resources = vr.create_validation_resources( |
| self.os, keypair=True, floating_ip=True, security_group=True, |
| security_group_rules=True, ethertype='IPv6', use_neutron=True, |
| floating_network_id=expected_floating_network_id, |
| floating_network_name=expected_floating_network_name) |
| # Keypair calls |
| self.assertGreater(self.mock_kp.mock.call_count, 0) |
| # Floating IP calls |
| self.assertEqual(self.mock_fip_compute.mock.call_count, 0) |
| self.assertGreater(self.mock_fip_network.mock.call_count, 0) |
| for call in self.mock_fip_compute.mock.call_args_list[1:]: |
| self.assertIn(expected_floating_network_id, call[1].values()) |
| self.assertNotIn(expected_floating_network_name, call[1].values()) |
| # SG calls |
| mock_create_sg.assert_called_once() |
| # Resources |
| for resource in ['keypair', 'floating_ip', 'security_group']: |
| self.assertIn(resource, resources) |
| self.assertEqual(FAKE_KEYPAIR['keypair'], resources['keypair']) |
| self.assertEqual(FAKE_SECURITY_GROUP['security_group'], |
| resources['security_group']) |
| self.assertIn('ip', resources['floating_ip']) |
| self.assertEqual(resources['floating_ip']['ip'], |
| FAKE_FIP_NEUTRON['floatingip']['floating_ip_address']) |
| self.assertEqual(resources['floating_ip']['id'], |
| FAKE_FIP_NEUTRON['floatingip']['id']) |
| |
| |
| class TestClearValidationResourcesFixture(base.TestCase): |
| |
| def setUp(self): |
| super(TestClearValidationResourcesFixture, self).setUp() |
| self.useFixture(registry_fixture.RegistryFixture()) |
| self.mock_sg_compute = self.useFixture(fixtures.MockPatch( |
| SG_CLIENT % ('compute', 'delete_security_group'), autospec=True)) |
| self.mock_sg_network = self.useFixture(fixtures.MockPatch( |
| SG_CLIENT % ('network', 'delete_security_group'), autospec=True)) |
| self.mock_sg_wait_compute = self.useFixture(fixtures.MockPatch( |
| SG_CLIENT % ('compute', 'wait_for_resource_deletion'), |
| autospec=True)) |
| self.mock_sg_wait_network = self.useFixture(fixtures.MockPatch( |
| SG_CLIENT % ('network', 'wait_for_resource_deletion'), |
| autospec=True)) |
| self.mock_kp = self.useFixture(fixtures.MockPatch( |
| KP_CLIENT % 'delete_keypair', autospec=True)) |
| self.mock_fip_compute = self.useFixture(fixtures.MockPatch( |
| FIP_CLIENT % ('compute', 'delete_floating_ip'), autospec=True)) |
| self.mock_fip_network = self.useFixture(fixtures.MockPatch( |
| FIP_CLIENT % ('network', 'delete_floatingip'), autospec=True)) |
| self.os = clients.ServiceClients( |
| fake_credentials.FakeKeystoneV3Credentials(), 'fake_uri') |
| |
| def test_clear_validation_resources_nova_net(self): |
| vr.clear_validation_resources( |
| self.os, |
| floating_ip=FAKE_FIP_NOVA_NET['floating_ip'], |
| security_group=FAKE_SECURITY_GROUP['security_group'], |
| keypair=FAKE_KEYPAIR['keypair'], |
| use_neutron=False) |
| self.assertGreater(self.mock_kp.mock.call_count, 0) |
| for call in self.mock_kp.mock.call_args_list[1:]: |
| self.assertIn(FAKE_KEYPAIR['keypair']['name'], call[1].values()) |
| self.assertGreater(self.mock_sg_compute.mock.call_count, 0) |
| for call in self.mock_sg_compute.mock.call_args_list[1:]: |
| self.assertIn(FAKE_SECURITY_GROUP['security_group']['id'], |
| call[1].values()) |
| self.assertGreater(self.mock_sg_wait_compute.mock.call_count, 0) |
| for call in self.mock_sg_wait_compute.mock.call_args_list[1:]: |
| self.assertIn(FAKE_SECURITY_GROUP['security_group']['id'], |
| call[1].values()) |
| self.assertEqual(self.mock_sg_network.mock.call_count, 0) |
| self.assertEqual(self.mock_sg_wait_network.mock.call_count, 0) |
| self.assertGreater(self.mock_fip_compute.mock.call_count, 0) |
| for call in self.mock_fip_compute.mock.call_args_list[1:]: |
| self.assertIn(FAKE_FIP_NOVA_NET['floating_ip']['id'], |
| call[1].values()) |
| self.assertEqual(self.mock_fip_network.mock.call_count, 0) |
| |
| def test_clear_validation_resources_neutron(self): |
| vr.clear_validation_resources( |
| self.os, |
| floating_ip=FAKE_FIP_NEUTRON['floatingip'], |
| security_group=FAKE_SECURITY_GROUP['security_group'], |
| keypair=FAKE_KEYPAIR['keypair'], |
| use_neutron=True) |
| self.assertGreater(self.mock_kp.mock.call_count, 0) |
| for call in self.mock_kp.mock.call_args_list[1:]: |
| self.assertIn(FAKE_KEYPAIR['keypair']['name'], call[1].values()) |
| self.assertGreater(self.mock_sg_network.mock.call_count, 0) |
| for call in self.mock_sg_network.mock.call_args_list[1:]: |
| self.assertIn(FAKE_SECURITY_GROUP['security_group']['id'], |
| call[1].values()) |
| self.assertGreater(self.mock_sg_wait_network.mock.call_count, 0) |
| for call in self.mock_sg_wait_network.mock.call_args_list[1:]: |
| self.assertIn(FAKE_SECURITY_GROUP['security_group']['id'], |
| call[1].values()) |
| self.assertEqual(self.mock_sg_compute.mock.call_count, 0) |
| self.assertEqual(self.mock_sg_wait_compute.mock.call_count, 0) |
| self.assertGreater(self.mock_fip_network.mock.call_count, 0) |
| for call in self.mock_fip_network.mock.call_args_list[1:]: |
| self.assertIn(FAKE_FIP_NEUTRON['floatingip']['id'], |
| call[1].values()) |
| self.assertEqual(self.mock_fip_compute.mock.call_count, 0) |
| |
| def test_clear_validation_resources_exceptions(self): |
| # Test that even with exceptions all cleanups are invoked and that only |
| # the first exception is reported. |
| # NOTE(andreaf) There's not way of knowing which exception is going to |
| # be raised first unless we enforce which resource is cleared first, |
| # which is not really interesting, but also not harmful. keypair first. |
| self.mock_kp.mock.side_effect = Exception('keypair exception') |
| self.mock_sg_network.mock.side_effect = Exception('sg exception') |
| self.mock_fip_network.mock.side_effect = Exception('fip exception') |
| with testtools.ExpectedException(Exception, value_re='keypair'): |
| vr.clear_validation_resources( |
| self.os, |
| floating_ip=FAKE_FIP_NEUTRON['floatingip'], |
| security_group=FAKE_SECURITY_GROUP['security_group'], |
| keypair=FAKE_KEYPAIR['keypair'], |
| use_neutron=True) |
| # Clients calls are still made, but not the wait call |
| self.assertGreater(self.mock_kp.mock.call_count, 0) |
| self.assertGreater(self.mock_sg_network.mock.call_count, 0) |
| self.assertGreater(self.mock_fip_network.mock.call_count, 0) |
| |
| def test_clear_validation_resources_wait_not_found_wait(self): |
| # Test that a not found on wait is not an exception |
| self.mock_sg_wait_network.mock.side_effect = lib_exc.NotFound('yay') |
| vr.clear_validation_resources( |
| self.os, |
| floating_ip=FAKE_FIP_NEUTRON['floatingip'], |
| security_group=FAKE_SECURITY_GROUP['security_group'], |
| keypair=FAKE_KEYPAIR['keypair'], |
| use_neutron=True) |
| # Clients calls are still made, but not the wait call |
| self.assertGreater(self.mock_kp.mock.call_count, 0) |
| self.assertGreater(self.mock_sg_network.mock.call_count, 0) |
| self.assertGreater(self.mock_sg_wait_network.mock.call_count, 0) |
| self.assertGreater(self.mock_fip_network.mock.call_count, 0) |
| |
| def test_clear_validation_resources_wait_not_found_delete(self): |
| # Test that a not found on delete is not an exception |
| self.mock_kp.mock.side_effect = lib_exc.NotFound('yay') |
| self.mock_sg_network.mock.side_effect = lib_exc.NotFound('yay') |
| self.mock_fip_network.mock.side_effect = lib_exc.NotFound('yay') |
| vr.clear_validation_resources( |
| self.os, |
| floating_ip=FAKE_FIP_NEUTRON['floatingip'], |
| security_group=FAKE_SECURITY_GROUP['security_group'], |
| keypair=FAKE_KEYPAIR['keypair'], |
| use_neutron=True) |
| # Clients calls are still made, but not the wait call |
| self.assertGreater(self.mock_kp.mock.call_count, 0) |
| self.assertGreater(self.mock_sg_network.mock.call_count, 0) |
| self.assertEqual(self.mock_sg_wait_network.mock.call_count, 0) |
| self.assertGreater(self.mock_fip_network.mock.call_count, 0) |
| |
| |
| class TestValidationResourcesFixture(base.TestCase): |
| |
| @mock.patch.object(vr, 'create_validation_resources', autospec=True) |
| def test_use_fixture(self, mock_vr): |
| exp_vr = dict(keypair='keypair', |
| floating_ip='floating_ip', |
| security_group='security_group') |
| mock_vr.return_value = exp_vr |
| exp_clients = 'clients' |
| exp_parameters = dict(keypair=True, floating_ip=True, |
| security_group=True, security_group_rules=True, |
| ethertype='v6', use_neutron=True, |
| floating_network_id='fnid', |
| floating_network_name='fnname') |
| # First mock cleanup |
| self.useFixture(fixtures.MockPatchObject( |
| vr, 'clear_validation_resources', autospec=True)) |
| # And then use vr fixture, so when the fixture is cleaned-up, the mock |
| # is still there |
| vr_fixture = self.useFixture(vr.ValidationResourcesFixture( |
| exp_clients, **exp_parameters)) |
| # Assert vr have been provisioned |
| mock_vr.assert_called_once_with(exp_clients, **exp_parameters) |
| # Assert vr have been setup in the fixture |
| self.assertEqual(exp_vr, vr_fixture.resources) |
| |
| @mock.patch.object(vr, 'clear_validation_resources', autospec=True) |
| @mock.patch.object(vr, 'create_validation_resources', autospec=True) |
| def test_use_fixture_context(self, mock_vr, mock_clear): |
| exp_vr = dict(keypair='keypair', |
| floating_ip='floating_ip', |
| security_group='security_group') |
| mock_vr.return_value = exp_vr |
| exp_clients = 'clients' |
| exp_parameters = dict(keypair=True, floating_ip=True, |
| security_group=True, security_group_rules=True, |
| ethertype='v6', use_neutron=True, |
| floating_network_id='fnid', |
| floating_network_name='fnname') |
| with vr.ValidationResourcesFixture(exp_clients, |
| **exp_parameters) as vr_fixture: |
| # Assert vr have been provisioned |
| mock_vr.assert_called_once_with(exp_clients, **exp_parameters) |
| # Assert vr have been setup in the fixture |
| self.assertEqual(exp_vr, vr_fixture.resources) |
| # After context manager is closed, clear is invoked |
| exp_vr['use_neutron'] = exp_parameters['use_neutron'] |
| mock_clear.assert_called_once_with(exp_clients, **exp_vr) |