KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use stop_mock method in lettuce-tools

Best Python code snippet using lettuce-tools_python

test_server_start_stop.py

Source:test_server_start_stop.py

1# Copyright (c) 2012 Midokura Japan K.K.2#3#    Licensed under the Apache License, Version 2.0 (the "License"); you may4#    not use this file except in compliance with the License. You may obtain5#    a copy of the License at6#7#         http://www.apache.org/licenses/LICENSE-2.08#9#    Unless required by applicable law or agreed to in writing, software10#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT11#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the12#    License for the specific language governing permissions and limitations13#    under the License.14import mock15from oslo_policy import policy as oslo_policy16import six17import webob18from nova.api.openstack.compute import servers \19        as server_v2120from nova.compute import api as compute_api21from nova import db22from nova import exception23from nova import policy24from nova import test25from nova.tests import fixtures as nova_fixtures26from nova.tests.unit.api.openstack import fakes27from nova.tests import uuidsentinel as uuids28class ServerStartStopTestV21(test.TestCase):29    def setUp(self):30        super(ServerStartStopTestV21, self).setUp()31        self._setup_controller()32        self.req = fakes.HTTPRequest.blank('')33        self.useFixture(nova_fixtures.SingleCellSimple())34        self.stub_out('nova.db.instance_get_by_uuid',35                      fakes.fake_instance_get())36    def _setup_controller(self):37        self.controller = server_v21.ServersController()38    @mock.patch.object(compute_api.API, 'start')39    def test_start(self, start_mock):40        body = dict(start="")41        self.controller._start_server(self.req, uuids.instance, body)42        start_mock.assert_called_once_with(mock.ANY, mock.ANY)43    @mock.patch.object(compute_api.API, 'start',44                       side_effect=exception.InstanceNotReady(45                           instance_id=uuids.instance))46    def test_start_not_ready(self, start_mock):47        body = dict(start="")48        self.assertRaises(webob.exc.HTTPConflict,49            self.controller._start_server, self.req, uuids.instance, body)50    @mock.patch.object(compute_api.API, 'start',51                       side_effect=exception.InstanceIsLocked(52                           instance_uuid=uuids.instance))53    def test_start_locked_server(self, start_mock):54        body = dict(start="")55        self.assertRaises(webob.exc.HTTPConflict,56            self.controller._start_server, self.req, uuids.instance, body)57    @mock.patch.object(compute_api.API, 'start',58                       side_effect=exception.InstanceIsLocked(59                           instance_uuid=uuids.instance))60    def test_start_invalid_state(self, start_mock):61        body = dict(start="")62        ex = self.assertRaises(webob.exc.HTTPConflict,63            self.controller._start_server, self.req, uuids.instance, body)64        self.assertIn('is locked', six.text_type(ex))65    @mock.patch.object(compute_api.API, 'stop')66    def test_stop(self, stop_mock):67        body = dict(stop="")68        self.controller._stop_server(self.req, uuids.instance, body)69        stop_mock.assert_called_once_with(mock.ANY, mock.ANY)70    @mock.patch.object(compute_api.API, 'stop',71                       side_effect=exception.InstanceNotReady(72                           instance_id=uuids.instance))73    def test_stop_not_ready(self, stop_mock):74        body = dict(stop="")75        self.assertRaises(webob.exc.HTTPConflict,76            self.controller._stop_server, self.req, uuids.instance, body)77    @mock.patch.object(compute_api.API, 'stop',78                       side_effect=exception.InstanceIsLocked(79                           instance_uuid=uuids.instance))80    def test_stop_locked_server(self, stop_mock):81        body = dict(stop="")82        ex = self.assertRaises(webob.exc.HTTPConflict,83            self.controller._stop_server, self.req, uuids.instance, body)84        self.assertIn('is locked', six.text_type(ex))85    @mock.patch.object(compute_api.API, 'stop',86                       side_effect=exception.InstanceIsLocked(87                           instance_uuid=uuids.instance))88    def test_stop_invalid_state(self, stop_mock):89        body = dict(start="")90        self.assertRaises(webob.exc.HTTPConflict,91            self.controller._stop_server, self.req, uuids.instance, body)92    @mock.patch.object(db, 'instance_get_by_uuid',93                       side_effect=exception.InstanceNotFound(94                           instance_id=uuids.instance))95    def test_start_with_bogus_id(self, get_mock):96        body = dict(start="")97        self.assertRaises(webob.exc.HTTPNotFound,98            self.controller._start_server, self.req, uuids.instance, body)99    @mock.patch.object(db, 'instance_get_by_uuid',100                       side_effect=exception.InstanceNotFound(101                           instance_id=uuids.instance))102    def test_stop_with_bogus_id(self, get_mock):103        body = dict(stop="")104        self.assertRaises(webob.exc.HTTPNotFound,105            self.controller._stop_server, self.req, uuids.instance, body)106class ServerStartStopPolicyEnforcementV21(test.TestCase):107    start_policy = "os_compute_api:servers:start"108    stop_policy = "os_compute_api:servers:stop"109    def setUp(self):110        super(ServerStartStopPolicyEnforcementV21, self).setUp()111        self.controller = server_v21.ServersController()112        self.req = fakes.HTTPRequest.blank('')113        self.useFixture(nova_fixtures.SingleCellSimple())114        self.stub_out(115            'nova.db.instance_get_by_uuid',116            fakes.fake_instance_get(117                project_id=self.req.environ['nova.context'].project_id))118    def test_start_policy_failed(self):119        rules = {120            self.start_policy: "project_id:non_fake"121        }122        policy.set_rules(oslo_policy.Rules.from_dict(rules))123        body = dict(start="")124        exc = self.assertRaises(exception.PolicyNotAuthorized,125                                self.controller._start_server,126                                self.req, uuids.instance, body)127        self.assertIn(self.start_policy, exc.format_message())128    def test_start_overridden_policy_failed_with_other_user_in_same_project(129        self):130        rules = {131            self.start_policy: "user_id:%(user_id)s"132        }133        policy.set_rules(oslo_policy.Rules.from_dict(rules))134        # Change the user_id in request context.135        self.req.environ['nova.context'].user_id = 'other-user'136        body = dict(start="")137        exc = self.assertRaises(exception.PolicyNotAuthorized,138                                self.controller._start_server,139                                self.req, uuids.instance, body)140        self.assertIn(self.start_policy, exc.format_message())141    @mock.patch('nova.compute.api.API.start')142    def test_start_overridden_policy_pass_with_same_user(self, start_mock):143        rules = {144            self.start_policy: "user_id:%(user_id)s"145        }146        policy.set_rules(oslo_policy.Rules.from_dict(rules))147        body = dict(start="")148        self.controller._start_server(self.req, uuids.instance, body)149        start_mock.assert_called_once_with(mock.ANY, mock.ANY)150    def test_stop_policy_failed_with_other_project(self):151        rules = {152            self.stop_policy: "project_id:%(project_id)s"153        }154        policy.set_rules(oslo_policy.Rules.from_dict(rules))155        body = dict(stop="")156        # Change the project_id in request context.157        self.req.environ['nova.context'].project_id = 'other-project'158        exc = self.assertRaises(exception.PolicyNotAuthorized,159                                self.controller._stop_server,160                                self.req, uuids.instance, body)161        self.assertIn(self.stop_policy, exc.format_message())162    @mock.patch('nova.compute.api.API.stop')163    def test_stop_overridden_policy_pass_with_same_project(self, stop_mock):164        rules = {165            self.stop_policy: "project_id:%(project_id)s"166        }167        policy.set_rules(oslo_policy.Rules.from_dict(rules))168        body = dict(stop="")169        self.controller._stop_server(self.req, uuids.instance, body)170        stop_mock.assert_called_once_with(mock.ANY, mock.ANY)171    def test_stop_overridden_policy_failed_with_other_user_in_same_project(172        self):173        rules = {174            self.stop_policy: "user_id:%(user_id)s"175        }176        policy.set_rules(oslo_policy.Rules.from_dict(rules))177        # Change the user_id in request context.178        self.req.environ['nova.context'].user_id = 'other-user'179        body = dict(stop="")180        exc = self.assertRaises(exception.PolicyNotAuthorized,181                                self.controller._stop_server,182                                self.req, uuids.instance, body)183        self.assertIn(self.stop_policy, exc.format_message())184    @mock.patch('nova.compute.api.API.stop')185    def test_stop_overridden_policy_pass_with_same_user(self, stop_mock):186        rules = {187            self.stop_policy: "user_id:%(user_id)s"188        }189        policy.set_rules(oslo_policy.Rules.from_dict(rules))190        body = dict(stop="")191        self.controller._stop_server(self.req, uuids.instance, body)...

test_services.py

Source:test_services.py

1import unittest2from unittest.mock import (3    AsyncMock,4)5from aiomisc import (6    Service,7)8from minos.common.testing import (9    PostgresAsyncTestCase,10)11from minos.networks import (12    PeriodicTaskScheduler,13    PeriodicTaskSchedulerService,14)15from tests.utils import (16    BASE_PATH,17)18class TestPeriodicTaskSchedulerService(PostgresAsyncTestCase):19    CONFIG_FILE_PATH = BASE_PATH / "test_config.yml"20    def test_is_instance(self):21        service = PeriodicTaskSchedulerService(config=self.config)22        self.assertIsInstance(service, Service)23    def test_dispatcher(self):24        service = PeriodicTaskSchedulerService(config=self.config)25        self.assertIsInstance(service.scheduler, PeriodicTaskScheduler)26    async def test_start_stop(self):27        service = PeriodicTaskSchedulerService(config=self.config)28        setup_mock = AsyncMock()29        destroy_mock = AsyncMock()30        start_mock = AsyncMock()31        stop_mock = AsyncMock()32        service.scheduler.setup = setup_mock33        service.scheduler.destroy = destroy_mock34        service.scheduler.start = start_mock35        service.scheduler.stop = stop_mock36        await service.start()37        self.assertEqual(1, setup_mock.call_count)38        self.assertEqual(1, start_mock.call_count)39        self.assertEqual(0, stop_mock.call_count)40        self.assertEqual(0, destroy_mock.call_count)41        setup_mock.reset_mock()42        destroy_mock.reset_mock()43        start_mock.reset_mock()44        stop_mock.reset_mock()45        await service.stop()46        self.assertEqual(0, setup_mock.call_count)47        self.assertEqual(0, start_mock.call_count)48        self.assertEqual(1, stop_mock.call_count)49        self.assertEqual(1, destroy_mock.call_count)50if __name__ == "__main__":...

Automation Testing Tutorials

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.