Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 17 additions & 2 deletions logpilot/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@
from typing import List, Optional

import typer
from typer.main import get_command

try:
import tomllib
Expand Down Expand Up @@ -40,6 +41,18 @@ def version_callback(value: bool):
app = typer.Typer(no_args_is_help=True)


def _get_analyze_help() -> str:
command = get_command(app)
ctx = typer.Context(command)
get_subcommand = getattr(command, "get_command", None)
if get_subcommand is None:
return ""
analyze_command = get_subcommand(ctx, "analyze")
if analyze_command is None:
return ""
return analyze_command.get_help(typer.Context(analyze_command))


@app.command()
def analyze(
path: str = typer.Argument(..., help="Path to log file or directory"),
Expand All @@ -63,8 +76,10 @@ def analyze(
import os

if path == "-":
typer.echo("Error: stdin is not supported.\n", err=True)
typer.echo(analyze.get_help(typer.Context(analyze)))
typer.echo("Error: stdin is not supported.", err=True)
help_text = _get_analyze_help()
if help_text:
typer.echo(help_text)
raise typer.Exit(1)

# 1. Read log lines
Expand Down
43 changes: 42 additions & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -9,9 +9,50 @@ skip-magic-trailing-comma = false
[tool.ruff]
line-length = 88
indent-width = 1
exclude = [".git", "__pycache__", ".venv", "venv", "env", "build", "dist"]

[tool.ruff.lint]
select = ["E", "F", "W", "I"]
ignore = []
exclude = [".git", "__pycache__", ".venv", "venv", "env", "build", "dist"]

[tool.ruff.format]
indent-style = "tab"

[tool.pytest.ini_options]
testpaths = ["tests"]
python_files = ["test_*.py"]
python_classes = ["Test*"]
python_functions = ["test_*"]
addopts = [
"-v",
"--strict-markers",
"--tb=short",
]
markers = [
"unit: Unit tests that run in isolation",
"integration: Integration tests that test the full CLI",
"slow: Tests that take a long time to run",
]

[tool.coverage.run]
source = ["logpilot"]
branch = true
omit = [
"*/tests/*",
"*/__pycache__/*",
"*/.venv/*",
]

[tool.coverage.report]
show_missing = true
skip_covered = false
fail_under = 80
exclude_lines = [
"pragma: no cover",
"def __repr__",
"if __name__ == .__main__.:",
"raise AssertionError",
"raise NotImplementedError",
"if TYPE_CHECKING:",
"@abstractmethod",
]
2 changes: 2 additions & 0 deletions requirements-dev.txt
Original file line number Diff line number Diff line change
@@ -1,3 +1,5 @@
black
ruff
pytest
pytest-mock
pytest-cov
205 changes: 205 additions & 0 deletions tests/conftest.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,205 @@
"""Shared fixtures for tests."""

import os
from pathlib import Path
from typing import List

import pytest

from logpilot.log_parser import LogEntry

# Try to enable mock LLM if it's not already set
if "LOGPILOT_MOCK_LLM" not in os.environ:
os.environ["LOGPILOT_MOCK_LLM"] = "1"


@pytest.fixture
def sample_log_entries() -> List[LogEntry]:
"""Provides a list of sample LogEntry objects for testing."""
return [
LogEntry(
timestamp="2024-01-01T10:00:00Z",
level="INFO",
source="app.main",
message="Application started",
raw=(
'{"timestamp":"2024-01-01T10:00:00Z",'
'"level":"INFO","source":"app.main",'
'"message":"Application started"}'
),
),
LogEntry(
timestamp="2024-01-01T10:00:01Z",
level="WARNING",
source="app.auth",
message="Failed login attempt",
raw=(
'{"timestamp":"2024-01-01T10:00:01Z",'
'"level":"WARNING","source":"app.auth",'
'"message":"Failed login attempt"}'
),
),
LogEntry(
timestamp="2024-01-01T10:00:02Z",
level="ERROR",
source="app.db",
message="Connection timeout",
raw=(
'{"timestamp":"2024-01-01T10:00:02Z",'
'"level":"ERROR","source":"app.db",'
'"message":"Connection timeout"}'
),
),
LogEntry(
timestamp="2024-01-01T10:00:03Z",
level="DEBUG",
source="app.cache",
message="Cache miss for key: user_123",
raw=(
'{"timestamp":"2024-01-01T10:00:03Z",'
'"level":"DEBUG","source":"app.cache",'
'"message":"Cache miss for key: user_123"}'
),
),
]


@pytest.fixture
def sample_json_logs() -> List[str]:
"""Provides sample JSON-formatted log strings."""
return [
(
'{"timestamp":"2024-01-01T10:00:00Z","level":"INFO",'
'"source":"app.main","message":"Application started"}'
),
(
'{"timestamp":"2024-01-01T10:00:01Z","level":"WARNING",'
'"source":"app.auth","message":"Failed login attempt"}'
),
(
'{"timestamp":"2024-01-01T10:00:02Z","level":"ERROR",'
'"source":"app.db","message":"Connection timeout"}'
),
(
'{"timestamp":"2024-01-01T10:00:03Z","level":"DEBUG",'
'"source":"app.cache","message":"Cache miss"}'
),
]


@pytest.fixture
def sample_text_logs() -> List[str]:
"""Provides sample CSV-style text log strings."""
return [
"2024-01-01T10:00:00Z,INFO,app.main,Application started",
"2024-01-01T10:00:01Z,WARNING,app.auth,Failed login attempt",
"2024-01-01T10:00:02Z,ERROR,app.db,Connection timeout",
"2024-01-01T10:00:03Z,DEBUG,app.cache,Cache miss",
]


@pytest.fixture
def malformed_logs() -> dict:
"""Provides various malformed log strings for edge case testing."""
return {
"empty": "",
"whitespace_only": " \t\n ",
"malformed_json": '{"timestamp":"2024-01-01","level":',
"json_missing_message": '{"timestamp":"2024-01-01","level":"INFO"}',
"text_no_commas": "This is just plain text without commas",
"text_one_comma": "2024-01-01,INFO only two fields",
# Missing message (needs 3+ commas)
"text_two_commas": "2024-01-01,INFO,app.main",
"json_array": '[{"message":"test"}]',
"unicode": "2024-01-01,INFO,app,Message with unicode: 你好世界",
"very_long": "2024-01-01,INFO,app," + "A" * 10000,
}


@pytest.fixture
def temp_log_files(tmp_path: Path):
"""Factory fixture to create temporary log files with given content."""

def _create_log_file(filename: str, content: str, subdirectory: str = None) -> Path:
"""
Create a temporary log file with the given content.

Args:
filename: Name of the file to create
content: Content to write to the file
subdirectory: Optional subdirectory within tmp_path

Returns:
Path to the created file
"""
if subdirectory:
dir_path = tmp_path / subdirectory
dir_path.mkdir(parents=True, exist_ok=True)
file_path = dir_path / filename
else:
file_path = tmp_path / filename

file_path.write_text(content, encoding="utf-8")
return file_path

return _create_log_file


@pytest.fixture
def sample_log_file_structure(tmp_path: Path) -> dict:
"""
Creates a sample directory structure with various log files.

Returns a dict with paths to different files and directories for testing.
"""
# Create directory structure
# tmp_path/
# ├── app.log
# ├── error.log
# ├── logs/
# │ ├── access.log
# │ ├── debug.log
# │ └── nested/
# │ └── deep.log
# └── data.txt (non-log file)

# Root level files
app_log = tmp_path / "app.log"
app_log.write_text(
"2024-01-01,INFO,app,Test log 1\n2024-01-01,INFO,app,Test log 2\n"
)

error_log = tmp_path / "error.log"
error_log.write_text("2024-01-01,ERROR,app,Error occurred\n")

data_txt = tmp_path / "data.txt"
data_txt.write_text("This is not a log file\n")

# Subdirectory
logs_dir = tmp_path / "logs"
logs_dir.mkdir()

access_log = logs_dir / "access.log"
access_log.write_text("2024-01-01,INFO,web,GET /api/users\n")

debug_log = logs_dir / "debug.log"
debug_log.write_text("2024-01-01,DEBUG,app,Debug info\n")

# Nested subdirectory
nested_dir = logs_dir / "nested"
nested_dir.mkdir()

deep_log = nested_dir / "deep.log"
deep_log.write_text("2024-01-01,INFO,deep,Deep nested log\n")

return {
"root": tmp_path,
"app_log": app_log,
"error_log": error_log,
"data_txt": data_txt,
"logs_dir": logs_dir,
"access_log": access_log,
"debug_log": debug_log,
"nested_dir": nested_dir,
"deep_log": deep_log,
}
Loading