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

How to use delete_function method in localstack

Best Python code snippet using localstack_python

sync_client.py

Source:sync_client.py

1import distutils2import distutils.spawn3import logging4import subprocess5import tempfile6import types7from shlex import quote8from ray.tune.error import TuneError9logger = logging.getLogger(__name__)10S3_PREFIX = "s3://"11GS_PREFIX = "gs://"12HDFS_PREFIX = "hdfs://"13ALLOWED_REMOTE_PREFIXES = (S3_PREFIX, GS_PREFIX, HDFS_PREFIX)14noop_template = ": {target}"  # noop in bash15def noop(*args):16    return17def get_sync_client(sync_function, delete_function=None):18    """Returns a sync client.19    Args:20        sync_function (Optional[str|function]): Sync function.21        delete_function (Optional[str|function]): Delete function. Must be22            the same type as sync_function if it is provided.23    Raises:24        ValueError if sync_function or delete_function are malformed.25    """26    if sync_function is None:27        return None28    if delete_function and type(sync_function) != type(delete_function):29        raise ValueError("Sync and delete functions must be of same type.")30    if isinstance(sync_function, types.FunctionType):31        delete_function = delete_function or noop32        client_cls = FunctionBasedClient33    elif isinstance(sync_function, str):34        delete_function = delete_function or noop_template35        client_cls = CommandBasedClient36    else:37        raise ValueError("Sync function {} must be string or function".format(38            sync_function))39    return client_cls(sync_function, sync_function, delete_function)40def get_cloud_sync_client(remote_path):41    """Returns a CommandBasedClient that can sync to/from remote storage.42    Args:43        remote_path (str): Path to remote storage (S3, GS or HDFS).44    Raises:45        ValueError if malformed remote_dir.46    """47    if remote_path.startswith(S3_PREFIX):48        if not distutils.spawn.find_executable("aws"):49            raise ValueError(50                "Upload uri starting with '{}' requires awscli tool"51                " to be installed".format(S3_PREFIX))52        sync_up_template = "aws s3 sync {source} {target} --only-show-errors"53        sync_down_template = sync_up_template54        delete_template = "aws s3 rm {target} --recursive --only-show-errors"55    elif remote_path.startswith(GS_PREFIX):56        if not distutils.spawn.find_executable("gsutil"):57            raise ValueError(58                "Upload uri starting with '{}' requires gsutil tool"59                " to be installed".format(GS_PREFIX))60        sync_up_template = "gsutil rsync -r {source} {target}"61        sync_down_template = sync_up_template62        delete_template = "gsutil rm -r {target}"63    elif remote_path.startswith(HDFS_PREFIX):64        if not distutils.spawn.find_executable("hdfs"):65            raise ValueError("Upload uri starting with '{}' requires hdfs tool"66                             " to be installed".format(HDFS_PREFIX))67        sync_up_template = "hdfs dfs -put -f {source} {target}"68        sync_down_template = "hdfs dfs -get -f {target} {source}"69        delete_template = "hdfs dfs -rm -r {target}"70    else:71        raise ValueError("Upload uri must start with one of: {}"72                         "".format(ALLOWED_REMOTE_PREFIXES))73    return CommandBasedClient(sync_up_template, sync_down_template,74                              delete_template)75class SyncClient:76    """Client interface for interacting with remote storage options."""77    def sync_up(self, source, target):78        """Syncs up from source to target.79        Args:80            source (str): Source path.81            target (str): Target path.82        Returns:83            True if sync initiation successful, False otherwise.84        """85        raise NotImplementedError86    def sync_down(self, source, target):87        """Syncs down from source to target.88        Args:89            source (str): Source path.90            target (str): Target path.91        Returns:92            True if sync initiation successful, False otherwise.93        """94        raise NotImplementedError95    def delete(self, target):96        """Deletes target.97        Args:98            target (str): Target path.99        Returns:100            True if delete initiation successful, False otherwise.101        """102        raise NotImplementedError103    def wait(self):104        """Waits for current sync to complete, if asynchronously started."""105        pass106    def reset(self):107        """Resets state."""108        pass109class FunctionBasedClient(SyncClient):110    def __init__(self, sync_up_func, sync_down_func, delete_func=None):111        self.sync_up_func = sync_up_func112        self.sync_down_func = sync_down_func113        self.delete_func = delete_func or noop114    def sync_up(self, source, target):115        self.sync_up_func(source, target)116        return True117    def sync_down(self, source, target):118        self.sync_down_func(source, target)119        return True120    def delete(self, target):121        self.delete_func(target)122        return True123NOOP = FunctionBasedClient(noop, noop)124class CommandBasedClient(SyncClient):125    def __init__(self,126                 sync_up_template,127                 sync_down_template,128                 delete_template=noop_template):129        """Syncs between two directories with the given command.130        Arguments:131            sync_up_template (str): A runnable string template; needs to132                include replacement fields '{source}' and '{target}'.133            sync_down_template (str): A runnable string template; needs to134                include replacement fields '{source}' and '{target}'.135            delete_template (Optional[str]): A runnable string template; needs136                to include replacement field '{target}'. Noop by default.137        """138        self._validate_sync_string(sync_up_template)139        self._validate_sync_string(sync_down_template)140        self.sync_up_template = sync_up_template141        self.sync_down_template = sync_down_template142        self.delete_template = delete_template143        self.logfile = None144        self.cmd_process = None145    def set_logdir(self, logdir):146        """Sets the directory to log sync execution output in.147        Args:148            logdir (str): Log directory.149        """150        self.logfile = tempfile.NamedTemporaryFile(151            prefix="log_sync_out", dir=logdir, suffix=".log", delete=False)152    def sync_up(self, source, target):153        return self._execute(self.sync_up_template, source, target)154    def sync_down(self, source, target):155        return self._execute(self.sync_down_template, source, target)156    def delete(self, target):157        if self.is_running:158            logger.warning("Last sync client cmd still in progress, skipping.")159            return False160        final_cmd = self.delete_template.format(target=quote(target))161        logger.debug("Running delete: {}".format(final_cmd))162        self.cmd_process = subprocess.Popen(163            final_cmd, shell=True, stderr=subprocess.PIPE, stdout=self.logfile)164        return True165    def wait(self):166        if self.cmd_process:167            _, error_msg = self.cmd_process.communicate()168            error_msg = error_msg.decode("ascii")169            code = self.cmd_process.returncode170            args = self.cmd_process.args171            self.cmd_process = None172            if code != 0:173                raise TuneError("Sync error. Ran command: {}\n"174                                "Error message ({}): {}".format(175                                    args, code, error_msg))176    def reset(self):177        if self.is_running:178            logger.warning("Sync process still running but resetting anyways.")179        self.cmd_process = None180    @property181    def is_running(self):182        """Returns whether a sync or delete process is running."""183        if self.cmd_process:184            self.cmd_process.poll()185            return self.cmd_process.returncode is None186        return False187    def _execute(self, sync_template, source, target):188        """Executes sync_template on source and target."""189        if self.is_running:190            logger.warning("Last sync client cmd still in progress, skipping.")191            return False192        final_cmd = sync_template.format(193            source=quote(source), target=quote(target))194        logger.debug("Running sync: {}".format(final_cmd))195        self.cmd_process = subprocess.Popen(196            final_cmd, shell=True, stderr=subprocess.PIPE, stdout=self.logfile)197        return True198    @staticmethod199    def _validate_sync_string(sync_string):200        if not isinstance(sync_string, str):201            raise ValueError("{} is not a string.".format(sync_string))202        if "{source}" not in sync_string:203            raise ValueError("Sync template missing '{source}'.")204        if "{target}" not in sync_string:...

resources.py

Source:resources.py

...22    def delete(self):23        """24        @summary: Deletes the resource25        """26        self.delete_function(self.resource_id)27class ResourcePool(object):28    """29    @summary: Pool of resources to be tracked for deletion.30    """31    def __init__(self):32        self.resources = []33    def add(self, resource_id, delete_function):34        """35        @summary: Adds a resource to the resource pool36        @param resource_id: Unique identifier of resource37        @type resource_id: string38        @param delete_function: The function to be called to delete a server39        @type delete_function: Function Pointer40        """...

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.