def setUp(self):
     super(TestEndpoint, self).setUp()
     self.task_cls = utils.TaskOneReturn
     self.task_uuid = 'task-uuid'
     self.task_args = {'context': 'context'}
     self.task_cls_name = reflection.get_class_name(self.task_cls)
     self.task_ep = ep.Endpoint(self.task_cls)
     self.task_result = 1

def prettify_failures(failures, limit=-1):
    """Prettifies a checked commits failures (ignores sensitive data...).

    Example input and output:

    >>> from taskflow.utils import kazoo_utils
    >>> conf = {"hosts": ['localhost:2181']}
    >>> c = kazoo_utils.make_client(conf)
    >>> c.start(timeout=1)
    >>> txn = c.transaction()
    >>> txn.create("/test")
    >>> txn.check("/test", 2)
    >>> txn.delete("/test")
    >>> try:
    ...     kazoo_utils.checked_commit(txn)
    ... except kazoo_utils.KazooTransactionException as e:
    ...     print(kazoo_utils.prettify_failures(e.failures, limit=1))
    ...
    [email protected](path='/test') and 2 more...
    >>> c.stop()
    >>> c.close()
    """
    prettier = []
    for (op, r) in failures:
        pretty_op = reflection.get_class_name(op, fully_qualified=False)
        # Pick off a few attributes that are meaningful (but one that don't
        # show actual data, which might not be desired to show...).
        selected_attrs = [
            "path=%r" % op.path,
        ]
        try:
            if op.version != -1:
                selected_attrs.append("version=%s" % op.version)
        except AttributeError:
            pass
        pretty_op += "(%s)" % (", ".join(selected_attrs))
        pretty_cause = reflection.get_class_name(r, fully_qualified=False)
        prettier.append("%[email protected]%s" % (pretty_cause, pretty_op))
    if limit <= 0 or len(prettier) <= limit:
        return ", ".join(prettier)
    else:
        leftover = prettier[limit:]
        prettier = prettier[0:limit]
        return ", ".join(prettier) + " and %s more..." % len(leftover)

def parse_uri(uri, query_duplicates=False):
    """Parses a uri into its components and returns a dictionary containing
    those components.
    """
    # Do some basic validation before continuing...
    if not isinstance(uri, six.string_types):
        raise TypeError("Can only parse string types to uri data, "
                        "and not an object of type %s"
                        % reflection.get_class_name(uri))
    match = _SCHEME_REGEX.match(uri)
    if not match:
        raise ValueError("Uri %r does not start with a RFC 3986 compliant"
                         " scheme" % (uri))
    parsed = network_utils.urlsplit(uri)
    if parsed.query:
        query_params = urlparse.parse_qsl(parsed.query)
        if not query_duplicates:
            query_params = dict(query_params)
        else:
            # Retain duplicates in a list for keys which have duplicates, but
            # for items which are not duplicated, just associate the key with
            # the value.
            tmp_query_params = {}
            for (k, v) in query_params:
                if k in tmp_query_params:
                    p_v = tmp_query_params[k]
                    if isinstance(p_v, list):
                        p_v.append(v)
                    else:
                        p_v = [p_v, v]
                        tmp_query_params[k] = p_v
                else:
                    tmp_query_params[k] = v
            query_params = tmp_query_params
    else:
        query_params = {}
    uri_pieces = {
        'scheme': parsed.scheme,
        'username': parsed.username,
        'password': parsed.password,
        'fragment': parsed.fragment,
        'path': parsed.path,
        'params': query_params,
    }
    for k in ('hostname', 'port'):
        try:
            uri_pieces[k] = getattr(parsed, k)
        except (IndexError, ValueError):
            # The underlying network_utils throws when the host string is empty
            # which it may be in cases where it is not provided.
            #
            # NOTE(harlowja): when https://review.openstack.org/#/c/86921/ gets
            # merged we can just remove this since that error will no longer
            # occur.
            uri_pieces[k] = None
    return AttrDict(**uri_pieces)

 def __init__(self, name=None, provides=None, requires=None,
              auto_extract=True, rebind=None):
     """Initialize task instance"""
     if name is None:
         name = reflection.get_class_name(self)
     if provides is None:
         provides = self.default_provides
     super(Task, self).__init__(name,
                                provides=provides)
     self.rebind = _build_arg_mapping(self.name, requires, rebind,
                                      self.execute, auto_extract)

 def __init__(self, task, uuid, action, arguments, progress_callback,
              timeout, **kwargs):
     self._task = task
     self._task_cls = reflection.get_class_name(task)
     self._uuid = uuid
     self._action = action
     self._event = ACTION_TO_EVENT[action]
     self._arguments = arguments
     self._progress_callback = progress_callback
     self._kwargs = kwargs
     self._watch = time.StopWatch(duration=timeout).start()
     self._state = WAITING
     self.result = futures.Future()

 def wrapper(self, *args, **kwargs):
     base_name = reflection.get_class_name(self, fully_qualified=False)
     if fully_qualified:
         old_name = old_attribute_name
     else:
         old_name = ".".join((base_name, old_attribute_name))
     new_name = ".".join((base_name, new_attribute_name))
     prefix = _KIND_MOVED_PREFIX_TPL % (kind, old_name, new_name)
     out_message = _generate_moved_message(
         prefix, message=message,
         version=version, removal_version=removal_version)
     deprecation(out_message, stacklevel=3)
     return f(self, *args, **kwargs)

def moved_class(new_class, old_class_name, old_module_name, message=None,
                version=None, removal_version=None):
    """Deprecates a class that was moved to another location.

    This will emit warnings when the old locations class is initialized,
    telling where the new and improved location for the old class now is.
    """
    old_name = ".".join((old_module_name, old_class_name))
    new_name = reflection.get_class_name(new_class)
    out_message = _generate_moved_message('Class', old_name, new_name,
                                          message=message, version=version,
                                          removal_version=removal_version)
    return MovedClassProxy(new_class, out_message, 3)

 def __init__(self, task, uuid, action, arguments, progress_callback,
              timeout, **kwargs):
     self._task = task
     self._name = reflection.get_class_name(task)
     self._uuid = uuid
     self._action = action
     self._event = pr.ACTION_TO_EVENT[action]
     self._arguments = arguments
     self._progress_callback = progress_callback
     self._timeout = timeout
     self._kwargs = kwargs
     self._time = time.time()
     self._state = pr.PENDING
     self.result = futures.Future()

 def _atomdetail_by_name(self, atom_name, expected_type=None):
     try:
         ad = self._flowdetail.find(self._atom_name_to_uuid[atom_name])
     except KeyError:
         raise exceptions.NotFound("Unknown atom name: %s" % atom_name)
     else:
         # TODO(harlowja): we need to figure out how to get away from doing
         # these kinds of type checks in general (since they likely mean
         # we aren't doing something right).
         if expected_type and not isinstance(ad, expected_type):
             raise TypeError("Atom %s is not of the expected type: %s"
                             % (atom_name,
                                reflection.get_class_name(expected_type)))
         return ad

def parse_uri(uri):
    """Parses a uri into its components."""
    # Do some basic validation before continuing...
    if not isinstance(uri, six.string_types):
        raise TypeError("Can only parse string types to uri data, "
                        "and not an object of type %s"
                        % reflection.get_class_name(uri))
    match = _SCHEME_REGEX.match(uri)
    if not match:
        raise ValueError("Uri %r does not start with a RFC 3986 compliant"
                         " scheme" % (uri))
    split = netutils.urlsplit(uri)
    return ModifiedSplitResult(scheme=split.scheme, fragment=split.fragment,
                               path=split.path, netloc=split.netloc,
                               query=split.query)

    def check(self, *exc_classes):
        """Check if any of ``exc_classes`` caused the failure.

        Arguments of this method can be exception types or type
        names (stings). If captured exception is instance of
        exception of given type, the corresponding argument is
        returned. Else, None is returned.
        """
        for cls in exc_classes:
            if isinstance(cls, type):
                err = reflection.get_class_name(cls)
            else:
                err = cls
            if err in self._exc_type_names:
                return cls
        return None

def parse_uri(uri, query_duplicates=False):
    """Parses a uri into its components."""
    # Do some basic validation before continuing...
    if not isinstance(uri, six.string_types):
        raise TypeError("Can only parse string types to uri data, "
                        "and not an object of type %s"
                        % reflection.get_class_name(uri))
    match = _SCHEME_REGEX.match(uri)
    if not match:
        raise ValueError("Uri %r does not start with a RFC 3986 compliant"
                         " scheme" % (uri))
    parsed = netutils.urlsplit(uri)
    if parsed.query:
        query_params = urlparse.parse_qsl(parsed.query)
        if not query_duplicates:
            query_params = dict(query_params)
        else:
            # Retain duplicates in a list for keys which have duplicates, but
            # for items which are not duplicated, just associate the key with
            # the value.
            tmp_query_params = {}
            for (k, v) in query_params:
                if k in tmp_query_params:
                    p_v = tmp_query_params[k]
                    if isinstance(p_v, list):
                        p_v.append(v)
                    else:
                        p_v = [p_v, v]
                        tmp_query_params[k] = p_v
                else:
                    tmp_query_params[k] = v
            query_params = tmp_query_params
    else:
        query_params = {}
    return AttrDict(
        scheme=parsed.scheme,
        username=parsed.username,
        password=parsed.password,
        fragment=parsed.fragment,
        path=parsed.path,
        params=query_params,
        hostname=parsed.hostname,
        port=parsed.port)

    def setUp(self):
        super(TestWorker, self).setUp()
        self.task_cls = utils.DummyTask
        self.task_name = reflection.get_class_name(self.task_cls)
        self.broker_url = "test-url"
        self.exchange = "test-exchange"
        self.topic = "test-topic"
        self.threads_count = 5
        self.endpoint_count = 22

        # patch classes
        self.executor_mock, self.executor_inst_mock = self.patchClass(
            worker.futures, "ThreadPoolExecutor", attach_as="executor"
        )
        self.server_mock, self.server_inst_mock = self.patchClass(worker.server, "Server")

        # other mocking
        self.threads_count_mock = self.patch("taskflow.engines.worker_based.worker.tu.get_optimal_thread_count")
        self.threads_count_mock.return_value = self.threads_count

    def setUp(self):
        super(TestWorker, self).setUp()
        self.task_cls = utils.DummyTask
        self.task_name = reflection.get_class_name(self.task_cls)
        self.broker_url = 'test-url'
        self.exchange = 'test-exchange'
        self.topic = 'test-topic'
        self.threads_count = 5
        self.endpoint_count = 21

        # patch classes
        self.executor_mock, self.executor_inst_mock = self._patch_class(
            worker.futures, 'ThreadPoolExecutor', attach_as='executor')
        self.server_mock, self.server_inst_mock = self._patch_class(
            worker.server, 'Server')

        # other mocking
        self.threads_count_mock = self._patch(
            'taskflow.engines.worker_based.worker.tu.get_optimal_thread_count')
        self.threads_count_mock.return_value = self.threads_count

 def _generate_banner(self):
     """Generates a banner that can be useful to display before running."""
     tpl_params = {}
     connection_details = self._server.connection_details
     transport = connection_details.transport
     if transport.driver_version:
         transport_driver = "%s v%s" % (transport.driver_name,
                                        transport.driver_version)
     else:
         transport_driver = transport.driver_name
     tpl_params['transport_driver'] = transport_driver
     tpl_params['exchange'] = self._exchange
     tpl_params['topic'] = self._topic
     tpl_params['transport_type'] = transport.driver_type
     tpl_params['connection_uri'] = connection_details.uri
     tpl_params['executor_type'] = reflection.get_class_name(self._executor)
     if self._threads_count != -1:
         tpl_params['executor_thread_count'] = self._threads_count
     if self._endpoints:
         pretty_endpoints = []
         for ep in self._endpoints:
             pretty_endpoints.append("  - %s" % ep)
         # This ensures there is a newline before the list...
         tpl_params['endpoints'] = "\n" + "\n".join(pretty_endpoints)
     try:
         tpl_params['hostname'] = socket.getfqdn()
     except socket.error:
         pass
     try:
         tpl_params['pid'] = os.getpid()
     except OSError:
         pass
     tpl_params['platform'] = platform.platform()
     tpl_params['thread_id'] = tu.get_ident()
     return BANNER_TEMPLATE.substitute(BANNER_TEMPLATE.defaults,
                                       **tpl_params)

def moved_class(new_class, old_class_name, old_module_name, message=None,
                version=None, removal_version=None):
    """Deprecates a class that was moved to another location.

    This will emit warnings when the old locations class is initialized,
    telling where the new and improved location for the old class now is.
    """
    old_name = ".".join((old_module_name, old_class_name))
    new_name = reflection.get_class_name(new_class)
    message_components = [
        "Class '%s' has moved to '%s'" % (old_name, new_name),
    ]
    if version:
        message_components.append(" in version '%s'" % version)
    if removal_version:
        if removal_version == "?":
            message_components.append(" and will be removed in a future"
                                      " version")
        else:
            message_components.append(" and will be removed in version '%s'"
                                      % removal_version)
    if message:
        message_components.append(": %s" % message)
    return MovedClassProxy(new_class, "".join(message_components), 3)

 def test_std_exception(self):
     name = reflection.get_class_name(RuntimeError)
     self.assertEqual(name, 'RuntimeError')

 def test_instance(self):
     name = reflection.get_class_name(Class())
     self.assertEqual(name, ".".join((__name__, "Class")))

 def test_class(self):
     name = reflection.get_class_name(Class)
     self.assertEquals(name, '.'.join((__name__, 'Class')))

 def __repr__(self):
     return '<%s %s>' % (reflection.get_class_name(self), self)