12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226 |
- # -*- 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.config
- # Import Salt Libs
- import salt.log.setup as log
- import salt.syspaths
- import salt.utils.parsers
- import salt.utils.platform
- from tests.support.mock import MagicMock, patch
- from tests.support.runtests import RUNTIME_VARS
- # Import Salt Testing Libs
- from tests.support.unit import TestCase, skipIf
- 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()
|