superset2/tests/integration_tests/charts/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.

import uuid
from io import BytesIO
from unittest import mock
from unittest.mock import patch
from zipfile import is_zipfile

import pytest
import rison
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.models.sql_lab import SavedQuery
from superset.reports.models import ReportSchedule, ReportScheduleType
from superset.subjects.models import Subject
from superset.subjects.types import SubjectType
from superset.tags.models import ObjectType, Tag, TaggedObject, TagType
from superset.utils import json
from superset.utils.core import get_example_default_schema
from superset.utils.database import get_example_database
from tests.integration_tests.base_api_tests import ApiEditorsTestCaseMixin
from tests.integration_tests.base_tests import (
    subjects_from_users,
    SupersetTestCase,
    user_is_editor,
)
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,
    database_config,
    dataset_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

CHARTS_FIXTURE_COUNT = 10


class TestChartApi(ApiEditorsTestCaseMixin, InsertChartMixin, SupersetTestCase):
    resource_name = "chart"
    subject_types_config_key = "SUBJECTS_RELATED_TYPES_CHARTS"

    @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,
            )
            # SQLAlchemy 2.0 removes the legacy cascade_backrefs behavior, so
            # assigning `chart=chart` on a transient ReportSchedule no longer
            # implicitly adds it to the session via the Slice.report_schedules
            # backref - it must be added explicitly.
            db.session.add(report_schedule)
            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.editors = subjects_from_users([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.editors = subjects_from_users([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:
                if db.session.query(TaggedObject).filter_by(id=association.id).first():
                    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={rison.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 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")
        assert rv.status_code == 200
        model = db.session.query(Slice).get(chart_id)
        assert model is None

    def test_delete_bulk_charts(self):
        """
        Chart API: Test delete bulk
        """
        admin = self.get_user("admin")
        chart_count = 4
        chart_ids = list()  # noqa: C408
        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={rison.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 {chart_count} charts"}
        assert response == expected_response
        for chart_id in chart_ids:
            model = db.session.query(Slice).get(chart_id)
            assert model is 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={rison.dumps(argument)}"
        rv = self.delete_assert_metric(uri, "bulk_delete")
        assert 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")
        assert 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"))
        assert rv.status_code == 422
        expected_response = {
            "message": (
                "This chart is used by alerts or reports: report_with_chart. "
                "Detach or delete them first."
            )
        }
        assert 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={rison.dumps(chart_ids)}"
        rv = self.delete_assert_metric(uri, "bulk_delete")
        assert 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={rison.dumps(chart_ids)}"
        rv = self.client.delete(uri)
        response = json.loads(rv.data.decode("utf-8"))
        assert rv.status_code == 422
        expected_response = {
            "message": (
                'Chart "chart_report" is used by alerts or reports: '
                "report_with_chart. Detach or delete them first."
            )
        }
        assert 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")
        assert rv.status_code == 200
        model = db.session.query(Slice).get(chart_id)
        assert model is 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()  # noqa: C408
        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={rison.dumps(argument)}"
        rv = self.delete_assert_metric(uri, "bulk_delete")
        response = json.loads(rv.data.decode("utf-8"))
        assert rv.status_code == 200
        expected_response = {"message": f"Deleted {chart_count} charts"}
        assert response == expected_response

        for chart_id in chart_ids:
            model = db.session.query(Slice).get(chart_id)
            assert model is 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")  # noqa: S106
        uri = f"api/v1/chart/{chart.id}"
        rv = self.delete_assert_metric(uri, "delete")
        assert 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()  # noqa: C408
        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")  # noqa: S106

        # verify we can't delete not owned charts
        arguments = [chart.id for chart in charts]
        uri = f"api/v1/chart/?q={rison.dumps(arguments)}"
        rv = self.delete_assert_metric(uri, "bulk_delete")
        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 charts
        arguments = [chart.id for chart in charts] + [owned_chart.id]
        uri = f"api/v1/chart/?q={rison.dumps(arguments)}"
        rv = self.delete_assert_metric(uri, "bulk_delete")
        assert rv.status_code == 403
        response = json.loads(rv.data.decode("utf-8"))
        expected_response = {"message": "Forbidden"}
        assert 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"])
        chart_data = {
            "slice_name": "name1",
            "description": "description1",
            "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")
        assert rv.status_code == 201
        data = json.loads(rv.data.decode("utf-8"))
        model = db.session.query(Slice).get(data.get("id"))
        # uuid should be returned in the response
        assert "uuid" in data
        assert str(model.uuid) == str(data["uuid"])
        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")
        assert 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_editors(self):
        """
        Chart API: Test create validate editors (subjects)
        """
        chart_data = {
            "slice_name": "title1",
            "datasource_id": 1,
            "datasource_type": "table",
            "editors": [1000],
        }
        self.login(ADMIN_USERNAME)
        uri = "api/v1/chart/"
        rv = self.post_assert_metric(uri, chart_data, "post")
        assert rv.status_code == 422
        response = json.loads(rv.data.decode("utf-8"))
        expected_response = {"message": {"editors": ["Subjects are invalid"]}}
        assert 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")
        assert 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")
        assert rv.status_code == 400
        response = json.loads(rv.data.decode("utf-8"))
        assert response == {
            "message": {
                "datasource_type": [
                    "Must be one of: table, dataset, query, saved_query, view, "
                    "semantic_view."
                ]
            }
        }
        chart_data = {
            "slice_name": "title1",
            "datasource_id": 0,
            "datasource_type": "table",
        }
        rv = self.post_assert_metric("/api/v1/chart/", chart_data, "post")
        assert rv.status_code == 422
        response = json.loads(rv.data.decode("utf-8"))
        assert response == {"message": {"datasource_id": ["Datasource does not exist"]}}

    def test_create_chart_from_saved_query_rejected_cleanly(self):
        """
        Chart API: creating a chart with datasource_type="saved_query" must
        fail with a clean validation error, not the unhandled 500 "Fatal
        error" reported in apache/superset#29697. Slice.datasource only
        ever resolves the "table" relationship, so even a chart that
        "created" successfully with this datasource_type could never
        actually render -- "saved_query" is a real, existing row here
        (not a bad ID), reproducing the original report exactly rather
        than a not-found case.
        """
        self.login(ADMIN_USERNAME)
        example_db = get_example_database()
        saved_query = SavedQuery(
            db_id=example_db.id,
            label="issue-29697-repro",
            schema=get_example_default_schema(),
            sql="SELECT 1 AS value",
        )
        db.session.add(saved_query)
        db.session.commit()
        saved_query_id = saved_query.id

        chart_data = {
            "slice_name": "issue-29697-repro-chart",
            "datasource_id": saved_query_id,
            "datasource_type": "saved_query",
            "viz_type": "table",
        }
        rv = self.post_assert_metric("/api/v1/chart/", chart_data, "post")

        db.session.delete(db.session.query(SavedQuery).get(saved_query_id))
        db.session.commit()

        assert rv.status_code == 422
        response = json.loads(rv.data.decode("utf-8"))
        assert response == {
            "message": {"datasource_type": ["Datasource type is invalid"]}
        }

    @pytest.mark.usefixtures("load_world_bank_dashboard_with_slices")
    def test_create_chart_validate_user_is_dashboard_editor(self):
        """
        Chart API: Test create validates user is dashboard editor
        """
        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")
        assert rv.status_code == 403
        response = json.loads(rv.data.decode("utf-8"))
        assert 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")
        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",
            "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")
        assert rv.status_code == 200
        model = db.session.query(Slice).get(chart_id)
        related_dashboard = db.session.query(Dashboard).filter_by(slug="births").first()
        assert model.created_by == admin
        assert model.slice_name == "title1_changed"
        assert model.description == "description1"
        assert user_is_editor(admin, model)
        assert model.viz_type == "viz_type1"
        assert model.params == '{"a": 1}'
        assert model.cache_timeout == 1000
        assert model.datasource_id == birth_names_table_id
        assert model.datasource_type == "table"
        assert model.datasource_name == full_table_name
        assert model.certified_by == "Mario Rossi"
        assert model.certification_details == "Edited certification"
        assert model.id in [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"),
        }
        self.login(ADMIN_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)

        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]
        assert current_chart["slice_name"] == new_name
        assert "username" not in current_chart["changed_by"].keys()
        assert len(current_chart["editors"]) > 0

        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 in editors
        """
        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"),
        }
        self.login(ADMIN_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)

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

        assert res["slice_name"] == new_name
        assert len(res["editors"]) > 0
        assert "username" not in res["editors"][0].keys()

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

    def test_update_chart_preserves_editors_not_admin(self):
        """
        Chart API: Test update preserves editors when non-admin updates chart
        """
        gamma = self.get_user("gamma_no_csv")
        chart_id = self.insert_chart("title", [gamma.id], 1).id
        chart_data = {
            "slice_name": (new_name := "title1_changed"),
        }
        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 user_is_editor(gamma, model)
        db.session.delete(model)
        db.session.commit()

    def test_update_chart_preserves_editors_admin(self):
        """
        Chart API: Test update as admin preserves editors
        """
        admin = self.get_user("admin")
        chart_id = self.insert_chart("title", [admin.id], 1).id
        chart_data = {"slice_name": "title1_changed"}
        self.login(ADMIN_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 user_is_editor(admin, model)
        db.session.delete(model)
        db.session.commit()

    @pytest.mark.usefixtures("add_dashboard_to_chart")
    def test_update_chart_preserves_editors(self):
        """
        Chart API: Test update chart preserves editors (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")
        assert rv.status_code == 200
        assert len(self.chart.editors) == 1
        assert user_is_editor(admin, self.chart)

    @pytest.mark.usefixtures("add_dashboard_to_chart")
    def test_update_chart_clear_editor_list(self):
        """
        Chart API: Test update chart admin can clear editor list
        """
        chart_data = {"slice_name": "title1_changed", "editors": []}
        self.login(username="admin")
        uri = f"api/v1/chart/{self.chart.id}"
        rv = self.put_assert_metric(uri, chart_data, "put")
        assert rv.status_code == 200
        assert self.chart.editors == []

    def test_update_chart_populate_editor(self):
        """
        Chart API: Test update admin can update chart with
        no editors to a different editor
        """
        gamma = self.get_user("gamma")
        chart_id = self.insert_chart("title", [], 1).id
        model = db.session.query(Slice).get(chart_id)
        assert model.editors == []
        gamma_subject = (
            db.session.query(Subject)
            .filter_by(user_id=gamma.id, type=SubjectType.USER)
            .first()
        )
        chart_data = {"editors": [gamma_subject.id]}
        self.login(username="admin")
        uri = f"api/v1/chart/{chart_id}"
        rv = self.put_assert_metric(uri, chart_data, "put")
        assert rv.status_code == 200
        model_updated = db.session.query(Slice).get(chart_id)
        assert user_is_editor(gamma, model_updated)
        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")
        assert rv.status_code == 200
        assert self.new_dashboard in self.chart.dashboards
        assert self.original_dashboard not in 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")
        assert rv.status_code == 200
        assert self.original_dashboard in self.chart.dashboards
        assert 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")  # noqa: S106
        chart_data = {"slice_name": "title1_changed"}
        uri = f"api/v1/chart/{chart.id}"
        rv = self.put_assert_metric(uri, chart_data, "put")
        assert 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.editors = subjects_from_users([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.editors = subjects_from_users([user_alpha2])
        new_dashboard.slices = [chart]
        new_dashboard.published = False
        db.session.add(new_dashboard)

        self.login(username="alpha1", password="password")  # noqa: S106
        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")
        assert rv.status_code == 422
        response = json.loads(rv.data.decode("utf-8"))
        expected_response = {"message": {"dashboards": ["Dashboards do not exist"]}}
        assert response == expected_response

        rv = self.put_assert_metric(uri, chart_data, "put")
        assert 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", editor_user_ids=[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")
        assert rv.status_code == 400
        response = json.loads(rv.data.decode("utf-8"))
        assert response == {
            "message": {
                "datasource_type": [
                    "Must be one of: table, dataset, query, saved_query, view, "
                    "semantic_view."
                ]
            }
        }

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

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

    def test_update_chart_validate_editors(self):
        """
        Chart API: Test update validate editors (subjects)
        """
        chart_data = {
            "slice_name": "title1",
            "datasource_id": 1,
            "datasource_type": "table",
            "editors": [1000],
        }
        self.login(ADMIN_USERNAME)
        uri = "api/v1/chart/"  # noqa: F541
        rv = self.client.post(uri, json=chart_data)
        assert rv.status_code == 422
        response = json.loads(rv.data.decode("utf-8"))
        expected_response = {"message": {"editors": ["Subjects are invalid"]}}
        assert 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")
        assert rv.status_code == 200
        from unittest.mock import ANY

        expected_result = {
            "cache_timeout": None,
            "certified_by": None,
            "certification_details": None,
            "dashboards": [],
            "description": None,
            "editors": [
                {
                    "id": ANY,
                    "label": "admin user",
                    "secondary_label": "admin@fab.org",
                    "type": 1,
                    "img": ANY,
                }
            ],
            "viewers": [],
            "params": None,
            "slice_name": "title",
            "tags": [],
            "viz_type": None,
            "query_context": None,
            "is_managed_externally": False,
        }
        data = json.loads(rv.data.decode("utf-8"))
        assert "changed_on_delta_humanized" in data["result"]
        assert "id" in data["result"]
        assert "thumbnail_url" in data["result"]
        assert "url" in 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",
                "uuid",
                "datasource_id",
                "datasource_name_text",
                "datasource_type",
                "datasource_url",
                "datasource_uuid",
            ):
                assert 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")
        assert rv.status_code == 404

    @parameterized.expand(
        [
            ("by_id", lambda chart: str(chart.id), "id"),
            (
                "by_uuid",
                lambda chart: str(chart.uuid) if chart.uuid else pytest.skip("No UUID"),
                "uuid",
            ),
        ]
    )
    def test_slice_get_existing(self, test_name, get_identifier, field_type):
        """Test Slice.get() successfully retrieves existing charts."""
        admin = self.get_user("admin")
        chart = self.insert_chart(f"test_slice_get_{field_type}", [admin.id], 1)

        identifier = get_identifier(chart)
        result = Slice.get(identifier)

        assert result is not None
        assert result.id == chart.id
        if field_type == "uuid" and chart.uuid:
            assert result.uuid == chart.uuid

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

    @parameterized.expand(
        [
            ("nonexistent_id", "999999"),
            ("nonexistent_uuid", str(uuid.uuid4())),
        ]
    )
    def test_slice_get_not_found(self, test_name, identifier):
        """Test Slice.get() returns None for non-existent identifiers."""
        result = Slice.get(identifier)
        assert result is None

    @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)
        assert rv.status_code == 404

    @pytest.mark.usefixtures("load_energy_table_with_slice")
    def test_get_deck_layers(self):
        """
        Chart API: Test get deck.gl Multiple Layers container's declared layers

        The layer charts sit on no dashboard of their own, so they are
        resolved without the base filter, gated only on access to the
        container -- mirroring how the legacy explore_json pipeline
        resolved them server-side.
        """
        admin = self.get_user("admin")
        layer_one = self.insert_chart(
            "layer one",
            [admin.id],
            1,
            viz_type="deck_scatter",
            params=json.dumps({"viz_type": "deck_scatter"}),
        )
        layer_two = self.insert_chart(
            "layer two",
            [admin.id],
            1,
            viz_type="deck_scatter",
            params=json.dumps({"viz_type": "deck_scatter"}),
        )
        container = self.insert_chart(
            "deck multi container",
            [admin.id],
            1,
            viz_type="deck_multi",
            params=json.dumps(
                {
                    "viz_type": "deck_multi",
                    "deck_slices": [layer_one.id, layer_two.id],
                }
            ),
        )
        self.login(ADMIN_USERNAME)
        uri = f"api/v1/chart/{container.id}/deck_layers/"
        rv = self.get_assert_metric(uri, "deck_layers")
        assert rv.status_code == 200
        data = json.loads(rv.data.decode("utf-8"))
        assert [layer["slice_id"] for layer in data["result"]] == [
            layer_one.id,
            layer_two.id,
        ]
        assert data["result"][0]["viz_type"] == "deck_scatter"

        db.session.delete(layer_one)
        db.session.delete(layer_two)
        db.session.delete(container)
        db.session.commit()

    @pytest.mark.usefixtures("load_energy_table_with_slice")
    def test_get_deck_layers_no_container_access(self):
        """
        Chart API: Test get deck layers 404s when the container itself
        isn't accessible, regardless of the layers' own access.
        """
        admin = self.get_user("admin")
        layer_one = self.insert_chart(
            "layer one no access",
            [admin.id],
            1,
            viz_type="deck_scatter",
            params=json.dumps({"viz_type": "deck_scatter"}),
        )
        container = self.insert_chart(
            "deck multi container no access",
            [admin.id],
            1,
            viz_type="deck_multi",
            params=json.dumps(
                {"viz_type": "deck_multi", "deck_slices": [layer_one.id]}
            ),
        )
        self.login(GAMMA_USERNAME)
        uri = f"api/v1/chart/{container.id}/deck_layers/"
        rv = self.client.get(uri)
        assert rv.status_code == 404

        db.session.delete(layer_one)
        db.session.delete(container)
        db.session.commit()

    @pytest.mark.usefixtures("load_energy_table_with_slice")
    def test_get_deck_layers_omits_inaccessible_layer_for_ordinary_user(self):
        """
        Chart API: An ordinary (non-guest) user with access to the deck_multi
        container must not have an inaccessible layer's params/datasource
        leaked just because it's named in the container's `deck_slices` --
        that layer is silently omitted from the result instead.
        """
        admin = self.get_user("admin")
        gamma = self.get_user("gamma")
        layer_visible = self.insert_chart(
            "layer visible",
            [gamma.id],
            1,
            viz_type="deck_scatter",
            params=json.dumps({"viz_type": "deck_scatter"}),
        )
        layer_hidden = self.insert_chart(
            "layer hidden from gamma",
            [admin.id],
            1,
            viz_type="deck_scatter",
            params=json.dumps({"viz_type": "deck_scatter"}),
        )
        container = self.insert_chart(
            "deck multi container for gamma",
            [gamma.id],
            1,
            viz_type="deck_multi",
            params=json.dumps(
                {
                    "viz_type": "deck_multi",
                    "deck_slices": [layer_visible.id, layer_hidden.id],
                }
            ),
        )
        self.login(GAMMA_USERNAME)
        uri = f"api/v1/chart/{container.id}/deck_layers/"
        rv = self.get_assert_metric(uri, "deck_layers")
        assert rv.status_code == 200
        data = json.loads(rv.data.decode("utf-8"))
        assert [layer["slice_id"] for layer in data["result"]] == [
            layer_visible.id,
        ]

        db.session.delete(layer_visible)
        db.session.delete(layer_hidden)
        db.session.delete(container)
        db.session.commit()

    @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")
        assert rv.status_code == 200
        data = json.loads(rv.data.decode("utf-8"))
        assert 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={rison.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]["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={rison.dumps(arguments)}"
        rv = self.get_assert_metric(uri, "get_list")
        assert 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)
            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("chart_tags", tag.name)
            response_by_name = self.get_list("chart", 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_charts)
            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_charts)  # noqa: C401

    def test_get_charts_changed_on_delta_humanized_sort_monotonic(self):
        """Regression for #27500: sorting the chart list by
        `changed_on_delta_humanized` desc must yield results whose underlying
        `changed_on` timestamps are monotonically non-increasing. The original
        report shows the humanized column visually out of order, suggesting
        the sort key didn't actually reflect the timestamp."""
        from datetime import datetime, timedelta

        admin = self.get_user("admin")
        # Insert two charts with distinct changed_on timestamps. Use raw UPDATE
        # to force the values since assignment alone can be overridden by the
        # before-update hook.
        chart_older = self.insert_chart(
            "regression_27500_older", [admin.id], 1, description="z"
        )
        chart_newer = self.insert_chart(
            "regression_27500_newer", [admin.id], 1, description="z"
        )
        # Use timestamps whose humanized strings sort DIFFERENTLY from their
        # real timestamps under a naive lexical sort. "3 hours ago" vs
        # "5 hours ago": lexical-desc puts "5..." first (older), but
        # timestamp-desc must put "3..." first (newer). If the API ever
        # accidentally sorts by the humanized text instead of the column,
        # this test fails. (Pairs like "now"/"2 days ago" don't discriminate
        # because 'n' > '2' lexically agrees with newest-first.)
        now = datetime.utcnow()
        chart_older.changed_on = now - timedelta(hours=5)
        chart_newer.changed_on = now - timedelta(hours=3)
        db.session.commit()

        try:
            self.login(ADMIN_USERNAME)
            arguments = {
                "order_column": "changed_on_delta_humanized",
                "order_direction": "desc",
                "filters": [
                    {
                        "col": "slice_name",
                        "opr": "sw",
                        "value": "regression_27500_",
                    }
                ],
            }
            uri = f"api/v1/chart/?q={rison.dumps(arguments)}"
            rv = self.get_assert_metric(uri, "get_list")
            assert rv.status_code == 200
            data = json.loads(rv.data.decode("utf-8"))

            results = data["result"]
            assert len(results) >= 2, f"expected at least 2 results, got {results}"

            # The two inserted charts should appear in newest-first order.
            indices = {
                row["slice_name"]: i
                for i, row in enumerate(results)
                if row["slice_name"].startswith("regression_27500_")
            }
            ordering = [
                row["slice_name"]
                for row in results
                if row["slice_name"].startswith("regression_27500_")
            ]
            assert (
                indices["regression_27500_newer"] < indices["regression_27500_older"]
            ), f"changed_on_delta_humanized desc sort is wrong: {ordering}; see #27500"
        finally:
            db.session.delete(chart_older)
            db.session.delete(chart_newer)
            db.session.commit()

    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={rison.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"] 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={rison.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"] == 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={rison.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"] == 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"]):
            assert item["description"] == expected_response[index]["description"]
            assert item["slice_name"] == expected_response[index]["slice_name"]
            assert 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={rison.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={rison.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"] == 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={rison.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"] == 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={rison.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("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")
            assert rv.status_code == 200
            data = rv.get_json()
            return data["count"]

        with self.temporary_user(gamma_user, login=True):
            assert 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):
            assert 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={rison.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={rison.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={rison.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={rison.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={rison.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={rison.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={rison.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={rison.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={rison.dumps(humanize_time_range)}"
        rv = self.client.get(uri)
        data = json.loads(rv.data.decode("utf-8"))
        assert 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={rison.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={rison.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"))
        assert rv.status_code == 200
        assert rv.content_type == "application/json; charset=utf-8"
        if slice:
            assert data["slice_id"] == slice.id

    @pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
    def test_query_form_data_no_data_access(self):
        """
        Chart API: query_form_data must refuse callers without
        datasource_access on the chart's underlying dataset. Mirrors
        the existing test_get_chart_no_data_access guard on
        ChartRestApi (which also returns 404 to avoid leaking chart
        existence to unauthorised callers).
        """
        self.login(GAMMA_USERNAME)
        chart_no_access = (
            db.session.query(Slice)
            .filter_by(slice_name="Girl Name Cloud")
            .one_or_none()
        )
        assert chart_no_access is not None, (
            "fixture load_birth_names_dashboard_with_slices did not "
            "create the 'Girl Name Cloud' slice"
        )
        uri = f"api/v1/form_data/?slice_id={chart_no_access.id}"
        rv = self.client.get(uri)
        # Match ChartRestApi.get: 404 for both missing AND forbidden so
        # the endpoint cannot be used to enumerate chart IDs.
        assert rv.status_code == 404, (
            f"Gamma user without datasource_access should get 404 "
            f"(status={rv.status_code}, body={rv.data[:200]!r})"
        )
        # Defence in depth: even if a future regression returns a non-
        # 200 status with a partially-filled error envelope, ensure the
        # caller cannot recover form_data fields.
        assert b"datasource" not in rv.data
        assert b"adhoc_filters" not in rv.data
        assert b"viz_type" not in rv.data

    def test_query_form_data_missing_slice(self):
        """
        Chart API: a non-existent slice_id must return the same 404 as a
        forbidden one, so the status code cannot be used to enumerate
        which slice IDs exist.
        """
        self.login(ADMIN_USERNAME)
        max_id = db.session.query(func.max(Slice.id)).scalar() or 0
        uri = f"api/v1/form_data/?slice_id={max_id + 10_000}"
        rv = self.client.get(uri)
        assert rv.status_code == 404

    @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={rison.dumps(arguments)}"
        rv = self.client.get(uri)
        assert rv.status_code == 200
        data = json.loads(rv.data.decode("utf-8"))
        assert len(data["result"]) == 10

        arguments = {"page_size": 10, "page": 3}
        uri = f"api/v1/chart/?q={rison.dumps(arguments)}"
        rv = self.get_assert_metric(uri, "get_list")
        assert rv.status_code == 200
        data = json.loads(rv.data.decode("utf-8"))
        assert 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")
        assert rv.status_code == 200
        data = json.loads(rv.data.decode("utf-8"))
        assert 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={rison.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={rison.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={rison.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_import_v1_zip_file("chart")
        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_import_v1_zip_file("chart")
        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_import_v1_zip_file("chart")
        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 len(response["errors"]) == 1
        error = response["errors"][0]
        assert error["message"].startswith("Error importing chart")
        assert error["error_type"] == "GENERIC_COMMAND_ERROR"
        assert error["level"] == "warning"
        assert "charts/chart.yaml" in str(error["extra"])
        assert "Chart already exists and `overwrite=true` was not passed" in str(
            error["extra"]
        )
        assert error["extra"]["issue_codes"][0]["code"] == 1010

        # import with overwrite flag
        buf = self.create_import_v1_zip_file("chart")
        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 = self.create_import_v1_zip_file("dataset", charts=[chart_config])
        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 len(response["errors"]) == 1
        error = response["errors"][0]
        assert error["message"].startswith("Error importing chart")
        assert error["error_type"] == "GENERIC_COMMAND_ERROR"
        assert error["level"] == "warning"
        assert "metadata.yaml" in error["extra"]
        assert error["extra"]["metadata.yaml"] == {"type": ["Must be equal to Slice."]}
        assert error["extra"]["issue_codes"][0]["code"] == 1010

    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={rison.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"] == 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={rison.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"] == 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={rison.dumps(arguments)}")
        assert rv.status_code == 200
        data = json.loads(rv.data.decode("utf-8"))

        # Verify the fixture charts are in the results
        result_names = {r["slice_name"] for r in data["result"]}
        assert "name0" in result_names

    @parameterized.expand(
        [
            "Pivot Table v2",  # Non-legacy charts
        ],
    )
    @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})
        assert rv.status_code == 200
        data = json.loads(rv.data.decode("utf-8"))

        assert 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},
        )
        assert rv.status_code == 200
        data = json.loads(rv.data.decode("utf-8"))
        assert 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"]}]
                ),
            },
        )
        assert rv.status_code == 200
        data = json.loads(rv.data.decode("utf-8"))
        assert data["result"] == [
            {"chart_id": slc.id, "viz_error": None, "viz_status": "success"}
        ]

    @pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
    def test_warm_up_cache_native_defaults_hit_browser_query_cache(self) -> None:
        self.login(ADMIN_USERNAME)
        chart = self.get_slice("Pivot Table v2")
        dashboard = self.get_dash_by_slug("births")

        saved_query_context = json.loads(chart.query_context)
        chart_filter = {"col": "name", "op": "!=", "val": "__missing_name__"}
        for query in saved_query_context["queries"]:
            query["filters"] = [*(query.get("filters") or []), chart_filter]
        chart.query_context = json.dumps(saved_query_context)

        metadata = json.loads(dashboard.json_metadata or "{}")
        legacy_filter = {"col": "name", "op": "in", "val": ["Alice"]}
        metadata["default_filters"] = json.dumps(
            {"-1": {legacy_filter["col"]: legacy_filter["val"]}}
        )
        metadata["filter_scopes"] = {}
        native_filter = {"col": "gender", "op": "IN", "val": ["girl"]}
        metadata["native_filter_configuration"] = [
            {
                "id": "NATIVE_FILTER-gender",
                "name": "Gender",
                "type": "NATIVE_FILTER",
                "filterType": "filter_select",
                "scope": {"rootPath": ["ROOT_ID"], "excluded": []},
                "targets": [
                    {
                        "datasetId": chart.datasource_id,
                        "column": {"name": "gender"},
                    }
                ],
                "defaultDataMask": {
                    "extraFormData": {"filters": [native_filter]},
                    "filterState": {"value": ["girl"]},
                },
                "controlValues": {},
            }
        ]
        dashboard.json_metadata = json.dumps(metadata)
        db.session.commit()

        warm_up_response = self.client.put(
            "/api/v1/chart/warm_up_cache",
            json={"chart_id": chart.id, "dashboard_id": dashboard.id},
        )
        assert warm_up_response.status_code == 200
        assert warm_up_response.json["result"] == [
            {"chart_id": chart.id, "viz_error": None, "viz_status": "success"}
        ]

        browser_query_context = json.loads(chart.query_context)
        browser_query_context["force"] = False
        for query in browser_query_context["queries"]:
            query["filters"] = [
                legacy_filter,
                native_filter,
                *(query.get("filters") or []),
            ]

        assert browser_query_context["queries"][0]["filters"] == [
            legacy_filter,
            native_filter,
            chart_filter,
        ]

        chart_data_response = self.client.post(
            "/api/v1/chart/data",
            json=browser_query_context,
        )
        assert chart_data_response.status_code == 200
        assert chart_data_response.json["result"][0]["is_cached"] is True

    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})
        assert rv.status_code == 400
        data = json.loads(rv.data.decode("utf-8"))
        assert 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})
        assert rv.status_code == 404
        data = json.loads(rv.data.decode("utf-8"))
        assert 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},
        )
        assert rv.status_code == 400
        data = json.loads(rv.data.decode("utf-8"))
        assert 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. Open the "
                        "chart in Explore once (or re-save it) to generate it.",
                        "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 query context does not exist. Open the "
                        "chart in Explore once (or re-save it) to generate it.",
                        "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.add(new_tag.id)
        update_payload = {"tags": list(new_tags)}

        uri = f"api/v1/chart/{chart.id}"
        rv = self.put_assert_metric(uri, update_payload, "put")
        assert 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}
        assert 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")
        assert 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]
        assert tag_list == new_tags

    @pytest.mark.usefixtures("create_chart_with_tag")
    def test_update_chart_add_tags_can_tag_on_chart(self):
        """
        Validates an editor 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.add(new_tag.id)
        update_payload = {"tags": list(new_tags)}

        uri = f"api/v1/chart/{chart.id}"
        rv = self.put_assert_metric(uri, update_payload, "put")
        assert 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}
        assert 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 editor 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")
        assert 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]
        assert 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 editor 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")
        assert rv.status_code == 403
        assert (
            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 editor 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")
        assert rv.status_code == 403
        assert (
            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 editor 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")
        assert rv.status_code == 200

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

    def test_related_editors_allowed_for_write_user(self):
        """
        Chart API: GET /api/v1/chart/related/editors returns 200 for Admin.
        """
        self.login(ADMIN_USERNAME)
        rv = self.client.get("api/v1/chart/related/editors")
        assert rv.status_code == 200