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

Copied to Clipboard!

How to use cleanup_tmp_files method in localstack

Best Python code snippet using localstack_python

AutoPkgCacheCleanup.py

Source:AutoPkgCacheCleanup.py

1#!/usr/local/autopkg/python2#3# James Stewart @JGStew - 20214#5# Related:6# - https://github.com/jgstew/bigfix_prefetch/blob/master/prefetch_from_dictionary.py7#8"""See docstring for AutoPkgCacheCleanup class"""9import datetime10import pathlib11from autopkglib import (  # pylint: disable=import-error,wrong-import-position,unused-import12    Processor,13    ProcessorError,14)15__all__ = ["AutoPkgCacheCleanup"]16def get_file_mtime(file_path):17    return datetime.datetime.fromtimestamp(file_path.stat().st_mtime)18def get_file_age(file_path):19    return datetime.datetime.today() - get_file_mtime(file_path)20class AutoPkgCacheCleanup(Processor):  # pylint: disable=invalid-name21    """checks that assert_string is within input_string"""22    description = __doc__23    input_variables = {24        "cleanup_max_age_days": {25            "required": False,26            "default": 30,27            "description": "Delete cached downloads older than `cleanup_max_age_days` days",28        },29        "cleanup_min_size_mb": {30            "required": False,31            "default": 2,32            "description": "Delete cached files greater than `cleanup_min_size` size. Defaults to 2. (2MB)",33        },34        "cleanup_empty_files": {35            "required": False,36            "default": True,37            "description": "Delete empty files? Default `True`",38        },39        "cleanup_tmp_files": {40            "required": False,41            "default": True,42            "description": "Delete tmp files older than `cleanup_max_age_days` days? Default `True`",43        },44        "cleanup_cache_all": {45            "required": False,46            "default": False,47            "description": "Cleanup cache for all recipes? Default `False`",48        },49    }50    output_variables = {51        "num_files_deleted": {"description": ("The number of files deleted")},52        "total_cache_size_gb": {53            "description": ("Total size of the download cache in GB.")54        },55    }56    __doc__ = description57    def main(self):58        """Execution starts here"""59        cleanup_max_age_days = int(self.env.get("cleanup_max_age_days", 30))60        cleanup_min_size = int(self.env.get("cleanup_min_size_mb", 2)) * 1024 * 102461        cleanup_empty_files = bool(self.env.get("cleanup_empty_files", True))62        cleanup_tmp_files = bool(self.env.get("cleanup_tmp_files", True))63        cleanup_cache_all = bool(self.env.get("cleanup_cache_all", True))64        RECIPE_CACHE_DIR = self.env.get("RECIPE_CACHE_DIR", None)65        num_files_deleted = 066        total_cache_size = 067        path_cache = pathlib.Path(RECIPE_CACHE_DIR)68        file_cache_pattern = "downloads/*"69        if cleanup_cache_all:70            path_cache = pathlib.Path(RECIPE_CACHE_DIR).parent71            file_cache_pattern = "*/downloads/*"72        self.output(path_cache.absolute(), 3)73        for file_path in path_cache.glob(file_cache_pattern):74            file_size = file_path.stat().st_size75            total_cache_size = total_cache_size + file_size76            file_age_days = get_file_age(file_path).days77            if cleanup_empty_files and file_size == 0:78                # delete empty files:79                file_path.unlink(missing_ok=True)80                num_files_deleted = num_files_deleted + 181            else:82                if file_age_days > cleanup_max_age_days:83                    # example tmp filename: tmpadsflsdf84                    if (85                        cleanup_tmp_files86                        and file_path.name.startswith("tmp")87                        and "." not in file_path.name88                    ):89                        # delete old tmp files:90                        file_path.unlink(missing_ok=True)91                        num_files_deleted = num_files_deleted + 192                    else:93                        if file_size > cleanup_min_size:94                            # delete old files:95                            file_path.unlink(missing_ok=True)96                            num_files_deleted = num_files_deleted + 197        self.env["num_files_deleted"] = num_files_deleted98        self.env["total_cache_size_gb"] = round(99            total_cache_size / (1024 * 1024 * 1024), 2100        )101if __name__ == "__main__":102    PROCESSOR = AutoPkgCacheCleanup()...

verify_files.py

Source:verify_files.py

...29    print(f"Found {len(tmp_files)} .wav files in tmp directory")30    return tmp_files31def get_verified_tmp_files(verification_results):32    return [result["path"] for result in verification_results if result["verified"]]33def cleanup_tmp_files():34    tmp_files = get_tmp_files()35    results = verify(tmp_files)36    37    verified_files = get_verified_tmp_files(results)38    for verified in verified_files:39        info = get_wav_info(verified)40        month = info["start_time"].strftime("%B")41        data_dir = config._GLIDER_DATASET_DIRECTORY.joinpath(month)42        if not data_dir.exists() and not data_dir.is_file():43            data_dir.mkdir(parents=False, exist_ok=False)44        filepath = data_dir.joinpath(info["filename"].name)45        src, dest = info["filename"], filepath46        print(f"Moving verified file {src.name} from {src} to {dest}")47        shutil.move(src, dest)48def verify_non_tmp_files():49    audiofiles = list(config.DATASET_DIRECTORY.glob("**/*.wav"))50    results = verify(audiofiles)51    not_verified = [result["path"] for result in results if not result["verified"]]52    for file in not_verified:53        print(f"File {str(file)} could not be verified, removing...")54        os.remove(file)55def main():56    cleanup_tmp_files()57    verify_non_tmp_files()58if __name__ == "__main__":59    print(__file__)...

base.py

Source:base.py

1import logging2import os3log = logging.getLogger(__name__)4class BaseFetch(object):5    fetch_method = 'base'6    def __init__(self):7        self.local_path = None8        self.cleanup_tmp_files = True9        # remote_resource is whatever was fetched, ie a url or galaxy content spec10        # just an identifier to use in messages11        self.remote_resource = None12    def find(self):13        '''Lookup the requested spec, and if found, return a RepositorySpec14        More or less resolve the ambiquious RequirementSpec to a particular concrete15        RepositorySpec'''16        raise NotImplementedError17    def fetch(self, find_results=None):18        '''Get the content archive, save it to a file locally and return the path to that file'''19        raise NotImplementedError20    def cleanup(self):21        if not self.cleanup_tmp_files:22            log.info("cleanup_tmp_files is false, Not removing the tmp file %s fetched from %s",23                     self.local_path, self.remote_resource)24            return25        log.debug("Removing the tmp file %s fetched from %s",26                  self.local_path, self.remote_resource)27        try:28            os.unlink(self.local_path)29        except (OSError, IOError) as e:...

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.