How to use create_snapshot_schedule method in localstack

Best Python code snippet using localstack_python

dellemc_isilon_snapshotschedule.py

Source:dellemc_isilon_snapshotschedule.py

...360                         ['schedules'][0]['duration'])361                snapshot_schedule_modify['duration'] = retention_in_sec362                modified = True363        return modified, snapshot_schedule_modify364    def create_snapshot_schedule(self, name, alias, effective_path,365                                 desired_retention, retention_unit,366                                 pattern, schedule):367        """Create snapshot schedule"""368        if effective_path is None:369            self.module.fail_json(msg="Path is mandatory while creating "370                                      "snapshot schedule")371        if pattern is None:372            self.module.fail_json(msg="Pattern is mandatory while creating "373                                      "snapshot schedule")374        if schedule is None:375            self.module.fail_json(msg="Schedule is mandatory while creating "376                                      "snapshot schedule")377        duration = 0378        if desired_retention is not None:379            if retention_unit == 'hours':380                duration = desired_retention * 60 * 60381            else:382                duration = desired_retention * 24 * 60 * 60383            if duration < 7200:384                self.module.fail_json(msg="The snapshot desired retention "385                                          "must be at least 2 hours")386        try:387            snapshot_schedule_create_param = utils.isi_sdk.\388                SnapshotScheduleCreateParams(name=name, alias=alias,389                                             path=effective_path,390                                             duration=duration,391                                             pattern=pattern,392                                             schedule=schedule)393            self.api_instance.create_snapshot_schedule(394                snapshot_schedule_create_param)395            return True396        except Exception as e:397            error_msg = self.determine_error(error_obj=e)398            error_message = 'Failed to create snapshot schedule: {0} for ' \399                            'path {1} with error: {2}'.format(400                                name, effective_path, error_msg)401            LOG.error(error_message)402            self.module.fail_json(msg=error_message)403    def validate_new_name(self, new_name):404        """validate if the snapshot schedule with new_name already exists"""405        snapshot_schedules = self.get_details(new_name)406        if snapshot_schedules is not None:407            error_message = 'Snapshot schedule with name {0} already exists'.\408                format(new_name)409            LOG.error(error_message)410            self.module.fail_json(msg=error_message)411    def rename_snapshot_schedule(self, snapshot_schedule_details, new_name):412        """Rename snapshot schedule"""413        try:414            snapshot_schedule_update_param = utils.isi_sdk.SnapshotSchedule(415                name=new_name)416            self.api_instance.update_snapshot_schedule(417                snapshot_schedule_update_param,418                snapshot_schedule_details['schedules'][0]['name'])419            return True420        except Exception as e:421            error_msg = self.determine_error(error_obj=e)422            error_message = 'Failed to rename snapshot schedule {0} with ' \423                            'error : {1}'.\424                format(snapshot_schedule_details['schedules'][0]['name'],425                       error_msg)426            LOG.error(error_message)427            self.module.fail_json(msg=error_message)428    def modify_snapshot_schedule(self, name,429                                 snapshot_schedule_modification_details):430        """Modify snapshot schedule"""431        snapshot_schedule_update_param = self.isi_sdk.SnapshotSchedule()432        try:433            if 'alias' in snapshot_schedule_modification_details:434                snapshot_schedule_update_param.alias = \435                    snapshot_schedule_modification_details['alias']436            if 'pattern' in snapshot_schedule_modification_details:437                snapshot_schedule_update_param.pattern = \438                    snapshot_schedule_modification_details['pattern']439            if 'schedule' in snapshot_schedule_modification_details:440                snapshot_schedule_update_param.schedule = \441                    snapshot_schedule_modification_details['schedule']442            if 'duration' in snapshot_schedule_modification_details:443                snapshot_schedule_update_param.duration = \444                    snapshot_schedule_modification_details['duration']445            self.api_instance.update_snapshot_schedule(446                snapshot_schedule_update_param, name)447            return True448        except Exception as e:449            error_msg = self.determine_error(error_obj=e)450            error_message = 'Failed to modify snapshot schedule {0} with ' \451                            'error : {1}'.format(name, error_msg)452            LOG.error(error_message)453            self.module.fail_json(msg=error_message)454    def delete_snapshot_schedule(self, name):455        """Delete snapshot schedule"""456        try:457            self.api_instance.delete_snapshot_schedule(name)458            return True459        except Exception as e:460            error_msg = self.determine_error(error_obj=e)461            error_message = 'Failed to delete snapshot schedule: {0} with ' \462                            'error: {1}'.format(name, error_msg)463            LOG.error(error_message)464            self.module.fail_json(msg=error_message)465    def determine_error(self, error_obj):466        """Determine the error message to return"""467        if isinstance(error_obj, utils.ApiException):468            error = re.sub("[\n \"]+", ' ', str(error_obj.body))469        else:470            error = error_obj471        return error472    def perform_module_operation(self):473        """474        Perform different actions on snapshot schedule module based on475        parameters chosen in playbook476        """477        name = self.module.params['name']478        state = self.module.params['state']479        access_zone = self.module.params['access_zone']480        path = self.module.params['path']481        new_name = self.module.params['new_name']482        pattern = self.module.params['pattern']483        schedule = self.module.params['schedule']484        desired_retention = self.module.params['desired_retention']485        retention_unit = self.module.params['retention_unit']486        alias = self.module.params['alias']487        # result is a dictionary that contains changed status and snapshot488        # schedule details489        result = dict(490            changed=False,491            snapshot_schedule_details=''492        )493        effective_path = None494        if access_zone is not None:495            if access_zone.lower() == 'system':496                effective_path = path497            else:498                if path:499                    effective_path = self.get_zone_base_path(access_zone) + \500                        path501        if desired_retention:502            self.validate_desired_retention(desired_retention)503        snapshot_schedule_details = self.get_details(name)504        is_schedule_modified = False505        snapshot_schedule_modification_details = dict()506        if snapshot_schedule_details is not None:507            is_schedule_modified, snapshot_schedule_modification_details = \508                self.check_snapshot_schedule_modified(509                    snapshot_schedule_details, alias, desired_retention,510                    retention_unit, effective_path, pattern, schedule)511        if state == 'present' and not snapshot_schedule_details:512            LOG.info("Creating new snapshot schedule: %s for path: %s",513                     name, effective_path)514            result['changed'] = self.create_snapshot_schedule(515                name, alias, effective_path, desired_retention,516                retention_unit, pattern, schedule) or result['changed']517        if state == 'present' and new_name is not None:518            if len(new_name) == 0:519                self.module.fail_json(msg="Please provide valid string for "520                                          "new_name")521            if snapshot_schedule_details is None:522                self.module.fail_json(msg="Snapshot schedule not found.")523            if new_name != snapshot_schedule_details['schedules'][0]['name']:524                self.validate_new_name(new_name)525                LOG.info("Renaming snapshot schedule %s to new name %s",526                         name, new_name)527                result['changed'] = self.rename_snapshot_schedule(528                    snapshot_schedule_details, new_name) or result['changed']...

snapshotschedule.py

Source:snapshotschedule.py

...344                         ['schedules'][0]['duration'])345                snapshot_schedule_modify['duration'] = retention_in_sec346                modified = True347        return modified, snapshot_schedule_modify348    def create_snapshot_schedule(self, name, alias, effective_path,349                                 desired_retention, retention_unit,350                                 pattern, schedule):351        """Create snapshot schedule"""352        if effective_path is None:353            self.module.fail_json(msg="Path is mandatory while creating "354                                      "snapshot schedule")355        if pattern is None:356            self.module.fail_json(msg="Pattern is mandatory while creating "357                                      "snapshot schedule")358        if schedule is None:359            self.module.fail_json(msg="Schedule is mandatory while creating "360                                      "snapshot schedule")361        duration = 0362        if desired_retention is not None:363            if retention_unit == 'hours':364                duration = desired_retention * 60 * 60365            else:366                duration = desired_retention * 24 * 60 * 60367            if duration < 7200:368                self.module.fail_json(msg="The snapshot desired retention "369                                          "must be at least 2 hours")370        try:371            snapshot_schedule_create_param = utils.isi_sdk.\372                SnapshotScheduleCreateParams(name=name, alias=alias,373                                             path=effective_path,374                                             duration=duration,375                                             pattern=pattern,376                                             schedule=schedule)377            self.api_instance.create_snapshot_schedule(378                snapshot_schedule_create_param)379            return True380        except Exception as e:381            error_msg = self.determine_error(error_obj=e)382            error_message = 'Failed to create snapshot schedule: {0} for ' \383                            'path {1} with error: {2}'.format(384                                name, effective_path, error_msg)385            LOG.error(error_message)386            self.module.fail_json(msg=error_message)387    def validate_new_name(self, new_name):388        """validate if the snapshot schedule with new_name already exists"""389        snapshot_schedules = self.get_details(new_name)390        if snapshot_schedules is not None:391            error_message = 'Snapshot schedule with name {0} already exists'.\392                format(new_name)393            LOG.error(error_message)394            self.module.fail_json(msg=error_message)395    def rename_snapshot_schedule(self, snapshot_schedule_details, new_name):396        """Rename snapshot schedule"""397        try:398            snapshot_schedule_update_param = utils.isi_sdk.SnapshotSchedule(399                name=new_name)400            self.api_instance.update_snapshot_schedule(401                snapshot_schedule_update_param,402                snapshot_schedule_details['schedules'][0]['name'])403            return True404        except Exception as e:405            error_msg = self.determine_error(error_obj=e)406            error_message = 'Failed to rename snapshot schedule {0} with ' \407                            'error : {1}'.\408                format(snapshot_schedule_details['schedules'][0]['name'],409                       error_msg)410            LOG.error(error_message)411            self.module.fail_json(msg=error_message)412    def modify_snapshot_schedule(self, name,413                                 snapshot_schedule_modification_details):414        """Modify snapshot schedule"""415        snapshot_schedule_update_param = self.isi_sdk.SnapshotSchedule()416        try:417            if 'alias' in snapshot_schedule_modification_details:418                snapshot_schedule_update_param.alias = \419                    snapshot_schedule_modification_details['alias']420            if 'pattern' in snapshot_schedule_modification_details:421                snapshot_schedule_update_param.pattern = \422                    snapshot_schedule_modification_details['pattern']423            if 'schedule' in snapshot_schedule_modification_details:424                snapshot_schedule_update_param.schedule = \425                    snapshot_schedule_modification_details['schedule']426            if 'duration' in snapshot_schedule_modification_details:427                snapshot_schedule_update_param.duration = \428                    snapshot_schedule_modification_details['duration']429            self.api_instance.update_snapshot_schedule(430                snapshot_schedule_update_param, name)431            return True432        except Exception as e:433            error_msg = self.determine_error(error_obj=e)434            error_message = 'Failed to modify snapshot schedule {0} with ' \435                            'error : {1}'.format(name, error_msg)436            LOG.error(error_message)437            self.module.fail_json(msg=error_message)438    def delete_snapshot_schedule(self, name):439        """Delete snapshot schedule"""440        try:441            self.api_instance.delete_snapshot_schedule(name)442            return True443        except Exception as e:444            error_msg = self.determine_error(error_obj=e)445            error_message = 'Failed to delete snapshot schedule: {0} with ' \446                            'error: {1}'.format(name, error_msg)447            LOG.error(error_message)448            self.module.fail_json(msg=error_message)449    def determine_error(self, error_obj):450        """Determine the error message to return"""451        if isinstance(error_obj, utils.ApiException):452            error = re.sub("[\n \"]+", ' ', str(error_obj.body))453        else:454            error = error_obj455        return error456    def perform_module_operation(self):457        """458        Perform different actions on snapshot schedule module based on459        parameters chosen in playbook460        """461        name = self.module.params['name']462        state = self.module.params['state']463        access_zone = self.module.params['access_zone']464        path = self.module.params['path']465        new_name = self.module.params['new_name']466        pattern = self.module.params['pattern']467        schedule = self.module.params['schedule']468        desired_retention = self.module.params['desired_retention']469        retention_unit = self.module.params['retention_unit']470        alias = self.module.params['alias']471        # result is a dictionary that contains changed status and snapshot472        # schedule details473        result = dict(474            changed=False,475            snapshot_schedule_details=''476        )477        effective_path = None478        if path:479            if access_zone.lower() == 'system':480                effective_path = path.rstrip("/")481            else:482                effective_path = self.get_zone_base_path(access_zone) + \483                    path.rstrip("/")484        if desired_retention:485            self.validate_desired_retention(desired_retention)486        snapshot_schedule_details = self.get_details(name)487        is_schedule_modified = False488        snapshot_schedule_modification_details = dict()489        if snapshot_schedule_details is not None:490            is_schedule_modified, snapshot_schedule_modification_details = \491                self.check_snapshot_schedule_modified(492                    snapshot_schedule_details, alias, desired_retention,493                    retention_unit, effective_path, pattern, schedule)494        if state == 'present' and not snapshot_schedule_details:495            LOG.debug("Creating new snapshot schedule: %s for path: %s",496                      name, effective_path)497            result['changed'] = self.create_snapshot_schedule(498                name, alias, effective_path, desired_retention,499                retention_unit, pattern, schedule) or result['changed']500        if state == 'present' and new_name is not None:501            if len(new_name) == 0:502                self.module.fail_json(msg="Please provide valid string for "503                                          "new_name")504            if snapshot_schedule_details is None:505                self.module.fail_json(msg="Snapshot schedule not found.")506            if new_name != snapshot_schedule_details['schedules'][0]['name']:507                self.validate_new_name(new_name)508                LOG.info("Renaming snapshot schedule %s to new name %s",509                         name, new_name)510                result['changed'] = self.rename_snapshot_schedule(511                    snapshot_schedule_details, new_name) or result['changed']...

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.