Skip to content

Commit

Permalink
log module unit test coverage (#260)
Browse files Browse the repository at this point in the history 
* 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 <[email protected]>
  • Loading branch information
@NeonDaniel
NeonDaniel and NeonDaniel authored Sep 2, 2024
1 parent d0d5587 commit e4a3ff7
Show file tree
Hide file tree
Showing 3 changed files with 235 additions and 31 deletions.
72 changes: 44 additions & 28 deletions ovos_utils/log.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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:
"""
Expand Down Expand Up @@ -205,19 +194,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
Expand All @@ -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.<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
Expand Down Expand Up @@ -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:
Expand All @@ -342,17 +347,17 @@ 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())

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
Expand All @@ -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

Expand All @@ -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
Expand Down
194 changes: 191 additions & 3 deletions test/unittests/test_log.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down Expand Up @@ -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):
Expand Down Expand Up @@ -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.<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_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(), [])
Empty file added test/unittests/test_logs/real.log
View file
Empty file.

0 comments on commit e4a3ff7

Please sign in to comment.