12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154 |
- # -*- coding: utf-8 -*-
- '''
- :codeauthor: Denys Havrysh <denys.gavrysh@gmail.com>
- '''
- # Import python libs
- from __future__ import absolute_import, print_function, unicode_literals
- import os
- import shutil
- import tempfile
- # Import Salt Testing Libs
- from tests.support.unit import skipIf, TestCase
- from tests.support.runtests import RUNTIME_VARS
- from tests.support.mock import (
- MagicMock,
- patch,
- )
- # Import Salt Libs
- import salt.log.setup as log
- import salt.config
- import salt.syspaths
- import salt.utils.parsers
- import salt.utils.platform
- class ErrorMock(object): # pylint: disable=too-few-public-methods
- '''
- Error handling
- '''
- def __init__(self):
- '''
- init
- '''
- self.msg = None
- def error(self, msg):
- '''
- Capture error message
- '''
- self.msg = msg
- class LogSetupMock(object):
- '''
- Logger setup
- '''
- def __init__(self):
- '''
- init
- '''
- self.log_level = None
- self.log_file = None
- self.log_level_logfile = None
- self.config = {}
- self.temp_log_level = None
- def setup_console_logger(self, log_level='error', **kwargs): # pylint: disable=unused-argument
- '''
- Set console loglevel
- '''
- self.log_level = log_level
- def setup_extended_logging(self, opts):
- '''
- Set opts
- '''
- self.config = opts
- def setup_logfile_logger(self, logfile, loglevel, **kwargs): # pylint: disable=unused-argument
- '''
- Set logfile and loglevel
- '''
- self.log_file = logfile
- self.log_level_logfile = loglevel
- @staticmethod
- def get_multiprocessing_logging_queue(): # pylint: disable=invalid-name
- '''
- Mock
- '''
- import multiprocessing
- return multiprocessing.Queue()
- def setup_multiprocessing_logging_listener(self, opts, *args): # pylint: disable=invalid-name,unused-argument
- '''
- Set opts
- '''
- self.config = opts
- def setup_temp_logger(self, log_level='error'):
- '''
- Set temp loglevel
- '''
- self.temp_log_level = log_level
- class ObjectView(object): # pylint: disable=too-few-public-methods
- '''
- Dict object view
- '''
- def __init__(self, d):
- self.__dict__ = d
- class ParserBase(object):
- '''
- Unit Tests for Log Level Mixin with Salt parsers
- '''
- args = []
- skip_console_logging_config = False
- log_setup = None
- # Set config option names
- loglevel_config_setting_name = 'log_level'
- logfile_config_setting_name = 'log_file'
- logfile_loglevel_config_setting_name = 'log_level_logfile' # pylint: disable=invalid-name
- @classmethod
- def setUpClass(cls):
- cls.root_dir = tempfile.mkdtemp(dir=RUNTIME_VARS.TMP)
- @classmethod
- def tearDownClass(cls):
- shutil.rmtree(cls.root_dir, ignore_errors=True)
- def setup_log(self):
- '''
- Mock logger functions
- '''
- testing_config = self.default_config.copy()
- testing_config['root_dir'] = self.root_dir
- for name in ('pki_dir', 'cachedir'):
- testing_config[name] = name
- testing_config[self.logfile_config_setting_name] = getattr(self, self.logfile_config_setting_name, self.log_file)
- self.testing_config = testing_config
- self.addCleanup(setattr, self, 'testing_config', None)
- self.log_setup = LogSetupMock()
- patcher = patch.multiple(
- log,
- setup_console_logger=self.log_setup.setup_console_logger,
- setup_extended_logging=self.log_setup.setup_extended_logging,
- setup_logfile_logger=self.log_setup.setup_logfile_logger,
- get_multiprocessing_logging_queue=self.log_setup.get_multiprocessing_logging_queue,
- setup_multiprocessing_logging_listener=self.log_setup.setup_multiprocessing_logging_listener,
- setup_temp_logger=self.log_setup.setup_temp_logger
- )
- patcher.start()
- self.addCleanup(patcher.stop)
- self.addCleanup(setattr, self, 'log_setup', None)
- # log level configuration tests
- def test_get_log_level_cli(self):
- '''
- Tests that log level match command-line specified value
- '''
- # Set defaults
- default_log_level = self.testing_config[self.loglevel_config_setting_name]
- # Set log level in CLI
- log_level = 'critical'
- args = ['--log-level', log_level] + self.args
- parser = self.parser()
- with patch(self.config_func, MagicMock(return_value=self.testing_config)):
- parser.parse_args(args)
- with patch('salt.utils.parsers.is_writeable', MagicMock(return_value=True)):
- parser.setup_logfile_logger()
- console_log_level = getattr(parser.options, self.loglevel_config_setting_name)
- # Check console log level setting
- self.assertEqual(console_log_level, log_level)
- # Check console loggger log level
- self.assertEqual(self.log_setup.log_level, log_level)
- self.assertEqual(self.log_setup.config[self.loglevel_config_setting_name],
- log_level)
- self.assertEqual(self.log_setup.temp_log_level, log_level)
- # Check log file logger log level
- self.assertEqual(self.log_setup.log_level_logfile, default_log_level)
- def test_get_log_level_config(self):
- '''
- Tests that log level match the configured value
- '''
- args = self.args
- # Set log level in config
- log_level = 'info'
- opts = self.testing_config.copy()
- opts.update({self.loglevel_config_setting_name: log_level})
- parser = self.parser()
- with patch(self.config_func, MagicMock(return_value=opts)):
- parser.parse_args(args)
- with patch('salt.utils.parsers.is_writeable', MagicMock(return_value=True)):
- parser.setup_logfile_logger()
- console_log_level = getattr(parser.options, self.loglevel_config_setting_name)
- # Check console log level setting
- self.assertEqual(console_log_level, log_level)
- # Check console loggger log level
- self.assertEqual(self.log_setup.log_level, log_level)
- self.assertEqual(self.log_setup.config[self.loglevel_config_setting_name],
- log_level)
- self.assertEqual(self.log_setup.temp_log_level, 'error')
- # Check log file logger log level
- self.assertEqual(self.log_setup.log_level_logfile, log_level)
- def test_get_log_level_default(self):
- '''
- Tests that log level match the default value
- '''
- # Set defaults
- log_level = default_log_level = self.testing_config[self.loglevel_config_setting_name]
- args = self.args
- parser = self.parser()
- with patch(self.config_func, MagicMock(return_value=self.testing_config)):
- parser.parse_args(args)
- with patch('salt.utils.parsers.is_writeable', MagicMock(return_value=True)):
- parser.setup_logfile_logger()
- console_log_level = getattr(parser.options, self.loglevel_config_setting_name)
- # Check log level setting
- self.assertEqual(console_log_level, log_level)
- # Check console loggger log level
- self.assertEqual(self.log_setup.log_level, log_level)
- # Check extended logger
- self.assertEqual(self.log_setup.config[self.loglevel_config_setting_name],
- log_level)
- self.assertEqual(self.log_setup.temp_log_level, 'error')
- # Check log file logger
- self.assertEqual(self.log_setup.log_level_logfile, default_log_level)
- # Check help message
- self.assertIn('Default: \'{0}\'.'.format(default_log_level),
- parser.get_option('--log-level').help)
- # log file configuration tests
- def test_get_log_file_cli(self):
- '''
- Tests that log file match command-line specified value
- '''
- # Set defaults
- log_level = self.testing_config[self.loglevel_config_setting_name]
- # Set log file in CLI
- log_file = '{0}_cli.log'.format(self.log_file)
- args = ['--log-file', log_file] + self.args
- parser = self.parser()
- with patch(self.config_func, MagicMock(return_value=self.testing_config)):
- parser.parse_args(args)
- with patch('salt.utils.parsers.is_writeable', MagicMock(return_value=True)):
- parser.setup_logfile_logger()
- log_file_option = getattr(parser.options, self.logfile_config_setting_name)
- if not self.skip_console_logging_config:
- # Check console loggger
- self.assertEqual(self.log_setup.log_level, log_level)
- # Check extended logger
- self.assertEqual(self.log_setup.config[self.loglevel_config_setting_name],
- log_level)
- self.assertEqual(self.log_setup.config[self.logfile_config_setting_name],
- log_file)
- # Check temp logger
- self.assertEqual(self.log_setup.temp_log_level, 'error')
- # Check log file setting
- self.assertEqual(log_file_option, log_file)
- # Check log file logger
- self.assertEqual(self.log_setup.log_file, log_file)
- def test_get_log_file_config(self):
- '''
- Tests that log file match the configured value
- '''
- # Set defaults
- log_level = self.testing_config[self.loglevel_config_setting_name]
- args = self.args
- # Set log file in config
- log_file = '{0}_config.log'.format(self.log_file)
- opts = self.testing_config.copy()
- opts.update({self.logfile_config_setting_name: log_file})
- parser = self.parser()
- with patch(self.config_func, MagicMock(return_value=opts)):
- parser.parse_args(args)
- with patch('salt.utils.parsers.is_writeable', MagicMock(return_value=True)):
- parser.setup_logfile_logger()
- log_file_option = getattr(parser.options, self.logfile_config_setting_name)
- if not self.skip_console_logging_config:
- # Check console loggger
- self.assertEqual(self.log_setup.log_level, log_level)
- # Check extended logger
- self.assertEqual(self.log_setup.config[self.loglevel_config_setting_name],
- log_level)
- self.assertEqual(self.log_setup.config[self.logfile_config_setting_name],
- log_file)
- # Check temp logger
- self.assertEqual(self.log_setup.temp_log_level, 'error')
- # Check log file setting
- self.assertEqual(log_file_option, log_file)
- # Check log file logger
- self.assertEqual(self.log_setup.log_file, log_file)
- def test_get_log_file_default(self):
- '''
- Tests that log file match the default value
- '''
- # Set defaults
- log_level = self.testing_config[self.loglevel_config_setting_name]
- log_file = self.testing_config[self.logfile_config_setting_name]
- default_log_file = self.default_config[self.logfile_config_setting_name]
- args = self.args
- parser = self.parser()
- with patch(self.config_func, MagicMock(return_value=self.testing_config)):
- parser.parse_args(args)
- with patch('salt.utils.parsers.is_writeable', MagicMock(return_value=True)):
- parser.setup_logfile_logger()
- log_file_option = getattr(parser.options, self.logfile_config_setting_name)
- if not self.skip_console_logging_config:
- # Check console loggger
- self.assertEqual(self.log_setup.log_level, log_level)
- # Check extended logger
- self.assertEqual(self.log_setup.config[self.loglevel_config_setting_name],
- log_level)
- self.assertEqual(self.log_setup.config[self.logfile_config_setting_name],
- log_file)
- # Check temp logger
- self.assertEqual(self.log_setup.temp_log_level, 'error')
- # Check log file setting
- self.assertEqual(log_file_option, log_file)
- # Check log file logger
- self.assertEqual(self.log_setup.log_file, log_file)
- # Check help message
- self.assertIn('Default: \'{0}\'.'.format(default_log_file),
- parser.get_option('--log-file').help)
- # log file log level configuration tests
- def test_get_log_file_level_cli(self):
- '''
- Tests that file log level match command-line specified value
- '''
- # Set defaults
- default_log_level = self.testing_config[self.loglevel_config_setting_name]
- # Set log file level in CLI
- log_level_logfile = 'error'
- args = ['--log-file-level', log_level_logfile] + self.args
- parser = self.parser()
- with patch(self.config_func, MagicMock(return_value=self.testing_config)):
- parser.parse_args(args)
- with patch('salt.utils.parsers.is_writeable', MagicMock(return_value=True)):
- parser.setup_logfile_logger()
- log_level_logfile_option = getattr(parser.options,
- self.logfile_loglevel_config_setting_name)
- if not self.skip_console_logging_config:
- # Check console loggger
- self.assertEqual(self.log_setup.log_level, default_log_level)
- # Check extended logger
- self.assertEqual(self.log_setup.config[self.loglevel_config_setting_name],
- default_log_level)
- self.assertEqual(self.log_setup.config[self.logfile_loglevel_config_setting_name],
- log_level_logfile)
- # Check temp logger
- self.assertEqual(self.log_setup.temp_log_level, 'error')
- # Check log file level setting
- self.assertEqual(log_level_logfile_option, log_level_logfile)
- # Check log file logger
- self.assertEqual(self.log_setup.log_level_logfile, log_level_logfile)
- def test_get_log_file_level_config(self):
- '''
- Tests that log file level match the configured value
- '''
- # Set defaults
- log_level = self.testing_config[self.loglevel_config_setting_name]
- args = self.args
- # Set log file level in config
- log_level_logfile = 'info'
- opts = self.testing_config.copy()
- opts.update({self.logfile_loglevel_config_setting_name: log_level_logfile})
- parser = self.parser()
- with patch(self.config_func, MagicMock(return_value=opts)):
- parser.parse_args(args)
- with patch('salt.utils.parsers.is_writeable', MagicMock(return_value=True)):
- parser.setup_logfile_logger()
- log_level_logfile_option = getattr(parser.options,
- self.logfile_loglevel_config_setting_name)
- if not self.skip_console_logging_config:
- # Check console loggger
- self.assertEqual(self.log_setup.log_level, log_level)
- # Check extended logger
- self.assertEqual(self.log_setup.config[self.loglevel_config_setting_name],
- log_level)
- self.assertEqual(self.log_setup.config[self.logfile_loglevel_config_setting_name],
- log_level_logfile)
- # Check temp logger
- self.assertEqual(self.log_setup.temp_log_level, 'error')
- # Check log file level setting
- self.assertEqual(log_level_logfile_option, log_level_logfile)
- # Check log file logger
- self.assertEqual(self.log_setup.log_level_logfile, log_level_logfile)
- def test_get_log_file_level_default(self):
- '''
- Tests that log file level match the default value
- '''
- # Set defaults
- default_log_level = self.testing_config[self.loglevel_config_setting_name]
- log_level = default_log_level
- log_level_logfile = default_log_level
- args = self.args
- parser = self.parser()
- with patch(self.config_func, MagicMock(return_value=self.testing_config)):
- parser.parse_args(args)
- with patch('salt.utils.parsers.is_writeable', MagicMock(return_value=True)):
- parser.setup_logfile_logger()
- log_level_logfile_option = getattr(parser.options,
- self.logfile_loglevel_config_setting_name)
- if not self.skip_console_logging_config:
- # Check console loggger
- self.assertEqual(self.log_setup.log_level, log_level)
- # Check extended logger
- self.assertEqual(self.log_setup.config[self.loglevel_config_setting_name],
- log_level)
- self.assertEqual(self.log_setup.config[self.logfile_loglevel_config_setting_name],
- log_level_logfile)
- # Check temp logger
- self.assertEqual(self.log_setup.temp_log_level, 'error')
- # Check log file level setting
- self.assertEqual(log_level_logfile_option, log_level_logfile)
- # Check log file logger
- self.assertEqual(self.log_setup.log_level_logfile, log_level_logfile)
- # Check help message
- self.assertIn('Default: \'{0}\'.'.format(default_log_level),
- parser.get_option('--log-file-level').help)
- def test_get_console_log_level_with_file_log_level(self): # pylint: disable=invalid-name
- '''
- Tests that both console log level and log file level setting are working together
- '''
- log_level = 'critical'
- log_level_logfile = 'debug'
- args = ['--log-file-level', log_level_logfile] + self.args
- opts = self.testing_config.copy()
- opts.update({self.loglevel_config_setting_name: log_level})
- parser = self.parser()
- with patch(self.config_func, MagicMock(return_value=opts)):
- parser.parse_args(args)
- with patch('salt.utils.parsers.is_writeable', MagicMock(return_value=True)):
- parser.setup_logfile_logger()
- log_level_logfile_option = getattr(parser.options,
- self.logfile_loglevel_config_setting_name)
- if not self.skip_console_logging_config:
- # Check console loggger
- self.assertEqual(self.log_setup.log_level, log_level)
- # Check extended logger
- self.assertEqual(self.log_setup.config[self.loglevel_config_setting_name],
- log_level)
- self.assertEqual(self.log_setup.config[self.logfile_loglevel_config_setting_name],
- log_level_logfile)
- # Check temp logger
- self.assertEqual(self.log_setup.temp_log_level, 'error')
- # Check log file level setting
- self.assertEqual(log_level_logfile_option, log_level_logfile)
- # Check log file logger
- self.assertEqual(self.log_setup.log_level_logfile, log_level_logfile)
- @skipIf(salt.utils.platform.is_windows(), 'Windows uses a logging listener')
- def test_log_created(self):
- '''
- Tests that log file is created
- '''
- args = self.args
- log_file = self.log_file
- log_file_name = self.logfile_config_setting_name
- opts = self.testing_config.copy()
- opts.update({'log_file': log_file})
- if log_file_name != 'log_file':
- opts.update({log_file_name: getattr(self, log_file_name)})
- if log_file_name == 'key_logfile':
- self.skipTest('salt-key creates log file outside of parse_args.')
- parser = self.parser()
- with patch(self.config_func, MagicMock(return_value=opts)):
- parser.parse_args(args)
- if log_file_name == 'log_file':
- self.assertEqual(os.path.getsize(log_file), 0)
- else:
- self.assertEqual(os.path.getsize(getattr(self, log_file_name)), 0)
- def test_callbacks_uniqueness(self):
- '''
- Test that the callbacks are only added once, no matter
- how many instances of the parser we create
- '''
- mixin_container_names = ('_mixin_setup_funcs',
- '_mixin_process_funcs',
- '_mixin_after_parsed_funcs',
- '_mixin_before_exit_funcs')
- parser = self.parser()
- nums_1 = {}
- for cb_container in mixin_container_names:
- obj = getattr(parser, cb_container)
- nums_1[cb_container] = len(obj)
- # The next time we instantiate the parser, the counts should be equal
- parser = self.parser()
- nums_2 = {}
- for cb_container in mixin_container_names:
- obj = getattr(parser, cb_container)
- nums_2[cb_container] = len(obj)
- self.assertDictEqual(nums_1, nums_2)
- @skipIf(salt.utils.platform.is_windows(), 'Windows uses a logging listener')
- class MasterOptionParserTestCase(ParserBase, TestCase):
- '''
- Tests parsing Salt Master options
- '''
- def setUp(self):
- '''
- Setting up
- '''
- # Set defaults
- self.default_config = salt.config.DEFAULT_MASTER_OPTS.copy()
- self.addCleanup(delattr, self, 'default_config')
- # Log file
- self.log_file = '/tmp/salt_master_parser_test'
- # Function to patch
- self.config_func = 'salt.config.master_config'
- # Mock log setup
- self.setup_log()
- # Assign parser
- self.parser = salt.utils.parsers.MasterOptionParser
- self.addCleanup(delattr, self, 'parser')
- def tearDown(self):
- if os.path.exists(self.log_file):
- os.unlink(self.log_file)
- @skipIf(salt.utils.platform.is_windows(), 'Windows uses a logging listener')
- class MinionOptionParserTestCase(ParserBase, TestCase):
- '''
- Tests parsing Salt Minion options
- '''
- def setUp(self):
- '''
- Setting up
- '''
- # Set defaults
- self.default_config = salt.config.DEFAULT_MINION_OPTS.copy()
- self.addCleanup(delattr, self, 'default_config')
- # Log file
- self.log_file = '/tmp/salt_minion_parser_test'
- # Function to patch
- self.config_func = 'salt.config.minion_config'
- # Mock log setup
- self.setup_log()
- # Assign parser
- self.parser = salt.utils.parsers.MinionOptionParser
- self.addCleanup(delattr, self, 'parser')
- def tearDown(self):
- if os.path.exists(self.log_file):
- os.unlink(self.log_file)
- class ProxyMinionOptionParserTestCase(ParserBase, TestCase):
- '''
- Tests parsing Salt Proxy Minion options
- '''
- def setUp(self):
- '''
- Setting up
- '''
- # Set defaults
- self.default_config = salt.config.DEFAULT_MINION_OPTS.copy()
- self.default_config.update(salt.config.DEFAULT_PROXY_MINION_OPTS)
- self.addCleanup(delattr, self, 'default_config')
- # Log file
- self.log_file = '/tmp/salt_proxy_minion_parser_test'
- # Function to patch
- self.config_func = 'salt.config.proxy_config'
- # Mock log setup
- self.setup_log()
- # Assign parser
- self.parser = salt.utils.parsers.ProxyMinionOptionParser
- self.addCleanup(delattr, self, 'parser')
- def tearDown(self):
- if os.path.exists(self.log_file):
- os.unlink(self.log_file)
- @skipIf(salt.utils.platform.is_windows(), 'Windows uses a logging listener')
- class SyndicOptionParserTestCase(ParserBase, TestCase):
- '''
- Tests parsing Salt Syndic options
- '''
- def setUp(self):
- '''
- Setting up
- '''
- # Set config option names
- self.logfile_config_setting_name = 'syndic_log_file'
- # Set defaults
- self.default_config = salt.config.DEFAULT_MASTER_OPTS.copy()
- self.addCleanup(delattr, self, 'default_config')
- # Log file
- self.log_file = '/tmp/salt_syndic_parser_test'
- self.syndic_log_file = '/tmp/salt_syndic_log'
- # Function to patch
- self.config_func = 'salt.config.syndic_config'
- # Mock log setup
- self.setup_log()
- # Assign parser
- self.parser = salt.utils.parsers.SyndicOptionParser
- self.addCleanup(delattr, self, 'parser')
- def tearDown(self):
- if os.path.exists(self.log_file):
- os.unlink(self.log_file)
- if os.path.exists(self.syndic_log_file):
- os.unlink(self.syndic_log_file)
- class SaltCMDOptionParserTestCase(ParserBase, TestCase):
- '''
- Tests parsing Salt CLI options
- '''
- def setUp(self):
- '''
- Setting up
- '''
- # Set mandatory CLI options
- self.args = ['foo', 'bar.baz']
- # Set defaults
- self.default_config = salt.config.DEFAULT_MASTER_OPTS.copy()
- self.addCleanup(delattr, self, 'default_config')
- # Log file
- self.log_file = '/tmp/salt_cmd_parser_test'
- # Function to patch
- self.config_func = 'salt.config.client_config'
- # Mock log setup
- self.setup_log()
- # Assign parser
- self.parser = salt.utils.parsers.SaltCMDOptionParser
- self.addCleanup(delattr, self, 'parser')
- def tearDown(self):
- if os.path.exists(self.log_file):
- os.unlink(self.log_file)
- class SaltCPOptionParserTestCase(ParserBase, TestCase):
- '''
- Tests parsing salt-cp options
- '''
- def setUp(self):
- '''
- Setting up
- '''
- # Set mandatory CLI options
- self.args = ['foo', 'bar', 'baz']
- # Set defaults
- self.default_config = salt.config.DEFAULT_MASTER_OPTS.copy()
- self.addCleanup(delattr, self, 'default_config')
- # Log file
- self.log_file = '/tmp/salt_cp_parser_test'
- # Function to patch
- self.config_func = 'salt.config.master_config'
- # Mock log setup
- self.setup_log()
- # Assign parser
- self.parser = salt.utils.parsers.SaltCPOptionParser
- self.addCleanup(delattr, self, 'parser')
- def tearDown(self):
- if os.path.exists(self.log_file):
- os.unlink(self.log_file)
- class SaltKeyOptionParserTestCase(ParserBase, TestCase):
- '''
- Tests parsing salt-key options
- '''
- def setUp(self):
- '''
- Setting up
- '''
- self.skip_console_logging_config = True
- # Set config option names
- self.logfile_config_setting_name = 'key_logfile'
- # Set defaults
- self.default_config = salt.config.DEFAULT_MASTER_OPTS.copy()
- self.addCleanup(delattr, self, 'default_config')
- # Log file
- self.log_file = '/tmp/salt_key_parser_test'
- self.key_logfile = '/tmp/key_logfile'
- # Function to patch
- self.config_func = 'salt.config.client_config'
- # Mock log setup
- self.setup_log()
- # Assign parser
- self.parser = salt.utils.parsers.SaltKeyOptionParser
- self.addCleanup(delattr, self, 'parser')
- # log level configuration tests
- def test_get_log_level_cli(self):
- '''
- Tests that console log level option is not recognized
- '''
- # No console log level will be actually set
- log_level = default_log_level = None
- option = '--log-level'
- args = self.args + [option, 'error']
- parser = self.parser()
- mock_err = ErrorMock()
- with patch('salt.utils.parsers.OptionParser.error', mock_err.error):
- parser.parse_args(args)
- # Check error msg
- self.assertEqual(mock_err.msg, 'no such option: {0}'.format(option))
- # Check console loggger has not been set
- self.assertEqual(self.log_setup.log_level, log_level)
- self.assertNotIn(self.loglevel_config_setting_name, self.log_setup.config)
- # Check temp logger
- self.assertEqual(self.log_setup.temp_log_level, 'error')
- # Check log file logger log level
- self.assertEqual(self.log_setup.log_level_logfile, default_log_level)
- def test_get_log_level_config(self):
- '''
- Tests that log level set in config is ignored
- '''
- log_level = 'info'
- args = self.args
- # Set log level in config and set additional mocked opts keys
- opts = {self.loglevel_config_setting_name: log_level,
- self.logfile_config_setting_name: 'key_logfile',
- 'log_fmt_logfile': None,
- 'log_datefmt_logfile': None,
- 'log_rotate_max_bytes': None,
- 'log_rotate_backup_count': None}
- parser = self.parser()
- with patch(self.config_func, MagicMock(return_value=opts)):
- parser.parse_args(args)
- with patch('salt.utils.parsers.is_writeable', MagicMock(return_value=True)):
- parser.setup_logfile_logger()
- # Check config name absence in options
- self.assertNotIn(self.loglevel_config_setting_name, parser.options.__dict__)
- # Check console loggger has not been set
- self.assertEqual(self.log_setup.log_level, None)
- self.assertNotIn(self.loglevel_config_setting_name, self.log_setup.config)
- # Check temp logger
- self.assertEqual(self.log_setup.temp_log_level, 'error')
- # Check log file logger log level
- self.assertEqual(self.log_setup.log_level_logfile, log_level)
- def test_get_log_level_default(self):
- '''
- Tests that log level default value is ignored
- '''
- # Set defaults
- default_log_level = self.testing_config[self.loglevel_config_setting_name]
- log_level = None
- args = self.args
- parser = self.parser()
- parser.parse_args(args)
- with patch('salt.utils.parsers.is_writeable', MagicMock(return_value=True)):
- parser.setup_logfile_logger()
- # Check config name absence in options
- self.assertNotIn(self.loglevel_config_setting_name, parser.options.__dict__)
- # Check console loggger has not been set
- self.assertEqual(self.log_setup.log_level, log_level)
- self.assertNotIn(self.loglevel_config_setting_name, self.log_setup.config)
- # Check temp logger
- self.assertEqual(self.log_setup.temp_log_level, 'error')
- # Check log file logger log level
- self.assertEqual(self.log_setup.log_level_logfile, default_log_level)
- def tearDown(self):
- if os.path.exists(self.log_file):
- os.unlink(self.log_file)
- if os.path.exists(self.key_logfile):
- os.unlink(self.key_logfile)
- class SaltCallOptionParserTestCase(ParserBase, TestCase):
- '''
- Tests parsing Salt Minion options
- '''
- def setUp(self):
- '''
- Setting up
- '''
- # Set mandatory CLI options
- self.args = ['foo.bar']
- # Set defaults
- self.default_config = salt.config.DEFAULT_MINION_OPTS.copy()
- self.addCleanup(delattr, self, 'default_config')
- # Log file
- self.log_file = '/tmp/salt_call_parser_test'
- # Function to patch
- self.config_func = 'salt.config.minion_config'
- # Mock log setup
- self.setup_log()
- # Assign parser
- self.parser = salt.utils.parsers.SaltCallOptionParser
- self.addCleanup(delattr, self, 'parser')
- def tearDown(self):
- if os.path.exists(self.log_file):
- os.unlink(self.log_file)
- class SaltRunOptionParserTestCase(ParserBase, TestCase):
- '''
- Tests parsing Salt Master options
- '''
- def setUp(self):
- '''
- Setting up
- '''
- # Set mandatory CLI options
- self.args = ['foo.bar']
- # Set defaults
- self.default_config = salt.config.DEFAULT_MASTER_OPTS.copy()
- self.addCleanup(delattr, self, 'default_config')
- # Log file
- self.log_file = '/tmp/salt_run_parser_test'
- # Function to patch
- self.config_func = 'salt.config.master_config'
- # Mock log setup
- self.setup_log()
- # Assign parser
- self.parser = salt.utils.parsers.SaltRunOptionParser
- self.addCleanup(delattr, self, 'parser')
- def tearDown(self):
- if os.path.exists(self.log_file):
- os.unlink(self.log_file)
- class SaltSSHOptionParserTestCase(ParserBase, TestCase):
- '''
- Tests parsing Salt Master options
- '''
- def setUp(self):
- '''
- Setting up
- '''
- # Set mandatory CLI options
- self.args = ['foo', 'bar.baz']
- # Set config option names
- self.logfile_config_setting_name = 'ssh_log_file'
- # Set defaults
- self.default_config = salt.config.DEFAULT_MASTER_OPTS.copy()
- self.addCleanup(delattr, self, 'default_config')
- # Log file
- self.log_file = '/tmp/salt_ssh_parser_test'
- self.ssh_log_file = '/tmp/ssh_logfile'
- # Function to patch
- self.config_func = 'salt.config.master_config'
- # Mock log setup
- self.setup_log()
- # Assign parser
- self.parser = salt.utils.parsers.SaltSSHOptionParser
- self.addCleanup(delattr, self, 'parser')
- def tearDown(self):
- if os.path.exists(self.log_file):
- os.unlink(self.log_file)
- if os.path.exists(self.ssh_log_file):
- os.unlink(self.ssh_log_file)
- class SaltCloudParserTestCase(ParserBase, TestCase):
- '''
- Tests parsing Salt Cloud options
- '''
- def setUp(self):
- '''
- Setting up
- '''
- # Set mandatory CLI options
- self.args = ['-p', 'foo', 'bar']
- # Set default configs
- # Cloud configs are merged with master configs in
- # config/__init__.py, so we'll do that here as well
- # As we need the 'user' key later on.
- self.default_config = salt.config.DEFAULT_MASTER_OPTS.copy()
- self.default_config.update(salt.config.DEFAULT_CLOUD_OPTS)
- self.addCleanup(delattr, self, 'default_config')
- # Log file
- self.log_file = '/tmp/salt_cloud_parser_test'
- # Function to patch
- self.config_func = 'salt.config.cloud_config'
- # Mock log setup
- self.setup_log()
- # Assign parser
- self.parser = salt.utils.parsers.SaltCloudParser
- self.addCleanup(delattr, self, 'parser')
- def tearDown(self):
- if os.path.exists(self.log_file):
- os.unlink(self.log_file)
- class SPMParserTestCase(ParserBase, TestCase):
- '''
- Tests parsing Salt Cloud options
- '''
- def setUp(self):
- '''
- Setting up
- '''
- # Set mandatory CLI options
- self.args = ['foo', 'bar']
- # Set config option names
- self.logfile_config_setting_name = 'spm_logfile'
- # Set defaults
- self.default_config = salt.config.DEFAULT_MASTER_OPTS.copy()
- self.default_config.update(salt.config.DEFAULT_SPM_OPTS)
- self.addCleanup(delattr, self, 'default_config')
- # Log file
- self.log_file = '/tmp/spm_parser_test'
- self.spm_logfile = '/tmp/spm_logfile'
- # Function to patch
- self.config_func = 'salt.config.spm_config'
- # Mock log setup
- self.setup_log()
- # Assign parser
- self.parser = salt.utils.parsers.SPMParser
- self.addCleanup(delattr, self, 'parser')
- def tearDown(self):
- if os.path.exists(self.log_file):
- os.unlink(self.log_file)
- if os.path.exists(self.spm_logfile):
- os.unlink(self.spm_logfile)
- class SaltAPIParserTestCase(ParserBase, TestCase):
- '''
- Tests parsing Salt Cloud options
- '''
- def setUp(self):
- '''
- Setting up
- '''
- # Set mandatory CLI options
- self.args = []
- # Set config option names
- self.logfile_config_setting_name = 'api_logfile'
- # Set defaults
- self.default_config = salt.config.DEFAULT_MASTER_OPTS.copy()
- self.default_config.update(salt.config.DEFAULT_API_OPTS)
- self.addCleanup(delattr, self, 'default_config')
- # Log file
- self.log_file = '/tmp/salt_api_parser_test'
- self.api_logfile = '/tmp/api_logfile'
- # Function to patch
- self.config_func = 'salt.config.api_config'
- # Mock log setup
- self.setup_log()
- # Assign parser
- self.parser = salt.utils.parsers.SaltAPIParser
- self.addCleanup(delattr, self, 'parser')
- def tearDown(self):
- if os.path.exists(self.log_file):
- os.unlink(self.log_file)
- if os.path.exists(self.api_logfile):
- os.unlink(self.api_logfile)
- class DaemonMixInTestCase(TestCase):
- '''
- Tests the PIDfile deletion in the DaemonMixIn.
- '''
- def setUp(self):
- '''
- Setting up
- '''
- # Setup mixin
- self.daemon_mixin = salt.utils.parsers.DaemonMixIn()
- self.daemon_mixin.config = {}
- self.daemon_mixin.config['pidfile'] = '/some/fake.pid'
- def tearDown(self):
- '''
- Tear down test
- :return:
- '''
- del self.daemon_mixin
- @patch('os.unlink', MagicMock())
- @patch('os.path.isfile', MagicMock(return_value=True))
- @patch('salt.utils.parsers.logger', MagicMock())
- def test_pid_file_deletion(self):
- '''
- PIDfile deletion without exception.
- '''
- self.daemon_mixin._mixin_before_exit()
- assert salt.utils.parsers.os.unlink.call_count == 1
- salt.utils.parsers.logger.info.assert_not_called()
- salt.utils.parsers.logger.debug.assert_not_called()
- @patch('os.unlink', MagicMock(side_effect=OSError()))
- @patch('os.path.isfile', MagicMock(return_value=True))
- @patch('salt.utils.parsers.logger', MagicMock())
- def test_pid_deleted_oserror_as_root(self):
- '''
- PIDfile deletion with exception, running as root.
- '''
- if salt.utils.platform.is_windows():
- patch_args = ('salt.utils.win_functions.is_admin',
- MagicMock(return_value=True))
- else:
- patch_args = ('os.getuid', MagicMock(return_value=0))
- with patch(*patch_args):
- self.daemon_mixin._mixin_before_exit()
- assert salt.utils.parsers.os.unlink.call_count == 1
- salt.utils.parsers.logger.info.assert_called_with(
- 'PIDfile could not be deleted: %s',
- format(self.daemon_mixin.config['pidfile'])
- )
- salt.utils.parsers.logger.debug.assert_called()
- @patch('os.unlink', MagicMock(side_effect=OSError()))
- @patch('os.path.isfile', MagicMock(return_value=True))
- @patch('salt.utils.parsers.logger', MagicMock())
- def test_pid_deleted_oserror_as_non_root(self):
- '''
- PIDfile deletion with exception, running as non-root.
- '''
- if salt.utils.platform.is_windows():
- patch_args = ('salt.utils.win_functions.is_admin',
- MagicMock(return_value=False))
- else:
- patch_args = ('os.getuid', MagicMock(return_value=1000))
- with patch(*patch_args):
- self.daemon_mixin._mixin_before_exit()
- assert salt.utils.parsers.os.unlink.call_count == 1
- salt.utils.parsers.logger.info.assert_not_called()
- salt.utils.parsers.logger.debug.assert_not_called()
|