salt/tests/integration/states/mysql.py

# -*- coding: utf-8 -*-
'''
Tests for the MySQL states
'''

# Import python libs
from __future__ import absolute_import
import logging

# Import Salt Testing libs
from salttesting import skipIf
from salttesting.helpers import (
    destructiveTest,
    ensure_in_syspath
)
ensure_in_syspath('../../')

# Import salt libs
import integration
import salt.utils
import salt.ext.six as six
from salt.modules import mysql as mysqlmod

log = logging.getLogger(__name__)

NO_MYSQL = False
try:
    import MySQLdb  # pylint: disable=import-error,unused-import
except ImportError:
    NO_MYSQL = True

if not salt.utils.which('mysqladmin'):
    NO_MYSQL = True


@skipIf(
    NO_MYSQL,
    'Please install MySQL bindings and a MySQL Server before running'
    'MySQL integration tests.'
)
class MysqlDatabaseStateTest(integration.ModuleCase,
                             integration.SaltReturnAssertsMixIn):
    '''
    Validate the mysql_database state
    '''

    user = 'root'
    password = 'poney'

    @destructiveTest
    def setUp(self):
        '''
        Test presence of MySQL server, enforce a root password
        '''
        super(MysqlDatabaseStateTest, self).setUp()
        NO_MYSQL_SERVER = True
        # now ensure we know the mysql root password
        # one of theses two at least should work
        ret1 = self.run_state(
            'cmd.run',
             name='mysqladmin --host="localhost" -u '
               + self.user
               + ' flush-privileges password "'
               + self.password
               + '"'
        )
        ret2 = self.run_state(
            'cmd.run',
             name='mysqladmin --host="localhost" -u '
               + self.user
               + ' --password="'
               + self.password
               + '" flush-privileges password "'
               + self.password
               + '"'
        )
        key, value = ret2.popitem()
        if value['result']:
            NO_MYSQL_SERVER = False
        else:
            self.skipTest('No MySQL Server running, or no root access on it.')

    def _test_database(self, db_name, second_db_name, test_conn, **kwargs):
        '''
        Create db two times, test conn, remove it two times
        '''
        # In case of...
        ret = self.run_state('mysql_database.absent',
                             name=db_name,
                             **kwargs
        )
        ret = self.run_state('mysql_database.present',
                             name=db_name,
                             **kwargs
        )
        self.assertSaltTrueReturn(ret)
        self.assertInSaltComment(
            'The database ' + db_name + ' has been created',
            ret
        )
        #2nd run
        ret = self.run_state('mysql_database.present',
                             name=second_db_name,
                             **kwargs
        )
        self.assertSaltTrueReturn(ret)
        self.assertInSaltComment(
            'Database ' + db_name + ' is already present',
            ret
        )
        if test_conn:
            # test root connection
            ret = self.run_function(
                'mysql.query',
                database=db_name,
                query='SELECT 1',
                **kwargs
            )
            if not isinstance(ret, dict) or 'results' not in ret:
                raise AssertionError(
                    ('Unexpected result while testing connection'
                    ' on db \'{0}\': {1}').format(
                        db_name,
                        repr(ret)
                    )
                )
            self.assertEqual([['1']], ret['results'])

        # Now removing databases
        kwargs.pop('character_set')
        kwargs.pop('collate')
        ret = self.run_state('mysql_database.absent',
                             name=db_name,
                             **kwargs
        )
        self.assertSaltTrueReturn(ret)
        self.assertInSaltComment(
            'Database ' + db_name + ' has been removed',
            ret
        )
        #2nd run
        ret = self.run_state('mysql_database.absent',
                             name=second_db_name,
                            ** kwargs
        )
        self.assertSaltTrueReturn(ret)
        self.assertInSaltComment(
            'Database ' + db_name + ' is not present, so it cannot be removed',
            ret
        )
        self.assertSaltStateChangesEqual(ret, {})

    @destructiveTest
    def test_present_absent(self):
        '''
        mysql_database.present
        '''
        self._test_database(
            'testdb1',
            'testdb1',
            test_conn=True,
            character_set='utf8',
            collate='utf8_general_ci',
            connection_user=self.user,
            connection_pass=self.password,
            connection_charset='utf8'
        )

    # TODO: test with variations on collate and charset, check for db alter
    # once it will be done in mysql_database.present state

    @destructiveTest
    def test_present_absent_fuzzy(self):
        '''
        mysql_database.present with utf-8 andf fuzzy db name
        '''
        # this is : ":() ;,?@=`&'\
        dbname_fuzzy = '":() ;,?@=`&/\'\\'
        # \xe6\xa8\x99\ = \u6a19 = 標
        # this is : "();,?:@=`&/標'\
        dbname_utf8 = '"();,?@=`&//\xe6\xa8\x99\'\\'
        dbname_unicode = u'"();,?@=`&//\u6a19\'\\'

        self._test_database(
            dbname_fuzzy,
            dbname_fuzzy,
            test_conn=True,
            character_set='utf8',
            collate='utf8_general_ci',
            connection_user=self.user,
            connection_pass=self.password,
            connection_charset='utf8'
        )

        # FIXME: MySQLdb bugs on dbnames with utf-8?
        self._test_database(
            dbname_utf8,
            dbname_unicode,
            test_conn=False,
            character_set='utf8',
            collate='utf8_general_ci',
            connection_user=self.user,
            connection_pass=self.password,
            connection_charset='utf8',
            #saltenv={"LC_ALL": "en_US.utf8"}
        )

    @destructiveTest
    @skipIf(True, 'This tests needs issue #8947 to be fixed first')
    def test_utf8_from_sls_file(self):
        '''
        Try to create/destroy an utf-8 database name from an sls file #8947
        '''
        expected_result = {
            'mysql_database_|-A_|-foo \xe6\xba\x96`bar_|-present': {
                '__run_num__': 0,
                'comment': 'The database foo \xe6\xba\x96`bar has been created',
                'result': True},
            'mysql_database_|-B_|-foo \xe6\xba\x96`bar_|-absent': {
                '__run_num__': 1,
                'comment': 'Database foo \xe6\xba\x96`bar has been removed',
                'result': True},
        }
        result = {}
        ret = self.run_function('state.sls', mods='mysql_utf8')
        if not isinstance(ret, dict):
            raise AssertionError(
                ('Unexpected result while testing external mysql utf8 sls'
                ': {0}').format(
                    repr(ret)
                )
            )
        for item, descr in six.iteritems(ret):
            result[item] = {
                '__run_num__': descr['__run_num__'],
                'comment': descr['comment'],
                'result': descr['result']
            }
        self.assertEqual(expected_result, result)


@skipIf(
    NO_MYSQL,
    'Please install MySQL bindings and a MySQL Server before running'
    'MySQL integration tests.'
)
class MysqlGrantsStateTest(integration.ModuleCase,
                           integration.SaltReturnAssertsMixIn):
    '''
    Validate the mysql_grants states
    '''

    user = 'root'
    password = 'poney'
    # yep, theses are valid MySQL db names
    # very special chars are _ % and .
    testdb1 = 'tes.t\'"saltdb'
    testdb2 = 't_st `(:=salt%b)'
    testdb3 = 'test `(:=salteeb)'
    table1 = 'foo'
    table2 = "foo `\'%_bar"
    users = {
        'user1': {
            'name': 'foo',
            'pwd': 'bar',
        },
        'user2': {
            'name': 'user ";--,?:&/\\',
            'pwd': '";--(),?:@=&/\\',
        },
        # this is : passwd 標標
        'user3': {
            'name': 'user( @ )=foobar',
            'pwd': '\xe6\xa8\x99\xe6\xa8\x99',
        },
        # this is : user/password containing 標標
        'user4': {
            'name': 'user \xe6\xa8\x99',
            'pwd': '\xe6\xa8\x99\xe6\xa8\x99',
        },
    }

    @destructiveTest
    def setUp(self):
        '''
        Test presence of MySQL server, enforce a root password
        '''
        super(MysqlGrantsStateTest, self).setUp()
        NO_MYSQL_SERVER = True
        # now ensure we know the mysql root password
        # one of theses two at least should work
        ret1 = self.run_state(
            'cmd.run',
             name='mysqladmin --host="localhost" -u '
               + self.user
               + ' flush-privileges password "'
               + self.password
               + '"'
        )
        ret2 = self.run_state(
            'cmd.run',
             name='mysqladmin --host="localhost" -u '
               + self.user
               + ' --password="'
               + self.password
               + '" flush-privileges password "'
               + self.password
               + '"'
        )
        key, value = ret2.popitem()
        if value['result']:
            NO_MYSQL_SERVER = False
        else:
            self.skipTest('No MySQL Server running, or no root access on it.')
        # Create some users and a test db
        for user, userdef in six.iteritems(self.users):
            self._userCreation(uname=userdef['name'], password=userdef['pwd'])
        self.run_state(
            'mysql_database.present',
            name=self.testdb1,
            character_set='utf8',
            collate='utf8_general_ci',
            connection_user=self.user,
            connection_pass=self.password,
        )
        self.run_state(
            'mysql_database.present',
            name=self.testdb2,
            character_set='utf8',
            collate='utf8_general_ci',
            connection_user=self.user,
            connection_pass=self.password,
        )
        create_query = ('CREATE TABLE {tblname} ('
            ' id INT NOT NULL AUTO_INCREMENT PRIMARY KEY,'
            ' data VARCHAR(100)) ENGINE={engine};'.format(
            tblname=mysqlmod.quote_identifier(self.table1),
            engine='MYISAM',
        ))
        log.info('Adding table \'{0}\''.format(self.table1,))
        self.run_function(
            'mysql.query',
            database=self.testdb2,
            query=create_query,
            connection_user=self.user,
            connection_pass=self.password
        )
        create_query = ('CREATE TABLE {tblname} ('
            ' id INT NOT NULL AUTO_INCREMENT PRIMARY KEY,'
            ' data VARCHAR(100)) ENGINE={engine};'.format(
            tblname=mysqlmod.quote_identifier(self.table2),
            engine='MYISAM',
        ))
        log.info('Adding table \'{0}\''.format(self.table2,))
        self.run_function(
            'mysql.query',
            database=self.testdb2,
            query=create_query,
            connection_user=self.user,
            connection_pass=self.password
        )

    @destructiveTest
    def tearDown(self):
        '''
        Removes created users and db
        '''
        for user, userdef in six.iteritems(self.users):
            self._userRemoval(uname=userdef['name'], password=userdef['pwd'])
        self.run_state(
            'mysql_database.absent',
            name=self.testdb1,
            connection_user=self.user,
            connection_pass=self.password,
       )
        self.run_function(
            'mysql_database.absent',
            name=self.testdb2,
            connection_user=self.user,
            connection_pass=self.password,
       )

    def _userCreation(self,
                      uname,
                      password=None):
        '''
        Create a test user
        '''
        self.run_state(
            'mysql_user.present',
            name=uname,
            host='localhost',
            password=password,
            connection_user=self.user,
            connection_pass=self.password,
            connection_charset='utf8',
            saltenv={"LC_ALL": "en_US.utf8"}
        )

    def _userRemoval(self,
                     uname,
                     password=None):
        '''
        Removes a test user
        '''
        self.run_state(
            'mysql_user.absent',
            name=uname,
            host='localhost',
            connection_user=self.user,
            connection_pass=self.password,
            connection_charset='utf8',
            saltenv={"LC_ALL": "en_US.utf8"}
        )

    @destructiveTest
    def test_grant_present_absent(self):
        '''
        mysql_database.present
        '''
        ret = self.run_state(
            'mysql_grants.present',
            name='grant test 1',
            grant='SELECT, INSERT',
            database=self.testdb1 + '.*',
            user=self.users['user1']['name'],
            host='localhost',
            grant_option=True,
            revoke_first=True,
            connection_user=self.user,
            connection_pass=self.password,
            connection_charset='utf8'
        )
        self.assertSaltTrueReturn(ret)
        ret = self.run_state(
            'mysql_grants.present',
            name='grant test 2',
            grant='SELECT, ALTER,CREATE TEMPORARY tables, execute',
            database=self.testdb1 + '.*',
            user=self.users['user1']['name'],
            host='localhost',
            grant_option=True,
            revoke_first=True,
            connection_user=self.user,
            connection_pass=self.password,
            connection_charset='utf8'
        )
        self.assertSaltTrueReturn(ret)
        ret = self.run_state(
            'mysql_grants.present',
            name='grant test 3',
            grant='SELECT, INSERT',
            database=self.testdb2 + '.' + self.table2,
            user=self.users['user2']['name'],
            host='localhost',
            grant_option=True,
            revoke_first=True,
            connection_user=self.user,
            connection_pass=self.password,
            connection_charset='utf8'
        )
        self.assertSaltTrueReturn(ret)
        ret = self.run_state(
            'mysql_grants.present',
            name='grant test 4',
            grant='SELECT, INSERT',
            database=self.testdb2 + '.' + self.table2,
            user=self.users['user2']['name'],
            host='localhost',
            grant_option=True,
            revoke_first=True,
            connection_user=self.user,
            connection_pass=self.password,
            connection_charset='utf8'
        )
        self.assertSaltTrueReturn(ret)
        ret = self.run_state(
            'mysql_grants.present',
            name='grant test 5',
            grant='SELECT, UPDATE',
            database=self.testdb2 + '.*',
            user=self.users['user1']['name'],
            host='localhost',
            grant_option=True,
            revoke_first=False,
            connection_user=self.user,
            connection_pass=self.password,
            connection_charset='utf8'
        )
        self.assertSaltTrueReturn(ret)
        ret = self.run_state(
            'mysql_grants.absent',
            name='grant test 6',
            grant='SELECT,update',
            database=self.testdb2 + '.*',
            user=self.users['user1']['name'],
            host='localhost',
            grant_option=True,
            revoke_first=False,
            connection_user=self.user,
            connection_pass=self.password,
            connection_charset='utf8'
        )
        self.assertSaltTrueReturn(ret)


if __name__ == '__main__':
    from integration import run_tests
    run_tests(MysqlDatabaseStateTest, MysqlGrantsStateTest)