Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

How to use execute_with_retries method in molecule

Best Python code snippet using molecule_python

images.py

Source:images.py

...17from sahara.utils.openstack import images as sahara_images18conductor = c.API19# Image Registry20def get_images(name, tags):21    return b.execute_with_retries(22        sahara_images.image_manager().list_registered, name, tags)23def get_image(**kwargs):24    if len(kwargs) == 1 and 'id' in kwargs:25        return b.execute_with_retries(26            sahara_images.image_manager().get, kwargs['id'])27    else:28        return b.execute_with_retries(29            sahara_images.image_manager().find, **kwargs)30def get_registered_image(id):31    return b.execute_with_retries(32        sahara_images.image_manager().get_registered_image, id)33def register_image(image_id, username, description=None):34    manager = sahara_images.image_manager()35    b.execute_with_retries(36        manager.set_image_info, image_id, username, description)37    return b.execute_with_retries(manager.get, image_id)38def unregister_image(image_id):39    manager = sahara_images.image_manager()40    b.execute_with_retries(manager.unset_image_info, image_id)41    return b.execute_with_retries(manager.get, image_id)42def get_image_tags(image_id):43    return b.execute_with_retries(44        sahara_images.image_manager().get, image_id).tags45def set_image_tags(image_id, tags):46    manager = sahara_images.image_manager()47    image_obj = b.execute_with_retries(manager.get, image_id)48    org_tags = frozenset(image_obj.tags)49    new_tags = frozenset(tags)50    to_add = list(new_tags - org_tags)51    to_remove = list(org_tags - new_tags)52    if to_add:53        b.execute_with_retries(manager.tag, image_id, to_add)54    if to_remove:55        b.execute_with_retries(manager.untag, image_id, to_remove)56    return b.execute_with_retries(manager.get, image_id)57def remove_image_tags(image_id):58    manager = sahara_images.image_manager()59    image_obj = b.execute_with_retries(manager.get, image_id)60    tags = image_obj.tags61    b.execute_with_retries(manager.untag, image_id, tags)...

base.py

Source:base.py

...15        try:16            sleep(1)17            logger.info("Running job '%s'", self.name)18            start = time()19            retries_left, result = self.execute_with_retries()20            end = time()21            self.notify_success(22                result=result,23                runtime=round(end - start),24                retries_left=retries_left,25            )26        except Exception as exc:27            logger.error("Error running job '%s': %s", self.name, str(exc))28            self.notify_failure(exc)29    def execute(self):30        raise NotImplementedError()31    def execute_with_retries(self, retries: int = MAX_RETRIES):32        try:33            return retries, self.execute()34        except Exception as exc:35            if retries:36                return self.execute_with_retries(retries - 1)37            raise exc38    def notify_success(self, result: str, runtime: int, retries_left: int):39        if not settings.SLACK_WEBHOOK_URL:40            return41        message = Message(42            text=f"{self.name} > finished successfully.",43            blocks=[44                Section(45                    text=MarkdownText(46                        text=f"*{self.name.upper()}* > finished successfully after {runtime} seconds. \n\n```{result}```",47                    ),48                ),49            ],50        )...

actionqueue.py

Source:actionqueue.py

...23        """24        self._state_machine.transition_to_execute()25        for action in self._actions:26            self._executed_actions.append(action)27            self.execute_with_retries(action, lambda a: a.execute())28        self._state_machine.transition_to_execute_complete()29    def rollback(self):30        """Call rollback on executed actions."""31        self._state_machine.transition_to_rollback()32        for action in reversed(self._executed_actions):33            try:34                self.execute_with_retries(action, lambda a: a.rollback())35            except:  # pylint: disable=bare-except36                pass  # on exception, carry on with rollback of other steps37        self._state_machine.transition_to_rollback_complete()38    def execute_with_retries(self, action, f):39        """Execute function f with single argument action. Retry if40        ActionRetryException is raised.41        """42        # Run action until either it succeeds or throws an exception43        # that's not an ActionRetryException44        retry = True45        while retry:46            retry = False47            try:48                f(action)49            except ActionRetryException as ex:  # other exceptions should bubble out50                retry = True...

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.