# 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 unittest from sqlalchemy.dialects import mysql from sqlalchemy.dialects.mysql import DATE, NVARCHAR, TEXT, VARCHAR from superset.db_engine_specs.mysql import MySQLEngineSpec from superset.errors import ErrorLevel, SupersetError, SupersetErrorType from superset.models.sql_lab import Query from superset.utils.core import GenericDataType from tests.integration_tests.db_engine_specs.base_tests import ( assert_generic_types, TestDbEngineSpec, ) class TestMySQLEngineSpecsDbEngineSpec(TestDbEngineSpec): @unittest.skipUnless( TestDbEngineSpec.is_module_installed("MySQLdb"), "mysqlclient not installed" ) def test_get_datatype_mysql(self): """Tests related to datatype mapping for MySQL""" self.assertEqual("TINY", MySQLEngineSpec.get_datatype(1)) self.assertEqual("VARCHAR", MySQLEngineSpec.get_datatype(15)) def test_convert_dttm(self): dttm = self.get_dttm() self.assertEqual( MySQLEngineSpec.convert_dttm("DATE", dttm), "STR_TO_DATE('2019-01-02', '%Y-%m-%d')", ) self.assertEqual( MySQLEngineSpec.convert_dttm("DATETIME", dttm), "STR_TO_DATE('2019-01-02 03:04:05.678900', '%Y-%m-%d %H:%i:%s.%f')", ) def test_column_datatype_to_string(self): test_cases = ( (DATE(), "DATE"), (VARCHAR(length=255), "VARCHAR(255)"), ( VARCHAR(length=255, charset="latin1", collation="utf8mb4_general_ci"), "VARCHAR(255)", ), (NVARCHAR(length=128), "NATIONAL VARCHAR(128)"), (TEXT(), "TEXT"), ) for original, expected in test_cases: actual = MySQLEngineSpec.column_datatype_to_string( original, mysql.dialect() ) self.assertEqual(actual, expected) def test_generic_type(self): type_expectations = ( # Numeric ("TINYINT", GenericDataType.NUMERIC), ("SMALLINT", GenericDataType.NUMERIC), ("MEDIUMINT", GenericDataType.NUMERIC), ("INT", GenericDataType.NUMERIC), ("BIGINT", GenericDataType.NUMERIC), ("DECIMAL", GenericDataType.NUMERIC), ("FLOAT", GenericDataType.NUMERIC), ("DOUBLE", GenericDataType.NUMERIC), ("BIT", GenericDataType.NUMERIC), # String ("CHAR", GenericDataType.STRING), ("VARCHAR", GenericDataType.STRING), ("TINYTEXT", GenericDataType.STRING), ("MEDIUMTEXT", GenericDataType.STRING), ("LONGTEXT", GenericDataType.STRING), # Temporal ("DATE", GenericDataType.TEMPORAL), ("DATETIME", GenericDataType.TEMPORAL), ("TIMESTAMP", GenericDataType.TEMPORAL), ("TIME", GenericDataType.TEMPORAL), ) assert_generic_types(MySQLEngineSpec, type_expectations) def test_extract_error_message(self): from MySQLdb._exceptions import OperationalError message = "Unknown table 'BIRTH_NAMES1' in information_schema" exception = OperationalError(message) extracted_message = MySQLEngineSpec._extract_error_message(exception) assert extracted_message == message exception = OperationalError(123, message) extracted_message = MySQLEngineSpec._extract_error_message(exception) assert extracted_message == message def test_extract_errors(self): """ Test that custom error messages are extracted correctly. """ msg = "mysql: Access denied for user 'test'@'testuser.com'" result = MySQLEngineSpec.extract_errors(Exception(msg)) assert result == [ SupersetError( error_type=SupersetErrorType.CONNECTION_ACCESS_DENIED_ERROR, message='Either the username "test" or the password is incorrect.', level=ErrorLevel.ERROR, extra={ "invalid": ["username", "password"], "engine_name": "MySQL", "issue_codes": [ { "code": 1014, "message": "Issue 1014 - Either the" " username or the password is wrong.", }, { "code": 1015, "message": "Issue 1015 - Either the database is " "spelled incorrectly or does not exist.", }, ], }, ) ] msg = "mysql: Unknown MySQL server host 'badhostname.com'" result = MySQLEngineSpec.extract_errors(Exception(msg)) assert result == [ SupersetError( error_type=SupersetErrorType.CONNECTION_INVALID_HOSTNAME_ERROR, message='Unknown MySQL server host "badhostname.com".', level=ErrorLevel.ERROR, extra={ "invalid": ["host"], "engine_name": "MySQL", "issue_codes": [ { "code": 1007, "message": "Issue 1007 - The hostname" " provided can't be resolved.", } ], }, ) ] msg = "mysql: Can't connect to MySQL server on 'badconnection.com'" result = MySQLEngineSpec.extract_errors(Exception(msg)) assert result == [ SupersetError( error_type=SupersetErrorType.CONNECTION_HOST_DOWN_ERROR, message='The host "badconnection.com" might be ' "down and can't be reached.", level=ErrorLevel.ERROR, extra={ "invalid": ["host", "port"], "engine_name": "MySQL", "issue_codes": [ { "code": 1007, "message": "Issue 1007 - The hostname provided" " can't be resolved.", } ], }, ) ] msg = "mysql: Can't connect to MySQL server on '93.184.216.34'" result = MySQLEngineSpec.extract_errors(Exception(msg)) assert result == [ SupersetError( error_type=SupersetErrorType.CONNECTION_HOST_DOWN_ERROR, message='The host "93.184.216.34" might be down and can\'t be reached.', level=ErrorLevel.ERROR, extra={ "invalid": ["host", "port"], "engine_name": "MySQL", "issue_codes": [ { "code": 10007, "message": "Issue 1007 - The hostname provided " "can't be resolved.", } ], }, ) ] msg = "mysql: Unknown database 'badDB'" result = MySQLEngineSpec.extract_errors(Exception(msg)) assert result == [ SupersetError( message='Unable to connect to database "badDB".', error_type=SupersetErrorType.CONNECTION_UNKNOWN_DATABASE_ERROR, level=ErrorLevel.ERROR, extra={ "invalid": ["database"], "engine_name": "MySQL", "issue_codes": [ { "code": 1015, "message": "Issue 1015 - Either the database is spelled incorrectly or does not exist.", } ], }, ) ] msg = "check the manual that corresponds to your MySQL server version for the right syntax to use near 'fromm" result = MySQLEngineSpec.extract_errors(Exception(msg)) assert result == [ SupersetError( message='Please check your query for syntax errors near "fromm". Then, try running your query again.', error_type=SupersetErrorType.SYNTAX_ERROR, level=ErrorLevel.ERROR, extra={ "engine_name": "MySQL", "issue_codes": [ { "code": 1030, "message": "Issue 1030 - The query has a syntax error.", } ], }, ) ] @unittest.mock.patch("sqlalchemy.engine.Engine.connect") def test_get_cancel_query_id(self, engine_mock): query = Query() cursor_mock = engine_mock.return_value.__enter__.return_value cursor_mock.fetchone.return_value = [123] assert MySQLEngineSpec.get_cancel_query_id(cursor_mock, query) == 123 @unittest.mock.patch("sqlalchemy.engine.Engine.connect") def test_cancel_query(self, engine_mock): query = Query() cursor_mock = engine_mock.return_value.__enter__.return_value assert MySQLEngineSpec.cancel_query(cursor_mock, query, 123) is True @unittest.mock.patch("sqlalchemy.engine.Engine.connect") def test_cancel_query_failed(self, engine_mock): query = Query() cursor_mock = engine_mock.raiseError.side_effect = Exception() assert MySQLEngineSpec.cancel_query(cursor_mock, query, 123) is False