# -*- coding: utf-8 -*- ''' :codeauthor: Denys Havrysh ''' # Import python libs from __future__ import absolute_import, print_function, unicode_literals import os # Import Salt Testing Libs from tests.support.unit import skipIf, TestCase from tests.support.helpers import destructiveTest, skip_if_not_root from tests.support.mock import ( MagicMock, patch, NO_MOCK, NO_MOCK_REASON ) # Import Salt Libs import salt.log.setup as log import salt.config import salt.syspaths import salt.utils.parsers import salt.utils.platform try: import pytest except ImportError: pytest = None 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 @destructiveTest @skip_if_not_root class LogSettingsParserTests(TestCase): ''' 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 def setup_log(self): ''' Mock logger functions ''' 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.default_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.default_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.default_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.default_config[self.loglevel_config_setting_name] args = self.args parser = self.parser() with patch(self.config_func, MagicMock(return_value=self.default_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.default_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.default_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.default_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.default_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.default_config[self.loglevel_config_setting_name] log_file = 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.default_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.default_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.default_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.default_config[self.loglevel_config_setting_name] args = self.args # Set log file level in config log_level_logfile = 'info' opts = self.default_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.default_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.default_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.default_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.default_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) @skipIf(NO_MOCK, NO_MOCK_REASON) @skipIf(salt.utils.platform.is_windows(), 'Windows uses a logging listener') class MasterOptionParserTestCase(LogSettingsParserTests): ''' Tests parsing Salt Master options ''' def setUp(self): ''' Setting up ''' # Set defaults self.default_config = salt.config.DEFAULT_MASTER_OPTS 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') @skipIf(NO_MOCK, NO_MOCK_REASON) @skipIf(salt.utils.platform.is_windows(), 'Windows uses a logging listener') class MinionOptionParserTestCase(LogSettingsParserTests): ''' Tests parsing Salt Minion options ''' def setUp(self): ''' Setting up ''' # Set defaults self.default_config = salt.config.DEFAULT_MINION_OPTS 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') @skipIf(NO_MOCK, NO_MOCK_REASON) class ProxyMinionOptionParserTestCase(LogSettingsParserTests): ''' 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') @skipIf(NO_MOCK, NO_MOCK_REASON) @skipIf(salt.utils.platform.is_windows(), 'Windows uses a logging listener') class SyndicOptionParserTestCase(LogSettingsParserTests): ''' 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 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') @skipIf(NO_MOCK, NO_MOCK_REASON) class SaltCMDOptionParserTestCase(LogSettingsParserTests): ''' 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 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') @skipIf(NO_MOCK, NO_MOCK_REASON) class SaltCPOptionParserTestCase(LogSettingsParserTests): ''' 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 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') @skipIf(NO_MOCK, NO_MOCK_REASON) class SaltKeyOptionParserTestCase(LogSettingsParserTests): ''' 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 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.master_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.default_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) @skipIf(NO_MOCK, NO_MOCK_REASON) class SaltCallOptionParserTestCase(LogSettingsParserTests): ''' 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 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') @skipIf(NO_MOCK, NO_MOCK_REASON) class SaltRunOptionParserTestCase(LogSettingsParserTests): ''' 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 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') @skipIf(NO_MOCK, NO_MOCK_REASON) class SaltSSHOptionParserTestCase(LogSettingsParserTests): ''' 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 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') @skipIf(NO_MOCK, NO_MOCK_REASON) class SaltCloudParserTestCase(LogSettingsParserTests): ''' 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') @skipIf(NO_MOCK, NO_MOCK_REASON) class SPMParserTestCase(LogSettingsParserTests): ''' 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') @skipIf(NO_MOCK, NO_MOCK_REASON) class SaltAPIParserTestCase(LogSettingsParserTests): ''' 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') @skipIf(not pytest, False) @skipIf(NO_MOCK, NO_MOCK_REASON) 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() # Hide the class from unittest framework when it searches for TestCase classes in the module del LogSettingsParserTests