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

ovos_utils/log.py

@@ -205,19 +205,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
@@ -342,9 +354,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())
@@ -375,7 +387,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

@@ -60,9 +60,61 @@ 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")
+        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")
+        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"))
+
+    @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 +167,46 @@ 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)
+
+    def test_monitor_log_level(self):
+        from ovos_utils.log import _monitor_log_level
+        # TODO
+
+    def test_get_logs_config(self):
+        from ovos_utils.log import get_logs_config
+
+        # Test original config with `logs` section and no `logging` section
+
+        # Test `logging.logs` config with no service config
+
+        # Test `logging.logs` config with `logging.<service>` overrides
+
+        # Test `logs` config with `logging.<service>` overrides
+
+        # Test `logging.<service>` config with no `logs` or `logging.logs`
+
+    def test_get_log_path(self):
+        from ovos_utils.log import get_log_path
+
+        # Test with multiple populated directories
+
+        # Test with specified empty directory
+
+        # Test path from configuration
+
+    @patch('ovos_utils.log.get_log_path')
+    def test_get_log_paths(self, get_log_path):
+        from ovos_utils.log import get_log_paths
+
+        # Test services with different configured paths
+
+    @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
+
+        # Test with no log directories
+        self.assertEqual(get_available_logs([dirname(__file__)]), [])
+        get_log_paths.return_value = []
+        self.assertEqual(get_available_logs(), [])