valitydev/salt
14
0
Fork 0
mirror of https://github.com/valitydev/salt.git synced 2024-11-07 08:58:59 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

adding states/mysql_query unit test case

Browse Source
This commit is contained in:
Jayesh Kariya 2015-05-25 15:59:25 +05:30
parent 6b43c9a8cb
commit ec50450460
1 changed files with 131 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

131
tests/unit/states/mysql_query_test.py Normal file
View File

@ -0,0 +1,131 @@
# -*- coding: utf-8 -*-
'''
:codeauthor: :email:`Jayesh Kariya <jayeshk@saltstack.com>`
'''
# Import Python libs
from __future__ import absolute_import
# Import Salt Testing Libs
from salttesting import skipIf, TestCase
from salttesting.mock import (
NO_MOCK,
NO_MOCK_REASON,
MagicMock,
patch)
from salttesting.helpers import ensure_in_syspath
ensure_in_syspath('../../')
# Import Salt Libs
from salt.states import mysql_query
import os
mysql_query.__salt__ = {}
mysql_query.__opts__ = {}
@skipIf(NO_MOCK, NO_MOCK_REASON)
class MysqlQueryTestCase(TestCase):
'''
Test cases for salt.states.mysql_query
'''
# 'run' function tests: 1
def test_run(self):
'''
Test to execute an arbitrary query on the specified database.
'''
name = 'query_id'
database = 'my_database'
query = "SELECT * FROM table;"
ret = {'name': name,
'result': True,
'comment': '',
'changes': {}}
mock_t = MagicMock(return_value=True)
mock_f = MagicMock(return_value=False)
mock_str = MagicMock(return_value='salt')
mock_none = MagicMock(return_value=None)
mock_dict = MagicMock(return_value={'salt': 'SALT'})
mock_lst = MagicMock(return_value=['grain'])
with patch.dict(mysql_query.__salt__, {'mysql.db_exists': mock_f}):
with patch.object(mysql_query, '_get_mysql_error', mock_str):
ret.update({'comment': 'salt', 'result': False})
self.assertDictEqual(mysql_query.run(name, database, query),
ret)
with patch.object(mysql_query, '_get_mysql_error', mock_none):
comt = ('Database {0} is not present'.format(name))
ret.update({'comment': comt, 'result': None})
self.assertDictEqual(mysql_query.run(name, database, query),
ret)
with patch.dict(mysql_query.__salt__, {'mysql.db_exists': mock_t,
'grains.ls': mock_lst,
'grains.get': mock_dict,
'mysql.query': mock_str}):
comt = ('No execution needed. Grain grain already set')
ret.update({'comment': comt, 'result': True})
self.assertDictEqual(mysql_query.run(name, database, query,
output='grain', grain='grain',
overwrite=False), ret)
with patch.dict(mysql_query.__opts__, {'test': True}):
comt = ('Query would execute, storing result in grain: grain')
ret.update({'comment': comt, 'result': None})
self.assertDictEqual(mysql_query.run(name, database, query,
output='grain',
grain='grain'), ret)
comt = ('Query would execute, storing result'
' in grain: grain:salt')
ret.update({'comment': comt})
self.assertDictEqual(mysql_query.run(name, database, query,
output='grain',
grain='grain',
key='salt'), ret)
comt = ('Query would execute, storing result in file: salt')
ret.update({'comment': comt})
self.assertDictEqual(mysql_query.run(name, database, query,
output='salt',
grain='grain'), ret)
comt = ('Query would execute, not storing result')
ret.update({'comment': comt})
self.assertDictEqual(mysql_query.run(name, database, query),
ret)
comt = ('No execution needed. Grain grain:salt already set')
ret.update({'comment': comt, 'result': True})
self.assertDictEqual(mysql_query.run(name, database, query,
output='grain', grain='grain',
key='salt', overwrite=False),
ret)
comt = ("Error: output type 'grain' needs the grain parameter\n")
ret.update({'comment': comt, 'result': False})
self.assertDictEqual(mysql_query.run(name, database, query,
output='grain'), ret)
with patch.object(os.path, 'isfile', mock_t):
comt = ('No execution needed. File salt already set')
ret.update({'comment': comt, 'result': True})
self.assertDictEqual(mysql_query.run(name, database, query,
output='salt',
grain='grain',
overwrite=False), ret)
with patch.dict(mysql_query.__opts__, {'test': False}):
ret.update({'comment': 'salt',
'changes': {'query': 'Executed'}})
self.assertDictEqual(mysql_query.run(name, database, query),
ret)
if __name__ == '__main__':
from integration import run_tests
run_tests(MysqlQueryTestCase, needs_daemon=False)