From e4a3ff7e16c4c0843e537947a03c6f6614304732 Mon Sep 17 00:00:00 2001 From: Daniel McKnight <34697904+NeonDaniel@users.noreply.github.com> Date: Mon, 2 Sep 2024 08:20:59 -0700 Subject: [PATCH] log module unit test coverage (#260) * Add docstrings and type annotations to `log` module Add test coverage for log rotation Outline test coverage for added log module functions Relates to #2 #239 #233 #250 #253 * Update logs to resolve order-related failure * Add test coverage for `get_logs_config` Refactor variable in `get_logs_config` for clarity Refactor `get_logs_config` for more predictable handling of an empty service name * Add test coverage for `get_log_path` * Add test coverage for `get_log_paths` Update `get_log_paths` to check all log names with `-` normalized to `_` (appeared to be the original intent) Add `enclosure` and `admin` service logs used by Neon and legacy Mycroft/OVOS setups * Add test coverage for `get_available_logs` * Add test coverage for `_monitor_log_level` * Add test of `get_config` call count * Update `get_log_paths` to reference config directly instead of test some well-known values Update unit test for `get_log_paths` * Remove unused `ALL_SERVICES` variable * Remove unused import --------- Co-authored-by: Daniel McKnight --- ovos_utils/log.py | 72 ++++++----- test/unittests/test_log.py | 194 +++++++++++++++++++++++++++++- test/unittests/test_logs/real.log | 0 3 files changed, 235 insertions(+), 31 deletions(-) create mode 100644 test/unittests/test_logs/real.log diff --git a/ovos_utils/log.py b/ovos_utils/log.py index 70cbe7ca..6f4c69a0 100644 --- a/ovos_utils/log.py +++ b/ovos_utils/log.py @@ -21,17 +21,6 @@ from pathlib import Path from typing import Optional, List, Set -ALL_SERVICES = {"bus", - "audio", - "skills", - "voice", - "gui", - "ovos", - "phal", - "phal-admin", - "hivemind", - "hivemind-voice-sat"} - class LOG: """ @@ -205,11 +194,16 @@ def _monitor_log_level(): _monitor_log_level.config_hash = None -def init_service_logger(service_name): +def init_service_logger(service_name: str): + """ + Initialize `LOG` for the specified service + @param service_name: Name of service to configure `LOG` for + """ _logs_conf = get_logs_config(service_name) - _monitor_log_level.config_hash = hash(json.dumps(_logs_conf, sort_keys=True, indent=2)) + _monitor_log_level.config_hash = hash(json.dumps(_logs_conf, sort_keys=True, + indent=2)) LOG.name = service_name - LOG.init(_logs_conf) # setup the LOG instance + LOG.init(_logs_conf) # set up the LOG instance try: from ovos_config import Configuration Configuration.set_config_watcher(_monitor_log_level) @@ -217,7 +211,14 @@ def init_service_logger(service_name): LOG.warning("Can not monitor config LOG level changes") -def get_logs_config(service_name=None, _cfg=None) -> dict: +def get_logs_config(service_name: Optional[str] = None, + _cfg: Optional[dict] = None) -> dict: + """ + Get logging configuration for the specified service + @param service_name: Name of service to get logging configuration for + @param _cfg: Configuration to parse + @return: dict logging configuration for the specified service + """ if _cfg is None: try: from ovos_config import Configuration @@ -227,20 +228,23 @@ def get_logs_config(service_name=None, _cfg=None) -> dict: _cfg = {} # First try and get the "logging" section - log_config = _cfg.get("logging") + logging_conf = _cfg.get("logging") # For compatibility we try to get the "logs" from the root level # and default to empty which is used in case there is no logging # section _logs_conf = _cfg.get("logs") or {} - if log_config: # We found a logging section + if logging_conf: # We found a logging section # if "logs" is defined in "logging" use that as the default # where per-service "logs" are not defined - _logs_conf = log_config.get("logs") or _logs_conf + _logs_conf = logging_conf.get("logs") or _logs_conf # Now get our config by service name if service_name: - _cfg = log_config.get(service_name) or log_config - # and if "logs" is redefined in "logging." use that - _logs_conf = _cfg.get("logs") or _logs_conf + _cfg = logging_conf.get(service_name) or logging_conf + else: + # No service name specified, use `logging` configuration + _cfg = logging_conf + # and if "logs" is redefined in "logging." use that + _logs_conf = _cfg.get("logs") or _logs_conf # Grab the log level from whatever section we found, defaulting to INFO _log_level = _cfg.get("log_level", "INFO") _logs_conf["level"] = _log_level @@ -326,6 +330,7 @@ def get_log_path(service: str, directories: Optional[List[str]] = None) \ Returns: path to log directory for service + (returned path may be `None` if `directories` is specified) """ if directories: for directory in directories: @@ -342,9 +347,9 @@ def get_log_path(service: str, directories: Optional[List[str]] = None) \ xdg_base = os.environ.get("OVOS_CONFIG_BASE_FOLDER", "mycroft") return os.path.join(xdg_state_home(), xdg_base) - config = Configuration().get("logging", dict()).get("logs", dict()) + config = get_logs_config(service_name=service) # service specific config or default config location - path = config.get(service, {}).get("path") or config.get("path") + path = config.get("path") # default xdg location if not path: path = os.path.join(xdg_state_home(), get_xdg_base()) @@ -352,7 +357,7 @@ def get_log_path(service: str, directories: Optional[List[str]] = None) \ return path -def get_log_paths() -> Set[str]: +def get_log_paths(config: Optional[dict] = None) -> Set[str]: """ Get all log paths for all service logs Different services may have different log paths @@ -361,9 +366,20 @@ def get_log_paths() -> Set[str]: set of paths to log directories """ paths = set() - ALL_SERVICES.union({s.replace("-", "_") for s in ALL_SERVICES}) - for service in ALL_SERVICES: - paths.add(get_log_path(service)) + if not config: + try: + from ovos_config import Configuration + config = Configuration() + except ImportError: + LOG.warning("ovos_config not available. Falling back to defaults") + config = dict() + + for name, service_config in config.get("logging", {}).items(): + if not isinstance(service_config, dict) or name == "logs": + continue + if service_config.get("path"): + paths.add(service_config.get("path")) + paths.add(get_log_path("")) return paths @@ -375,7 +391,7 @@ def get_available_logs(directories: Optional[List[str]] = None) -> List[str]: directories: (optional) list of directories to check for service Returns: - list of log files + list of log file basenames (i.e. "audio", "skills") """ directories = directories or get_log_paths() return [Path(f).stem for path in directories diff --git a/test/unittests/test_log.py b/test/unittests/test_log.py index 751f96ae..f816cf1b 100644 --- a/test/unittests/test_log.py +++ b/test/unittests/test_log.py @@ -17,6 +17,7 @@ def tearDownClass(cls) -> None: def test_log(self): import ovos_utils.log + importlib.reload(ovos_utils.log) from ovos_utils.log import LOG # Default log config self.assertEqual(LOG.base_path, "stdout") @@ -60,9 +61,76 @@ def test_log(self): self.assertEqual(len(lines), 1) self.assertTrue(lines[0].endswith("This will print\n")) - def test_init_service_logger(self): - from ovos_utils.log import init_service_logger - # TODO + # Init with backup + test_config['max_bytes'] = 2 + test_config['backup_count'] = 1 + test_config['level'] = 'INFO' + LOG.init(test_config) + LOG.name = "rotate" + LOG.info("first") + LOG.info("second") + LOG.debug("third") + log_1 = join(LOG.base_path, f"{LOG.name}.log.1") + log = join(LOG.base_path, f"{LOG.name}.log") + + # Log rotated once + with open(log_1) as f: + lines = f.readlines() + self.assertEqual(len(lines), 1) + self.assertTrue(lines[0].endswith("first\n")) + with open(log) as f: + lines = f.readlines() + self.assertEqual(len(lines), 1) + self.assertTrue(lines[0].endswith("second\n")) + + LOG.info("fourth") + # Log rotated again + with open(log_1) as f: + lines = f.readlines() + self.assertEqual(len(lines), 1) + self.assertTrue(lines[0].endswith("second\n")) + with open(log) as f: + lines = f.readlines() + self.assertEqual(len(lines), 1) + self.assertTrue(lines[0].endswith("fourth\n")) + + # Multiple log rotations within a short period of time + for i in range(100): + LOG.info(str(i)) + with open(log_1) as f: + lines = f.readlines() + self.assertEqual(len(lines), 1) + self.assertTrue(lines[0].endswith("98\n")) + with open(log) as f: + lines = f.readlines() + self.assertEqual(len(lines), 1) + self.assertTrue(lines[0].endswith("99\n")) + + @patch("ovos_utils.log.get_logs_config") + @patch("ovos_config.Configuration.set_config_watcher") + def test_init_service_logger(self, set_config_watcher, log_config): + from ovos_utils.log import init_service_logger, LOG + + # Test log init with default config + log_config.return_value = dict() + LOG.level = "ERROR" + init_service_logger("default") + from ovos_utils.log import LOG + set_config_watcher.assert_called_once() + self.assertEqual(LOG.name, "default") + self.assertEqual(LOG.level, "ERROR") + + # Test log init with config + set_config_watcher.reset_mock() + log_config.return_value = {"path": self.test_dir, + "level": "DEBUG"} + init_service_logger("configured") + from ovos_utils.log import LOG + set_config_watcher.assert_called_once() + self.assertEqual(LOG.name, "configured") + self.assertEqual(LOG.level, "DEBUG") + LOG.debug("This will print") + self.assertTrue(isfile(join(self.test_dir, "configured.log"))) @patch("ovos_utils.log.LOG.create_logger") def test_log_deprecation(self, create_logger): @@ -115,3 +183,123 @@ def _deprecated_function(test_arg): log_msg = log_warning.call_args[0][0] self.assertIn('version=1.0.0', log_msg, log_msg) self.assertIn('test deprecation', log_msg, log_msg) + + @patch("ovos_utils.log.get_logs_config") + @patch("ovos_utils.log.LOG") + def test_monitor_log_level(self, log, get_config): + from ovos_utils.log import _monitor_log_level + + log.name = "TEST" + get_config.return_value = {"changed": False} + + _monitor_log_level() + get_config.assert_called_once_with("TEST") + log.init.assert_called_once_with(get_config.return_value) + log.info.assert_called_once() + + # Callback with no change + _monitor_log_level() + self.assertEqual(get_config.call_count, 2) + log.init.assert_called_once_with(get_config.return_value) + log.info.assert_called_once() + + # Callback with change + get_config.return_value["changed"] = True + _monitor_log_level() + self.assertEqual(get_config.call_count, 3) + self.assertEqual(log.init.call_count, 2) + log.init.assert_called_with(get_config.return_value) + + def test_get_logs_config(self): + from ovos_utils.log import get_logs_config + valid_config = {"level": "DEBUG", + "path": self.test_dir, + "max_bytes": 1000, + "backup_count": 2, + "diagnostic": False} + valid_config_2 = {"max_bytes": 100000, + "diagnostic": True} + logs_config = {"path": self.test_dir, + "max_bytes": 1000, + "backup_count": 2, + "diagnostic": False} + legacy_config = {"log_level": "DEBUG", + "logs": logs_config} + + logging_config = {"logging": {"log_level": "DEBUG", + "logs": logs_config, + "test_service": {"log_level": "WARNING", + "logs": valid_config_2} + } + } + + # Test original config with `logs` section and no `logging` section + self.assertEqual(get_logs_config("", legacy_config), valid_config) + + # Test `logging.logs` config with no service config + self.assertEqual(get_logs_config("service", logging_config), valid_config) + + # Test `logging.logs` config with `logging.` overrides + expected_config = {**valid_config_2, **{"level": "WARNING"}} + self.assertEqual(get_logs_config("test_service", logging_config), + expected_config) + + # Test `logs` config with `logging.` overrides + logging_config["logs"] = logging_config["logging"].pop("logs") + self.assertEqual(get_logs_config("test_service", logging_config), + expected_config) + + # Test `logging.` config with no `logs` or `logging.logs` + logging_config["logging"].pop("log_level") + logging_config.pop("logs") + self.assertEqual(get_logs_config("test_service", logging_config), + expected_config) + + @patch("ovos_utils.log.get_logs_config") + def test_get_log_path(self, get_config): + from ovos_utils.log import get_log_path + + real_log_path = join(dirname(__file__), "test_logs") + test_paths = [self.test_dir, dirname(__file__), real_log_path] + + # Test with multiple populated directories + self.assertEqual(get_log_path("real", test_paths), real_log_path) + self.assertIsNone(get_log_path("fake", test_paths)) + + # Test path from configuration + get_config.return_value = {"path": self.test_dir} + self.assertEqual(get_log_path("test"), self.test_dir) + get_config.assert_called_once_with(service_name="test") + + @patch('ovos_config.Configuration') + def test_get_log_paths(self, config): + from ovos_utils.log import get_log_paths + + config_no_modules = {"logging": {"logs": {"path": "default_path"}}} + + # Test default config path from Configuration (no module overrides) + config.return_value = config_no_modules + self.assertEqual(get_log_paths(), {"default_path"}) + + # Test services with different configured paths + config_multi_modules = {"logging": {"logs": {"path": "default_path"}, + "module_1": {"path": "path_1"}, + "module_2": {"path": "path_2"}, + "module_3": {"path": "path_1"}}} + self.assertEqual(get_log_paths(config_multi_modules), + {"default_path", "path_1", "path_2"}) + + + @patch('ovos_utils.log.get_log_paths') + def test_get_available_logs(self, get_log_paths): + from ovos_utils.log import get_available_logs + + # Test with specified directories containing logs and other files + real_log_path = join(dirname(__file__), "test_logs") + get_log_paths.return_value = [dirname(__file__), real_log_path] + self.assertEqual(get_available_logs(), ["real"]) + + # Test with no log directories + self.assertEqual(get_available_logs([dirname(__file__)]), []) + get_log_paths.return_value = [] + self.assertEqual(get_available_logs(), []) diff --git a/test/unittests/test_logs/real.log b/test/unittests/test_logs/real.log new file mode 100644 index 00000000..e69de29b