1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225 |
- # -*- coding: utf-8 -*-
- """
- :codeauthor: Denys Havrysh <denys.gavrysh@gmail.com>
- """
- from __future__ import absolute_import, print_function, unicode_literals
- import os
- import shutil
- import tempfile
- import pytest
- import salt.config
- import salt.log.setup
- 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
- 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
- @pytest.mark.destructive_test
- @pytest.mark.skip_if_not_root
- 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(
- salt.log.setup,
- 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)
- @pytest.mark.slow_test(seconds=1) # Test takes >0.1 and <=1 seconds
- 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()
|