QT/superset2
1
0
Fork 0
mirror of https://github.com/apache/superset.git synced 2026-04-20 08:34:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(file uploads): List only allowed schemas in the file uploads dialog (#32702)

Browse Source
This commit is contained in:
Vitor Avila
2025-03-17 23:36:16 -03:00
committed by GitHub
parent 4adf44a43c
commit e35145c816
5 changed files with 110 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

87
tests/integration_tests/databases/api_tests.py
View File

@@ -2088,6 +2088,93 @@ class TestDatabaseApi(SupersetTestCase):
)
assert rv.status_code == 400
def test_database_schemas_upload_allowed_filter(self):
"""
Database API: Test database schemas when filtering for upload allowed
and there is not schema restriction
"""
with self.create_app().app_context():
example_db = get_example_database()
extra = {
"metadata_params": {},
"engine_params": {},
"metadata_cache_timeout": {},
"schemas_allowed_for_file_upload": [],
}
self.login(ADMIN_USERNAME)
database = self.insert_database(
"database_with_upload",
example_db.sqlalchemy_uri_decrypted,
extra=json.dumps(extra),
allow_file_upload=True,
)
db.session.commit()
yield database
mock_schemas = ["schema_1", "schema_2", "schema_3"]
mock.patch.object(
database, "get_all_schema_names", return_value=mock_schemas
)
arguments = {"upload_allowed": True}
uri = f"api/v1/database/{database.id}/schemas/?q={prison.dumps(arguments)}"
rv = self.client.get(uri)
data = json.loads(rv.data.decode("utf-8"))
assert data["result"] == mock_schemas
db.session.delete(database)
db.session.commit()
def test_database_schemas_upload_allowed_filter_specific_schemas(self):
"""
Database API: Test database schemas when filtering for upload allowed
with an schema restriction set
"""
with self.create_app().app_context():
example_db = get_example_database()
extra = {
"metadata_params": {},
"engine_params": {},
"metadata_cache_timeout": {},
"schemas_allowed_for_file_upload": ["schema_2"],
}
self.login(ADMIN_USERNAME)
database = self.insert_database(
"database_with_upload",
example_db.sqlalchemy_uri_decrypted,
extra=json.dumps(extra),
allow_file_upload=True,
)
db.session.commit()
yield database
mock.patch.object(
database,
"get_all_schema_names",
return_value=["schema_1", "schema_2", "schema_3"],
)
arguments = {"upload_allowed": True}
uri = f"api/v1/database/{database.id}/schemas/?q={prison.dumps(arguments)}"
rv = self.client.get(uri)
data = json.loads(rv.data.decode("utf-8"))
assert data["result"] == ["schema_2"]
db.session.delete(database)
db.session.commit()
def test_database_schemas_upload_allowed_filter_disabled(self):
"""
Database API: Test database schemas when filtering for upload allowed
for a DB connection that has file uploads disabled
"""
database = db.session.query(Database).filter_by(database_name="examples").one()
self.login(ADMIN_USERNAME)
arguments = {"upload_allowed": True}
uri = f"api/v1/database/{database.id}/schemas/?q={prison.dumps(arguments)}"
rv = self.client.get(uri)
assert rv.status_code == 200
data = json.loads(rv.data.decode("utf-8"))
assert data["result"] == []
def test_database_tables(self):
"""
Database API: Test database tables