Google Git
Sign in
apache / superset / refs/heads/fix/remove-csrf-validation-embedded-downloads / . / tests / integration_tests / dashboards / api_tests.py
blob: 2bab17d05cfe7714e16780a3b1ad8e35cb5992f7 [file] [log] [blame]
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
# isort:skip_file
"""Unit tests for Superset"""
from io import BytesIO
from time import sleep
from unittest.mock import ANY, patch
from zipfile import is_zipfile, ZipFile
from tests.integration_tests.insert_chart_mixin import InsertChartMixin
import pytest
import prison
import yaml
from freezegun import freeze_time
from sqlalchemy import and_
from superset import db, security_manager # noqa: F401
from superset.models.dashboard import Dashboard
from superset.models.core import FavStar, FavStarClassName
from superset.reports.models import ReportSchedule, ReportScheduleType
from superset.models.slice import Slice
from superset.tags.models import Tag, TaggedObject, TagType, ObjectType
from superset.utils.core import backend, override_user
from superset.utils import json
from tests.integration_tests.base_api_tests import ApiOwnersTestCaseMixin
from tests.integration_tests.base_tests import SupersetTestCase
from tests.integration_tests.conftest import with_feature_flags
from tests.integration_tests.constants import (
ADMIN_USERNAME,
ALPHA_USERNAME,
GAMMA_USERNAME,
)
from tests.integration_tests.fixtures.importexport import (
chart_config,
database_config,
dashboard_config,
dashboard_export,
dashboard_metadata_config,
dataset_config,
dataset_metadata_config,
)
from tests.integration_tests.fixtures.tags import (
create_custom_tags, # noqa: F401
get_filter_params,
)
from tests.integration_tests.utils.get_dashboards import get_dashboards_ids
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.world_bank_dashboard import (
load_world_bank_dashboard_with_slices, # noqa: F401
load_world_bank_data, # noqa: F401
)
DASHBOARDS_FIXTURE_COUNT = 10
class TestDashboardApi(ApiOwnersTestCaseMixin, InsertChartMixin, SupersetTestCase):
resource_name = "dashboard"
dashboards: list[Dashboard] = []
dashboard_data = {
"dashboard_title": "title1_changed",
"slug": "slug1_changed",
"position_json": '{"b": "B"}',
"css": "css_changed",
"json_metadata": '{"refresh_frequency": 30, "timed_refresh_immune_slices": [], "expanded_slices": {}, "color_scheme": "", "label_colors": {}, "shared_label_colors": [], "map_label_colors": {}, "color_scheme_domain": [], "cross_filters_enabled": false}', # noqa: E501
"published": False,
}
dashboard_put_filters_data = {
"modified": [
{"id": "native_filter_1", "name": "Filter 1"},
{"id": "native_filter_2", "name": "Filter 2"},
],
"deleted": [],
"reordered": [],
}
@pytest.fixture
def create_dashboards(self):
with self.create_app().app_context():
dashboards = []
admin = self.get_user("admin")
charts = []
half_dash_count = round(DASHBOARDS_FIXTURE_COUNT / 2)
for cx in range(DASHBOARDS_FIXTURE_COUNT):
dashboard = self.insert_dashboard(
f"title{cx}",
f"slug{cx}",
[admin.id],
slices=charts if cx < half_dash_count else [],
certified_by="John Doe",
certification_details="Sample certification",
)
if cx < half_dash_count:
chart = self.insert_chart(f"slice{cx}", [admin.id], 1, params="{}")
charts.append(chart)
dashboard.slices = [chart]
db.session.add(dashboard)
dashboards.append(dashboard)
fav_dashboards = []
for cx in range(half_dash_count):
fav_star = FavStar(
user_id=admin.id, class_name="Dashboard", obj_id=dashboards[cx].id
)
db.session.add(fav_star)
db.session.commit()
fav_dashboards.append(fav_star)
self.dashboards = dashboards
yield dashboards
# rollback changes
for chart in charts:
db.session.delete(chart)
for dashboard in dashboards:
db.session.delete(dashboard)
for fav_dashboard in fav_dashboards:
db.session.delete(fav_dashboard)
db.session.commit()
@pytest.fixture
def create_created_by_gamma_dashboards(self):
with self.create_app().app_context():
dashboards = []
gamma = self.get_user("gamma")
for cx in range(2):
dashboard = self.insert_dashboard(
f"create_title{cx}",
f"create_slug{cx}",
[gamma.id],
created_by=gamma,
)
sleep(1)
dashboards.append(dashboard)
yield dashboards
for dashboard in dashboards:
db.session.delete(dashboard)
db.session.commit()
@pytest.fixture
def create_dashboard_with_report(self):
with self.create_app().app_context():
admin = self.get_user("admin")
dashboard = self.insert_dashboard(
"dashboard_report",
"dashboard_report",
[admin.id], # noqa: F541
)
report_schedule = ReportSchedule(
type=ReportScheduleType.REPORT,
name="report_with_dashboard",
crontab="* * * * *",
dashboard=dashboard,
)
db.session.commit()
yield dashboard
# rollback changes
db.session.delete(report_schedule)
db.session.delete(dashboard)
db.session.commit()
@pytest.fixture
def create_dashboard_with_tag(self, create_custom_tags): # noqa: F811
with self.create_app().app_context():
gamma = self.get_user("gamma")
dashboard = self.insert_dashboard(
"dash with tag",
None,
[gamma.id],
)
tag = db.session.query(Tag).filter(Tag.name == "first_tag").first()
tag_association = TaggedObject(
object_id=dashboard.id,
object_type=ObjectType.dashboard,
tag=tag,
)
db.session.add(tag_association)
db.session.commit()
yield dashboard
# rollback changes
db.session.delete(tag_association)
db.session.delete(dashboard)
db.session.commit()
@pytest.fixture
def create_dashboards_some_with_tags(self, create_custom_tags): # noqa: F811
"""
Fixture that creates 4 dashboards:
- ``first_dashboard`` is associated with ``first_tag``
- ``second_dashboard`` is associated with ``second_tag``
- ``third_dashboard`` is associated with both ``first_tag`` and ``second_tag``
- ``fourth_dashboard`` is not associated with any tag
Relies on the ``create_custom_tags`` fixture for the tag creation.
""" # noqa: E501
with self.create_app().app_context():
admin_user = self.get_user(ADMIN_USERNAME)
tags = {
"first_tag": db.session.query(Tag)
.filter(Tag.name == "first_tag")
.first(),
"second_tag": db.session.query(Tag)
.filter(Tag.name == "second_tag")
.first(),
}
dashboard_names = [
"first_dashboard",
"second_dashboard",
"third_dashboard",
"fourth_dashboard",
]
dashboards = [
self.insert_dashboard(name, None, [admin_user.id])
for name in dashboard_names
]
tag_associations = [
TaggedObject(
object_id=dashboards[0].id,
object_type=ObjectType.chart,
tag=tags["first_tag"],
),
TaggedObject(
object_id=dashboards[1].id,
object_type=ObjectType.chart,
tag=tags["second_tag"],
),
TaggedObject(
object_id=dashboards[2].id,
object_type=ObjectType.chart,
tag=tags["first_tag"],
),
TaggedObject(
object_id=dashboards[2].id,
object_type=ObjectType.chart,
tag=tags["second_tag"],
),
]
for association in tag_associations:
db.session.add(association)
db.session.commit()
yield dashboards
# rollback changes
for association in tag_associations:
db.session.delete(association)
for chart in dashboards:
db.session.delete(chart)
db.session.commit()
@pytest.mark.usefixtures("load_world_bank_dashboard_with_slices")
@patch("superset.utils.log.logger")
def test_get_dashboard_datasets(self, logger_mock):
self.login(ADMIN_USERNAME)
uri = "api/v1/dashboard/world_health/datasets"
response = self.get_assert_metric(uri, "get_datasets")
assert response.status_code == 200
data = json.loads(response.data.decode("utf-8"))
dashboard = Dashboard.get("world_health")
expected_dataset_ids = {s.datasource_id for s in dashboard.slices}
result = data["result"]
actual_dataset_ids = {dataset["id"] for dataset in result}
assert actual_dataset_ids == expected_dataset_ids
expected_values = [0, 1] if backend() == "presto" else [0, 1, 2]
assert result[0]["column_types"] == expected_values
logger_mock.warning.assert_not_called()
@pytest.mark.usefixtures("load_world_bank_dashboard_with_slices")
@patch("superset.dashboards.schemas.security_manager.has_guest_access")
@patch("superset.dashboards.schemas.security_manager.is_guest_user")
def test_get_dashboard_datasets_as_guest(self, is_guest_user, has_guest_access):
self.login(ADMIN_USERNAME)
uri = "api/v1/dashboard/world_health/datasets"
response = self.get_assert_metric(uri, "get_datasets")
assert response.status_code == 200
data = json.loads(response.data.decode("utf-8"))
dashboard = Dashboard.get("world_health")
expected_dataset_ids = {s.datasource_id for s in dashboard.slices}
result = data["result"]
actual_dataset_ids = {dataset["id"] for dataset in result}
assert actual_dataset_ids == expected_dataset_ids
for dataset in result:
for excluded_key in ["database", "owners"]:
assert excluded_key not in dataset
@pytest.mark.usefixtures("load_world_bank_dashboard_with_slices")
@patch("superset.utils.log.logger")
def test_get_dashboard_datasets_not_found(self, logger_mock):
self.login(ALPHA_USERNAME)
uri = "api/v1/dashboard/not_found/datasets"
response = self.get_assert_metric(uri, "get_datasets")
assert response.status_code == 404
logger_mock.warning.assert_called_once_with(
"Dashboard not found.", exc_info=True
)
@pytest.mark.usefixtures("load_world_bank_dashboard_with_slices")
@patch("superset.utils.log.logger")
@patch("superset.daos.dashboard.DashboardDAO.get_datasets_for_dashboard")
def test_get_dashboard_datasets_invalid_schema(
self, dashboard_datasets_mock, logger_mock
):
dashboard_datasets_mock.side_effect = TypeError("Invalid schema")
self.login(ADMIN_USERNAME)
uri = "api/v1/dashboard/world_health/datasets"
response = self.get_assert_metric(uri, "get_datasets")
assert response.status_code == 422
logger_mock.warning.assert_called_once_with(
"Dataset schema is invalid, caused by: Invalid schema", exc_info=True
)
@pytest.mark.usefixtures("create_dashboards")
def test_get_gamma_dashboard_datasets(self):
"""
Check that a gamma user with data access can access dashboard/datasets
"""
from superset.connectors.sqla.models import SqlaTable
# Set correct role permissions
gamma_role = security_manager.find_role("Gamma")
fixture_dataset = db.session.query(SqlaTable).get(1)
data_access_pvm = security_manager.add_permission_view_menu(
"datasource_access", fixture_dataset.perm
)
gamma_role.permissions.append(data_access_pvm)
db.session.commit()
self.login(GAMMA_USERNAME)
dashboard = self.dashboards[0]
dashboard.published = True
db.session.commit()
uri = f"api/v1/dashboard/{dashboard.id}/datasets"
response = self.get_assert_metric(uri, "get_datasets")
assert response.status_code == 200
# rollback permission change
data_access_pvm = security_manager.find_permission_view_menu(
"datasource_access", fixture_dataset.perm
)
security_manager.del_permission_role(gamma_role, data_access_pvm)
@pytest.mark.usefixtures("create_dashboards")
def get_dashboard_by_slug(self):
self.login(ADMIN_USERNAME)
dashboard = self.dashboards[0]
uri = f"api/v1/dashboard/{dashboard.slug}"
response = self.get_assert_metric(uri, "get")
assert response.status_code == 200
data = json.loads(response.data.decode("utf-8"))
assert data["id"] == dashboard.id
@pytest.mark.usefixtures("create_dashboards")
def get_dashboard_by_bad_slug(self):
self.login(ADMIN_USERNAME)
dashboard = self.dashboards[0]
uri = f"api/v1/dashboard/{dashboard.slug}-bad-slug"
response = self.get_assert_metric(uri, "get")
assert response.status_code == 404
@pytest.mark.usefixtures("create_dashboards")
def get_draft_dashboard_by_slug(self):
"""
All users should have access to dashboards without roles
"""
self.login(GAMMA_USERNAME)
dashboard = self.dashboards[0]
uri = f"api/v1/dashboard/{dashboard.slug}"
response = self.get_assert_metric(uri, "get")
assert response.status_code == 200
@pytest.mark.usefixtures("create_dashboards")
def test_get_dashboard_charts(self):
"""
Dashboard API: Test getting charts belonging to a dashboard
"""
self.login(ADMIN_USERNAME)
dashboard = self.dashboards[0]
uri = f"api/v1/dashboard/{dashboard.id}/charts"
response = self.get_assert_metric(uri, "get_charts")
assert response.status_code == 200
data = json.loads(response.data.decode("utf-8"))
assert len(data["result"]) == 1
result = data["result"][0]
assert set(result.keys()) == {
"cache_timeout",
"certification_details",
"certified_by",
"changed_on",
"description",
"description_markeddown",
"form_data",
"id",
"slice_name",
"slice_url",
}
assert result["id"] == dashboard.slices[0].id
assert result["slice_name"] == dashboard.slices[0].slice_name
@pytest.mark.usefixtures("create_dashboards")
def test_get_dashboard_charts_by_slug(self):
"""
Dashboard API: Test getting charts belonging to a dashboard
"""
self.login(ADMIN_USERNAME)
dashboard = self.dashboards[0]
uri = f"api/v1/dashboard/{dashboard.slug}/charts"
response = self.get_assert_metric(uri, "get_charts")
assert response.status_code == 200
data = json.loads(response.data.decode("utf-8"))
assert len(data["result"]) == 1
assert data["result"][0]["slice_name"] == dashboard.slices[0].slice_name
@pytest.mark.usefixtures("create_dashboards")
def test_get_dashboard_charts_not_found(self):
"""
Dashboard API: Test getting charts belonging to a dashboard that does not exist
"""
self.login(ADMIN_USERNAME)
bad_id = self.get_nonexistent_numeric_id(Dashboard)
uri = f"api/v1/dashboard/{bad_id}/charts"
response = self.get_assert_metric(uri, "get_charts")
assert response.status_code == 404
@pytest.mark.usefixtures("load_world_bank_dashboard_with_slices")
def test_get_dashboard_datasets_not_allowed(self):
self.login(GAMMA_USERNAME)
uri = "api/v1/dashboard/world_health/datasets"
response = self.get_assert_metric(uri, "get_datasets")
assert response.status_code == 404
@pytest.mark.usefixtures("create_dashboards")
def test_get_gamma_dashboard_charts(self):
"""
Check that a gamma user with data access can access dashboard/charts
"""
from superset.connectors.sqla.models import SqlaTable
# Set correct role permissions
gamma_role = security_manager.find_role("Gamma")
fixture_dataset = db.session.query(SqlaTable).get(1)
data_access_pvm = security_manager.add_permission_view_menu(
"datasource_access", fixture_dataset.perm
)
gamma_role.permissions.append(data_access_pvm)
db.session.commit()
self.login(GAMMA_USERNAME)
dashboard = self.dashboards[0]
dashboard.published = True
db.session.commit()
uri = f"api/v1/dashboard/{dashboard.id}/charts"
response = self.get_assert_metric(uri, "get_charts")
assert response.status_code == 200
# rollback permission change
data_access_pvm = security_manager.find_permission_view_menu(
"datasource_access", fixture_dataset.perm
)
security_manager.del_permission_role(gamma_role, data_access_pvm)
@pytest.mark.usefixtures("create_dashboards")
def test_get_dashboard_charts_empty(self):
"""
Dashboard API: Test getting charts belonging to a dashboard without any charts
"""
self.login(ADMIN_USERNAME)
# the fixture setup assigns no charts to the second half of dashboards
uri = f"api/v1/dashboard/{self.dashboards[-1].id}/charts"
response = self.get_assert_metric(uri, "get_charts")
assert response.status_code == 200
data = json.loads(response.data.decode("utf-8"))
assert data["result"] == []
def test_get_dashboard(self):
"""
Dashboard API: Test get dashboard
"""
admin = self.get_user("admin")
dashboard = self.insert_dashboard(
"title", "slug1", [admin.id], created_by=admin
)
self.login(ADMIN_USERNAME)
uri = f"api/v1/dashboard/{dashboard.id}"
rv = self.get_assert_metric(uri, "get")
assert rv.status_code == 200
with override_user(admin):
expected_result = {
"certified_by": None,
"certification_details": None,
"changed_by": None,
"changed_by_name": "",
"charts": [],
"created_by": {
"id": 1,
"first_name": "admin",
"last_name": "user",
},
"id": dashboard.id,
"css": "",
"dashboard_title": "title",
"datasources": [],
"json_metadata": "",
"owners": [
{
"id": 1,
"first_name": "admin",
"last_name": "user",
}
],
"roles": [],
"position_json": "",
"published": False,
"url": "/superset/dashboard/slug1/",
"slug": "slug1",
"tags": [],
"thumbnail_url": dashboard.thumbnail_url,
"is_managed_externally": False,
}
data = json.loads(rv.data.decode("utf-8"))
assert "changed_on" in data["result"]
assert "changed_on_delta_humanized" in data["result"]
assert "created_on_delta_humanized" in data["result"]
for key, value in data["result"].items():
# We can't assert timestamp values
if key not in (
"changed_on",
"changed_on_delta_humanized",
"created_on_delta_humanized",
):
assert value == expected_result[key]
# rollback changes
db.session.delete(dashboard)
db.session.commit()
@patch("superset.dashboards.schemas.security_manager.has_guest_access")
@patch("superset.dashboards.schemas.security_manager.is_guest_user")
def test_get_dashboard_as_guest(self, is_guest_user, has_guest_access):
"""
Dashboard API: Test get dashboard as guest
"""
admin = self.get_user("admin")
dashboard = self.insert_dashboard(
"title", "slug1", [admin.id], created_by=admin
)
is_guest_user.return_value = True
has_guest_access.return_value = True
self.login(ADMIN_USERNAME)
uri = f"api/v1/dashboard/{dashboard.id}"
rv = self.get_assert_metric(uri, "get")
assert rv.status_code == 200
data = json.loads(rv.data.decode("utf-8"))
for excluded_key in ["changed_by", "changed_by_name", "owners"]:
assert excluded_key not in data["result"]
# rollback changes
db.session.delete(dashboard)
db.session.commit()
def test_info_dashboard(self):
"""
Dashboard API: Test info
"""
self.login(ADMIN_USERNAME)
uri = "api/v1/dashboard/_info"
rv = self.get_assert_metric(uri, "info")
assert rv.status_code == 200
def test_info_security_dashboard(self):
"""
Dashboard API: Test info security
"""
self.login(ADMIN_USERNAME)
params = {"keys": ["permissions"]}
uri = f"api/v1/dashboard/_info?q={prison.dumps(params)}"
rv = self.get_assert_metric(uri, "info")
data = json.loads(rv.data.decode("utf-8"))
assert rv.status_code == 200
assert set(data["permissions"]) == {
"can_read",
"can_write",
"can_export",
"can_get_embedded",
"can_delete_embedded",
"can_set_embedded",
"can_cache_dashboard_screenshot",
}
@pytest.mark.usefixtures("load_world_bank_dashboard_with_slices")
def test_get_dashboard_not_found(self):
"""
Dashboard API: Test get dashboard not found
"""
bad_id = self.get_nonexistent_numeric_id(Dashboard)
self.login(ADMIN_USERNAME)
uri = f"api/v1/dashboard/{bad_id}"
rv = self.get_assert_metric(uri, "get")
assert rv.status_code == 404
def test_get_dashboard_no_data_access(self):
"""
Dashboard API: Test get dashboard without data access
"""
admin = self.get_user("admin")
dashboard = self.insert_dashboard("title", "slug1", [admin.id])
self.login(GAMMA_USERNAME)
uri = f"api/v1/dashboard/{dashboard.id}"
rv = self.client.get(uri)
assert rv.status_code == 404
# rollback changes
db.session.delete(dashboard)
db.session.commit()
def test_get_dashboards_changed_on(self):
"""
Dashboard API: Test get dashboards changed on
"""
from datetime import datetime
import humanize
with freeze_time("2020-01-01T00:00:00Z"):
admin = self.get_user("admin")
dashboard = self.insert_dashboard("title", "slug1", [admin.id])
self.login(ADMIN_USERNAME)
arguments = {
"order_column": "changed_on_delta_humanized",
"order_direction": "desc",
}
uri = f"api/v1/dashboard/?q={prison.dumps(arguments)}"
rv = self.get_assert_metric(uri, "get_list")
assert rv.status_code == 200
data = json.loads(rv.data.decode("utf-8"))
assert data["result"][0][
"changed_on_delta_humanized"
] == humanize.naturaltime(datetime.now())
# rollback changes
db.session.delete(dashboard)
db.session.commit()
def test_get_dashboards_filter(self):
"""
Dashboard API: Test get dashboards filter
"""
admin = self.get_user("admin")
gamma = self.get_user("gamma")
dashboard = self.insert_dashboard("title", "slug1", [admin.id, gamma.id])
self.login(ADMIN_USERNAME)
arguments = {
"filters": [{"col": "dashboard_title", "opr": "sw", "value": "ti"}]
}
uri = f"api/v1/dashboard/?q={prison.dumps(arguments)}"
rv = self.get_assert_metric(uri, "get_list")
assert rv.status_code == 200
data = json.loads(rv.data.decode("utf-8"))
assert data["count"] == 1
arguments = {
"filters": [
{"col": "owners", "opr": "rel_m_m", "value": [admin.id, gamma.id]}
]
}
uri = f"api/v1/dashboard/?q={prison.dumps(arguments)}"
rv = self.client.get(uri)
assert rv.status_code == 200
data = json.loads(rv.data.decode("utf-8"))
assert data["count"] == 1
# rollback changes
db.session.delete(dashboard)
db.session.commit()
@pytest.mark.usefixtures("create_dashboards")
def test_get_dashboards_title_or_slug_filter(self):
"""
Dashboard API: Test get dashboards title or slug filter
"""
# Test title filter with ilike
arguments = {
"filters": [
{"col": "dashboard_title", "opr": "title_or_slug", "value": "title1"}
],
"order_column": "dashboard_title",
"order_direction": "asc",
"keys": ["none"],
"columns": ["dashboard_title", "slug"],
}
self.login(ADMIN_USERNAME)
uri = f"api/v1/dashboard/?q={prison.dumps(arguments)}"
rv = self.client.get(uri)
assert rv.status_code == 200
data = json.loads(rv.data.decode("utf-8"))
assert data["count"] == 1
expected_response = [
{"slug": "slug1", "dashboard_title": "title1"},
]
assert data["result"] == expected_response
# Test slug filter with ilike
arguments["filters"][0]["value"] = "slug2"
uri = f"api/v1/dashboard/?q={prison.dumps(arguments)}"
rv = self.client.get(uri)
assert rv.status_code == 200
data = json.loads(rv.data.decode("utf-8"))
assert data["count"] == 1
expected_response = [
{"slug": "slug2", "dashboard_title": "title2"},
]
assert data["result"] == expected_response
self.logout()
self.login(GAMMA_USERNAME)
uri = f"api/v1/dashboard/?q={prison.dumps(arguments)}"
rv = self.client.get(uri)
assert rv.status_code == 200
data = json.loads(rv.data.decode("utf-8"))
assert data["count"] == 0
@pytest.mark.usefixtures("create_dashboards")
def test_get_dashboards_favorite_filter(self):
"""
Dashboard API: Test get dashboards favorite filter
"""
admin = self.get_user("admin")
users_favorite_query = db.session.query(FavStar.obj_id).filter(
and_(FavStar.user_id == admin.id, FavStar.class_name == "Dashboard")
)
expected_models = (
db.session.query(Dashboard)
.filter(and_(Dashboard.id.in_(users_favorite_query)))
.order_by(Dashboard.dashboard_title.asc())
.all()
)
arguments = {
"filters": [{"col": "id", "opr": "dashboard_is_favorite", "value": True}],
"order_column": "dashboard_title",
"order_direction": "asc",
"keys": ["none"],
"columns": ["dashboard_title"],
}
self.login(ADMIN_USERNAME)
uri = f"api/v1/dashboard/?q={prison.dumps(arguments)}"
rv = self.client.get(uri)
assert rv.status_code == 200
data = json.loads(rv.data.decode("utf-8"))
assert len(expected_models) == data["count"]
for i, expected_model in enumerate(expected_models):
assert (
expected_model.dashboard_title == data["result"][i]["dashboard_title"]
)
@pytest.mark.usefixtures("create_dashboards_some_with_tags")
def test_get_dashboards_tag_filters(self):
"""
Dashboard API: Test get dashboards with tag filters
"""
# Get custom tags relationship
tags = {
"first_tag": db.session.query(Tag).filter(Tag.name == "first_tag").first(),
"second_tag": db.session.query(Tag)
.filter(Tag.name == "second_tag")
.first(),
"third_tag": db.session.query(Tag).filter(Tag.name == "third_tag").first(),
}
dashboard_tag_relationship = {
tag.name: db.session.query(Dashboard.id)
.join(Dashboard.tags)
.filter(Tag.id == tag.id)
.all()
for tag in tags.values()
}
# Validate API results for each tag
for tag_name, tag in tags.items():
expected_dashboards = dashboard_tag_relationship[tag_name]
# Filter by tag ID
filter_params = get_filter_params("dashboard_tag_id", tag.id)
response_by_id = self.get_list("dashboard", filter_params)
assert response_by_id.status_code == 200
data_by_id = json.loads(response_by_id.data.decode("utf-8"))
# Filter by tag name
filter_params = get_filter_params("dashboard_tags", tag.name)
response_by_name = self.get_list("dashboard", filter_params)
assert response_by_name.status_code == 200
data_by_name = json.loads(response_by_name.data.decode("utf-8"))
# Compare results
assert data_by_id["count"] == data_by_name["count"], len(
expected_dashboards
)
assert set(chart["id"] for chart in data_by_id["result"]) == set( # noqa: C401
chart["id"] for chart in data_by_name["result"]
), set(chart.id for chart in expected_dashboards) # noqa: C401
@pytest.mark.usefixtures("create_dashboards")
def test_get_current_user_favorite_status(self):
"""
Dataset API: Test get current user favorite stars
"""
admin = self.get_user("admin")
users_favorite_ids = [
star.obj_id
for star in db.session.query(FavStar.obj_id)
.filter(
and_(
FavStar.user_id == admin.id,
FavStar.class_name == FavStarClassName.DASHBOARD,
)
)
.all()
]
assert users_favorite_ids
arguments = [dash.id for dash in db.session.query(Dashboard.id).all()]
self.login(ADMIN_USERNAME)
uri = f"api/v1/dashboard/favorite_status/?q={prison.dumps(arguments)}"
rv = self.client.get(uri)
data = json.loads(rv.data.decode("utf-8"))
assert rv.status_code == 200
for res in data["result"]:
if res["id"] in users_favorite_ids:
assert res["value"]
def test_add_favorite(self):
"""
Dataset API: Test add dashboard to favorites
"""
dashboard = Dashboard(
id=100,
dashboard_title="test_dashboard",
slug="test_slug",
slices=[],
published=True,
)
db.session.add(dashboard)
db.session.commit()
self.login(ADMIN_USERNAME)
uri = f"api/v1/dashboard/favorite_status/?q={prison.dumps([dashboard.id])}"
rv = self.client.get(uri)
data = json.loads(rv.data.decode("utf-8"))
for res in data["result"]:
assert res["value"] is False
uri = f"api/v1/dashboard/{dashboard.id}/favorites/"
self.client.post(uri)
uri = f"api/v1/dashboard/favorite_status/?q={prison.dumps([dashboard.id])}"
rv = self.client.get(uri)
data = json.loads(rv.data.decode("utf-8"))
for res in data["result"]:
assert res["value"] is True
db.session.delete(dashboard)
db.session.commit()
def test_remove_favorite(self):
"""
Dataset API: Test remove dashboard from favorites
"""
dashboard = Dashboard(
id=100,
dashboard_title="test_dashboard",
slug="test_slug",
slices=[],
published=True,
)
db.session.add(dashboard)
db.session.commit()
self.login(ADMIN_USERNAME)
uri = f"api/v1/dashboard/{dashboard.id}/favorites/"
self.client.post(uri)
uri = f"api/v1/dashboard/favorite_status/?q={prison.dumps([dashboard.id])}"
rv = self.client.get(uri)
data = json.loads(rv.data.decode("utf-8"))
for res in data["result"]:
assert res["value"] is True
uri = f"api/v1/dashboard/{dashboard.id}/favorites/"
self.client.delete(uri)
uri = f"api/v1/dashboard/favorite_status/?q={prison.dumps([dashboard.id])}"
rv = self.client.get(uri)
data = json.loads(rv.data.decode("utf-8"))
for res in data["result"]:
assert res["value"] is False
db.session.delete(dashboard)
db.session.commit()
@pytest.mark.usefixtures("create_dashboards")
def test_get_dashboards_not_favorite_filter(self):
"""
Dashboard API: Test get dashboards not favorite filter
"""
admin = self.get_user("admin")
users_favorite_query = db.session.query(FavStar.obj_id).filter(
and_(FavStar.user_id == admin.id, FavStar.class_name == "Dashboard")
)
expected_models = (
db.session.query(Dashboard)
.filter(and_(~Dashboard.id.in_(users_favorite_query)))
.order_by(Dashboard.dashboard_title.asc())
.all()
)
arguments = {
"filters": [{"col": "id", "opr": "dashboard_is_favorite", "value": False}],
"order_column": "dashboard_title",
"order_direction": "asc",
"keys": ["none"],
"columns": ["dashboard_title"],
}
uri = f"api/v1/dashboard/?q={prison.dumps(arguments)}"
self.login(ADMIN_USERNAME)
rv = self.client.get(uri)
data = json.loads(rv.data.decode("utf-8"))
assert rv.status_code == 200
assert len(expected_models) == data["count"]
for i, expected_model in enumerate(expected_models):
assert (
expected_model.dashboard_title == data["result"][i]["dashboard_title"]
)
@pytest.mark.usefixtures("create_dashboards")
def test_gets_certified_dashboards_filter(self):
arguments = {
"filters": [
{
"col": "id",
"opr": "dashboard_is_certified",
"value": True,
}
],
"keys": ["none"],
"columns": ["dashboard_title"],
}
self.login(ADMIN_USERNAME)
uri = f"api/v1/dashboard/?q={prison.dumps(arguments)}"
rv = self.get_assert_metric(uri, "get_list")
assert rv.status_code == 200
data = json.loads(rv.data.decode("utf-8"))
assert data["count"] == DASHBOARDS_FIXTURE_COUNT
@pytest.mark.usefixtures("create_dashboards")
def test_gets_not_certified_dashboards_filter(self):
arguments = {
"filters": [
{
"col": "id",
"opr": "dashboard_is_certified",
"value": False,
}
],
"keys": ["none"],
"columns": ["dashboard_title"],
}
self.login(ADMIN_USERNAME)
uri = f"api/v1/dashboard/?q={prison.dumps(arguments)}"
rv = self.get_assert_metric(uri, "get_list")
assert rv.status_code == 200
data = json.loads(rv.data.decode("utf-8"))
assert data["count"] == 0
@pytest.mark.usefixtures("create_created_by_gamma_dashboards")
def test_get_dashboards_created_by_me(self):
"""
Dashboard API: Test get dashboards created by current user
"""
query = {
"columns": ["created_on_delta_humanized", "dashboard_title", "url"],
"filters": [
{"col": "created_by", "opr": "dashboard_created_by_me", "value": "me"}
],
"order_column": "changed_on",
"order_direction": "desc",
"page": 0,
"page_size": 100,
}
uri = f"api/v1/dashboard/?q={prison.dumps(query)}"
self.login(GAMMA_USERNAME)
rv = self.client.get(uri)
data = json.loads(rv.data.decode("utf-8"))
assert rv.status_code == 200
assert len(data["result"]) == 2
assert list(data["result"][0].keys()) == query["columns"]
expected_results = [
{
"dashboard_title": "create_title1",
"url": "/superset/dashboard/create_slug1/",
},
{
"dashboard_title": "create_title0",
"url": "/superset/dashboard/create_slug0/",
},
]
for idx, response_item in enumerate(data["result"]):
for key, value in expected_results[idx].items():
assert response_item[key] == value
def test_get_dashboard_tabs(self):
"""
Dashboard API: Test get dashboard tabs
"""
position_data = {
"GRID_ID": {"children": [], "id": "GRID_ID", "type": "GRID"},
"ROOT_ID": {
"children": ["TABS-tDGEcwZ82u"],
"id": "ROOT_ID",
"type": "ROOT",
},
"TAB-0TkqQRxzg7": {
"children": [],
"id": "TAB-0TkqQRxzg7",
"meta": {"text": "P2 - T1"},
"type": "TAB",
},
"TAB-1iG_yOlKA2": {
"children": [],
"id": "TAB-1iG_yOlKA2",
"meta": {"text": "P1 - T1"},
"type": "TAB",
},
"TAB-2dgADEurF": {
"children": ["TABS-LsyXZWG2rk"],
"id": "TAB-2dgADEurF",
"meta": {"text": "P1 - T2"},
"type": "TAB",
},
"TAB-BJIt5SdCx3": {
"children": [],
"id": "TAB-BJIt5SdCx3",
"meta": {"text": "P1 - T2 - T1"},
"type": "TAB",
},
"TAB-CjZlNL5Uz": {
"children": ["TABS-Ji_K1ZBE0M"],
"id": "TAB-CjZlNL5Uz",
"meta": {"text": "Parent Tab 2"},
"type": "TAB",
},
"TAB-Nct5fiHtn": {
"children": [],
"id": "TAB-Nct5fiHtn",
"meta": {"text": "P1 - T2 - T3"},
"type": "TAB",
},
"TAB-PumuDkWKq": {
"children": [],
"id": "TAB-PumuDkWKq",
"meta": {"text": "P2 - T2"},
"type": "TAB",
},
"TAB-hyTv5L7zz": {
"children": [],
"id": "TAB-hyTv5L7zz",
"meta": {"text": "P1 - T2 - T2"},
"type": "TAB",
},
"TAB-qL7fSzr3jl": {
"children": ["TABS-N8ODUqp2sE"],
"id": "TAB-qL7fSzr3jl",
"meta": {"text": "Parent Tab 1"},
"type": "TAB",
},
"TABS-Ji_K1ZBE0M": {
"children": ["TAB-0TkqQRxzg7", "TAB-PumuDkWKq"],
"id": "TABS-Ji_K1ZBE0M",
"meta": {},
"type": "TABS",
},
"TABS-LsyXZWG2rk": {
"children": ["TAB-BJIt5SdCx3", "TAB-hyTv5L7zz", "TAB-Nct5fiHtn"],
"id": "TABS-LsyXZWG2rk",
"meta": {},
"type": "TABS",
},
"TABS-N8ODUqp2sE": {
"children": ["TAB-1iG_yOlKA2", "TAB-2dgADEurF"],
"id": "TABS-N8ODUqp2sE",
"meta": {},
"type": "TABS",
},
"TABS-tDGEcwZ82u": {
"children": ["TAB-qL7fSzr3jl", "TAB-CjZlNL5Uz"],
"id": "TABS-tDGEcwZ82u",
"meta": {},
"type": "TABS",
},
}
admin_id = self.get_user("admin").id
dashboard = self.insert_dashboard(
"title", "slug", [admin_id], position_json=json.dumps(position_data)
)
self.login(ADMIN_USERNAME)
uri = f"api/v1/dashboard/{dashboard.id}/tabs"
rv = self.get_assert_metric(uri, "get_tabs")
response = json.loads(rv.data.decode("utf-8"))
expected_response = {
"result": {
"all_tabs": {
"TAB-0TkqQRxzg7": "P2 - T1",
"TAB-1iG_yOlKA2": "P1 - T1",
"TAB-2dgADEurF": "P1 - T2",
"TAB-BJIt5SdCx3": "P1 - T2 - T1",
"TAB-CjZlNL5Uz": "Parent Tab 2",
"TAB-Nct5fiHtn": "P1 - T2 - T3",
"TAB-PumuDkWKq": "P2 - T2",
"TAB-hyTv5L7zz": "P1 - T2 - T2",
"TAB-qL7fSzr3jl": "Parent Tab 1",
},
"tab_tree": [
{
"children": [
{
"children": [],
"title": "P1 - T1",
"value": "TAB-1iG_yOlKA2",
},
{
"children": [
{
"children": [],
"title": "P1 - T2 - T1",
"value": "TAB-BJIt5SdCx3",
},
{
"children": [],
"title": "P1 - T2 - T2",
"value": "TAB-hyTv5L7zz",
},
{
"children": [],
"title": "P1 - T2 - T3",
"value": "TAB-Nct5fiHtn",
},
],
"title": "P1 - T2",
"value": "TAB-2dgADEurF",
},
],
"title": "Parent Tab 1",
"value": "TAB-qL7fSzr3jl",
},
{
"children": [
{
"children": [],
"title": "P2 - T1",
"value": "TAB-0TkqQRxzg7",
},
{
"children": [],
"title": "P2 - T2",
"value": "TAB-PumuDkWKq",
},
],
"title": "Parent Tab 2",
"value": "TAB-CjZlNL5Uz",
},
],
}
}
assert rv.status_code == 200
assert response == expected_response
db.session.delete(dashboard)
db.session.commit()
@pytest.mark.usefixtures("load_world_bank_dashboard_with_slices")
def test_get_dashboard_tabs_not_found(self):
"""
Dashboard API: Test get dashboard tabs not found
"""
bad_id = self.get_nonexistent_numeric_id(Dashboard)
self.login(ADMIN_USERNAME)
uri = f"api/v1/dashboard/{bad_id}/tabs"
rv = self.get_assert_metric(uri, "get_tabs")
assert rv.status_code == 404
def create_dashboard_import(self):
buf = BytesIO()
with ZipFile(buf, "w") as bundle:
with bundle.open("dashboard_export/metadata.yaml", "w") as fp:
fp.write(yaml.safe_dump(dashboard_metadata_config).encode())
with bundle.open(
"dashboard_export/databases/imported_database.yaml", "w"
) as fp:
fp.write(yaml.safe_dump(database_config).encode())
with bundle.open(
"dashboard_export/datasets/imported_dataset.yaml", "w"
) as fp:
fp.write(yaml.safe_dump(dataset_config).encode())
with bundle.open("dashboard_export/charts/imported_chart.yaml", "w") as fp:
fp.write(yaml.safe_dump(chart_config).encode())
with bundle.open(
"dashboard_export/dashboards/imported_dashboard.yaml", "w"
) as fp:
fp.write(yaml.safe_dump(dashboard_config).encode())
buf.seek(0)
return buf
def create_invalid_dashboard_import(self):
buf = BytesIO()
with ZipFile(buf, "w") as bundle:
with bundle.open("sql/dump.sql", "w") as fp:
fp.write(b"CREATE TABLE foo (bar INT)")
buf.seek(0)
return buf
def test_delete_dashboard(self):
"""
Dashboard API: Test delete
"""
admin_id = self.get_user("admin").id
dashboard_id = self.insert_dashboard("title", "slug1", [admin_id]).id
self.login(ADMIN_USERNAME)
uri = f"api/v1/dashboard/{dashboard_id}"
rv = self.delete_assert_metric(uri, "delete")
assert rv.status_code == 200
model = db.session.query(Dashboard).get(dashboard_id)
assert model is None
def test_delete_bulk_dashboards(self):
"""
Dashboard API: Test delete bulk
"""
admin_id = self.get_user("admin").id
dashboard_count = 4
dashboard_ids = list() # noqa: C408
for dashboard_name_index in range(dashboard_count):
dashboard_ids.append(
self.insert_dashboard(
f"title{dashboard_name_index}",
f"slug{dashboard_name_index}",
[admin_id],
).id
)
self.login(ADMIN_USERNAME)
argument = dashboard_ids
uri = f"api/v1/dashboard/?q={prison.dumps(argument)}"
rv = self.delete_assert_metric(uri, "bulk_delete")
assert rv.status_code == 200
response = json.loads(rv.data.decode("utf-8"))
expected_response = {"message": f"Deleted {dashboard_count} dashboards"}
assert response == expected_response
for dashboard_id in dashboard_ids:
model = db.session.query(Dashboard).get(dashboard_id)
assert model is None
def test_delete_bulk_embedded_dashboards(self):
"""
Dashboard API: Test delete bulk embedded
"""
user = self.get_user("admin")
dashboard_count = 4
dashboard_ids = list() # noqa: C408
for dashboard_name_index in range(dashboard_count):
dashboard_ids.append(
self.insert_dashboard(
f"title{dashboard_name_index}",
None,
[user.id],
).id
)
self.login(username=user.username)
for dashboard_id in dashboard_ids:
# post succeeds and returns value
allowed_domains = ["test.example", "embedded.example"]
resp = self.post_assert_metric(
f"api/v1/dashboard/{dashboard_id}/embedded",
{"allowed_domains": allowed_domains},
"set_embedded",
)
assert resp.status_code == 200
result = json.loads(resp.data.decode("utf-8"))["result"]
assert result["uuid"] is not None
assert result["uuid"] != ""
assert result["allowed_domains"] == allowed_domains
argument = dashboard_ids
uri = f"api/v1/dashboard/?q={prison.dumps(argument)}"
rv = self.delete_assert_metric(uri, "bulk_delete")
assert rv.status_code == 200
response = json.loads(rv.data.decode("utf-8"))
expected_response = {"message": f"Deleted {dashboard_count} dashboards"}
assert response == expected_response
for dashboard_id in dashboard_ids:
model = db.session.query(Dashboard).get(dashboard_id)
assert model is None
def test_delete_bulk_dashboards_bad_request(self):
"""
Dashboard API: Test delete bulk bad request
"""
dashboard_ids = [1, "a"]
self.login(ADMIN_USERNAME)
argument = dashboard_ids
uri = f"api/v1/dashboard/?q={prison.dumps(argument)}"
rv = self.client.delete(uri)
assert rv.status_code == 400
def test_delete_not_found_dashboard(self):
"""
Dashboard API: Test not found delete
"""
self.login(ADMIN_USERNAME)
dashboard_id = 1000
uri = f"api/v1/dashboard/{dashboard_id}"
rv = self.client.delete(uri)
assert rv.status_code == 404
@pytest.mark.usefixtures("create_dashboard_with_report")
def test_delete_dashboard_with_report(self):
"""
Dashboard API: Test delete with associated report
"""
self.login(ADMIN_USERNAME)
dashboard = (
db.session.query(Dashboard.id)
.filter(Dashboard.dashboard_title == "dashboard_report")
.one_or_none()
)
uri = f"api/v1/dashboard/{dashboard.id}"
rv = self.client.delete(uri)
response = json.loads(rv.data.decode("utf-8"))
assert rv.status_code == 422
expected_response = {
"message": "There are associated alerts or reports: report_with_dashboard"
}
assert response == expected_response
def test_delete_bulk_dashboards_not_found(self):
"""
Dashboard API: Test delete bulk not found
"""
dashboard_ids = [1001, 1002]
self.login(ADMIN_USERNAME)
argument = dashboard_ids
uri = f"api/v1/dashboard/?q={prison.dumps(argument)}"
rv = self.client.delete(uri)
assert rv.status_code == 404
@pytest.mark.usefixtures("create_dashboard_with_report", "create_dashboards")
def test_delete_bulk_dashboard_with_report(self):
"""
Dashboard API: Test bulk delete with associated report
"""
self.login(ADMIN_USERNAME)
dashboard_with_report = (
db.session.query(Dashboard.id)
.filter(Dashboard.dashboard_title == "dashboard_report")
.one_or_none()
)
dashboards = (
db.session.query(Dashboard)
.filter(Dashboard.dashboard_title.like("title%"))
.all()
)
dashboard_ids = [dashboard.id for dashboard in dashboards]
dashboard_ids.append(dashboard_with_report.id)
uri = f"api/v1/dashboard/?q={prison.dumps(dashboard_ids)}"
rv = self.client.delete(uri)
response = json.loads(rv.data.decode("utf-8"))
assert rv.status_code == 422
expected_response = {
"message": "There are associated alerts or reports: report_with_dashboard"
}
assert response == expected_response
def test_delete_dashboard_admin_not_owned(self):
"""
Dashboard API: Test admin delete not owned
"""
gamma_id = self.get_user("gamma").id
dashboard_id = self.insert_dashboard("title", "slug1", [gamma_id]).id
self.login(ADMIN_USERNAME)
uri = f"api/v1/dashboard/{dashboard_id}"
rv = self.client.delete(uri)
assert rv.status_code == 200
model = db.session.query(Dashboard).get(dashboard_id)
assert model is None
def test_delete_bulk_dashboard_admin_not_owned(self):
"""
Dashboard API: Test admin delete bulk not owned
"""
gamma_id = self.get_user("gamma").id
dashboard_count = 4
dashboard_ids = list() # noqa: C408
for dashboard_name_index in range(dashboard_count):
dashboard_ids.append(
self.insert_dashboard(
f"title{dashboard_name_index}",
f"slug{dashboard_name_index}",
[gamma_id],
).id
)
self.login(ADMIN_USERNAME)
argument = dashboard_ids
uri = f"api/v1/dashboard/?q={prison.dumps(argument)}"
rv = self.client.delete(uri)
response = json.loads(rv.data.decode("utf-8"))
assert rv.status_code == 200
expected_response = {"message": f"Deleted {dashboard_count} dashboards"}
assert response == expected_response
for dashboard_id in dashboard_ids:
model = db.session.query(Dashboard).get(dashboard_id)
assert model is None
@pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
def test_delete_dashboard_not_owned(self):
"""
Dashboard API: Test delete try not owned
"""
user_alpha1 = self.create_user(
"alpha1", "password", "Alpha", email="alpha1@superset.org"
)
user_alpha2 = self.create_user(
"alpha2", "password", "Alpha", email="alpha2@superset.org"
)
existing_slice = (
db.session.query(Slice).filter_by(slice_name="Girl Name Cloud").first()
)
dashboard = self.insert_dashboard(
"title", "slug1", [user_alpha1.id], slices=[existing_slice], published=True
)
self.login(username="alpha2", password="password") # noqa: S106
uri = f"api/v1/dashboard/{dashboard.id}"
rv = self.client.delete(uri)
assert rv.status_code == 403
db.session.delete(dashboard)
db.session.delete(user_alpha1)
db.session.delete(user_alpha2)
db.session.commit()
@pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
def test_delete_bulk_dashboard_not_owned(self):
"""
Dashboard API: Test delete bulk try not owned
"""
user_alpha1 = self.create_user(
"alpha1", "password", "Alpha", email="alpha1@superset.org"
)
user_alpha2 = self.create_user(
"alpha2", "password", "Alpha", email="alpha2@superset.org"
)
existing_slice = (
db.session.query(Slice).filter_by(slice_name="Girl Name Cloud").first()
)
dashboard_count = 4
dashboards = list() # noqa: C408
for dashboard_name_index in range(dashboard_count):
dashboards.append(
self.insert_dashboard(
f"title{dashboard_name_index}",
f"slug{dashboard_name_index}",
[user_alpha1.id],
slices=[existing_slice],
published=True,
)
)
owned_dashboard = self.insert_dashboard(
"title_owned",
"slug_owned",
[user_alpha2.id],
slices=[existing_slice],
published=True,
)
self.login(username="alpha2", password="password") # noqa: S106
# verify we can't delete not owned dashboards
arguments = [dashboard.id for dashboard in dashboards]
uri = f"api/v1/dashboard/?q={prison.dumps(arguments)}"
rv = self.client.delete(uri)
assert rv.status_code == 403
response = json.loads(rv.data.decode("utf-8"))
expected_response = {"message": "Forbidden"}
assert response == expected_response
# nothing is deleted in bulk with a list of owned and not owned dashboards
arguments = [dashboard.id for dashboard in dashboards] + [owned_dashboard.id]
uri = f"api/v1/dashboard/?q={prison.dumps(arguments)}"
rv = self.client.delete(uri)
assert rv.status_code == 403
response = json.loads(rv.data.decode("utf-8"))
expected_response = {"message": "Forbidden"}
assert response == expected_response
for dashboard in dashboards:
db.session.delete(dashboard)
db.session.delete(owned_dashboard)
db.session.delete(user_alpha1)
db.session.delete(user_alpha2)
db.session.commit()
def test_create_dashboard(self):
"""
Dashboard API: Test create dashboard
"""
admin_id = self.get_user("admin").id
dashboard_data = {
"dashboard_title": "title1",
"slug": "slug1",
"owners": [admin_id],
"position_json": '{"a": "A"}',
"css": "css",
"json_metadata": '{"refresh_frequency": 30}',
"published": True,
}
self.login(ADMIN_USERNAME)
uri = "api/v1/dashboard/"
rv = self.post_assert_metric(uri, dashboard_data, "post")
assert rv.status_code == 201
data = json.loads(rv.data.decode("utf-8"))
model = db.session.query(Dashboard).get(data.get("id"))
db.session.delete(model)
db.session.commit()
def test_create_simple_dashboard(self):
"""
Dashboard API: Test create simple dashboard
"""
dashboard_data = {"dashboard_title": "title1"}
self.login(ADMIN_USERNAME)
uri = "api/v1/dashboard/"
rv = self.client.post(uri, json=dashboard_data)
assert rv.status_code == 201
data = json.loads(rv.data.decode("utf-8"))
model = db.session.query(Dashboard).get(data.get("id"))
db.session.delete(model)
db.session.commit()
def test_create_dashboard_empty(self):
"""
Dashboard API: Test create empty
"""
dashboard_data = {}
self.login(ADMIN_USERNAME)
uri = "api/v1/dashboard/"
rv = self.client.post(uri, json=dashboard_data)
assert rv.status_code == 201
data = json.loads(rv.data.decode("utf-8"))
model = db.session.query(Dashboard).get(data.get("id"))
db.session.delete(model)
db.session.commit()
dashboard_data = {"dashboard_title": ""}
self.login(ADMIN_USERNAME)
uri = "api/v1/dashboard/"
rv = self.client.post(uri, json=dashboard_data)
assert rv.status_code == 201
data = json.loads(rv.data.decode("utf-8"))
model = db.session.query(Dashboard).get(data.get("id"))
db.session.delete(model)
db.session.commit()
def test_create_dashboard_validate_title(self):
"""
Dashboard API: Test create dashboard validate title
"""
dashboard_data = {"dashboard_title": "a" * 600}
self.login(ADMIN_USERNAME)
uri = "api/v1/dashboard/"
rv = self.post_assert_metric(uri, dashboard_data, "post")
assert rv.status_code == 400
response = json.loads(rv.data.decode("utf-8"))
expected_response = {
"message": {"dashboard_title": ["Length must be between 0 and 500."]}
}
assert response == expected_response
def test_create_dashboard_validate_slug(self):
"""
Dashboard API: Test create validate slug
"""
admin_id = self.get_user("admin").id
dashboard = self.insert_dashboard("title1", "slug1", [admin_id])
self.login(ADMIN_USERNAME)
# Check for slug uniqueness
dashboard_data = {"dashboard_title": "title2", "slug": "slug1"}
uri = "api/v1/dashboard/"
rv = self.client.post(uri, json=dashboard_data)
assert rv.status_code == 422
response = json.loads(rv.data.decode("utf-8"))
expected_response = {"message": {"slug": ["Must be unique"]}}
assert response == expected_response
# Check for slug max size
dashboard_data = {"dashboard_title": "title2", "slug": "a" * 256}
uri = "api/v1/dashboard/"
rv = self.client.post(uri, json=dashboard_data)
assert rv.status_code == 400
response = json.loads(rv.data.decode("utf-8"))
expected_response = {"message": {"slug": ["Length must be between 1 and 255."]}}
assert response == expected_response
db.session.delete(dashboard)
db.session.commit()
def test_create_dashboard_validate_owners(self):
"""
Dashboard API: Test create validate owners
"""
dashboard_data = {"dashboard_title": "title1", "owners": [1000]}
self.login(ADMIN_USERNAME)
uri = "api/v1/dashboard/"
rv = self.client.post(uri, json=dashboard_data)
assert rv.status_code == 422
response = json.loads(rv.data.decode("utf-8"))
expected_response = {"message": {"owners": ["Owners are invalid"]}}
assert response == expected_response
def test_create_dashboard_validate_roles(self):
"""
Dashboard API: Test create validate roles
"""
dashboard_data = {"dashboard_title": "title1", "roles": [1000]}
self.login(ADMIN_USERNAME)
uri = "api/v1/dashboard/"
rv = self.client.post(uri, json=dashboard_data)
assert rv.status_code == 422
response = json.loads(rv.data.decode("utf-8"))
expected_response = {"message": {"roles": ["Some roles do not exist"]}}
assert response == expected_response
def test_create_dashboard_validate_json(self):
"""
Dashboard API: Test create validate json
"""
dashboard_data = {"dashboard_title": "title1", "position_json": '{"A:"a"}'}
self.login(ADMIN_USERNAME)
uri = "api/v1/dashboard/"
rv = self.client.post(uri, json=dashboard_data)
assert rv.status_code == 400
dashboard_data = {"dashboard_title": "title1", "json_metadata": '{"A:"a"}'}
self.login(ADMIN_USERNAME)
uri = "api/v1/dashboard/"
rv = self.client.post(uri, json=dashboard_data)
assert rv.status_code == 400
dashboard_data = {
"dashboard_title": "title1",
"json_metadata": '{"refresh_frequency": "A"}',
}
self.login(ADMIN_USERNAME)
uri = "api/v1/dashboard/"
rv = self.client.post(uri, json=dashboard_data)
assert rv.status_code == 400
def test_update_dashboard(self):
"""
Dashboard API: Test update
"""
admin = self.get_user("admin")
admin_role = self.get_role("Admin")
dashboard_id = self.insert_dashboard(
"title1", "slug1", [admin.id], roles=[admin_role.id]
).id
self.login(ADMIN_USERNAME)
uri = f"api/v1/dashboard/{dashboard_id}"
rv = self.put_assert_metric(uri, self.dashboard_data, "put")
assert rv.status_code == 200
model = db.session.query(Dashboard).get(dashboard_id)
assert model.dashboard_title == self.dashboard_data["dashboard_title"]
assert model.slug == self.dashboard_data["slug"]
assert model.position_json == self.dashboard_data["position_json"]
assert model.css == self.dashboard_data["css"]
assert model.json_metadata == self.dashboard_data["json_metadata"]
assert model.published == self.dashboard_data["published"]
assert model.owners == [admin]
assert model.roles == [admin_role]
db.session.delete(model)
db.session.commit()
def test_add_dashboard_filters(self):
"""
Dashboard API: Test that a filter was added
"""
admin = self.get_user("admin")
admin_role = self.get_role("Admin")
dashboard_id = self.insert_dashboard(
"title1", "slug1", [admin.id], roles=[admin_role.id]
).id
self.login(ADMIN_USERNAME)
uri = f"api/v1/dashboard/{dashboard_id}/filters"
rv = self.put_assert_metric(uri, self.dashboard_put_filters_data, "put_filters")
assert rv.status_code == 200
model = db.session.query(Dashboard).get(dashboard_id)
json_metadata = model.json_metadata
native_filter_config = json.loads(json_metadata)["native_filter_configuration"]
assert native_filter_config[0]["name"] == "Filter 1"
db.session.delete(model)
db.session.commit()
def test_modify_dashboard_filters_values(self):
"""
Dashboard API: Test that a filter was added
"""
admin = self.get_user("admin")
admin_role = self.get_role("Admin")
json_metadata = {
"native_filter_configuration": [
{
"id": "native_filter_1",
"name": "Filter X",
"filterType": "filter_select",
"cascadeParentIds": [],
}
]
}
dashboard_id = self.insert_dashboard(
"title1",
"slug1",
[admin.id],
roles=[admin_role.id],
json_metadata=json.dumps(json_metadata),
).id
self.login(ADMIN_USERNAME)
uri = f"api/v1/dashboard/{dashboard_id}/filters"
rv = self.put_assert_metric(uri, self.dashboard_put_filters_data, "put_filters")
assert rv.status_code == 200
model = db.session.query(Dashboard).get(dashboard_id)
json_metadata = model.json_metadata
native_filter_config = json.loads(json_metadata)["native_filter_configuration"]
assert native_filter_config[0]["name"] == "Filter 1"
db.session.delete(model)
db.session.commit()
def test_modfify_dashboard_filters_order(self):
"""
Dashboard API: Test filters reordered
"""
admin = self.get_user("admin")
admin_role = self.get_role("Admin")
json_metadata = {
"native_filter_configuration": [
{
"id": "native_filter_1",
"name": "Filter 1",
"filterType": "filter_select",
"cascadeParentIds": [],
},
{
"id": "native_filter_2",
"name": "Filter 2",
"filterType": "filter_select",
"cascadeParentIds": [],
},
]
}
dashboard_id = self.insert_dashboard(
"title1",
"slug1",
[admin.id],
roles=[admin_role.id],
json_metadata=json.dumps(json_metadata),
).id
self.login(ADMIN_USERNAME)
uri = f"api/v1/dashboard/{dashboard_id}/filters"
put_data = {
**self.dashboard_put_filters_data,
"reordered": ["native_filter_2", "native_filter_1"],
}
rv = self.put_assert_metric(uri, put_data, "put_filters")
assert rv.status_code == 200
model = db.session.query(Dashboard).get(dashboard_id)
json_metadata = model.json_metadata
native_filter_config = json.loads(json_metadata)["native_filter_configuration"]
assert native_filter_config[0]["name"] == "Filter 2"
db.session.delete(model)
db.session.commit()
def test_dashboard_filters_deleted(self):
"""
Dashboard API: Test filters deleted
"""
admin = self.get_user("admin")
admin_role = self.get_role("Admin")
json_metadata = {
"native_filter_configuration": [
{
"id": "native_filter_1",
"name": "Filter 1",
"filterType": "filter_select",
"cascadeParentIds": [],
},
{
"id": "native_filter_2",
"name": "Filter 2",
"filterType": "filter_select",
"cascadeParentIds": [],
},
]
}
dashboard_id = self.insert_dashboard(
"title1",
"slug1",
[admin.id],
roles=[admin_role.id],
json_metadata=json.dumps(json_metadata),
).id
self.login(ADMIN_USERNAME)
uri = f"api/v1/dashboard/{dashboard_id}/filters"
put_data = {
**self.dashboard_put_filters_data,
"deleted": ["native_filter_1"],
}
rv = self.put_assert_metric(uri, put_data, "put_filters")
assert rv.status_code == 200
model = db.session.query(Dashboard).get(dashboard_id)
json_metadata = model.json_metadata
native_filter_config = json.loads(json_metadata)["native_filter_configuration"]
assert native_filter_config[0]["name"] == "Filter 2"
db.session.delete(model)
db.session.commit()
def test_modify_dashboard_filters_invalid_data(self):
"""
Dashboard API: Test modify filters with invalid data
"""
admin = self.get_user("admin")
admin_role = self.get_role("Admin")
dashboard_id = self.insert_dashboard(
"title1", "slug1", [admin.id], roles=[admin_role.id]
).id
self.login(ADMIN_USERNAME)
uri = f"api/v1/dashboard/{dashboard_id}/filters"
put_data = {"invalid_key": "invalid_value"}
rv = self.put_assert_metric(uri, put_data, "put_filters")
assert rv.status_code == 400
model = db.session.query(Dashboard).get(dashboard_id)
db.session.delete(model)
db.session.commit()
def test_dashboard_get_list_no_username(self):
"""
Dashboard API: Tests that no username is returned
"""
admin = self.get_user("admin")
admin_role = self.get_role("Admin")
dashboard_id = self.insert_dashboard(
"title1", "slug1", [admin.id], roles=[admin_role.id]
).id
model = db.session.query(Dashboard).get(dashboard_id)
self.login(ADMIN_USERNAME)
uri = f"api/v1/dashboard/{dashboard_id}"
dashboard_data = {"dashboard_title": "title2"}
rv = self.client.put(uri, json=dashboard_data)
assert rv.status_code == 200
response = self.get_assert_metric("api/v1/dashboard/", "get_list")
res = json.loads(response.data.decode("utf-8"))["result"]
current_dash = [d for d in res if d["id"] == dashboard_id][0]
assert current_dash["dashboard_title"] == "title2"
assert "username" not in current_dash["changed_by"].keys()
assert "username" not in current_dash["owners"][0].keys()
db.session.delete(model)
db.session.commit()
def test_dashboard_get_no_username(self):
"""
Dashboard API: Tests that no username is returned
"""
admin = self.get_user("admin")
admin_role = self.get_role("Admin")
dashboard_id = self.insert_dashboard(
"title1", "slug1", [admin.id], roles=[admin_role.id]
).id
model = db.session.query(Dashboard).get(dashboard_id)
self.login(ADMIN_USERNAME)
uri = f"api/v1/dashboard/{dashboard_id}"
dashboard_data = {"dashboard_title": "title2"}
rv = self.client.put(uri, json=dashboard_data)
assert rv.status_code == 200
response = self.get_assert_metric(uri, "get")
res = json.loads(response.data.decode("utf-8"))["result"]
assert res["dashboard_title"] == "title2"
assert "username" not in res["changed_by"].keys()
assert "username" not in res["owners"][0].keys()
db.session.delete(model)
db.session.commit()
@pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
def test_update_dashboard_chart_owners_propagation(self):
"""
Dashboard API: Test update chart owners propagation
"""
user_alpha1 = self.create_user(
"alpha1",
"password",
"Alpha",
email="alpha1@superset.org",
first_name="alpha1",
)
admin = self.get_user("admin")
slices = []
slices.append(db.session.query(Slice).filter_by(slice_name="Trends").one())
slices.append(db.session.query(Slice).filter_by(slice_name="Boys").one())
# Insert dashboard with admin as owner
dashboard = self.insert_dashboard(
"title1",
"slug1",
[admin.id],
slices=slices,
)
# Updates dashboard without Boys in json_metadata positions
# and user_alpha1 as owner
dashboard_data = {
"owners": [user_alpha1.id],
"json_metadata": json.dumps(
{
"positions": {
f"{slices[0].id}": {
"type": "CHART",
"meta": {"chartId": slices[0].id},
},
}
}
),
}
self.login(ADMIN_USERNAME)
uri = f"api/v1/dashboard/{dashboard.id}"
rv = self.client.put(uri, json=dashboard_data)
assert rv.status_code == 200
# Check that chart named Boys does not contain alpha 1 in its owners
boys = db.session.query(Slice).filter_by(slice_name="Boys").one()
assert user_alpha1 not in boys.owners
# Revert owners on slice
for slice in slices:
slice.owners = []
db.session.commit()
# Rollback changes
db.session.delete(dashboard)
db.session.delete(user_alpha1)
db.session.commit()
def test_update_partial_dashboard(self):
"""
Dashboard API: Test update partial
"""
admin_id = self.get_user("admin").id
dashboard_id = self.insert_dashboard("title1", "slug1", [admin_id]).id
self.login(ADMIN_USERNAME)
uri = f"api/v1/dashboard/{dashboard_id}"
rv = self.client.put(
uri, json={"json_metadata": self.dashboard_data["json_metadata"]}
)
assert rv.status_code == 200
rv = self.client.put(
uri, json={"dashboard_title": self.dashboard_data["dashboard_title"]}
)
assert rv.status_code == 200
rv = self.client.put(uri, json={"slug": self.dashboard_data["slug"]})
assert rv.status_code == 200
model = db.session.query(Dashboard).get(dashboard_id)
assert model.json_metadata == self.dashboard_data["json_metadata"]
assert model.dashboard_title == self.dashboard_data["dashboard_title"]
assert model.slug == self.dashboard_data["slug"]
db.session.delete(model)
db.session.commit()
def test_update_dashboard_new_owner_not_admin(self):
"""
Dashboard API: Test update set new owner implicitly adds logged in owner
"""
gamma = self.get_user("gamma")
alpha = self.get_user("alpha")
dashboard_id = self.insert_dashboard("title1", "slug1", [alpha.id]).id
dashboard_data = {"dashboard_title": "title1_changed", "owners": [gamma.id]}
self.login(ALPHA_USERNAME)
uri = f"api/v1/dashboard/{dashboard_id}"
rv = self.client.put(uri, json=dashboard_data)
assert rv.status_code == 200
model = db.session.query(Dashboard).get(dashboard_id)
assert gamma in model.owners
assert alpha in model.owners
for slc in model.slices:
assert gamma in slc.owners
assert alpha in slc.owners
db.session.delete(model)
db.session.commit()
def test_update_dashboard_new_owner_admin(self):
"""
Dashboard API: Test update set new owner as admin to other than current user
"""
gamma = self.get_user("gamma")
admin = self.get_user("admin")
dashboard_id = self.insert_dashboard("title1", "slug1", [admin.id]).id
dashboard_data = {"dashboard_title": "title1_changed", "owners": [gamma.id]}
self.login(ADMIN_USERNAME)
uri = f"api/v1/dashboard/{dashboard_id}"
rv = self.client.put(uri, json=dashboard_data)
assert rv.status_code == 200
model = db.session.query(Dashboard).get(dashboard_id)
assert gamma in model.owners
assert admin not in model.owners
for slc in model.slices:
assert gamma in slc.owners
assert admin not in slc.owners
db.session.delete(model)
db.session.commit()
def test_update_dashboard_clear_owner_list(self):
"""
Dashboard API: Test update admin can clear up owners list
"""
admin = self.get_user("admin")
dashboard_id = self.insert_dashboard("title1", "slug1", [admin.id]).id
self.login(username="admin")
uri = f"api/v1/dashboard/{dashboard_id}"
dashboard_data = {"owners": []}
rv = self.client.put(uri, json=dashboard_data)
assert rv.status_code == 200
model = db.session.query(Dashboard).get(dashboard_id)
assert [] == model.owners
db.session.delete(model)
db.session.commit()
def test_update_dashboard_populate_owner(self):
"""
Dashboard API: Test update admin can update dashboard with
no owners to a different owner
"""
self.login(username="admin")
gamma = self.get_user("gamma")
dashboard = self.insert_dashboard(
"title1",
"slug1",
[],
)
uri = f"api/v1/dashboard/{dashboard.id}"
dashboard_data = {"owners": [gamma.id]}
rv = self.client.put(uri, json=dashboard_data)
assert rv.status_code == 200
model = db.session.query(Dashboard).get(dashboard.id)
assert [gamma] == model.owners
db.session.delete(model)
db.session.commit()
def test_update_dashboard_slug_formatting(self):
"""
Dashboard API: Test update slug formatting
"""
admin_id = self.get_user("admin").id
dashboard_id = self.insert_dashboard("title1", "slug1", [admin_id]).id
dashboard_data = {"dashboard_title": "title1_changed", "slug": "slug1 changed"}
self.login(ADMIN_USERNAME)
uri = f"api/v1/dashboard/{dashboard_id}"
rv = self.client.put(uri, json=dashboard_data)
assert rv.status_code == 200
model = db.session.query(Dashboard).get(dashboard_id)
assert model.dashboard_title == "title1_changed"
assert model.slug == "slug1-changed"
db.session.delete(model)
db.session.commit()
def test_update_dashboard_validate_slug(self):
"""
Dashboard API: Test update validate slug
"""
admin_id = self.get_user("admin").id
dashboard1 = self.insert_dashboard("title1", "slug-1", [admin_id])
dashboard2 = self.insert_dashboard("title2", "slug-2", [admin_id])
self.login(ADMIN_USERNAME)
# Check for slug uniqueness
dashboard_data = {"dashboard_title": "title2", "slug": "slug 1"}
uri = f"api/v1/dashboard/{dashboard2.id}"
rv = self.client.put(uri, json=dashboard_data)
assert rv.status_code == 422
response = json.loads(rv.data.decode("utf-8"))
expected_response = {"message": {"slug": ["Must be unique"]}}
assert response == expected_response
db.session.delete(dashboard1)
db.session.delete(dashboard2)
db.session.commit()
dashboard1 = self.insert_dashboard("title1", None, [admin_id])
dashboard2 = self.insert_dashboard("title2", None, [admin_id])
self.login(ADMIN_USERNAME)
# Accept empty slugs and don't validate them has unique
dashboard_data = {"dashboard_title": "title2_changed", "slug": ""}
uri = f"api/v1/dashboard/{dashboard2.id}"
rv = self.client.put(uri, json=dashboard_data)
assert rv.status_code == 200
db.session.delete(dashboard1)
db.session.delete(dashboard2)
db.session.commit()
def test_update_published(self):
"""
Dashboard API: Test update published patch
"""
admin = self.get_user("admin")
gamma = self.get_user("gamma")
dashboard = self.insert_dashboard("title1", "slug1", [admin.id, gamma.id])
dashboard_data = {"published": True}
self.login(ADMIN_USERNAME)
uri = f"api/v1/dashboard/{dashboard.id}"
rv = self.client.put(uri, json=dashboard_data)
assert rv.status_code == 200
model = db.session.query(Dashboard).get(dashboard.id)
assert model.published is True
assert model.slug == "slug1"
assert admin in model.owners
assert gamma in model.owners
db.session.delete(model)
db.session.commit()
@pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
def test_update_dashboard_not_owned(self):
"""
Dashboard API: Test update dashboard not owned
"""
user_alpha1 = self.create_user(
"alpha1", "password", "Alpha", email="alpha1@superset.org"
)
user_alpha2 = self.create_user(
"alpha2", "password", "Alpha", email="alpha2@superset.org"
)
existing_slice = (
db.session.query(Slice).filter_by(slice_name="Girl Name Cloud").first()
)
dashboard = self.insert_dashboard(
"title", "slug1", [user_alpha1.id], slices=[existing_slice], published=True
)
self.login(username="alpha2", password="password") # noqa: S106
dashboard_data = {"dashboard_title": "title1_changed", "slug": "slug1 changed"}
uri = f"api/v1/dashboard/{dashboard.id}"
rv = self.put_assert_metric(uri, dashboard_data, "put")
assert rv.status_code == 403
db.session.delete(dashboard)
db.session.delete(user_alpha1)
db.session.delete(user_alpha2)
db.session.commit()
@pytest.mark.usefixtures(
"load_world_bank_dashboard_with_slices",
"load_birth_names_dashboard_with_slices",
)
@freeze_time("2022-01-01")
def test_export(self):
"""
Dashboard API: Test dashboard export
"""
self.login(ADMIN_USERNAME)
dashboards_ids = get_dashboards_ids(["world_health", "births"])
uri = f"api/v1/dashboard/export/?q={prison.dumps(dashboards_ids)}"
rv = self.get_assert_metric(uri, "export")
headers = "attachment; filename=dashboard_export_20220101T000000.zip" # noqa: F541
assert rv.status_code == 200
assert rv.headers["Content-Disposition"] == headers
def test_export_not_found(self):
"""
Dashboard API: Test dashboard export not found
"""
self.login(ADMIN_USERNAME)
argument = [1000]
uri = f"api/v1/dashboard/export/?q={prison.dumps(argument)}"
rv = self.client.get(uri)
assert rv.status_code == 404
def test_export_not_allowed(self):
"""
Dashboard API: Test dashboard export not allowed
"""
admin_id = self.get_user("admin").id
dashboard = self.insert_dashboard("title", "slug1", [admin_id], published=False)
self.login(GAMMA_USERNAME)
argument = [dashboard.id]
uri = f"api/v1/dashboard/export/?q={prison.dumps(argument)}"
rv = self.client.get(uri)
assert rv.status_code == 404
db.session.delete(dashboard)
db.session.commit()
def test_export_bundle(self):
"""
Dashboard API: Test dashboard export
"""
dashboards_ids = get_dashboards_ids(["world_health", "births"])
uri = f"api/v1/dashboard/export/?q={prison.dumps(dashboards_ids)}"
self.login(ADMIN_USERNAME)
rv = self.client.get(uri)
assert rv.status_code == 200
buf = BytesIO(rv.data)
assert is_zipfile(buf)
def test_export_bundle_not_found(self):
"""
Dashboard API: Test dashboard export not found
"""
self.login(ADMIN_USERNAME)
argument = [1000]
uri = f"api/v1/dashboard/export/?q={prison.dumps(argument)}"
rv = self.client.get(uri)
assert rv.status_code == 404
def test_export_bundle_not_allowed(self):
"""
Dashboard API: Test dashboard export not allowed
"""
admin_id = self.get_user("admin").id
dashboard = self.insert_dashboard("title", "slug1", [admin_id], published=False)
self.login(GAMMA_USERNAME)
argument = [dashboard.id]
uri = f"api/v1/dashboard/export/?q={prison.dumps(argument)}"
rv = self.client.get(uri)
assert rv.status_code == 404
db.session.delete(dashboard)
db.session.commit()
@patch("superset.commands.database.importers.v1.utils.add_permissions")
def test_import_dashboard(self, mock_add_permissions):
"""
Dashboard API: Test import dashboard
"""
self.login(ADMIN_USERNAME)
uri = "api/v1/dashboard/import/"
buf = self.create_dashboard_import()
form_data = {
"formData": (buf, "dashboard_export.zip"),
}
rv = self.client.post(uri, data=form_data, content_type="multipart/form-data")
response = json.loads(rv.data.decode("utf-8"))
assert rv.status_code == 200
assert response == {"message": "OK"}
dashboard = (
db.session.query(Dashboard).filter_by(uuid=dashboard_config["uuid"]).one()
)
assert dashboard.dashboard_title == "Test dash"
assert len(dashboard.slices) == 1
chart = dashboard.slices[0]
assert str(chart.uuid) == chart_config["uuid"]
dataset = chart.table
assert str(dataset.uuid) == dataset_config["uuid"]
database = dataset.database
assert str(database.uuid) == database_config["uuid"]
db.session.delete(dashboard)
db.session.delete(chart)
db.session.delete(dataset)
db.session.delete(database)
db.session.commit()
def test_import_dashboard_invalid_file(self):
"""
Dashboard API: Test import invalid dashboard file
"""
self.login(ADMIN_USERNAME)
uri = "api/v1/dashboard/import/"
buf = self.create_invalid_dashboard_import()
form_data = {
"formData": (buf, "dashboard_export.zip"),
}
rv = self.client.post(uri, data=form_data, content_type="multipart/form-data")
response = json.loads(rv.data.decode("utf-8"))
assert rv.status_code == 400
assert response == {
"errors": [
{
"message": "No valid import files were found",
"error_type": "GENERIC_COMMAND_ERROR",
"level": "warning",
"extra": {
"issue_codes": [
{
"code": 1010,
"message": (
"Issue 1010 - Superset encountered an "
"error while running a command."
),
}
]
},
}
]
}
def test_import_dashboard_v0_export(self):
num_dashboards = db.session.query(Dashboard).count()
self.login(ADMIN_USERNAME)
uri = "api/v1/dashboard/import/"
buf = BytesIO()
buf.write(json.dumps(dashboard_export).encode())
buf.seek(0)
form_data = {
"formData": (buf, "20201119_181105.json"),
}
rv = self.client.post(uri, data=form_data, content_type="multipart/form-data")
response = json.loads(rv.data.decode("utf-8"))
assert rv.status_code == 200
assert response == {"message": "OK"}
assert db.session.query(Dashboard).count() == num_dashboards + 1
dashboard = (
db.session.query(Dashboard).filter_by(dashboard_title="Births 2").one()
)
chart = dashboard.slices[0]
dataset = chart.table
db.session.delete(dashboard)
db.session.delete(chart)
db.session.delete(dataset)
db.session.commit()
@patch("superset.commands.database.importers.v1.utils.add_permissions")
def test_import_dashboard_overwrite(self, mock_add_permissions):
"""
Dashboard API: Test import existing dashboard
"""
self.login(ADMIN_USERNAME)
uri = "api/v1/dashboard/import/"
buf = self.create_dashboard_import()
form_data = {
"formData": (buf, "dashboard_export.zip"),
}
rv = self.client.post(uri, data=form_data, content_type="multipart/form-data")
response = json.loads(rv.data.decode("utf-8"))
assert rv.status_code == 200
assert response == {"message": "OK"}
# import again without overwrite flag
buf = self.create_dashboard_import()
form_data = {
"formData": (buf, "dashboard_export.zip"),
}
rv = self.client.post(uri, data=form_data, content_type="multipart/form-data")
response = json.loads(rv.data.decode("utf-8"))
assert rv.status_code == 422
assert (
response
== {
"errors": [
{
"message": "Error importing dashboard",
"error_type": "GENERIC_COMMAND_ERROR",
"level": "warning",
"extra": {
"dashboards/imported_dashboard.yaml": "Dashboard already exists and `overwrite=true` was not passed", # noqa: E501
"issue_codes": [
{
"code": 1010,
"message": (
"Issue 1010 - Superset encountered an "
"error while running a command."
),
}
],
},
}
]
}
)
# import with overwrite flag
buf = self.create_dashboard_import()
form_data = {
"formData": (buf, "dashboard_export.zip"),
"overwrite": "true",
}
rv = self.client.post(uri, data=form_data, content_type="multipart/form-data")
response = json.loads(rv.data.decode("utf-8"))
assert rv.status_code == 200
assert response == {"message": "OK"}
# cleanup
dashboard = (
db.session.query(Dashboard).filter_by(uuid=dashboard_config["uuid"]).one()
)
chart = dashboard.slices[0]
dataset = chart.table
database = dataset.database
db.session.delete(dashboard)
db.session.delete(chart)
db.session.delete(dataset)
db.session.delete(database)
db.session.commit()
def test_import_dashboard_invalid(self):
"""
Dashboard API: Test import invalid dashboard
"""
self.login(ADMIN_USERNAME)
uri = "api/v1/dashboard/import/"
buf = BytesIO()
with ZipFile(buf, "w") as bundle:
with bundle.open("dashboard_export/metadata.yaml", "w") as fp:
fp.write(yaml.safe_dump(dataset_metadata_config).encode())
with bundle.open(
"dashboard_export/databases/imported_database.yaml", "w"
) as fp:
fp.write(yaml.safe_dump(database_config).encode())
with bundle.open(
"dashboard_export/datasets/imported_dataset.yaml", "w"
) as fp:
fp.write(yaml.safe_dump(dataset_config).encode())
with bundle.open("dashboard_export/charts/imported_chart.yaml", "w") as fp:
fp.write(yaml.safe_dump(chart_config).encode())
with bundle.open(
"dashboard_export/dashboards/imported_dashboard.yaml", "w"
) as fp:
fp.write(yaml.safe_dump(dashboard_config).encode())
buf.seek(0)
form_data = {
"formData": (buf, "dashboard_export.zip"),
}
rv = self.client.post(uri, data=form_data, content_type="multipart/form-data")
response = json.loads(rv.data.decode("utf-8"))
assert rv.status_code == 422
assert response == {
"errors": [
{
"message": "Error importing dashboard",
"error_type": "GENERIC_COMMAND_ERROR",
"level": "warning",
"extra": {
"metadata.yaml": {"type": ["Must be equal to Dashboard."]},
"issue_codes": [
{
"code": 1010,
"message": (
"Issue 1010 - Superset encountered "
"an error while running a command."
),
}
],
},
}
]
}
def test_get_all_related_roles(self):
"""
API: Test get filter related roles
"""
self.login(ADMIN_USERNAME)
uri = "api/v1/dashboard/related/roles" # noqa: F541
rv = self.client.get(uri)
assert rv.status_code == 200
response = json.loads(rv.data.decode("utf-8"))
roles = db.session.query(security_manager.role_model).all()
expected_roles = [str(role) for role in roles]
assert response["count"] == len(roles)
response_roles = [result["text"] for result in response["result"]]
for expected_role in expected_roles:
assert expected_role in response_roles
def test_get_filter_related_roles(self):
"""
API: Test get filter related roles
"""
self.login(ADMIN_USERNAME)
argument = {"filter": "alpha"}
uri = f"api/v1/dashboard/related/roles?q={prison.dumps(argument)}"
rv = self.client.get(uri)
assert rv.status_code == 200
response = json.loads(rv.data.decode("utf-8"))
assert response["count"] == 1
response_roles = [result["text"] for result in response["result"]]
assert "Alpha" in response_roles
def test_get_all_related_roles_with_with_extra_filters(self):
"""
API: Test get filter related roles with extra related query filters
"""
self.login(ADMIN_USERNAME)
def _base_filter(query):
return query.filter_by(name="Alpha")
with patch.dict(
"superset.views.filters.current_app.config",
{"EXTRA_RELATED_QUERY_FILTERS": {"role": _base_filter}},
):
uri = "api/v1/dashboard/related/roles" # noqa: F541
rv = self.client.get(uri)
assert rv.status_code == 200
response = json.loads(rv.data.decode("utf-8"))
response_roles = [result["text"] for result in response["result"]]
assert response_roles == ["Alpha"]
@pytest.mark.usefixtures("load_world_bank_dashboard_with_slices")
def test_embedded_dashboards(self):
self.login(ADMIN_USERNAME)
uri = "api/v1/dashboard/world_health/embedded"
# initial get should return 404
resp = self.get_assert_metric(uri, "get_embedded")
assert resp.status_code == 404
# post succeeds and returns value
allowed_domains = ["test.example", "embedded.example"]
resp = self.post_assert_metric(
uri,
{"allowed_domains": allowed_domains},
"set_embedded",
)
assert resp.status_code == 200
result = json.loads(resp.data.decode("utf-8"))["result"]
assert result["uuid"] is not None
assert result["uuid"] != ""
assert result["allowed_domains"] == allowed_domains
# get returns value
resp = self.get_assert_metric(uri, "get_embedded")
assert resp.status_code == 200
result = json.loads(resp.data.decode("utf-8"))["result"]
assert result["uuid"] is not None
assert result["uuid"] != ""
assert result["allowed_domains"] == allowed_domains
# save uuid for later
original_uuid = result["uuid"]
# put succeeds and returns value
resp = self.post_assert_metric(uri, {"allowed_domains": []}, "set_embedded")
assert resp.status_code == 200
result = json.loads(resp.data.decode("utf-8"))["result"]
assert resp.status_code == 200
assert result["uuid"] is not None
assert result["uuid"] != ""
assert result["allowed_domains"] == []
# get returns changed value
resp = self.get_assert_metric(uri, "get_embedded")
assert resp.status_code == 200
result = json.loads(resp.data.decode("utf-8"))["result"]
assert result["uuid"] == original_uuid
assert result["allowed_domains"] == []
# delete succeeds
resp = self.delete_assert_metric(uri, "delete_embedded")
assert resp.status_code == 200
# get returns 404
resp = self.get_assert_metric(uri, "get_embedded")
assert resp.status_code == 404
@pytest.mark.usefixtures("create_created_by_gamma_dashboards")
def test_gets_created_by_user_dashboards_filter(self):
expected_models = (
db.session.query(Dashboard)
.filter(Dashboard.created_by_fk.isnot(None))
.all()
)
arguments = {
"filters": [
{"col": "created_by", "opr": "dashboard_has_created_by", "value": True}
],
"keys": ["none"],
"columns": ["dashboard_title"],
}
self.login(ADMIN_USERNAME)
uri = f"api/v1/dashboard/?q={prison.dumps(arguments)}"
rv = self.get_assert_metric(uri, "get_list")
assert rv.status_code == 200
data = json.loads(rv.data.decode("utf-8"))
assert data["count"] == len(expected_models)
def test_gets_not_created_by_user_dashboards_filter(self):
dashboard = self.insert_dashboard("title", "slug", []) # noqa: F541
expected_models = (
db.session.query(Dashboard).filter(Dashboard.created_by_fk.is_(None)).all()
)
arguments = {
"filters": [
{"col": "created_by", "opr": "dashboard_has_created_by", "value": False}
],
"keys": ["none"],
"columns": ["dashboard_title"],
}
self.login(ADMIN_USERNAME)
uri = f"api/v1/dashboard/?q={prison.dumps(arguments)}"
rv = self.get_assert_metric(uri, "get_list")
assert rv.status_code == 200
data = json.loads(rv.data.decode("utf-8"))
assert data["count"] == len(expected_models)
db.session.delete(dashboard)
db.session.commit()
@pytest.mark.usefixtures("load_world_bank_dashboard_with_slices")
def test_copy_dashboard(self):
self.login(ADMIN_USERNAME)
original_dash = (
db.session.query(Dashboard).filter_by(slug="world_health").first()
)
data = {
"dashboard_title": "copied dash",
"css": "<css>",
"duplicate_slices": False,
"json_metadata": json.dumps(
{
"positions": original_dash.position,
"color_namespace": "Color Namespace Test",
"color_scheme": "Color Scheme Test",
}
),
}
pk = original_dash.id
uri = f"api/v1/dashboard/{pk}/copy/"
rv = self.client.post(uri, json=data)
assert rv.status_code == 200
response = json.loads(rv.data.decode("utf-8"))
assert response == {"result": {"id": ANY, "last_modified_time": ANY}}
dash = (
db.session.query(Dashboard)
.filter(Dashboard.id == response["result"]["id"])
.one()
)
assert dash.id != original_dash.id
assert len(dash.position) == len(original_dash.position)
assert dash.dashboard_title == "copied dash"
assert dash.css == "<css>"
assert dash.owners == [security_manager.find_user("admin")]
self.assertCountEqual(dash.slices, original_dash.slices) # noqa: PT009
assert dash.params_dict["color_namespace"] == "Color Namespace Test"
assert dash.params_dict["color_scheme"] == "Color Scheme Test"
db.session.delete(dash)
db.session.commit()
@pytest.mark.usefixtures("load_world_bank_dashboard_with_slices")
def test_copy_dashboard_duplicate_slices(self):
self.login(ADMIN_USERNAME)
original_dash = (
db.session.query(Dashboard).filter_by(slug="world_health").first()
)
data = {
"dashboard_title": "copied dash",
"css": "<css>",
"duplicate_slices": True,
"json_metadata": json.dumps(
{
"positions": original_dash.position,
"color_namespace": "Color Namespace Test",
"color_scheme": "Color Scheme Test",
}
),
}
pk = original_dash.id
uri = f"api/v1/dashboard/{pk}/copy/"
rv = self.client.post(uri, json=data)
assert rv.status_code == 200
response = json.loads(rv.data.decode("utf-8"))
assert response == {"result": {"id": ANY, "last_modified_time": ANY}}
dash = (
db.session.query(Dashboard)
.filter(Dashboard.id == response["result"]["id"])
.one()
)
assert dash.id != original_dash.id
assert len(dash.position) == len(original_dash.position)
assert dash.dashboard_title == "copied dash"
assert dash.css == "<css>"
assert dash.owners == [security_manager.find_user("admin")]
assert dash.params_dict["color_namespace"] == "Color Namespace Test"
assert dash.params_dict["color_scheme"] == "Color Scheme Test"
assert len(dash.slices) == len(original_dash.slices)
for original_slc in original_dash.slices:
for slc in dash.slices:
assert slc.id != original_slc.id
for slc in dash.slices:
db.session.delete(slc)
db.session.delete(dash)
db.session.commit()
@pytest.mark.usefixtures("create_dashboard_with_tag")
def test_update_dashboard_add_tags_can_write_on_tag(self):
"""
Validates a user with can write on tag permission can
add tags while updating a dashboard
"""
self.login(ADMIN_USERNAME)
dashboard = (
db.session.query(Dashboard)
.filter(Dashboard.dashboard_title == "dash with tag")
.first()
)
new_tag = db.session.query(Tag).filter(Tag.name == "second_tag").one()
# get existing tag and add a new one
new_tags = [tag.id for tag in dashboard.tags if tag.type == TagType.custom]
new_tags.append(new_tag.id)
update_payload = {"tags": new_tags}
uri = f"api/v1/dashboard/{dashboard.id}"
rv = self.put_assert_metric(uri, update_payload, "put")
assert rv.status_code == 200
model = db.session.query(Dashboard).get(dashboard.id)
# Clean up system tags
tag_list = [tag.id for tag in model.tags if tag.type == TagType.custom]
assert sorted(tag_list) == sorted(new_tags)
@pytest.mark.usefixtures("create_dashboard_with_tag")
def test_update_dashboard_remove_tags_can_write_on_tag(self):
"""
Validates a user with can write on tag permission can
remove tags while updating a dashboard
"""
self.login(ADMIN_USERNAME)
dashboard = (
db.session.query(Dashboard)
.filter(Dashboard.dashboard_title == "dash with tag")
.first()
)
# get existing tag and add a new one
new_tags = [tag.id for tag in dashboard.tags if tag.type == TagType.custom]
new_tags.pop()
update_payload = {"tags": new_tags}
uri = f"api/v1/dashboard/{dashboard.id}"
rv = self.put_assert_metric(uri, update_payload, "put")
assert rv.status_code == 200
model = db.session.query(Dashboard).get(dashboard.id)
# Clean up system tags
tag_list = [tag.id for tag in model.tags if tag.type == TagType.custom]
assert tag_list == new_tags
@pytest.mark.usefixtures("create_dashboard_with_tag")
def test_update_dashboard_add_tags_can_tag_on_dashboard(self):
"""
Validates an owner with can tag on dashboard permission can
add tags while updating a dashboard
"""
self.login(GAMMA_USERNAME)
write_tags_perm = security_manager.add_permission_view_menu("can_write", "Tag")
gamma_role = security_manager.find_role("Gamma")
security_manager.del_permission_role(gamma_role, write_tags_perm)
assert "can tag on Dashboard" in str(gamma_role.permissions)
dashboard = (
db.session.query(Dashboard)
.filter(Dashboard.dashboard_title == "dash with tag")
.first()
)
new_tag = db.session.query(Tag).filter(Tag.name == "second_tag").one()
# get existing tag and add a new one
new_tags = [tag.id for tag in dashboard.tags if tag.type == TagType.custom]
new_tags.append(new_tag.id)
update_payload = {"tags": new_tags}
uri = f"api/v1/dashboard/{dashboard.id}"
rv = self.put_assert_metric(uri, update_payload, "put")
assert rv.status_code == 200
model = db.session.query(Dashboard).get(dashboard.id)
# Clean up system tags
tag_list = [tag.id for tag in model.tags if tag.type == TagType.custom]
assert sorted(tag_list) == sorted(new_tags)
security_manager.add_permission_role(gamma_role, write_tags_perm)
@pytest.mark.usefixtures("create_dashboard_with_tag")
def test_update_dashboard_remove_tags_can_tag_on_dashboard(self):
"""
Validates an owner with can tag on dashboard permission can
remove tags from a dashboard
"""
self.login(GAMMA_USERNAME)
write_tags_perm = security_manager.add_permission_view_menu("can_write", "Tag")
gamma_role = security_manager.find_role("Gamma")
security_manager.del_permission_role(gamma_role, write_tags_perm)
assert "can tag on Dashboard" in str(gamma_role.permissions)
dashboard = (
db.session.query(Dashboard)
.filter(Dashboard.dashboard_title == "dash with tag")
.first()
)
update_payload = {"tags": []}
uri = f"api/v1/dashboard/{dashboard.id}"
rv = self.put_assert_metric(uri, update_payload, "put")
assert rv.status_code == 200
model = db.session.query(Dashboard).get(dashboard.id)
# Clean up system tags
tag_list = [tag.id for tag in model.tags if tag.type == TagType.custom]
assert tag_list == []
security_manager.add_permission_role(gamma_role, write_tags_perm)
@pytest.mark.usefixtures("create_dashboard_with_tag")
def test_update_dashboard_add_tags_missing_permission(self):
"""
Validates an owner can't add tags to a dashboard if they don't
have permission to it
"""
self.login(GAMMA_USERNAME)
write_tags_perm = security_manager.add_permission_view_menu("can_write", "Tag")
tag_dashboards_perm = security_manager.add_permission_view_menu(
"can_tag", "Dashboard"
)
gamma_role = security_manager.find_role("Gamma")
security_manager.del_permission_role(gamma_role, write_tags_perm)
security_manager.del_permission_role(gamma_role, tag_dashboards_perm)
dashboard = (
db.session.query(Dashboard)
.filter(Dashboard.dashboard_title == "dash with tag")
.first()
)
new_tag = db.session.query(Tag).filter(Tag.name == "second_tag").one()
# get existing tag and add a new one
new_tags = [tag.id for tag in dashboard.tags if tag.type == TagType.custom]
new_tags.append(new_tag.id)
update_payload = {"tags": new_tags}
uri = f"api/v1/dashboard/{dashboard.id}"
rv = self.put_assert_metric(uri, update_payload, "put")
assert rv.status_code == 403
assert (
rv.json["message"]
== "You do not have permission to manage tags on dashboards"
)
security_manager.add_permission_role(gamma_role, write_tags_perm)
security_manager.add_permission_role(gamma_role, tag_dashboards_perm)
@pytest.mark.usefixtures("create_dashboard_with_tag")
def test_update_dashboard_remove_tags_missing_permission(self):
"""
Validates an owner can't remove tags from a dashboard if they don't
have permission to it
"""
self.login(GAMMA_USERNAME)
write_tags_perm = security_manager.add_permission_view_menu("can_write", "Tag")
tag_dashboards_perm = security_manager.add_permission_view_menu(
"can_tag", "Dashboard"
)
gamma_role = security_manager.find_role("Gamma")
security_manager.del_permission_role(gamma_role, write_tags_perm)
security_manager.del_permission_role(gamma_role, tag_dashboards_perm)
dashboard = (
db.session.query(Dashboard)
.filter(Dashboard.dashboard_title == "dash with tag")
.first()
)
update_payload = {"tags": []}
uri = f"api/v1/dashboard/{dashboard.id}"
rv = self.put_assert_metric(uri, update_payload, "put")
assert rv.status_code == 403
assert (
rv.json["message"]
== "You do not have permission to manage tags on dashboards"
)
security_manager.add_permission_role(gamma_role, write_tags_perm)
security_manager.add_permission_role(gamma_role, tag_dashboards_perm)
@pytest.mark.usefixtures("create_dashboard_with_tag")
def test_update_dashboard_no_tag_changes(self):
"""
Validates an owner without permission to change tags is able to
update a dashboard when tags haven't changed
"""
self.login(GAMMA_USERNAME)
write_tags_perm = security_manager.add_permission_view_menu("can_write", "Tag")
tag_dashboards_perm = security_manager.add_permission_view_menu(
"can_tag", "Dashboard"
)
gamma_role = security_manager.find_role("Gamma")
security_manager.del_permission_role(gamma_role, write_tags_perm)
security_manager.del_permission_role(gamma_role, tag_dashboards_perm)
dashboard = (
db.session.query(Dashboard)
.filter(Dashboard.dashboard_title == "dash with tag")
.first()
)
existing_tags = [tag.id for tag in dashboard.tags if tag.type == TagType.custom]
update_payload = {"tags": existing_tags}
uri = f"api/v1/dashboard/{dashboard.id}"
rv = self.put_assert_metric(uri, update_payload, "put")
assert rv.status_code == 200
security_manager.add_permission_role(gamma_role, write_tags_perm)
security_manager.add_permission_role(gamma_role, tag_dashboards_perm)
def _cache_screenshot(self, dashboard_id, payload=None):
if payload is None:
payload = {"dataMask": {}, "activeTabs": [], "anchor": "", "urlParams": []}
uri = f"/api/v1/dashboard/{dashboard_id}/cache_dashboard_screenshot/"
return self.client.post(uri, json=payload)
def _get_screenshot(self, dashboard_id, cache_key, download_format):
uri = f"/api/v1/dashboard/{dashboard_id}/screenshot/{cache_key}/?download_format={download_format}" # noqa: E501
return self.client.get(uri)
@with_feature_flags(THUMBNAILS=True, ENABLE_DASHBOARD_SCREENSHOT_ENDPOINTS=True)
@pytest.mark.usefixtures("create_dashboard_with_tag")
def test_cache_dashboard_screenshot_success(self):
self.login(ADMIN_USERNAME)
dashboard = (
db.session.query(Dashboard)
.filter(Dashboard.dashboard_title == "dash with tag")
.first()
)
response = self._cache_screenshot(dashboard.id)
assert response.status_code == 202
@with_feature_flags(THUMBNAILS=True, ENABLE_DASHBOARD_SCREENSHOT_ENDPOINTS=True)
@pytest.mark.usefixtures("create_dashboard_with_tag")
def test_cache_dashboard_screenshot_dashboard_validation(self):
self.login(ADMIN_USERNAME)
dashboard = (
db.session.query(Dashboard)
.filter(Dashboard.dashboard_title == "dash with tag")
.first()
)
invalid_payload = {
"dataMask": ["should be a dict"],
"activeTabs": "should be a list",
"anchor": 1,
"urlParams": "should be a list",
}
response = self._cache_screenshot(dashboard.id, invalid_payload)
assert response.status_code == 400
@with_feature_flags(THUMBNAILS=True, ENABLE_DASHBOARD_SCREENSHOT_ENDPOINTS=True)
def test_cache_dashboard_screenshot_dashboard_not_found(self):
self.login(ADMIN_USERNAME)
non_existent_id = 999
response = self._cache_screenshot(non_existent_id)
assert response.status_code == 404
@with_feature_flags(THUMBNAILS=True, ENABLE_DASHBOARD_SCREENSHOT_ENDPOINTS=True)
@pytest.mark.usefixtures("create_dashboard_with_tag")
@patch("superset.dashboards.api.cache_dashboard_screenshot")
@patch("superset.dashboards.api.DashboardScreenshot.get_from_cache_key")
def test_screenshot_success_png(self, mock_get_cache, mock_cache_task):
"""
Validate screenshot returns png
"""
self.login(ADMIN_USERNAME)
mock_cache_task.return_value = None
mock_get_cache.return_value = BytesIO(b"fake image data")
dashboard = (
db.session.query(Dashboard)
.filter(Dashboard.dashboard_title == "dash with tag")
.first()
)
cache_resp = self._cache_screenshot(dashboard.id)
assert cache_resp.status_code == 202
cache_key = json.loads(cache_resp.data.decode("utf-8"))["cache_key"]
response = self._get_screenshot(dashboard.id, cache_key, "png")
assert response.status_code == 200
assert response.mimetype == "image/png"
assert response.data == b"fake image data"
@with_feature_flags(THUMBNAILS=True, ENABLE_DASHBOARD_SCREENSHOT_ENDPOINTS=True)
@pytest.mark.usefixtures("create_dashboard_with_tag")
@patch("superset.dashboards.api.cache_dashboard_screenshot")
@patch("superset.dashboards.api.build_pdf_from_screenshots")
@patch("superset.dashboards.api.DashboardScreenshot.get_from_cache_key")
def test_screenshot_success_pdf(
self, mock_get_from_cache, mock_build_pdf, mock_cache_task
):
"""
Validate screenshot can return pdf.
"""
self.login(ADMIN_USERNAME)
mock_cache_task.return_value = None
mock_get_from_cache.return_value = BytesIO(b"fake image data")
mock_build_pdf.return_value = b"fake pdf data"
dashboard = (
db.session.query(Dashboard)
.filter(Dashboard.dashboard_title == "dash with tag")
.first()
)
cache_resp = self._cache_screenshot(dashboard.id)
assert cache_resp.status_code == 202
cache_key = json.loads(cache_resp.data.decode("utf-8"))["cache_key"]
response = self._get_screenshot(dashboard.id, cache_key, "pdf")
assert response.status_code == 200
assert response.mimetype == "application/pdf"
assert response.data == b"fake pdf data"
@with_feature_flags(THUMBNAILS=True, ENABLE_DASHBOARD_SCREENSHOT_ENDPOINTS=True)
@pytest.mark.usefixtures("create_dashboard_with_tag")
@patch("superset.dashboards.api.cache_dashboard_screenshot")
@patch("superset.dashboards.api.DashboardScreenshot.get_from_cache_key")
def test_screenshot_not_in_cache(self, mock_get_cache, mock_cache_task):
self.login(ADMIN_USERNAME)
mock_cache_task.return_value = None
mock_get_cache.return_value = None
dashboard = (
db.session.query(Dashboard)
.filter(Dashboard.dashboard_title == "dash with tag")
.first()
)
cache_resp = self._cache_screenshot(dashboard.id)
assert cache_resp.status_code == 202
cache_key = json.loads(cache_resp.data.decode("utf-8"))["cache_key"]
response = self._get_screenshot(dashboard.id, cache_key, "pdf")
assert response.status_code == 404
@with_feature_flags(THUMBNAILS=True, ENABLE_DASHBOARD_SCREENSHOT_ENDPOINTS=True)
def test_screenshot_dashboard_not_found(self):
self.login(ADMIN_USERNAME)
non_existent_id = 999
response = self._get_screenshot(non_existent_id, "some_cache_key", "png")
assert response.status_code == 404
@with_feature_flags(THUMBNAILS=True, ENABLE_DASHBOARD_SCREENSHOT_ENDPOINTS=True)
@pytest.mark.usefixtures("create_dashboard_with_tag")
@patch("superset.dashboards.api.cache_dashboard_screenshot")
@patch("superset.dashboards.api.DashboardScreenshot.get_from_cache_key")
def test_screenshot_invalid_download_format(self, mock_get_cache, mock_cache_task):
self.login(ADMIN_USERNAME)
mock_cache_task.return_value = None
mock_get_cache.return_value = BytesIO(b"fake png data")
dashboard = (
db.session.query(Dashboard)
.filter(Dashboard.dashboard_title == "dash with tag")
.first()
)
cache_resp = self._cache_screenshot(dashboard.id)
assert cache_resp.status_code == 202
cache_key = json.loads(cache_resp.data.decode("utf-8"))["cache_key"]
response = self._get_screenshot(dashboard.id, cache_key, "invalid")
assert response.status_code == 404
@with_feature_flags(THUMBNAILS=False, ENABLE_DASHBOARD_SCREENSHOT_ENDPOINTS=True)
@pytest.mark.usefixtures("create_dashboard_with_tag")
def test_cache_dashboard_screenshot_feature_thumbnails_ff_disabled(self):
self.login(ADMIN_USERNAME)
dashboard = (
db.session.query(Dashboard)
.filter(Dashboard.dashboard_title == "dash with tag")
.first()
)
assert dashboard is not None
response = self._cache_screenshot(dashboard.id)
assert response.status_code == 404
@with_feature_flags(THUMBNAILS=True, ENABLE_DASHBOARD_SCREENSHOT_ENDPOINTS=False)
@pytest.mark.usefixtures("create_dashboard_with_tag")
def test_cache_dashboard_screenshot_feature_screenshot_ff_disabled(self):
self.login(ADMIN_USERNAME)
dashboard = (
db.session.query(Dashboard)
.filter(Dashboard.dashboard_title == "dash with tag")
.first()
)
assert dashboard is not None
response = self._cache_screenshot(dashboard.id)
assert response.status_code == 404
@with_feature_flags(THUMBNAILS=False, ENABLE_DASHBOARD_SCREENSHOT_ENDPOINTS=False)
@pytest.mark.usefixtures("create_dashboard_with_tag")
def test_cache_dashboard_screenshot_feature_both_ff_disabled(self):
self.login(ADMIN_USERNAME)
dashboard = (
db.session.query(Dashboard)
.filter(Dashboard.dashboard_title == "dash with tag")
.first()
)
assert dashboard is not None
response = self._cache_screenshot(dashboard.id)
assert response.status_code == 404
@pytest.mark.usefixtures("load_world_bank_dashboard_with_slices")
def test_put_dashboard_colors(self):
"""
Dashboard API: Test updating dashboard colors
"""
self.login(ADMIN_USERNAME)
dashboard = Dashboard.get("world_health")
colors = {
"label_colors": {"Sales": "#FF0000", "Profit": "#00FF00"},
"shared_label_colors": ["#0000FF", "#FFFF00"],
"map_label_colors": {"Revenue": "#FFFFFF"},
"color_scheme": "d3Category10",
}
uri = f"api/v1/dashboard/{dashboard.id}/colors"
rv = self.client.put(uri, json=colors)
assert rv.status_code == 200
updated_dashboard = db.session.query(Dashboard).get(dashboard.id)
updated_label_colors = json.loads(updated_dashboard.json_metadata).get(
"label_colors"
)
updated_shared_label_colors = json.loads(updated_dashboard.json_metadata).get(
"shared_label_colors"
)
updated_map_label_colors = json.loads(updated_dashboard.json_metadata).get(
"map_label_colors"
)
updated_color_scheme = json.loads(updated_dashboard.json_metadata).get(
"color_scheme"
)
assert updated_label_colors == colors["label_colors"]
assert updated_shared_label_colors == colors["shared_label_colors"]
assert updated_map_label_colors == colors["map_label_colors"]
assert updated_color_scheme == colors["color_scheme"]
@pytest.mark.usefixtures("load_world_bank_dashboard_with_slices")
def test_put_dashboard_colors_no_mark_updated(self):
"""
Dashboard API: Test updating dashboard colors without marking the dashboard as updated
""" # noqa: E501
self.login(ADMIN_USERNAME)
dashboard = Dashboard.get("world_health")
colors = {"color_scheme": "d3Category10"}
previous_changed_on = dashboard.changed_on
uri = f"api/v1/dashboard/{dashboard.id}/colors?mark_updated=false"
rv = self.client.put(uri, json=colors)
assert rv.status_code == 200
updated_dashboard = db.session.query(Dashboard).get(dashboard.id)
updated_color_scheme = json.loads(updated_dashboard.json_metadata).get(
"color_scheme"
)
assert updated_color_scheme == colors["color_scheme"]
assert updated_dashboard.changed_on == previous_changed_on
def test_put_dashboard_colors_not_found(self):
"""
Dashboard API: Test updating colors for dashboard that does not exist
"""
self.login(ADMIN_USERNAME)
colors = {"label_colors": {"Sales": "#FF0000"}}
invalid_id = self.get_nonexistent_numeric_id(Dashboard)
uri = f"api/v1/dashboard/{invalid_id}/colors"
rv = self.client.put(uri, json=colors)
assert rv.status_code == 404
@pytest.mark.usefixtures("load_world_bank_dashboard_with_slices")
def test_put_dashboard_colors_invalid(self):
"""
Dashboard API: Test updating dashboard colors with invalid color format
"""
self.login(ADMIN_USERNAME)
dashboard = Dashboard.get("world_health")
colors = {"test_invalid_prop": {"Sales": "invalid"}}
uri = f"api/v1/dashboard/{dashboard.id}/colors"
rv = self.client.put(uri, json=colors)
assert rv.status_code == 400
def test_put_dashboard_colors_not_authorized(self):
"""
Dashboard API: Test updating colors without authorization
"""
with self.create_app().app_context():
admin = security_manager.find_user("admin")
dashboard = self.insert_dashboard("title", None, [admin.id])
assert dashboard.id is not None
colors = {"label_colors": {"Sales": "#FF0000"}}
self.login(GAMMA_USERNAME)
uri = f"api/v1/dashboard/{dashboard.id}/colors"
rv = self.client.put(uri, json=colors)
assert rv.status_code == 403
yield dashboard
# Cleanup
db.session.delete(dashboard)
db.session.commit()