How to use delete_backup method in tempest

Best Python code snippet using tempest_python

application.py

Source:application.py

1from any_project import __module__2from any_project import Setup3from any_project.template import yaml_template4from any_project.constant import Constant5from any_project.internal import BuildActionData, InternalActions6from collections import OrderedDict7import oyaml as yaml8from pymsgprompt.logger import pinfo, perror, pwarn9from pymsgprompt.prompt import ask10import os, re11from any_project.setup import ValidationError12from any_project.setup import ValidationResult131415class Actions(object):16    @staticmethod17    def init(target_dir, project_name, template_generator=None):18        YAML_CODE_DICT = yaml_template(project_name,19            os.path.relpath(target_dir), template_generator)20        try:21            os.makedirs(target_dir, exist_ok=True)22        except FileNotFoundError as e:23            perror(f"{type(e).__name__} -> {e}")24            return None25        project_structure_yaml = os.path.join(target_dir, 'project-structure.yaml')26        overwrite_option = None27        if os.path.isfile(project_structure_yaml):28            overwrite_option = ask('Do you want to overwrite the existing project structure?', \29                choices=['yes', 'no'], default='no', on_error=lambda *argv: True)30        else:31            overwrite_option = 'yes'32        if overwrite_option == 'yes':33            with open(project_structure_yaml, 'w') as f:34                yaml.dump(YAML_CODE_DICT, f, indent=4)35        else:36            pwarn('Action has been prevented by user!')37            return None38        return project_structure_yaml39        40    @staticmethod41    @InternalActions.post_build_activity42    def build(structure, action_name, tasks, delete_backup=None):43        # Complete the precheck before building the structure project44        precheck_result = InternalActions.any_project_precheck(structure)45        if precheck_result is None:46            return BuildActionData(status=False)47        48        # Initialize the local variables49        yaml_data, working_dir, project_name = precheck_result50        backup_zip, root = None, None51        52        # Load the environment variables53        try:54            envionment = yaml_data.get('environment')55            if envionment is not None:56                if not isinstance(envionment, (dict, OrderedDict)):57                    raise TypeError(f'Invalid data of "environment" inside "{structure}"')58                pinfo('Adding environment variables')59                os.environ.update({60                    env_key.strip() : env_val \61                        for env_key, env_val in envionment.items()62                })63        except TypeError as e:64            perror(f'{e}')65            return BuildActionData(status=False)6667        # Load the any-project constants68        try:69            temp_constants = yaml_data.get('constants')70            if temp_constants is None:71                constants = type('Constants', (object, ), {})72            else:73                if not isinstance(temp_constants, (dict, OrderedDict)):74                    raise TypeError(f'Invalid data of constants inside "{structure}"')75                def invalid_constant(key):76                    raise KeyError(f'Invalid any-project constant {key}')77                def valid_constant(key, val):78                    pinfo(f'Adding any-project constant "{key}={val}"')79                    return Constant(val)80                k_pat = r'^[a-z_][a-z0-9_]*$'81                constants = type('Constants', (object, ), {82                    key.strip() : valid_constant(key.strip(), val) \83                        if re.match(k_pat, key.strip(), flags=re.I) \84                            else invalid_constant(key.strip()) \85                                for key, val in temp_constants.items()              86                })87        except (TypeError, KeyError) as e:88            perror(f'{e}')89            return BuildActionData(status=False)90        91        # Get the project root path92        root = os.path.relpath(os.path.join(working_dir, project_name))93        94        # Take the safety backup95        backup_zip, error_occured = InternalActions.take_safe_backup(96            working_dir, project_name97        )98        if error_occured:99            if backup_zip is not None:100                pinfo('Deleting corrupted safety backup')101                os.unlink(backup_zip)102            return BuildActionData(103                status=False,104                root=root,105                cwd=working_dir,106                backup_zip=None,107                delete_backup=False108            )109        110        # Get the action's architecture111        action = None112        setup_obj = None113        try:114            if action_name in yaml_data['boilerplates'].keys():115                action = yaml_data['boilerplates'][action_name]116            else:117                raise KeyError(f'Could not find the action "{action_name}"')118        except KeyError as e:119            perror(f'{type(e).__name__} occurred -> {e}')120            return BuildActionData(121                status=False,122                root=root,123                cwd=working_dir,124                backup_zip=backup_zip,125                delete_backup=delete_backup126            )127        128        # Get the setup code from yaml129        setup_code = action.get('setup')130        if setup_code is not None:131            # Validation of setup code132            if not isinstance(setup_code, str):133                perror(f'Invalid setup code in action "{action_name}"')134                pinfo('Setup code should be a valid python code')135                return BuildActionData(136                    status=False,137                    root=root,138                    cwd=working_dir,139                    backup_zip=backup_zip,140                    delete_backup=delete_backup141                )142143            # Get the source code, class name and it's location 144            # in terms of line number in the source code145            try:146                actual_source, setup_class_name, class_def_line_no = \147                    InternalActions.get_setup_source_code(148                        setup_code, action_name149                    )150            except (SyntaxError, ImportError) as e:151                perror(f'{e}')152                return BuildActionData(153                    status=False,154                    root=root,155                    cwd=working_dir,156                    backup_zip=backup_zip,157                    delete_backup=delete_backup158                )159160            # Execute the setup source code and create object161            # of the setup class of the action162            try:163                exec(actual_source, globals())164                SetupClass = eval(setup_class_name)165                if not issubclass(SetupClass, Setup):166                    raise SyntaxError(167                        f'Setup class "{SetupClass.__name__}" ' + \168                            f'at line {class_def_line_no} '+ \169                                'must inherit from "any_project.Setup" class'170                    )171172                # Creating object of the setup class173                setup_obj = SetupClass(action_name)174175                # Completing the pre-validations of the setup176                try:177                    result = setup_obj.pre_validation()178                except Exception as e:179                    raise RuntimeError(180                        f'<{type(e).__name__}> {e}, while executing' + \181                            'pre_validation()')182183                if not isinstance(result, ValidationResult):184                    result = ValidationResult(185                        False, 'Illegal type of pre-validation result.'186                    )187                if not result.successful:188                    raise ValidationError(result.message)189                190                # Asking the prompt value from the user191                try:192                    setup_obj.set_prompts()193                except Exception as e:194                    raise RuntimeError(195                        f'<{type(e).__name__}> {e}, while executing ' + \196                            'set_prompts()')197198                # Executing all the tasks199                for task in tasks:200                    try:201                        setup_obj.on_task(task=task)202                    except Exception as e:203                        raise RuntimeError(204                            f'<{type(e).__name__}> {e}, while executing' + \205                                f'on_task(task="{task}")')206            except (SyntaxError, ValidationError, RuntimeError) as e:207                # Handling RuntimeError is important to prevent any208                # mess up of the code, after the setup functions209                # get any exception210                perror(f'{type(e).__name__} occurred -> {e}')211                return BuildActionData(212                    status=False,213                    root=root,214                    cwd=working_dir,215                    backup_zip=backup_zip,216                    delete_backup=delete_backup217                )218219        # Get the structure from yaml220        structure_ = action.get('structure')221        if structure_ is not None:222            if not isinstance(structure_, (dict, OrderedDict)):223                perror(f'Invalid file structure in action "{action_name}"')224                return BuildActionData(225                    status=False,226                    root=root,227                    cwd=working_dir,228                    backup_zip=backup_zip,229                    delete_backup=delete_backup230                )231            232            # Creating the ROOT directory if does not exist233            if not os.path.isdir(root):234                pinfo(f'Creating ROOT directory: "{root}"')235            os.makedirs(root, exist_ok=True)236237            # Expand the file structure238            if not InternalActions.expand_file_structure( \239                root, structure_, setup_obj, constants):240                perror('Could not expand the file structure!')241                return BuildActionData(242                    status=False,243                    root=root,244                    cwd=working_dir,245                    backup_zip=backup_zip,246                    delete_backup=delete_backup247                )248            249            # Git repo functionality250            try:251                is_git_repo = constants.git_repo252                if not isinstance(is_git_repo, bool):253                    is_git_repo = False254            except AttributeError:255                is_git_repo = False256257            # Get the git commit message258            git_commit = action.get('git-commit')259        260            if not isinstance(git_commit, str):261                # Handle invalid value of git-commit262                if git_commit is None and \263                    action_name.strip() == 'default':264                    git_commit = 'Commit for default action'265                else:266                    if is_git_repo:267                        pwarn(268                            f'Value of "git-commit" under ' + \269                                f'"{action_name}" action should be a string'270                        )271                    git_commit = None272            if is_git_repo and git_commit is not None:273                git_commit = os.path.expandvars(git_commit).format(274                    prompts = None if setup_obj \275                        is None else setup_obj.prompts,276                    consts = constants277                )278279                # Create a git repo if does not exist280                # and add the files with a commit message281                exc = InternalActions.add_git_commit(282                    root, git_commit283                )284                if exc is not None:285                    perror(f"{type(exc).__name__} occurred -> {exc}")286                    pwarn('Commit may not be occurred successfully!')287                    # Even if git commit is failed, it should not be 288                    # rolled back. If rolling back is necessary, it289                    # can be done using post-validation290291        # Do the post build validations292        if setup_obj is not None:293            try:294                try:295                    result = setup_obj.post_validation()296                except Exception as e:297                    raise RuntimeError(298                        f'<{type(e).__name__}> {e}, while executing ' + \299                            'post_validtion()')300                if not isinstance(result, ValidationResult):301                    result = ValidationResult(302                        False, 'Illegal type of post-validation result.'303                    )304                if not result.successful:305                    raise ValidationError(result.message)306            except (ValidationError, RuntimeError) as e:307                perror(f'{type(e).__name__} occurred -> {e}')308                return BuildActionData(309                    status=False,310                    root=root,311                    cwd=working_dir,312                    backup_zip=backup_zip,313                    delete_backup=delete_backup314                )315316        return BuildActionData(317            status=True,318            root=root,319            cwd=working_dir,320            backup_zip=backup_zip,321            delete_backup=delete_backup 
...

glance_delete_image_request_body.py

Source:glance_delete_image_request_body.py

...29        self.discriminator = None30        if delete_backup is not None:31            self.delete_backup = delete_backup32    @property33    def delete_backup(self):34        """Gets the delete_backup of this GlanceDeleteImageRequestBody.35        åå¼ä¸ºï¼trueåfalse trueï¼è¡¨ç¤ºå é¤æ´æºéåæ¶ï¼åæ¶å é¤å¶å³èçäºæå¡å¨å¤ä»½ã falseï¼è¡¨ç¤ºåªå é¤æ´æºéåï¼ä¸å é¤å¶å³èçäºæå¡å¨å¤ä»½ã36        :return: The delete_backup of this GlanceDeleteImageRequestBody.37        :rtype: bool38        """39        return self._delete_backup40    @delete_backup.setter41    def delete_backup(self, delete_backup):42        """Sets the delete_backup of this GlanceDeleteImageRequestBody.43        åå¼ä¸ºï¼trueåfalse trueï¼è¡¨ç¤ºå é¤æ´æºéåæ¶ï¼åæ¶å é¤å¶å³èçäºæå¡å¨å¤ä»½ã falseï¼è¡¨ç¤ºåªå é¤æ´æºéåï¼ä¸å é¤å¶å³èçäºæå¡å¨å¤ä»½ã44        :param delete_backup: The delete_backup of this GlanceDeleteImageRequestBody.45        :type delete_backup: bool46        """47        self._delete_backup = delete_backup48    def to_dict(self):49        """Returns the model properties as a dict"""50        result = {}51        for attr, _ in six.iteritems(self.openapi_types):52            value = getattr(self, attr)53            if isinstance(value, list):54                result[attr] = list(map(55                    lambda x: x.to_dict() if hasattr(x, "to_dict") else x,...

backup-server.py

Source:backup-server.py

...40            self.wfile.write("ERR: btrfs receive returncode {}\n".format(process.returncode).encode("UTF-8"))41        # clean up if we had error and ended up with leftover backup-new42        if os.path.exists(directory + "/BACKUP-new"):43            subprocess.call(("/bin/btrfs", "subvolume", "delete", directory + "/BACKUP-new"))44    def delete_backup(self, date):45        name = directory + date.strftime(".%Y-%m-%dT%H")46        if os.path.exists(name):47            subprocess.call(("/bin/btrfs", "subvolume", "delete", name))48    def archive(self):49        today = datetime.datetime.today();50        subprocess.call(("/bin/btrfs", "subvolume", "snapshot", "-r", directory + "/BACKUP", directory + today.strftime(".%Y-%m-%dT%H")))51        # delete all but midnight's backups after 1 day52        date = today - datetime.timedelta(1)53        if date.hour != 0:54            self.delete_backup(date)55        # delete all but first week's backups after 1 month56        date = today.date() - datetime.timedelta(7*4)57        if date.weekday() != 0 or date.day > 7:58            self.delete_backup(date)59        # unconditionally delete after 365 days60        date = today.date() - datetime.timedelta(365)61        self.delete_backup(date)62if __name__ == "__main__":63    host, port, directory = sys.argv[1:]64    port = int(port)65    server = socketserver.TCPServer((host, port), BackupServer)66    server.allow_reuse_address = 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.