Russell Ballestrini
7883ae0eba
Phase 1 of module restructuring as outlined in docs/REFACTOR.md: New neopig/ package modules: - live.py: Live media queue (get_live_queue, emit_live_media) - state.py: AppendOnlyStateLog, state file helpers - html_utils.py: trim_html_wrapper, extract_meta_from_html - logging.py: TqdmLoggingHandler, job logging functions - backfill/: markdown and screenshot backfill operations Package features: - Lazy import of NeoPig/main from neopig.py via __getattr__ - Full backwards compatibility with existing imports - 42 new unit tests for extracted modules Total: 458 tests passing
131 lines
3.8 KiB
Python
131 lines
3.8 KiB
Python
"""
|
|
Tests for neopig.logging module.
|
|
|
|
Tests logging utilities and job log capture.
|
|
"""
|
|
|
|
import pytest
|
|
import tempfile
|
|
import shutil
|
|
import os
|
|
import logging
|
|
from pathlib import Path
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
from neopig.logging import (
|
|
TqdmLoggingHandler,
|
|
setup_logging,
|
|
start_job_logging,
|
|
stop_job_logging,
|
|
get_job_logs,
|
|
LOGS_PATH,
|
|
JOB_LOG_HANDLERS,
|
|
)
|
|
|
|
|
|
class TestTqdmLoggingHandler:
|
|
"""Test TqdmLoggingHandler class."""
|
|
|
|
@patch('neopig.logging.tqdm')
|
|
def test_emit_writes_through_tqdm(self, mock_tqdm):
|
|
"""Test that emit writes through tqdm.write."""
|
|
handler = TqdmLoggingHandler()
|
|
handler.setFormatter(logging.Formatter('%(message)s'))
|
|
|
|
record = logging.LogRecord(
|
|
name='test', level=logging.INFO, pathname='', lineno=0,
|
|
msg='Test message', args=(), exc_info=None
|
|
)
|
|
|
|
handler.emit(record)
|
|
mock_tqdm.write.assert_called_once_with('Test message')
|
|
|
|
|
|
class TestSetupLogging:
|
|
"""Test setup_logging function."""
|
|
|
|
def test_configures_root_logger(self):
|
|
"""Test that setup_logging configures root logger."""
|
|
setup_logging(level=logging.DEBUG)
|
|
|
|
root = logging.getLogger()
|
|
assert root.level == logging.DEBUG
|
|
assert len(root.handlers) >= 1
|
|
assert any(isinstance(h, TqdmLoggingHandler) for h in root.handlers)
|
|
|
|
|
|
class TestJobLogging:
|
|
"""Test job-specific logging functions."""
|
|
|
|
def setup_method(self):
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
# Patch LOGS_PATH for tests
|
|
import neopig.logging
|
|
self.original_logs_path = neopig.logging.LOGS_PATH
|
|
neopig.logging.LOGS_PATH = Path(self.temp_dir)
|
|
# Clear handlers
|
|
neopig.logging.JOB_LOG_HANDLERS.clear()
|
|
|
|
def teardown_method(self):
|
|
# Restore and cleanup
|
|
import neopig.logging
|
|
neopig.logging.LOGS_PATH = self.original_logs_path
|
|
neopig.logging.JOB_LOG_HANDLERS.clear()
|
|
if os.path.exists(self.temp_dir):
|
|
shutil.rmtree(self.temp_dir)
|
|
|
|
def test_start_job_logging_creates_file(self):
|
|
"""Test that start_job_logging creates log file."""
|
|
import neopig.logging
|
|
start_job_logging(123)
|
|
|
|
log_file = Path(self.temp_dir) / "123.log"
|
|
assert log_file.exists()
|
|
assert 123 in neopig.logging.JOB_LOG_HANDLERS
|
|
|
|
# Cleanup
|
|
stop_job_logging(123)
|
|
|
|
def test_stop_job_logging_removes_handler(self):
|
|
"""Test that stop_job_logging removes handler."""
|
|
import neopig.logging
|
|
start_job_logging(456)
|
|
assert 456 in neopig.logging.JOB_LOG_HANDLERS
|
|
|
|
stop_job_logging(456)
|
|
assert 456 not in neopig.logging.JOB_LOG_HANDLERS
|
|
|
|
def test_stop_nonexistent_job(self):
|
|
"""Test that stopping non-existent job doesn't raise."""
|
|
stop_job_logging(999) # Should not raise
|
|
|
|
def test_get_job_logs_empty(self):
|
|
"""Test getting logs for non-existent job."""
|
|
logs = get_job_logs(999)
|
|
assert logs == ""
|
|
|
|
def test_get_job_logs_content(self):
|
|
"""Test getting logs content."""
|
|
import neopig.logging
|
|
log_file = Path(self.temp_dir) / "100.log"
|
|
log_file.write_text("Line 1\nLine 2\nLine 3\n")
|
|
|
|
logs = get_job_logs(100)
|
|
assert "Line 1" in logs
|
|
assert "Line 2" in logs
|
|
assert "Line 3" in logs
|
|
|
|
def test_get_job_logs_tail(self):
|
|
"""Test getting last N lines of logs."""
|
|
import neopig.logging
|
|
log_file = Path(self.temp_dir) / "200.log"
|
|
log_file.write_text("Line 1\nLine 2\nLine 3\nLine 4\nLine 5\n")
|
|
|
|
logs = get_job_logs(200, tail=2)
|
|
assert "Line 4" in logs
|
|
assert "Line 5" in logs
|
|
assert "Line 1" not in logs
|
|
|
|
|
|
if __name__ == '__main__':
|
|
pytest.main([__file__, '-v'])
|