api_tests.py

# 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.
"""Unit tests for Superset"""

from io import BytesIO
from unittest import mock
from unittest.mock import patch
from zipfile import is_zipfile, ZipFile

import prison
import pytest
import yaml
from flask_babel import lazy_gettext as _
from parameterized import parameterized
from sqlalchemy import and_
from sqlalchemy.sql import func

from superset.commands.chart.data.get_data_command import ChartDataCommand
from superset.commands.chart.exceptions import ChartDataQueryFailedError
from superset.connectors.sqla.models import SqlaTable
from superset.extensions import cache_manager, db, security_manager
from superset.models.core import Database, FavStar, FavStarClassName
from superset.models.dashboard import Dashboard
from superset.models.slice import Slice
from superset.reports.models import ReportSchedule, ReportScheduleType
from superset.tags.models import ObjectType, Tag, TaggedObject, TagType
from superset.utils import json
from superset.utils.core import get_example_default_schema
from tests.integration_tests.base_api_tests import ApiOwnersTestCaseMixin
from tests.integration_tests.base_tests import SupersetTestCase
from tests.integration_tests.constants import (
    ADMIN_USERNAME,
    ALPHA_USERNAME,
    GAMMA_USERNAME,
)
from tests.integration_tests.fixtures.birth_names_dashboard import (
    load_birth_names_dashboard_with_slices,  # noqa: F401
    load_birth_names_data,  # noqa: F401
)
from tests.integration_tests.fixtures.energy_dashboard import (
    load_energy_table_data,  # noqa: F401
    load_energy_table_with_slice,  # noqa: F401
)
from tests.integration_tests.fixtures.importexport import (
    chart_config,
    chart_metadata_config,
    database_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.fixtures.unicode_dashboard import (
    load_unicode_dashboard_with_slice,  # noqa: F401
    load_unicode_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
)
from tests.integration_tests.insert_chart_mixin import InsertChartMixin
from tests.integration_tests.test_app import app
from tests.integration_tests.utils.get_dashboards import get_dashboards_ids

CHART_DATA_URI = "api/v1/chart/data"
CHARTS_FIXTURE_COUNT = 10


class TestChartApi(ApiOwnersTestCaseMixin, InsertChartMixin, SupersetTestCase):
    resource_name = "chart"

    @pytest.fixture(autouse=True)
    def clear_data_cache(self):
        with app.app_context():
            cache_manager.data_cache.clear()
            yield

    @pytest.fixture()
    def create_charts(self):
        with self.create_app().app_context():
            charts = []
            admin = self.get_user("admin")
            for cx in range(CHARTS_FIXTURE_COUNT - 1):
                charts.append(self.insert_chart(f"name{cx}", [admin.id], 1))
            fav_charts = []
            for cx in range(round(CHARTS_FIXTURE_COUNT / 2)):
                fav_star = FavStar(
                    user_id=admin.id, class_name="slice", obj_id=charts[cx].id
                )
                db.session.add(fav_star)
                db.session.commit()
                fav_charts.append(fav_star)
            yield charts

            # rollback changes
            for chart in charts:
                db.session.delete(chart)
            for fav_chart in fav_charts:
                db.session.delete(fav_chart)
            db.session.commit()

    @pytest.fixture()
    def create_charts_created_by_gamma(self):
        with self.create_app().app_context():
            charts = []
            user = self.get_user("gamma")
            for cx in range(CHARTS_FIXTURE_COUNT - 1):
                charts.append(self.insert_chart(f"gamma{cx}", [user.id], 1))
            yield charts
            # rollback changes
            for chart in charts:
                db.session.delete(chart)
            db.session.commit()

    @pytest.fixture()
    def create_certified_charts(self):
        with self.create_app().app_context():
            certified_charts = []
            admin = self.get_user("admin")
            for cx in range(CHARTS_FIXTURE_COUNT):
                certified_charts.append(
                    self.insert_chart(
                        f"certified{cx}",
                        [admin.id],
                        1,
                        certified_by="John Doe",
                        certification_details="Sample certification",
                    )
                )

            yield certified_charts

            # rollback changes
            for chart in certified_charts:
                db.session.delete(chart)
            db.session.commit()

    @pytest.fixture()
    def create_chart_with_report(self):
        with self.create_app().app_context():
            admin = self.get_user("admin")
            chart = self.insert_chart("chart_report", [admin.id], 1)  # noqa: F541
            report_schedule = ReportSchedule(
                type=ReportScheduleType.REPORT,
                name="report_with_chart",
                crontab="* * * * *",
                chart=chart,
            )
            db.session.commit()

            yield chart

            # rollback changes
            db.session.delete(report_schedule)
            db.session.delete(chart)
            db.session.commit()

    @pytest.fixture()
    def add_dashboard_to_chart(self):
        with self.create_app().app_context():
            admin = self.get_user("admin")

            self.chart = self.insert_chart("My chart", [admin.id], 1)

            self.original_dashboard = Dashboard()
            self.original_dashboard.dashboard_title = "Original Dashboard"
            self.original_dashboard.slug = "slug"
            self.original_dashboard.owners = [admin]
            self.original_dashboard.slices = [self.chart]
            self.original_dashboard.published = False
            db.session.add(self.original_dashboard)

            self.new_dashboard = Dashboard()
            self.new_dashboard.dashboard_title = "New Dashboard"
            self.new_dashboard.slug = "new_slug"
            self.new_dashboard.owners = [admin]
            self.new_dashboard.published = False
            db.session.add(self.new_dashboard)

            db.session.commit()

            yield self.chart

            db.session.delete(self.original_dashboard)
            db.session.delete(self.new_dashboard)
            db.session.delete(self.chart)
            db.session.commit()

    @pytest.fixture
    def create_chart_with_tag(self, create_custom_tags):  # noqa: F811
        with self.create_app().app_context():
            alpha_user = self.get_user(ALPHA_USERNAME)

            chart = self.insert_chart(
                "chart with tag",
                [alpha_user.id],
                1,
            )

            tag = db.session.query(Tag).filter(Tag.name == "first_tag").first()
            tag_association = TaggedObject(
                object_id=chart.id,
                object_type=ObjectType.chart,
                tag=tag,
            )

            db.session.add(tag_association)
            db.session.commit()

            yield chart

            # rollback changes
            db.session.delete(tag_association)
            db.session.delete(chart)
            db.session.commit()

    @pytest.fixture
    def create_charts_some_with_tags(self, create_custom_tags):  # noqa: F811
        """
        Fixture that creates 4 charts:
            - ``first_chart`` is associated with ``first_tag``
            - ``second_chart`` is associated with ``second_tag``
            - ``third_chart`` is associated with both ``first_tag`` and ``second_tag``
            - ``fourth_chart`` is not associated with any tag

        Relies on the ``create_custom_tags`` fixture for the tag creation.
        """
        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(),
            }

            chart_names = ["first_chart", "second_chart", "third_chart", "fourth_chart"]
            charts = [
                self.insert_chart(name, [admin_user.id], 1) for name in chart_names
            ]

            tag_associations = [
                TaggedObject(
                    object_id=charts[0].id,
                    object_type=ObjectType.chart,
                    tag=tags["first_tag"],
                ),
                TaggedObject(
                    object_id=charts[1].id,
                    object_type=ObjectType.chart,
                    tag=tags["second_tag"],
                ),
                TaggedObject(
                    object_id=charts[2].id,
                    object_type=ObjectType.chart,
                    tag=tags["first_tag"],
                ),
                TaggedObject(
                    object_id=charts[2].id,
                    object_type=ObjectType.chart,
                    tag=tags["second_tag"],
                ),
            ]

            for association in tag_associations:
                db.session.add(association)
            db.session.commit()

            yield charts

            # rollback changes
            for association in tag_associations:
                db.session.delete(association)
            for chart in charts:
                db.session.delete(chart)
            db.session.commit()

    def test_info_security_chart(self):
        """
        Chart API: Test info security
        """
        self.login(ADMIN_USERNAME)
        params = {"keys": ["permissions"]}
        uri = f"api/v1/chart/_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_warm_up_cache",
        }

    def create_chart_import(self):
        buf = BytesIO()
        with ZipFile(buf, "w") as bundle:
            with bundle.open("chart_export/metadata.yaml", "w") as fp:
                fp.write(yaml.safe_dump(chart_metadata_config).encode())
            with bundle.open(
                "chart_export/databases/imported_database.yaml", "w"
            ) as fp:
                fp.write(yaml.safe_dump(database_config).encode())
            with bundle.open("chart_export/datasets/imported_dataset.yaml", "w") as fp:
                fp.write(yaml.safe_dump(dataset_config).encode())
            with bundle.open("chart_export/charts/imported_chart.yaml", "w") as fp:
                fp.write(yaml.safe_dump(chart_config).encode())
        buf.seek(0)
        return buf

    def test_delete_chart(self):
        """
        Chart API: Test delete
        """
        admin_id = self.get_user("admin").id
        chart_id = self.insert_chart("name", [admin_id], 1).id
        self.login(ADMIN_USERNAME)
        uri = f"api/v1/chart/{chart_id}"
        rv = self.delete_assert_metric(uri, "delete")
        self.assertEqual(rv.status_code, 200)
        model = db.session.query(Slice).get(chart_id)
        self.assertEqual(model, None)

    def test_delete_bulk_charts(self):
        """
        Chart API: Test delete bulk
        """
        admin = self.get_user("admin")
        chart_count = 4
        chart_ids = list()
        for chart_name_index in range(chart_count):
            chart_ids.append(
                self.insert_chart(f"title{chart_name_index}", [admin.id], 1, admin).id
            )
        self.login(ADMIN_USERNAME)
        argument = chart_ids
        uri = f"api/v1/chart/?q={prison.dumps(argument)}"
        rv = self.delete_assert_metric(uri, "bulk_delete")
        self.assertEqual(rv.status_code, 200)
        response = json.loads(rv.data.decode("utf-8"))
        expected_response = {"message": f"Deleted {chart_count} charts"}
        self.assertEqual(response, expected_response)
        for chart_id in chart_ids:
            model = db.session.query(Slice).get(chart_id)
            self.assertEqual(model, None)

    def test_delete_bulk_chart_bad_request(self):
        """
        Chart API: Test delete bulk bad request
        """
        chart_ids = [1, "a"]
        self.login(ADMIN_USERNAME)
        argument = chart_ids
        uri = f"api/v1/chart/?q={prison.dumps(argument)}"
        rv = self.delete_assert_metric(uri, "bulk_delete")
        self.assertEqual(rv.status_code, 400)

    def test_delete_not_found_chart(self):
        """
        Chart API: Test not found delete
        """
        self.login(ADMIN_USERNAME)
        chart_id = 1000
        uri = f"api/v1/chart/{chart_id}"
        rv = self.delete_assert_metric(uri, "delete")
        self.assertEqual(rv.status_code, 404)

    @pytest.mark.usefixtures("create_chart_with_report")
    def test_delete_chart_with_report(self):
        """
        Chart API: Test delete with associated report
        """
        self.login(ADMIN_USERNAME)
        chart = (
            db.session.query(Slice)
            .filter(Slice.slice_name == "chart_report")
            .one_or_none()
        )
        uri = f"api/v1/chart/{chart.id}"
        rv = self.client.delete(uri)
        response = json.loads(rv.data.decode("utf-8"))
        self.assertEqual(rv.status_code, 422)
        expected_response = {
            "message": "There are associated alerts or reports: report_with_chart"
        }
        self.assertEqual(response, expected_response)

    def test_delete_bulk_charts_not_found(self):
        """
        Chart API: Test delete bulk not found
        """
        max_id = db.session.query(func.max(Slice.id)).scalar()
        chart_ids = [max_id + 1, max_id + 2]
        self.login(ADMIN_USERNAME)
        uri = f"api/v1/chart/?q={prison.dumps(chart_ids)}"
        rv = self.delete_assert_metric(uri, "bulk_delete")
        self.assertEqual(rv.status_code, 404)

    @pytest.mark.usefixtures("create_chart_with_report", "create_charts")
    def test_bulk_delete_chart_with_report(self):
        """
        Chart API: Test bulk delete with associated report
        """
        self.login(ADMIN_USERNAME)
        chart_with_report = (
            db.session.query(Slice.id)
            .filter(Slice.slice_name == "chart_report")
            .one_or_none()
        )

        charts = db.session.query(Slice.id).filter(Slice.slice_name.like("name%")).all()
        chart_ids = [chart.id for chart in charts]
        chart_ids.append(chart_with_report.id)

        uri = f"api/v1/chart/?q={prison.dumps(chart_ids)}"
        rv = self.client.delete(uri)
        response = json.loads(rv.data.decode("utf-8"))
        self.assertEqual(rv.status_code, 422)
        expected_response = {
            "message": "There are associated alerts or reports: report_with_chart"
        }
        self.assertEqual(response, expected_response)

    def test_delete_chart_admin_not_owned(self):
        """
        Chart API: Test admin delete not owned
        """
        gamma_id = self.get_user("gamma").id
        chart_id = self.insert_chart("title", [gamma_id], 1).id

        self.login(ADMIN_USERNAME)
        uri = f"api/v1/chart/{chart_id}"
        rv = self.delete_assert_metric(uri, "delete")
        self.assertEqual(rv.status_code, 200)
        model = db.session.query(Slice).get(chart_id)
        self.assertEqual(model, None)

    def test_delete_bulk_chart_admin_not_owned(self):
        """
        Chart API: Test admin delete bulk not owned
        """
        gamma_id = self.get_user("gamma").id
        chart_count = 4
        chart_ids = list()
        for chart_name_index in range(chart_count):
            chart_ids.append(
                self.insert_chart(f"title{chart_name_index}", [gamma_id], 1).id
            )

        self.login(ADMIN_USERNAME)
        argument = chart_ids
        uri = f"api/v1/chart/?q={prison.dumps(argument)}"
        rv = self.delete_assert_metric(uri, "bulk_delete")
        response = json.loads(rv.data.decode("utf-8"))
        self.assertEqual(rv.status_code, 200)
        expected_response = {"message": f"Deleted {chart_count} charts"}
        self.assertEqual(response, expected_response)

        for chart_id in chart_ids:
            model = db.session.query(Slice).get(chart_id)
            self.assertEqual(model, None)

    def test_delete_chart_not_owned(self):
        """
        Chart 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"
        )
        chart = self.insert_chart("title", [user_alpha1.id], 1)
        self.login(username="alpha2", password="password")
        uri = f"api/v1/chart/{chart.id}"
        rv = self.delete_assert_metric(uri, "delete")
        self.assertEqual(rv.status_code, 403)
        db.session.delete(chart)
        db.session.delete(user_alpha1)
        db.session.delete(user_alpha2)
        db.session.commit()

    def test_delete_bulk_chart_not_owned(self):
        """
        Chart 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"
        )

        chart_count = 4
        charts = list()
        for chart_name_index in range(chart_count):
            charts.append(
                self.insert_chart(f"title{chart_name_index}", [user_alpha1.id], 1)
            )

        owned_chart = self.insert_chart("title_owned", [user_alpha2.id], 1)

        self.login(username="alpha2", password="password")

        # verify we can't delete not owned charts
        arguments = [chart.id for chart in charts]
        uri = f"api/v1/chart/?q={prison.dumps(arguments)}"
        rv = self.delete_assert_metric(uri, "bulk_delete")
        self.assertEqual(rv.status_code, 403)
        response = json.loads(rv.data.decode("utf-8"))
        expected_response = {"message": "Forbidden"}
        self.assertEqual(response, expected_response)

        # # nothing is deleted in bulk with a list of owned and not owned charts
        arguments = [chart.id for chart in charts] + [owned_chart.id]
        uri = f"api/v1/chart/?q={prison.dumps(arguments)}"
        rv = self.delete_assert_metric(uri, "bulk_delete")
        self.assertEqual(rv.status_code, 403)
        response = json.loads(rv.data.decode("utf-8"))
        expected_response = {"message": "Forbidden"}
        self.assertEqual(response, expected_response)

        for chart in charts:
            db.session.delete(chart)
        db.session.delete(owned_chart)
        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",
    )
    def test_create_chart(self):
        """
        Chart API: Test create chart
        """
        dashboards_ids = get_dashboards_ids(["world_health", "births"])
        admin_id = self.get_user("admin").id
        chart_data = {
            "slice_name": "name1",
            "description": "description1",
            "owners": [admin_id],
            "viz_type": "viz_type1",
            "params": "1234",
            "cache_timeout": 1000,
            "datasource_id": 1,
            "datasource_type": "table",
            "dashboards": dashboards_ids,
            "certified_by": "John Doe",
            "certification_details": "Sample certification",
        }
        self.login(ADMIN_USERNAME)
        uri = "api/v1/chart/"
        rv = self.post_assert_metric(uri, chart_data, "post")
        self.assertEqual(rv.status_code, 201)
        data = json.loads(rv.data.decode("utf-8"))
        model = db.session.query(Slice).get(data.get("id"))
        db.session.delete(model)
        db.session.commit()

    def test_create_simple_chart(self):
        """
        Chart API: Test create simple chart
        """
        chart_data = {
            "slice_name": "title1",
            "datasource_id": 1,
            "datasource_type": "table",
        }
        self.login(ADMIN_USERNAME)
        uri = "api/v1/chart/"
        rv = self.post_assert_metric(uri, chart_data, "post")
        self.assertEqual(rv.status_code, 201)
        data = json.loads(rv.data.decode("utf-8"))
        model = db.session.query(Slice).get(data.get("id"))
        db.session.delete(model)
        db.session.commit()

    def test_create_chart_validate_owners(self):
        """
        Chart API: Test create validate owners
        """
        chart_data = {
            "slice_name": "title1",
            "datasource_id": 1,
            "datasource_type": "table",
            "owners": [1000],
        }
        self.login(ADMIN_USERNAME)
        uri = "api/v1/chart/"
        rv = self.post_assert_metric(uri, chart_data, "post")
        self.assertEqual(rv.status_code, 422)
        response = json.loads(rv.data.decode("utf-8"))
        expected_response = {"message": {"owners": ["Owners are invalid"]}}
        self.assertEqual(response, expected_response)

    def test_create_chart_validate_params(self):
        """
        Chart API: Test create validate params json
        """
        chart_data = {
            "slice_name": "title1",
            "datasource_id": 1,
            "datasource_type": "table",
            "params": '{"A:"a"}',
        }
        self.login(ADMIN_USERNAME)
        uri = "api/v1/chart/"
        rv = self.post_assert_metric(uri, chart_data, "post")
        self.assertEqual(rv.status_code, 400)

    def test_create_chart_validate_datasource(self):
        """
        Chart API: Test create validate datasource
        """
        self.login(ADMIN_USERNAME)
        chart_data = {
            "slice_name": "title1",
            "datasource_id": 1,
            "datasource_type": "unknown",
        }
        rv = self.post_assert_metric("/api/v1/chart/", chart_data, "post")
        self.assertEqual(rv.status_code, 400)
        response = json.loads(rv.data.decode("utf-8"))
        self.assertEqual(
            response,
            {
                "message": {
                    "datasource_type": [
                        "Must be one of: table, dataset, query, saved_query, view."
                    ]
                }
            },
        )
        chart_data = {
            "slice_name": "title1",
            "datasource_id": 0,
            "datasource_type": "table",
        }
        rv = self.post_assert_metric("/api/v1/chart/", chart_data, "post")
        self.assertEqual(rv.status_code, 422)
        response = json.loads(rv.data.decode("utf-8"))
        self.assertEqual(
            response, {"message": {"datasource_id": ["Datasource does not exist"]}}
        )

    @pytest.mark.usefixtures("load_world_bank_dashboard_with_slices")
    def test_create_chart_validate_user_is_dashboard_owner(self):
        """
        Chart API: Test create validate user is dashboard owner
        """
        dash = db.session.query(Dashboard).filter_by(slug="world_health").first()
        # Must be published so that alpha user has read access to dash
        dash.published = True
        db.session.commit()
        chart_data = {
            "slice_name": "title1",
            "datasource_id": 1,
            "datasource_type": "table",
            "dashboards": [dash.id],
        }
        self.login(ALPHA_USERNAME)
        uri = "api/v1/chart/"
        rv = self.post_assert_metric(uri, chart_data, "post")
        self.assertEqual(rv.status_code, 403)
        response = json.loads(rv.data.decode("utf-8"))
        self.assertEqual(
            response,
            {"message": "Changing one or more of these dashboards is forbidden"},
        )

    @pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
    def test_update_chart(self):
        """
        Chart API: Test update
        """
        schema = get_example_default_schema()
        full_table_name = f"{schema}.birth_names" if schema else "birth_names"

        admin = self.get_user("admin")
        gamma = self.get_user("gamma")
        birth_names_table_id = SupersetTestCase.get_table(name="birth_names").id
        chart_id = self.insert_chart(
            "title", [admin.id], birth_names_table_id, admin
        ).id
        dash_id = db.session.query(Dashboard.id).filter_by(slug="births").first()[0]
        chart_data = {
            "slice_name": "title1_changed",
            "description": "description1",
            "owners": [gamma.id],
            "viz_type": "viz_type1",
            "params": """{"a": 1}""",
            "cache_timeout": 1000,
            "datasource_id": birth_names_table_id,
            "datasource_type": "table",
            "dashboards": [dash_id],
            "certified_by": "Mario Rossi",
            "certification_details": "Edited certification",
        }
        self.login(ADMIN_USERNAME)
        uri = f"api/v1/chart/{chart_id}"
        rv = self.put_assert_metric(uri, chart_data, "put")
        self.assertEqual(rv.status_code, 200)
        model = db.session.query(Slice).get(chart_id)
        related_dashboard = db.session.query(Dashboard).filter_by(slug="births").first()
        self.assertEqual(model.created_by, admin)
        self.assertEqual(model.slice_name, "title1_changed")
        self.assertEqual(model.description, "description1")
        self.assertNotIn(admin, model.owners)
        self.assertIn(gamma, model.owners)
        self.assertEqual(model.viz_type, "viz_type1")
        self.assertEqual(model.params, """{"a": 1}""")
        self.assertEqual(model.cache_timeout, 1000)
        self.assertEqual(model.datasource_id, birth_names_table_id)
        self.assertEqual(model.datasource_type, "table")
        self.assertEqual(model.datasource_name, full_table_name)
        self.assertEqual(model.certified_by, "Mario Rossi")
        self.assertEqual(model.certification_details, "Edited certification")
        self.assertIn(model.id, [slice.id for slice in related_dashboard.slices])
        db.session.delete(model)
        db.session.commit()

    @pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
    def test_chart_get_list_no_username(self):
        """
        Chart API: Tests that no username is returned
        """
        admin = self.get_user("admin")
        birth_names_table_id = SupersetTestCase.get_table(name="birth_names").id
        chart_id = self.insert_chart("title", [admin.id], birth_names_table_id).id
        chart_data = {
            "slice_name": (new_name := "title1_changed"),
            "owners": [admin.id],
        }
        self.login(ADMIN_USERNAME)
        uri = f"api/v1/chart/{chart_id}"
        rv = self.put_assert_metric(uri, chart_data, "put")
        self.assertEqual(rv.status_code, 200)
        model = db.session.query(Slice).get(chart_id)

        response = self.get_assert_metric("api/v1/chart/", "get_list")
        res = json.loads(response.data.decode("utf-8"))["result"]

        current_chart = [d for d in res if d["id"] == chart_id][0]
        self.assertEqual(current_chart["slice_name"], new_name)
        self.assertNotIn("username", current_chart["changed_by"].keys())
        self.assertNotIn("username", current_chart["owners"][0].keys())

        db.session.delete(model)
        db.session.commit()

    @pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
    def test_chart_get_no_username(self):
        """
        Chart API: Tests that no username is returned
        """
        admin = self.get_user("admin")
        birth_names_table_id = SupersetTestCase.get_table(name="birth_names").id
        chart_id = self.insert_chart("title", [admin.id], birth_names_table_id).id
        chart_data = {
            "slice_name": (new_name := "title1_changed"),
            "owners": [admin.id],
        }
        self.login(ADMIN_USERNAME)
        uri = f"api/v1/chart/{chart_id}"
        rv = self.put_assert_metric(uri, chart_data, "put")
        self.assertEqual(rv.status_code, 200)
        model = db.session.query(Slice).get(chart_id)

        response = self.get_assert_metric(uri, "get")
        res = json.loads(response.data.decode("utf-8"))["result"]

        self.assertEqual(res["slice_name"], new_name)
        self.assertNotIn("username", res["owners"][0].keys())

        db.session.delete(model)
        db.session.commit()

    def test_update_chart_new_owner_not_admin(self):
        """
        Chart API: Test update set new owner implicitly adds logged in owner
        """
        gamma = self.get_user("gamma_no_csv")
        alpha = self.get_user("alpha")
        chart_id = self.insert_chart("title", [gamma.id], 1).id
        chart_data = {
            "slice_name": (new_name := "title1_changed"),
            "owners": [alpha.id],
        }
        self.login(gamma.username)
        uri = f"api/v1/chart/{chart_id}"
        rv = self.put_assert_metric(uri, chart_data, "put")
        assert rv.status_code == 200
        model = db.session.query(Slice).get(chart_id)
        assert model.slice_name == new_name
        assert alpha in model.owners
        assert gamma in model.owners
        db.session.delete(model)
        db.session.commit()

    def test_update_chart_new_owner_admin(self):
        """
        Chart API: Test update set new owner as admin to other than current user
        """
        gamma = self.get_user("gamma")
        admin = self.get_user("admin")
        chart_id = self.insert_chart("title", [admin.id], 1).id
        chart_data = {"slice_name": "title1_changed", "owners": [gamma.id]}
        self.login(ADMIN_USERNAME)
        uri = f"api/v1/chart/{chart_id}"
        rv = self.put_assert_metric(uri, chart_data, "put")
        self.assertEqual(rv.status_code, 200)
        model = db.session.query(Slice).get(chart_id)
        self.assertNotIn(admin, model.owners)
        self.assertIn(gamma, model.owners)
        db.session.delete(model)
        db.session.commit()

    @pytest.mark.usefixtures("add_dashboard_to_chart")
    def test_update_chart_preserve_ownership(self):
        """
        Chart API: Test update chart preserves owner list (if un-changed)
        """
        chart_data = {
            "slice_name": "title1_changed",
        }
        admin = self.get_user("admin")
        self.login(username="admin")
        uri = f"api/v1/chart/{self.chart.id}"
        rv = self.put_assert_metric(uri, chart_data, "put")
        self.assertEqual(rv.status_code, 200)
        self.assertEqual([admin], self.chart.owners)

    @pytest.mark.usefixtures("add_dashboard_to_chart")
    def test_update_chart_clear_owner_list(self):
        """
        Chart API: Test update chart admin can clear owner list
        """
        chart_data = {"slice_name": "title1_changed", "owners": []}
        self.get_user("admin")  # noqa: F841
        self.login(username="admin")
        uri = f"api/v1/chart/{self.chart.id}"
        rv = self.put_assert_metric(uri, chart_data, "put")
        self.assertEqual(rv.status_code, 200)
        self.assertEqual([], self.chart.owners)

    def test_update_chart_populate_owner(self):
        """
        Chart API: Test update admin can update chart with
        no owners to a different owner
        """
        gamma = self.get_user("gamma")
        admin = self.get_user("admin")
        chart_id = self.insert_chart("title", [], 1).id
        model = db.session.query(Slice).get(chart_id)
        self.assertEqual(model.owners, [])
        chart_data = {"owners": [gamma.id]}
        self.login(username="admin")
        uri = f"api/v1/chart/{chart_id}"
        rv = self.put_assert_metric(uri, chart_data, "put")
        self.assertEqual(rv.status_code, 200)
        model_updated = db.session.query(Slice).get(chart_id)
        self.assertNotIn(admin, model_updated.owners)
        self.assertIn(gamma, model_updated.owners)
        db.session.delete(model_updated)
        db.session.commit()

    @pytest.mark.usefixtures("add_dashboard_to_chart")
    def test_update_chart_new_dashboards(self):
        """
        Chart API: Test update chart associating it with new dashboard
        """
        chart_data = {
            "slice_name": "title1_changed",
            "dashboards": [self.new_dashboard.id],
        }
        self.login(ADMIN_USERNAME)
        uri = f"api/v1/chart/{self.chart.id}"
        rv = self.put_assert_metric(uri, chart_data, "put")
        self.assertEqual(rv.status_code, 200)
        self.assertIn(self.new_dashboard, self.chart.dashboards)
        self.assertNotIn(self.original_dashboard, self.chart.dashboards)

    @pytest.mark.usefixtures("add_dashboard_to_chart")
    def test_not_update_chart_none_dashboards(self):
        """
        Chart API: Test update chart without changing dashboards configuration
        """
        chart_data = {"slice_name": "title1_changed_again"}
        self.login(ADMIN_USERNAME)
        uri = f"api/v1/chart/{self.chart.id}"
        rv = self.put_assert_metric(uri, chart_data, "put")
        self.assertEqual(rv.status_code, 200)
        self.assertIn(self.original_dashboard, self.chart.dashboards)
        self.assertEqual(len(self.chart.dashboards), 1)

    def test_update_chart_not_owned(self):
        """
        Chart API: Test update 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"
        )
        chart = self.insert_chart("title", [user_alpha1.id], 1)

        self.login(username="alpha2", password="password")
        chart_data = {"slice_name": "title1_changed"}
        uri = f"api/v1/chart/{chart.id}"
        rv = self.put_assert_metric(uri, chart_data, "put")
        self.assertEqual(rv.status_code, 403)
        db.session.delete(chart)
        db.session.delete(user_alpha1)
        db.session.delete(user_alpha2)
        db.session.commit()

    def test_update_chart_linked_with_not_owned_dashboard(self):
        """
        Chart API: Test update chart which is linked to not owned dashboard
        """
        user_alpha1 = self.create_user(
            "alpha1", "password", "Alpha", email="alpha1@superset.org"
        )
        user_alpha2 = self.create_user(
            "alpha2", "password", "Alpha", email="alpha2@superset.org"
        )
        chart = self.insert_chart("title", [user_alpha1.id], 1)

        original_dashboard = Dashboard()
        original_dashboard.dashboard_title = "Original Dashboard"
        original_dashboard.slug = "slug"
        original_dashboard.owners = [user_alpha1]
        original_dashboard.slices = [chart]
        original_dashboard.published = False
        db.session.add(original_dashboard)

        new_dashboard = Dashboard()
        new_dashboard.dashboard_title = "Cloned Dashboard"
        new_dashboard.slug = "new_slug"
        new_dashboard.owners = [user_alpha2]
        new_dashboard.slices = [chart]
        new_dashboard.published = False
        db.session.add(new_dashboard)

        self.login(username="alpha1", password="password")
        chart_data_with_invalid_dashboard = {
            "slice_name": "title1_changed",
            "dashboards": [original_dashboard.id, 0],
        }
        chart_data = {
            "slice_name": "title1_changed",
            "dashboards": [original_dashboard.id, new_dashboard.id],
        }
        uri = f"api/v1/chart/{chart.id}"

        rv = self.put_assert_metric(uri, chart_data_with_invalid_dashboard, "put")
        self.assertEqual(rv.status_code, 422)
        response = json.loads(rv.data.decode("utf-8"))
        expected_response = {"message": {"dashboards": ["Dashboards do not exist"]}}
        self.assertEqual(response, expected_response)

        rv = self.put_assert_metric(uri, chart_data, "put")
        self.assertEqual(rv.status_code, 200)

        db.session.delete(chart)
        db.session.delete(original_dashboard)
        db.session.delete(new_dashboard)
        db.session.delete(user_alpha1)
        db.session.delete(user_alpha2)
        db.session.commit()

    def test_update_chart_validate_datasource(self):
        """
        Chart API: Test update validate datasource
        """
        admin = self.get_user("admin")
        chart = self.insert_chart("title", owners=[admin.id], datasource_id=1)
        self.login(ADMIN_USERNAME)

        chart_data = {"datasource_id": 1, "datasource_type": "unknown"}
        rv = self.put_assert_metric(f"/api/v1/chart/{chart.id}", chart_data, "put")
        self.assertEqual(rv.status_code, 400)
        response = json.loads(rv.data.decode("utf-8"))
        self.assertEqual(
            response,
            {
                "message": {
                    "datasource_type": [
                        "Must be one of: table, dataset, query, saved_query, view."
                    ]
                }
            },
        )

        chart_data = {"datasource_id": 0, "datasource_type": "table"}
        rv = self.put_assert_metric(f"/api/v1/chart/{chart.id}", chart_data, "put")
        self.assertEqual(rv.status_code, 422)
        response = json.loads(rv.data.decode("utf-8"))
        self.assertEqual(
            response, {"message": {"datasource_id": ["Datasource does not exist"]}}
        )

        db.session.delete(chart)
        db.session.commit()

    def test_update_chart_validate_owners(self):
        """
        Chart API: Test update validate owners
        """
        chart_data = {
            "slice_name": "title1",
            "datasource_id": 1,
            "datasource_type": "table",
            "owners": [1000],
        }
        self.login(ADMIN_USERNAME)
        uri = "api/v1/chart/"  # noqa: F541
        rv = self.client.post(uri, json=chart_data)
        self.assertEqual(rv.status_code, 422)
        response = json.loads(rv.data.decode("utf-8"))
        expected_response = {"message": {"owners": ["Owners are invalid"]}}
        self.assertEqual(response, expected_response)

    @pytest.mark.usefixtures("load_world_bank_dashboard_with_slices")
    def test_get_chart(self):
        """
        Chart API: Test get chart
        """
        admin = self.get_user("admin")
        chart = self.insert_chart("title", [admin.id], 1)
        self.login(ADMIN_USERNAME)
        uri = f"api/v1/chart/{chart.id}"
        rv = self.get_assert_metric(uri, "get")
        self.assertEqual(rv.status_code, 200)
        expected_result = {
            "cache_timeout": None,
            "certified_by": None,
            "certification_details": None,
            "dashboards": [],
            "description": None,
            "owners": [
                {
                    "id": 1,
                    "first_name": "admin",
                    "last_name": "user",
                }
            ],
            "params": None,
            "slice_name": "title",
            "tags": [],
            "viz_type": None,
            "query_context": None,
            "is_managed_externally": False,
        }
        data = json.loads(rv.data.decode("utf-8"))
        self.assertIn("changed_on_delta_humanized", data["result"])
        self.assertIn("id", data["result"])
        self.assertIn("thumbnail_url", data["result"])
        self.assertIn("url", data["result"])
        for key, value in data["result"].items():
            # We can't assert timestamp values or id/urls
            if key not in (
                "changed_on_delta_humanized",
                "id",
                "thumbnail_url",
                "url",
            ):
                self.assertEqual(value, expected_result[key])
        db.session.delete(chart)
        db.session.commit()

    def test_get_chart_not_found(self):
        """
        Chart API: Test get chart not found
        """
        chart_id = 1000
        self.login(ADMIN_USERNAME)
        uri = f"api/v1/chart/{chart_id}"
        rv = self.get_assert_metric(uri, "get")
        self.assertEqual(rv.status_code, 404)

    @pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
    def test_get_chart_no_data_access(self):
        """
        Chart API: Test get chart without data access
        """
        self.login(GAMMA_USERNAME)
        chart_no_access = (
            db.session.query(Slice)
            .filter_by(slice_name="Girl Name Cloud")
            .one_or_none()
        )
        uri = f"api/v1/chart/{chart_no_access.id}"
        rv = self.client.get(uri)
        self.assertEqual(rv.status_code, 404)

    @pytest.mark.usefixtures(
        "load_energy_table_with_slice",
        "load_birth_names_dashboard_with_slices",
        "load_unicode_dashboard_with_slice",
        "load_world_bank_dashboard_with_slices",
    )
    def test_get_charts(self):
        """
        Chart API: Test get charts
        """
        self.login(ADMIN_USERNAME)
        uri = "api/v1/chart/"  # noqa: F541
        rv = self.get_assert_metric(uri, "get_list")
        self.assertEqual(rv.status_code, 200)
        data = json.loads(rv.data.decode("utf-8"))
        self.assertEqual(data["count"], 33)

    @pytest.mark.usefixtures("load_energy_table_with_slice", "add_dashboard_to_chart")
    def test_get_charts_dashboards(self):
        """
        Chart API: Test get charts with related dashboards
        """
        self.login(ADMIN_USERNAME)
        arguments = {
            "filters": [
                {"col": "slice_name", "opr": "eq", "value": self.chart.slice_name}
            ]
        }
        uri = f"api/v1/chart/?q={prison.dumps(arguments)}"
        rv = self.get_assert_metric(uri, "get_list")
        self.assertEqual(rv.status_code, 200)
        data = json.loads(rv.data.decode("utf-8"))
        assert data["result"][0]["dashboards"] == [
            {
                "id": self.original_dashboard.id,
                "dashboard_title": self.original_dashboard.dashboard_title,
            }
        ]

    @pytest.mark.usefixtures("load_energy_table_with_slice", "add_dashboard_to_chart")
    def test_get_charts_dashboard_filter(self):
        """
        Chart API: Test get charts with dashboard filter
        """
        self.login(ADMIN_USERNAME)
        arguments = {
            "filters": [
                {
                    "col": "dashboards",
                    "opr": "rel_m_m",
                    "value": self.original_dashboard.id,
                }
            ]
        }
        uri = f"api/v1/chart/?q={prison.dumps(arguments)}"
        rv = self.get_assert_metric(uri, "get_list")
        self.assertEqual(rv.status_code, 200)
        data = json.loads(rv.data.decode("utf-8"))
        result = data["result"]
        assert len(result) == 1
        assert result[0]["slice_name"] == self.chart.slice_name

    @pytest.mark.usefixtures("create_charts_some_with_tags")
    def test_get_charts_tag_filters(self):
        """
        Chart API: Test get charts 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(),
        }
        chart_tag_relationship = {
            tag.name: db.session.query(Slice.id)
            .join(Slice.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_charts = chart_tag_relationship[tag_name]

            # Filter by tag ID
            filter_params = get_filter_params("chart_tag_id", tag.id)
            response_by_id = self.get_list("chart", filter_params)
            self.assertEqual(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("chart_tags", tag.name)
            response_by_name = self.get_list("chart", filter_params)
            self.assertEqual(response_by_name.status_code, 200)
            data_by_name = json.loads(response_by_name.data.decode("utf-8"))

            # Compare results
            self.assertEqual(
                data_by_id["count"],
                data_by_name["count"],
                len(expected_charts),
            )
            self.assertEqual(
                set(chart["id"] for chart in data_by_id["result"]),
                set(chart["id"] for chart in data_by_name["result"]),
                set(chart.id for chart in expected_charts),
            )

    def test_get_charts_changed_on(self):
        """
        Dashboard API: Test get charts changed on
        """
        admin = self.get_user("admin")
        chart = self.insert_chart("foo_a", [admin.id], 1, description="ZY_bar")

        self.login(ADMIN_USERNAME)

        arguments = {
            "order_column": "changed_on_delta_humanized",
            "order_direction": "desc",
        }
        uri = f"api/v1/chart/?q={prison.dumps(arguments)}"

        rv = self.get_assert_metric(uri, "get_list")
        self.assertEqual(rv.status_code, 200)
        data = json.loads(rv.data.decode("utf-8"))
        assert data["result"][0]["changed_on_delta_humanized"] in (
            "now",
            "a second ago",
        )

        # rollback changes
        db.session.delete(chart)
        db.session.commit()

    @pytest.mark.usefixtures(
        "load_world_bank_dashboard_with_slices",
        "load_birth_names_dashboard_with_slices",
    )
    def test_get_charts_filter(self):
        """
        Chart API: Test get charts filter
        """
        self.login(ADMIN_USERNAME)
        arguments = {"filters": [{"col": "slice_name", "opr": "sw", "value": "G"}]}
        uri = f"api/v1/chart/?q={prison.dumps(arguments)}"
        rv = self.get_assert_metric(uri, "get_list")
        self.assertEqual(rv.status_code, 200)
        data = json.loads(rv.data.decode("utf-8"))
        self.assertEqual(data["count"], 5)

    @pytest.fixture()
    def load_energy_charts(self):
        with app.app_context():
            admin = self.get_user("admin")
            energy_table = (
                db.session.query(SqlaTable)
                .filter_by(table_name="energy_usage")
                .one_or_none()
            )
            energy_table_id = 1
            if energy_table:
                energy_table_id = energy_table.id
            chart1 = self.insert_chart(
                "foo_a", [admin.id], energy_table_id, description="ZY_bar"
            )
            chart2 = self.insert_chart(
                "zy_foo", [admin.id], energy_table_id, description="desc1"
            )
            chart3 = self.insert_chart(
                "foo_b", [admin.id], energy_table_id, description="desc1zy_"
            )
            chart4 = self.insert_chart(
                "foo_c", [admin.id], energy_table_id, viz_type="viz_zy_"
            )
            chart5 = self.insert_chart(
                "bar", [admin.id], energy_table_id, description="foo"
            )

            yield
            # rollback changes
            db.session.delete(chart1)
            db.session.delete(chart2)
            db.session.delete(chart3)
            db.session.delete(chart4)
            db.session.delete(chart5)
            db.session.commit()

    @pytest.mark.usefixtures("load_energy_charts")
    def test_get_charts_custom_filter(self):
        """
        Chart API: Test get charts custom filter
        """

        arguments = {
            "filters": [{"col": "slice_name", "opr": "chart_all_text", "value": "zy_"}],
            "order_column": "slice_name",
            "order_direction": "asc",
            "keys": ["none"],
            "columns": ["slice_name", "description", "viz_type"],
        }
        self.login(ADMIN_USERNAME)
        uri = f"api/v1/chart/?q={prison.dumps(arguments)}"
        rv = self.get_assert_metric(uri, "get_list")
        self.assertEqual(rv.status_code, 200)
        data = json.loads(rv.data.decode("utf-8"))
        self.assertEqual(data["count"], 4)

        expected_response = [
            {"description": "ZY_bar", "slice_name": "foo_a", "viz_type": None},
            {"description": "desc1zy_", "slice_name": "foo_b", "viz_type": None},
            {"description": None, "slice_name": "foo_c", "viz_type": "viz_zy_"},
            {"description": "desc1", "slice_name": "zy_foo", "viz_type": None},
        ]
        for index, item in enumerate(data["result"]):
            self.assertEqual(
                item["description"], expected_response[index]["description"]
            )
            self.assertEqual(item["slice_name"], expected_response[index]["slice_name"])
            self.assertEqual(item["viz_type"], expected_response[index]["viz_type"])

    @pytest.mark.usefixtures("load_energy_table_with_slice", "load_energy_charts")
    def test_admin_gets_filtered_energy_slices(self):
        # test filtering on datasource_name
        arguments = {
            "filters": [
                {
                    "col": "slice_name",
                    "opr": "chart_all_text",
                    "value": "energy",
                }
            ],
            "keys": ["none"],
            "columns": ["slice_name", "description", "table.table_name"],
        }
        self.login(ADMIN_USERNAME)

        uri = f"api/v1/chart/?q={prison.dumps(arguments)}"
        rv = self.get_assert_metric(uri, "get_list")
        data = rv.json
        assert rv.status_code == 200
        assert data["count"] > 0
        for chart in data["result"]:
            assert (
                "energy"
                in " ".join(
                    [
                        chart["slice_name"] or "",
                        chart["description"] or "",
                        chart["table"]["table_name"] or "",
                    ]
                ).lower()
            )

    @pytest.mark.usefixtures("create_certified_charts")
    def test_gets_certified_charts_filter(self):
        arguments = {
            "filters": [
                {
                    "col": "id",
                    "opr": "chart_is_certified",
                    "value": True,
                }
            ],
            "keys": ["none"],
            "columns": ["slice_name"],
        }
        self.login(ADMIN_USERNAME)

        uri = f"api/v1/chart/?q={prison.dumps(arguments)}"
        rv = self.get_assert_metric(uri, "get_list")
        self.assertEqual(rv.status_code, 200)
        data = json.loads(rv.data.decode("utf-8"))
        self.assertEqual(data["count"], CHARTS_FIXTURE_COUNT)

    @pytest.mark.usefixtures("create_charts")
    def test_gets_not_certified_charts_filter(self):
        arguments = {
            "filters": [
                {
                    "col": "id",
                    "opr": "chart_is_certified",
                    "value": False,
                }
            ],
            "keys": ["none"],
            "columns": ["slice_name"],
        }
        self.login(ADMIN_USERNAME)

        uri = f"api/v1/chart/?q={prison.dumps(arguments)}"
        rv = self.get_assert_metric(uri, "get_list")
        self.assertEqual(rv.status_code, 200)
        data = json.loads(rv.data.decode("utf-8"))
        self.assertEqual(data["count"], 17)

    @pytest.mark.usefixtures("load_energy_charts")
    def test_user_gets_none_filtered_energy_slices(self):
        # test filtering on datasource_name
        arguments = {
            "filters": [
                {
                    "col": "slice_name",
                    "opr": "chart_all_text",
                    "value": "energy",
                }
            ],
            "keys": ["none"],
            "columns": ["slice_name"],
        }

        self.login(GAMMA_USERNAME)
        uri = f"api/v1/chart/?q={prison.dumps(arguments)}"
        rv = self.get_assert_metric(uri, "get_list")
        self.assertEqual(rv.status_code, 200)
        data = json.loads(rv.data.decode("utf-8"))
        self.assertEqual(data["count"], 0)

    @pytest.mark.usefixtures("load_energy_charts")
    def test_user_gets_all_charts(self):
        # test filtering on datasource_name
        gamma_user = security_manager.find_user(username="gamma")

        def count_charts():
            uri = "api/v1/chart/"
            rv = self.client.get(uri, "get_list")
            self.assertEqual(rv.status_code, 200)
            data = rv.get_json()
            return data["count"]

        with self.temporary_user(gamma_user, login=True):
            self.assertEqual(count_charts(), 0)

        perm = ("all_database_access", "all_database_access")
        with self.temporary_user(gamma_user, extra_pvms=[perm], login=True):
            assert count_charts() > 0

        perm = ("all_datasource_access", "all_datasource_access")
        with self.temporary_user(gamma_user, extra_pvms=[perm], login=True):
            assert count_charts() > 0

        # Back to normal
        with self.temporary_user(gamma_user, login=True):
            self.assertEqual(count_charts(), 0)

    @pytest.mark.usefixtures("create_charts")
    def test_get_charts_favorite_filter(self):
        """
        Chart API: Test get charts 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 == "slice")
        )
        expected_models = (
            db.session.query(Slice)
            .filter(and_(Slice.id.in_(users_favorite_query)))
            .order_by(Slice.slice_name.asc())
            .all()
        )

        arguments = {
            "filters": [{"col": "id", "opr": "chart_is_favorite", "value": True}],
            "order_column": "slice_name",
            "order_direction": "asc",
            "keys": ["none"],
            "columns": ["slice_name"],
        }
        self.login(ADMIN_USERNAME)
        uri = f"api/v1/chart/?q={prison.dumps(arguments)}"
        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.slice_name == data["result"][i]["slice_name"]

        # Test not favorite charts
        expected_models = (
            db.session.query(Slice)
            .filter(and_(~Slice.id.in_(users_favorite_query)))
            .order_by(Slice.slice_name.asc())
            .all()
        )
        arguments["filters"][0]["value"] = False
        uri = f"api/v1/chart/?q={prison.dumps(arguments)}"
        rv = self.client.get(uri)
        data = json.loads(rv.data.decode("utf-8"))
        assert rv.status_code == 200
        assert len(expected_models) == data["count"]

    @pytest.mark.usefixtures("create_charts_created_by_gamma")
    def test_get_charts_created_by_me_filter(self):
        """
        Chart API: Test get charts with created by me special filter
        """
        gamma_user = self.get_user("gamma")
        expected_models = (
            db.session.query(Slice).filter(Slice.created_by_fk == gamma_user.id).all()
        )
        arguments = {
            "filters": [
                {"col": "created_by", "opr": "chart_created_by_me", "value": "me"}
            ],
            "order_column": "slice_name",
            "order_direction": "asc",
            "keys": ["none"],
            "columns": ["slice_name"],
        }
        self.login(gamma_user.username)
        uri = f"api/v1/chart/?q={prison.dumps(arguments)}"
        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.slice_name == data["result"][i]["slice_name"]

    @pytest.mark.usefixtures("create_charts")
    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.CHART,
                )
            )
            .all()
        ]

        assert users_favorite_ids
        arguments = [s.id for s in db.session.query(Slice.id).all()]
        self.login(ADMIN_USERNAME)
        uri = f"api/v1/chart/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 chart to favorites
        """
        chart = Slice(
            id=100,
            datasource_id=1,
            datasource_type="table",
            datasource_name="tmp_perm_table",
            slice_name="slice_name",
        )
        db.session.add(chart)
        db.session.commit()

        self.login(ADMIN_USERNAME)
        uri = f"api/v1/chart/favorite_status/?q={prison.dumps([chart.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/chart/{chart.id}/favorites/"
        self.client.post(uri)

        uri = f"api/v1/chart/favorite_status/?q={prison.dumps([chart.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(chart)
        db.session.commit()

    def test_remove_favorite(self):
        """
        Dataset API: Test remove chart from favorites
        """
        chart = Slice(
            id=100,
            datasource_id=1,
            datasource_type="table",
            datasource_name="tmp_perm_table",
            slice_name="slice_name",
        )
        db.session.add(chart)
        db.session.commit()

        self.login(ADMIN_USERNAME)
        uri = f"api/v1/chart/{chart.id}/favorites/"
        self.client.post(uri)

        uri = f"api/v1/chart/favorite_status/?q={prison.dumps([chart.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/chart/{chart.id}/favorites/"
        self.client.delete(uri)

        uri = f"api/v1/chart/favorite_status/?q={prison.dumps([chart.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(chart)
        db.session.commit()

    def test_get_time_range(self):
        """
        Chart API: Test get actually time range from human readable string
        """
        self.login(ADMIN_USERNAME)
        humanize_time_range = "100 years ago : now"
        uri = f"api/v1/time_range/?q={prison.dumps(humanize_time_range)}"
        rv = self.client.get(uri)
        data = json.loads(rv.data.decode("utf-8"))
        self.assertEqual(rv.status_code, 200)
        assert "since" in data["result"][0]
        assert "until" in data["result"][0]
        assert "timeRange" in data["result"][0]

        humanize_time_range = [
            {"timeRange": "2021-01-01 : 2022-02-01"},
            {"timeRange": "2022-01-01 : 2023-02-01"},
        ]
        uri = f"api/v1/time_range/?q={prison.dumps(humanize_time_range)}"
        rv = self.client.get(uri)
        data = json.loads(rv.data.decode("utf-8"))
        assert rv.status_code == 200
        assert len(data["result"]) == 2
        assert "since" in data["result"][0]
        assert "until" in data["result"][0]
        assert "timeRange" in data["result"][0]

        humanize_time_range = [
            {"timeRange": "2021-01-01 : 2022-02-01", "shift": "1 year ago"},
            {"timeRange": "2022-01-01 : 2023-02-01", "shift": "2 year ago"},
        ]
        uri = f"api/v1/time_range/?q={prison.dumps(humanize_time_range)}"
        rv = self.client.get(uri)
        data = json.loads(rv.data.decode("utf-8"))
        assert rv.status_code == 200
        assert len(data["result"]) == 2
        assert "since" in data["result"][0]
        assert "until" in data["result"][0]
        assert "timeRange" in data["result"][0]
        assert "shift" in data["result"][0]

    def test_query_form_data(self):
        """
        Chart API: Test query form data
        """
        self.login(ADMIN_USERNAME)
        slice = db.session.query(Slice).first()
        uri = f"api/v1/form_data/?slice_id={slice.id if slice else None}"
        rv = self.client.get(uri)
        data = json.loads(rv.data.decode("utf-8"))
        self.assertEqual(rv.status_code, 200)
        self.assertEqual(rv.content_type, "application/json")
        if slice:
            self.assertEqual(data["slice_id"], slice.id)

    @pytest.mark.usefixtures(
        "load_unicode_dashboard_with_slice",
        "load_energy_table_with_slice",
        "load_world_bank_dashboard_with_slices",
        "load_birth_names_dashboard_with_slices",
    )
    def test_get_charts_page(self):
        """
        Chart API: Test get charts filter
        """
        # Assuming we have 33 sample charts
        self.login(ADMIN_USERNAME)
        arguments = {"page_size": 10, "page": 0}
        uri = f"api/v1/chart/?q={prison.dumps(arguments)}"
        rv = self.client.get(uri)
        self.assertEqual(rv.status_code, 200)
        data = json.loads(rv.data.decode("utf-8"))
        self.assertEqual(len(data["result"]), 10)

        arguments = {"page_size": 10, "page": 3}
        uri = f"api/v1/chart/?q={prison.dumps(arguments)}"
        rv = self.get_assert_metric(uri, "get_list")
        self.assertEqual(rv.status_code, 200)
        data = json.loads(rv.data.decode("utf-8"))
        self.assertEqual(len(data["result"]), 3)

    def test_get_charts_no_data_access(self):
        """
        Chart API: Test get charts no data access
        """
        self.login(GAMMA_USERNAME)
        uri = "api/v1/chart/"
        rv = self.get_assert_metric(uri, "get_list")
        self.assertEqual(rv.status_code, 200)
        data = json.loads(rv.data.decode("utf-8"))
        self.assertEqual(data["count"], 0)

    def test_export_chart(self):
        """
        Chart API: Test export chart
        """
        example_chart = db.session.query(Slice).all()[0]
        argument = [example_chart.id]
        uri = f"api/v1/chart/export/?q={prison.dumps(argument)}"

        self.login(ADMIN_USERNAME)
        rv = self.get_assert_metric(uri, "export")

        assert rv.status_code == 200

        buf = BytesIO(rv.data)
        assert is_zipfile(buf)

    def test_export_chart_not_found(self):
        """
        Chart API: Test export chart not found
        """
        # Just one does not exist and we get 404
        argument = [-1, 1]
        uri = f"api/v1/chart/export/?q={prison.dumps(argument)}"
        self.login(ADMIN_USERNAME)
        rv = self.get_assert_metric(uri, "export")

        assert rv.status_code == 404

    def test_export_chart_gamma(self):
        """
        Chart API: Test export chart has gamma
        """
        example_chart = db.session.query(Slice).all()[0]
        argument = [example_chart.id]
        uri = f"api/v1/chart/export/?q={prison.dumps(argument)}"

        self.login(GAMMA_USERNAME)
        rv = self.client.get(uri)

        assert rv.status_code == 404

    @patch("superset.commands.database.importers.v1.utils.add_permissions")
    def test_import_chart(self, mock_add_permissions):
        """
        Chart API: Test import chart
        """
        self.login(ADMIN_USERNAME)
        uri = "api/v1/chart/import/"

        buf = self.create_chart_import()
        form_data = {
            "formData": (buf, "chart_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"}

        database = (
            db.session.query(Database).filter_by(uuid=database_config["uuid"]).one()
        )
        assert database.database_name == "imported_database"

        assert len(database.tables) == 1
        dataset = database.tables[0]
        assert dataset.table_name == "imported_dataset"
        assert str(dataset.uuid) == dataset_config["uuid"]

        chart = db.session.query(Slice).filter_by(uuid=chart_config["uuid"]).one()
        assert chart.table == dataset

        db.session.delete(chart)
        db.session.commit()
        db.session.delete(dataset)
        db.session.commit()
        db.session.delete(database)
        db.session.commit()

    @patch("superset.commands.database.importers.v1.utils.add_permissions")
    def test_import_chart_overwrite(self, mock_add_permissions):
        """
        Chart API: Test import existing chart
        """
        self.login(ADMIN_USERNAME)
        uri = "api/v1/chart/import/"

        buf = self.create_chart_import()
        form_data = {
            "formData": (buf, "chart_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_chart_import()
        form_data = {
            "formData": (buf, "chart_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 chart",
                    "error_type": "GENERIC_COMMAND_ERROR",
                    "level": "warning",
                    "extra": {
                        "charts/imported_chart.yaml": "Chart already exists and `overwrite=true` was not passed",
                        "issue_codes": [
                            {
                                "code": 1010,
                                "message": "Issue 1010 - Superset encountered an error while running a command.",
                            }
                        ],
                    },
                }
            ]
        }

        # import with overwrite flag
        buf = self.create_chart_import()
        form_data = {
            "formData": (buf, "chart_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"}

        # clean up
        database = (
            db.session.query(Database).filter_by(uuid=database_config["uuid"]).one()
        )
        dataset = database.tables[0]
        chart = db.session.query(Slice).filter_by(uuid=chart_config["uuid"]).one()

        db.session.delete(chart)
        db.session.commit()
        db.session.delete(dataset)
        db.session.commit()
        db.session.delete(database)
        db.session.commit()

    @patch("superset.commands.database.importers.v1.utils.add_permissions")
    def test_import_chart_invalid(self, mock_add_permissions):
        """
        Chart API: Test import invalid chart
        """
        self.login(ADMIN_USERNAME)
        uri = "api/v1/chart/import/"

        buf = BytesIO()
        with ZipFile(buf, "w") as bundle:
            with bundle.open("chart_export/metadata.yaml", "w") as fp:
                fp.write(yaml.safe_dump(dataset_metadata_config).encode())
            with bundle.open(
                "chart_export/databases/imported_database.yaml", "w"
            ) as fp:
                fp.write(yaml.safe_dump(database_config).encode())
            with bundle.open("chart_export/datasets/imported_dataset.yaml", "w") as fp:
                fp.write(yaml.safe_dump(dataset_config).encode())
            with bundle.open("chart_export/charts/imported_chart.yaml", "w") as fp:
                fp.write(yaml.safe_dump(chart_config).encode())
        buf.seek(0)

        form_data = {
            "formData": (buf, "chart_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 chart",
                    "error_type": "GENERIC_COMMAND_ERROR",
                    "level": "warning",
                    "extra": {
                        "metadata.yaml": {"type": ["Must be equal to Slice."]},
                        "issue_codes": [
                            {
                                "code": 1010,
                                "message": (
                                    "Issue 1010 - Superset encountered an "
                                    "error while running a command."
                                ),
                            }
                        ],
                    },
                }
            ]
        }

    def test_gets_created_by_user_charts_filter(self):
        arguments = {
            "filters": [{"col": "id", "opr": "chart_has_created_by", "value": True}],
            "keys": ["none"],
            "columns": ["slice_name"],
        }
        self.login(ADMIN_USERNAME)

        uri = f"api/v1/chart/?q={prison.dumps(arguments)}"
        rv = self.get_assert_metric(uri, "get_list")
        self.assertEqual(rv.status_code, 200)
        data = json.loads(rv.data.decode("utf-8"))
        self.assertEqual(data["count"], 8)

    def test_gets_not_created_by_user_charts_filter(self):
        arguments = {
            "filters": [{"col": "id", "opr": "chart_has_created_by", "value": False}],
            "keys": ["none"],
            "columns": ["slice_name"],
        }
        self.login(ADMIN_USERNAME)

        uri = f"api/v1/chart/?q={prison.dumps(arguments)}"
        rv = self.get_assert_metric(uri, "get_list")
        self.assertEqual(rv.status_code, 200)
        data = json.loads(rv.data.decode("utf-8"))
        self.assertEqual(data["count"], 8)

    @pytest.mark.usefixtures("create_charts")
    def test_gets_owned_created_favorited_by_me_filter(self):
        """
        Chart API: Test ChartOwnedCreatedFavoredByMeFilter
        """
        self.login(ADMIN_USERNAME)
        arguments = {
            "filters": [
                {
                    "col": "id",
                    "opr": "chart_owned_created_favored_by_me",
                    "value": True,
                }
            ],
            "order_column": "slice_name",
            "order_direction": "asc",
            "page": 0,
            "page_size": 25,
        }
        rv = self.client.get(f"api/v1/chart/?q={prison.dumps(arguments)}")
        self.assertEqual(rv.status_code, 200)
        data = json.loads(rv.data.decode("utf-8"))

        assert data["result"][0]["slice_name"] == "name0"
        assert data["result"][0]["datasource_id"] == 1

    @parameterized.expand(
        [
            "Top 10 Girl Name Share",  # Legacy chart
            "Pivot Table v2",  # Non-legacy chart
        ],
    )
    @pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
    def test_warm_up_cache(self, slice_name):
        self.login(ADMIN_USERNAME)
        slc = self.get_slice(slice_name)
        rv = self.client.put("/api/v1/chart/warm_up_cache", json={"chart_id": slc.id})
        self.assertEqual(rv.status_code, 200)
        data = json.loads(rv.data.decode("utf-8"))

        self.assertEqual(
            data["result"],
            [{"chart_id": slc.id, "viz_error": None, "viz_status": "success"}],
        )

        dashboard = self.get_dash_by_slug("births")

        rv = self.client.put(
            "/api/v1/chart/warm_up_cache",
            json={"chart_id": slc.id, "dashboard_id": dashboard.id},
        )
        self.assertEqual(rv.status_code, 200)
        data = json.loads(rv.data.decode("utf-8"))
        self.assertEqual(
            data["result"],
            [{"chart_id": slc.id, "viz_error": None, "viz_status": "success"}],
        )

        rv = self.client.put(
            "/api/v1/chart/warm_up_cache",
            json={
                "chart_id": slc.id,
                "dashboard_id": dashboard.id,
                "extra_filters": json.dumps(
                    [{"col": "name", "op": "in", "val": ["Jennifer"]}]
                ),
            },
        )
        self.assertEqual(rv.status_code, 200)
        data = json.loads(rv.data.decode("utf-8"))
        self.assertEqual(
            data["result"],
            [{"chart_id": slc.id, "viz_error": None, "viz_status": "success"}],
        )

    def test_warm_up_cache_chart_id_required(self):
        self.login(ADMIN_USERNAME)
        rv = self.client.put("/api/v1/chart/warm_up_cache", json={"dashboard_id": 1})
        self.assertEqual(rv.status_code, 400)
        data = json.loads(rv.data.decode("utf-8"))
        self.assertEqual(
            data,
            {"message": {"chart_id": ["Missing data for required field."]}},
        )

    def test_warm_up_cache_chart_not_found(self):
        self.login(ADMIN_USERNAME)
        rv = self.client.put("/api/v1/chart/warm_up_cache", json={"chart_id": 99999})
        self.assertEqual(rv.status_code, 404)
        data = json.loads(rv.data.decode("utf-8"))
        self.assertEqual(data, {"message": "Chart not found"})

    def test_warm_up_cache_payload_validation(self):
        self.login(ADMIN_USERNAME)
        rv = self.client.put(
            "/api/v1/chart/warm_up_cache",
            json={"chart_id": "id", "dashboard_id": "id", "extra_filters": 4},
        )
        self.assertEqual(rv.status_code, 400)
        data = json.loads(rv.data.decode("utf-8"))
        self.assertEqual(
            data,
            {
                "message": {
                    "chart_id": ["Not a valid integer."],
                    "dashboard_id": ["Not a valid integer."],
                    "extra_filters": ["Not a valid string."],
                }
            },
        )

    @pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
    def test_warm_up_cache_error(self) -> None:
        self.login(ADMIN_USERNAME)
        slc = self.get_slice("Pivot Table v2")

        with mock.patch.object(ChartDataCommand, "run") as mock_run:
            mock_run.side_effect = ChartDataQueryFailedError(
                _(
                    "Error: %(error)s",
                    error=_("Empty query?"),
                )
            )

            assert json.loads(
                self.client.put(
                    "/api/v1/chart/warm_up_cache",
                    json={"chart_id": slc.id},
                ).data
            ) == {
                "result": [
                    {
                        "chart_id": slc.id,
                        "viz_error": "Error: Empty query?",
                        "viz_status": None,
                    },
                ],
            }

    @pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
    def test_warm_up_cache_no_query_context(self) -> None:
        self.login(ADMIN_USERNAME)
        slc = self.get_slice("Pivot Table v2")

        with mock.patch.object(Slice, "get_query_context") as mock_get_query_context:
            mock_get_query_context.return_value = None

            assert json.loads(
                self.client.put(
                    "/api/v1/chart/warm_up_cache",  # noqa: F541
                    json={"chart_id": slc.id},
                ).data
            ) == {
                "result": [
                    {
                        "chart_id": slc.id,
                        "viz_error": "Chart's query context does not exist",
                        "viz_status": None,
                    },
                ],
            }

    @pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
    def test_warm_up_cache_no_datasource(self) -> None:
        self.login(ADMIN_USERNAME)
        slc = self.get_slice("Top 10 Girl Name Share")

        with mock.patch.object(
            Slice,
            "datasource",
            new_callable=mock.PropertyMock,
        ) as mock_datasource:
            mock_datasource.return_value = None

            assert json.loads(
                self.client.put(
                    "/api/v1/chart/warm_up_cache",  # noqa: F541
                    json={"chart_id": slc.id},
                ).data
            ) == {
                "result": [
                    {
                        "chart_id": slc.id,
                        "viz_error": "Chart's datasource does not exist",
                        "viz_status": None,
                    },
                ],
            }

    @pytest.mark.usefixtures("create_chart_with_tag")
    def test_update_chart_add_tags_can_write_on_tag(self):
        """
        Validates a user with can write on tag permission can
        add tags while updating a chart
        """
        self.login(ADMIN_USERNAME)

        chart = (
            db.session.query(Slice).filter(Slice.slice_name == "chart 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 chart.tags if tag.type == TagType.custom]
        new_tags.append(new_tag.id)
        update_payload = {"tags": new_tags}

        uri = f"api/v1/chart/{chart.id}"
        rv = self.put_assert_metric(uri, update_payload, "put")
        self.assertEqual(rv.status_code, 200)
        model = db.session.query(Slice).get(chart.id)

        # Clean up system tags
        tag_list = [tag.id for tag in model.tags if tag.type == TagType.custom]
        self.assertEqual(tag_list, new_tags)

    @pytest.mark.usefixtures("create_chart_with_tag")
    def test_update_chart_remove_tags_can_write_on_tag(self):
        """
        Validates a user with can write on tag permission can
        remove tags while updating a chart
        """
        self.login(ADMIN_USERNAME)

        chart = (
            db.session.query(Slice).filter(Slice.slice_name == "chart with tag").first()
        )

        # get existing tag and add a new one
        new_tags = [tag.id for tag in chart.tags if tag.type == TagType.custom]
        new_tags.pop()

        update_payload = {"tags": new_tags}

        uri = f"api/v1/chart/{chart.id}"
        rv = self.put_assert_metric(uri, update_payload, "put")
        self.assertEqual(rv.status_code, 200)
        model = db.session.query(Slice).get(chart.id)

        # Clean up system tags
        tag_list = [tag.id for tag in model.tags if tag.type == TagType.custom]
        self.assertEqual(tag_list, new_tags)

    @pytest.mark.usefixtures("create_chart_with_tag")
    def test_update_chart_add_tags_can_tag_on_chart(self):
        """
        Validates an owner with can tag on chart permission can
        add tags while updating a chart
        """
        self.login(ALPHA_USERNAME)

        alpha_role = security_manager.find_role("Alpha")
        write_tags_perm = security_manager.add_permission_view_menu("can_write", "Tag")
        security_manager.del_permission_role(alpha_role, write_tags_perm)
        assert "can tag on Chart" in str(alpha_role.permissions)

        chart = (
            db.session.query(Slice).filter(Slice.slice_name == "chart 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 chart.tags if tag.type == TagType.custom]
        new_tags.append(new_tag.id)
        update_payload = {"tags": new_tags}

        uri = f"api/v1/chart/{chart.id}"
        rv = self.put_assert_metric(uri, update_payload, "put")
        self.assertEqual(rv.status_code, 200)
        model = db.session.query(Slice).get(chart.id)

        # Clean up system tags
        tag_list = [tag.id for tag in model.tags if tag.type == TagType.custom]
        self.assertEqual(tag_list, new_tags)

        security_manager.add_permission_role(alpha_role, write_tags_perm)

    @pytest.mark.usefixtures("create_chart_with_tag")
    def test_update_chart_remove_tags_can_tag_on_chart(self):
        """
        Validates an owner with can tag on chart permission can
        remove tags from a chart
        """
        self.login(ALPHA_USERNAME)

        alpha_role = security_manager.find_role("Alpha")
        write_tags_perm = security_manager.add_permission_view_menu("can_write", "Tag")
        security_manager.del_permission_role(alpha_role, write_tags_perm)
        assert "can tag on Chart" in str(alpha_role.permissions)

        chart = (
            db.session.query(Slice).filter(Slice.slice_name == "chart with tag").first()
        )

        update_payload = {"tags": []}

        uri = f"api/v1/chart/{chart.id}"
        rv = self.put_assert_metric(uri, update_payload, "put")
        self.assertEqual(rv.status_code, 200)
        model = db.session.query(Slice).get(chart.id)

        # Clean up system tags
        tag_list = [tag.id for tag in model.tags if tag.type == TagType.custom]
        self.assertEqual(tag_list, [])

        security_manager.add_permission_role(alpha_role, write_tags_perm)

    @pytest.mark.usefixtures("create_chart_with_tag")
    def test_update_chart_add_tags_missing_permission(self):
        """
        Validates an owner can't add tags to a chart if they don't
        have permission to it
        """
        self.login(ALPHA_USERNAME)

        alpha_role = security_manager.find_role("Alpha")
        write_tags_perm = security_manager.add_permission_view_menu("can_write", "Tag")
        tag_charts_perm = security_manager.add_permission_view_menu("can_tag", "Chart")
        security_manager.del_permission_role(alpha_role, write_tags_perm)
        security_manager.del_permission_role(alpha_role, tag_charts_perm)

        chart = (
            db.session.query(Slice).filter(Slice.slice_name == "chart 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 chart.tags if tag.type == TagType.custom]
        new_tags.append(new_tag.id)
        update_payload = {"tags": new_tags}

        uri = f"api/v1/chart/{chart.id}"
        rv = self.put_assert_metric(uri, update_payload, "put")
        self.assertEqual(rv.status_code, 403)
        self.assertEqual(
            rv.json["message"],
            "You do not have permission to manage tags on charts",
        )

        security_manager.add_permission_role(alpha_role, write_tags_perm)
        security_manager.add_permission_role(alpha_role, tag_charts_perm)

    @pytest.mark.usefixtures("create_chart_with_tag")
    def test_update_chart_remove_tags_missing_permission(self):
        """
        Validates an owner can't remove tags from a chart if they don't
        have permission to it
        """
        self.login(ALPHA_USERNAME)

        alpha_role = security_manager.find_role("Alpha")
        write_tags_perm = security_manager.add_permission_view_menu("can_write", "Tag")
        tag_charts_perm = security_manager.add_permission_view_menu("can_tag", "Chart")
        security_manager.del_permission_role(alpha_role, write_tags_perm)
        security_manager.del_permission_role(alpha_role, tag_charts_perm)

        chart = (
            db.session.query(Slice).filter(Slice.slice_name == "chart with tag").first()
        )

        update_payload = {"tags": []}

        uri = f"api/v1/chart/{chart.id}"
        rv = self.put_assert_metric(uri, update_payload, "put")
        self.assertEqual(rv.status_code, 403)
        self.assertEqual(
            rv.json["message"],
            "You do not have permission to manage tags on charts",
        )

        security_manager.add_permission_role(alpha_role, write_tags_perm)
        security_manager.add_permission_role(alpha_role, tag_charts_perm)

    @pytest.mark.usefixtures("create_chart_with_tag")
    def test_update_chart_no_tag_changes(self):
        """
        Validates an owner without permission to change tags is able to
        update a chart when tags haven't changed
        """
        self.login(ALPHA_USERNAME)

        alpha_role = security_manager.find_role("Alpha")
        write_tags_perm = security_manager.add_permission_view_menu("can_write", "Tag")
        tag_charts_perm = security_manager.add_permission_view_menu("can_tag", "Chart")
        security_manager.del_permission_role(alpha_role, write_tags_perm)
        security_manager.del_permission_role(alpha_role, tag_charts_perm)

        chart = (
            db.session.query(Slice).filter(Slice.slice_name == "chart with tag").first()
        )
        existing_tags = [tag.id for tag in chart.tags if tag.type == TagType.custom]
        update_payload = {"tags": existing_tags}

        uri = f"api/v1/chart/{chart.id}"
        rv = self.put_assert_metric(uri, update_payload, "put")
        self.assertEqual(rv.status_code, 200)

        security_manager.add_permission_role(alpha_role, write_tags_perm)
        security_manager.add_permission_role(alpha_role, tag_charts_perm)