iop
/
superset
4
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix(databases): test connection api endpoint (#10824) 

* fix test connection with extra

* fix lint and allow_none server_cert

* update test connection tests
This commit is contained in:
Lily Kuang 2020-09-10 13:49:14 -07:00 committed by GitHub
parent ac2937a6c5
commit f0545bfe50
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 23 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
superset/databases/commands/test_connection.py
View File

@ -18,7 +18,6 @@ import logging
from contextlib import closing
from typing import Any, Dict, Optional
import simplejson as json
from flask_appbuilder.security.sqla.models import User
from sqlalchemy import select
@ -46,9 +45,9 @@ class TestConnectionDatabaseCommand(BaseCommand):
database = DatabaseDAO.build_db_for_connection_test(
server_cert=self._properties.get("server_cert", ""),
extra=json.dumps(self._properties.get("extra", {})),
extra=self._properties.get("extra", "{}"),
impersonate_user=self._properties.get("impersonate_user", False),
encrypted_extra=json.dumps(self._properties.get("encrypted_extra", {})),
encrypted_extra=self._properties.get("encrypted_extra", "{}"),
)
if database is not None:
database.set_sqlalchemy_uri(uri)

8
superset/databases/schemas.py
View File

@ -298,10 +298,14 @@ class DatabaseTestConnectionSchema(Schema):
impersonate_user = fields.Boolean(description=impersonate_user_description)
extra = fields.String(description=extra_description, validate=extra_validator)
encrypted_extra = fields.String(
description=encrypted_extra_description, validate=encrypted_extra_validator
description=encrypted_extra_description,
validate=encrypted_extra_validator,
allow_none=True,
)
server_cert = fields.String(
description=server_cert_description, validate=server_cert_validator
description=server_cert_description,
allow_none=True,
validate=server_cert_validator,
)
sqlalchemy_uri = fields.String(
description=sqlalchemy_uri_description,

16
tests/databases/api_tests.py
View File

@ -656,15 +656,24 @@ class TestDatabaseApi(SupersetTestCase):
"""
Database API: Test test connection
"""
extra = {
"metadata_params": {},
"engine_params": {},
"metadata_cache_timeout": {},
"schemas_allowed_for_csv_upload": [],
}
# need to temporarily allow sqlite dbs, teardown will undo this
app.config["PREVENT_UNSAFE_DB_CONNECTIONS"] = False
self.login("admin")
example_db = get_example_database()
# validate that the endpoint works with the password-masked sqlalchemy uri
data = {
"sqlalchemy_uri": example_db.safe_sqlalchemy_uri(),
"database_name": "examples",
"encrypted_extra": "{}",
"extra": json.dumps(extra),
"impersonate_user": False,
"sqlalchemy_uri": example_db.safe_sqlalchemy_uri(),
"server_cert": ssl_certificate,
}
url = f"api/v1/database/test_connection"
rv = self.post_assert_metric(url, data, "test_connection")
@ -676,6 +685,8 @@ class TestDatabaseApi(SupersetTestCase):
"sqlalchemy_uri": example_db.sqlalchemy_uri_decrypted,
"database_name": "examples",
"impersonate_user": False,
"extra": json.dumps(extra),
"server_cert": None,
}
rv = self.post_assert_metric(url, data, "test_connection")
self.assertEqual(rv.status_code, 200)
@ -691,6 +702,7 @@ class TestDatabaseApi(SupersetTestCase):
"sqlalchemy_uri": "broken://url",
"database_name": "examples",
"impersonate_user": False,
"server_cert": None,
}
url = f"api/v1/database/test_connection"
rv = self.post_assert_metric(url, data, "test_connection")
@ -707,6 +719,7 @@ class TestDatabaseApi(SupersetTestCase):
"sqlalchemy_uri": "mssql+pymssql://url",
"database_name": "examples",
"impersonate_user": False,
"server_cert": None,
}
rv = self.post_assert_metric(url, data, "test_connection")
self.assertEqual(rv.status_code, 400)
@ -729,6 +742,7 @@ class TestDatabaseApi(SupersetTestCase):
"sqlalchemy_uri": "sqlite:///home/superset/unsafe.db",
"database_name": "unsafe",
"impersonate_user": False,
"server_cert": None,
}
url = f"api/v1/database/test_connection"
rv = self.post_assert_metric(url, data, "test_connection")