backup

tests/unit/test_log_manager.py

@@ -0,0 +1,209 @@
+"""Unit tests for log manager."""
+
+import tempfile
+from datetime import datetime, timedelta
+from pathlib import Path
+
+import pytest
+
+from src.server.utils.log_manager import LogManager, get_log_manager
+
+
+@pytest.fixture
+def temp_log_env():
+    """Create temporary log environment."""
+    with tempfile.TemporaryDirectory() as tmpdir:
+        yield tmpdir
+
+
+def test_log_manager_initialization(temp_log_env):
+    """Test log manager initialization."""
+    manager = LogManager(log_dir=temp_log_env)
+
+    assert manager is not None
+    assert manager.log_dir.exists()
+    assert manager.archived_dir.exists()
+
+
+def test_get_log_files(temp_log_env):
+    """Test getting list of log files."""
+    manager = LogManager(log_dir=temp_log_env)
+
+    # Create test log files
+    (Path(temp_log_env) / "app.log").write_text("log content 1")
+    (Path(temp_log_env) / "error.log").write_text("log content 2")
+    (Path(temp_log_env) / "other.txt").write_text("not a log")
+
+    log_files = manager.get_log_files()
+
+    assert len(log_files) == 2
+    assert log_files[0].filename in ["app.log", "error.log"]
+
+
+def test_rotate_log(temp_log_env):
+    """Test log file rotation."""
+    manager = LogManager(log_dir=temp_log_env)
+
+    log_file = Path(temp_log_env) / "app.log"
+    large_content = "x" * (11 * 1024 * 1024)  # 11MB
+    log_file.write_text(large_content)
+
+    result = manager.rotate_log("app.log", max_size_bytes=10485760)
+
+    assert result is True
+    assert not log_file.exists()  # Original file rotated
+
+
+def test_rotate_log_not_found(temp_log_env):
+    """Test rotation of non-existent log."""
+    manager = LogManager(log_dir=temp_log_env)
+
+    result = manager.rotate_log("nonexistent.log")
+
+    assert result is False
+
+
+def test_rotate_log_small_file(temp_log_env):
+    """Test rotation of small log file."""
+    manager = LogManager(log_dir=temp_log_env)
+
+    log_file = Path(temp_log_env) / "app.log"
+    log_file.write_text("small content")
+
+    result = manager.rotate_log("app.log", max_size_bytes=1048576)
+
+    assert result is False
+    assert log_file.exists()
+
+
+def test_archive_old_logs(temp_log_env):
+    """Test archiving old log files."""
+    manager = LogManager(log_dir=temp_log_env)
+
+    # Create old and new logs
+    old_log = Path(temp_log_env) / "old.log"
+    old_log.write_text("old log")
+    old_log.touch()
+
+    new_log = Path(temp_log_env) / "new.log"
+    new_log.write_text("new log")
+
+    archived = manager.archive_old_logs(days_old=30)
+
+    assert archived > 0
+
+
+def test_search_logs(temp_log_env):
+    """Test searching logs."""
+    manager = LogManager(log_dir=temp_log_env)
+
+    # Create test logs
+    (Path(temp_log_env) / "app.log").write_text(
+        "Error occurred\nWarning message\nError again"
+    )
+    (Path(temp_log_env) / "debug.log").write_text(
+        "Debug info\nError in debug"
+    )
+
+    results = manager.search_logs("Error", case_sensitive=False)
+
+    assert len(results) >= 1
+    assert any("Error" in line for lines in results.values()
+               for line in lines)
+
+
+def test_search_logs_case_sensitive(temp_log_env):
+    """Test case-sensitive log search."""
+    manager = LogManager(log_dir=temp_log_env)
+
+    (Path(temp_log_env) / "app.log").write_text("ERROR\nerror\nError")
+
+    results = manager.search_logs("ERROR", case_sensitive=True)
+
+    assert "app.log" in results
+    # Should only find uppercase ERROR
+    assert len(results["app.log"]) == 1
+
+
+def test_export_logs(temp_log_env):
+    """Test exporting logs."""
+    manager = LogManager(log_dir=temp_log_env)
+
+    # Create test logs
+    (Path(temp_log_env) / "app.log").write_text("log content 1")
+    (Path(temp_log_env) / "error.log").write_text("log content 2")
+
+    output_file = Path(temp_log_env) / "export.tar.gz"
+    result = manager.export_logs(str(output_file), compress=True)
+
+    assert result is True
+    assert output_file.exists()
+
+
+def test_export_logs_uncompressed(temp_log_env):
+    """Test exporting logs without compression."""
+    manager = LogManager(log_dir=temp_log_env)
+
+    (Path(temp_log_env) / "app.log").write_text("log content")
+
+    output_file = Path(temp_log_env) / "export.txt"
+    result = manager.export_logs(str(output_file), compress=False)
+
+    assert result is True
+    assert output_file.exists()
+    assert "log content" in output_file.read_text()
+
+
+def test_get_log_stats(temp_log_env):
+    """Test getting log statistics."""
+    manager = LogManager(log_dir=temp_log_env)
+
+    # Create test logs
+    (Path(temp_log_env) / "app.log").write_text("x" * 1000)
+    (Path(temp_log_env) / "error.log").write_text("y" * 2000)
+
+    stats = manager.get_log_stats()
+
+    assert stats["total_files"] == 2
+    assert stats["total_size_bytes"] >= 3000
+
+
+def test_get_log_stats_empty(temp_log_env):
+    """Test getting stats with no logs."""
+    manager = LogManager(log_dir=temp_log_env)
+
+    stats = manager.get_log_stats()
+
+    assert stats["total_files"] == 0
+    assert stats["total_size_bytes"] == 0
+
+
+def test_cleanup_logs(temp_log_env):
+    """Test log cleanup."""
+    manager = LogManager(log_dir=temp_log_env)
+
+    # Create multiple logs
+    for i in range(10):
+        (Path(temp_log_env) / f"log_{i}.log").write_text("x" * 1000)
+
+    deleted = manager.cleanup_logs(max_total_size_mb=0.01, keep_files=2)
+
+    assert deleted > 0
+
+
+def test_set_log_level():
+    """Test setting log level."""
+    manager = LogManager()
+
+    result = manager.set_log_level("test_logger", "DEBUG")
+
+    assert result is True
+
+
+def test_get_log_manager_singleton():
+    """Test singleton log manager."""
+    manager1 = get_log_manager()
+    manager2 = get_log_manager()
+
+    assert manager1 is manager2
+    assert isinstance(manager1, LogManager)