74 lines
2.8 KiB
Python
74 lines
2.8 KiB
Python
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"
|