def test_merge_user_password(self):
     url = "http://josh:[email protected]/"
     parsed = misc.parse_uri(url)
     joined = misc.merge_uri(parsed, {})
     self.assertEqual('www.yahoo.com', joined.get('hostname'))
     self.assertEqual('josh', joined.get('username'))
     self.assertEqual('harlow', joined.get('password'))

 def test_merge(self):
     url = "http://www.yahoo.com/?a=b&c=d"
     parsed = misc.parse_uri(url)
     joined = misc.merge_uri(parsed, {})
     self.assertEqual('b', joined.get('a'))
     self.assertEqual('d', joined.get('c'))
     self.assertEqual('www.yahoo.com', joined.get('hostname'))

 def test_port_provided(self):
     url = "rabbitmq://www.yahoo.com:5672"
     parsed = misc.parse_uri(url)
     self.assertEqual('rabbitmq', parsed.scheme)
     self.assertEqual('www.yahoo.com', parsed.hostname)
     self.assertEqual(5672, parsed.port)
     self.assertEqual('', parsed.path)

def _extract_engine(**kwargs):
    """Extracts the engine kind and any associated options."""
    options = {}
    kind = kwargs.pop('engine', None)
    engine_conf = kwargs.pop('engine_conf', None)
    if engine_conf is not None:
        warnings.warn("Using the 'engine_conf' argument is"
                      " deprecated and will be removed in a future version,"
                      " please use the 'engine' argument instead.",
                      DeprecationWarning)
        if isinstance(engine_conf, six.string_types):
            kind = engine_conf
        else:
            options.update(engine_conf)
            kind = options.pop('engine', None)
    if not kind:
        kind = ENGINE_DEFAULT
    # See if it's a URI and if so, extract any further options...
    try:
        uri = misc.parse_uri(kind)
    except (TypeError, ValueError):
        pass
    else:
        kind = uri.scheme
        options = misc.merge_uri(uri, options.copy())
    # Merge in any leftover **kwargs into the options, this makes it so that
    # the provided **kwargs override any URI or engine_conf specific options.
    options.update(kwargs)
    return (kind, options)

 def _compat_extract(**kwargs):
     options = {}
     kind = kwargs.pop('engine', None)
     engine_conf = kwargs.pop('engine_conf', None)
     if engine_conf is not None:
         if isinstance(engine_conf, six.string_types):
             kind = engine_conf
         else:
             options.update(engine_conf)
             kind = options.pop('engine', None)
     if not kind:
         kind = ENGINE_DEFAULT
     # See if it's a URI and if so, extract any further options...
     try:
         uri = misc.parse_uri(kind)
     except (TypeError, ValueError):
         pass
     else:
         kind = uri.scheme
         options = misc.merge_uri(uri, options.copy())
     # Merge in any leftover **kwargs into the options, this makes it so
     # that the provided **kwargs override any URI or engine_conf specific
     # options.
     options.update(kwargs)
     return (kind, options)

 def test_parse(self):
     url = "zookeeper://192.168.0.1:2181/a/b/?c=d"
     parsed = misc.parse_uri(url)
     self.assertEqual('zookeeper', parsed.scheme)
     self.assertEqual(2181, parsed.port)
     self.assertEqual('192.168.0.1', parsed.hostname)
     self.assertEqual('', parsed.fragment)
     self.assertEqual('/a/b/', parsed.path)
     self.assertEqual({'c': 'd'}, parsed.params)

 def test_merge_user_password_existing(self):
     url = "http://josh:[email protected]/"
     parsed = misc.parse_uri(url)
     existing = {
         'username': 'joe',
         'password': 'biggie',
     }
     joined = misc.merge_uri(parsed, existing)
     self.assertEqual('www.yahoo.com', joined.get('hostname'))
     self.assertEqual('joe', joined.get('username'))
     self.assertEqual('biggie', joined.get('password'))

def fetch(conf, namespace=BACKEND_NAMESPACE, **kwargs):
    """Fetch a persistence backend with the given configuration.

    This fetch method will look for the entrypoint name in the entrypoint
    namespace, and then attempt to instantiate that entrypoint using the
    provided configuration and any persistence backend specific kwargs.

    NOTE(harlowja): to aid in making it easy to specify configuration and
    options to a backend the configuration (which is typical just a dictionary)
    can also be a URI string that identifies the entrypoint name and any
    configuration specific to that backend.

    For example, given the following configuration URI::

        mysql://<not-used>/?a=b&c=d

    This will look for the entrypoint named 'mysql' and will provide
    a configuration object composed of the URI's components, in this case that
    is ``{'a': 'b', 'c': 'd'}`` to the constructor of that persistence backend
    instance.
    """
    if isinstance(conf, six.string_types):
        conf = {'connection': conf}
    backend_name = conf['connection']
    try:
        uri = misc.parse_uri(backend_name)
    except (TypeError, ValueError):
        pass
    else:
        backend_name = uri.scheme
        conf = misc.merge_uri(uri, conf.copy())
    # If the backend is like 'mysql+pymysql://...' which informs the
    # backend to use a dialect (supported by sqlalchemy at least) we just want
    # to look at the first component to find our entrypoint backend name...
    if backend_name.find("+") != -1:
        backend_name = backend_name.split("+", 1)[0]
    LOG.debug('Looking for %r backend driver in %r', backend_name, namespace)
    try:
        mgr = driver.DriverManager(namespace, backend_name,
                                   invoke_on_load=True,
                                   invoke_args=(conf,),
                                   invoke_kwds=kwargs)
        return mgr.driver
    except RuntimeError as e:
        raise exc.NotFound("Could not find backend %s: %s" % (backend_name, e))

def _extract_engine(engine, **kwargs):
    """Extracts the engine kind and any associated options."""
    kind = engine
    if not kind:
        kind = ENGINE_DEFAULT

    # See if it's a URI and if so, extract any further options...
    options = {}
    try:
        uri = misc.parse_uri(kind)
    except (TypeError, ValueError):
        pass
    else:
        kind = uri.scheme
        options = misc.merge_uri(uri, options.copy())

    # Merge in any leftover **kwargs into the options, this makes it so
    # that the provided **kwargs override any URI/engine specific
    # options.
    options.update(kwargs)
    return (kind, options)

def fetch(conf, namespace=BACKEND_NAMESPACE, **kwargs):
    """Fetches a given backend using the given configuration (and any backend
    specific kwargs) in the given entrypoint namespace.
    """
    backend_name = conf['connection']
    try:
        pieces = misc.parse_uri(backend_name)
    except (TypeError, ValueError):
        pass
    else:
        backend_name = pieces['scheme']
        conf = misc.merge_uri(pieces, conf.copy())
    LOG.debug('Looking for %r backend driver in %r', backend_name, namespace)
    try:
        mgr = driver.DriverManager(namespace, backend_name,
                                   invoke_on_load=True,
                                   invoke_args=(conf,),
                                   invoke_kwds=kwargs)
        return mgr.driver
    except RuntimeError as e:
        raise exc.NotFound("Could not find backend %s: %s" % (backend_name, e))

def fetch(name, conf, namespace=BACKEND_NAMESPACE, **kwargs):
    """Fetch a jobboard backend with the given configuration.

    This fetch method will look for the entrypoint name in the entrypoint
    namespace, and then attempt to instantiate that entrypoint using the
    provided name, configuration and any board specific kwargs.

    NOTE(harlowja): to aid in making it easy to specify configuration and
    options to a board the configuration (which is typical just a dictionary)
    can also be a URI string that identifies the entrypoint name and any
    configuration specific to that board.

    For example, given the following configuration URI::

        zookeeper://<not-used>/?a=b&c=d

    This will look for the entrypoint named 'zookeeper' and will provide
    a configuration object composed of the URI's components, in this case that
    is ``{'a': 'b', 'c': 'd'}`` to the constructor of that board
    instance (also including the name specified).
    """
    if isinstance(conf, six.string_types):
        conf = {'board': conf}
    board = conf['board']
    try:
        uri = misc.parse_uri(board)
    except (TypeError, ValueError):
        pass
    else:
        board = uri.scheme
        conf = misc.merge_uri(uri, conf.copy())
    LOG.debug('Looking for %r jobboard driver in %r', board, namespace)
    try:
        mgr = driver.DriverManager(namespace, board,
                                   invoke_on_load=True,
                                   invoke_args=(name, conf),
                                   invoke_kwds=kwargs)
        return mgr.driver
    except RuntimeError as e:
        raise exc.NotFound("Could not find jobboard %s" % (board), e)

def fetch(conf, namespace=BACKEND_NAMESPACE, **kwargs):
    """Fetch a persistence backend with the given configuration.

    This fetch method will look for the entrypoint name in the entrypoint
    namespace, and then attempt to instantiate that entrypoint using the
    provided configuration and any persistence backend specific kwargs.

    NOTE(harlowja): to aid in making it easy to specify configuration and
    options to a backend the configuration (which is typical just a dictionary)
    can also be a uri string that identifies the entrypoint name and any
    configuration specific to that backend.

    For example, given the following configuration uri:

    mysql://<not-used>/?a=b&c=d

    This will look for the entrypoint named 'mysql' and will provide
    a configuration object composed of the uris parameters, in this case that
    is {'a': 'b', 'c': 'd'} to the constructor of that persistence backend
    instance.
    """
    backend_name = conf['connection']
    try:
        uri = misc.parse_uri(backend_name)
    except (TypeError, ValueError):
        pass
    else:
        backend_name = uri.scheme
        conf = misc.merge_uri(uri, conf.copy())
    LOG.debug('Looking for %r backend driver in %r', backend_name, namespace)
    try:
        mgr = driver.DriverManager(namespace, backend_name,
                                   invoke_on_load=True,
                                   invoke_args=(conf,),
                                   invoke_kwds=kwargs)
        return mgr.driver
    except RuntimeError as e:
        raise exc.NotFound("Could not find backend %s: %s" % (backend_name, e))

def fetch(name, conf, namespace=BACKEND_NAMESPACE, **kwargs):
    """Fetch a jobboard backend with the given configuration (and any board
    specific kwargs) in the given entrypoint namespace and create it with the
    given name.
    """
    if isinstance(conf, six.string_types):
        conf = {'board': conf}
    board = conf['board']
    try:
        pieces = misc.parse_uri(board)
    except (TypeError, ValueError):
        pass
    else:
        board = pieces['scheme']
        conf = misc.merge_uri(pieces, conf.copy())
    LOG.debug('Looking for %r jobboard driver in %r', board, namespace)
    try:
        mgr = driver.DriverManager(namespace, board,
                                   invoke_on_load=True,
                                   invoke_args=(name, conf),
                                   invoke_kwds=kwargs)
        return mgr.driver
    except RuntimeError as e:
        raise exc.NotFound("Could not find jobboard %s" % (board), e)

 def test_ipv6_host(self):
     url = "rsync://[2001:db8:0:1]:873"
     parsed = misc.parse_uri(url)
     self.assertEqual('rsync', parsed.scheme)
     self.assertEqual('2001:db8:0:1', parsed.hostname)
     self.assertEqual(873, parsed.port)

 def test_multi_params(self):
     url = "mysql://www.yahoo.com:3306/a/b/?c=d&c=e"
     parsed = misc.parse_uri(url, query_duplicates=True)
     self.assertEqual({'c': ['d', 'e']}, parsed.params)

 def test_merge_existing_hostname(self):
     url = "http://www.yahoo.com/"
     parsed = misc.parse_uri(url)
     joined = misc.merge_uri(parsed, {'hostname': 'b.com'})
     self.assertEqual('b.com', joined.get('hostname'))

 def test_user_password(self):
     url = "rsync://test:[email protected]:873"
     parsed = misc.parse_uri(url)
     self.assertEqual('test', parsed.username)
     self.assertEqual('test_pw', parsed.password)
     self.assertEqual('www.yahoo.com', parsed.hostname)

 def test_user(self):
     url = "rsync://[email protected]:873"
     parsed = misc.parse_uri(url)
     self.assertEqual('test', parsed.username)
     self.assertEqual(None, parsed.password)

def load(flow, store=None, flow_detail=None, book=None,
         engine_conf=None, backend=None, namespace=ENGINES_NAMESPACE,
         **kwargs):
    """Load a flow into an engine.

    This function creates and prepares engine to run the
    flow. All that is left is to run the engine with 'run()' method.

    Which engine to load is specified in 'engine_conf' parameter. It
    can be a string that names engine type or a dictionary which holds
    engine type (with 'engine' key) and additional engine-specific
    configuration.

    Which storage backend to use is defined by backend parameter. It
    can be backend itself, or a dictionary that is passed to
    taskflow.persistence.backends.fetch to obtain backend.

    :param flow: flow to load
    :param store: dict -- data to put to storage to satisfy flow requirements
    :param flow_detail: FlowDetail that holds the state of the flow (if one is
        not provided then one will be created for you in the provided backend)
    :param book: LogBook to create flow detail in if flow_detail is None
    :param engine_conf: engine type and configuration configuration
    :param backend: storage backend to use or configuration
    :param namespace: driver namespace for stevedore (default is fine
       if you don't know what is it)
    :returns: engine
    """

    if engine_conf is None:
        engine_conf = {'engine': 'default'}

    # NOTE(imelnikov): this allows simpler syntax.
    if isinstance(engine_conf, six.string_types):
        engine_conf = {'engine': engine_conf}

    engine_name = engine_conf['engine']
    try:
        pieces = misc.parse_uri(engine_name)
    except (TypeError, ValueError):
        pass
    else:
        engine_name = pieces['scheme']
        engine_conf = misc.merge_uri(pieces, engine_conf.copy())

    if isinstance(backend, dict):
        backend = p_backends.fetch(backend)

    if flow_detail is None:
        flow_detail = p_utils.create_flow_detail(flow, book=book,
                                                 backend=backend)

    try:
        mgr = stevedore.driver.DriverManager(
            namespace, engine_name,
            invoke_on_load=True,
            invoke_args=(flow, flow_detail, backend, engine_conf),
            invoke_kwds=kwargs)
        engine = mgr.driver
    except RuntimeError as e:
        raise exc.NotFound("Could not find engine %s" % (engine_name), e)
    else:
        if store:
            engine.storage.inject(store)
        return engine