salt/tests/integration/states/test_mysql_grants.py

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

from __future__ import absolute_import, print_function, unicode_literals

import logging

import pytest
import salt.utils.path
from salt.ext import six
from salt.modules import mysql as mysqlmod
from tests.support.case import ModuleCase
from tests.support.mixins import SaltReturnAssertsMixin
from tests.support.unit import skipIf

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.path.which("mysqladmin"):
    NO_MYSQL = True


@skipIf(
    NO_MYSQL,
    "Please install MySQL bindings and a MySQL Server before running"
    "MySQL integration tests.",
)
class MysqlGrantsStateTest(ModuleCase, 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"},
    }

    @pytest.mark.destructive_test
    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 '%s'", 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 '%s'", self.table2)
        self.run_function(
            "mysql.query",
            database=self.testdb2,
            query=create_query,
            connection_user=self.user,
            connection_pass=self.password,
        )

    @pytest.mark.destructive_test
    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"},
        )

    @pytest.mark.destructive_test
    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)