def test_comparison(self):
        components = (
            "drivername",
            "username",
            "password",
            "host",
            "database",
            "query",
            "port",
        )

        common_url = (
            "dbtype://username:password"
            "@[2001:da8:2004:1000:202:116:160:90]:80/database?foo=bar"
        )
        other_url = "dbtype://uname:[email protected]/"

        url1 = url.make_url(common_url)
        url2 = url.make_url(common_url)
        url3 = url.make_url(other_url)

        is_true(url1 == url2)
        is_false(url1 != url2)
        is_true(url1 != url3)
        is_false(url1 == url3)

        for curr_component in components:
            setattr(url2, curr_component, "new_changed_value")
            is_true(url1 != url2)
            is_false(url1 == url2)
            setattr(url2, curr_component, getattr(url1, curr_component))

    def test_rfc1738_password(self):
        u = url.make_url("dbtype://user:pass word + other%[email protected]/dbname")
        eq_(u.password, "pass word + other:words")
        eq_(str(u), "dbtype://user:pass word + other%[email protected]/dbname")

        u = url.make_url(
            "dbtype://username:apples%[email protected]/database"
        )
        eq_(u.password, "apples/oranges")
        eq_(str(u), "dbtype://username:apples%[email protected]/database")

        u = url.make_url(
            "dbtype://username:apples%40oranges%40%[email protected]/database"
        )
        eq_(u.password, "[email protected]@@")
        eq_(
            str(u),
            "dbtype://username:apples%40oranges%40%[email protected]/database",
        )

        u = url.make_url("dbtype://username%40:@hostspec/database")
        eq_(u.password, "")
        eq_(u.username, "[email protected]")
        eq_(str(u), "dbtype://username%40:@hostspec/database")

        u = url.make_url("dbtype://username:pass%[email protected]/database")
        eq_(u.password, "pass/word")
        eq_(str(u), "dbtype://username:pass%[email protected]/database")

def _create_connection(db):
    if isinstance(db, str):
        url = make_url(connections.databases[db]['ENGINE'])
    elif isinstance(db, dict):
        url = make_url(db['ENGINE'])
    else:
        url = db

    db_engine = url.drivername.split('+')[0]
    database = url.database
    kwargs = {'autocommit': True}

    if db_engine == 'sqlite':
        return
    elif db_engine == 'postgresql':
        database = 'postgres'
    elif db_engine == 'mssql':
        kwargs['isolation_level'] = 'AUTOCOMMIT'
        database = 'master'
    elif db_engine == 'mysql':
        database = None
    elif db_engine == 'oracle':
        database = connections.databases[DEFAULT_DB_ALIAS].get('SYSTEM_DB', 'SYSTEM')

    url = URL(url.drivername, url.username, url.password, url.host, url.port, database, url.query)
    return create_engine(url).connect().execution_options(**kwargs)

    def __init__(self):

        try:

            super(SQLRepository, self).__init__()

            self.conf = cfg.CONF
            url = None
            if self.conf.mysql.database_name is not None:
                settings_db = (self.conf.mysql.username,
                               self.conf.mysql.password,
                               self.conf.mysql.hostname,
                               self.conf.mysql.database_name)
                url = make_url("mysql+pymysql://%s:%[email protected]%s/%s" % settings_db)
            else:
                if self.conf.database.url is not None:
                    url = make_url(self.conf.database.url)
                else:
                    database_conf = dict(self.conf.database)
                    if 'url' in database_conf:
                        del database_conf['url']
                    url = URL(**database_conf)

            from sqlalchemy import create_engine
            self._db_engine = create_engine(url, pool_recycle=3600)

            self.metadata = MetaData()

        except Exception as ex:
            LOG.exception(ex)
            raise exceptions.RepositoryException(ex)

    def test_pymssql_port_setting(self):
        dialect = pymssql.dialect()

        u = url.make_url("mssql+pymssql://scott:[email protected]/test")
        connection = dialect.create_connect_args(u)
        eq_(
            [
                [],
                {
                    "host": "somehost",
                    "password": "tiger",
                    "user": "scott",
                    "database": "test",
                },
            ],
            connection,
        )

        u = url.make_url("mssql+pymssql://scott:[email protected]:5000/test")
        connection = dialect.create_connect_args(u)
        eq_(
            [
                [],
                {
                    "host": "somehost:5000",
                    "password": "tiger",
                    "user": "scott",
                    "database": "test",
                },
            ],
            connection,
        )

    def configure(cls, connection=None, url=None, dialect_name=None, opts={}):
        """Create a new :class:`.MigrationContext`.

        This is a factory method usually called
        by :meth:`.EnvironmentContext.configure`.

        :param connection: a :class:`~sqlalchemy.engine.Connection`
         to use for SQL execution in "online" mode.  When present,
         is also used to determine the type of dialect in use.
        :param url: a string database url, or a
         :class:`sqlalchemy.engine.url.URL` object.
         The type of dialect to be used will be derived from this if
         ``connection`` is not passed.
        :param dialect_name: string name of a dialect, such as
         "postgresql", "mssql", etc.  The type of dialect to be used will be
         derived from this if ``connection`` and ``url`` are not passed.
        :param opts: dictionary of options.  Most other options
         accepted by :meth:`.EnvironmentContext.configure` are passed via
         this dictionary.

        """
        if connection:
            dialect = connection.dialect
        elif url:
            url = sqla_url.make_url(url)
            dialect = url.get_dialect()()
        elif dialect_name:
            url = sqla_url.make_url("%s://" % dialect_name)
            dialect = url.get_dialect()()
        else:
            raise Exception("Connection, url, or dialect_name is required.")

        return MigrationContext(dialect, connection, opts)

def all_to_intermediary(filename_or_input, schema=None):
    """ Dispatch the filename_or_input to the different function to produce the intermediary syntax.
    All the supported classes names are in `swich_input_class_to_method`.
    The input can also be a list of strings in markdown format or a filename finishing by '.er' containing markdown
    format.
    """
    # Try to convert from the name of the class
    input_class_name = filename_or_input.__class__.__name__
    try:
        this_to_intermediary = switch_input_class_to_method[input_class_name]
        tables, relationships = this_to_intermediary(filename_or_input)
        return tables, relationships
    except KeyError:
        pass

    # try to read markdown file.
    if isinstance(filename_or_input, basestring):
        if filename_or_input.split('.')[-1] == 'er':
            return markdown_file_to_intermediary(filename_or_input)

    # try to read a markdown in a string
    if not isinstance(filename_or_input, basestring):
        if all(isinstance(e, basestring) for e in filename_or_input):
            return line_iterator_to_intermediary(filename_or_input)

    # try to read DB URI.
    try:
        make_url(filename_or_input)
        return database_to_intermediary(filename_or_input, schema=schema)
    except ArgumentError:
        pass

    msg = 'Cannot process filename_or_input {}'.format(input_class_name)
    raise ValueError(msg)

def normalize_url(raw):
    """
    Build a URL from a string according to obvious rules.

    Converts *.sqlite paths to sqlite:/// URLs, etc.

    >>> str(normalize_url("/tmp/foo.sqlite"))
    'sqlite:////tmp/foo.sqlite'
    >>> str(normalize_url("postgresql://[email protected]/database"))
    'postgresql://[email protected]/database'
    """

    from sqlalchemy.engine.url import make_url
    from sqlalchemy.exceptions import ArgumentError

    try:
        return make_url(raw)
    except ArgumentError:
        from os.path import (
            abspath,
            splitext,
            )

        (root, extension) = splitext(raw)

        if extension == ".sqlite":
            return make_url("sqlite:///%s" % abspath(raw))
        else:
            raise ArgumentError("could not parse supposed URL \"%s\"" % raw)

    def test_database_schema_hive(self):
        sqlalchemy_uri = 'hive://[email protected]:10000/default?auth=NOSASL'
        model = Database(sqlalchemy_uri=sqlalchemy_uri)
        db = make_url(model.get_sqla_engine().url).database
        self.assertEquals('default', db)

        db = make_url(model.get_sqla_engine(schema='core_db').url).database
        self.assertEquals('core_db', db)

    def test_database_schema_mysql(self):
        sqlalchemy_uri = 'mysql://[email protected]/superset'
        model = Database(sqlalchemy_uri=sqlalchemy_uri)

        db = make_url(model.get_sqla_engine().url).database
        self.assertEquals('superset', db)

        db = make_url(model.get_sqla_engine(schema='staging').url).database
        self.assertEquals('staging', db)

    def test_database_schema_postgres(self):
        sqlalchemy_uri = 'postgresql+psycopg2://postgres.airbnb.io:5439/prod'
        model = Database(sqlalchemy_uri=sqlalchemy_uri)

        db = make_url(model.get_sqla_engine().url).database
        self.assertEquals('prod', db)

        db = make_url(model.get_sqla_engine(schema='foo').url).database
        self.assertEquals('prod', db)

def get_absolute_uri_for_session(session):
    engine = session.get_bind()
    url_ = engine.url
    if url_.drivername.startswith('sqlite'):
        database_path = url_.database
        abs_database_path = os.path.abspath(database_path)
        abs_database_path = '/'.join(abs_database_path.split(os.sep))
        return make_url("sqlite:///%s" % abs_database_path)
    else:
        return make_url(url_)

    def test_plugin_multiple_url_registration(self):
        from sqlalchemy.dialects import sqlite

        global MyEnginePlugin1
        global MyEnginePlugin2

        def side_effect_1(url, kw):
            eq_(kw, {"logging_name": "foob"})
            kw["logging_name"] = "bar"
            url.query.pop("myplugin1_arg", None)
            return MyEnginePlugin1

        def side_effect_2(url, kw):
            url.query.pop("myplugin2_arg", None)
            return MyEnginePlugin2

        MyEnginePlugin1 = Mock(side_effect=side_effect_1)
        MyEnginePlugin2 = Mock(side_effect=side_effect_2)

        plugins.register("engineplugin1", __name__, "MyEnginePlugin1")
        plugins.register("engineplugin2", __name__, "MyEnginePlugin2")

        e = create_engine(
            "sqlite:///?plugin=engineplugin1&foo=bar&myplugin1_arg=bat"
            "&plugin=engineplugin2&myplugin2_arg=hoho",
            logging_name="foob",
        )
        eq_(e.dialect.name, "sqlite")
        eq_(e.logging_name, "bar")

        # plugin args are removed from URL.
        eq_(e.url.query, {"foo": "bar"})
        assert isinstance(e.dialect, sqlite.dialect)

        eq_(
            MyEnginePlugin1.mock_calls,
            [
                call(url.make_url("sqlite:///?foo=bar"), {}),
                call.handle_dialect_kwargs(sqlite.dialect, mock.ANY),
                call.handle_pool_kwargs(mock.ANY, {"dialect": e.dialect}),
                call.engine_created(e),
            ],
        )

        eq_(
            MyEnginePlugin2.mock_calls,
            [
                call(url.make_url("sqlite:///?foo=bar"), {}),
                call.handle_dialect_kwargs(sqlite.dialect, mock.ANY),
                call.handle_pool_kwargs(mock.ANY, {"dialect": e.dialect}),
                call.engine_created(e),
            ],
        )

 def make_connection_manager(self, path):
     if isinstance(path, ConnectionManager):
         return path
     try:
         path = str(path)
         if path == "":
             return SQLiteConnectionManager("sqlite://")
         # Ensure that the path is in fact a URI
         make_url(path)
         return ConnectionManager(path)
     except:
         return self.connection_manager_type(path)

    def test_database_impersonate_user(self):
        uri = 'mysql://[email protected]'
        example_user = 'giuseppe'
        model = Database(sqlalchemy_uri=uri)

        model.impersonate_user = True
        user_name = make_url(model.get_sqla_engine(user_name=example_user).url).username
        self.assertEquals(example_user, user_name)

        model.impersonate_user = False
        user_name = make_url(model.get_sqla_engine(user_name=example_user).url).username
        self.assertNotEquals(example_user, user_name)

def _assert_db_exists(engine):
    db_url = make_url(engine.url)
    db_name = db_url.database
    try:
        logger.debug('try to connect to database "%s": %s', db_name, db_url)
        connection = engine.connect()
        connection.close()
    except sqlalchemy.exc.OperationalError as err:
        db_url = make_url(engine.url)
        db_name = db_url.database
        logger.error('could not connect to database "%s": %s', db_name, str(err))
        raise ValueError('Cannot connect to database "%s".' % db_name)

    def test_query_string(self):
        u = url.make_url("dialect://user:[email protected]/db?arg1=param1&arg2=param2")
        eq_(u.query, {"arg1": "param1", "arg2": "param2"})
        eq_(str(u), "dialect://user:[email protected]/db?arg1=param1&arg2=param2")

        u = url.make_url(
            "dialect://user:[email protected]/db?arg1=param1&arg2=param2&arg2=param3"
        )
        eq_(u.query, {"arg1": "param1", "arg2": ["param2", "param3"]})
        eq_(
            str(u),
            "dialect://user:[email protected]/db?arg1=param1&arg2=param2&arg2=param3",
        )

 def _url_from_target(self, target):
     if isinstance(target, Connectable):
         return target.engine.url
     elif isinstance(target, six.string_types):
         if "://" not in target:
             target_url = sa_url.make_url("%s://" % target)
         else:
             target_url = sa_url.make_url(target)
         return target_url
     elif isinstance(target, sa_url.URL):
         return target
     else:
         raise ValueError("Invalid target type: %r" % target)

def check_db_url(db_url):
    """Check for the correct format of the DB URL.  Return none if check
    passed.  Raise exception otherwise.
    :param db_url: The DB URL to be checked
    :raise exc.ArPrDataBadDBURLFormatException:
    :rtype : None
    """
    try:
        make_url(db_url)
    except Exception:
        raise exc.ArPrDataBadDBURLFormatException(
            "Incorrect DB URL Format.  Check http://docs.sqlalchemy.org"
            "/en/latest/core/engines.html?highlight=url#database-urls")

    def __init__(self, url):
        self.table = Table('__tablename__', MetaData(),
                           Column('taskid', String(64), primary_key=True, nullable=False),
                           Column('project', String(64)),
                           Column('url', String(1024)),
                           Column('status', Integer),
                           Column('schedule', LargeBinary),
                           Column('fetch', LargeBinary),
                           Column('process', LargeBinary),
                           Column('track', LargeBinary),
                           Column('lastcrawltime', Float(32)),
                           Column('updatetime', Float(32)),
                           mysql_engine='InnoDB',
                           mysql_charset='utf8'
                           )

        self.url = make_url(url)
        if self.url.database:
            database = self.url.database
            self.url.database = None
            try:
                engine = create_engine(self.url, convert_unicode=True)
                engine.execute("CREATE DATABASE IF NOT EXISTS %s" % database)
            except sqlalchemy.exc.SQLAlchemyError:
                pass
            self.url.database = database
        self.engine = create_engine(self.url, convert_unicode=True)

        self._list_project()