KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference
How to use my_task method in locust

Best Python code snippet using locust
test_task.py
Source:test_task.py
1# -*- coding: utf-8 -*-2#    Copyright 2015 Hewlett-Packard Development Company, L.P.3#    Copyright (C) 2013 Yahoo! Inc. All Rights Reserved.4#5#    Licensed under the Apache License, Version 2.0 (the "License"); you may6#    not use this file except in compliance with the License. You may obtain7#    a copy of the License at8#9#         http://www.apache.org/licenses/LICENSE-2.010#11#    Unless required by applicable law or agreed to in writing, software12#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT13#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the14#    License for the specific language governing permissions and limitations15#    under the License.16from taskflow import task17from taskflow import test18from taskflow.test import mock19from taskflow.types import notifier20class MyTask(task.Task):21    def execute(self, context, spam, eggs):22        pass23class KwargsTask(task.Task):24    def execute(self, spam, **kwargs):25        pass26class DefaultArgTask(task.Task):27    def execute(self, spam, eggs=()):28        pass29class DefaultProvidesTask(task.Task):30    default_provides = 'def'31    def execute(self):32        return None33class ProgressTask(task.Task):34    def execute(self, values, **kwargs):35        for value in values:36            self.update_progress(value)37class SeparateRevertTask(task.Task):38    def execute(self, execute_arg):39        pass40    def revert(self, revert_arg, result, flow_failures):41        pass42class SeparateRevertOptionalTask(task.Task):43    def execute(self, execute_arg=None):44        pass45    def revert(self, result, flow_failures, revert_arg=None):46        pass47class RevertKwargsTask(task.Task):48    def execute(self, execute_arg1, execute_arg2):49        pass50    def revert(self, execute_arg1, *args, **kwargs):51        pass52class TaskTest(test.TestCase):53    def test_passed_name(self):54        my_task = MyTask(name='my name')55        self.assertEqual('my name', my_task.name)56    def test_generated_name(self):57        my_task = MyTask()58        self.assertEqual('%s.%s' % (__name__, 'MyTask'),59                         my_task.name)60    def test_task_str(self):61        my_task = MyTask(name='my')62        self.assertEqual('my==1.0', str(my_task))63    def test_task_repr(self):64        my_task = MyTask(name='my')65        self.assertEqual('<%s.MyTask my==1.0>' % __name__, repr(my_task))66    def test_no_provides(self):67        my_task = MyTask()68        self.assertEqual({}, my_task.save_as)69    def test_provides(self):70        my_task = MyTask(provides='food')71        self.assertEqual({'food': None}, my_task.save_as)72    def test_multi_provides(self):73        my_task = MyTask(provides=('food', 'water'))74        self.assertEqual({'food': 0, 'water': 1}, my_task.save_as)75    def test_unpack(self):76        my_task = MyTask(provides=('food',))77        self.assertEqual({'food': 0}, my_task.save_as)78    def test_bad_provides(self):79        self.assertRaisesRegex(TypeError, '^Atom provides',80                               MyTask, provides=object())81    def test_requires_by_default(self):82        my_task = MyTask()83        expected = {84            'spam': 'spam',85            'eggs': 'eggs',86            'context': 'context'87        }88        self.assertEqual(expected,89                         my_task.rebind)90        self.assertEqual(set(['spam', 'eggs', 'context']),91                         my_task.requires)92    def test_requires_amended(self):93        my_task = MyTask(requires=('spam', 'eggs'))94        expected = {95            'spam': 'spam',96            'eggs': 'eggs',97            'context': 'context'98        }99        self.assertEqual(expected, my_task.rebind)100    def test_requires_explicit(self):101        my_task = MyTask(auto_extract=False,102                         requires=('spam', 'eggs', 'context'))103        expected = {104            'spam': 'spam',105            'eggs': 'eggs',106            'context': 'context'107        }108        self.assertEqual(expected, my_task.rebind)109    def test_requires_explicit_not_enough(self):110        self.assertRaisesRegex(ValueError, '^Missing arguments',111                               MyTask,112                               auto_extract=False, requires=('spam', 'eggs'))113    def test_requires_ignores_optional(self):114        my_task = DefaultArgTask()115        self.assertEqual(set(['spam']), my_task.requires)116        self.assertEqual(set(['eggs']), my_task.optional)117    def test_requires_allows_optional(self):118        my_task = DefaultArgTask(requires=('spam', 'eggs'))119        self.assertEqual(set(['spam', 'eggs']), my_task.requires)120        self.assertEqual(set(), my_task.optional)121    def test_rebind_includes_optional(self):122        my_task = DefaultArgTask()123        expected = {124            'spam': 'spam',125            'eggs': 'eggs',126        }127        self.assertEqual(expected, my_task.rebind)128    def test_rebind_all_args(self):129        my_task = MyTask(rebind={'spam': 'a', 'eggs': 'b', 'context': 'c'})130        expected = {131            'spam': 'a',132            'eggs': 'b',133            'context': 'c'134        }135        self.assertEqual(expected, my_task.rebind)136        self.assertEqual(set(['a', 'b', 'c']),137                         my_task.requires)138    def test_rebind_partial(self):139        my_task = MyTask(rebind={'spam': 'a', 'eggs': 'b'})140        expected = {141            'spam': 'a',142            'eggs': 'b',143            'context': 'context'144        }145        self.assertEqual(expected, my_task.rebind)146        self.assertEqual(set(['a', 'b', 'context']),147                         my_task.requires)148    def test_rebind_unknown(self):149        self.assertRaisesRegex(ValueError, '^Extra arguments',150                               MyTask, rebind={'foo': 'bar'})151    def test_rebind_unknown_kwargs(self):152        my_task = KwargsTask(rebind={'foo': 'bar'})153        expected = {154            'foo': 'bar',155            'spam': 'spam'156        }157        self.assertEqual(expected, my_task.rebind)158    def test_rebind_list_all(self):159        my_task = MyTask(rebind=('a', 'b', 'c'))160        expected = {161            'context': 'a',162            'spam': 'b',163            'eggs': 'c'164        }165        self.assertEqual(expected, my_task.rebind)166        self.assertEqual(set(['a', 'b', 'c']),167                         my_task.requires)168    def test_rebind_list_partial(self):169        my_task = MyTask(rebind=('a', 'b'))170        expected = {171            'context': 'a',172            'spam': 'b',173            'eggs': 'eggs'174        }175        self.assertEqual(expected, my_task.rebind)176        self.assertEqual(set(['a', 'b', 'eggs']),177                         my_task.requires)178    def test_rebind_list_more(self):179        self.assertRaisesRegex(ValueError, '^Extra arguments',180                               MyTask, rebind=('a', 'b', 'c', 'd'))181    def test_rebind_list_more_kwargs(self):182        my_task = KwargsTask(rebind=('a', 'b', 'c'))183        expected = {184            'spam': 'a',185            'b': 'b',186            'c': 'c'187        }188        self.assertEqual(expected, my_task.rebind)189        self.assertEqual(set(['a', 'b', 'c']),190                         my_task.requires)191    def test_rebind_list_bad_value(self):192        self.assertRaisesRegex(TypeError, '^Invalid rebind value',193                               MyTask, rebind=object())194    def test_default_provides(self):195        my_task = DefaultProvidesTask()196        self.assertEqual(set(['def']), my_task.provides)197        self.assertEqual({'def': None}, my_task.save_as)198    def test_default_provides_can_be_overridden(self):199        my_task = DefaultProvidesTask(provides=('spam', 'eggs'))200        self.assertEqual(set(['spam', 'eggs']), my_task.provides)201        self.assertEqual({'spam': 0, 'eggs': 1}, my_task.save_as)202    def test_update_progress_within_bounds(self):203        values = [0.0, 0.5, 1.0]204        result = []205        def progress_callback(event_type, details):206            result.append(details.pop('progress'))207        a_task = ProgressTask()208        a_task.notifier.register(task.EVENT_UPDATE_PROGRESS, progress_callback)209        a_task.execute(values)210        self.assertEqual(values, result)211    @mock.patch.object(task.LOG, 'warning')212    def test_update_progress_lower_bound(self, mocked_warning):213        result = []214        def progress_callback(event_type, details):215            result.append(details.pop('progress'))216        a_task = ProgressTask()217        a_task.notifier.register(task.EVENT_UPDATE_PROGRESS, progress_callback)218        a_task.execute([-1.0, -0.5, 0.0])219        self.assertEqual([0.0, 0.0, 0.0], result)220        self.assertEqual(2, mocked_warning.call_count)221    @mock.patch.object(task.LOG, 'warning')222    def test_update_progress_upper_bound(self, mocked_warning):223        result = []224        def progress_callback(event_type, details):225            result.append(details.pop('progress'))226        a_task = ProgressTask()227        a_task.notifier.register(task.EVENT_UPDATE_PROGRESS, progress_callback)228        a_task.execute([1.0, 1.5, 2.0])229        self.assertEqual([1.0, 1.0, 1.0], result)230        self.assertEqual(2, mocked_warning.call_count)231    @mock.patch.object(notifier.LOG, 'warning')232    def test_update_progress_handler_failure(self, mocked_warning):233        def progress_callback(*args, **kwargs):234            raise Exception('Woot!')235        a_task = ProgressTask()236        a_task.notifier.register(task.EVENT_UPDATE_PROGRESS, progress_callback)237        a_task.execute([0.5])238        self.assertEqual(1, mocked_warning.call_count)239    def test_register_handler_is_none(self):240        a_task = MyTask()241        self.assertRaises(ValueError, a_task.notifier.register,242                          task.EVENT_UPDATE_PROGRESS, None)243        self.assertEqual(0, len(a_task.notifier))244    def test_deregister_any_handler(self):245        a_task = MyTask()246        self.assertEqual(0, len(a_task.notifier))247        a_task.notifier.register(task.EVENT_UPDATE_PROGRESS,248                                 lambda event_type, details: None)249        self.assertEqual(1, len(a_task.notifier))250        a_task.notifier.deregister_event(task.EVENT_UPDATE_PROGRESS)251        self.assertEqual(0, len(a_task.notifier))252    def test_deregister_any_handler_empty_listeners(self):253        a_task = MyTask()254        self.assertEqual(0, len(a_task.notifier))255        self.assertFalse(a_task.notifier.deregister_event(256            task.EVENT_UPDATE_PROGRESS))257        self.assertEqual(0, len(a_task.notifier))258    def test_deregister_non_existent_listener(self):259        handler1 = lambda event_type, details: None260        handler2 = lambda event_type, details: None261        a_task = MyTask()262        a_task.notifier.register(task.EVENT_UPDATE_PROGRESS, handler1)263        self.assertEqual(1, len(list(a_task.notifier.listeners_iter())))264        a_task.notifier.deregister(task.EVENT_UPDATE_PROGRESS, handler2)265        self.assertEqual(1, len(list(a_task.notifier.listeners_iter())))266        a_task.notifier.deregister(task.EVENT_UPDATE_PROGRESS, handler1)267        self.assertEqual(0, len(list(a_task.notifier.listeners_iter())))268    def test_bind_not_callable(self):269        a_task = MyTask()270        self.assertRaises(ValueError, a_task.notifier.register,271                          task.EVENT_UPDATE_PROGRESS, 2)272    def test_copy_no_listeners(self):273        handler1 = lambda event_type, details: None274        a_task = MyTask()275        a_task.notifier.register(task.EVENT_UPDATE_PROGRESS, handler1)276        b_task = a_task.copy(retain_listeners=False)277        self.assertEqual(1, len(a_task.notifier))278        self.assertEqual(0, len(b_task.notifier))279    def test_copy_listeners(self):280        handler1 = lambda event_type, details: None281        handler2 = lambda event_type, details: None282        a_task = MyTask()283        a_task.notifier.register(task.EVENT_UPDATE_PROGRESS, handler1)284        b_task = a_task.copy()285        self.assertEqual(1, len(b_task.notifier))286        self.assertTrue(a_task.notifier.deregister_event(287            task.EVENT_UPDATE_PROGRESS))288        self.assertEqual(0, len(a_task.notifier))289        self.assertEqual(1, len(b_task.notifier))290        b_task.notifier.register(task.EVENT_UPDATE_PROGRESS, handler2)291        listeners = dict(list(b_task.notifier.listeners_iter()))292        self.assertEqual(2, len(listeners[task.EVENT_UPDATE_PROGRESS]))293        self.assertEqual(0, len(a_task.notifier))294    def test_separate_revert_args(self):295        my_task = SeparateRevertTask(rebind=('a',), revert_rebind=('b',))296        self.assertEqual({'execute_arg': 'a'}, my_task.rebind)297        self.assertEqual({'revert_arg': 'b'}, my_task.revert_rebind)298        self.assertEqual(set(['a', 'b']),299                         my_task.requires)300        my_task = SeparateRevertTask(requires='execute_arg',301                                     revert_requires='revert_arg')302        self.assertEqual({'execute_arg': 'execute_arg'}, my_task.rebind)303        self.assertEqual({'revert_arg': 'revert_arg'}, my_task.revert_rebind)304        self.assertEqual(set(['execute_arg', 'revert_arg']),305                         my_task.requires)306    def test_separate_revert_optional_args(self):307        my_task = SeparateRevertOptionalTask()308        self.assertEqual(set(['execute_arg']), my_task.optional)309        self.assertEqual(set(['revert_arg']), my_task.revert_optional)310    def test_revert_kwargs(self):311        my_task = RevertKwargsTask()312        expected_rebind = {'execute_arg1': 'execute_arg1',313                           'execute_arg2': 'execute_arg2'}314        self.assertEqual(expected_rebind, my_task.rebind)315        expected_rebind = {'execute_arg1': 'execute_arg1'}316        self.assertEqual(expected_rebind, my_task.revert_rebind)317        self.assertEqual(set(['execute_arg1', 'execute_arg2']),318                         my_task.requires)319class FunctorTaskTest(test.TestCase):320    def test_creation_with_version(self):321        version = (2, 0)322        f_task = task.FunctorTask(lambda: None, version=version)323        self.assertEqual(version, f_task.version)324    def test_execute_not_callable(self):325        self.assertRaises(ValueError, task.FunctorTask, 2)326    def test_revert_not_callable(self):327        self.assertRaises(ValueError, task.FunctorTask, lambda: None,328                          revert=2)329class ReduceFunctorTaskTest(test.TestCase):330    def test_invalid_functor(self):331        # Functor not callable332        self.assertRaises(ValueError, task.ReduceFunctorTask, 2, requires=5)333        # Functor takes no arguments334        self.assertRaises(ValueError, task.ReduceFunctorTask, lambda: None,335                          requires=5)336        # Functor takes too few arguments337        self.assertRaises(ValueError, task.ReduceFunctorTask, lambda x: None,338                          requires=5)339    def test_functor_invalid_requires(self):340        # Invalid type, requires is not iterable341        self.assertRaises(TypeError, task.ReduceFunctorTask,342                          lambda x, y: None, requires=1)343        # Too few elements in requires344        self.assertRaises(ValueError, task.ReduceFunctorTask,345                          lambda x, y: None, requires=[1])346class MapFunctorTaskTest(test.TestCase):347    def test_invalid_functor(self):348        # Functor not callable349        self.assertRaises(ValueError, task.MapFunctorTask, 2, requires=5)350        # Functor takes no arguments351        self.assertRaises(ValueError, task.MapFunctorTask, lambda: None,352                          requires=5)353        # Functor takes too many arguments354        self.assertRaises(ValueError, task.MapFunctorTask, lambda x, y: None,355                          requires=5)356    def test_functor_invalid_requires(self):357        # Invalid type, requires is not iterable358        self.assertRaises(TypeError, task.MapFunctorTask, lambda x: None,...
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.