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

How to use on_success method in locust

Best Python code snippet using locust

dialogue.py

Source:dialogue.py

1import sys2import os3from actions import Action, get_answer, get_bool_answer4import messages5from validators import get_list_validators6class StackIsEmpty(Exception):7    pass8class Command:9    def execute(self, stack):10        pass11    def undo(self):12        pass13    @staticmethod14    def go_back(stack):15        while len(stack) > 1:16            stack.pop()17            if stack[-1].undo():18                return stack.pop()19        raise StackIsEmpty()20class FinalCommand(Command):21    def __init__(self, field):22        self.field = field23    def execute(self, stack):24        print(messages.DONE)25        try:26            yaml = self.field.get_yaml()27        except Exception as e:28            print(messages.INTERNAL_ERROR, e)29            sys.exit(1)30        while True:31            print(messages.CONFIG_FILE_NAME)32            action, answer = get_answer(self.field)33            if action == Action.UNDO:34                return self.go_back(stack)35            elif action != Action.CONTINUE:36                print(messages.INCORRECT_ACTION)37                continue38            path = os.path.abspath(answer)39            if not os.path.exists(os.path.dirname(path)):40                try:41                    os.makedirs(os.path.dirname(path))42                except Exception as e:43                    print(messages.CANT_CREATE_DIRECTORY, e)44                    continue45            try:46                with open(path, 'w') as f:47                    f.write(yaml)48            except Exception as e:49                print(messages.CANT_SAVE_CONFIG, e)50                continue51            sys.exit(0)52    def undo(self):53        pass54class Continue(Command):55    def __init__(self, current_command, resume_command):56        self.current_command = current_command57        self.resume_command = resume_command58    def execute(self, stack):59        next_command = self.current_command.execute(stack)60        if next_command is not None:61            return Continue(next_command, self.resume_command)62        return self.resume_command63    def undo(self):64        return self.current_command.undo()65class UpdateYaml(Command):66    def __init__(self, field, on_success):67        self.field = field68        self.on_success = on_success69        self.old_yaml = None70    def execute(self, stack):71        self.old_yaml = self.field.update_yaml()72        print(messages.SUCCESS + self.field.get_name())73        return self.on_success74    def undo(self):75        self.field.set_yaml(self.old_yaml)76        return False77class AddFieldAndContinue(Command):78    def __init__(self, field, sub_field, resume_command):79        self.field = field80        self.sub_field = sub_field81        self.resume_command = resume_command82    def execute(self, stack):83        self.field.add_field(self.sub_field)84        return self.resume_command85    def undo(self):86        self.field.pop_field()87        return False88class ManualFillSubfieldGenerators(Command):89    def __init__(self, field, on_success, idx=0):90        self.field = field91        self.on_success = on_success92        self.idx = idx93    def execute(self, stack):94        if self.idx + 1 < len(self.field.get_field_generators()):95            on_fail = ManualFillSubfieldGenerators(self.field, self.on_success, self.idx + 1)96        else:97            on_fail = UpdateYaml(self.field, self.on_success)98        sub_field = self.field.get_field_generators()[self.idx].copy()99        sub_field.set_parent(self.field)100        on_success = ManualFillSubfieldGenerators(self.field, self.on_success, self.idx)101        on_success = AddFieldAndContinue(self.field, sub_field, on_success)102        return SkipNotMandatoryField(sub_field, on_success, on_fail)103    def undo(self):104        return False105class ManualFillSubfields(Command):106    def __init__(self, field, on_success, idx=0):107        self.field = field108        self.on_success = on_success109        self.idx = idx110    def execute(self, stack):111        if self.idx == 0:112            print(messages.HAS_SUBFIELDS)113        if self.idx + 1 < len(self.field.get_fields()):114            resume_command = ManualFillSubfields(self.field, self.on_success, self.idx + 1)115        elif len(self.field.get_field_generators()) > 0:116            resume_command = ManualFillSubfieldGenerators(self.field, self.on_success)117        else:118            resume_command = UpdateYaml(self.field, self.on_success)119        return Continue(SkipNotMandatoryField(self.field.get_fields()[self.idx], None, None), resume_command)120    def undo(self):121        return False122class SetLeafValue(Command):123    def __init__(self, field, value, on_success):124        self.field = field125        self.value = value126        self.on_success = on_success127        self.old_yaml = None128    def execute(self, stack):129        typed_value = self.field.validate(self.value)130        if typed_value is None:131            raise ValueError('Validation failed')132        self.old_yaml = self.field.set_leaf_value(typed_value)133        print(messages.SUCCESS + self.field.get_name())134        return self.on_success135    def undo(self):136        self.field.set_yaml(self.old_yaml)137        return False138class ReadArray(Command):139    def __init__(self, field, validators, array, on_success):140        self.field = field141        self.validators = validators142        self.array = array143        self.got_value = False144        self.on_success = on_success145    def execute(self, stack):146        print(messages.FILLING_ARRAY, self.array)147        self.got_value = False148        while True:149            action, answer = get_answer(self.field)150            if action == Action.UNDO:151                return self.go_back(stack)152            elif action == Action.READ_ARRAY:153                validators = []154                for validator in self.validators:155                    validators += validator.get_list_validators()156                if len(validators) == 0:157                    print(messages.INCORRECT_VALUE)158                    continue159                self.got_value = True160                self.array.append([])161                return ReadArray(self.field, validators, self.array[-1],162                                 ReadArray(self.field, self.validators, self.array, self.on_success))163            elif action == Action.END_READ_ARRAY:164                success = False165                for validator in self.validators:166                    if validator.validate(self.array) is not None:167                        success = True168                if not success:169                    print(messages.INCORRECT_VALUE)170                    continue171                return self.on_success172            elif action != Action.CONTINUE:173                print(messages.INCORRECT_ACTION)174                continue175            next_validators = []176            for validator in self.validators:177                if validator.validate([answer]) is not None:178                    next_validators.append(validator)179            if len(next_validators) == 0:180                print(messages.INCORRECT_VALUE)181                continue182            self.got_value = True183            self.array.append(answer)184            return ReadArray(self.field, next_validators, self.array, self.on_success)185    def undo(self):186        if self.got_value:187            self.array.pop()188        return True189class ManualFillLeaf(Command):190    def __init__(self, field, on_success):191        self.field = field192        self.on_success = on_success193    def execute(self, stack):194        if self.field.has_values():195            while True:196                print(messages.AVAILABLE_VALUES)197                for idx, value in enumerate(self.field.get_values()):198                    print(idx, value, sep=': ')199                try:200                    action, value_idx = get_answer(self.field)201                    if action == Action.UNDO:202                        return self.go_back(stack)203                    elif action != Action.CONTINUE:204                        print(messages.INCORRECT_ACTION)205                        continue206                    value = self.field.get_values()[int(value_idx)]207                    self.old_yaml = self.field.set_leaf_value(value)208                    print(messages.SUCCESS + self.field.get_name())209                    return self.on_success210                except:211                    print(messages.INCORRECT_VALUE_IDX)212        while True:213            print(messages.ALLOWED_VALUES, end=' ')214            print(self.field.get_type_description())215            print(messages.ENTER_VALUE)216            action, answer = get_answer(self.field)217            if action == Action.UNDO:218                return self.go_back(stack)219            elif action == Action.READ_ARRAY:220                validators = self.field.get_list_validators()221                if len(validators) == 0:222                    print(messages.INCORRECT_VALUE)223                    continue224                array = []225                return ReadArray(self.field, validators, array, SetLeafValue(self.field, array, self.on_success))226            elif action != Action.CONTINUE:227                print(messages.INCORRECT_ACTION)228                continue229            if self.field.validate(answer) is None:230                print(messages.INCORRECT_VALUE)231                continue232            return SetLeafValue(self.field, answer, self.on_success)233    def undo(self):234        return True235class UseDefaultValue(Command):236    def __init__(self, field, on_success):237        self.field = field238        self.on_success = on_success239        self.had_answer = False240        self.updated_yaml_key = False241    def execute(self, stack):242        self.had_answer = False243        self.updated_yaml_key = False244        if self.field.has_default_value():245            self.had_answer = True246            print(messages.DEFAULT_VALUE_AVAILABLE, end='')247            print(self.field.get_default_value())248            print(messages.USE_DEFAULT_VALUE)249            while True:250                action, answer = get_bool_answer(self.field)251                if action == Action.UNDO:252                    return self.go_back(stack)253                elif action != Action.CONTINUE:254                    print(messages.INCORRECT_ACTION)255                    continue256                break257            if answer:258                self.updated_yaml_key = True259                self.field.update_yaml_key()260                return self.on_success261        if self.field.is_leaf():262            return ManualFillLeaf(self.field, self.on_success)263        elif len(self.field.get_fields()) > 0:264            return ManualFillSubfields(self.field, self.on_success)265        return ManualFillSubfieldGenerators(self.field, self.on_success)266    def undo(self):267        if self.updated_yaml_key:268            self.field.undo_update_yaml_key()269        return self.had_answer270class FillKey(Command):271    def __init__(self, field, on_success):272        self.field = field273        self.on_success = on_success274        self.had_answer = False275    def execute(self, stack):276        self.had_answer = False277        if not self.field.has_key() and not self.field.is_root():278            self.had_answer = True279            print(messages.KEY_REQUIRED)280            while True:281                action, answer = get_answer(self.field)282                if action == Action.UNDO:283                    return self.go_back(stack)284                elif action != Action.CONTINUE:285                    print(messages.INCORRECT_ACTION)286                    continue287                if not self.field.set_key(answer):288                    print(messages.KEY_ALREADY_USED)289                    continue290                break291        return UseDefaultValue(self.field, self.on_success)292    def undo(self):293        if self.had_answer:294            self.field.properties.pop('key')295        return self.had_answer296class SkipNotMandatoryField(Command):297    def __init__(self, field, on_success, on_fail):298        self.field = field299        self.on_success = on_success300        self.on_fail = on_fail301        self.old_yaml = None302        self.updated_yaml = False303        self.had_answer = False304    def execute(self, stack):305        self.old_yaml = None306        self.updated_yaml = False307        self.had_answer = False308        print(messages.SETTING_UP_FIELD + self.field.get_name() + self.field.get_optional_key())309        if self.field.has_hint():310            print(self.field.get_hint())311        if not self.field.is_mandatory():312            self.had_answer = True313            print(messages.FILL_NOT_MANDATORY_FIELD)314            while True:315                action, answer = get_bool_answer(self.field)316                if action == Action.UNDO:317                    return self.go_back(stack)318                elif action != Action.CONTINUE:319                    print(messages.INCORRECT_ACTION)320                    continue321                break322            if not answer:323                self.updated_yaml = True324                self.old_yaml = self.field.reset_yaml()325                return self.on_fail326        return FillKey(self.field, self.on_success)327    def undo(self):328        if self.updated_yaml:329            self.field.set_yaml(self.old_yaml)330        return self.had_answer331def run_dialogue(root):332    print(messages.WELCOME)333    print(messages.SEPARATOR)334    try:335        root.initialize()336    except Exception as e:337        print(messages.INTERNAL_ERROR, e)338        sys.exit(1)339    while True:340        try:341            final_command = FinalCommand(root)342            current_command = SkipNotMandatoryField(root, final_command, final_command)343            stack = [current_command]344            while True:345                current_command = current_command.execute(stack)346                stack.append(current_command)347        except StackIsEmpty:348            continue349        except Exception as e:350            print(messages.INTERNAL_ERROR, e)...

aplus.py

Source:aplus.py

...96        return97    try:98        result = future.result()99        if on_success:100            result = on_success(result)101        if isinstance(result, Future):102            _chain_future(new_future, result)103        else:104            new_future.set_result(result)105    except BaseException:106        exception = _sys_exception()107        if not on_failure:108            new_future.set_exception(exception)109        else:110            try:111                result = on_failure(exception)112                if isinstance(result, BaseException):113                    new_future.set_exception(result)114                else:...

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.