QT/superset2
1
0
Fork 0
mirror of https://github.com/apache/superset.git synced 2026-04-18 15:44:57 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor: Moving get_user_datasources to security manager (#15467)

Browse Source 
Co-authored-by: John Bodley <john.bodley@airbnb.com>
This commit is contained in:
John Bodley
2021-06-30 09:51:11 -07:00
committed by GitHub
parent cad5ba828c
commit ffa51753e3
6 changed files with 140 additions and 121 deletions
Download Patch File Download Diff File Expand all files Collapse all files

79
tests/access_tests.py
View File

@@ -18,7 +18,6 @@
"""Unit tests for Superset"""
import json
import unittest
from collections import namedtuple
from unittest import mock
from tests.fixtures.birth_names_dashboard import load_birth_names_dashboard_with_slices
@@ -627,83 +626,5 @@ class TestRequestAccess(SupersetTestCase):
session.commit()
class TestDatasources(SupersetTestCase):
def test_get_user_datasources_admin(self):
Datasource = namedtuple("Datasource", ["database", "schema", "name"])
mock_session = mock.MagicMock()
mock_session.query.return_value.filter.return_value.all.return_value = []
with mock.patch("superset.security_manager") as mock_security_manager:
mock_security_manager.can_access_database.return_value = True
with mock.patch.object(
ConnectorRegistry, "get_all_datasources"
) as mock_get_all_datasources:
mock_get_all_datasources.return_value = [
Datasource("database1", "schema1", "table1"),
Datasource("database1", "schema1", "table2"),
Datasource("database2", None, "table1"),
]
datasources = ConnectorRegistry.get_user_datasources(mock_session)
assert sorted(datasources) == [
Datasource("database1", "schema1", "table1"),
Datasource("database1", "schema1", "table2"),
Datasource("database2", None, "table1"),
]
def test_get_user_datasources_gamma(self):
Datasource = namedtuple("Datasource", ["database", "schema", "name"])
mock_session = mock.MagicMock()
mock_session.query.return_value.filter.return_value.all.return_value = []
with mock.patch("superset.security_manager") as mock_security_manager:
mock_security_manager.can_access_database.return_value = False
with mock.patch.object(
ConnectorRegistry, "get_all_datasources"
) as mock_get_all_datasources:
mock_get_all_datasources.return_value = [
Datasource("database1", "schema1", "table1"),
Datasource("database1", "schema1", "table2"),
Datasource("database2", None, "table1"),
]
datasources = ConnectorRegistry.get_user_datasources(mock_session)
assert datasources == []
def test_get_user_datasources_gamma_with_schema(self):
Datasource = namedtuple("Datasource", ["database", "schema", "name"])
mock_session = mock.MagicMock()
mock_session.query.return_value.filter.return_value.all.return_value = [
Datasource("database1", "schema1", "table1"),
Datasource("database1", "schema1", "table2"),
]
with mock.patch("superset.security_manager") as mock_security_manager:
mock_security_manager.can_access_database.return_value = False
with mock.patch.object(
ConnectorRegistry, "get_all_datasources"
) as mock_get_all_datasources:
mock_get_all_datasources.return_value = [
Datasource("database1", "schema1", "table1"),
Datasource("database1", "schema1", "table2"),
Datasource("database2", None, "table1"),
]
datasources = ConnectorRegistry.get_user_datasources(mock_session)
assert sorted(datasources) == [
Datasource("database1", "schema1", "table1"),
Datasource("database1", "schema1", "table2"),
]
if __name__ == "__main__":
unittest.main()