"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__)