def delete_replica_set(name: str, ns: str = "default",
                       label_selector: str = "name in ({name})",
                       secrets: Secrets = None):
    """
    Delete a replica set by `name` in the namespace `ns`.

    The replica set is deleted without a graceful period to trigger an abrupt
    termination.

    The selected resources are matched by the given `label_selector`.
    """
    label_selector = label_selector.format(name=name)
    api = create_k8s_api_client(secrets)
    v1 = client.ExtensionsV1beta1Api(api)
    if label_selector:
        ret = v1.list_namespaced_replica_set(ns, label_selector=label_selector)
    else:
        ret = v1.list_namespaced_replica_set(ns)

    logger.debug("Found {d} replica sets named '{n}'".format(
        d=len(ret.items), n=name))

    body = client.V1DeleteOptions()
    for r in ret.items:
        v1.delete_namespaced_replica_set(r.metadata.name, ns, body=body) 

def delete_pods(name: str, ns: str = "default",
                label_selector: str = "name in ({name})",
                secrets: Secrets = None):
    """
    Delete pods by `name` in the namespace `ns`.

    The pods are deleted without a graceful period to trigger an abrupt
    termination.

    The selected resources are matched by the given `label_selector`.
    """
    label_selector = label_selector.format(name=name)
    api = create_k8s_api_client(secrets)
    v1 = client.CoreV1Api(api)
    if label_selector:
        ret = v1.list_namespaced_pod(ns, label_selector=label_selector)
    else:
        ret = v1.list_namespaced_pod(ns)

    logger.debug("Found {d} pods named '{n}'".format(
        d=len(ret.items), n=name))

    body = client.V1DeleteOptions()
    for p in ret.items:
        v1.delete_namespaced_pod(p.metadata.name, ns, body=body) 

def service_is_initialized(name: str, ns: str = "default",
                           label_selector: str = "name in ({name})",
                           secrets: Secrets = None):
    """
    Lookup a service endpoint by its name and raises :exc:`FailedProbe` when
    the service was not found or not initialized.
    """
    label_selector = label_selector.format(name=name)
    api = create_k8s_api_client(secrets)

    v1 = client.CoreV1Api(api)
    if label_selector:
        ret = v1.list_namespaced_service(ns, label_selector=label_selector)
    else:
        ret = v1.list_namespaced_service(ns)

    logger.debug("Found {d} service(s) named '{n}' ins ns '{s}'".format(
        d=len(ret.items), n=name, s=ns))

    if not ret.items:
        raise ActivityFailed(
            "service '{name}' is not initialized".format(name=name))

    return True 

def _drain_queue(self):
        """
        Logic:
            - send message to server when server is alive
            - update local db
        """
        while True:
            message = await self._queue.get()
            if message is None:
                logger.info("Resent messages: %s", self._db)
                for _, v in self._db.items():
                    await self._ws.write_message(v)
                continue

            if 'udid' in message:  # ping消息不包含在裡面
                udid = message['udid']
                update_recursive(self._db, {udid: message})
            self._queue.task_done()

            if self._ws:
                try:
                    await self._ws.write_message(message)
                    logger.debug("websocket send: %s", message)
                except TypeError as e:
                    logger.info("websocket write_message error: %s", e) 

def _push_stf(self,
                  path: str,
                  dest: str,
                  zipfile_path: str,
                  mode=0o755):
        """ push minicap and minitouch from zip """
        with zipfile.ZipFile(zipfile_path) as z:
            if path not in z.namelist():
                logger.warning("stf stuff %s not found", path)
                return
            src_info = z.getinfo(path)
            dest_info = self._device.sync.stat(dest)
            if dest_info.size == src_info.file_size and dest_info.mode & mode == mode:
                logger.debug("%s already pushed %s", self, path)
                return
            with z.open(path) as f:
                self._device.sync.push(f, dest, mode) 

def mirror_download(url: str, target: str) -> str:
    """
    Returns:
        target path
    """
    if os.path.exists(target):
        return target
    github_host = "https://github.com"
    if url.startswith(github_host):
        mirror_url = "http://tool.appetizer.io" + url[len(
            github_host):]  # mirror of github
        try:
            return download(mirror_url, target)
        except (requests.RequestException, ValueError) as e:
            logger.debug("download from mirror error, use origin source")

    return download(url, target) 

def send_click(self, x, y):
        if self._click_before_delay:
            self.logger.debug("click before delay %.1f seconds",
                              self._click_after_delay)
            time.sleep(self._click_before_delay)

        # TODO(ssx): should use a better way
        # event callbacks for report generate
        for callback_func in self._event_callbacks['send_click']:
            callback_func(x, y)

        self._d.click(x, y)

        if self._click_after_delay:
            self.logger.debug("click after delay %.1f seconds",
                              self._click_after_delay)
            time.sleep(self._click_after_delay) 

def mirror_download(url: str, filename=None, logger: logging.Logger = logger):
    """
    Download from mirror, then fallback to origin url
    """
    storepath = gen_cachepath(url)
    if not filename:
        filename = os.path.basename(url)
    github_host = "https://github.com"
    if url.startswith(github_host):
        mirror_url = "https://tool.appetizer.io" + url[len(
            github_host):]  # mirror of github
        try:
            return cache_download(mirror_url,
                                  filename,
                                  timeout=60,
                                  storepath=storepath,
                                  logger=logger)
        except (requests.RequestException, FileNotFoundError,
                AssertionError) as e:
            logger.debug("download error from mirror(%s), use origin source", e)

    return cache_download(url, filename, storepath=storepath, logger=logger) 

def push_url(self, url, dest=None, mode=0o755, tgz=False, extract_name=None):  # yapf: disable
        path = mirror_download(url,
                               filename=os.path.basename(url),
                               logger=self.logger)
        if tgz:
            tar = tarfile.open(path, 'r:gz')
            path = os.path.join(os.path.dirname(path), extract_name)
            tar.extract(extract_name,
                        os.path.dirname(path))  # zlib.error may raise

        if not dest:
            dest = "/data/local/tmp/" + os.path.basename(path)

        self.logger.debug("Push to %s:0%o", dest, mode)
        self._device.sync.push(path, dest, mode=mode)
        return dest 

def is_atx_agent_outdated(self):
        """
        Returns:
            bool
        """
        agent_version = self._device.shell(self.atx_agent_path +
                                           " version").strip()
        if agent_version == "dev":
            self.logger.info("skip version check for atx-agent dev")
            return False

        # semver major.minor.patch
        try:
            real_ver = list(map(int, agent_version.split(".")))
            want_ver = list(map(int, __atx_agent_version__.split(".")))
        except ValueError:
            return True

        self.logger.debug("Real version: %s, Expect version: %s", real_ver,
                          want_ver)

        if real_ver[:2] != want_ver[:2]:
            return True

        return real_ver[2] < want_ver[2] 

def validate(ctx: click.Context, source: str,
             no_verify_tls: bool = False) -> Experiment:
    """Validate the experiment at SOURCE."""
    settings = load_settings(ctx.obj["settings_path"])

    try:
        experiment = load_experiment(
            source, settings, verify_tls=not no_verify_tls)
    except InvalidSource as x:
        logger.error(str(x))
        logger.debug(x)
        ctx.exit(1)

    try:
        notify(settings, ValidateFlowEvent.ValidateStarted, experiment)
        ensure_experiment_is_valid(experiment)
        notify(settings, ValidateFlowEvent.ValidateCompleted, experiment)
        logger.info("experiment syntax and semantic look valid")
    except ChaosException as x:
        notify(settings, ValidateFlowEvent.ValidateFailed, experiment, x)
        logger.error(str(x))
        logger.debug(x)
        ctx.exit(1)

    return experiment 

def discover(ctx: click.Context, package: str,
             discovery_path: str = "./discovery.json",
             no_system_info: bool = False,
             no_install: bool = False) -> Discovery:
    """Discover capabilities and experiments."""
    settings = load_settings(ctx.obj["settings_path"])
    try:
        notify(settings, DiscoverFlowEvent.DiscoverStarted, package)
        discovery = disco(
            package_name=package, discover_system=not no_system_info,
            download_and_install=not no_install)
    except DiscoveryFailed as err:
        notify(settings, DiscoverFlowEvent.DiscoverFailed, package, err)
        logger.debug("Failed to discover {}".format(package), exc_info=err)
        logger.fatal(str(err))
        return

    with open(discovery_path, "w") as d:
        d.write(json.dumps(discovery, indent=2, default=encoder))
    logger.info("Discovery outcome saved in {p}".format(
        p=discovery_path))

    notify(settings, DiscoverFlowEvent.DiscoverCompleted, discovery)
    return discovery 

def device_watch(wda_directory: str):
    """
    When iOS device plugin, launch WDA
    """
    lock = locks.Lock()  # WDA launch one by one

    async for event in idb.track_devices():
        if event.udid.startswith("ffffffffffffffffff"):
            logger.debug("Invalid event: %s", event)
            continue
        logger.debug("Event: %s", event)
        if event.present:
            d = idb.WDADevice(event.udid, lock=lock, callback=_device_callback)
            d.wda_directory = wda_directory
            idevices[event.udid] = d
            d.start()
        else:  # offline
            await idevices[event.udid].stop()
            idevices.pop(event.udid) 

def wda_status(self):
        """
        Returns:
            dict or None
        """
        try:
            request = httpclient.HTTPRequest(self.wda_device_url + "/status",
                                             connect_timeout=3,
                                             request_timeout=15)
            client = httpclient.AsyncHTTPClient()
            resp = await client.fetch(request)
            info = json.loads(resp.body)
            self.__wda_info = info
            return info
        except httpclient.HTTPError as e:
            logger.debug("%s request wda/status error: %s", self, e)
            return None
        except (ConnectionResetError, ConnectionRefusedError):
            logger.debug("%s waiting for wda", self)
            return None
        except Exception as e:
            logger.warning("%s ping wda unknown error: %s %s", self, type(e),
                           e)
            return None 

def _drain_queue(self):
        """
        Logic:
            - send message to server when server is alive
            - update local db
        """
        while True:
            message = await self._queue.get()
            if message is None:
                logger.info("Resent messages: %s", self._db)
                for _, v in self._db.items():
                    await self._ws.write_message(v)
                continue

            if 'udid' in message:  # ping消息不包含在裡面
                udid = message['udid']
                update_recursive(self._db, {udid: message})
            self._queue.task_done()

            if self._ws:
                try:
                    await self._ws.write_message(message)
                    logger.debug("websocket send: %s", message)
                except TypeError as e:
                    logger.info("websocket write_message error: %s", e) 

def deregister_target(tg_name: str,
                      configuration: Configuration = None,
                      secrets: Secrets = None) -> AWSResponse:
    """Deregisters one random target from target group"""
    client = aws_client('elbv2', configuration, secrets)
    tg_arn = get_target_group_arns(tg_names=[tg_name], client=client)
    tg_health = get_targets_health_description(tg_arns=tg_arn, client=client)
    random_target = random.choice(
        tg_health[tg_name]['TargetHealthDescriptions'])

    logger.debug("Deregistering target {} from target group {}".format(
        random_target['Target']['Id'], tg_name))

    try:
        return client.deregister_targets(
            TargetGroupArn=tg_arn[tg_name],
            Targets=[{
                'Id': random_target['Target']['Id'],
                'Port': random_target['Target']['Port']
            }]
        )
    except ClientError as e:
        raise FailedActivity('Exception detaching %s: %s' % (
            tg_name, e.response['Error']['Message'])) 

def delete_load_balancer(load_balancer_names: List[str],
                         configuration: Configuration = None,
                         secrets: Secrets = None):
    """
    Deletes the provided load balancer(s).

    Parameters:
        - load_balancer_names: a list of load balancer names
    """
    client = aws_client('elbv2', configuration, secrets)
    load_balancers = get_load_balancer_arns(load_balancer_names, client)

    for k, v in load_balancers.items():
        if k not in ('application', 'network'):
            continue

        for l in v:
            logger.debug('Deleting load balancer %s' % l)
            client.delete_load_balancer(LoadBalancerArn=l)


###############################################################################
# Private functions
############################################################################### 

def get_target_group_arns(tg_names: List[str],
                          client: boto3.client) -> Dict:
    """
    Return list of target group ARNs based on list of target group names

    return structure:
    {
        "TargetGroupName": "TargetGroupArn",
        ....
    }
    """
    logger.debug("Target group name(s): {} Looking for ARN"
                 .format(str(tg_names)))
    res = client.describe_target_groups(Names=tg_names)
    tg_arns = {}

    for tg in res['TargetGroups']:
        tg_arns[tg['TargetGroupName']] = tg['TargetGroupArn']
    logger.debug("Target groups ARN: {}".format(str(tg_arns)))

    return tg_arns 

def get_target_group_arns(tg_names: List[str],
                          client: boto3.client) -> Dict:
    """
    Return list of target group ARNs based on list of target group names

    return structure:
    {
        "TargetGroupName": "TargetGroupArn",
        ....
    }
    """
    logger.debug("Target group name(s): {} Looking for ARN"
                 .format(str(tg_names)))
    res = client.describe_target_groups(Names=tg_names)
    tg_arns = {}

    for tg in res['TargetGroups']:
        tg_arns[tg['TargetGroupName']] = tg['TargetGroupArn']
    logger.debug("Target groups ARNs: {}".format(str(tg_arns)))

    return tg_arns 

def pods_in_phase(label_selector: str, phase: str = "Running",
                  ns: str = "default", secrets: Secrets = None) -> bool:
    """
    Lookup a pod by `label_selector` in the namespace `ns`.

    Raises :exc:`chaoslib.exceptions.ActivityFailed` when the state is not
    as expected.
    """
    api = create_k8s_api_client(secrets)

    v1 = client.CoreV1Api(api)
    if label_selector:
        ret = v1.list_namespaced_pod(ns, label_selector=label_selector)
        logger.debug("Found {d} pods matching label '{n}' in ns '{s}'".format(
            d=len(ret.items), n=label_selector, s=ns))
    else:
        ret = v1.list_namespaced_pod(ns)
        logger.debug("Found {d} pods in ns '{n}'".format(
            d=len(ret.items), n=ns))

    if not ret.items:
        raise ActivityFailed(
            "no pods '{name}' were found".format(name=label_selector))

    for d in ret.items:
        if d.status.phase != phase:
            raise ActivityFailed(
                "pod '{name}' is in phase '{s}' but should be '{p}'".format(
                    name=label_selector, s=d.status.phase, p=phase))

    return True 

def pods_not_in_phase(label_selector: str, phase: str = "Running",
                      ns: str = "default", secrets: Secrets = None) -> bool:
    """
    Lookup a pod by `label_selector` in the namespace `ns`.

    Raises :exc:`chaoslib.exceptions.ActivityFailed` when the pod is in the
    given phase and should not have.
    """
    api = create_k8s_api_client(secrets)

    v1 = client.CoreV1Api(api)
    if label_selector:
        ret = v1.list_namespaced_pod(ns, label_selector=label_selector)
        logger.debug("Found {d} pods matching label '{n}' in ns '{s}'".format(
            d=len(ret.items), n=label_selector, s=ns))
    else:
        ret = v1.list_namespaced_pod(ns)
        logger.debug("Found {d} pods in ns '{n}'".format(
            d=len(ret.items), n=ns))

    if not ret.items:
        raise ActivityFailed(
            "no pods '{name}' were found".format(name=label_selector))

    for d in ret.items:
        if d.status.phase == phase:
            raise ActivityFailed(
                "pod '{name}' should not be in phase '{s}'".format(
                    name=label_selector, s=d.status.phase))

    return True 

def count_pods(label_selector: str, phase: str = None,
               ns: str = "default", secrets: Secrets = None) -> int:
    """
    Count the number of pods matching the given selector in a given `phase`, if
    one is given.
    """
    api = create_k8s_api_client(secrets)

    v1 = client.CoreV1Api(api)
    if label_selector:
        ret = v1.list_namespaced_pod(ns, label_selector=label_selector)
        logger.debug("Found {d} pods matching label '{n}' in ns '{s}'".format(
            d=len(ret.items), n=label_selector, s=ns))
    else:
        ret = v1.list_namespaced_pod(ns)
        logger.debug("Found {d} pods in ns '{n}'".format(
            d=len(ret.items), n=ns))

    if not ret.items:
        return 0

    if not phase:
        return len(ret.items)

    count = 0
    for d in ret.items:
        if d.status.phase == phase:
            count = count + 1

    return count 

def pod_is_not_available(name: str, ns: str = "default",
                         label_selector: str = "name in ({name})",
                         secrets: Secrets = None) -> bool:
    """
    Lookup pods with a `name` label set to the given `name` in the specified
    `ns`.

    Raises :exc:`chaoslib.exceptions.ActivityFailed` when one of the pods
    with the specified `name` is in the `"Running"` phase.
    """
    label_selector = label_selector.format(name=name)
    api = create_k8s_api_client(secrets)

    v1 = client.CoreV1Api(api)
    if label_selector:
        ret = v1.list_namespaced_pod(ns, label_selector=label_selector)
    else:
        ret = v1.list_namespaced_pod(ns)

    logger.debug("Found {d} pod(s) named '{n}' in ns '{s}".format(
        d=len(ret.items), n=name, s=ns))

    for p in ret.items:
        phase = p.status.phase
        logger.debug("Pod '{p}' has status '{s}'".format(
            p=p.metadata.name, s=phase))
        if phase == "Running":
            raise ActivityFailed(
                "pod '{name}' is actually running".format(name=name))

    return True 

def all_pods_healthy(ns: str = "default",
                     secrets: Secrets = None) -> MicroservicesStatus:
    """
    Check all pods in the system are running and available.

    Raises :exc:`chaoslib.exceptions.ActivityFailed` when the state is not
    as expected.
    """
    api = create_k8s_api_client(secrets)
    not_ready = []
    failed = []

    v1 = client.CoreV1Api(api)
    ret = v1.list_namespaced_pod(namespace=ns)
    for p in ret.items:
        phase = p.status.phase
        if phase == "Failed":
            failed.append(p)
        elif phase not in ("Running", "Succeeded"):
            not_ready.append(p)

    logger.debug("Found {d} failed and {n} not ready pods".format(
        d=len(failed), n=len(not_ready)))

    # we probably should list them in the message
    if failed or not_ready:
        raise ActivityFailed("the system is unhealthy")

    return True 

def delete_deployment(name: str, ns: str = "default",
                      label_selector: str = "name in ({name})",
                      secrets: Secrets = None):
    """
    Delete a deployment by `name` in the namespace `ns`.

    The deployment is deleted without a graceful period to trigger an abrupt
    termination.

    The selected resources are matched by the given `label_selector`.
    """
    label_selector = label_selector.format(name=name)
    api = create_k8s_api_client(secrets)

    v1 = client.AppsV1beta1Api(api)
    if label_selector:
        ret = v1.list_namespaced_deployment(ns, label_selector=label_selector)
    else:
        ret = v1.list_namespaced_deployment(ns)

    logger.debug("Found {d} deployments named '{n}'".format(
        d=len(ret.items), n=name))

    body = client.V1DeleteOptions()
    for d in ret.items:
        v1.delete_namespaced_deployment(d.metadata.name, ns, body=body) 

def create_node(meta: Dict[str, Any] = None, spec: Dict[str, Any] = None,
                secrets: Secrets = None) -> client.V1Node:
    """
    Create one new node in the cluster.

    Due to the way things work on certain cloud providers, you won't be able
    to use this meaningfully on them. For instance on GCE, this will likely
    fail.

    See also: https://github.com/kubernetes/community/blob/master/contributors/devel/api-conventions.md#idempotency
    """  # noqa: E501
    api = create_k8s_api_client(secrets)

    v1 = client.CoreV1Api(api)
    body = client.V1Node()

    body.metadata = client.V1ObjectMeta(**meta) if meta else None
    body.spec = client.V1NodeSpec(**spec) if spec else None

    try:
        res = v1.create_node(body)
    except ApiException as x:
        raise ActivityFailed("Creating new node failed: {}".format(x.body))

    logger.debug("Node '{}' created".format(res.metadata.name))

    return res 

def cordon_node(name: str = None, label_selector: str = None,
                secrets: Secrets = None):
    """
    Cordon nodes matching the given label or name, so that no pods
    are scheduled on them any longer.
    """
    api = create_k8s_api_client(secrets)

    v1 = client.CoreV1Api(api)

    nodes = _select_nodes(name=name, label_selector=label_selector,
                          secrets=secrets)

    body = {
        "spec": {
            "unschedulable": True
        }
    }

    for n in nodes:
        try:
            v1.patch_node(n.metadata.name, body)
        except ApiException as x:
            logger.debug("Unscheduling node '{}' failed: {}".format(
                n.metadata.name, x.body))
            raise ActivityFailed("Failed to unschedule node '{}': {}".format(
                n.metadata.name, x.body)) 

def remove_statefulset(name: str = None, ns: str = "default",
                       label_selector: str = None, secrets: Secrets = None):
    """
    Remove a statefulset by `name` in the namespace `ns`.

    The statefulset is removed by deleting it without
        a graceful period to trigger an abrupt termination.

    The selected resources are matched by the given `label_selector`.
    """
    field_selector = "metadata.name={name}".format(name=name)
    api = create_k8s_api_client(secrets)

    v1 = client.AppsV1Api(api)
    if label_selector:
        ret = v1.list_namespaced_stateful_set(
            ns, field_selector=field_selector,
            label_selector=label_selector)
    else:
        ret = v1.list_namespaced_stateful_set(ns,
                                              field_selector=field_selector)

    logger.debug("Found {d} statefulset(s) named '{n}' in ns '{s}'".format(
        d=len(ret.items), n=name, s=ns))

    body = client.V1DeleteOptions()
    for d in ret.items:
        res = v1.delete_namespaced_stateful_set(
            d.metadata.name, ns, body=body) 

def schedule_conversation_deletion(self, peer, delay=5):
        await asyncio.sleep(delay)
        self.send(DeleteHistory(await self.resolve_peer(peer), max_id=999999999, just_clear=True))
        log.debug("Deleted conversation with {}".format(peer))

    # def delete_all_conversations(self):
    #     all_peers = [utils.resolve_id(x[0]) for x in self.session.entities.get_input_list()]
    #     for peer in all_peers:
    #         log.debug("Deleting conversation with {}...".format(peer))
    #         try:
    #             input_entity = self.client.session.entities.get_input_entity(peer[0])
    #             self.client(DeleteHistoryRequest(input_entity, max_id=9999999999999999))
    #         except:
    #             log.error("Couldn't find {}".format(peer[0])) 

def download_profile_photo(self, bot: BotModel, photo_path):
        tmp_file = os.path.join(TMP_DIR, bot.username.replace('@', '') + '.jpg')
        photos = self.get_user_profile_photos(bot.chat_id).photos
        if photos:
            photo_size_object = photos[0][-1]

            await self.__photos_lock.acquire()
            try:
                try:
                    self.download_media(
                        photo_size_object,
                        file_name=tmp_file,
                        block=True
                    )
                except FloodWait as e:
                    # TODO: as the error happens inside of the update worker, this won't work (yet)
                    # Leaving it in as the default behavior should be to raise the FloodWait
                    # when block=True
                    log.debug(f"FloodWait for downloading media ({e.x})")

                if os.path.exists(tmp_file):
                    try:
                        similar = filecmp.cmp(tmp_file, photo_path, shallow=False)
                    except FileNotFoundError:
                        similar = False

                    if not similar:
                        shutil.copy(tmp_file, photo_path)
            finally:
                self.__photos_lock.release()