feat(logging): Implement LogPruner for cleaning up old insignificant logs

log_pruner.py

@@ -0,0 +1,60 @@
+import os
+import shutil
+from datetime import datetime, timedelta
+from log_registry import LogRegistry
+
+class LogPruner:
+    def __init__(self, log_registry: LogRegistry, logs_dir: str):
+        """
+        Initializes the LogPruner.
+
+        Args:
+            log_registry: An instance of LogRegistry to check session data.
+            logs_dir: The path to the directory containing session sub-directories.
+        """
+        self.log_registry = log_registry
+        self.logs_dir = logs_dir
+
+    def prune(self):
+        """
+        Prunes old and small session directories from the logs directory.
+
+        Deletes session directories that meet the following criteria:
+        1. The session start time is older than 24 hours (based on data from LogRegistry).
+        2. The session name is NOT in the whitelist provided by the LogRegistry.
+        3. The total size of all files within the session directory is less than 2KB (2048 bytes).
+        """
+        now = datetime.now()
+        cutoff_time = now - timedelta(hours=24)
+
+        # Ensure the base logs directory exists.
+        if not os.path.isdir(self.logs_dir):
+            return
+
+        # Get sessions that are old and not whitelisted from the registry
+        old_sessions_to_check = self.log_registry.get_old_non_whitelisted_sessions(cutoff_time)
+
+        # Prune sessions if their size is less than 2048 bytes
+        for session_info in old_sessions_to_check:
+            session_id = session_info['session_id']
+            session_path = session_info['path']
+            
+            if not session_path or not os.path.isdir(session_path):
+                continue
+
+            # Calculate total size of files in the directory
+            total_size = 0
+            try:
+                for entry in os.scandir(session_path):
+                    if entry.is_file():
+                        total_size += entry.stat().st_size
+            except OSError:
+                continue
+
+            # Prune if the total size is less than 2KB (2048 bytes)
+            if total_size < 2048: # 2KB
+                try:
+                    shutil.rmtree(session_path)
+                    # print(f"Pruned session '{session_id}' (Size: {total_size} bytes)")
+                except OSError:
+                    pass

tests/test_log_pruner.py

@@ -0,0 +1,55 @@
+import os
+import shutil
+import pytest
+from pathlib import Path
+from datetime import datetime, timedelta
+from log_registry import LogRegistry
+from log_pruner import LogPruner
+
+@pytest.fixture
+def pruner_setup(tmp_path):
+    logs_dir = tmp_path / "logs"
+    logs_dir.mkdir()
+    registry_path = logs_dir / "log_registry.toml"
+    registry = LogRegistry(str(registry_path))
+    pruner = LogPruner(registry, str(logs_dir))
+    return pruner, registry, logs_dir
+
+def test_prune_old_insignificant_logs(pruner_setup):
+    pruner, registry, logs_dir = pruner_setup
+    
+    # 1. Old and small (insignificant) -> should be pruned
+    session_id_old_small = "old_small"
+    dir_old_small = logs_dir / session_id_old_small
+    dir_old_small.mkdir()
+    (dir_old_small / "comms.log").write_text("small") # < 2KB
+    registry.register_session(session_id_old_small, str(dir_old_small), datetime.now() - timedelta(days=2))
+    
+    # 2. Old and large (significant) -> should NOT be pruned
+    session_id_old_large = "old_large"
+    dir_old_large = logs_dir / session_id_old_large
+    dir_old_large.mkdir()
+    (dir_old_large / "comms.log").write_text("x" * 3000) # > 2KB
+    registry.register_session(session_id_old_large, str(dir_old_large), datetime.now() - timedelta(days=2))
+    
+    # 3. Recent and small -> should NOT be pruned
+    session_id_recent_small = "recent_small"
+    dir_recent_small = logs_dir / session_id_recent_small
+    dir_recent_small.mkdir()
+    (dir_recent_small / "comms.log").write_text("small")
+    registry.register_session(session_id_recent_small, str(dir_recent_small), datetime.now() - timedelta(hours=2))
+    
+    # 4. Old and whitelisted -> should NOT be pruned
+    session_id_old_whitelisted = "old_whitelisted"
+    dir_old_whitelisted = logs_dir / session_id_old_whitelisted
+    dir_old_whitelisted.mkdir()
+    (dir_old_whitelisted / "comms.log").write_text("small")
+    registry.register_session(session_id_old_whitelisted, str(dir_old_whitelisted), datetime.now() - timedelta(days=2))
+    registry.update_session_metadata(session_id_old_whitelisted, 0, 0, 0, True, "Manual")
+    
+    pruner.prune()
+    
+    assert not dir_old_small.exists()
+    assert dir_old_large.exists()
+    assert dir_recent_small.exists()
+    assert dir_old_whitelisted.exists()