"Fossies" - the Fresh Open Source Software Archive  

Source code changes of the file "tests/unit/modules/test_mysql.py" between
salt-3002.1.tar.gz and salt-3002.2.tar.gz

About: SaltStack is a systems management software for data center automation, cloud orchestration, server provisioning, configuration management and more. Community version.

test_mysql.py  (salt-3002.1):test_mysql.py  (salt-3002.2)
# -*- coding: utf-8 -*-
""" """
:codeauthor: Mike Place (mp@saltstack.com) :codeauthor: Mike Place (mp@saltstack.com)
tests.unit.modules.mysql tests.unit.modules.mysql
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
""" """
# Import Python libs
from __future__ import absolute_import, print_function, unicode_literals
import logging import logging
# Import salt libs
import salt.modules.mysql as mysql import salt.modules.mysql as mysql
# Import Salt Testing libs
from tests.support.mixins import LoaderModuleMockMixin from tests.support.mixins import LoaderModuleMockMixin
from tests.support.mock import MagicMock, call, patch from tests.support.mock import MagicMock, call, patch
from tests.support.unit import TestCase, skipIf from tests.support.unit import TestCase, skipIf
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
NO_MYSQL = False NO_MYSQL = False
NO_PyMYSQL = False NO_PyMYSQL = False
try: try:
import MySQLdb # pylint: disable=W0611 import MySQLdb # pylint: disable=W0611
except ImportError: except ImportError:
skipping to change at line 82 skipping to change at line 75
"SHOW DATABASES", "SHOW DATABASES",
"SHOW VIEW", "SHOW VIEW",
"SHUTDOWN", "SHUTDOWN",
"SUPER", "SUPER",
"SYSTEM_VARIABLES_ADMIN", "SYSTEM_VARIABLES_ADMIN",
"TRIGGER", "TRIGGER",
"UPDATE", "UPDATE",
"XA_RECOVER_ADMIN", "XA_RECOVER_ADMIN",
] ]
class MockMySQLConnect(object): class MockMySQLConnect:
def __init__(self, *args, **kwargs): def __init__(self, *args, **kwargs):
self.args = args self.args = args
self.kwargs = kwargs self.kwargs = kwargs
def autocommit(self, *args, **kwards): def autocommit(self, *args, **kwards):
return True return True
@skipIf(NO_MYSQL, "Install MySQL bindings before running MySQL unit tests.") @skipIf(NO_MYSQL, "Install MySQL bindings before running MySQL unit tests.")
class MySQLTestCase(TestCase, LoaderModuleMockMixin): class MySQLTestCase(TestCase, LoaderModuleMockMixin):
def setup_loader_modules(self): def setup_loader_modules(self):
skipping to change at line 201 skipping to change at line 194
"host": "localhost", "host": "localhost",
"password": "BLUECOW", "password": "BLUECOW",
"user": "mytestuser", "user": "mytestuser",
}, },
}, },
user="mytestuser", user="mytestuser",
host="localhost", host="localhost",
password="BLUECOW", password="BLUECOW",
) )
with patch.object(
mysql, "version", side_effect=["", "10.2.21-MariaDB", "10.2.21-Maria
DB"]
):
self._test_call(
mysql.user_exists,
{
"sql": (
"SELECT User,Host FROM mysql.user WHERE "
"User = %(user)s AND Host = %(host)s AND "
"Password = PASSWORD(%(password)s)"
),
"sql_args": {
"host": "localhost",
"password": "new_pass",
"user": "root",
},
},
user="root",
host="localhost",
password="new_pass",
connection_user="root",
connection_pass="old_pass",
)
# test_user_create_when_user_exists(self): # test_user_create_when_user_exists(self):
# ensure we don't try to create a user when one already exists # ensure we don't try to create a user when one already exists
# mock the version of MySQL # mock the version of MySQL
with patch.object(mysql, "version", return_value="8.0.10"): with patch.object(mysql, "version", return_value="8.0.10"):
with patch.object(mysql, "user_exists", MagicMock(return_value=True) ): with patch.object(mysql, "user_exists", MagicMock(return_value=True) ):
with patch.dict(mysql.__salt__, {"config.option": MagicMock()}): with patch.dict(mysql.__salt__, {"config.option": MagicMock()}):
ret = mysql.user_create("testuser") ret = mysql.user_create("testuser")
self.assertEqual(False, ret) self.assertEqual(False, ret)
# test_user_create_when_user_exists(self): # test_user_create_when_user_exists(self):
skipping to change at line 292 skipping to change at line 309
mysql.user_create, mysql.user_create,
{ {
"sql": "CREATE USER %(user)s@%(host)s IDENTIFIED VIA uni x_socket", "sql": "CREATE USER %(user)s@%(host)s IDENTIFIED VIA uni x_socket",
"sql_args": {"user": "testuser", "host": "localhost"}, "sql_args": {"user": "testuser", "host": "localhost"},
}, },
"testuser", "testuser",
allow_passwordless=True, allow_passwordless=True,
unix_socket=True, unix_socket=True,
) )
with patch.object(mysql, "version", side_effect=["", "8.0.10", "8.0.10"]
):
with patch.object(
mysql, "user_exists", MagicMock(return_value=False)
), patch.object(
mysql,
"__get_auth_plugin",
MagicMock(return_value="mysql_native_password"),
):
self._test_call(
mysql.user_create,
{
"sql": "CREATE USER %(user)s@%(host)s IDENTIFIED BY %(pa
ssword)s",
"sql_args": {
"password": "new_pass",
"user": "root",
"host": "localhost",
},
},
"root",
password="new_pass",
connection_user="root",
connection_pass="old_pass",
)
def test_user_chpass(self): def test_user_chpass(self):
""" """
Test changing a MySQL user password in mysql exec module Test changing a MySQL user password in mysql exec module
""" """
connect_mock = MagicMock() connect_mock = MagicMock()
with patch.object(mysql, "_connect", connect_mock): with patch.object(mysql, "_connect", connect_mock):
with patch.object(mysql, "version", return_value="8.0.10"): with patch.object(mysql, "version", return_value="8.0.10"):
with patch.object(mysql, "user_exists", MagicMock(return_value=T rue)): with patch.object(mysql, "user_exists", MagicMock(return_value=T rue)):
with patch.dict(mysql.__salt__, {"config.option": MagicMock( )}): with patch.dict(mysql.__salt__, {"config.option": MagicMock( )}):
mysql.user_chpass("testuser", password="BLUECOW") mysql.user_chpass("testuser", password="BLUECOW")
skipping to change at line 338 skipping to change at line 379
{ {
"password": "BLUECOW", "password": "BLUECOW",
"user": "testuser", "user": "testuser",
"host": "localhost", "host": "localhost",
}, },
), ),
call().cursor().execute("FLUSH PRIVILEGES;"), call().cursor().execute("FLUSH PRIVILEGES;"),
) )
connect_mock.assert_has_calls(calls, any_order=True) connect_mock.assert_has_calls(calls, any_order=True)
connect_mock = MagicMock()
with patch.object(mysql, "_connect", connect_mock):
with patch.object(mysql, "version", side_effect=["", "8.0.11", "8.0.
11"]):
with patch.object(mysql, "user_exists", MagicMock(return_value=T
rue)):
with patch.dict(mysql.__salt__, {"config.option": MagicMock(
)}):
mysql.user_chpass(
"root",
password="new_pass",
connection_user="root",
connection_pass="old_pass",
)
calls = (
call()
.cursor()
.execute(
"ALTER USER %(user)s@%(host)s IDENTIFIED BY %(pa
ssword)s;",
{
"password": "new_pass",
"user": "root",
"host": "localhost",
},
),
call().cursor().execute("FLUSH PRIVILEGES;"),
)
connect_mock.assert_has_calls(calls, any_order=True)
def test_user_remove(self): def test_user_remove(self):
""" """
Test the removal of a MySQL user in mysql exec module Test the removal of a MySQL user in mysql exec module
""" """
with patch.object(mysql, "user_exists", MagicMock(return_value=True)): with patch.object(mysql, "user_exists", MagicMock(return_value=True)):
self._test_call( self._test_call(
mysql.user_remove, mysql.user_remove,
{ {
"sql": "DROP USER %(user)s@%(host)s", "sql": "DROP USER %(user)s@%(host)s",
"sql_args": {"user": "testuser", "host": "localhost"}, "sql_args": {"user": "testuser", "host": "localhost"},
skipping to change at line 679 skipping to change at line 746
def _test_call(self, function, expected_sql, *args, **kwargs): def _test_call(self, function, expected_sql, *args, **kwargs):
connect_mock = MagicMock() connect_mock = MagicMock()
with patch.object(mysql, "_connect", connect_mock): with patch.object(mysql, "_connect", connect_mock):
with patch.dict(mysql.__salt__, {"config.option": MagicMock()}): with patch.dict(mysql.__salt__, {"config.option": MagicMock()}):
function(*args, **kwargs) function(*args, **kwargs)
if isinstance(expected_sql, dict): if isinstance(expected_sql, dict):
calls = ( calls = (
call() call()
.cursor() .cursor()
.execute( .execute(
"{0}".format(expected_sql["sql"]), expected_sql["sql _args"] "{}".format(expected_sql["sql"]), expected_sql["sql_ args"]
) )
) )
else: else:
calls = call().cursor().execute("{0}".format(expected_sql)) calls = call().cursor().execute("{}".format(expected_sql))
connect_mock.assert_has_calls((calls,), True) connect_mock.assert_has_calls((calls,), True)
@skipIf( @skipIf(
NO_PyMYSQL, "Install pymysql bindings before running test__connect_pymys ql." NO_PyMYSQL, "Install pymysql bindings before running test__connect_pymys ql."
) )
def test__connect_pymysql_exception(self): def test__connect_pymysql_exception(self):
""" """
Test the _connect function in the MySQL module Test the _connect function in the MySQL module
""" """
with patch.dict(mysql.__salt__, {"config.option": MagicMock()}): with patch.dict(mysql.__salt__, {"config.option": MagicMock()}):
 End of changes. 10 change blocks. 
10 lines changed or deleted 84 lines changed or added

Home  |  About  |  Features  |  All  |  Newest  |  Dox  |  Diffs  |  RSS Feeds  |  Screenshots  |  Comments  |  Imprint  |  Privacy  |  HTTP(S)