Best Python code snippet using slash
test_test.py
Source:test_test.py  
1# Copyright 2017 IBM Corp2# All Rights Reserved.3#4#    Licensed under the Apache License, Version 2.0 (the "License"); you may5#    not use this file except in compliance with the License. You may obtain6#    a copy of the License at7#8#         http://www.apache.org/licenses/LICENSE-2.09#10#    Unless required by applicable law or agreed to in writing, software11#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT12#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the13#    License for the specific language governing permissions and limitations14#    under the License.15import os16import sys17import mock18from oslo_config import cfg19import testtools20from tempest import clients21from tempest import config22from tempest.lib.common import validation_resources as vr23from tempest.lib import exceptions as lib_exc24from tempest import test25from tempest.tests import base26from tempest.tests import fake_config27from tempest.tests.lib import fake_credentials28from tempest.tests.lib.services import registry_fixture29if sys.version_info >= (2, 7):30    import unittest31else:32    import unittest2 as unittest33class LoggingTestResult(testtools.TestResult):34    def __init__(self, log, *args, **kwargs):35        super(LoggingTestResult, self).__init__(*args, **kwargs)36        self.log = log37    def addError(self, test, err=None, details=None):38        self.log.append((test, err, details))39class TestValidationResources(base.TestCase):40    validation_resources_module = 'tempest.lib.common.validation_resources'41    def setUp(self):42        super(TestValidationResources, self).setUp()43        self.useFixture(fake_config.ConfigFixture())44        self.useFixture(registry_fixture.RegistryFixture())45        self.patchobject(config, 'TempestConfigPrivate',46                         fake_config.FakePrivate)47        class TestTestClass(test.BaseTestCase):48            pass49        self.test_test_class = TestTestClass50    def test_validation_resources_no_validation(self):51        cfg.CONF.set_default('run_validation', False, 'validation')52        creds = fake_credentials.FakeKeystoneV3Credentials()53        osclients = clients.Manager(creds)54        vr = self.test_test_class.get_class_validation_resources(osclients)55        self.assertIsNone(vr)56    def test_validation_resources_exists(self):57        cfg.CONF.set_default('run_validation', True, 'validation')58        creds = fake_credentials.FakeKeystoneV3Credentials()59        osclients = clients.Manager(creds)60        expected_vr = 'expected_validation_resources'61        self.test_test_class._validation_resources[osclients] = expected_vr62        obtained_vr = self.test_test_class.get_class_validation_resources(63            osclients)64        self.assertEqual(expected_vr, obtained_vr)65    @mock.patch(validation_resources_module + '.create_validation_resources',66                autospec=True)67    def test_validation_resources_new(self, mock_create_vr):68        cfg.CONF.set_default('run_validation', True, 'validation')69        cfg.CONF.set_default('neutron', True, 'service_available')70        creds = fake_credentials.FakeKeystoneV3Credentials()71        osclients = clients.Manager(creds)72        expected_vr = {'expected_validation_resources': None}73        mock_create_vr.return_value = expected_vr74        with mock.patch.object(75                self.test_test_class,76                'addClassResourceCleanup') as mock_add_class_cleanup:77            obtained_vr = self.test_test_class.get_class_validation_resources(78                osclients)79            self.assertEqual(1, mock_add_class_cleanup.call_count)80            self.assertEqual(mock.call(vr.clear_validation_resources,81                                       osclients,82                                       use_neutron=True,83                                       **expected_vr),84                             mock_add_class_cleanup.call_args)85        self.assertEqual(mock_create_vr.call_count, 1)86        self.assertIn(osclients, mock_create_vr.call_args_list[0][0])87        self.assertEqual(expected_vr, obtained_vr)88        self.assertIn(osclients, self.test_test_class._validation_resources)89        self.assertEqual(expected_vr,90                         self.test_test_class._validation_resources[osclients])91    def test_validation_resources_invalid_config(self):92        invalid_version = 99993        cfg.CONF.set_default('run_validation', True, 'validation')94        cfg.CONF.set_default('ip_version_for_ssh', invalid_version,95                             'validation')96        cfg.CONF.set_default('neutron', True, 'service_available')97        creds = fake_credentials.FakeKeystoneV3Credentials()98        osclients = clients.Manager(creds)99        with testtools.ExpectedException(100                lib_exc.InvalidConfiguration,101                value_re='^.*\n.*' + str(invalid_version)):102            self.test_test_class.get_class_validation_resources(osclients)103    @mock.patch(validation_resources_module + '.create_validation_resources',104                autospec=True)105    def test_validation_resources_invalid_config_nova_net(self,106                                                          mock_create_vr):107        invalid_version = 999108        cfg.CONF.set_default('run_validation', True, 'validation')109        cfg.CONF.set_default('ip_version_for_ssh', invalid_version,110                             'validation')111        cfg.CONF.set_default('neutron', False, 'service_available')112        creds = fake_credentials.FakeKeystoneV3Credentials()113        osclients = clients.Manager(creds)114        expected_vr = {'expected_validation_resources': None}115        mock_create_vr.return_value = expected_vr116        obtained_vr = self.test_test_class.get_class_validation_resources(117            osclients)118        self.assertEqual(mock_create_vr.call_count, 1)119        self.assertIn(osclients, mock_create_vr.call_args_list[0][0])120        self.assertEqual(expected_vr, obtained_vr)121        self.assertIn(osclients, self.test_test_class._validation_resources)122        self.assertEqual(expected_vr,123                         self.test_test_class._validation_resources[osclients])124    @mock.patch(validation_resources_module + '.create_validation_resources',125                autospec=True)126    @mock.patch(validation_resources_module + '.clear_validation_resources',127                autospec=True)128    def test_validation_resources_fixture(self, mock_clean_vr, mock_create_vr):129        class TestWithRun(self.test_test_class):130            def runTest(self):131                pass132        cfg.CONF.set_default('run_validation', True, 'validation')133        test_case = TestWithRun()134        creds = fake_credentials.FakeKeystoneV3Credentials()135        osclients = clients.Manager(creds)136        test_case.get_test_validation_resources(osclients)137        self.assertEqual(1, mock_create_vr.call_count)138        self.assertEqual(0, mock_clean_vr.call_count)139class TestSetNetworkResources(base.TestCase):140    def setUp(self):141        super(TestSetNetworkResources, self).setUp()142        class ParentTest(test.BaseTestCase):143            @classmethod144            def setup_credentials(cls):145                cls.set_network_resources(dhcp=True)146                super(ParentTest, cls).setup_credentials()147            def runTest(self):148                pass149        self.parent_class = ParentTest150    def test_set_network_resources_child_only(self):151        class ChildTest(self.parent_class):152            @classmethod153            def setup_credentials(cls):154                cls.set_network_resources(router=True)155                super(ChildTest, cls).setup_credentials()156        child_test = ChildTest()157        child_test.setUpClass()158        # Assert that the parents network resources are not set159        self.assertFalse(child_test._network_resources['dhcp'])160        # Assert that the child network resources are set161        self.assertTrue(child_test._network_resources['router'])162    def test_set_network_resources_right_order(self):163        class ChildTest(self.parent_class):164            @classmethod165            def setup_credentials(cls):166                super(ChildTest, cls).setup_credentials()167                cls.set_network_resources(router=True)168        child_test = ChildTest()169        with testtools.ExpectedException(RuntimeError,170                                         value_re='set_network_resources'):171            child_test.setUpClass()172    def test_set_network_resources_children(self):173        class ChildTest(self.parent_class):174            @classmethod175            def setup_credentials(cls):176                cls.set_network_resources(router=True)177                super(ChildTest, cls).setup_credentials()178        class GrandChildTest(ChildTest):179            pass180        # Invoke setupClass on both and check that the setup_credentials181        # call check mechanism does not report any false negative.182        child_test = ChildTest()183        child_test.setUpClass()184        grandchild_test = GrandChildTest()185        grandchild_test.setUpClass()186class TestTempestBaseTestClass(base.TestCase):187    def setUp(self):188        super(TestTempestBaseTestClass, self).setUp()189        self.useFixture(fake_config.ConfigFixture())190        self.patchobject(config, 'TempestConfigPrivate',191                         fake_config.FakePrivate)192        class ParentTest(test.BaseTestCase):193            def runTest(self):194                pass195        self.parent_test = ParentTest196    def test_resource_cleanup(self):197        cfg.CONF.set_default('neutron', False, 'service_available')198        exp_args = (1, 2,)199        exp_kwargs = {'a': 1, 'b': 2}200        mock1 = mock.Mock()201        mock2 = mock.Mock()202        exp_functions = [mock1, mock2]203        class TestWithCleanups(self.parent_test):204            @classmethod205            def resource_setup(cls):206                for fn in exp_functions:207                    cls.addClassResourceCleanup(fn, *exp_args,208                                                **exp_kwargs)209        test_cleanups = TestWithCleanups()210        suite = unittest.TestSuite((test_cleanups,))211        log = []212        result = LoggingTestResult(log)213        suite.run(result)214        # No exception raised - error log is empty215        self.assertFalse(log)216        # All stacked resource cleanups invoked217        mock1.assert_called_once_with(*exp_args, **exp_kwargs)218        mock2.assert_called_once_with(*exp_args, **exp_kwargs)219        # Cleanup stack is empty220        self.assertEqual(0, len(test_cleanups._class_cleanups))221    def test_resource_cleanup_failures(self):222        cfg.CONF.set_default('neutron', False, 'service_available')223        exp_args = (1, 2,)224        exp_kwargs = {'a': 1, 'b': 2}225        mock1 = mock.Mock()226        mock1.side_effect = Exception('mock1 resource cleanup failure')227        mock2 = mock.Mock()228        mock3 = mock.Mock()229        mock3.side_effect = Exception('mock3 resource cleanup failure')230        exp_functions = [mock1, mock2, mock3]231        class TestWithFailingCleanups(self.parent_test):232            @classmethod233            def resource_setup(cls):234                for fn in exp_functions:235                    cls.addClassResourceCleanup(fn, *exp_args,236                                                **exp_kwargs)237        test_cleanups = TestWithFailingCleanups()238        suite = unittest.TestSuite((test_cleanups,))239        log = []240        result = LoggingTestResult(log)241        suite.run(result)242        # One multiple exception captured243        self.assertEqual(1, len(log))244        # [0]: test, err, details [1] -> exc_info245        # Type, Exception, traceback [1] -> MultipleException246        found_exc = log[0][1][1]247        self.assertTrue(isinstance(found_exc, testtools.MultipleExceptions))248        self.assertEqual(2, len(found_exc.args))249        # Each arg is exc_info - match messages and order250        self.assertIn('mock3 resource', str(found_exc.args[0][1]))251        self.assertIn('mock1 resource', str(found_exc.args[1][1]))252        # All stacked resource cleanups invoked253        mock1.assert_called_once_with(*exp_args, **exp_kwargs)254        mock2.assert_called_once_with(*exp_args, **exp_kwargs)255        # Cleanup stack is empty256        self.assertEqual(0, len(test_cleanups._class_cleanups))257    def test_super_resource_cleanup_not_invoked(self):258        class BadResourceCleanup(self.parent_test):259            @classmethod260            def resource_cleanup(cls):261                pass262        bad_class = BadResourceCleanup()263        suite = unittest.TestSuite((bad_class,))264        log = []265        result = LoggingTestResult(log)266        suite.run(result)267        # One multiple exception captured268        self.assertEqual(1, len(log))269        # [0]: test, err, details [1] -> exc_info270        # Type, Exception, traceback [1] -> RuntimeError271        found_exc = log[0][1][1]272        self.assertTrue(isinstance(found_exc, RuntimeError))273        self.assertIn(BadResourceCleanup.__name__, str(found_exc))274    def test_super_skip_checks_not_invoked(self):275        class BadSkipChecks(self.parent_test):276            @classmethod277            def skip_checks(cls):278                pass279        bad_class = BadSkipChecks()280        with testtools.ExpectedException(281                RuntimeError,282                value_re='^.* ' + BadSkipChecks.__name__):283            bad_class.setUpClass()284    def test_super_setup_credentials_not_invoked(self):285        class BadSetupCredentials(self.parent_test):286            @classmethod287            def skip_checks(cls):288                pass289        bad_class = BadSetupCredentials()290        with testtools.ExpectedException(291                RuntimeError,292                value_re='^.* ' + BadSetupCredentials.__name__):293            bad_class.setUpClass()294    def test_grandparent_skip_checks_not_invoked(self):295        class BadSkipChecks(self.parent_test):296            @classmethod297            def skip_checks(cls):298                pass299        class SonOfBadSkipChecks(BadSkipChecks):300            pass301        bad_class = SonOfBadSkipChecks()302        with testtools.ExpectedException(303                RuntimeError,304                value_re='^.* ' + SonOfBadSkipChecks.__name__):305            bad_class.setUpClass()306    @mock.patch('tempest.common.credentials_factory.is_admin_available',307                autospec=True, return_value=True)308    def test_skip_checks_admin(self, mock_iaa):309        identity_version = 'identity_version'310        class NeedAdmin(self.parent_test):311            credentials = ['admin']312            @classmethod313            def get_identity_version(cls):314                return identity_version315        NeedAdmin().skip_checks()316        mock_iaa.assert_called_once_with('identity_version')317    @mock.patch('tempest.common.credentials_factory.is_admin_available',318                autospec=True, return_value=False)319    def test_skip_checks_admin_not_available(self, mock_iaa):320        identity_version = 'identity_version'321        class NeedAdmin(self.parent_test):322            credentials = ['admin']323            @classmethod324            def get_identity_version(cls):325                return identity_version326        with testtools.ExpectedException(testtools.testcase.TestSkipped):327            NeedAdmin().skip_checks()328        mock_iaa.assert_called_once_with('identity_version')329    def test_skip_checks_identity_v2_not_available(self):330        cfg.CONF.set_default('api_v2', False, 'identity-feature-enabled')331        class NeedV2(self.parent_test):332            identity_version = 'v2'333        with testtools.ExpectedException(testtools.testcase.TestSkipped):334            NeedV2().skip_checks()335    def test_skip_checks_identity_v3_not_available(self):336        cfg.CONF.set_default('api_v3', False, 'identity-feature-enabled')337        class NeedV3(self.parent_test):338            identity_version = 'v3'339        with testtools.ExpectedException(testtools.testcase.TestSkipped):340            NeedV3().skip_checks()341    def test_setup_credentials_all(self):342        expected_creds = ['string', ['list', 'role1', 'role2']]343        class AllCredentials(self.parent_test):344            credentials = expected_creds345        expected_clients = 'clients'346        with mock.patch.object(347                AllCredentials,348                'get_client_manager') as mock_get_client_manager:349            mock_get_client_manager.return_value = expected_clients350            all_creds = AllCredentials()351            all_creds.setup_credentials()352        self.assertTrue(hasattr(all_creds, 'os_string'))353        self.assertEqual(expected_clients, all_creds.os_string)354        self.assertTrue(hasattr(all_creds, 'os_roles_list'))355        self.assertEqual(expected_clients, all_creds.os_roles_list)356        self.assertEqual(2, mock_get_client_manager.call_count)357        self.assertEqual(358            expected_creds[0],359            mock_get_client_manager.mock_calls[0][2]['credential_type'])360        self.assertEqual(361            expected_creds[1][1:],362            mock_get_client_manager.mock_calls[1][2]['roles'])363    def test_setup_class_overwritten(self):364        class OverridesSetup(self.parent_test):365            @classmethod366            def setUpClass(cls):  # noqa367                pass368        overrides_setup = OverridesSetup()369        suite = unittest.TestSuite((overrides_setup,))370        log = []371        result = LoggingTestResult(log)372        suite.run(result)373        # Record 0, test (error holder). The error generates during test run.374        self.assertIn('runTest', str(log[0][0]))375        # Record 0, traceback376        self.assertRegex(377            str(log[0][2]['traceback']).replace('\n', ' '),378            RuntimeError.__name__ + ': .* ' + OverridesSetup.__name__)379class TestTempestBaseTestClassFixtures(base.TestCase):380    SETUP_FIXTURES = [test.BaseTestCase.setUpClass.__name__,381                      test.BaseTestCase.skip_checks.__name__,382                      test.BaseTestCase.setup_credentials.__name__,383                      test.BaseTestCase.setup_clients.__name__,384                      test.BaseTestCase.resource_setup.__name__]385    TEARDOWN_FIXTURES = [test.BaseTestCase.tearDownClass.__name__,386                         test.BaseTestCase.resource_cleanup.__name__,387                         test.BaseTestCase.clear_credentials.__name__]388    def setUp(self):389        super(TestTempestBaseTestClassFixtures, self).setUp()390        self.mocks = {}391        for fix in self.SETUP_FIXTURES + self.TEARDOWN_FIXTURES:392            self.mocks[fix] = mock.Mock()393        def tracker_builder(name):394            def tracker(cls):395                # Track that the fixture was invoked396                cls.fixtures_invoked.append(name)397                # Run the fixture398                getattr(super(TestWithClassFixtures, cls), name)()399                # Run a mock we can use for side effects400                self.mocks[name]()401            return tracker402        class TestWithClassFixtures(test.BaseTestCase):403            credentials = []404            fixtures_invoked = []405            def runTest(_self):406                pass407        # Decorate all test class fixtures with tracker_builder408        for method_name in self.SETUP_FIXTURES + self.TEARDOWN_FIXTURES:409            setattr(TestWithClassFixtures, method_name,410                    classmethod(tracker_builder(method_name)))411        self.test = TestWithClassFixtures()412    def test_no_error_flow(self):413        # If all setup fixtures are executed, all cleanup fixtures are414        # executed too415        suite = unittest.TestSuite((self.test,))416        log = []417        result = LoggingTestResult(log)418        suite.run(result)419        self.assertEqual(self.SETUP_FIXTURES + self.TEARDOWN_FIXTURES,420                         self.test.fixtures_invoked)421    def test_skip_only(self):422        # If a skip condition is hit in the test, no credentials or resource423        # is provisioned / cleaned-up424        self.mocks['skip_checks'].side_effect = (425            testtools.TestCase.skipException())426        suite = unittest.TestSuite((self.test,))427        log = []428        result = LoggingTestResult(log)429        suite.run(result)430        # If we trigger a skip condition, teardown is not invoked at all431        self.assertEqual((self.SETUP_FIXTURES[:2] +432                          [self.TEARDOWN_FIXTURES[0]]),433                         self.test.fixtures_invoked)434    def test_skip_credentials_fails(self):435        expected_exc = 'sc exploded'436        self.mocks['setup_credentials'].side_effect = Exception(expected_exc)437        suite = unittest.TestSuite((self.test,))438        log = []439        result = LoggingTestResult(log)440        suite.run(result)441        # If setup_credentials explodes, we invoked teardown class and442        # clear credentials, and re-raise443        self.assertEqual((self.SETUP_FIXTURES[:3] +444                          [self.TEARDOWN_FIXTURES[i] for i in (0, 2)]),445                         self.test.fixtures_invoked)446        found_exc = log[0][1][1]447        self.assertIn(expected_exc, str(found_exc))448    def test_skip_credentials_fails_clear_fails(self):449        # If cleanup fails on failure, we log the exception and do not450        # re-raise it. Note that since the exception happens outside of451        # the Tempest test setUp, logging is not captured on the Tempest452        # test side, it will be captured by the unit test instead.453        expected_exc = 'sc exploded'454        clear_exc = 'clear exploded'455        self.mocks['setup_credentials'].side_effect = Exception(expected_exc)456        self.mocks['clear_credentials'].side_effect = Exception(clear_exc)457        suite = unittest.TestSuite((self.test,))458        log = []459        result = LoggingTestResult(log)460        suite.run(result)461        # If setup_credentials explodes, we invoked teardown class and462        # clear credentials, and re-raise463        self.assertEqual((self.SETUP_FIXTURES[:3] +464                          [self.TEARDOWN_FIXTURES[i] for i in (0, 2)]),465                         self.test.fixtures_invoked)466        found_exc = log[0][1][1]467        self.assertIn(expected_exc, str(found_exc))468        # Since log capture depends on OS_LOG_CAPTURE, we can only assert if469        # logging was captured470        if os.environ.get('OS_LOG_CAPTURE'):471            self.assertIn(clear_exc, self.log_fixture.logger.output)472    def test_skip_credentials_clients_resources_credentials_clear_fails(self):473        # If cleanup fails with no previous failure, we re-raise the exception.474        expected_exc = 'clear exploded'475        self.mocks['clear_credentials'].side_effect = Exception(expected_exc)476        suite = unittest.TestSuite((self.test,))477        log = []478        result = LoggingTestResult(log)479        suite.run(result)480        # If setup_credentials explodes, we invoked teardown class and481        # clear credentials, and re-raise482        self.assertEqual(self.SETUP_FIXTURES + self.TEARDOWN_FIXTURES,483                         self.test.fixtures_invoked)484        found_exc = log[0][1][1]485        self.assertIn(expected_exc, str(found_exc))486    def test_skip_credentials_clients_fails(self):487        expected_exc = 'clients exploded'488        self.mocks['setup_clients'].side_effect = Exception(expected_exc)489        suite = unittest.TestSuite((self.test,))490        log = []491        result = LoggingTestResult(log)492        suite.run(result)493        # If setup_clients explodes, we invoked teardown class and494        # clear credentials, and re-raise495        self.assertEqual((self.SETUP_FIXTURES[:4] +496                          [self.TEARDOWN_FIXTURES[i] for i in (0, 2)]),497                         self.test.fixtures_invoked)498        found_exc = log[0][1][1]499        self.assertIn(expected_exc, str(found_exc))500    def test_skip_credentials_clients_resources_fails(self):501        expected_exc = 'resource setup exploded'502        self.mocks['resource_setup'].side_effect = Exception(expected_exc)503        suite = unittest.TestSuite((self.test,))504        log = []505        result = LoggingTestResult(log)506        suite.run(result)507        # If resource_setup explodes, we invoked teardown class and508        # clear credentials and resource cleanup, and re-raise509        self.assertEqual(self.SETUP_FIXTURES + self.TEARDOWN_FIXTURES,510                         self.test.fixtures_invoked)511        found_exc = log[0][1][1]...Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!
