tests/integration_tests/databases/schema_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.

from unittest import mock

from marshmallow import fields, Schema, ValidationError

from superset.databases.schemas import DatabaseParametersSchemaMixin
from superset.db_engine_specs.base import BasicParametersMixin
from superset.models.core import ConfigurationMethod


class DummySchema(Schema, DatabaseParametersSchemaMixin):
    sqlalchemy_uri = fields.String()


class DummyEngine(BasicParametersMixin):
    engine = "dummy"
    default_driver = "dummy"


class InvalidEngine:
    pass


@mock.patch("superset.databases.schemas.get_engine_specs")
def test_database_parameters_schema_mixin(get_engine_specs):
    get_engine_specs.return_value = {"dummy_engine": DummyEngine}
    payload = {
        "engine": "dummy_engine",
        "configuration_method": ConfigurationMethod.DYNAMIC_FORM,
        "parameters": {
            "username": "username",
            "password": "password",
            "host": "localhost",
            "port": 12345,
            "database": "dbname",
        },
    }
    schema = DummySchema()
    result = schema.load(payload)
    assert result == {
        "configuration_method": ConfigurationMethod.DYNAMIC_FORM,
        "sqlalchemy_uri": "dummy+dummy://username:password@localhost:12345/dbname",
    }


def test_database_parameters_schema_mixin_no_engine():
    payload = {
        "configuration_method": ConfigurationMethod.DYNAMIC_FORM,
        "parameters": {
            "username": "username",
            "password": "password",
            "host": "localhost",
            "port": 12345,
            "dbname": "dbname",
        },
    }
    schema = DummySchema()
    try:
        schema.load(payload)
    except ValidationError as err:
        assert err.messages == {
            "_schema": [
                "An engine must be specified when passing individual parameters to a database."
            ]
        }


@mock.patch("superset.databases.schemas.get_engine_specs")
def test_database_parameters_schema_mixin_invalid_engine(get_engine_specs):
    get_engine_specs.return_value = {}
    payload = {
        "engine": "dummy_engine",
        "configuration_method": ConfigurationMethod.DYNAMIC_FORM,
        "parameters": {
            "username": "username",
            "password": "password",
            "host": "localhost",
            "port": 12345,
            "dbname": "dbname",
        },
    }
    schema = DummySchema()
    try:
        schema.load(payload)
    except ValidationError as err:
        assert err.messages == {
            "_schema": ['Engine "dummy_engine" is not a valid engine.']
        }


@mock.patch("superset.databases.schemas.get_engine_specs")
def test_database_parameters_schema_no_mixin(get_engine_specs):
    get_engine_specs.return_value = {"invalid_engine": InvalidEngine}
    payload = {
        "engine": "invalid_engine",
        "configuration_method": ConfigurationMethod.DYNAMIC_FORM,
        "parameters": {
            "username": "username",
            "password": "password",
            "host": "localhost",
            "port": 12345,
            "database": "dbname",
        },
    }
    schema = DummySchema()
    try:
        schema.load(payload)
    except ValidationError as err:
        assert err.messages == {
            "_schema": [
                (
                    'Engine spec "InvalidEngine" does not support '
                    "being configured via individual parameters."
                )
            ]
        }


@mock.patch("superset.databases.schemas.get_engine_specs")
def test_database_parameters_schema_mixin_invalid_type(get_engine_specs):
    get_engine_specs.return_value = {"dummy_engine": DummyEngine}
    payload = {
        "engine": "dummy_engine",
        "configuration_method": ConfigurationMethod.DYNAMIC_FORM,
        "parameters": {
            "username": "username",
            "password": "password",
            "host": "localhost",
            "port": "badport",
            "database": "dbname",
        },
    }
    schema = DummySchema()
    try:
        schema.load(payload)
    except ValidationError as err:
        assert err.messages == {"port": ["Not a valid integer."]}