"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

---