base_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.
# isort:skip_file
import json

from flask_appbuilder.models.sqla.interface import SQLAInterface
import prison

import tests.test_app
from superset import db, security_manager
from superset.extensions import appbuilder
from superset.models.dashboard import Dashboard
from superset.views.base_api import BaseSupersetModelRestApi

from .base_tests import SupersetTestCase


class Model1Api(BaseSupersetModelRestApi):
    datamodel = SQLAInterface(Dashboard)
    allow_browser_login = True
    class_permission_name = "Dashboard"
    method_permission_name = {
        "get_list": "read",
        "get": "read",
        "export": "read",
        "post": "write",
        "put": "write",
        "delete": "write",
        "bulk_delete": "write",
        "info": "read",
        "related": "read",
    }


appbuilder.add_api(Model1Api)


class TestOpenApiSpec(SupersetTestCase):
    def test_open_api_spec(self):
        """
        API: Test validate OpenAPI spec
        :return:
        """
        from openapi_spec_validator import validate_spec

        self.login(username="admin")
        uri = "api/v1/_openapi"
        rv = self.client.get(uri)
        self.assertEqual(rv.status_code, 200)
        response = json.loads(rv.data.decode("utf-8"))
        validate_spec(response)


class TestBaseModelRestApi(SupersetTestCase):
    def test_default_missing_declaration_get(self):
        """
        API: Test default missing declaration on get

        We want to make sure that not declared list_columns will
        not render all columns by default but just the model's pk
        """
        # Check get list response
        self.login(username="admin")
        uri = "api/v1/model1api/"
        rv = self.client.get(uri)
        self.assertEqual(rv.status_code, 200)
        response = json.loads(rv.data.decode("utf-8"))
        self.assertEqual(response["list_columns"], ["id"])
        for result in response["result"]:
            self.assertEqual(list(result.keys()), ["id"])

        # Check get response
        dashboard = db.session.query(Dashboard).first()
        uri = f"api/v1/model1api/{dashboard.id}"
        rv = self.client.get(uri)
        self.assertEqual(rv.status_code, 200)
        response = json.loads(rv.data.decode("utf-8"))
        self.assertEqual(response["show_columns"], ["id"])
        self.assertEqual(list(response["result"].keys()), ["id"])

    def test_default_missing_declaration_put_spec(self):
        """
        API: Test default missing declaration on put openapi spec

        We want to make sure that not declared edit_columns will
        not render all columns by default but just the model's pk
        """
        self.login(username="admin")
        uri = "api/v1/_openapi"
        rv = self.client.get(uri)
        # dashboard model accepts all fields are null
        self.assertEqual(rv.status_code, 200)
        response = json.loads(rv.data.decode("utf-8"))
        expected_mutation_spec = {
            "properties": {"id": {"format": "int32", "type": "integer"}},
            "type": "object",
        }
        self.assertEqual(
            response["components"]["schemas"]["Model1Api.post"], expected_mutation_spec
        )
        self.assertEqual(
            response["components"]["schemas"]["Model1Api.put"], expected_mutation_spec
        )

    def test_default_missing_declaration_post(self):
        """
        API: Test default missing declaration on post

        We want to make sure that not declared add_columns will
        not accept all columns by default
        """
        dashboard_data = {
            "dashboard_title": "title1",
            "slug": "slug1",
            "position_json": '{"a": "A"}',
            "css": "css",
            "json_metadata": '{"b": "B"}',
            "published": True,
        }
        self.login(username="admin")
        uri = "api/v1/model1api/"
        rv = self.client.post(uri, json=dashboard_data)
        response = json.loads(rv.data.decode("utf-8"))
        self.assertEqual(rv.status_code, 422)
        expected_response = {
            "message": {
                "css": ["Unknown field."],
                "dashboard_title": ["Unknown field."],
                "json_metadata": ["Unknown field."],
                "position_json": ["Unknown field."],
                "published": ["Unknown field."],
                "slug": ["Unknown field."],
            }
        }
        self.assertEqual(response, expected_response)

    def test_default_missing_declaration_put(self):
        """
        API: Test default missing declaration on put

        We want to make sure that not declared edit_columns will
        not accept all columns by default
        """
        dashboard = db.session.query(Dashboard).first()
        dashboard_data = {"dashboard_title": "CHANGED", "slug": "CHANGED"}
        self.login(username="admin")
        uri = f"api/v1/model1api/{dashboard.id}"
        rv = self.client.put(uri, json=dashboard_data)
        response = json.loads(rv.data.decode("utf-8"))
        self.assertEqual(rv.status_code, 422)
        expected_response = {
            "message": {
                "dashboard_title": ["Unknown field."],
                "slug": ["Unknown field."],
            }
        }
        self.assertEqual(response, expected_response)


class ApiOwnersTestCaseMixin:
    """
    Implements shared tests for owners related field
    """

    resource_name: str = ""

    def test_get_related_owners(self):
        """
            API: Test get related owners
        """
        self.login(username="admin")
        uri = f"api/v1/{self.resource_name}/related/owners"
        rv = self.client.get(uri)
        self.assertEqual(rv.status_code, 200)
        response = json.loads(rv.data.decode("utf-8"))
        users = db.session.query(security_manager.user_model).all()
        expected_users = [str(user) for user in users]
        self.assertEqual(response["count"], len(users))
        # This needs to be implemented like this, because ordering varies between
        # postgres and mysql
        response_users = [result["text"] for result in response["result"]]
        for expected_user in expected_users:
            self.assertIn(expected_user, response_users)

    def test_get_filter_related_owners(self):
        """
            API: Test get filter related owners
        """
        self.login(username="admin")
        argument = {"filter": "gamma"}
        uri = f"api/v1/{self.resource_name}/related/owners?q={prison.dumps(argument)}"

        rv = self.client.get(uri)
        self.assertEqual(rv.status_code, 200)
        response = json.loads(rv.data.decode("utf-8"))
        self.assertEqual(3, response["count"])
        sorted_results = sorted(response["result"], key=lambda value: value["text"])
        expected_results = [
            {"text": "gamma user", "value": 2},
            {"text": "gamma2 user", "value": 3},
            {"text": "gamma_sqllab user", "value": 4},
        ]
        self.assertEqual(expected_results, sorted_results)

    def test_get_related_fail(self):
        """
            API: Test get related fail
        """
        self.login(username="admin")
        uri = f"api/v1/{self.resource_name}/related/owner"

        rv = self.client.get(uri)
        self.assertEqual(rv.status_code, 404)