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

ovos_utils/log.py

@@ -239,20 +239,23 @@ def get_logs_config(service_name: Optional[str] = None,
             _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

test/unittests/test_log.py

@@ -72,6 +72,8 @@ def test_log(self):
         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)
@@ -82,6 +84,7 @@ def test_log(self):
         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)
@@ -91,6 +94,18 @@ def test_log(self):
         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):
@@ -175,16 +190,48 @@ def test_monitor_log_level(self):
 
     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)
 
     def test_get_log_path(self):
         from ovos_utils.log import get_log_path