2017-02-27 01:54:48 -05:00
|
|
|
import unittest
|
|
|
|
|
|
|
|
from sqlalchemy.engine.url import make_url
|
|
|
|
|
2017-03-10 12:11:51 -05:00
|
|
|
from superset.models.core import Database
|
2017-02-27 01:54:48 -05:00
|
|
|
|
|
|
|
|
|
|
|
class DatabaseModelTestCase(unittest.TestCase):
|
2017-04-10 18:36:58 -04:00
|
|
|
|
|
|
|
def test_database_schema_presto(self):
|
2017-02-27 01:54:48 -05:00
|
|
|
sqlalchemy_uri = 'presto://presto.airbnb.io:8080/hive/default'
|
|
|
|
model = Database(sqlalchemy_uri=sqlalchemy_uri)
|
|
|
|
|
2017-04-10 18:36:58 -04:00
|
|
|
db = make_url(model.get_sqla_engine().url).database
|
|
|
|
self.assertEquals('hive/default', db)
|
|
|
|
|
|
|
|
db = make_url(model.get_sqla_engine(schema='core_db').url).database
|
|
|
|
self.assertEquals('hive/core_db', db)
|
|
|
|
|
|
|
|
sqlalchemy_uri = 'presto://presto.airbnb.io:8080/hive'
|
2017-02-27 01:54:48 -05:00
|
|
|
model = Database(sqlalchemy_uri=sqlalchemy_uri)
|
|
|
|
|
2017-04-10 18:36:58 -04:00
|
|
|
db = make_url(model.get_sqla_engine().url).database
|
|
|
|
self.assertEquals('hive', db)
|
|
|
|
|
|
|
|
db = make_url(model.get_sqla_engine(schema='core_db').url).database
|
|
|
|
self.assertEquals('hive/core_db', db)
|
|
|
|
|
|
|
|
def test_database_schema_postgres(self):
|
|
|
|
sqlalchemy_uri = 'postgresql+psycopg2://postgres.airbnb.io:5439/prod'
|
2017-02-27 01:54:48 -05:00
|
|
|
model = Database(sqlalchemy_uri=sqlalchemy_uri)
|
|
|
|
|
2017-04-10 18:36:58 -04:00
|
|
|
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 test_database_schema_hive(self):
|
2017-04-18 15:29:13 -04:00
|
|
|
sqlalchemy_uri = 'hive://hive@hive.airbnb.io:10000/default?auth=NOSASL'
|
2017-02-27 01:54:48 -05:00
|
|
|
model = Database(sqlalchemy_uri=sqlalchemy_uri)
|
2017-04-10 18:36:58 -04:00
|
|
|
db = make_url(model.get_sqla_engine().url).database
|
2017-04-18 15:29:13 -04:00
|
|
|
self.assertEquals('default', db)
|
2017-04-10 18:36:58 -04:00
|
|
|
|
|
|
|
db = make_url(model.get_sqla_engine(schema='core_db').url).database
|
2017-04-18 15:29:13 -04:00
|
|
|
self.assertEquals('core_db', db)
|
2017-02-27 01:54:48 -05:00
|
|
|
|
2017-04-10 18:36:58 -04:00
|
|
|
def test_database_schema_mysql(self):
|
|
|
|
sqlalchemy_uri = 'mysql://root@localhost/superset'
|
2017-02-27 01:54:48 -05:00
|
|
|
model = Database(sqlalchemy_uri=sqlalchemy_uri)
|
2017-04-10 18:36:58 -04:00
|
|
|
|
|
|
|
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)
|
2017-09-18 12:52:29 -04:00
|
|
|
|
|
|
|
def test_database_impersonate_user(self):
|
|
|
|
uri = 'mysql://root@localhost'
|
|
|
|
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)
|