"""Tests for utility modules."""
from datetime import datetime
from pathlib import Path
import pytest
from codesdevs_log_analyzer.models import LogLevel, ParsedLogEntry
from codesdevs_log_analyzer.utils.file_handler import (
count_lines,
detect_encoding,
get_file_info,
is_gzip_file,
read_tail,
stream_file,
)
from codesdevs_log_analyzer.utils.formatters import (
format_as_json,
format_as_markdown,
truncate_for_context,
)
from codesdevs_log_analyzer.utils.time_utils import (
extract_timestamp_from_line,
format_timestamp,
parse_relative_time,
parse_timestamp,
time_ago,
)
class TestTimeUtils:
"""Tests for time_utils module."""
def test_parse_iso_timestamp(self) -> None:
"""Test parsing ISO 8601 timestamp."""
result = parse_timestamp("2026-01-15T10:30:00Z")
assert result is not None
assert result.year == 2026
assert result.month == 1
assert result.day == 15
assert result.hour == 10
assert result.minute == 30
def test_parse_iso_with_milliseconds(self) -> None:
"""Test parsing ISO timestamp with milliseconds."""
result = parse_timestamp("2026-01-15T10:30:00.123Z")
assert result is not None
assert result.microsecond > 0
def test_parse_iso_with_timezone(self) -> None:
"""Test parsing ISO timestamp with timezone offset."""
result = parse_timestamp("2026-01-15T10:30:00+05:00")
assert result is not None
assert result.tzinfo is not None
def test_parse_syslog_timestamp(self) -> None:
"""Test parsing syslog format timestamp."""
result = parse_timestamp("Jan 15 10:30:00", default_year=2026)
assert result is not None
assert result.month == 1
assert result.day == 15
def test_parse_apache_timestamp(self) -> None:
"""Test parsing Apache format timestamp."""
result = parse_timestamp("15/Jan/2026:10:30:00 +0000")
assert result is not None
assert result.year == 2026
assert result.month == 1
assert result.day == 15
def test_parse_unix_epoch_seconds(self) -> None:
"""Test parsing Unix epoch timestamp in seconds."""
result = parse_timestamp("1736934600")
assert result is not None
assert result.year == 2025 # Epoch is in past
def test_parse_unix_epoch_milliseconds(self) -> None:
"""Test parsing Unix epoch timestamp in milliseconds."""
result = parse_timestamp("1736934600000")
assert result is not None
def test_parse_invalid_timestamp(self) -> None:
"""Test parsing invalid timestamp returns None."""
result = parse_timestamp("not a timestamp")
assert result is None
def test_parse_empty_timestamp(self) -> None:
"""Test parsing empty string returns None."""
result = parse_timestamp("")
assert result is None
def test_format_timestamp_iso(self) -> None:
"""Test formatting timestamp as ISO."""
dt = datetime(2026, 1, 15, 10, 30, 0)
result = format_timestamp(dt, format_style="iso")
assert "2026-01-15" in result
assert "10:30:00" in result
def test_format_timestamp_human(self) -> None:
"""Test formatting timestamp as human-readable."""
dt = datetime(2026, 1, 15, 10, 30, 0)
result = format_timestamp(dt, format_style="human")
assert result == "2026-01-15 10:30:00"
def test_format_timestamp_none(self) -> None:
"""Test formatting None returns empty string."""
result = format_timestamp(None)
assert result == ""
def test_parse_relative_time_hours(self) -> None:
"""Test parsing relative time in hours."""
ref = datetime(2026, 1, 15, 12, 0, 0)
result = parse_relative_time("2h ago", reference=ref)
assert result is not None
assert result.hour == 10
def test_parse_relative_time_days(self) -> None:
"""Test parsing relative time in days."""
ref = datetime(2026, 1, 15, 12, 0, 0)
result = parse_relative_time("1d ago", reference=ref)
assert result is not None
assert result.day == 14
def test_parse_relative_time_yesterday(self) -> None:
"""Test parsing 'yesterday' keyword."""
ref = datetime(2026, 1, 15, 12, 0, 0)
result = parse_relative_time("yesterday", reference=ref)
assert result is not None
assert result.day == 14
assert result.hour == 0
def test_parse_relative_time_today(self) -> None:
"""Test parsing 'today' keyword."""
ref = datetime(2026, 1, 15, 12, 30, 45)
result = parse_relative_time("today", reference=ref)
assert result is not None
assert result.day == 15
assert result.hour == 0
def test_time_ago_seconds(self) -> None:
"""Test time_ago for seconds."""
ref = datetime(2026, 1, 15, 10, 30, 30)
dt = datetime(2026, 1, 15, 10, 30, 0)
result = time_ago(dt, reference=ref)
assert "30 second" in result
def test_time_ago_minutes(self) -> None:
"""Test time_ago for minutes."""
ref = datetime(2026, 1, 15, 10, 35, 0)
dt = datetime(2026, 1, 15, 10, 30, 0)
result = time_ago(dt, reference=ref)
assert "5 minute" in result
def test_extract_timestamp_from_line(self) -> None:
"""Test extracting timestamp from log line."""
line = "2026-01-15T10:30:00Z INFO Application started"
result = extract_timestamp_from_line(line)
assert result is not None
assert result.year == 2026
class TestFileHandler:
"""Tests for file_handler module."""
def test_stream_file(self, temp_log_file: Path) -> None:
"""Test streaming file lines."""
lines = list(stream_file(str(temp_log_file)))
assert len(lines) == 3
assert lines[0][0] == 1 # Line number
assert "Test message 1" in lines[0][1]
def test_stream_file_max_lines(self, temp_log_file: Path) -> None:
"""Test streaming with max_lines limit."""
lines = list(stream_file(str(temp_log_file), max_lines=2))
assert len(lines) == 2
def test_stream_file_not_found(self) -> None:
"""Test streaming nonexistent file raises error."""
with pytest.raises(FileNotFoundError):
list(stream_file("/nonexistent/file.log"))
def test_read_tail(self, temp_log_file: Path) -> None:
"""Test reading last N lines."""
lines = read_tail(str(temp_log_file), n_lines=2)
assert len(lines) == 2
assert "error" in lines[0][1].lower() or "debug" in lines[0][1].lower()
def test_read_tail_more_than_file(self, temp_log_file: Path) -> None:
"""Test reading more lines than file contains."""
lines = read_tail(str(temp_log_file), n_lines=100)
assert len(lines) == 3
def test_detect_encoding_utf8(self, temp_log_file: Path) -> None:
"""Test encoding detection for UTF-8 file."""
encoding = detect_encoding(str(temp_log_file))
assert encoding in ("utf-8", "ascii")
def test_is_gzip_file_regular(self, temp_log_file: Path) -> None:
"""Test gzip detection for regular file."""
assert is_gzip_file(str(temp_log_file)) is False
def test_is_gzip_file_compressed(self, gzip_temp_file: Path) -> None:
"""Test gzip detection for compressed file."""
assert is_gzip_file(str(gzip_temp_file)) is True
def test_stream_gzip_file(self, gzip_temp_file: Path) -> None:
"""Test streaming gzip compressed file."""
lines = list(stream_file(str(gzip_temp_file)))
assert len(lines) == 2
assert "Compressed" in lines[0][1]
def test_count_lines(self, temp_log_file: Path) -> None:
"""Test counting lines in file."""
count = count_lines(str(temp_log_file))
assert count == 3
def test_get_file_info(self, temp_log_file: Path) -> None:
"""Test getting file information."""
info = get_file_info(str(temp_log_file))
assert info["path"] == str(temp_log_file)
assert info["size_bytes"] > 0
assert "encoding" in info
assert info["is_compressed"] is False
class TestFormatters:
"""Tests for formatters module."""
def test_format_as_json_dict(self) -> None:
"""Test formatting dict as JSON."""
data = {"key": "value", "number": 42}
result = format_as_json(data)
assert '"key": "value"' in result
assert '"number": 42' in result
def test_format_as_json_with_datetime(self) -> None:
"""Test formatting dict with datetime as JSON."""
data = {"timestamp": datetime(2026, 1, 15, 10, 30, 0)}
result = format_as_json(data)
assert "2026-01-15" in result
def test_format_as_markdown_entry(self) -> None:
"""Test formatting ParsedLogEntry as markdown."""
_entry = ParsedLogEntry( # noqa: F841
line_number=1,
raw_line="2026-01-15 INFO Test",
message="Test message",
level=LogLevel.INFO,
timestamp=datetime(2026, 1, 15, 10, 30, 0),
metadata={},
)
# format_as_markdown expects specific result types
# Test with a dict instead
result = format_as_markdown({"entry": "test"})
assert "Entry" in result
def test_truncate_for_context_short(self) -> None:
"""Test truncation of short content."""
content = "Short content"
result, was_truncated = truncate_for_context(content, max_chars=1000)
assert result == content
assert was_truncated is False
def test_truncate_for_context_long(self) -> None:
"""Test truncation of long content."""
content = "x" * 1000
result, was_truncated = truncate_for_context(content, max_chars=100)
assert len(result) < len(content)
assert was_truncated is True
assert "truncated" in result.lower()
class TestIntegration:
"""Integration tests across utility modules."""
def test_parse_and_format_timestamp(self) -> None:
"""Test parsing and reformatting timestamp."""
original = "2026-01-15T10:30:00Z"
parsed = parse_timestamp(original)
assert parsed is not None
formatted = format_timestamp(parsed, format_style="human")
assert "2026-01-15" in formatted
assert "10:30:00" in formatted
def test_stream_and_count(self, temp_log_file: Path) -> None:
"""Test streaming matches count."""
stream_count = sum(1 for _ in stream_file(str(temp_log_file)))
direct_count = count_lines(str(temp_log_file))
assert stream_count == direct_count
