log module unit test coverage

ovos_utils/log.py

@@ -29,6 +29,8 @@
                 "ovos",
                 "phal",
                 "phal-admin",
+                "enclosure",
+                "admin",
                 "hivemind",
                 "hivemind-voice-sat"}
 
@@ -205,19 +207,31 @@ 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)
     except ImportError:
         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 +241,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.<service_name>" 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.<service_name>" 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 +343,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 +360,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())
@@ -361,8 +379,8 @@ 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:
+    svc_names = ALL_SERVICES.union({s.replace("-", "_") for s in ALL_SERVICES})
+    for service in svc_names:
         paths.add(get_log_path(service))
 
     return paths
@@ -375,7 +393,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

test/unittests/test_log.py

@@ -2,6 +2,7 @@
 import shutil
 import unittest
 import importlib
+from copy import deepcopy
 
 from os.path import join, dirname, isdir, isfile
 from unittest.mock import patch, Mock
@@ -17,6 +18,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 +62,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 +184,126 @@ 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.<service>` overrides
+        expected_config = {**valid_config_2, **{"level": "WARNING"}}
+        self.assertEqual(get_logs_config("test_service", logging_config),
+                         expected_config)
+
+        # Test `logs` config with `logging.<service>` overrides
+        logging_config["logs"] = logging_config["logging"].pop("logs")
+        self.assertEqual(get_logs_config("test_service", logging_config),
+                         expected_config)
+
+        # Test `logging.<service>` 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_utils.log.get_log_path')
+    def test_get_log_paths(self, get_log_path):
+        from ovos_utils.log import get_log_paths, ALL_SERVICES
+
+        def mock_get_path_different(service) -> str:
+            return service
+
+        def mock_get_path_same(service) -> str:
+            return "log_path"
+
+        # Test services with different configured paths
+        get_log_path.side_effect = mock_get_path_different
+        paths = get_log_paths()
+        for svc in ALL_SERVICES:
+            self.assertIn(svc, paths)
+            if '-' in svc:
+                self.assertIn(svc.replace('-', '_'), paths)
+
+        # Test services with same path
+        get_log_path.side_effect = mock_get_path_same
+        self.assertEqual(get_log_paths(), {"log_path"})
+
+    @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(), [])