diff --git a/tests/test_logging_e2e.py b/tests/test_logging_e2e.py new file mode 100644 index 0000000..e26fc1e --- /dev/null +++ b/tests/test_logging_e2e.py @@ -0,0 +1,73 @@ +import os +import shutil +import pytest +from pathlib import Path +from datetime import datetime, timedelta +from unittest.mock import patch +import session_logger +import tomllib +from log_registry import LogRegistry +from log_pruner import LogPruner + +@pytest.fixture +def e2e_setup(tmp_path): + # Mock _LOG_DIR and _SCRIPTS_DIR in session_logger + original_log_dir = session_logger._LOG_DIR + session_logger._LOG_DIR = tmp_path / "logs" + session_logger._LOG_DIR.mkdir(parents=True, exist_ok=True) + + original_scripts_dir = session_logger._SCRIPTS_DIR + session_logger._SCRIPTS_DIR = tmp_path / "scripts" / "generated" + session_logger._SCRIPTS_DIR.mkdir(parents=True, exist_ok=True) + + yield tmp_path + + # Cleanup + session_logger.close_session() + session_logger._LOG_DIR = original_log_dir + session_logger._SCRIPTS_DIR = original_scripts_dir + +def test_logging_e2e(e2e_setup): + tmp_path = e2e_setup + logs_dir = tmp_path / "logs" + + # Step 1: Initialize (open_session) + session_logger.open_session(label="E2E_Test") + session_id = session_logger._session_id + session_dir = session_logger._session_dir + + # Step 2: Simulate logs (write 'ERROR' to 'comms.log') + # Use log_comms which writes to comms.log + session_logger.log_comms({"level": "ERROR", "message": "Something went wrong"}) + + # Step 3: Shutdown (close_session) + session_logger.close_session() + + # Step 4: Verify 'log_registry.toml' has the session whitelisted due to 'ERROR' + registry_path = logs_dir / "log_registry.toml" + registry = LogRegistry(str(registry_path)) + assert registry.is_session_whitelisted(session_id), "Current session should be whitelisted due to ERROR keyword" + + # Step 5: Simulate an OLD insignificant session in the registry and directory + old_session_id = "20200101_120000_OLD" + old_session_dir = logs_dir / old_session_id + old_session_dir.mkdir() + (old_session_dir / "comms.log").write_text("nothing special") # < 2KB + + old_start_time = datetime.now() - timedelta(days=2) + registry.register_session(old_session_id, str(old_session_dir), old_start_time) + + # Step 6: Trigger 'LogPruner.prune()' + pruner = LogPruner(registry, str(logs_dir)) + pruner.prune() + + # Step 7: Verify the OLD session is deleted but the NEW (whitelisted) session is kept. + assert not old_session_dir.exists(), "Old insignificant session should have been pruned" + assert session_dir.exists(), "New whitelisted session should have been kept" + + # Extra check: Whitelisted sessions should be kept even if old + # Manually backdate the current session + registry.data[session_id]['start_time'] = (datetime.now() - timedelta(days=2)).isoformat() + registry.save_registry() + pruner.prune() + assert session_dir.exists(), "Whitelisted session should be kept even if it is old and small"