mirror of
https://github.com/valitydev/salt.git
synced 2024-11-08 17:33:54 +00:00
305 lines
11 KiB
Python
305 lines
11 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
'''
|
|
Tests for the MySQL state
|
|
'''
|
|
|
|
# Import python libs
|
|
import logging
|
|
|
|
|
|
# Import Salt Testing libs
|
|
from salttesting import skipIf
|
|
from salttesting.helpers import (
|
|
destructiveTest,
|
|
ensure_in_syspath
|
|
)
|
|
ensure_in_syspath('../../')
|
|
|
|
# Import salt libs
|
|
import integration
|
|
import salt.utils
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
NO_MYSQL = False
|
|
try:
|
|
import MySQLdb
|
|
except Exception:
|
|
NO_MYSQL = True
|
|
|
|
@skipIf(
|
|
NO_MYSQL,
|
|
'Please install MySQL bindings and a MySQL Server before running'
|
|
'MySQL integration tests.'
|
|
)
|
|
class MysqlDatabaseStateTest(integration.ModuleCase,
|
|
integration.SaltReturnAssertsMixIn):
|
|
'''
|
|
Validate the mysql_database state
|
|
'''
|
|
|
|
user = 'root'
|
|
password = 'poney'
|
|
|
|
|
|
@destructiveTest
|
|
def setUp(self):
|
|
'''
|
|
Test presence of MySQL server, enforce a root password
|
|
'''
|
|
super(MysqlDatabaseStateTest, 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 -u '
|
|
+ self.user
|
|
+ ' flush-privileges password "'
|
|
+ self.password
|
|
+ '"'
|
|
)
|
|
ret2 = self.run_state(
|
|
'cmd.run',
|
|
name='mysqladmin -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.')
|
|
|
|
@destructiveTest
|
|
def test_present_absent(self):
|
|
'''
|
|
mysql_database.present
|
|
'''
|
|
# In case of...
|
|
ret = self.run_state('mysql_database.absent',
|
|
name='testdb1',
|
|
character_set='utf8',
|
|
collate='utf8_general_ci',
|
|
connection_user=self.user,
|
|
connection_pass=self.password
|
|
)
|
|
ret = self.run_state('mysql_database.present',
|
|
name='testdb1',
|
|
character_set='utf8',
|
|
collate='utf8_general_ci',
|
|
connection_user=self.user,
|
|
connection_pass=self.password
|
|
)
|
|
self.assertSaltTrueReturn(ret)
|
|
self.assertInSaltComment('The database testdb1 has been created', ret)
|
|
#2nd run
|
|
ret = self.run_state('mysql_database.present',
|
|
name='testdb1',
|
|
character_set='utf8',
|
|
collate='utf8_general_ci',
|
|
connection_user=self.user,
|
|
connection_pass=self.password
|
|
)
|
|
self.assertSaltTrueReturn(ret)
|
|
self.assertInSaltComment('Database testdb1 is already present', ret)
|
|
self.assertSaltStateChangesEqual(ret, {})
|
|
|
|
# Test root connection on db
|
|
ret = self.run_function(
|
|
'mysql.query',
|
|
database='testdb1',
|
|
query='SELECT 1',
|
|
connection_user=self.user,
|
|
connection_pass=self.password,
|
|
connection_host='localhost'
|
|
)
|
|
if not isinstance(ret, dict) or not 'results' in ret:
|
|
raise AssertionError(
|
|
('Unexpected result while testing connection'
|
|
' on db {0!r}: {1}').format(
|
|
'testdb1',
|
|
repr(ret)
|
|
)
|
|
)
|
|
self.assertEqual([['1']], ret['results'])
|
|
|
|
# TODO: test with variations on collate and charset, check for db alter
|
|
# once it will be done in mysql_database.present state
|
|
|
|
ret = self.run_state('mysql_database.absent',
|
|
name='testdb1',
|
|
connection_user=self.user,
|
|
connection_pass=self.password
|
|
)
|
|
self.assertSaltTrueReturn(ret)
|
|
self.assertInSaltComment(
|
|
'Database testdb1 has been removed',
|
|
ret
|
|
)
|
|
#2nd run
|
|
ret = self.run_state('mysql_database.absent',
|
|
name='testdb1',
|
|
connection_user=self.user,
|
|
connection_pass=self.password
|
|
)
|
|
self.assertSaltTrueReturn(ret)
|
|
self.assertInSaltComment(
|
|
'Database testdb1 is not present, so it cannot be removed',
|
|
ret
|
|
)
|
|
self.assertSaltStateChangesEqual(ret, {})
|
|
|
|
|
|
@destructiveTest
|
|
def test_present_absent_fuzzy(self):
|
|
'''
|
|
mysql_database.present with utf-8 andf fuzzy db name
|
|
'''
|
|
# \xe6\xa8\x99\ = \u6a19 = 標
|
|
# this is : ":;,?:@=`&/標'\
|
|
dbname_utf8 = '":;,?:@=`&/\xe6\xa8\x99\'\\'
|
|
dbname_unicode = u'":;,?:@=`&/\u6a19\'\\'
|
|
#dbname_utf8 = '":;,?:@=`&/\u6a19\'\\'
|
|
#dbname_unicode = '":;,?:@=`&/\u6a19\'\\'
|
|
# In case of...
|
|
ret = self.run_state('mysql_database.absent',
|
|
name=dbname_utf8,
|
|
character_set='utf8',
|
|
collate='utf8_general_ci',
|
|
connection_user=self.user,
|
|
connection_pass=self.password,
|
|
connection_use_unicode=True,
|
|
connection_charset='utf8',
|
|
saltenv={"LC_ALL": "en_US.utf8"}
|
|
)
|
|
ret = self.run_state('mysql_database.present',
|
|
name=dbname_utf8,
|
|
character_set='utf8',
|
|
collate='utf8_general_ci',
|
|
connection_user=self.user,
|
|
connection_pass=self.password,
|
|
connection_use_unicode=True,
|
|
connection_charset='utf8',
|
|
saltenv={"LC_ALL": "en_US.utf8"}
|
|
)
|
|
self.assertSaltTrueReturn(ret)
|
|
# 'The database ":;,?: has been created' not found in 'Database ":;,?: is already present'
|
|
self.assertInSaltComment(
|
|
'The database ' + dbname_utf8 + ' has been created',
|
|
ret
|
|
)
|
|
#2nd run
|
|
ret = self.run_state('mysql_database.present',
|
|
name=dbname_unicode,
|
|
character_set='utf8',
|
|
collate='utf8_general_ci',
|
|
connection_user=self.user,
|
|
connection_pass=self.password,
|
|
connection_use_unicode=True,
|
|
connection_charset='utf8',
|
|
saltenv={"LC_ALL": "en_US.utf8"}
|
|
)
|
|
self.assertSaltTrueReturn(ret)
|
|
self.assertInSaltComment(
|
|
'Database ' + dbname_utf8 + ' is already present',
|
|
ret
|
|
)
|
|
self.assertSaltStateChangesEqual(ret, {})
|
|
|
|
# Test root connection on db
|
|
ret = self.run_function(
|
|
'mysql.query',
|
|
database=dbname_utf8,
|
|
query='SELECT 1',
|
|
connection_user=self.user,
|
|
connection_pass=self.password,
|
|
connection_host='localhost',
|
|
connection_use_unicode=True,
|
|
connection_charset='utf8',
|
|
)
|
|
if not isinstance(ret, dict) or not 'results' in ret:
|
|
raise AssertionError(
|
|
('Unexpected result while testing connection'
|
|
' on db {0!r}: {1}').format(
|
|
dbname_utf8,
|
|
repr(ret)
|
|
)
|
|
)
|
|
self.assertEqual([['1']], ret['results'])
|
|
|
|
ret = self.run_state('mysql_database.absent',
|
|
name=dbname_utf8,
|
|
character_set='utf8',
|
|
collate='utf8_general_ci',
|
|
connection_user=self.user,
|
|
connection_pass=self.password,
|
|
connection_use_unicode=True,
|
|
connection_charset='utf8',
|
|
saltenv={"LC_ALL": "en_US.utf8"}
|
|
)
|
|
self.assertSaltTrueReturn(ret)
|
|
self.assertInSaltComment(
|
|
'Database ' + dbname_utf8 + ' has been removed',
|
|
ret
|
|
)
|
|
#2nd run
|
|
ret = self.run_state('mysql_database.absent',
|
|
name=dbname_unicode,
|
|
character_set='utf8',
|
|
collate='utf8_general_ci',
|
|
connection_user=self.user,
|
|
connection_pass=self.password,
|
|
connection_use_unicode=True,
|
|
connection_charset='utf8',
|
|
saltenv={"LC_ALL": "en_US.utf8"}
|
|
)
|
|
self.assertSaltTrueReturn(ret)
|
|
self.assertInSaltComment(
|
|
'Database ' + dbname_utf8 + ' is not present, so it cannot be removed',
|
|
ret
|
|
)
|
|
self.assertSaltStateChangesEqual(ret, {})
|
|
|
|
@destructiveTest
|
|
@skipIf(True, 'This tests needs issue #8947 to be fixed first')
|
|
def test_utf8_from_sls_file(self):
|
|
'''
|
|
Try to create/destroy an utf-8 database name from an sls file #8947
|
|
'''
|
|
expected_result = {
|
|
'mysql_database_|-A_|-foo \xe6\xba\x96`bar_|-present': {
|
|
'__run_num__': 0,
|
|
'comment': 'The database foo \xe6\xba\x96`bar has been created',
|
|
'result': True},
|
|
'mysql_database_|-B_|-foo \xe6\xba\x96`bar_|-absent': {
|
|
'__run_num__': 1,
|
|
'comment': 'Database foo \xe6\xba\x96`bar has been removed',
|
|
'result': True},
|
|
}
|
|
result = {}
|
|
ret = self.run_function('state.sls', mods='mysql_utf8')
|
|
if not isinstance(ret, dict):
|
|
raise AssertionError(
|
|
('Unexpected result while testing external mysql utf8 sls'
|
|
': {0}').format(
|
|
repr(ret)
|
|
)
|
|
)
|
|
for item, descr in ret.iteritems():
|
|
result[item] = {
|
|
'__run_num__': descr['__run_num__'],
|
|
'comment': descr['comment'],
|
|
'result': descr['result']
|
|
}
|
|
self.assertEqual(expected_result, result)
|
|
|
|
if __name__ == '__main__':
|
|
from integration import run_tests
|
|
run_tests(MysqlDatabaseStateTest)
|