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

feat: Enable drilling in embedded (#34319)

Browse Source
This commit is contained in:
Vitor Avila
2025-08-05 02:23:00 -03:00
committed by GitHub
parent 791ea9860d
commit 49689eec6c
29 changed files with 1510 additions and 336 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
tests/integration_tests/base_tests.py
View File

@@ -160,12 +160,13 @@ class SupersetTestCase(TestCase):
return user_to_create
@contextmanager
def temporary_user(
def temporary_user( # noqa: C901
self,
clone_user=None,
username=None,
extra_roles=None,
extra_pvms=None,
pvms_to_remove=None,
login=False,
):
"""
@@ -180,33 +181,39 @@ class SupersetTestCase(TestCase):
temp_user = ab_models.User(
username=username, email=f"{username}@temp.com", active=True
)
pvms = []
if clone_user:
temp_user.roles = clone_user.roles
temp_user.first_name = clone_user.first_name
temp_user.last_name = clone_user.last_name
temp_user.password = clone_user.password
if clone_user.roles:
for role in clone_user.roles:
pvms.extend(role.permissions)
else:
temp_user.first_name = temp_user.last_name = username
if clone_user:
temp_user.roles = clone_user.roles
if extra_roles:
temp_user.roles.extend(extra_roles)
for role in extra_roles:
pvms.extend(role.permissions)
pvms = []
temp_role = None
if extra_pvms:
temp_role = ab_models.Role(name=f"tmp_role_{shortid()}")
for pvm in extra_pvms:
if isinstance(pvm, (tuple, list)):
pvms.append(security_manager.find_permission_view_menu(*pvm))
else:
pvms.append(pvm)
temp_role.permissions = pvms
temp_user.roles.append(temp_role)
db.session.add(temp_role)
db.session.commit()
for pvm in extra_pvms or []:
if isinstance(pvm, (tuple, list)):
pvms.append(security_manager.find_permission_view_menu(*pvm))
else:
pvms.append(pvm)
for pvm in pvms_to_remove or []:
if isinstance(pvm, (tuple, list)):
pvm = security_manager.find_permission_view_menu(*pvm)
if pvm in pvms:
pvms.remove(pvm)
temp_role = ab_models.Role(name=f"tmp_role_{shortid()}")
temp_role.permissions = pvms
temp_user.roles.append(temp_role)
db.session.add(temp_role)
db.session.commit()
# Add the temp user to the session and commit to apply changes for the test
db.session.add(temp_user)

94
tests/integration_tests/charts/data/api_tests.py
View File

@@ -18,6 +18,7 @@
import copy
import time
import unittest
from contextlib import contextmanager
from datetime import datetime
from io import BytesIO
from typing import Any, Optional
@@ -135,6 +136,32 @@ class BaseTestChartDataApi(SupersetTestCase):
)
return name
@contextmanager
def set_column_groupby_false(self, column_name: str):
"""
Context manager to temporarily set a column's groupby property to false.
"""
birth_names_table = self.get_birth_names_dataset()
target_column = None
original_groupby_value = None
for col in birth_names_table.columns:
if col.column_name == column_name:
target_column = col
original_groupby_value = col.groupby
break
if target_column:
target_column.groupby = False
db.session.commit()
try:
yield target_column
finally:
if target_column and original_groupby_value is not False:
target_column.groupby = original_groupby_value
db.session.commit()
@pytest.mark.chart_data_flow
class TestPostChartDataApi(BaseTestChartDataApi):
@@ -972,6 +999,73 @@ class TestPostChartDataApi(BaseTestChartDataApi):
assert "name" in result["query"]
assert list(result["data"][0].keys()) == ["name", "num divide by 10"]
@pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
def test_drill_by_allowed_column(self):
"""
Chart data API: Test that user can drill by column with
isDimension set to True
"""
request_payload = self.query_context_payload
request_payload["queries"][0]["columns"] = ["name"]
rv = self.post_assert_metric(CHART_DATA_URI, request_payload, "data")
assert rv.status_code == 200
@pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
def test_drill_by_disallowed_column_regular_user(self):
"""
Chart data API: Test that user can still drill by column with
isDimension set to False (given the dataset access)
"""
with self.set_column_groupby_false("num_girls"):
self.query_context_payload["queries"][0]["columns"] = ["num_girls"]
rv = self.post_assert_metric(
CHART_DATA_URI, self.query_context_payload, "data"
)
assert rv.status_code == 200
@pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
@mock.patch("superset.security.manager.SupersetSecurityManager.has_guest_access")
@mock.patch("superset.security.manager.SupersetSecurityManager.is_guest_user")
@with_feature_flags(EMBEDDED_SUPERSET=True)
def test_embedded_user_drill_by_allowed_column(
self, mock_is_guest_user, mock_has_guest_access
):
"""
Chart data API: Test that embedded user can drill by column with
isDimension set to True.
"""
g.user.rls = []
mock_has_guest_access.return_value = True
mock_is_guest_user.return_value = True
request_payload = self.query_context_payload
request_payload["queries"][0]["columns"] = ["name"]
rv = self.post_assert_metric(CHART_DATA_URI, request_payload, "data")
assert rv.status_code == 200
@pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
@mock.patch("superset.security.manager.SupersetSecurityManager.has_guest_access")
@mock.patch("superset.security.manager.SupersetSecurityManager.is_guest_user")
@with_feature_flags(EMBEDDED_SUPERSET=True)
def test_embedded_user_drill_by_disallowed_column(
self, mock_is_guest_user, mock_has_guest_access
):
"""
Chart data API: Test that embedded user can't drill by column with
isDimension set to False.
"""
self.logout()
self.login(GAMMA_USERNAME)
with self.set_column_groupby_false("num_girls"):
g.user.rls = []
mock_has_guest_access.return_value = True
mock_is_guest_user.return_value = True
self.query_context_payload["queries"][0]["columns"] = ["num_girls"]
rv = self.post_assert_metric(
CHART_DATA_URI, self.query_context_payload, "data"
)
assert rv.status_code == 403
@pytest.mark.chart_data_flow
class TestGetChartDataApi(BaseTestChartDataApi):

539
tests/integration_tests/datasets/api_tests.py
View File

@@ -84,16 +84,21 @@ class TestDatasetApi(SupersetTestCase):
def insert_dataset(
table_name: str,
owners: list[int],
database: Database,
database: Database | None = None,
sql: str | None = None,
schema: str | None = None,
catalog: str | None = None,
fetch_metadata: bool = True,
columns: list[TableColumn] | None = None,
metrics: list[SqlMetric] | None = None,
extra: str | None = None,
) -> SqlaTable:
obj_owners = list() # noqa: C408
for owner in owners:
user = db.session.query(security_manager.user_model).get(owner)
obj_owners.append(user)
database = database or get_example_database()
schema = schema or get_example_default_schema()
table = SqlaTable(
table_name=table_name,
schema=schema,
@@ -101,13 +106,36 @@ class TestDatasetApi(SupersetTestCase):
database=database,
sql=sql,
catalog=catalog,
extra=extra,
)
if columns:
table.columns = columns
if metrics:
table.metrics = metrics
db.session.add(table)
db.session.commit()
if fetch_metadata:
table.fetch_metadata()
return table
@staticmethod
def insert_chart(
chart_title: str,
dataset_id: int,
viz_type: str = "bar",
params: str = "{}",
) -> Slice:
chart = Slice(
slice_name=chart_title,
datasource_id=dataset_id,
datasource_type="table",
viz_type=viz_type,
params=params,
)
db.session.add(chart)
db.session.commit()
return chart
def insert_default_dataset(self):
return self.insert_dataset(
"ab_permission", [self.get_user("admin").id], get_main_database()
@@ -421,12 +449,11 @@ class TestDatasetApi(SupersetTestCase):
"""
Dataset API: Test get dataset with the render parameter.
"""
database = get_example_database()
dataset = SqlaTable(
dataset = self.insert_dataset(
table_name="test_sql_table_with_jinja",
database=database,
schema=get_example_default_schema(),
main_dttm_col="default_dttm",
owners=[],
sql="SELECT {{ current_user_id() }} as my_user_id",
fetch_metadata=False,
columns=[
TableColumn(
column_name="my_user_id",
@@ -446,10 +473,7 @@ class TestDatasetApi(SupersetTestCase):
expression="{{ url_param('multiplier') }} * 1.4",
)
],
sql="SELECT {{ current_user_id() }} as my_user_id",
)
db.session.add(dataset)
db.session.commit()
self.login(ADMIN_USERNAME)
admin = self.get_user(ADMIN_USERNAME)
@@ -493,12 +517,11 @@ class TestDatasetApi(SupersetTestCase):
Dataset API: Test get dataset with the render parameter
when rendering raises an exception.
"""
database = get_example_database()
dataset = SqlaTable(
dataset = self.insert_dataset(
table_name="test_sql_table_with_incorrect_jinja",
database=database,
schema=get_example_default_schema(),
main_dttm_col="default_dttm",
owners=[],
sql="SELECT {{ current_user_id() } as my_user_id",
fetch_metadata=False,
columns=[
TableColumn(
column_name="my_user_id",
@@ -518,10 +541,7 @@ class TestDatasetApi(SupersetTestCase):
expression="{{ url_param('multiplier') } * 1.4",
)
],
sql="SELECT {{ current_user_id() } as my_user_id",
)
db.session.add(dataset)
db.session.commit()
self.login(ADMIN_USERNAME)
@@ -679,6 +699,7 @@ class TestDatasetApi(SupersetTestCase):
"can_duplicate",
"can_get_or_create_dataset",
"can_warm_up_cache",
"can_get_drill_info",
}
def test_create_dataset_item(self):
@@ -2716,17 +2737,12 @@ class TestDatasetApi(SupersetTestCase):
"""
Dataset API: Test custom dataset_is_certified filter
"""
table_w_certification = SqlaTable(
table_w_certification = self.insert_dataset(
table_name="foo",
schema=None,
owners=[],
database=get_main_database(),
sql=None,
fetch_metadata=False,
extra='{"certification": 1}',
)
db.session.add(table_w_certification)
db.session.commit()
arguments = {
"filters": [{"col": "id", "opr": "dataset_is_certified", "value": True}]
@@ -2999,3 +3015,478 @@ class TestDatasetApi(SupersetTestCase):
assert data == {
"message": "The provided table was not found in the provided database"
}
def test_get_drill_info_admin_user(self):
"""
Dataset API: Test drill_info endpoint returns metadata for admin users, even
without a dashboard param.
"""
self.login(ADMIN_USERNAME)
dataset = self.insert_dataset(
table_name="test_drill_dataset",
owners=[],
columns=[
TableColumn(
column_name="category",
type="VARCHAR(255)",
verbose_name="Category Column",
groupby=True,
),
TableColumn(
column_name="region",
type="VARCHAR(255)",
groupby=True,
),
TableColumn(
column_name="value",
type="VARCHAR(255)",
groupby=False,
),
TableColumn(
column_name="description",
type="VARCHAR(255)",
groupby=False,
),
],
fetch_metadata=False,
)
# Test the drill_info endpoint
uri = f"api/v1/dataset/{dataset.id}/drill_info/"
rv = self.get_assert_metric(uri, "get_drill_info")
assert rv.status_code == 200
data = json.loads(rv.data.decode("utf-8"))
result = data["result"]
# Verify admin gets full dataset metadata
assert "created_by" in result
assert "created_on_humanized" in result
assert "changed_by" in result
assert "changed_on_humanized" in result
assert result["id"] == dataset.id
assert result["table_name"] == "test_drill_dataset"
assert result["owners"] == []
assert len(result["columns"]) == 2
assert result["columns"] == [
{"column_name": "category", "verbose_name": "Category Column"},
{"column_name": "region", "verbose_name": None},
]
self.items_to_delete = [dataset]
def test_get_drill_info_admin_user_dataset_not_found(self):
"""
Dataset API: Test drill_info endpoint returns 404 for non-existent dataset.
"""
self.login(ADMIN_USERNAME)
uri = "api/v1/dataset/99999/drill_info/"
rv = self.client.get(uri)
assert rv.status_code == 404
def test_get_drill_info_no_perm_to_drill(self):
"""
Dataset API: Test drill_info endpoint returns 403 for users without permission
to access the API.
"""
dataset = self.insert_dataset(table_name="foo", owners=[], fetch_metadata=False)
# Log in as alpha for dataset access but remove pvm access
with self.temporary_user(
clone_user=security_manager.find_user(username=ALPHA_USERNAME),
pvms_to_remove=[("can_get_drill_info", "Dataset")],
login=True,
):
uri = f"api/v1/dataset/{dataset.id}/drill_info/"
rv = self.client.get(uri)
assert rv.status_code == 403
self.items_to_delete = [dataset]
@patch("superset.security.manager.SupersetSecurityManager.has_guest_access")
@patch("superset.security.manager.SupersetSecurityManager.is_guest_user")
@with_feature_flags(EMBEDDED_SUPERSET=True)
def test_get_drill_info_embedded_user_no_perm_to_drill(
self, mock_is_guest_user, mock_has_guest_access
):
"""
Dataset API: Test drill_info endpoint returns 403 for embedded users when
the role does not have permission.
"""
dataset = self.insert_dataset(
table_name="test_embedded_dataset",
owners=[],
columns=[
TableColumn(
column_name="category",
type="VARCHAR(255)",
verbose_name="Category Column",
groupby=True,
),
TableColumn(
column_name="region",
type="VARCHAR(255)",
groupby=True,
),
],
fetch_metadata=False,
)
chart = self.insert_chart("Test Embedded Chart", dataset.id)
dash = self.insert_dashboard(
"Embedded Test Dashboard", "embedded-test-dashboard", [], slices=[chart]
)
# Log in to role without `can_get_drill_info` permission, and mock guest checks
with self.temporary_user(
clone_user=security_manager.find_user(username=GAMMA_USERNAME),
pvms_to_remove=[("can_get_drill_info", "Dataset")],
login=True,
):
mock_is_guest_user.return_value = True
mock_has_guest_access.return_value = True
uri = f"api/v1/dataset/{dataset.id}/drill_info/?q=(dashboard_id:{dash.id})"
rv = self.client.get(uri)
assert rv.status_code == 403
self.items_to_delete = [dash, chart, dataset]
@patch("superset.security.manager.SupersetSecurityManager.has_guest_access")
@patch("superset.security.manager.SupersetSecurityManager.is_guest_user")
@with_feature_flags(EMBEDDED_SUPERSET=True)
def test_get_drill_info_embedded_user_with_dashboard_id(
self, mock_is_guest_user, mock_has_guest_access
):
"""
Dataset API: Test drill_info endpoint with dashboard ID parameter for
embedded users.
"""
dataset = self.insert_dataset(
table_name="test_embedded_dataset",
owners=[],
columns=[
TableColumn(
column_name="category",
type="VARCHAR(255)",
verbose_name="Category Column",
groupby=True,
),
TableColumn(
column_name="region",
type="VARCHAR(255)",
groupby=True,
),
],
fetch_metadata=False,
)
chart = self.insert_chart("Test Embedded Chart", dataset.id)
dash = self.insert_dashboard(
"Embedded Test Dashboard", "embedded-test-dashboard", [], slices=[chart]
)
with self.temporary_user(
clone_user=security_manager.find_user(username=GAMMA_USERNAME),
login=True,
):
mock_is_guest_user.return_value = True
mock_has_guest_access.return_value = True
uri = f"api/v1/dataset/{dataset.id}/drill_info/?q=(dashboard_id:{dash.id})"
rv = self.client.get(uri)
assert rv.status_code == 200
data = json.loads(rv.data.decode("utf-8"))
result = data["result"]
assert result == {
"id": dataset.id,
"columns": [
{"column_name": "category", "verbose_name": "Category Column"},
{"column_name": "region", "verbose_name": None},
],
}
self.items_to_delete = [dash, chart, dataset]
@patch("superset.security.manager.SupersetSecurityManager.has_guest_access")
@patch("superset.security.manager.SupersetSecurityManager.is_guest_user")
@with_feature_flags(EMBEDDED_SUPERSET=True)
def test_get_drill_info_embedded_user_without_dashboard_parameter(
self, mock_is_guest_user, mock_has_guest_access
):
"""
Dataset API: Test drill_info endpoint without dashboard ID parameter
for embedded users.
"""
dataset = self.insert_dataset(
table_name="test_embedded_dataset",
owners=[],
columns=[
TableColumn(
column_name="category",
type="VARCHAR(255)",
verbose_name="Category Column",
groupby=True,
),
TableColumn(
column_name="region",
type="VARCHAR(255)",
groupby=True,
),
],
fetch_metadata=False,
)
chart = self.insert_chart("Test Embedded Chart", dataset.id)
dashboard = self.insert_dashboard(
"Embedded Test Dashboard", "embedded-test-dashboard", [], slices=[chart]
)
with self.temporary_user(
clone_user=security_manager.find_user(username=GAMMA_USERNAME),
login=True,
):
mock_is_guest_user.return_value = True
mock_has_guest_access.return_value = True
uri = f"api/v1/dataset/{dataset.id}/drill_info/"
rv = self.client.get(uri)
assert rv.status_code == 403
self.items_to_delete = [dashboard, chart, dataset]
@patch("superset.security.manager.SupersetSecurityManager.has_guest_access")
@patch("superset.security.manager.SupersetSecurityManager.is_guest_user")
@with_feature_flags(EMBEDDED_SUPERSET=True)
def test_get_drill_info_embedded_user_dashboard_without_dataset(
self, mock_is_guest_user, mock_has_guest_access
):
"""
Dataset API: Test drill_info with dashboard ID that user has access to but
does not contain the dataset.
"""
dataset = self.insert_dataset(
table_name="test_d2d_table",
owners=[],
columns=[
TableColumn(
column_name="category",
type="VARCHAR(255)",
groupby=True,
),
],
fetch_metadata=False,
)
dashboard_dataset = self.insert_dataset(
table_name="test_dashboard_dataset",
owners=[],
fetch_metadata=False,
)
chart = self.insert_chart("Dashboard Chart", dashboard_dataset.id)
dash = self.insert_dashboard(
"Dashboard Without Test Dataset",
"dashboard-without-test-dataset",
[],
slices=[chart],
)
with self.temporary_user(
clone_user=security_manager.find_user(username=GAMMA_USERNAME),
login=True,
):
mock_is_guest_user.return_value = True
mock_has_guest_access.return_value = True
uri = f"api/v1/dataset/{dataset.id}/drill_info/?q=(dashboard_id:{dash.id})"
rv = self.client.get(uri)
assert rv.status_code == 403
self.items_to_delete = [dash, chart, dataset, dashboard_dataset]
@with_feature_flags(DASHBOARD_RBAC=True)
def test_get_drill_info_dashboard_rbac_access_granted(self):
"""
Dataset API: Test drill_info with dashboard parameter when user has access
via the DASHBOARD_RBAC FF.
"""
with self.temporary_user(
clone_user=security_manager.find_user(username=GAMMA_USERNAME)
) as test_user:
user_role_ids = [role.id for role in test_user.roles]
# Login as admin to avoid FK issues during temp account deletion
self.login(ADMIN_USERNAME)
dataset = self.insert_dataset(
table_name="test_rbac_dataset",
owners=[],
columns=[
TableColumn(
column_name="restricted_column",
type="VARCHAR(255)",
verbose_name="Restricted Column",
groupby=True,
),
],
fetch_metadata=False,
)
chart = self.insert_chart("Test RBAC Chart", dataset.id)
dash = self.insert_dashboard(
"RBAC Test Dashboard",
"rbac-test-dashboard",
[],
roles=user_role_ids,
slices=[chart],
published=True,
)
self.logout()
self.login(test_user.username)
uri = f"api/v1/dataset/{dataset.id}/drill_info/?q=(dashboard_id:{dash.id})"
rv = self.client.get(uri)
assert rv.status_code == 200
data = json.loads(rv.data.decode("utf-8"))
result = data["result"]
assert "created_by" in result
assert "created_on_humanized" in result
assert "changed_by" in result
assert "changed_on_humanized" in result
assert result["id"] == dataset.id
assert result["table_name"] == "test_rbac_dataset"
assert len(result["columns"]) == 1
assert result["columns"][0]["column_name"] == "restricted_column"
self.items_to_delete = [dash, chart, dataset]
@with_feature_flags(DASHBOARD_RBAC=True)
def test_get_drill_info_dashboard_rbac_no_perm_to_drill(self):
"""
Dataset API: Test drill_info with dashboard parameter when user has
no permission to access the API.
"""
with self.temporary_user(
clone_user=security_manager.find_user(username=GAMMA_USERNAME),
pvms_to_remove=[("can_get_drill_info", "Dataset")],
) as test_user:
user_role_ids = [role.id for role in test_user.roles]
self.login(ADMIN_USERNAME)
dataset = self.insert_dataset(
table_name="test_rbac_dataset_denied",
owners=[],
columns=[
TableColumn(
column_name="restricted_column",
type="VARCHAR(255)",
groupby=True,
),
],
fetch_metadata=False,
)
chart = self.insert_chart("Test RBAC Chart second", dataset.id)
dash = self.insert_dashboard(
"RBAC Test Dashboard 2",
"rbac-test-dashboard-2",
[],
slices=[chart],
roles=user_role_ids,
published=True,
)
self.logout()
self.login(test_user.username)
uri = f"api/v1/dataset/{dataset.id}/drill_info/?q=(dashboard_id:{dash.id})"
rv = self.client.get(uri)
assert rv.status_code == 403
self.items_to_delete = [dash, chart, dataset]
@with_feature_flags(DASHBOARD_RBAC=True)
def test_get_drill_info_dashboard_rbac_no_access_on_dashboard(self):
"""
Dataset API: Test drill_info with dashboard parameter when user has
no access to the dashboard.
"""
dataset = self.insert_dataset(
table_name="test_rbac_dataset_denied",
owners=[],
columns=[
TableColumn(
column_name="restricted_column",
type="VARCHAR(255)",
groupby=True,
),
],
fetch_metadata=False,
)
chart = self.insert_chart("Test RBAC Chart second", dataset.id)
dash = self.insert_dashboard(
"RBAC Test Dashboard 2",
"rbac-test-dashboard-2",
[],
slices=[chart],
roles=[],
published=True,
)
with self.temporary_user(
clone_user=security_manager.find_user(username=GAMMA_USERNAME),
login=True,
username="test_new_account",
):
uri = f"api/v1/dataset/{dataset.id}/drill_info/?q=(dashboard_id:{dash.id})"
rv = self.client.get(uri)
assert rv.status_code == 403
self.items_to_delete = [dash, chart, dataset]
@with_feature_flags(DASHBOARD_RBAC=True)
def test_get_drill_info_dashboard_rbac_no_dashboard_id(self):
"""
Dataset API: Test drill_info without dashboard ID parameter falls back
to regular access control.
"""
with self.temporary_user(
clone_user=security_manager.find_user(username=GAMMA_USERNAME),
) as test_user:
self.login(ADMIN_USERNAME)
user_role_ids = [role.id for role in test_user.roles]
dataset = self.insert_dataset(
table_name="test_no_dashboard_id",
owners=[],
columns=[
TableColumn(
column_name="restricted_column",
type="VARCHAR(255)",
groupby=True,
),
],
fetch_metadata=False,
)
chart = self.insert_chart("Test RBAC Chart second", dataset.id)
dashboard = self.insert_dashboard(
"RBAC Test Dashboard 2",
"rbac-test-dashboard-2",
[],
slices=[chart],
roles=user_role_ids,
published=True,
)
self.logout()
self.login(test_user.username)
uri = f"api/v1/dataset/{dataset.id}/drill_info/"
rv = self.client.get(uri)
assert rv.status_code == 404
self.items_to_delete = [dashboard, chart, dataset]

72
tests/integration_tests/datasource_tests.py
View File

@@ -44,12 +44,16 @@ from superset.utils.database import ( # noqa: F401
)
from tests.integration_tests.base_tests import db_insert_temp_object, SupersetTestCase
from tests.integration_tests.conftest import with_feature_flags
from tests.integration_tests.constants import ADMIN_USERNAME
from tests.integration_tests.constants import ADMIN_USERNAME, GAMMA_USERNAME
from tests.integration_tests.fixtures.birth_names_dashboard import (
load_birth_names_dashboard_with_slices, # noqa: F401
load_birth_names_data, # noqa: F401
)
from tests.integration_tests.fixtures.datasource import get_datasource_post
from tests.integration_tests.fixtures.world_bank_dashboard import (
load_world_bank_dashboard_with_slices, # noqa: F401
load_world_bank_data, # noqa: F401
)
@contextmanager
@@ -516,6 +520,72 @@ class TestDatasource(SupersetTestCase):
resp = self.get_json_resp("/datasource/get/druid/500000/", raise_on_error=False)
assert resp.get("error") == "'druid' is not a valid DatasourceType"
@pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
@mock.patch(
"superset.security.manager.SupersetSecurityManager.get_guest_rls_filters"
)
@mock.patch("superset.security.manager.SupersetSecurityManager.is_guest_user")
@mock.patch("superset.security.manager.SupersetSecurityManager.has_guest_access")
@with_feature_flags(EMBEDDED_SUPERSET=True)
def test_get_samples_embedded_user(
self, mock_has_guest_access, mock_is_guest_user, mock_rls
):
"""
Embedded user can access the /samples view.
"""
self.login(ADMIN_USERNAME)
mock_is_guest_user.return_value = True
mock_has_guest_access.return_value = True
mock_rls.return_value = []
tbl = self.get_table(name="birth_names")
dash = self.get_dash_by_slug("births")
uri = f"/datasource/samples?datasource_id={tbl.id}&datasource_type=table&dashboard_id={dash.id}" # noqa: E501
resp = self.client.post(uri, json={})
assert resp.status_code == 200
@pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
@mock.patch(
"superset.security.manager.SupersetSecurityManager.get_guest_rls_filters"
)
@mock.patch("superset.security.manager.SupersetSecurityManager.is_guest_user")
@with_feature_flags(EMBEDDED_SUPERSET=True)
def test_get_samples_embedded_user_without_dash_id(
self, mock_is_guest_user, mock_rls
):
"""
Embedded user can't access the /samples view if not providing a dashboard ID.
"""
self.login(GAMMA_USERNAME)
mock_is_guest_user.return_value = True
mock_rls.return_value = []
tbl = self.get_table(name="birth_names")
uri = f"/datasource/samples?datasource_id={tbl.id}&datasource_type=table"
resp = self.client.post(uri, json={})
assert resp.status_code == 403
@pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
@pytest.mark.usefixtures("load_world_bank_dashboard_with_slices")
@mock.patch(
"superset.security.manager.SupersetSecurityManager.get_guest_rls_filters"
)
@mock.patch("superset.security.manager.SupersetSecurityManager.is_guest_user")
@with_feature_flags(EMBEDDED_SUPERSET=True)
def test_get_samples_embedded_user_dashboard_without_dataset(
self, mock_is_guest_user, mock_rls
):
"""
Embedded user can't access the /samples view when providing a dashboard ID that
does not include the target dataset.
"""
self.login(GAMMA_USERNAME)
mock_is_guest_user.return_value = True
mock_rls.return_value = []
tbl = self.get_table(name="birth_names")
dash = self.get_dash_by_slug("world_health")
uri = f"/datasource/samples?datasource_id={tbl.id}&datasource_type=table&dashboard_id={dash.id}" # noqa: E501
resp = self.client.post(uri, json={})
assert resp.status_code == 403
def test_get_samples(test_client, login_as_admin, virtual_dataset):
"""