From ee318553be610d6de31acb38768d9167d0aaf15e Mon Sep 17 00:00:00 2001 From: Wolfgang Fahl Date: Fri, 4 Oct 2024 13:46:43 +0200 Subject: [PATCH] fixes #135 --- lodstorage/__init__.py | 2 +- lodstorage/persistent_log.py | 116 +++++++++++++++++++++++++++++++++++ tests/test_persistent_log.py | 64 +++++++++++++++++++ 3 files changed, 181 insertions(+), 1 deletion(-) create mode 100644 lodstorage/persistent_log.py create mode 100644 tests/test_persistent_log.py diff --git a/lodstorage/__init__.py b/lodstorage/__init__.py index f23a6b3..7e0dc0e 100644 --- a/lodstorage/__init__.py +++ b/lodstorage/__init__.py @@ -1 +1 @@ -__version__ = "0.13.0" +__version__ = "0.13.1" diff --git a/lodstorage/persistent_log.py b/lodstorage/persistent_log.py new file mode 100644 index 0000000..cfdd704 --- /dev/null +++ b/lodstorage/persistent_log.py @@ -0,0 +1,116 @@ +""" +Created on 2024-10-04 + +@author: wf +""" +from lodstorage.yamlable import lod_storable +import logging +from collections import Counter +from dataclasses import field +from typing import List, Optional, Tuple +from datetime import datetime + +@lod_storable +class LogEntry: + """ + Represents a log entry with a message, kind, and log level name. + """ + msg: str + kind: str + level_name: str + timestamp: Optional[str]=None + + def __post_init__(self): + if self.timestamp is None: + self.timestamp = datetime.now().isoformat() + +@lod_storable +class Log: + """ + Wrapper for persistent logging. + """ + entries: List[LogEntry] = field(default_factory=list) + + def __post_init__(self): + """ + Initializes the log with level mappings and updates the level counts. + """ + self.do_log = True + self.do_print = False + self.levels = { + "❌": logging.ERROR, + "⚠️": logging.WARNING, + "✅": logging.INFO + } + self.level_names = { + logging.ERROR: "error", + logging.WARNING: "warn", + logging.INFO: "info", + } + self.update_level_counts() + + def clear(self): + """ + Clears all log entries. + """ + self.entries = [] + self.update_level_counts() + + def update_level_counts(self): + """ + Updates the counts for each log level based on the existing entries. + """ + self.level_counts = {"error": Counter(), "warn": Counter(), "info": Counter()} + for entry in self.entries: + counter = self.get_counter(entry.level_name) + if counter is not None: + counter[entry.kind] += 1 + + def get_counter(self, level: str) -> Counter: + """ + Returns the counter for the specified log level. + """ + return self.level_counts.get(level) + + def get_level_summary(self, level: str, limit: int = 7) -> Tuple[int, str]: + """ + Get a summary of the most common counts for the specified log level. + + Args: + level (str): The log level name ('error', 'warn', 'info'). + limit (int): The maximum number of most common entries to include in the summary (default is 7). + + Returns: + Tuple[int, str]: A tuple containing the count of log entries and a summary message. + """ + counter = self.get_counter(level) + if counter: + count = sum(counter.values()) + most_common_entries = dict(counter.most_common(limit)) # Get the top 'limit' entries + summary_msg = f"{level.capitalize()} entries: {most_common_entries}" + return count, summary_msg + return 0, f"No entries found for level: {level}" + + + def log(self, icon: str, kind: str, msg: str): + """ + Log a message with the specified icon and kind. + + Args: + icon (str): The icon representing the log level ('❌', '⚠️', '✅'). + kind (str): The category or type of the log message. + msg (str): The log message to record. + """ + level = self.levels.get(icon, logging.INFO) + level_name = self.level_names[level] + icon_msg = f"{icon}:{msg}" + log_entry = LogEntry(msg=icon_msg, level_name=level_name, kind=kind) + self.entries.append(log_entry) + + # Update level counts + self.level_counts[level_name][kind] += 1 + + if self.do_log: + logging.log(level, icon_msg) + if self.do_print: + print(icon_msg) diff --git a/tests/test_persistent_log.py b/tests/test_persistent_log.py new file mode 100644 index 0000000..e4adc94 --- /dev/null +++ b/tests/test_persistent_log.py @@ -0,0 +1,64 @@ +''' +Created on 2024-10-04 + +@author: wf +''' +from tests.basetest import Basetest +from lodstorage.persistent_log import Log + +class TestPersistentLog(Basetest): + """ + test the persistent log handling + """ + + def setUp(self, debug=True, profile=True): + Basetest.setUp(self, debug=debug, profile=profile) + self.log = Log() + if debug: + self.log.do_log=False + self.log.do_print=True + + def test_positive_logging(self): + """Test normal logging and level summary.""" + self.log.log("❌", "system", "An error occurred.") + self.log.log("⚠️", "validation", "A warning message.") + self.log.log("✅", "process", "Process completed successfully.") + + # Verify entries + self.assertEqual(len(self.log.entries), 3) + self.assertEqual(self.log.entries[0].msg, "❌:An error occurred.") + self.assertEqual(self.log.entries[1].msg, "⚠️:A warning message.") + self.assertEqual(self.log.entries[2].msg, "✅:Process completed successfully.") + + # Verify level counts + count, summary = self.log.get_level_summary("error") + self.assertEqual(count, 1) + self.assertIn("system", summary) + + count, summary = self.log.get_level_summary("warn") + self.assertEqual(count, 1) + self.assertIn("validation", summary) + + count, summary = self.log.get_level_summary("info") + self.assertEqual(count, 1) + self.assertIn("process", summary) + + yaml_file="/tmp/persistent_log_test.yaml" + self.log.save_to_yaml_file(yaml_file) + # Later or in another session + loaded_log = Log.load_from_yaml_file(yaml_file) + self.assertEqual(self.log,loaded_log) + + def test_robustness(self): + """Test clearing logs and handling unsupported icons.""" + self.log.log("❌", "system", "First error.") + self.log.log("⭐", "unknown", "Unknown log level.") # Unsupported icon + self.log.clear() + + # Verify entries are cleared + self.assertEqual(len(self.log.entries), 0) + + # Verify level counts are reset + count, summary = self.log.get_level_summary("error") + self.assertEqual(count, 0) + self.assertIn("No entries found", summary)