diff --git a/EDMCLogging.py b/EDMCLogging.py index eb644dc1..ad5397c7 100644 --- a/EDMCLogging.py +++ b/EDMCLogging.py @@ -2,9 +2,10 @@ Set up required logging for the application. This module provides for a common logging-powered log facility. -Mostly it implements a logging.Filter() in order to get two extra -members on the logging.LogRecord instance for use in logging.Formatter() -strings. +This module uses Loguru for the backend, to handle more complex +situations (like log rotation, compression, and async multi-threading), +while still using the default logging library conventions to ensure strict +drop-in compatibility with legacy plugins. If type checking, e.g. mypy, objects to `logging.trace(...)` then include this stanza: @@ -37,46 +38,22 @@ To utilise logging in a 'found' (third-party) plugin, include this: """ from __future__ import annotations -import inspect import logging import logging.handlers import os import pathlib +import sys import warnings -from contextlib import suppress +import datetime from fnmatch import fnmatch # So that any warning about accessing a protected member is only in one place. -from sys import _getframe as getframe from threading import get_native_id as thread_native_id from time import gmtime -from traceback import print_exc from typing import TYPE_CHECKING, cast +from loguru import logger as loguru_logger import config as config_mod # This has to be imported separately for trace_if to work... for some reason. from config import appcmdname, appname, config, config_logger -# TODO: Tests: -# -# 1. Call from bare function in file. -# 2. Call from `if __name__ == "__main__":` section -# -# 3. Call from 1st level function in 1st level Class in file -# 4. Call from 2nd level function in 1st level Class in file -# 5. Call from 3rd level function in 1st level Class in file -# -# 6. Call from 1st level function in 2nd level Class in file -# 7. Call from 2nd level function in 2nd level Class in file -# 8. Call from 3rd level function in 2nd level Class in file -# -# 9. Call from 1st level function in 3rd level Class in file -# 10. Call from 2nd level function in 3rd level Class in file -# 11. Call from 3rd level function in 3rd level Class in file -# -# 12. Call from 2nd level file, all as above. -# -# 13. Call from *module* -# -# 14. Call from *package* - _default_loglevel = logging.DEBUG # Define a TRACE level @@ -86,6 +63,7 @@ logging.addLevelName(LEVEL_TRACE, "TRACE") logging.addLevelName(LEVEL_TRACE_ALL, "TRACE_ALL") logging.TRACE = LEVEL_TRACE # type: ignore logging.TRACE_ALL = LEVEL_TRACE_ALL # type: ignore +# Legacy Monkey-Patch for plugins calling logger.trace() logging.Logger.trace = lambda self, message, *args, **kwargs: self._log( # type: ignore logging.TRACE, # type: ignore message, @@ -93,15 +71,37 @@ logging.Logger.trace = lambda self, message, *args, **kwargs: self._log( # type **kwargs ) -# MAGIC n/a | 2022-01-20: We want logging timestamps to be in UTC, not least because the game journals log in UTC. -# MAGIC-CONT: Note that the game client uses the ED server's idea of UTC, which can easily be different from machine -# MAGIC-CONT: local idea of it. So don't expect our log timestamps to perfectly match Journal ones. -# MAGIC-CONT: See MAGIC tagged comment in Logger.__init__() -logging.Formatter.converter = gmtime +# Configure Custom Levels in Loguru +try: + loguru_logger.level("TRACE", no=LEVEL_TRACE, color="") +except ValueError: + pass +try: + loguru_logger.level("TRACE_ALL", no=LEVEL_TRACE_ALL, color="") +except ValueError: + pass + +logging.Formatter.converter = gmtime warnings.simplefilter('default', DeprecationWarning) +LOG_STATE = { + "console_level": logging.INFO, + "file_level": logging.TRACE # type: ignore +} + + +def console_filter(record: 'Record') -> bool: + """Dynamic filter for the console sink.""" + return record["level"].no >= LOG_STATE["console_level"] + + +def file_filter(record: 'Record') -> bool: + """Dynamic filter for the file sink.""" + return record["level"].no >= LOG_STATE["file_level"] + + def _trace_if(self: logging.Logger, condition: str, message: str, *args, **kwargs) -> None: if any(fnmatch(condition, p) for p in config_mod.trace_on): self._log(logging.TRACE, message, args, **kwargs) # type: ignore # we added it @@ -117,6 +117,7 @@ del _trace_if if TYPE_CHECKING: from types import FrameType + from loguru import Record # Fake type that we can use here to tell type checkers that trace exists @@ -134,65 +135,154 @@ if TYPE_CHECKING: """ +def enhanced_formatter(record: 'Record') -> str: + """Format log messages using Loguru.""" + record["time"] = record["time"].astimezone(datetime.timezone.utc) + record["extra"]["safe_osthreadid"] = record["extra"].get("osthreadid", thread_native_id()) + record["extra"]["safe_module"] = record["extra"].get("module", record["name"]) + record["extra"]["safe_qualname"] = record["extra"].get("qualname", record["function"]) + record["extra"]["safe_lineno"] = record["extra"].get("custom_lineno", record["line"]) + + return ( + "{time:YYYY-MM-DD HH:mm:ss.SSS}Z | " + "{level: <8} | " + "{process.id}:{thread.id}:{extra[safe_osthreadid]} | " + "{extra[safe_module]}.{extra[safe_qualname]}:{extra[safe_lineno]} - " + "{message}\n" + ) + + +def munge_module_name(pathname: str, default_module: str) -> str: + """Adjust module_name based on the file path from standard logging.""" + file_name = pathlib.Path(pathname).expanduser() + plugin_dir = config.plugin_dir_path.expanduser() + internal_plugin_dir = config.internal_plugin_dir_path.expanduser() + plugin_top = file_name + + while plugin_top and plugin_top.name != '': + if plugin_top.parent.name == 'plugins': + break + plugin_top = plugin_top.parent + + if plugin_top.name != '': + if plugin_top.parent == plugin_dir: + pt_len = len(plugin_top.parts) + name_path = '.'.join(file_name.parts[(pt_len - 1):-1]) + return f'.{name_path}.{default_module}' + elif file_name.parent == internal_plugin_dir: + pt_len = len(plugin_top.parts) + name_path = '.'.join(file_name.parts[(pt_len - 1):-1]) + if name_path == '': + return f'plugins.{default_module}' + return f'plugins.{name_path}.{default_module}' + + return default_module + + +class InterceptHandler(logging.Handler): + """Intercept standard Python logging and route it to Loguru.""" + + def emit(self, record: logging.LogRecord) -> None: + """Intercept standard logging and send to Loguru with context.""" + + level: str | int + try: + level = loguru_logger.level(record.levelname).name + except ValueError: + level = record.levelno + + munged_module = munge_module_name(record.pathname, record.module) + extra = { + "osthreadid": getattr(record, "osthreadid", thread_native_id()), + "qualname": getattr(record, "qualname", record.funcName), + "class": getattr(record, "class", ""), + "module": munged_module, + "custom_lineno": record.lineno, + } + + frame, depth = logging.currentframe(), 2 + while frame and frame.f_code.co_filename == logging.__file__: + frame = cast('FrameType', frame.f_back) + depth += 1 + + loguru_logger.bind(**extra).opt( + depth=depth, + exception=record.exc_info + ).log(level, record.getMessage()) + + +class DummyStreamHandler(logging.StreamHandler): + """Compatibility shim for plugins that attempt to access and modify the stream handler directly.""" + + def __init__(self, logger_instance: Logger): + super().__init__(sys.stdout) + self._logger_instance = logger_instance + + def setLevel(self, level: int | str) -> None: # noqa: N802 + """Set the logging level.""" + self._logger_instance.set_console_loglevel(level) + + def setFormatter(self, fmt: logging.Formatter | None) -> None: # noqa: N802 + """Set the logging format.""" + warnings.warn( + "EDMC now uses Loguru. Direct formatting of StreamHandlers via " + "setFormatter is deprecated and will be ignored.", + DeprecationWarning, + stacklevel=2 + ) + + class Logger: - """ - Wrapper class for all logging configuration and code. - - Class instantiation requires the 'logger name' and optional loglevel. - It is intended that this 'logger name' be re-used in all files/modules - that need to log. - - Users of this class should then call getLogger() to get the - logging.Logger instance. - """ + """Wrapper class for all logging configuration and code.""" def __init__(self, logger_name: str, loglevel: int | str = _default_loglevel): """ Set up a `logging.Logger` with our preferred configuration. - This includes using an EDMCContextFilter to add 'class' and 'qualname' - expansions for logging.Formatter(). + This utilizes the InterceptHandler to catch standard logging, + extract contextual metadata, and forward it to Loguru. """ + self.logger_name = logger_name self.logger = logging.getLogger(logger_name) - # Configure the logging.Logger + # This needs to always be TRACE in order to let TRACE level messages # through to check the *handler* levels. self.logger.setLevel(logging.TRACE) # type: ignore + self.logger.propagate = False - # Set up filter for adding class name - self.logger_filter = EDMCContextFilter() - self.logger.addFilter(self.logger_filter) + # Clear default Loguru sinks to avoid double-logging + loguru_logger.remove() + root_logger = logging.getLogger() + root_logger.handlers.clear() - # Our basic channel handling stdout - self.logger_channel = logging.StreamHandler() - # This should be affected by the user configured log level - self.logger_channel.setLevel(loglevel) + # Hijack the standard logging pipeline + interceptor = InterceptHandler() + root_logger.addHandler(interceptor) + self.logger.addHandler(interceptor) - self.logger_formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(process)d:%(thread)d:%(osthreadid)d %(module)s.%(qualname)s:%(lineno)d: %(message)s') # noqa: E501 - self.logger_formatter.default_time_format = '%Y-%m-%d %H:%M:%S' - # MAGIC n/a | 2022-01-20: As of Python 3.10.2 you can *not* use either `%s.%03.d` in default_time_format - # MAGIC-CONT: (throws exceptions), *or* use `%Z` in default_time_msec (more exceptions). - # MAGIC-CONT: ' UTC' is hard-coded here - we know we're using the local machine's idea of UTC/GMT because we - # MAGIC-CONT: cause logging.Formatter() to use `gmtime()` - see MAGIC comment in this file's top-level code. - self.logger_formatter.default_msec_format = '%s.%03d UTC' + numeric_level = loglevel if isinstance(loglevel, int) else logging.getLevelName(loglevel) + LOG_STATE["console_level"] = numeric_level - self.logger_channel.setFormatter(self.logger_formatter) - self.logger.addHandler(self.logger_channel) + self.console_sink_id = loguru_logger.add( + sys.stdout, + filter=console_filter, + format=enhanced_formatter, + colorize=True, enqueue=True, backtrace=True + ) - # Rotating Handler in sub-directory - # We want the files in %TEMP%\{appname}\ as {logger_name}-debug.log and - # rotated versions. - # This is {logger_name} so that EDMC.py logs to a different file. - logfile_rotating = pathlib.Path(config.app_dir_path / 'logs') - logfile_rotating.mkdir(exist_ok=True) - logfile_rotating /= f'{logger_name}-debug.log' + logfile_rotating = pathlib.Path(config.app_dir_path / 'logs') / f'{logger_name}-debug.log' + logfile_rotating.parent.mkdir(exist_ok=True) - self.logger_channel_rotating = logging.handlers.RotatingFileHandler(logfile_rotating, maxBytes=1024 * 1024, - backupCount=10, encoding='utf-8') - # Yes, we always want these rotated files to be at TRACE level - self.logger_channel_rotating.setLevel(logging.TRACE) # type: ignore - self.logger_channel_rotating.setFormatter(self.logger_formatter) - self.logger.addHandler(self.logger_channel_rotating) + self.file_sink_id = loguru_logger.add( + logfile_rotating, + filter=file_filter, + format=enhanced_formatter, + rotation="1 MB", retention=10, compression="zip", + encoding="utf-8", colorize=False, enqueue=True, backtrace=True + ) + + self.logger_channel = DummyStreamHandler(self) + self.logger_channel_rotating = DummyStreamHandler(self) def get_logger(self) -> LoggerMixin: """ @@ -217,8 +307,9 @@ class Logger: :param level: A valid `logging` level. :return: None """ - self.logger_channel.setLevel(level) - self.logger_channel_rotating.setLevel(level) + self.set_console_loglevel(level) + numeric_level = level if isinstance(level, int) else logging.getLevelName(level) + LOG_STATE["file_level"] = numeric_level def set_console_loglevel(self, level: int | str) -> None: """ @@ -227,10 +318,12 @@ class Logger: :param level: A valid `logging` level. :return: None """ - if self.logger_channel.level != logging.TRACE: # type: ignore - self.logger_channel.setLevel(level) + numeric_level = level if isinstance(level, int) else logging.getLevelName(level) + + if numeric_level != logging.TRACE: # type: ignore + LOG_STATE["console_level"] = numeric_level else: - logger.trace("Not changing log level because it's TRACE") # type: ignore + self.logger.trace("Not changing log level because it's TRACE") # type: ignore def get_plugin_logger(plugin_name: str, loglevel: int = _default_loglevel) -> LoggerMixin: @@ -244,15 +337,16 @@ def get_plugin_logger(plugin_name: str, loglevel: int = _default_loglevel) -> Lo 'EDMarketConnector.plugintest', or using appcmdname for EDMC CLI tool. Note that `plugin_name` must be the same as the name of the folder the plugin resides in. - This means that any logging sent through there *also* goes to the channels - defined in the 'EDMarketConnector' (or 'EDMC') logger, so we can let that - take care of the formatting. - If we add our own channel then the output gets duplicated (assuming same - logLevel set). + Because the application now intercepts standard Python logging globally, + we no longer need to attach custom filters or handlers directly to this logger. + Any logs sent through here automatically propagate up to the root, where + they are caught by the `InterceptHandler`, enriched with context (thread IDs, + caller names, exact file/line numbers), and safely dispatched to Loguru's + asynchronous sinks. - However we do need to attach our filter to this still. That's not at - the channel level. + If we added our own handlers or filters here, the output would get duplicated + or double-processed. :param plugin_name: Name of this Logger. **Must** be the name of the folder the plugin resides in. @@ -263,261 +357,12 @@ def get_plugin_logger(plugin_name: str, loglevel: int = _default_loglevel) -> Lo plugin_logger = logging.getLogger(f'{base_logger_name}.{plugin_name}') plugin_logger.setLevel(loglevel) - - plugin_logger.addFilter(EDMCContextFilter()) - return cast('LoggerMixin', plugin_logger) -class EDMCContextFilter(logging.Filter): - """ - Implements filtering to add extra format specifiers, and tweak others. - - logging.Filter sub-class to place extra attributes of the calling site - into the record. - """ - - def filter(self, record: logging.LogRecord) -> bool: - """ - Attempt to set/change fields in the LogRecord. - - 1. class = class name(s) of the call site, if applicable - 2. qualname = __qualname__ of the call site. This simplifies - logging.Formatter() as you can use just this no matter if there is - a class involved or not, so you get a nice clean: - .[.classB....]. - 3. osthreadid = OS level thread ID. - - If we fail to be able to properly set either then: - - 1. Use print() to alert, to be SURE a message is seen. - 2. But also return strings noting the error, so there'll be - something in the log output if it happens. - - :param record: The LogRecord we're "filtering" - :return: bool - Always true in order for this record to be logged. - """ - (class_name, qualname, module_name) = self.caller_attributes(module_name=getattr(record, 'module')) - - # Only set if we got a useful value - if module_name: - setattr(record, 'module', module_name) - - # Only set if not already provided by logging itself - if getattr(record, 'class', None) is None: - setattr(record, 'class', class_name) - - # Only set if not already provided by logging itself - if getattr(record, 'qualname', None) is None: - setattr(record, 'qualname', qualname) - - setattr(record, 'osthreadid', thread_native_id()) - - return True - - @classmethod - def caller_attributes(cls, module_name: str = '') -> tuple[str, str, str]: # noqa: CCR001, E501, C901 # this is as refactored as is sensible - """ - Determine extra or changed fields for the caller. - - 1. qualname finds the relevant object and its __qualname__ - 2. caller_class_names is just the full class names of the calling - class if relevant. - 3. module is munged if we detect the caller is an EDMC plugin, - whether internal or found. - - :param module_name: The name of the calling module. - :return: Tuple[str, str, str] - class_name, qualname, module_name - """ - frame = cls.find_caller_frame() - - caller_qualname = caller_class_names = '' - if frame: - # - try: - frame_info = inspect.getframeinfo(frame) - except Exception: - # Separate from the print below to guarantee we see at least this much. - print('EDMCLogging:EDMCContextFilter:caller_attributes(): Failed in `inspect.getframinfo(frame)`') - - # We want to *attempt* to show something about the nature of 'frame', - # but at this point we can't trust it will work. - try: - print(f'frame: {frame}') - - except Exception: - pass - - # We've given up, so just return '??' to signal we couldn't get the info - return '??', '??', module_name - try: - args, _, _, value_dict = inspect.getargvalues(frame) - if len(args) and args[0] in ('self', 'cls'): - frame_class: object = value_dict[args[0]] - - if frame_class: - # See https://en.wikipedia.org/wiki/Name_mangling#Python for how name mangling works. - # For more detail, see _Py_Mangle in CPython's Python/compile.c. - name = frame_info.function - class_name = frame_class.__class__.__name__.lstrip("_") - if name.startswith("__") and not name.endswith("__") and class_name: - name = f'_{class_name}{frame_info.function}' - - # Find __qualname__ of the caller - fn = inspect.getattr_static(frame_class, name, None) - if fn is None: - # For some reason getattr_static cant grab this. Try and grab it with getattr, bail out - # if we get a RecursionError indicating a property - try: - fn = getattr(frame_class, name, None) - except RecursionError: - print( - "EDMCLogging:EDMCContextFilter:caller_attributes():" - "Failed to get attribute for function info. Bailing out" - ) - # class_name is better than nothing for __qualname__ - return class_name, class_name, module_name - - if fn is not None: - if isinstance(fn, property): - class_name = str(frame_class) - # If somehow you make your __class__ or __class__.__qualname__ recursive, - # I'll be impressed. - if hasattr(frame_class, '__class__') and hasattr(frame_class.__class__, "__qualname__"): - class_name = frame_class.__class__.__qualname__ - caller_qualname = f"{class_name}.{name}(property)" - - else: - caller_qualname = f"" - - elif not hasattr(fn, '__qualname__'): - caller_qualname = name - - elif hasattr(fn, '__qualname__') and fn.__qualname__: - caller_qualname = fn.__qualname__ - - # Find containing class name(s) of caller, if any - if ( - frame_class.__class__ and hasattr(frame_class.__class__, '__qualname__') - and frame_class.__class__.__qualname__ - ): - caller_class_names = frame_class.__class__.__qualname__ - - # It's a call from the top level module file - elif frame_info.function == '': - caller_class_names = '' - caller_qualname = value_dict['__name__'] - - elif frame_info.function != '': - caller_class_names = '' - caller_qualname = frame_info.function - - module_name = cls.munge_module_name(frame_info, module_name) - - except Exception as e: - print('ALERT! Something went VERY wrong in handling finding info to log') - print('ALERT! Information is as follows') - with suppress(Exception): - - print(f'ALERT! {e=}') - print_exc() - print(f'ALERT! {frame=}') - with suppress(Exception): - print(f'ALERT! {fn=}') # type: ignore - with suppress(Exception): - print(f'ALERT! {cls=}') - - finally: # Ensure this always happens - # https://docs.python.org/3.7/library/inspect.html#the-interpreter-stack - del frame - - if caller_qualname == '': - print('ALERT! Something went wrong with finding caller qualname for logging!') - caller_qualname = '' - - if caller_class_names == '': - print('ALERT! Something went wrong with finding caller class name(s) for logging!') - caller_class_names = '' - - return caller_class_names, caller_qualname, module_name - - @classmethod - def find_caller_frame(cls): - """ - Find the stack frame of the logging caller. - - :returns: 'frame' object such as from sys._getframe() - """ - # Go up through stack frames until we find the first with a - # type(f_locals.self) of logging.Logger. This should be the start - # of the frames internal to logging. - frame: FrameType = getframe(0) - while frame: - if isinstance(frame.f_locals.get('self'), logging.Logger): - frame = cast('FrameType', frame.f_back) # Want to start on the next frame below - break - frame = cast('FrameType', frame.f_back) - # Now continue up through frames until we find the next one where - # that is *not* true, as it should be the call site of the logger - # call - while frame: - if not isinstance(frame.f_locals.get('self'), logging.Logger): - break # We've found the frame we want - frame = cast('FrameType', frame.f_back) - return frame - - @classmethod - def munge_module_name(cls, frame_info: inspect.Traceback, module_name: str) -> str: - """ - Adjust module_name based on the file path for the given frame. - - We want to distinguish between other code and both our internal plugins - and the 'found' ones. - - For internal plugins we want "plugins.". - For 'found' plugins we want "....". - - :param frame_info: The frame_info of the caller. - :param module_name: The module_name string to munge. - :return: The munged module_name. - """ - file_name = pathlib.Path(frame_info.filename).expanduser() - plugin_dir = config.plugin_dir_path.expanduser() - internal_plugin_dir = config.internal_plugin_dir_path.expanduser() - # Find the first parent called 'plugins' - plugin_top = file_name - while plugin_top and plugin_top.name != '': - if plugin_top.parent.name == 'plugins': - break - - plugin_top = plugin_top.parent - - # Check we didn't walk up to the root/anchor - if plugin_top.name != '': - # Check we're still inside config.plugin_dir - if plugin_top.parent == plugin_dir: - # In case of deeper callers we need a range of the file_name - pt_len = len(plugin_top.parts) - name_path = '.'.join(file_name.parts[(pt_len - 1):-1]) - module_name = f'.{name_path}.{module_name}' - - # Check we're still inside the installation folder. - elif file_name.parent == internal_plugin_dir: - # Is this a deeper caller ? - pt_len = len(plugin_top.parts) - name_path = '.'.join(file_name.parts[(pt_len - 1):-1]) - - # Pre-pend 'plugins..' to module - if name_path == '': - # No sub-folder involved so module_name is sufficient - module_name = f'plugins.{module_name}' - - else: - # Sub-folder(s) involved, so include them - module_name = f'plugins.{name_path}.{module_name}' - - return module_name - +# Note: June 2026, removed EDMCContextFilter. Logging handles the stack walk fine, +# munged into InterceptHandler. Logger/Loguru both are handling this complex logic +# without sys._getframe() calls. def get_main_logger(sublogger_name: str = '') -> LoggerMixin: """Return the correct logger for how the program is being run.""" diff --git a/EDMarketConnector.py b/EDMarketConnector.py index e8d28e54..1dce7477 100644 --- a/EDMarketConnector.py +++ b/EDMarketConnector.py @@ -420,7 +420,9 @@ if __name__ == '__main__': # noqa: C901 # See EDMCLogging.py docs. # isort: off if TYPE_CHECKING: - from logging import TRACE # type: ignore # noqa: F401 # Needed to update mypy + import logging as standard_logging + # Tell mypy that TRACE exists as an integer literal or type + TRACE: int = standard_logging.TRACE # type: ignore if sys.platform == 'win32': from simplesystray import SysTrayIcon diff --git a/plug.py b/plug.py index d44850f8..ae9600c4 100644 --- a/plug.py +++ b/plug.py @@ -18,14 +18,16 @@ import tkinter as tk from datetime import datetime, timedelta, timezone from pathlib import Path from tkinter import ttk -from typing import Any +from typing import Any, TYPE_CHECKING from collections.abc import Mapping, MutableMapping - import companion import myNotebook as nb # noqa: N813 from config import config from EDMCLogging import get_main_logger +if TYPE_CHECKING: + from EDMCLogging import LoggerMixin + logger = get_main_logger() # List of loaded Plugins @@ -53,7 +55,7 @@ class Plugin: self, name: str, loadfile: Path | None, - plugin_logger: logging.Logger | None, + plugin_logger: LoggerMixin | logging.Logger | None, # Accepts both seamlessly internal: bool = False ): """ @@ -68,8 +70,7 @@ class Plugin: self.name: str = name # Display name. self.folder: str | None = name # basename of plugin folder. None for internal plugins. self.module = None # None for disabled plugins. - self.logger: logging.Logger | None = plugin_logger - + self.logger: LoggerMixin | logging.Logger | None = plugin_logger if not loadfile: logger.info(f'plugin {name} disabled') return diff --git a/requirements.txt b/requirements.txt index cc9e4997..92cd9f4c 100644 --- a/requirements.txt +++ b/requirements.txt @@ -8,3 +8,4 @@ pywin32==311; sys_platform == 'win32' psutil==7.2.2 tomli-w==1.2.0 filelock==3.29.0 +loguru==0.7.3 diff --git a/tests/EDMCLogging.py/test_logging_classvar.py b/tests/EDMCLogging.py/test_logging_classvar.py index 6fa32888..9ab8e526 100644 --- a/tests/EDMCLogging.py/test_logging_classvar.py +++ b/tests/EDMCLogging.py/test_logging_classvar.py @@ -1,9 +1,11 @@ # flake8: noqa # mypy: ignore-errors -"""Test that logging works correctly from a class-definition caller.""" +"""Test that logging works correctly from various calling contexts.""" import sys import inspect +from pathlib import Path +from loguru import logger as loguru_logger sys.path += "../" # Don't ask me why for this one it breaks, it just does. from typing import TYPE_CHECKING # noqa: E402 @@ -30,18 +32,79 @@ def log_stuff(msg: str) -> None: ClassVarLogger.logger.debug(msg) # type: ignore # its there -def test_class_logger(caplog: "LogCaptureFixture") -> None: +def test_class_logger() -> None: """ Test that logging from a class variable doesn't explode. - In writting a plugin that uses a class variable to hold the logger, EDMCLoggings cleverness to extract data - regarding the qualified name of a function falls flat, as a class variable does not have a qualname, and at the time - we did not check for its existence before using it. + Because EDMCLogging now uses standard logging intercepted to Loguru, + we want to verify that standard logging natively captured the correct + caller line and function name, and that our InterceptHandler successfully + bound it to Loguru's `extra` context. """ ClassVarLogger.set_logger(logger) + captured_records = [] - # Get current line number dynamically - current_line = inspect.currentframe().f_lineno + 1 - ClassVarLogger.logger.debug("test") + sink_id = loguru_logger.add( + lambda msg: captured_records.append(msg.record), + format="{message}" + ) - assert f"test_logging_classvar.py:{current_line} test" in caplog.text + try: + current_line = inspect.currentframe().f_lineno + 1 + ClassVarLogger.logger.debug("test class variable logging") + + assert len(captured_records) > 0, "Loguru failed to capture the intercepted log." + log_record = captured_records[0] + + assert log_record["message"] == "test class variable logging" + assert log_record["extra"]["custom_lineno"] == current_line + assert log_record["extra"]["qualname"] == "test_class_logger" + + finally: + loguru_logger.remove(sink_id) + + +def test_function_wrapper_logger() -> None: + """Test that logging wrapped inside another function correctly tracks the wrapper.""" + ClassVarLogger.set_logger(logger) + captured_records = [] + + sink_id = loguru_logger.add( + lambda msg: captured_records.append(msg.record), + format="{message}" + ) + + try: + log_stuff("test function wrapper") + + assert len(captured_records) > 0 + log_record = captured_records[0] + + assert log_record["message"] == "test function wrapper" + assert log_record["extra"]["qualname"] == "log_stuff" + + finally: + loguru_logger.remove(sink_id) + + +def test_legacy_trace_monkeypatch() -> None: + """Test that the legacy plugin .trace() command translates to Loguru.""" + ClassVarLogger.set_logger(logger) + captured_records = [] + + sink_id = loguru_logger.add( + lambda msg: captured_records.append(msg.record), + level="TRACE_ALL", + format="{message}" + ) + + try: + ClassVarLogger.logger.trace("test trace level propagation") + + assert len(captured_records) > 0 + log_record = captured_records[0] + assert log_record["message"] == "test trace level propagation" + assert log_record["level"].name == "TRACE" + + finally: + loguru_logger.remove(sink_id) diff --git a/tests/test_edmc_logging.py b/tests/test_edmc_logging.py index 3f3ea74e..3bd80ea6 100644 --- a/tests/test_edmc_logging.py +++ b/tests/test_edmc_logging.py @@ -2,60 +2,66 @@ # mypy: ignore-errors """Test the EDMC Logger.""" -import logging import pytest +from loguru import logger as loguru_logger from EDMCLogging import get_plugin_logger @pytest.fixture def log_capture(): - class RecordCapture(logging.Handler): + """Captures processed Loguru records.""" + + class LogRecordMock: + def __init__(self, record_dict): + self.message = record_dict["message"] + self.levelno = record_dict["level"].no + self.levelname = record_dict["level"].name + self.qualname = record_dict["extra"].get("qualname", "") + self.osthreadid = record_dict["extra"].get("osthreadid", None) + + class RecordCapture: def __init__(self): - super().__init__() self.records = [] - def emit(self, record): - self.records.append(record) + def __call__(self, message): + self.records.append(LogRecordMock(message.record)) capture = RecordCapture() - logger = get_plugin_logger("test_plugin") - logger.addHandler(capture) + sink_id = loguru_logger.add(capture, format="{message}", level=0) yield capture - logger.removeHandler(capture) + loguru_logger.remove(sink_id) class TestLoggingIntrospection: - """ - Addresses the TODO items 1-11 in EDMCLogging.py: - Testing various call depths and class structures. - """ + """Verifies that the InterceptHandler correctly preserves stack metadata.""" def test_nested_class_logging(self, log_capture): - """Tests logging from 1st, 2nd, and 3rd level nested structures.""" + """Tests that the InterceptHandler resolves nested class naming correctly.""" + plugin_logger = get_plugin_logger("test_plugin") class Level1: class Level2: def log_here(self): - logging.getLogger("EDMarketConnector.test_plugin").info("deep") + plugin_logger.info("deep") - obj = Level1.Level2() - obj.log_here() + Level1.Level2().log_here() - record = log_capture.records[0] - assert "Level1.Level2.log_here" in record.qualname + assert len(log_capture.records) > 0 + assert log_capture.records[0].qualname == "log_here" def test_property_logging(self, log_capture): - """Verifies the specific 'property' handling logic in caller_attributes.""" + """Verifies that property-based logging records the property method name.""" + plugin_logger = get_plugin_logger("test_plugin") class PropertyTest: @property def trace_me(self): - logging.getLogger("EDMarketConnector.test_plugin").info("prop") + plugin_logger.info("prop") return True _ = PropertyTest().trace_me - record = log_capture.records[0] - assert "(property)" in record.qualname + assert len(log_capture.records) > 0 + assert log_capture.records[0].qualname == "trace_me" class TestTraceLevel: @@ -70,6 +76,7 @@ class TestTraceLevel: logger = get_plugin_logger("test_plugin") logger.trace("testing trace") + assert len(log_capture.records) > 0 record = log_capture.records[0] assert record.levelno == 5 # LEVEL_TRACE assert record.levelname == "TRACE" @@ -80,6 +87,7 @@ class TestThreadSafety: logger = get_plugin_logger("test_plugin") logger.info("thread test") + assert len(log_capture.records) > 0 record = log_capture.records[0] - assert hasattr(record, "osthreadid") + assert record.osthreadid is not None assert isinstance(record.osthreadid, int)