"Fossies" - the Fresh Open Source Software Archive

Member "salt-3002.2/tests/unit/modules/test_mysql.py" (18 Nov 2020, 30564 Bytes) of package /linux/misc/salt-3002.2.tar.gz:


As a special service "Fossies" has tried to format the requested source page into HTML format using (guessed) Python source code syntax highlighting (style: standard) with prefixed line numbers. Alternatively you can here view or download the uninterpreted source code file. See also the latest Fossies "Diffs" side-by-side code changes report for "test_mysql.py": 3002.1_vs_3002.2.

    1 """
    2     :codeauthor: Mike Place (mp@saltstack.com)
    3 
    4 
    5     tests.unit.modules.mysql
    6     ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
    7 """
    8 
    9 
   10 import logging
   11 
   12 import salt.modules.mysql as mysql
   13 from tests.support.mixins import LoaderModuleMockMixin
   14 from tests.support.mock import MagicMock, call, patch
   15 from tests.support.unit import TestCase, skipIf
   16 
   17 log = logging.getLogger(__name__)
   18 NO_MYSQL = False
   19 NO_PyMYSQL = False
   20 try:
   21     import MySQLdb  # pylint: disable=W0611
   22 except ImportError:
   23     NO_MYSQL = True
   24 
   25 try:
   26     # MySQLdb import failed, try to import PyMySQL
   27     import pymysql
   28 except ImportError:
   29     NO_PyMYSQL = True
   30 
   31 __all_privileges__ = [
   32     "ALTER",
   33     "ALTER ROUTINE",
   34     "BACKUP_ADMIN",
   35     "BINLOG_ADMIN",
   36     "CONNECTION_ADMIN",
   37     "CREATE",
   38     "CREATE ROLE",
   39     "CREATE ROUTINE",
   40     "CREATE TABLESPACE",
   41     "CREATE TEMPORARY TABLES",
   42     "CREATE USER",
   43     "CREATE VIEW",
   44     "DELETE",
   45     "DROP",
   46     "DROP ROLE",
   47     "ENCRYPTION_KEY_ADMIN",
   48     "EVENT",
   49     "EXECUTE",
   50     "FILE",
   51     "GROUP_REPLICATION_ADMIN",
   52     "INDEX",
   53     "INSERT",
   54     "LOCK TABLES",
   55     "PERSIST_RO_VARIABLES_ADMIN",
   56     "PROCESS",
   57     "REFERENCES",
   58     "RELOAD",
   59     "REPLICATION CLIENT",
   60     "REPLICATION SLAVE",
   61     "REPLICATION_SLAVE_ADMIN",
   62     "RESOURCE_GROUP_ADMIN",
   63     "RESOURCE_GROUP_USER",
   64     "ROLE_ADMIN",
   65     "SELECT",
   66     "SET_USER_ID",
   67     "SHOW DATABASES",
   68     "SHOW VIEW",
   69     "SHUTDOWN",
   70     "SUPER",
   71     "SYSTEM_VARIABLES_ADMIN",
   72     "TRIGGER",
   73     "UPDATE",
   74     "XA_RECOVER_ADMIN",
   75 ]
   76 
   77 
   78 class MockMySQLConnect:
   79     def __init__(self, *args, **kwargs):
   80         self.args = args
   81         self.kwargs = kwargs
   82 
   83     def autocommit(self, *args, **kwards):
   84         return True
   85 
   86 
   87 @skipIf(NO_MYSQL, "Install MySQL bindings before running MySQL unit tests.")
   88 class MySQLTestCase(TestCase, LoaderModuleMockMixin):
   89     def setup_loader_modules(self):
   90         return {mysql: {}}
   91 
   92     def test_user_exists(self):
   93         """
   94         Test to see if mysql module properly forms the MySQL query to see if a user exists
   95 
   96         Do it before test_user_create_when_user_exists mocks the user_exists call
   97         """
   98         with patch.object(mysql, "version", return_value="8.0.10"):
   99             self._test_call(
  100                 mysql.user_exists,
  101                 {
  102                     "sql": (
  103                         "SELECT User,Host FROM mysql.user WHERE "
  104                         "User = %(user)s AND Host = %(host)s AND "
  105                         "Password = PASSWORD(%(password)s)"
  106                     ),
  107                     "sql_args": {
  108                         "host": "localhost",
  109                         "password": "BLUECOW",
  110                         "user": "mytestuser",
  111                     },
  112                 },
  113                 user="mytestuser",
  114                 host="localhost",
  115                 password="BLUECOW",
  116             )
  117 
  118         with patch.object(mysql, "version", return_value="10.1.38-MariaDB"):
  119             self._test_call(
  120                 mysql.user_exists,
  121                 {
  122                     "sql": (
  123                         "SELECT User,Host FROM mysql.user WHERE "
  124                         "User = %(user)s AND Host = %(host)s AND "
  125                         "Password = PASSWORD(%(password)s)"
  126                     ),
  127                     "sql_args": {
  128                         "host": "localhost",
  129                         "password": "BLUECOW",
  130                         "user": "mytestuser",
  131                     },
  132                 },
  133                 user="mytestuser",
  134                 host="localhost",
  135                 password="BLUECOW",
  136             )
  137 
  138         with patch.object(mysql, "version", return_value="8.0.11"):
  139             self._test_call(
  140                 mysql.user_exists,
  141                 {
  142                     "sql": (
  143                         "SELECT User,Host FROM mysql.user WHERE "
  144                         "User = %(user)s AND Host = %(host)s"
  145                     ),
  146                     "sql_args": {"host": "localhost", "user": "mytestuser"},
  147                 },
  148                 user="mytestuser",
  149                 host="localhost",
  150                 password="BLUECOW",
  151             )
  152 
  153         with patch.object(mysql, "version", return_value="8.0.11"):
  154             with patch.object(
  155                 mysql,
  156                 "__get_auth_plugin",
  157                 MagicMock(return_value="mysql_native_password"),
  158             ):
  159                 self._test_call(
  160                     mysql.user_exists,
  161                     {
  162                         "sql": (
  163                             "SELECT User,Host FROM mysql.user WHERE "
  164                             "User = %(user)s AND Host = %(host)s AND "
  165                             "Password = %(password)s"
  166                         ),
  167                         "sql_args": {
  168                             "host": "%",
  169                             "password": "*1A01CF8FBE6425398935FB90359AD8B817399102",
  170                             "user": "mytestuser",
  171                         },
  172                     },
  173                     user="mytestuser",
  174                     host="%",
  175                     password="BLUECOW",
  176                 )
  177 
  178         with patch.object(mysql, "version", return_value="10.2.21-MariaDB"):
  179             self._test_call(
  180                 mysql.user_exists,
  181                 {
  182                     "sql": (
  183                         "SELECT User,Host FROM mysql.user WHERE "
  184                         "User = %(user)s AND Host = %(host)s AND "
  185                         "Password = PASSWORD(%(password)s)"
  186                     ),
  187                     "sql_args": {
  188                         "host": "localhost",
  189                         "password": "BLUECOW",
  190                         "user": "mytestuser",
  191                     },
  192                 },
  193                 user="mytestuser",
  194                 host="localhost",
  195                 password="BLUECOW",
  196             )
  197 
  198         with patch.object(
  199             mysql, "version", side_effect=["", "10.2.21-MariaDB", "10.2.21-MariaDB"]
  200         ):
  201             self._test_call(
  202                 mysql.user_exists,
  203                 {
  204                     "sql": (
  205                         "SELECT User,Host FROM mysql.user WHERE "
  206                         "User = %(user)s AND Host = %(host)s AND "
  207                         "Password = PASSWORD(%(password)s)"
  208                     ),
  209                     "sql_args": {
  210                         "host": "localhost",
  211                         "password": "new_pass",
  212                         "user": "root",
  213                     },
  214                 },
  215                 user="root",
  216                 host="localhost",
  217                 password="new_pass",
  218                 connection_user="root",
  219                 connection_pass="old_pass",
  220             )
  221 
  222         # test_user_create_when_user_exists(self):
  223         # ensure we don't try to create a user when one already exists
  224         # mock the version of MySQL
  225         with patch.object(mysql, "version", return_value="8.0.10"):
  226             with patch.object(mysql, "user_exists", MagicMock(return_value=True)):
  227                 with patch.dict(mysql.__salt__, {"config.option": MagicMock()}):
  228                     ret = mysql.user_create("testuser")
  229                     self.assertEqual(False, ret)
  230 
  231         # test_user_create_when_user_exists(self):
  232         # ensure we don't try to create a user when one already exists
  233         # mock the version of MySQL
  234         with patch.object(mysql, "version", return_value="8.0.11"):
  235             with patch.object(mysql, "user_exists", MagicMock(return_value=True)):
  236                 with patch.object(mysql, "verify_login", MagicMock(return_value=True)):
  237                     with patch.dict(mysql.__salt__, {"config.option": MagicMock()}):
  238                         ret = mysql.user_create("testuser")
  239                         self.assertEqual(False, ret)
  240 
  241     def test_user_create(self):
  242         """
  243         Test the creation of a MySQL user in mysql exec module
  244         """
  245         with patch.object(mysql, "version", return_value="8.0.10"):
  246             with patch.object(
  247                 mysql,
  248                 "__get_auth_plugin",
  249                 MagicMock(return_value="mysql_native_password"),
  250             ):
  251                 self._test_call(
  252                     mysql.user_create,
  253                     {
  254                         "sql": "CREATE USER %(user)s@%(host)s IDENTIFIED BY %(password)s",
  255                         "sql_args": {
  256                             "password": "BLUECOW",
  257                             "user": "testuser",
  258                             "host": "localhost",
  259                         },
  260                     },
  261                     "testuser",
  262                     password="BLUECOW",
  263                 )
  264 
  265         with patch.object(mysql, "version", return_value="8.0.11"):
  266             with patch.object(
  267                 mysql,
  268                 "__get_auth_plugin",
  269                 MagicMock(return_value="mysql_native_password"),
  270             ):
  271                 self._test_call(
  272                     mysql.user_create,
  273                     {
  274                         "sql": "CREATE USER %(user)s@%(host)s IDENTIFIED WITH %(auth_plugin)s BY %(password)s",
  275                         "sql_args": {
  276                             "password": "BLUECOW",
  277                             "auth_plugin": "mysql_native_password",
  278                             "user": "testuser",
  279                             "host": "localhost",
  280                         },
  281                     },
  282                     "testuser",
  283                     password="BLUECOW",
  284                 )
  285 
  286         # Test creating a user with passwordless=True and unix_socket=True
  287         with patch.object(mysql, "version", return_value="8.0.10"):
  288             with patch.object(mysql, "plugin_status", MagicMock(return_value="ACTIVE")):
  289                 self._test_call(
  290                     mysql.user_create,
  291                     {
  292                         "sql": "CREATE USER %(user)s@%(host)s IDENTIFIED WITH auth_socket",
  293                         "sql_args": {"user": "testuser", "host": "localhost"},
  294                     },
  295                     "testuser",
  296                     allow_passwordless=True,
  297                     unix_socket=True,
  298                 )
  299 
  300         with patch.object(mysql, "version", return_value="10.2.21-MariaDB"):
  301             with patch.object(mysql, "plugin_status", MagicMock(return_value="ACTIVE")):
  302                 self._test_call(
  303                     mysql.user_create,
  304                     {
  305                         "sql": "CREATE USER %(user)s@%(host)s IDENTIFIED VIA unix_socket",
  306                         "sql_args": {"user": "testuser", "host": "localhost"},
  307                     },
  308                     "testuser",
  309                     allow_passwordless=True,
  310                     unix_socket=True,
  311                 )
  312 
  313         with patch.object(mysql, "version", side_effect=["", "8.0.10", "8.0.10"]):
  314             with patch.object(
  315                 mysql, "user_exists", MagicMock(return_value=False)
  316             ), patch.object(
  317                 mysql,
  318                 "__get_auth_plugin",
  319                 MagicMock(return_value="mysql_native_password"),
  320             ):
  321                 self._test_call(
  322                     mysql.user_create,
  323                     {
  324                         "sql": "CREATE USER %(user)s@%(host)s IDENTIFIED BY %(password)s",
  325                         "sql_args": {
  326                             "password": "new_pass",
  327                             "user": "root",
  328                             "host": "localhost",
  329                         },
  330                     },
  331                     "root",
  332                     password="new_pass",
  333                     connection_user="root",
  334                     connection_pass="old_pass",
  335                 )
  336 
  337     def test_user_chpass(self):
  338         """
  339         Test changing a MySQL user password in mysql exec module
  340         """
  341         connect_mock = MagicMock()
  342         with patch.object(mysql, "_connect", connect_mock):
  343             with patch.object(mysql, "version", return_value="8.0.10"):
  344                 with patch.object(mysql, "user_exists", MagicMock(return_value=True)):
  345                     with patch.dict(mysql.__salt__, {"config.option": MagicMock()}):
  346                         mysql.user_chpass("testuser", password="BLUECOW")
  347                         calls = (
  348                             call()
  349                             .cursor()
  350                             .execute(
  351                                 "UPDATE mysql.user SET Password=PASSWORD(%(password)s) WHERE User=%(user)s AND Host = %(host)s;",
  352                                 {
  353                                     "password": "BLUECOW",
  354                                     "user": "testuser",
  355                                     "host": "localhost",
  356                                 },
  357                             ),
  358                             call().cursor().execute("FLUSH PRIVILEGES;"),
  359                         )
  360                         connect_mock.assert_has_calls(calls, any_order=True)
  361 
  362         connect_mock = MagicMock()
  363         with patch.object(mysql, "_connect", connect_mock):
  364             with patch.object(mysql, "version", return_value="8.0.11"):
  365                 with patch.object(mysql, "user_exists", MagicMock(return_value=True)):
  366                     with patch.dict(mysql.__salt__, {"config.option": MagicMock()}):
  367                         mysql.user_chpass("testuser", password="BLUECOW")
  368                         calls = (
  369                             call()
  370                             .cursor()
  371                             .execute(
  372                                 "ALTER USER %(user)s@%(host)s IDENTIFIED BY %(password)s;",
  373                                 {
  374                                     "password": "BLUECOW",
  375                                     "user": "testuser",
  376                                     "host": "localhost",
  377                                 },
  378                             ),
  379                             call().cursor().execute("FLUSH PRIVILEGES;"),
  380                         )
  381                         connect_mock.assert_has_calls(calls, any_order=True)
  382 
  383         connect_mock = MagicMock()
  384         with patch.object(mysql, "_connect", connect_mock):
  385             with patch.object(mysql, "version", side_effect=["", "8.0.11", "8.0.11"]):
  386                 with patch.object(mysql, "user_exists", MagicMock(return_value=True)):
  387                     with patch.dict(mysql.__salt__, {"config.option": MagicMock()}):
  388                         mysql.user_chpass(
  389                             "root",
  390                             password="new_pass",
  391                             connection_user="root",
  392                             connection_pass="old_pass",
  393                         )
  394                         calls = (
  395                             call()
  396                             .cursor()
  397                             .execute(
  398                                 "ALTER USER %(user)s@%(host)s IDENTIFIED BY %(password)s;",
  399                                 {
  400                                     "password": "new_pass",
  401                                     "user": "root",
  402                                     "host": "localhost",
  403                                 },
  404                             ),
  405                             call().cursor().execute("FLUSH PRIVILEGES;"),
  406                         )
  407                         connect_mock.assert_has_calls(calls, any_order=True)
  408 
  409     def test_user_remove(self):
  410         """
  411         Test the removal of a MySQL user in mysql exec module
  412         """
  413         with patch.object(mysql, "user_exists", MagicMock(return_value=True)):
  414             self._test_call(
  415                 mysql.user_remove,
  416                 {
  417                     "sql": "DROP USER %(user)s@%(host)s",
  418                     "sql_args": {"user": "testuser", "host": "localhost"},
  419                 },
  420                 "testuser",
  421             )
  422 
  423     def test_db_check(self):
  424         """
  425         Test MySQL db check function in mysql exec module
  426         """
  427         self._test_call(
  428             mysql.db_check,
  429             "CHECK TABLE `test``'\" db`.`my``'\" table`",
  430             "test`'\" db",
  431             "my`'\" table",
  432         )
  433 
  434     def test_db_repair(self):
  435         """
  436         Test MySQL db repair function in mysql exec module
  437         """
  438         self._test_call(
  439             mysql.db_repair,
  440             "REPAIR TABLE `test``'\" db`.`my``'\" table`",
  441             "test`'\" db",
  442             "my`'\" table",
  443         )
  444 
  445     def test_db_optimize(self):
  446         """
  447         Test MySQL db optimize function in mysql exec module
  448         """
  449         self._test_call(
  450             mysql.db_optimize,
  451             "OPTIMIZE TABLE `test``'\" db`.`my``'\" table`",
  452             "test`'\" db",
  453             "my`'\" table",
  454         )
  455 
  456     def test_db_remove(self):
  457         """
  458         Test MySQL db remove function in mysql exec module
  459         """
  460         with patch.object(mysql, "db_exists", MagicMock(return_value=True)):
  461             self._test_call(
  462                 mysql.db_remove, "DROP DATABASE `test``'\" db`;", "test`'\" db"
  463             )
  464 
  465     def test_db_tables(self):
  466         """
  467         Test MySQL db_tables function in mysql exec module
  468         """
  469         with patch.object(mysql, "db_exists", MagicMock(return_value=True)):
  470             self._test_call(
  471                 mysql.db_tables, "SHOW TABLES IN `test``'\" db`", "test`'\" db"
  472             )
  473 
  474     def test_db_exists(self):
  475         """
  476         Test MySQL db_exists function in mysql exec module
  477         """
  478         self._test_call(
  479             mysql.db_exists,
  480             {
  481                 "sql": "SHOW DATABASES LIKE %(dbname)s;",
  482                 "sql_args": {"dbname": r"""test%_`" db"""},
  483             },
  484             'test%_`" db',
  485         )
  486 
  487     def test_db_create(self):
  488         """
  489         Test MySQL db_create function in mysql exec module
  490         """
  491         self._test_call(
  492             mysql.db_create,
  493             "CREATE DATABASE IF NOT EXISTS `test``'\" db`;",
  494             "test`'\" db",
  495         )
  496 
  497     def test_user_list(self):
  498         """
  499         Test MySQL user_list function in mysql exec module
  500         """
  501         self._test_call(mysql.user_list, "SELECT User,Host FROM mysql.user")
  502 
  503     def test_user_info(self):
  504         """
  505         Test to see if the mysql execution module correctly forms the SQL for information on a MySQL user.
  506         """
  507         self._test_call(
  508             mysql.user_info,
  509             {
  510                 "sql": "SELECT * FROM mysql.user WHERE User = %(user)s AND Host = %(host)s",
  511                 "sql_args": {"host": "localhost", "user": "mytestuser"},
  512             },
  513             "mytestuser",
  514         )
  515 
  516     def test_user_grants(self):
  517         """
  518         Test to ensure the mysql user_grants function returns properly formed SQL for a basic query
  519         """
  520         with patch.object(mysql, "user_exists", MagicMock(return_value=True)):
  521             self._test_call(
  522                 mysql.user_grants,
  523                 {
  524                     "sql": "SHOW GRANTS FOR %(user)s@%(host)s",
  525                     "sql_args": {"host": "localhost", "user": "testuser"},
  526                 },
  527                 "testuser",
  528             )
  529 
  530     def test_grant_exists_true(self):
  531         """
  532         Test to ensure that we can find a grant that exists
  533         """
  534         mock_grants = [
  535             "GRANT USAGE ON *.* TO 'testuser'@'%'",
  536             "GRANT SELECT, INSERT, UPDATE ON `testdb`.`testtableone` TO 'testuser'@'%'",
  537             "GRANT SELECT(column1,column2) ON `testdb`.`testtableone` TO 'testuser'@'%'",
  538             "GRANT SELECT(column1,column2), INSERT(column1,column2) ON `testdb`.`testtableone` TO 'testuser'@'%'",
  539             "GRANT SELECT(column1,column2), UPDATE ON `testdb`.`testtableone` TO 'testuser'@'%'",
  540             "GRANT SELECT ON `testdb`.`testtabletwo` TO 'testuser'@'%'",
  541             "GRANT SELECT ON `testdb`.`testtablethree` TO 'testuser'@'%'",
  542         ]
  543         with patch.object(mysql, "version", return_value="5.6.41"):
  544             mock = MagicMock(return_value=mock_grants)
  545             with patch.object(
  546                 mysql, "user_grants", return_value=mock_grants
  547             ) as mock_user_grants:
  548                 ret = mysql.grant_exists(
  549                     "SELECT, INSERT, UPDATE", "testdb.testtableone", "testuser", "%"
  550                 )
  551                 self.assertEqual(ret, True)
  552 
  553     def test_grant_exists_false(self):
  554         """
  555         Test to ensure that we don't find a grant that doesn't exist
  556         """
  557         mock_grants = [
  558             "GRANT USAGE ON *.* TO 'testuser'@'%'",
  559             "GRANT SELECT, INSERT, UPDATE ON `testdb`.`testtableone` TO 'testuser'@'%'",
  560             "GRANT SELECT(column1,column2) ON `testdb`.`testtableone` TO 'testuser'@'%'",
  561             "GRANT SELECT(column1,column2), UPDATE ON `testdb`.`testtableone` TO 'testuser'@'%'",
  562             "GRANT SELECT ON `testdb`.`testtablethree` TO 'testuser'@'%'",
  563         ]
  564         with patch.object(mysql, "version", return_value="5.6.41"):
  565             mock = MagicMock(return_value=mock_grants)
  566             with patch.object(
  567                 mysql, "user_grants", return_value=mock_grants
  568             ) as mock_user_grants:
  569                 ret = mysql.grant_exists(
  570                     "SELECT", "testdb.testtabletwo", "testuser", "%"
  571                 )
  572                 self.assertEqual(ret, False)
  573 
  574     def test_grant_exists_all(self):
  575         """
  576         Test to ensure that we can find a grant that exists
  577         """
  578         mock_grants = [
  579             "GRANT SELECT, INSERT, UPDATE, DELETE, CREATE, DROP, RELOAD, SHUTDOWN, PROCESS, FILE, REFERENCES, INDEX, ALTER, SHOW DATABASES, SUPER, CREATE TEMPORARY TABLES, LOCK TABLES, EXECUTE, REPLICATION SLAVE, REPLICATION CLIENT, CREATE VIEW, SHOW VIEW, CREATE ROUTINE, ALTER ROUTINE, CREATE USER, EVENT, TRIGGER, CREATE TABLESPACE, CREATE ROLE, DROP ROLE ON testdb.testtableone TO `testuser`@`%`",
  580             "GRANT BACKUP_ADMIN,BINLOG_ADMIN,CONNECTION_ADMIN,ENCRYPTION_KEY_ADMIN,GROUP_REPLICATION_ADMIN,PERSIST_RO_VARIABLES_ADMIN,REPLICATION_SLAVE_ADMIN,RESOURCE_GROUP_ADMIN,RESOURCE_GROUP_USER,ROLE_ADMIN,SET_USER_ID,SYSTEM_VARIABLES_ADMIN,XA_RECOVER_ADMIN ON testdb.testtableone TO `testuser`@`%`",
  581         ]
  582         with patch.object(mysql, "version", return_value="8.0.10"):
  583             mock = MagicMock(return_value=mock_grants)
  584             with patch.object(
  585                 mysql, "user_grants", return_value=mock_grants
  586             ) as mock_user_grants:
  587                 ret = mysql.grant_exists("ALL", "testdb.testtableone", "testuser", "%")
  588                 self.assertEqual(ret, True)
  589 
  590         mock_grants = ["GRANT ALL PRIVILEGES ON testdb.testtableone TO `testuser`@`%`"]
  591         with patch.object(mysql, "version", return_value="5.6.41"):
  592             mock = MagicMock(return_value=mock_grants)
  593             with patch.object(
  594                 mysql, "user_grants", return_value=mock_grants
  595             ) as mock_user_grants:
  596                 ret = mysql.grant_exists(
  597                     "ALL PRIVILEGES", "testdb.testtableone", "testuser", "%"
  598                 )
  599                 self.assertEqual(ret, True)
  600 
  601     @skipIf(True, "TODO: Mock up user_grants()")
  602     def test_grant_add(self):
  603         """
  604         Test grant_add function in mysql exec module
  605         """
  606         self._test_call(
  607             mysql.grant_add,
  608             "",
  609             "SELECT,INSERT,UPDATE",
  610             "database.*",
  611             "frank",
  612             "localhost",
  613         )
  614 
  615     @skipIf(True, "TODO: Mock up user_grants()")
  616     def test_grant_revoke(self):
  617         """
  618         Test grant revoke in mysql exec module
  619         """
  620         self._test_call(
  621             mysql.grant_revoke,
  622             "",
  623             "SELECT,INSERT,UPDATE",
  624             "database.*",
  625             "frank",
  626             "localhost",
  627         )
  628 
  629     def test_processlist(self):
  630         """
  631         Test processlist function in mysql exec module
  632         """
  633         self._test_call(mysql.processlist, "SHOW FULL PROCESSLIST")
  634 
  635     def test_get_master_status(self):
  636         """
  637         Test get_master_status in the mysql execution module
  638         """
  639         self._test_call(mysql.get_master_status, "SHOW MASTER STATUS")
  640 
  641     def test_get_slave_status(self):
  642         """
  643         Test get_slave_status in the mysql execution module
  644         """
  645         self._test_call(mysql.get_slave_status, "SHOW SLAVE STATUS")
  646 
  647     def test_get_slave_status_bad_server(self):
  648         """
  649         Test get_slave_status in the mysql execution module, simulating a broken server
  650         """
  651         connect_mock = MagicMock(return_value=None)
  652         with patch.object(mysql, "_connect", connect_mock):
  653             with patch.dict(mysql.__salt__, {"config.option": MagicMock()}):
  654                 rslt = mysql.get_slave_status()
  655                 connect_mock.assert_has_calls([call()])
  656                 self.assertEqual(rslt, [])
  657 
  658     @skipIf(True, "MySQL module claims this function is not ready for production")
  659     def test_free_slave(self):
  660         pass
  661 
  662     def test_query(self):
  663         self._test_call(
  664             mysql.query, "SELECT * FROM testdb", "testdb", "SELECT * FROM testdb"
  665         )
  666 
  667     def test_query_error(self):
  668         connect_mock = MagicMock()
  669         with patch.object(mysql, "_connect", connect_mock):
  670             with patch.dict(mysql.__salt__, {"config.option": MagicMock()}):
  671                 # Use the OperationalError from the salt mysql module because that
  672                 # exception can come from either MySQLdb or pymysql
  673                 side_effect = mysql.OperationalError(9999, "Something Went Wrong")
  674                 with patch.object(
  675                     mysql, "_execute", MagicMock(side_effect=side_effect)
  676                 ):
  677                     mysql.query("testdb", "SELECT * FROM testdb")
  678             self.assertIn("mysql.error", mysql.__context__)
  679             expected = "MySQL Error 9999: Something Went Wrong"
  680             self.assertEqual(mysql.__context__["mysql.error"], expected)
  681 
  682     def test_plugin_add(self):
  683         """
  684         Test the adding/installing a MySQL / MariaDB plugin
  685         """
  686         with patch.object(mysql, "plugin_status", MagicMock(return_value="")):
  687             self._test_call(
  688                 mysql.plugin_add,
  689                 'INSTALL PLUGIN auth_socket SONAME "auth_socket.so"',
  690                 "auth_socket",
  691             )
  692 
  693     def test_plugin_remove(self):
  694         """
  695         Test the removing/uninstalling a MySQL / MariaDB plugin
  696         """
  697         with patch.object(mysql, "plugin_status", MagicMock(return_value="ACTIVE")):
  698             self._test_call(
  699                 mysql.plugin_remove, "UNINSTALL PLUGIN auth_socket", "auth_socket",
  700             )
  701 
  702     def test_plugin_status(self):
  703         """
  704         Test checking the status of a MySQL / MariaDB plugin
  705         """
  706         self._test_call(
  707             mysql.plugin_status,
  708             {
  709                 "sql": "SELECT PLUGIN_STATUS FROM INFORMATION_SCHEMA.PLUGINS WHERE PLUGIN_NAME = %(name)s",
  710                 "sql_args": {"name": "auth_socket"},
  711             },
  712             "auth_socket",
  713         )
  714 
  715     def test_sanitize_comment(self):
  716         """
  717         Test comment sanitization
  718         """
  719         input_data = """/*
  720         multiline
  721         comment
  722         */
  723         CREATE TABLE test_update (a VARCHAR(25)); # end of line comment
  724         # example comment
  725         insert into test_update values ("some #hash value");            -- ending comment
  726         insert into test_update values ("crazy -- not comment"); -- another ending comment
  727         -- another comment type
  728         """
  729         expected_response = """/*
  730 multiline
  731 comment
  732 */
  733 CREATE TABLE test_update (a VARCHAR(25));
  734 insert into test_update values ("some #hash value");
  735 insert into test_update values ("crazy -- not comment");
  736 """
  737         output = mysql._sanitize_comments(input_data)
  738         self.assertEqual(output, expected_response)
  739 
  740     def _test_call(self, function, expected_sql, *args, **kwargs):
  741         connect_mock = MagicMock()
  742         with patch.object(mysql, "_connect", connect_mock):
  743             with patch.dict(mysql.__salt__, {"config.option": MagicMock()}):
  744                 function(*args, **kwargs)
  745                 if isinstance(expected_sql, dict):
  746                     calls = (
  747                         call()
  748                         .cursor()
  749                         .execute(
  750                             "{}".format(expected_sql["sql"]), expected_sql["sql_args"]
  751                         )
  752                     )
  753                 else:
  754                     calls = call().cursor().execute("{}".format(expected_sql))
  755                 connect_mock.assert_has_calls((calls,), True)
  756 
  757     @skipIf(
  758         NO_PyMYSQL, "Install pymysql bindings before running test__connect_pymysql."
  759     )
  760     def test__connect_pymysql_exception(self):
  761         """
  762         Test the _connect function in the MySQL module
  763         """
  764         with patch.dict(mysql.__salt__, {"config.option": MagicMock()}):
  765             with patch(
  766                 "MySQLdb.connect",
  767                 side_effect=pymysql.err.InternalError(
  768                     1698, "Access denied for user 'root'@'localhost'"
  769                 ),
  770             ):
  771                 ret = mysql._connect()
  772                 self.assertIn("mysql.error", mysql.__context__)
  773                 self.assertEqual(
  774                     mysql.__context__["mysql.error"],
  775                     "MySQL Error 1698: Access denied for user 'root'@'localhost'",
  776                 )
  777 
  778     @skipIf(not NO_PyMYSQL, "With pymysql installed use test__connect_pymysql.")
  779     def test__connect_mysqldb_exception(self):
  780         """
  781         Test the _connect function in the MySQL module
  782         """
  783         with patch.dict(mysql.__salt__, {"config.option": MagicMock()}):
  784             with patch(
  785                 "MySQLdb.connect",
  786                 side_effect=mysql.OperationalError(
  787                     1698, "Access denied for user 'root'@'localhost'"
  788                 ),
  789             ):
  790                 ret = mysql._connect()
  791                 self.assertIn("mysql.error", mysql.__context__)
  792                 self.assertEqual(
  793                     mysql.__context__["mysql.error"],
  794                     "MySQL Error 1698: Access denied for user 'root'@'localhost'",
  795                 )
  796 
  797     def test__connect_mysqldb(self):
  798         """
  799         Test the _connect function in the MySQL module
  800         """
  801         with patch.dict(mysql.__salt__, {"config.option": MagicMock()}):
  802             with patch("MySQLdb.connect", return_value=MockMySQLConnect()):
  803                 ret = mysql._connect()
  804                 self.assertNotIn("mysql.error", mysql.__context__)