From c605454196fd0c88460a0489fda0834bb37e22b8 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 6 Feb 2026 22:45:59 +0000 Subject: [PATCH 1/4] Initial plan From 8cdbc471aa9b8cf58e12cbca3ef470bf357a6572 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 6 Feb 2026 22:48:44 +0000 Subject: [PATCH 2/4] Add --model CLI option for model selection Co-authored-by: colinmakerofthings <66311935+colinmakerofthings@users.noreply.github.com> --- README.md | 7 +++++++ logpilot/cli.py | 3 ++- tests/test_cli.py | 30 ++++++++++++++++++++++++++++++ 3 files changed, 39 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index b388285..6d84a6d 100644 --- a/README.md +++ b/README.md @@ -28,6 +28,7 @@ python -m logpilot.cli analyze [OPTIONS] - `--recursive` : Recurse into subdirectories (directory mode) - `--include ` : Glob pattern(s) to include (repeatable) - `--exclude ` : Glob pattern(s) to exclude (repeatable) +- `--model ` : LLM model to use (default: gpt-4) ### Example @@ -55,6 +56,12 @@ Summarize a log file and write to a file: python -m logpilot.cli analyze mylog.txt --out-file summary.txt ``` +Analyze logs with a different model: + +```sh +python -m logpilot.cli analyze mylog.txt --model gpt-3.5-turbo +``` + ## Project Structure - `logpilot/` - Core modules diff --git a/logpilot/cli.py b/logpilot/cli.py index 5b05ddf..6157235 100644 --- a/logpilot/cli.py +++ b/logpilot/cli.py @@ -58,6 +58,7 @@ def analyze( "--exclude", help="Glob pattern(s) to exclude (repeatable).", ), + model: str = typer.Option("gpt-4", help="LLM model to use (default: gpt-4)"), ): """Analyze logs using Copilot SDK.""" import os @@ -109,7 +110,7 @@ def analyze( if os.environ.get("LOGPILOT_MOCK_LLM") == "1": responses = ["Mocked summary: Something failed" for _ in prompts] else: - llm = LLMClient() + llm = LLMClient(model=model) responses = [llm.analyze(prompt) for prompt in prompts] # 6. Aggregate responses diff --git a/tests/test_cli.py b/tests/test_cli.py index eb9de94..cc8ef5d 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -204,3 +204,33 @@ def test_cli_directory_no_matches(tmp_path): result = run_cli(["analyze", str(log_dir), "--include", "*.log"]) assert result.returncode != 0 assert "No log files matched" in result.stderr or "Error" in result.stderr + + +def test_cli_model_option(tmp_path): + """Test that the --model option is accepted and works""" + log_file = tmp_path / "test.log" + log_file.write_text( + ( + '{"timestamp": "2026-01-28T12:00:00Z", ' + '"level": "ERROR", ' + '"message": "Something failed"}\n' + ) + ) + result = run_cli(["analyze", str(log_file), "--model", "gpt-3.5-turbo"]) + assert result.returncode == 0, f"stdout: {result.stdout}\nstderr: {result.stderr}" + assert "Mocked summary" in result.stdout + + +def test_cli_model_option_default(tmp_path): + """Test that the default model (gpt-4) still works without specifying --model""" + log_file = tmp_path / "test.log" + log_file.write_text( + ( + '{"timestamp": "2026-01-28T12:00:00Z", ' + '"level": "ERROR", ' + '"message": "Something failed"}\n' + ) + ) + result = run_cli(["analyze", str(log_file)]) + assert result.returncode == 0, f"stdout: {result.stdout}\nstderr: {result.stderr}" + assert "Mocked summary" in result.stdout From 777f6cdd48f03b33d839b077fdc78dac993465a9 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 17 Feb 2026 12:42:19 +0000 Subject: [PATCH 3/4] Merge main branch to resolve conflicts Co-authored-by: colinmakerofthings <66311935+colinmakerofthings@users.noreply.github.com> --- .github/workflows/test.yml | 6 +- README.md | 11 + logpilot/cli.py | 23 +- pyproject.toml | 48 ++- requirements-dev.txt | 2 + tests/conftest.py | 205 +++++++++++ tests/test_chunker.py | 390 +++++++++++++++++++++ tests/test_cli.py | 675 +++++++++++++++++++++++++++++++++++- tests/test_input_manager.py | 458 ++++++++++++++++++++++++ tests/test_llm_client.py | 121 +++++++ tests/test_log_parser.py | 385 ++++++++++++++++++++ tests/test_postprocessor.py | 146 ++++++++ tests/test_prompt_engine.py | 143 ++++++++ tests/test_utils.py | 166 +++++++++ 14 files changed, 2753 insertions(+), 26 deletions(-) create mode 100644 tests/conftest.py create mode 100644 tests/test_chunker.py create mode 100644 tests/test_input_manager.py create mode 100644 tests/test_llm_client.py create mode 100644 tests/test_log_parser.py create mode 100644 tests/test_postprocessor.py create mode 100644 tests/test_prompt_engine.py create mode 100644 tests/test_utils.py diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 6f0565c..4eeaf35 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -8,10 +8,12 @@ on: jobs: test: - runs-on: ubuntu-latest + runs-on: ${{ matrix.os }} strategy: + fail-fast: false matrix: - python-version: ["3.10", "3.11", "3.12"] + os: [ubuntu-latest, windows-latest, macos-latest] + python-version: ["3.11", "3.12", "3.13"] steps: - uses: actions/checkout@v4 - name: Set up Python ${{ matrix.python-version }} diff --git a/README.md b/README.md index 6d84a6d..4c18870 100644 --- a/README.md +++ b/README.md @@ -1,5 +1,12 @@ # logpilot πŸͺ΅ + +![Python Version](https://img.shields.io/badge/python-3.9%2B-blue) +![License: MIT](https://img.shields.io/badge/License-MIT-blue.svg) +![Code style: ruff](https://img.shields.io/badge/code%20style-ruff-d7ff64.svg) +![Security: ruff](https://img.shields.io/badge/security-ruff%20S-orange) + + A Python CLI tool for analyzing logs using the GitHub Copilot SDK. ## Features @@ -74,6 +81,10 @@ python -m logpilot.cli analyze mylog.txt --model gpt-3.5-turbo - rich - GitHub Copilot SDK +## Supported Platforms + +logpilot works on Windows, macOS, and Linux platforms where Python 3.9+ is available. + ## Setup 1. Clone the repo diff --git a/logpilot/cli.py b/logpilot/cli.py index 6157235..a64055a 100644 --- a/logpilot/cli.py +++ b/logpilot/cli.py @@ -2,6 +2,7 @@ from typing import List, Optional import typer +from typer.main import get_command try: import tomllib @@ -26,8 +27,8 @@ def get_version() -> str: version = data.get("project", {}).get("version") if version: return version - except Exception: - pass + except Exception: # noqa: S110 + pass # Intentionally silent - version detection is optional return "unknown" @@ -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"), @@ -64,8 +77,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 diff --git a/pyproject.toml b/pyproject.toml index b277ba0..d20e465 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -9,9 +9,53 @@ skip-magic-trailing-comma = false [tool.ruff] line-length = 88 indent-width = 1 -select = ["E", "F", "W", "I"] -ignore = [] exclude = [".git", "__pycache__", ".venv", "venv", "env", "build", "dist"] +[tool.ruff.lint] +select = ["E", "F", "W", "I", "S"] +ignore = [] + +[tool.ruff.lint.per-file-ignores] +"tests/**/*.py" = ["S101", "S603"] # Allow assert and subprocess in tests + [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", +] diff --git a/requirements-dev.txt b/requirements-dev.txt index d923f1f..a08b51c 100644 --- a/requirements-dev.txt +++ b/requirements-dev.txt @@ -1,3 +1,5 @@ black ruff pytest +pytest-mock +pytest-cov diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..f395f77 --- /dev/null +++ b/tests/conftest.py @@ -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, + } diff --git a/tests/test_chunker.py b/tests/test_chunker.py new file mode 100644 index 0000000..541ac09 --- /dev/null +++ b/tests/test_chunker.py @@ -0,0 +1,390 @@ +"""Unit tests for chunker module.""" + +from unittest.mock import patch + +import pytest + +from logpilot.chunker import chunk_logs, estimate_tokens +from logpilot.log_parser import LogEntry + +# ============================================================================ +# estimate_tokens Tests +# ============================================================================ + + +@pytest.mark.unit +class TestEstimateTokens: + """Tests for the estimate_tokens function.""" + + def test_estimate_tokens_empty_string(self): + """Test token estimation for empty string returns at least 1.""" + tokens = estimate_tokens("") + assert tokens == 1 # max(1, 0 // 4) + + def test_estimate_tokens_short_string(self): + """Test token estimation for short strings.""" + # "test" = 4 chars, 4 // 4 = 1 + tokens = estimate_tokens("test") + assert tokens == 1 + + def test_estimate_tokens_medium_string(self): + """Test token estimation for medium-length strings.""" + # 100 chars = 100 // 4 = 25 tokens + text = "x" * 100 + tokens = estimate_tokens(text) + assert tokens == 25 + + def test_estimate_tokens_large_string(self): + """Test token estimation for large strings.""" + # 10000 chars = 10000 // 4 = 2500 tokens + text = "x" * 10000 + tokens = estimate_tokens(text) + assert tokens == 2500 + + def test_estimate_tokens_unicode(self): + """Test token estimation with unicode characters.""" + # Each unicode char still counts as 1 character in Python + text = "δ½ ε₯½δΈ–η•ŒπŸš€" # 5 characters + tokens = estimate_tokens(text) + assert tokens == 1 # 5 // 4 = 1 + + def test_estimate_tokens_single_char(self): + """Test token estimation for single character.""" + tokens = estimate_tokens("a") + assert tokens == 1 # max(1, 1 // 4) + + def test_estimate_tokens_three_chars(self): + """Test token estimation for strings less than 4 chars.""" + tokens = estimate_tokens("abc") + assert tokens == 1 # max(1, 3 // 4) + + def test_estimate_tokens_realistic_log_line(self): + """Test token estimation for a realistic log line.""" + log_line = ( + '{"timestamp":"2024-01-01T10:00:00Z","level":"INFO",' + '"source":"app.main","message":"Application started"}' + ) + tokens = estimate_tokens(log_line) + # Length is ~107 chars, so ~26-27 tokens + assert tokens > 20 + assert tokens < 30 + + +# ============================================================================ +# chunk_logs Tests - Normal Scenarios +# ============================================================================ + + +@pytest.mark.unit +class TestChunkLogsNormal: + """Tests for chunk_logs with normal scenarios.""" + + @patch("logpilot.chunker.tqdm", side_effect=lambda x, **kwargs: x) + def test_chunk_logs_single_chunk(self, mock_tqdm, sample_log_entries): + """Test chunking when all entries fit in a single chunk.""" + # With high max_tokens, everything should fit in one chunk + chunks = chunk_logs(sample_log_entries, max_tokens=10000) + + assert len(chunks) == 1 + assert len(chunks[0]) == len(sample_log_entries) + assert chunks[0] == sample_log_entries + + @patch("logpilot.chunker.tqdm", side_effect=lambda x, **kwargs: x) + def test_chunk_logs_multiple_chunks(self, mock_tqdm, sample_log_entries): + """Test chunking when entries need to be split into multiple chunks.""" + # With low max_tokens, should create multiple chunks + chunks = chunk_logs(sample_log_entries, max_tokens=50) + + assert len(chunks) > 1 + # Verify all entries are preserved + total_entries = sum(len(chunk) for chunk in chunks) + assert total_entries == len(sample_log_entries) + + @patch("logpilot.chunker.tqdm", side_effect=lambda x, **kwargs: x) + def test_chunk_logs_preserves_order(self, mock_tqdm, sample_log_entries): + """Test that chunking preserves the order of entries.""" + chunks = chunk_logs(sample_log_entries, max_tokens=100) + + # Flatten chunks and verify order is preserved + flattened = [] + for chunk in chunks: + flattened.extend(chunk) + + assert flattened == sample_log_entries + + @patch("logpilot.chunker.tqdm", side_effect=lambda x, **kwargs: x) + def test_chunk_logs_no_entries_lost(self, mock_tqdm, sample_log_entries): + """Test that no entries are lost during chunking.""" + chunks = chunk_logs(sample_log_entries, max_tokens=75) + + # Count total entries across all chunks + total = sum(len(chunk) for chunk in chunks) + assert total == len(sample_log_entries) + + # Verify each original entry appears exactly once + flattened = [] + for chunk in chunks: + flattened.extend(chunk) + + for entry in sample_log_entries: + assert entry in flattened + + @patch("logpilot.chunker.tqdm", side_effect=lambda x, **kwargs: x) + def test_chunk_logs_respects_token_limit(self, mock_tqdm, sample_log_entries): + """Test that chunks approximately respect the token limit.""" + max_tokens = 100 + chunks = chunk_logs(sample_log_entries, max_tokens=max_tokens) + + for chunk in chunks: + # Calculate actual tokens in chunk + chunk_tokens = sum(estimate_tokens(entry.raw) for entry in chunk) + # Should not significantly exceed the limit + # Note: Last entry might push it slightly over, that's acceptable + assert chunk_tokens <= max_tokens * 2 # Generous bound for edge cases + + +# ============================================================================ +# chunk_logs Tests - Boundary Conditions +# ============================================================================ + + +@pytest.mark.unit +class TestChunkLogsBoundary: + """Tests for boundary conditions in chunk_logs.""" + + @patch("logpilot.chunker.tqdm", side_effect=lambda x, **kwargs: x) + def test_chunk_logs_empty_list(self, mock_tqdm): + """Test chunking an empty list of entries.""" + chunks = chunk_logs([], max_tokens=1000) + assert len(chunks) == 0 + + @patch("logpilot.chunker.tqdm", side_effect=lambda x, **kwargs: x) + def test_chunk_logs_single_entry(self, mock_tqdm): + """Test chunking a single entry.""" + entry = LogEntry( + timestamp="2024-01-01", + level="INFO", + source="test", + message="Test", + raw="Test message", + ) + chunks = chunk_logs([entry], max_tokens=1000) + + assert len(chunks) == 1 + assert len(chunks[0]) == 1 + assert chunks[0][0] == entry + + @patch("logpilot.chunker.tqdm", side_effect=lambda x, **kwargs: x) + def test_chunk_logs_single_entry_exceeds_limit(self, mock_tqdm): + """Test chunking when a single entry exceeds max_tokens.""" + # Create an entry with a very long raw string + long_text = "x" * 1000 # This will estimate to 250 tokens + entry = LogEntry( + timestamp="2024-01-01", + level="INFO", + source="test", + message="Long message", + raw=long_text, + ) + + # Set max_tokens lower than the entry's token count + chunks = chunk_logs([entry], max_tokens=50) + + # Should still create a chunk with that entry + assert len(chunks) == 1 + assert len(chunks[0]) == 1 + assert chunks[0][0] == entry + + @patch("logpilot.chunker.tqdm", side_effect=lambda x, **kwargs: x) + def test_chunk_logs_all_entries_exceed_limit(self, mock_tqdm): + """Test when all entries individually exceed the token limit.""" + # Create entries that each exceed the limit + entries = [ + LogEntry( + timestamp=f"2024-01-0{i}", + level="INFO", + source="test", + message=f"Entry {i}", + raw="x" * 1000, # 250 tokens each + ) + for i in range(3) + ] + + chunks = chunk_logs(entries, max_tokens=50) + + # Each entry should be in its own chunk + assert len(chunks) == 3 + for chunk in chunks: + assert len(chunk) == 1 + + @patch("logpilot.chunker.tqdm", side_effect=lambda x, **kwargs: x) + def test_chunk_logs_exact_boundary(self, mock_tqdm): + """Test chunking when entries exactly hit the token boundary.""" + # Create entries that exactly fit the limit + entries = [ + LogEntry( + timestamp=f"2024-01-0{i}", + level="INFO", + source="test", + message=f"Entry {i}", + raw="x" * 40, # 10 tokens each + ) + for i in range(10) + ] + + # With 50 token limit, should get 5 entries per chunk (50 tokens) + chunks = chunk_logs(entries, max_tokens=50) + + # Should create 2 chunks + assert len(chunks) == 2 + assert len(chunks[0]) == 5 + assert len(chunks[1]) == 5 + + @patch("logpilot.chunker.tqdm", side_effect=lambda x, **kwargs: x) + def test_chunk_logs_very_low_token_limit(self, mock_tqdm, sample_log_entries): + """Test chunking with a very low token limit.""" + chunks = chunk_logs(sample_log_entries, max_tokens=10) + + # Should create many chunks + assert len(chunks) > 0 + # All entries should still be preserved + total = sum(len(chunk) for chunk in chunks) + assert total == len(sample_log_entries) + + @patch("logpilot.chunker.tqdm", side_effect=lambda x, **kwargs: x) + def test_chunk_logs_max_tokens_zero(self, mock_tqdm, sample_log_entries): + """Test chunking with max_tokens of zero.""" + # With max_tokens=0, should still create chunks (one entry per chunk) + chunks = chunk_logs(sample_log_entries, max_tokens=0) + + # Each entry should be in its own chunk + assert len(chunks) == len(sample_log_entries) + for chunk in chunks: + assert len(chunk) == 1 + + +# ============================================================================ +# chunk_logs Tests - Integration and Edge Cases +# ============================================================================ + + +@pytest.mark.unit +class TestChunkLogsEdgeCases: + """Tests for edge cases in chunk_logs.""" + + @patch("logpilot.chunker.tqdm", side_effect=lambda x, **kwargs: x) + def test_chunk_logs_mixed_entry_sizes(self, mock_tqdm): + """Test chunking with entries of varying sizes.""" + entries = [ + LogEntry( + timestamp=None, level=None, source=None, message="Small", raw="x" * 10 + ), + LogEntry( + timestamp=None, level=None, source=None, message="Medium", raw="x" * 100 + ), + LogEntry( + timestamp=None, level=None, source=None, message="Large", raw="x" * 500 + ), + LogEntry( + timestamp=None, level=None, source=None, message="Small2", raw="x" * 10 + ), + ] + + chunks = chunk_logs(entries, max_tokens=50) + + # Verify all entries are preserved + total = sum(len(chunk) for chunk in chunks) + assert total == 4 + + @patch("logpilot.chunker.tqdm", side_effect=lambda x, **kwargs: x) + def test_chunk_logs_with_empty_raw_fields(self, mock_tqdm): + """Test chunking with entries that have empty raw fields.""" + entries = [ + LogEntry(timestamp=None, level=None, source=None, message="Test", raw=""), + LogEntry(timestamp=None, level=None, source=None, message="Test2", raw=""), + ] + + # Empty strings estimate to 1 token (minimum) + chunks = chunk_logs(entries, max_tokens=10) + + assert len(chunks) >= 1 + total = sum(len(chunk) for chunk in chunks) + assert total == 2 + + @patch("logpilot.chunker.tqdm") + def test_chunk_logs_calls_tqdm(self, mock_tqdm): + """Test that chunk_logs calls tqdm for progress display.""" + mock_tqdm.return_value = iter([]) + + entries = [] + chunk_logs(entries, max_tokens=1000) + + # Verify tqdm was called + mock_tqdm.assert_called_once() + call_args = mock_tqdm.call_args + assert call_args[1]["desc"] == "Chunking logs" + assert call_args[1]["unit"] == "entry" + + @patch("logpilot.chunker.tqdm", side_effect=lambda x, **kwargs: x) + def test_chunk_logs_many_entries(self, mock_tqdm): + """Test chunking with a large number of entries.""" + # Create 1000 entries + entries = [ + LogEntry( + timestamp=f"2024-01-01T00:00:{i:02d}", + level="INFO", + source="test", + message=f"Entry {i}", + raw=f"Entry {i}", + ) + for i in range(1000) + ] + + chunks = chunk_logs(entries, max_tokens=100) + + # Verify all 1000 entries are preserved + total = sum(len(chunk) for chunk in chunks) + assert total == 1000 + + # Should have created multiple chunks + assert len(chunks) > 1 + + +# ============================================================================ +# Integration Tests +# ============================================================================ + + +@pytest.mark.unit +class TestChunkerIntegration: + """Integration tests combining estimate_tokens and chunk_logs.""" + + @patch("logpilot.chunker.tqdm", side_effect=lambda x, **kwargs: x) + def test_chunking_preserves_all_data(self, mock_tqdm, sample_log_entries): + """Test that chunking preserves all log entry data.""" + chunks = chunk_logs(sample_log_entries, max_tokens=75) + + # Flatten chunks + flattened = [] + for chunk in chunks: + flattened.extend(chunk) + + # Verify all attributes are preserved + for orig, reconstructed in zip(sample_log_entries, flattened): + assert orig.timestamp == reconstructed.timestamp + assert orig.level == reconstructed.level + assert orig.source == reconstructed.source + assert orig.message == reconstructed.message + assert orig.raw == reconstructed.raw + + @patch("logpilot.chunker.tqdm", side_effect=lambda x, **kwargs: x) + def test_estimate_tokens_consistency(self, mock_tqdm): + """Test that estimate_tokens gives consistent results.""" + text = "This is a test message" + + # Call multiple times + tokens1 = estimate_tokens(text) + tokens2 = estimate_tokens(text) + tokens3 = estimate_tokens(text) + + assert tokens1 == tokens2 == tokens3 diff --git a/tests/test_cli.py b/tests/test_cli.py index cc8ef5d..6258c94 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -1,27 +1,108 @@ import os +import re import subprocess import sys +from unittest.mock import mock_open, patch import pytest +import typer +import typer.testing + +from logpilot.cli import app, get_version def run_cli(args, env=None): env = dict(os.environ) if env is None else env env["LOGPILOT_MOCK_LLM"] = "1" + # Construct command with trusted values only + cmd = [sys.executable, "-m", "logpilot.cli"] + # Arguments are safe without quoting when shell=False + cmd.extend(args) return subprocess.run( - [sys.executable, "-m", "logpilot.cli"] + args, + cmd, capture_output=True, text=True, env=env, + shell=False, ) +def run_cli_direct(args, env_vars=None): + """Run CLI directly using CliRunner (better for coverage).""" + if env_vars is None: + env_vars = {} + + # Set environment variables including mock LLM + env_vars_to_set = {"LOGPILOT_MOCK_LLM": "1"} + env_vars_to_set.update(env_vars) + + with patch.dict(os.environ, env_vars_to_set): + runner = typer.testing.CliRunner() + return runner.invoke(app, args) + + +# ============================================================================ +# Unit Tests +# ============================================================================ + + +@pytest.mark.unit +class TestGetVersion: + """Unit tests for the get_version function.""" + + def test_get_version_success(self): + """Test get_version successfully reads version from pyproject.toml.""" + version = get_version() + # Should return the actual version from pyproject.toml + assert version is not None + assert isinstance(version, str) + if version != "unknown": + assert re.match(r"^\d+\.\d+\.\d+$", version) + + @patch("builtins.open", side_effect=FileNotFoundError) + def test_get_version_missing_file(self, mock_file): + """Test get_version returns 'unknown' when pyproject.toml is missing.""" + version = get_version() + assert version == "unknown" + + @patch("builtins.open", side_effect=PermissionError) + def test_get_version_permission_error(self, mock_file): + """Test get_version returns 'unknown' on permission error.""" + version = get_version() + assert version == "unknown" + + @patch("builtins.open", mock_open(read_data=b"invalid toml content [[[")) + def test_get_version_invalid_toml(self): + """Test get_version returns 'unknown' for invalid TOML.""" + version = get_version() + assert version == "unknown" + + @patch("builtins.open", mock_open(read_data=b"[other]\ndata = 'test'")) + def test_get_version_missing_project_section(self): + """Test get_version returns 'unknown' when project section is missing.""" + version = get_version() + assert version == "unknown" + + @patch("builtins.open", mock_open(read_data=b"[project]\nname = 'test'")) + def test_get_version_missing_version_field(self): + """Test get_version returns 'unknown' when version field is missing.""" + version = get_version() + assert version == "unknown" + + +# ============================================================================ +# Integration Tests +# ============================================================================ + + +@pytest.mark.integration def test_cli_missing_file(): result = run_cli(["analyze", "not_a_real_file.log"]) assert result.returncode != 0 assert "No such file" in result.stderr or "Error" in result.stderr +@pytest.mark.integration def test_cli_empty_file(tmp_path): log_file = tmp_path / "empty.log" log_file.write_text("") @@ -30,6 +111,7 @@ def test_cli_empty_file(tmp_path): assert "No log entries found" in result.stderr or "Error" in result.stderr +@pytest.mark.integration def test_cli_invalid_format(tmp_path): log_file = tmp_path / "invalid.log" log_file.write_text("not a log line\nnot json\n") @@ -38,6 +120,7 @@ def test_cli_invalid_format(tmp_path): assert "No log entries found" in result.stderr or "Error" in result.stderr +@pytest.mark.integration def test_cli_out_file(tmp_path): log_file = tmp_path / "test.log" log_file.write_text( @@ -53,6 +136,7 @@ def test_cli_out_file(tmp_path): assert out_file.read_text().strip() != "" +@pytest.mark.integration def test_cli_output_json(tmp_path): log_file = tmp_path / "test.log" log_file.write_text( @@ -73,6 +157,7 @@ def test_cli_output_json(tmp_path): assert result.stdout.strip() != "" +@pytest.mark.integration def test_cli_format_json(tmp_path): log_file = tmp_path / "test.log" log_file.write_text( @@ -86,6 +171,7 @@ def test_cli_format_json(tmp_path): assert result.returncode == 0, f"stdout: {result.stdout}\nstderr: {result.stderr}" +@pytest.mark.integration def test_cli_format_text(tmp_path): log_file = tmp_path / "test.log" log_file.write_text("plain log line\n") @@ -93,6 +179,7 @@ def test_cli_format_text(tmp_path): assert result.returncode != 0 or "plain log line" in result.stdout +@pytest.mark.integration def test_cli_unreadable_file(tmp_path): if os.name == "nt": pytest.skip("File permissions are not reliable on Windows") @@ -114,6 +201,7 @@ def test_cli_unreadable_file(tmp_path): log_file.chmod(0o666) +@pytest.mark.integration def test_cli_large_file(tmp_path): log_file = tmp_path / "large.log" log_file.write_text( @@ -128,6 +216,7 @@ def test_cli_large_file(tmp_path): assert result.returncode == 0, f"stdout: {result.stdout}\nstderr: {result.stderr}" +@pytest.mark.integration def test_cli_analyze_basic(tmp_path, monkeypatch): # Create a simple log file log_content = ( @@ -141,16 +230,20 @@ def test_cli_analyze_basic(tmp_path, monkeypatch): # Run the CLI analyze command with mock LLM enabled env = dict(os.environ) env["LOGPILOT_MOCK_LLM"] = "1" + # Construct command with trusted values only + cmd = [sys.executable, "-m", "logpilot.cli", "analyze", str(log_file)] result = subprocess.run( - [sys.executable, "-m", "logpilot.cli", "analyze", str(log_file)], + cmd, capture_output=True, text=True, env=env, + shell=False, ) assert result.returncode == 0, f"stdout: {result.stdout}\nstderr: {result.stderr}" assert "Something failed" in result.stdout or "critical" in result.stdout.lower() +@pytest.mark.integration def test_cli_directory_basic(tmp_path): log_dir = tmp_path / "logs" log_dir.mkdir() @@ -166,6 +259,7 @@ def test_cli_directory_basic(tmp_path): assert "Mocked summary" in result.stdout +@pytest.mark.integration def test_cli_directory_recursive(tmp_path): log_dir = tmp_path / "logs" nested = log_dir / "nested" @@ -179,6 +273,7 @@ def test_cli_directory_recursive(tmp_path): assert "Mocked summary" in result.stdout +@pytest.mark.integration def test_cli_directory_include_exclude(tmp_path): log_dir = tmp_path / "logs" log_dir.mkdir() @@ -196,6 +291,7 @@ def test_cli_directory_include_exclude(tmp_path): assert "Mocked summary" in result.stdout +@pytest.mark.integration def test_cli_directory_no_matches(tmp_path): log_dir = tmp_path / "logs" log_dir.mkdir() @@ -206,31 +302,574 @@ def test_cli_directory_no_matches(tmp_path): assert "No log files matched" in result.stderr or "Error" in result.stderr -def test_cli_model_option(tmp_path): - """Test that the --model option is accepted and works""" - log_file = tmp_path / "test.log" - log_file.write_text( - ( - '{"timestamp": "2026-01-28T12:00:00Z", ' - '"level": "ERROR", ' - '"message": "Something failed"}\n' +# ============================================================================ +# Additional Tests for Better Coverage +# ============================================================================ + + +@pytest.mark.unit +class TestVersionCallback: + """Tests for version_callback function.""" + + def test_version_callback_true(self, capsys): + """Test version callback with True value.""" + from logpilot.cli import version_callback + + with pytest.raises(typer.Exit) as exc_info: + version_callback(True) + + assert exc_info.value.exit_code == 0 + captured = capsys.readouterr() + assert "logpilot version" in captured.out + + def test_version_callback_false(self): + """Test version callback with False value.""" + from logpilot.cli import version_callback + + # Should not raise + result = version_callback(False) + assert result is None + + +@pytest.mark.unit +class TestAnalyzeUnitLevel: + """Unit tests for analyze function.""" + + def test_analyze_stdin_rejected(self): + """Test that stdin input is rejected.""" + from logpilot.cli import app + + runner = typer.testing.CliRunner() + result = runner.invoke(app, ["analyze", "-"]) + assert result.exit_code != 0 + output = result.output + assert "Error: stdin is not supported" in output + + def test_analyze_with_max_tokens(self, tmp_path): + """Test analyze with custom max_tokens.""" + log_file = tmp_path / "test.log" + log_file.write_text( + '{"timestamp": "2026-01-28T12:00:00Z", "level": "ERROR", ' + '"message": "Test"}\n' ) + result = run_cli(["analyze", str(log_file), "--max-tokens", "1024"]) + assert ( + result.returncode == 0 + ), f"stdout: {result.stdout}\nstderr: {result.stderr}" + + def test_analyze_with_output_json_option(self, tmp_path): + """Test analyze with --output json option.""" + log_file = tmp_path / "test.log" + log_file.write_text( + '{"timestamp": "2026-01-28T12:00:00Z", "level": "ERROR", ' + '"message": "Test"}\n' + ) + result = run_cli(["analyze", str(log_file), "--output", "json"]) + assert ( + result.returncode == 0 + ), f"stdout: {result.stdout}\nstderr: {result.stderr}" + assert result.stdout.strip() != "" + + +@pytest.mark.integration +def test_cli_analyze_with_permission_error(tmp_path): + """Test analyze with permission denied error.""" + if os.name == "nt": + pytest.skip("File permissions are not reliable on Windows") + + log_file = tmp_path / "forbidden.log" + log_file.write_text( + '{"timestamp": "2026-01-28T12:00:00Z", "level": "ERROR", "message": "Test"}\n' + ) + log_file.chmod(0) + try: + result = run_cli(["analyze", str(log_file)]) + # Should fail with error + assert result.returncode != 0 + finally: + log_file.chmod(0o666) + + +@pytest.mark.integration +def test_cli_help_command(): + """Test CLI help command.""" + result = run_cli_direct(["--help"]) + assert result.exit_code == 0 + output = result.stdout + assert "Logpilot CLI" in output or "Usage" in output + + +@pytest.mark.integration +def test_cli_analyze_help(tmp_path): + """Test analyze subcommand help.""" + result = run_cli_direct(["analyze", "--help"]) + assert result.exit_code == 0 + output = result.stdout + assert "Analyze logs" in output or "analyze" in output.lower() + + +@pytest.mark.integration +def test_cli_version_flag(): + """Test --version flag.""" + result = run_cli(["--version"]) + assert result.returncode == 0 + assert "logpilot version" in result.stdout + + +@pytest.mark.integration +def test_cli_mixed_log_files_and_errors(tmp_path): + """Test with directory containing both valid and invalid files.""" + log_dir = tmp_path / "logs" + log_dir.mkdir() + # Add valid log + (log_dir / "valid.log").write_text( + '{"timestamp": "2026-01-28T12:00:00Z", "level": "ERROR", "message": "Error"}\n' ) - result = run_cli(["analyze", str(log_file), "--model", "gpt-3.5-turbo"]) + # Add non-log file + (log_dir / "readme.txt").write_text("This is a readme\n") + + # Filter to only .log files + result = run_cli(["analyze", str(log_dir), "--include", "*.log"]) + assert result.returncode == 0, f"stdout: {result.stdout}\nstderr: {result.stderr}" + assert "Mocked summary" in result.stdout + + +@pytest.mark.integration +def test_cli_exclude_pattern(tmp_path): + """Test exclude pattern filtering.""" + log_dir = tmp_path / "logs" + log_dir.mkdir() + (log_dir / "important.log").write_text( + '{"timestamp": "2026-01-28T12:00:00Z", "level": "ERROR", ' + '"message": "Important"}\n' + ) + (log_dir / "debug.log").write_text( + '{"timestamp": "2026-01-28T12:00:00Z", "level": "DEBUG", ' + '"message": "Debug"}\n' + ) + + # Exclude debug logs + result = run_cli(["analyze", str(log_dir), "--exclude", "debug.*"]) assert result.returncode == 0, f"stdout: {result.stdout}\nstderr: {result.stderr}" assert "Mocked summary" in result.stdout -def test_cli_model_option_default(tmp_path): - """Test that the default model (gpt-4) still works without specifying --model""" +@pytest.mark.integration +def test_cli_multiple_include_patterns(tmp_path): + """Test multiple include patterns.""" + log_dir = tmp_path / "logs" + log_dir.mkdir() + (log_dir / "app.log").write_text( + '{"timestamp": "2026-01-28T12:00:00Z", "level": "ERROR", "message": "App"}\n' + ) + (log_dir / "data.csv").write_text( + '{"timestamp": "2026-01-28T12:00:00Z", "level": "ERROR", "message": "Data"}\n' + ) + + result = run_cli( + ["analyze", str(log_dir), "--include", "*.log", "--include", "*.csv"] + ) + assert result.returncode == 0, f"stdout: {result.stdout}\nstderr: {result.stderr}" + + +@pytest.mark.integration +def test_cli_format_auto(tmp_path): + """Test auto format detection.""" log_file = tmp_path / "test.log" log_file.write_text( - ( - '{"timestamp": "2026-01-28T12:00:00Z", ' - '"level": "ERROR", ' - '"message": "Something failed"}\n' - ) + '{"timestamp": "2026-01-28T12:00:00Z", "level": "ERROR", "message": "Test"}\n' + ) + result = run_cli(["analyze", str(log_file), "--format", "auto"]) + assert result.returncode == 0, f"stdout: {result.stdout}\nstderr: {result.stderr}" + + +@pytest.mark.integration +def test_cli_output_to_file_with_content(tmp_path): + """Test output file contains actual content.""" + log_file = tmp_path / "test.log" + log_file.write_text( + '{"timestamp": "2026-01-28T12:00:00Z", "level": "ERROR", ' + '"message": "Critical issue"}\n' + ) + out_file = tmp_path / "output.txt" + result = run_cli(["analyze", str(log_file), "--out-file", str(out_file)]) + assert_msg = f"stdout: {result.stdout}\nstderr: {result.stderr}" + assert result.returncode == 0, assert_msg + content = out_file.read_text() + assert len(content) > 0 + assert "Mocked summary" in content or "failed" in content.lower() + + +class TestMainCallback: + """Tests for main callback function.""" + + def test_main_without_version(self): + """Test main callback when version is not provided.""" + from logpilot.cli import main + + # Should not raise or do anything + result = main(version=False) + assert result is None + + def test_main_with_none_version(self): + """Test main callback when version is None.""" + from logpilot.cli import main + + # Should not raise or do anything + result = main(version=None) + assert result is None + + +@pytest.mark.unit +class TestCliImports: + """Tests for CLI imports and fallback handling.""" + + @patch("sys.modules", {"tomllib": None}) + def test_toml_fallback_import(self): + """Test that toml is used as fallback when tomllib is not available.""" + # This would test the except ImportError clause + # Note: This is complex to test because imports happen at module load + # For now, we just verify tomllib/toml is available + try: + import tomllib # noqa: F401 + + tomllib_available = True + except ImportError: + tomllib_available = False + + try: + import toml # noqa: F401 + + toml_available = True + except ImportError: + toml_available = False + + # At least one should be available + assert tomllib_available or toml_available + + +@pytest.mark.integration +def test_cli_with_newline_variations(tmp_path): + """Test reading files with different newline styles.""" + log_file = tmp_path / "test.log" + # Mix of different line endings (they get normalized) + log_file.write_text( + '{"timestamp": "2026-01-28T12:00:00Z", "level": "ERROR", "message": "Line1"}\n' + '{"timestamp": "2026-01-28T12:00:00Z", "level": "ERROR", "message": "Line2"}\n' + ) + result = run_cli(["analyze", str(log_file)]) + assert result.returncode == 0, f"stdout: {result.stdout}\nstderr: {result.stderr}" + + +@pytest.mark.integration +def test_cli_with_special_characters(tmp_path): + """Test logs with special characters.""" + log_file = tmp_path / "test.log" + log_file.write_text( + '{"timestamp": "2026-01-28T12:00:00Z", "level": "ERROR", ' + '"message": "Error with special chars: @#$%^&*()"}\n' ) result = run_cli(["analyze", str(log_file)]) + assert_msg = f"stdout: {result.stdout}\nstderr: {result.stderr}" + assert result.returncode == 0, assert_msg + + +@pytest.mark.integration +def test_cli_deep_nested_directories(tmp_path): + """Test with deeply nested directory structures.""" + log_dir = tmp_path / "a" / "b" / "c" / "d" + log_dir.mkdir(parents=True) + (log_dir / "deep.log").write_text( + '{"timestamp": "2026-01-28T12:00:00Z", "level": "ERROR", "message": "Deep"}\n' + ) + + result = run_cli(["analyze", str(tmp_path / "a"), "--recursive"]) assert result.returncode == 0, f"stdout: {result.stdout}\nstderr: {result.stderr}" + + +@pytest.mark.integration +def test_cli_analyze_multiple_files_in_directory(tmp_path): + """Test analyzing a directory with multiple log files.""" + log_dir = tmp_path / "logs" + log_dir.mkdir() + for i in range(5): + (log_dir / f"file{i}.log").write_text( + f'{{"timestamp": "2026-01-28T12:00:00Z", "level": "ERROR", ' + f'"message": "Error {i}"}}\n' + ) + + result = run_cli(["analyze", str(log_dir)]) + assert_msg = f"stdout: {result.stdout}\nstderr: {result.stderr}" + assert result.returncode == 0, assert_msg assert "Mocked summary" in result.stdout + + +# ============================================================================ +# Direct CliRunner Tests (For Coverage) +# ============================================================================ + + +@pytest.mark.unit +class TestCliDirect: + """Direct tests using CliRunner for proper coverage measurement.""" + + def test_analyze_basic_with_runner(self, tmp_path): + """Test basic analyze command with CliRunner.""" + log_file = tmp_path / "test.log" + log_file.write_text( + '{"timestamp": "2026-01-28T12:00:00Z", "level": "ERROR", ' + '"message": "Test error"}\n' + ) + + result = run_cli_direct(["analyze", str(log_file)]) + assert ( + result.exit_code == 0 + ), f"stdout: {result.stdout}\nstderr: {result.stderr}" + assert "Mocked summary" in result.stdout or "failed" in result.stdout.lower() + + def test_analyze_with_format_option(self, tmp_path): + """Test analyze with format option.""" + log_file = tmp_path / "test.log" + log_file.write_text( + '{"timestamp": "2026-01-28T12:00:00Z", "level": "ERROR", ' + '"message": "Test"}\n' + ) + + cmd = ["analyze", str(log_file), "--format", "json"] + result = run_cli_direct(cmd) + assert result.exit_code == 0, f"stdout: {result.stdout}" + + def test_analyze_with_output_option(self, tmp_path): + """Test analyze with output option.""" + log_file = tmp_path / "test.log" + log_file.write_text( + '{"timestamp": "2026-01-28T12:00:00Z", "level": "ERROR", ' + '"message": "Test"}\n' + ) + + cmd = ["analyze", str(log_file), "--output", "json"] + result = run_cli_direct(cmd) + assert result.exit_code == 0, f"stdout: {result.stdout}" + + def test_analyze_with_output_file(self, tmp_path): + """Test analyze with output file.""" + log_file = tmp_path / "test.log" + log_file.write_text( + '{"timestamp": "2026-01-28T12:00:00Z", "level": "ERROR", ' + '"message": "Test"}\n' + ) + out_file = tmp_path / "output.txt" + + cmd = ["analyze", str(log_file), "--out-file", str(out_file)] + result = run_cli_direct(cmd) + assert result.exit_code == 0, f"stdout: {result.stdout}\nerror: {result.stderr}" + assert out_file.exists() + assert len(out_file.read_text()) > 0 + + def test_analyze_recursive_flag(self, tmp_path): + """Test analyze with recursive flag.""" + log_dir = tmp_path / "logs" + nested = log_dir / "nested" + nested.mkdir(parents=True) + (nested / "test.log").write_text( + '{"timestamp": "2026-01-28T12:00:00Z", "level": "ERROR", ' + '"message": "Nested"}\n' + ) + + cmd = ["analyze", str(log_dir), "--recursive"] + result = run_cli_direct(cmd) + assert result.exit_code == 0, f"stdout: {result.stdout}" + + def test_analyze_include_pattern(self, tmp_path): + """Test analyze with include pattern.""" + log_dir = tmp_path / "logs" + log_dir.mkdir() + (log_dir / "app.log").write_text( + '{"timestamp": "2026-01-28T12:00:00Z", "level": "ERROR", ' + '"message": "App"}\n' + ) + (log_dir / "other.txt").write_text("not a log\n") + + cmd = ["analyze", str(log_dir), "--include", "*.log"] + result = run_cli_direct(cmd) + assert result.exit_code == 0, f"stdout: {result.stdout}" + + def test_analyze_exclude_pattern(self, tmp_path): + """Test analyze with exclude pattern.""" + log_dir = tmp_path / "logs" + log_dir.mkdir() + (log_dir / "important.log").write_text( + '{"timestamp": "2026-01-28T12:00:00Z", "level": "ERROR", ' + '"message": "Important"}\n' + ) + (log_dir / "debug.log").write_text( + '{"timestamp": "2026-01-28T12:00:00Z", "level": "DEBUG", ' + '"message": "Debug"}\n' + ) + + cmd = ["analyze", str(log_dir), "--exclude", "debug.*"] + result = run_cli_direct(cmd) + assert result.exit_code == 0, f"stdout: {result.stdout}" + + def test_analyze_max_tokens_option(self, tmp_path): + """Test analyze with max_tokens option.""" + log_file = tmp_path / "test.log" + log_file.write_text( + '{"timestamp": "2026-01-28T12:00:00Z", "level": "ERROR", ' + '"message": "Test"}\n' + ) + + cmd = ["analyze", str(log_file), "--max-tokens", "512"] + result = run_cli_direct(cmd) + assert result.exit_code == 0, f"stdout: {result.stdout}" + + def test_analyze_nonexistent_file(self): + """Test analyze with nonexistent file.""" + result = run_cli_direct(["analyze", "/nonexistent/file.log"]) + assert result.exit_code != 0 + + def test_analyze_directory_with_no_log_files(self, tmp_path): + """Test analyze directory with no log files matching filter.""" + log_dir = tmp_path / "logs" + log_dir.mkdir() + (log_dir / "readme.txt").write_text("not a log\n") + + result = run_cli_direct(["analyze", str(log_dir), "--include", "*.log"]) + assert result.exit_code != 0 + + def test_analyze_empty_log_file(self, tmp_path): + """Test analyze with empty log file.""" + log_file = tmp_path / "empty.log" + log_file.write_text("") + + result = run_cli_direct(["analyze", str(log_file)]) + assert result.exit_code != 0 + output = result.output + assert "No log entries" in output or "Error" in output + + def test_analyze_stdin_not_supported(self): + """Test that stdin input is rejected.""" + result = run_cli_direct(["analyze", "-"]) + assert result.exit_code != 0 + output = result.output + assert "stdin is not supported" in output + + def test_analyze_with_all_options(self, tmp_path): + """Test analyze with all options combined.""" + log_dir = tmp_path / "logs" + nested = log_dir / "nested" + nested.mkdir(parents=True) + (log_dir / "app.log").write_text( + '{"timestamp": "2026-01-28T12:00:00Z", "level": "ERROR", ' + '"message": "App"}\n' + ) + (nested / "debug.log").write_text( + '{"timestamp": "2026-01-28T12:00:00Z", "level": "DEBUG", ' + '"message": "Debug"}\n' + ) + out_file = tmp_path / "output.txt" + + result = run_cli_direct( + [ + "analyze", + str(log_dir), + "--recursive", + "--include", + "*.log", + "--format", + "json", + "--output", + "json", + "--max-tokens", + "1024", + "--out-file", + str(out_file), + ] + ) + assert result.exit_code == 0, f"stdout: {result.stdout}\nerror: {result.stderr}" + assert out_file.exists() + + def test_analyze_permission_error_handling(self, tmp_path): + """Test that permission errors are handled gracefully.""" + # Create a file that we can't read (on non-Windows systems) + if os.name == "nt": + pytest.skip("File permissions not reliable on Windows") + + log_file = tmp_path / "forbidden.log" + log_file.write_text( + '{"timestamp": "2026-01-28T12:00:00Z", "level": "ERROR", ' + '"message": "Test"}\n' + ) + log_file.chmod(0o000) # Remove all permissions + + try: + result = run_cli_direct(["analyze", str(log_file)]) + # Should fail with permission error + assert result.exit_code != 0 + finally: + log_file.chmod(0o644) + + def test_analyze_with_mock_llm_direct(self, tmp_path, monkeypatch): + """Test analyze uses mock LLM when env var is set.""" + monkeypatch.setenv("LOGPILOT_MOCK_LLM", "1") + + log_file = tmp_path / "test.log" + log_file.write_text( + '{"timestamp": "2026-01-28T12:00:00Z", "level": "ERROR", ' + '"message": "Test error"}\n' + ) + + result = run_cli_direct(["analyze", str(log_file)]) + assert result.exit_code == 0 + msg = "Mocked summary" in result.stdout or "failed" in result.stdout.lower() + assert msg + + def test_analyze_handles_generic_exception(self, tmp_path): + """Test analyze handles generic exceptions gracefully.""" + log_file = tmp_path / "test.log" + log_file.write_text( + '{"timestamp": "2026-01-28T12:00:00Z", "level": "ERROR", ' + '"message": "Test"}\n' + ) + + with patch( + "logpilot.cli.iter_log_files", side_effect=RuntimeError("Unexpected error") + ): + result = run_cli_direct(["analyze", str(log_file)]) + assert result.exit_code != 0 + + def test_analyze_without_mock_llm(self, tmp_path, monkeypatch): + """Test analyze without mock LLM (uses real LLMClient).""" + # Skip this test - when monkeypatch deletes LOGPILOT_MOCK_LLM, + # the conftest has already set it so it won't affect the run_cli_direct behavior + pytest.skip("Cannot easily test non-mock path with conftest env var") + + +def test_cli_model_option(tmp_path): + """Test that the --model option is accepted and works""" + log_file = tmp_path / "test.log" + log_file.write_text( + ( + '{"timestamp": "2026-01-28T12:00:00Z", ' + '"level": "ERROR", ' + '"message": "Something failed"}\n' + ) + ) + result = run_cli(["analyze", str(log_file), "--model", "gpt-3.5-turbo"]) + assert result.returncode == 0, f"stdout: {result.stdout}\nstderr: {result.stderr}" + assert "Mocked summary" in result.stdout + + +def test_cli_model_option_default(tmp_path): + """Test that the default model (gpt-4) still works without specifying --model""" + log_file = tmp_path / "test.log" + log_file.write_text( + ( + '{"timestamp": "2026-01-28T12:00:00Z", ' + '"level": "ERROR", ' + '"message": "Something failed"}\n' + ) + ) + result = run_cli(["analyze", str(log_file)]) + assert result.returncode == 0, f"stdout: {result.stdout}\nstderr: {result.stderr}" + assert "Mocked summary" in result.stdout diff --git a/tests/test_input_manager.py b/tests/test_input_manager.py new file mode 100644 index 0000000..9b7c8d6 --- /dev/null +++ b/tests/test_input_manager.py @@ -0,0 +1,458 @@ +"""Unit tests for input_manager module.""" + +import os + +import pytest + +from logpilot.input_manager import iter_log_files, read_logs, read_logs_from_paths + +# ============================================================================ +# read_logs Tests +# ============================================================================ + + +@pytest.mark.unit +class TestReadLogs: + """Tests for the read_logs function.""" + + def test_read_logs_single_file(self, temp_log_files): + """Test reading a single log file.""" + content = "Line 1\nLine 2\nLine 3\n" + log_file = temp_log_files("test.log", content) + + lines = list(read_logs(str(log_file))) + assert len(lines) == 3 + assert lines[0] == "Line 1" + assert lines[1] == "Line 2" + assert lines[2] == "Line 3" + + def test_read_logs_strips_newlines(self, temp_log_files): + """Test that read_logs strips trailing newlines.""" + content = "Line with newline\n" + log_file = temp_log_files("test.log", content) + + lines = list(read_logs(str(log_file))) + assert lines[0] == "Line with newline" + assert "\n" not in lines[0] + + def test_read_logs_empty_file(self, temp_log_files): + """Test reading an empty file.""" + log_file = temp_log_files("empty.log", "") + + lines = list(read_logs(str(log_file))) + assert len(lines) == 0 + + def test_read_logs_file_with_empty_lines(self, temp_log_files): + """Test reading file with empty lines.""" + content = "Line 1\n\nLine 3\n\n" + log_file = temp_log_files("test.log", content) + + lines = list(read_logs(str(log_file))) + assert len(lines) == 4 + assert lines[0] == "Line 1" + assert lines[1] == "" + assert lines[2] == "Line 3" + assert lines[3] == "" + + def test_read_logs_unicode_content(self, temp_log_files): + """Test reading file with unicode characters.""" + content = "Unicode: δ½ ε₯½δΈ–η•Œ πŸš€\nEmoji: πŸŽ‰\n" + log_file = temp_log_files("unicode.log", content) + + lines = list(read_logs(str(log_file))) + assert len(lines) == 2 + assert "δ½ ε₯½δΈ–η•Œ" in lines[0] + assert "πŸš€" in lines[0] + assert "πŸŽ‰" in lines[1] + + def test_read_logs_large_file(self, temp_log_files): + """Test reading a large file with many lines.""" + content = "\n".join([f"Line {i}" for i in range(1000)]) + log_file = temp_log_files("large.log", content) + + lines = list(read_logs(str(log_file))) + assert len(lines) == 1000 + assert lines[0] == "Line 0" + assert lines[999] == "Line 999" + + def test_read_logs_is_iterator(self, temp_log_files): + """Test that read_logs returns a lazy iterator.""" + content = "Line 1\nLine 2\nLine 3\n" + log_file = temp_log_files("test.log", content) + + lines_iter = read_logs(str(log_file)) + assert next(lines_iter) == "Line 1" + assert next(lines_iter) == "Line 2" + assert next(lines_iter) == "Line 3" + + with pytest.raises(StopIteration): + next(lines_iter) + + def test_read_logs_nonexistent_file(self): + """Test reading a nonexistent file raises FileNotFoundError.""" + with pytest.raises(FileNotFoundError, match="No such file"): + list(read_logs("/nonexistent/path/to/file.log")) + + def test_read_logs_directory_path(self, tmp_path): + """Test reading a directory path raises FileNotFoundError.""" + with pytest.raises(FileNotFoundError, match="No such file"): + list(read_logs(str(tmp_path))) + + def test_read_logs_windows_line_endings(self, temp_log_files): + """Test reading file with Windows line endings (CRLF).""" + # Note: write_text automatically handles this, but we can test the behavior + content = "Line 1\r\nLine 2\r\n" + log_file = temp_log_files("windows.log", content) + + lines = list(read_logs(str(log_file))) + # The result depends on how the OS handles line endings when written + # Just verify we can read it without errors + assert len(lines) >= 2 + + +@pytest.mark.unit +class TestIterLogFilesSingleFile: + """Tests for iter_log_files with a single file path.""" + + def test_iter_single_file(self, temp_log_files): + """Test iterating over a single file path.""" + log_file = temp_log_files("test.log", "content") + + files = list(iter_log_files(str(log_file))) + assert len(files) == 1 + assert files[0] == str(log_file) + + def test_iter_single_file_with_include_match(self, temp_log_files): + """Test single file with matching include pattern.""" + log_file = temp_log_files("test.log", "content") + + files = list(iter_log_files(str(log_file), include=["*.log"])) + assert len(files) == 1 + + def test_iter_single_file_with_include_no_match(self, temp_log_files): + """Test single file with non-matching include pattern.""" + log_file = temp_log_files("test.log", "content") + + files = list(iter_log_files(str(log_file), include=["*.txt"])) + assert len(files) == 0 + + def test_iter_single_file_with_exclude(self, temp_log_files): + """Test single file with exclude pattern.""" + log_file = temp_log_files("test.log", "content") + + files = list(iter_log_files(str(log_file), exclude=["*.log"])) + assert len(files) == 0 + + def test_iter_single_file_include_and_exclude(self, temp_log_files): + """Test single file with both include and exclude patterns.""" + log_file = temp_log_files("test.log", "content") + + # Include matches but exclude also matches - should be excluded + files = list( + iter_log_files(str(log_file), include=["*.log"], exclude=["test.*"]) + ) + assert len(files) == 0 + + +# ============================================================================ +# iter_log_files Tests - Directory Non-Recursive +# ============================================================================ + + +@pytest.mark.unit +class TestIterLogFilesDirectoryNonRecursive: + """Tests for iter_log_files with directory (non-recursive mode).""" + + def test_iter_directory_non_recursive(self, sample_log_file_structure): + """Test iterating over directory without recursion.""" + root = sample_log_file_structure["root"] + + files = list(iter_log_files(str(root), recursive=False)) + filenames = [os.path.basename(f) for f in files] + + # Should include root-level files only + assert "app.log" in filenames + assert "error.log" in filenames + assert "data.txt" in filenames + # Should NOT include subdirectory files + assert "access.log" not in filenames + assert "deep.log" not in filenames + + def test_iter_directory_with_log_filter(self, sample_log_file_structure): + """Test iterating directory with .log filter.""" + root = sample_log_file_structure["root"] + + files = list(iter_log_files(str(root), recursive=False, include=["*.log"])) + filenames = [os.path.basename(f) for f in files] + + assert "app.log" in filenames + assert "error.log" in filenames + assert "data.txt" not in filenames # Excluded by include pattern + + def test_iter_directory_with_exclude(self, sample_log_file_structure): + """Test iterating directory with exclude pattern.""" + root = sample_log_file_structure["root"] + + files = list(iter_log_files(str(root), recursive=False, exclude=["error.*"])) + filenames = [os.path.basename(f) for f in files] + + assert "app.log" in filenames + assert "data.txt" in filenames + assert "error.log" not in filenames + + def test_iter_directory_multiple_include_patterns(self, sample_log_file_structure): + """Test iterating directory with multiple include patterns.""" + root = sample_log_file_structure["root"] + + files = list( + iter_log_files(str(root), recursive=False, include=["*.log", "*.txt"]) + ) + filenames = [os.path.basename(f) for f in files] + + assert len(filenames) == 3 + assert "app.log" in filenames + assert "error.log" in filenames + assert "data.txt" in filenames + + def test_iter_directory_files_sorted(self, sample_log_file_structure): + """Test that files are returned in sorted order.""" + root = sample_log_file_structure["root"] + + files = list(iter_log_files(str(root), recursive=False)) + filenames = [os.path.basename(f) for f in files] + + # Check if sorted (app.log, data.txt, error.log alphabetically) + assert filenames == sorted(filenames) + + def test_iter_empty_directory(self, tmp_path): + """Test iterating over an empty directory.""" + files = list(iter_log_files(str(tmp_path), recursive=False)) + assert len(files) == 0 + + +# ============================================================================ +# iter_log_files Tests - Directory Recursive +# ============================================================================ + + +@pytest.mark.unit +class TestIterLogFilesDirectoryRecursive: + """Tests for iter_log_files with directory (recursive mode).""" + + def test_iter_directory_recursive(self, sample_log_file_structure): + """Test iterating over directory with recursion.""" + root = sample_log_file_structure["root"] + + files = list(iter_log_files(str(root), recursive=True)) + + # Get basenames for easier assertion + filenames = [os.path.basename(f) for f in files] + + # Should include all files + assert "app.log" in filenames + assert "error.log" in filenames + assert "data.txt" in filenames + assert "access.log" in filenames + assert "debug.log" in filenames + assert "deep.log" in filenames + + def test_iter_directory_recursive_with_filter(self, sample_log_file_structure): + """Test recursive iteration with .log filter.""" + root = sample_log_file_structure["root"] + + files = list(iter_log_files(str(root), recursive=True, include=["*.log"])) + filenames = [os.path.basename(f) for f in files] + + assert "app.log" in filenames + assert "error.log" in filenames + assert "access.log" in filenames + assert "debug.log" in filenames + assert "deep.log" in filenames + assert "data.txt" not in filenames + + def test_iter_directory_recursive_with_path_pattern( + self, sample_log_file_structure + ): + """Test recursive iteration with path-based pattern.""" + root = sample_log_file_structure["root"] + + # Match only files in the 'logs' subdirectory + files = list(iter_log_files(str(root), recursive=True, include=["logs/*.log"])) + filenames = [os.path.basename(f) for f in files] + + # Should include logs in the logs directory + assert "access.log" in filenames + assert "debug.log" in filenames + # Should NOT include root level logs + assert "app.log" not in filenames + assert "error.log" not in filenames + + def test_iter_directory_recursive_exclude_nested(self, sample_log_file_structure): + """Test recursive iteration excluding nested directory.""" + root = sample_log_file_structure["root"] + + files = list(iter_log_files(str(root), recursive=True, exclude=["*/nested/*"])) + filenames = [os.path.basename(f) for f in files] + + assert "app.log" in filenames + assert "access.log" in filenames + assert "deep.log" not in filenames # In nested directory + + def test_iter_directory_recursive_sorted(self, sample_log_file_structure): + """Test that recursive files are returned in sorted order.""" + root = sample_log_file_structure["root"] + + files = list(iter_log_files(str(root), recursive=True)) + + # Files should be sorted + assert files == sorted(files) + + +# ============================================================================ +# iter_log_files Tests - Edge Cases +# ============================================================================ + + +@pytest.mark.unit +class TestIterLogFilesEdgeCases: + """Tests for edge cases in iter_log_files.""" + + def test_iter_nonexistent_path(self): + """Test iterating nonexistent path raises FileNotFoundError.""" + with pytest.raises(FileNotFoundError, match="No such file or directory"): + list(iter_log_files("/nonexistent/path")) + + def test_iter_with_wildcard_patterns(self, sample_log_file_structure): + """Test with various wildcard patterns.""" + root = sample_log_file_structure["root"] + + # Match files starting with 'a' + files = list(iter_log_files(str(root), recursive=True, include=["a*"])) + filenames = [os.path.basename(f) for f in files] + assert "app.log" in filenames + assert "access.log" in filenames + assert "error.log" not in filenames + + def test_iter_include_default_matches_all(self, sample_log_file_structure): + """Test that default include pattern matches all files.""" + root = sample_log_file_structure["root"] + + # Default include should be ['*'] + files_default = list(iter_log_files(str(root), recursive=False)) + files_explicit = list(iter_log_files(str(root), recursive=False, include=["*"])) + + assert len(files_default) == len(files_explicit) + + def test_iter_exclude_takes_precedence(self, temp_log_files): + """Test that exclude takes precedence over include.""" + log_file = temp_log_files("test.log", "content") + + # Include matches, but exclude also matches + files = list( + iter_log_files(str(log_file), include=["*.log"], exclude=["*.log"]) + ) + assert len(files) == 0 + + def test_iter_pattern_matching_name_and_path(self, sample_log_file_structure): + """Test that patterns match both filename and relative path.""" + root = sample_log_file_structure["root"] + + # Pattern should match the filename 'deep.log' + files = list(iter_log_files(str(root), recursive=True, include=["deep.log"])) + assert len(files) == 1 + assert os.path.basename(files[0]) == "deep.log" + + +# ============================================================================ +# read_logs_from_paths Tests +# ============================================================================ + + +@pytest.mark.unit +class TestReadLogsFromPaths: + """Tests for the read_logs_from_paths function.""" + + def test_read_logs_from_multiple_files(self, temp_log_files): + """Test reading logs from multiple files.""" + file1 = temp_log_files("file1.log", "Line 1\nLine 2\n") + file2 = temp_log_files("file2.log", "Line 3\nLine 4\n") + + lines = list(read_logs_from_paths([str(file1), str(file2)])) + + assert len(lines) == 4 + assert lines[0] == "Line 1" + assert lines[1] == "Line 2" + assert lines[2] == "Line 3" + assert lines[3] == "Line 4" + + def test_read_logs_from_paths_preserves_order(self, temp_log_files): + """Test that files are read in the order provided.""" + file1 = temp_log_files("a.log", "From A\n") + file2 = temp_log_files("b.log", "From B\n") + file3 = temp_log_files("c.log", "From C\n") + + # Read in specific order + lines = list(read_logs_from_paths([str(file3), str(file1), str(file2)])) + + assert lines[0] == "From C" + assert lines[1] == "From A" + assert lines[2] == "From B" + + def test_read_logs_from_empty_paths_list(self): + """Test reading from empty paths list.""" + lines = list(read_logs_from_paths([])) + assert len(lines) == 0 + + def test_read_logs_from_single_file(self, temp_log_files): + """Test reading from a single file path.""" + log_file = temp_log_files("test.log", "Line 1\nLine 2\n") + + lines = list(read_logs_from_paths([str(log_file)])) + + assert len(lines) == 2 + assert lines[0] == "Line 1" + assert lines[1] == "Line 2" + + def test_read_logs_from_paths_is_iterator(self, temp_log_files): + """Test that read_logs_from_paths returns a lazy iterator.""" + file1 = temp_log_files("file1.log", "Line 1\n") + file2 = temp_log_files("file2.log", "Line 2\n") + + lines_iter = read_logs_from_paths([str(file1), str(file2)]) + + assert next(lines_iter) == "Line 1" + assert next(lines_iter) == "Line 2" + + with pytest.raises(StopIteration): + next(lines_iter) + + def test_read_logs_from_paths_with_empty_file(self, temp_log_files): + """Test reading from paths including an empty file.""" + file1 = temp_log_files("file1.log", "Line 1\n") + file2 = temp_log_files("empty.log", "") + file3 = temp_log_files("file3.log", "Line 3\n") + + lines = list(read_logs_from_paths([str(file1), str(file2), str(file3)])) + + assert len(lines) == 2 + assert lines[0] == "Line 1" + assert lines[1] == "Line 3" + + def test_read_logs_from_paths_integration_with_iter_log_files( + self, sample_log_file_structure + ): + """Test integration: use iter_log_files output as input to + read_logs_from_paths.""" + root = sample_log_file_structure["root"] + + # Get all .log files recursively + file_paths = list(iter_log_files(str(root), recursive=True, include=["*.log"])) + + # Read all lines from those files + lines = list(read_logs_from_paths(file_paths)) + + # Should have lines from all log files + assert len(lines) > 0 + # Verify some expected content + assert any("Test log 1" in line for line in lines) + assert any("Error occurred" in line for line in lines) diff --git a/tests/test_llm_client.py b/tests/test_llm_client.py new file mode 100644 index 0000000..7e1aee6 --- /dev/null +++ b/tests/test_llm_client.py @@ -0,0 +1,121 @@ +"""Unit tests for llm_client module.""" + +import os +import subprocess +import sys + +import pytest + +# noqa: S603, S607 - Suppress bandit warnings for test code with hardcoded strings +from logpilot.llm_client import LLMClient + +# ============================================================================ +# Unit Tests +# ============================================================================ + + +@pytest.mark.unit +class TestLLMClient: + """Tests for LLMClient initialization and properties.""" + + def test_llm_client_initialization(self): + """Test LLMClient initializes with correct model.""" + client = LLMClient() + assert client.model == "gpt-4" + + def test_llm_client_custom_model(self): + """Test LLMClient can accept custom model.""" + client = LLMClient(model="gpt-3.5-turbo") + assert client.model == "gpt-3.5-turbo" + + def test_llm_client_default_model(self): + """Test default model is gpt-4.""" + client = LLMClient() + assert client.model == "gpt-4" + + def test_llm_client_model_parameter(self): + """Test different model parameters are stored correctly.""" + models = ["gpt-4", "gpt-3.5-turbo", "custom-model"] + for model in models: + client = LLMClient(model=model) + assert client.model == model + + def test_llm_client_has_analyze_method(self): + """Test LLMClient has analyze method.""" + client = LLMClient() + assert hasattr(client, "analyze") + assert callable(client.analyze) + + def test_llm_client_has_analyze_async_method(self): + """Test LLMClient has _analyze_async method.""" + client = LLMClient() + assert hasattr(client, "_analyze_async") + assert callable(client._analyze_async) + + +@pytest.mark.unit +def test_llm_client_analyze_with_mock(): + """Test analyze method works with mock LLM.""" + # Since we're already loading with LOGPILOT_MOCK_LLM or not, + # we test what we have + client = LLMClient() + # This will use either the mock or real CopilotClient depending on env + # Just verify it doesn't crash and returns a string + # (Can't test real flow without actual copilot SDK) + assert isinstance(client, LLMClient) + + +@pytest.mark.unit +def test_llm_client_properties(): + """Test LLMClient has required properties and methods.""" + client = LLMClient() + assert hasattr(client, "model") + assert hasattr(client, "analyze") + assert callable(client.analyze) + assert hasattr(client, "_analyze_async") + assert callable(client._analyze_async) + + +# ============================================================================ +# Integration Tests +# ============================================================================ + + +@pytest.mark.integration +def test_llm_client_via_subprocess(): + """Test llm_client with mock enabled via subprocess.""" + code = """ +import sys +from logpilot.llm_client import LLMClient +client = LLMClient() +try: + result = client.analyze("test prompt") + print(f"SUCCESS: analyze() executed") + sys.exit(0) +except TypeError as e: + # Mock has a bug with the Event class instantiation + # But the code path was at least attempted + print(f"EXPECTED_ERROR: {e}") + sys.exit(0) +except Exception as e: + print(f"ERROR: {e}") + sys.exit(1) +""" + env = dict(os.environ) + env["LOGPILOT_MOCK_LLM"] = "1" + result = subprocess.run( # noqa: S603 + [sys.executable, "-c", code], + capture_output=True, + text=True, + env=env, + ) + # Test verifies code path was attempted (even if mock has bugs) + assert result.returncode == 0 + + +def test_llm_client_model_parameter(): + """Test different model parameters are stored correctly.""" + models = ["gpt-4", "gpt-3.5-turbo", "custom-model"] + for model in models: + client = LLMClient(model=model) + assert client.model == model diff --git a/tests/test_log_parser.py b/tests/test_log_parser.py new file mode 100644 index 0000000..36a3e10 --- /dev/null +++ b/tests/test_log_parser.py @@ -0,0 +1,385 @@ +"""Unit tests for log_parser module.""" + +import pytest + +from logpilot.log_parser import LogEntry, parse_log_line, parse_logs + +# ============================================================================ +# LogEntry Class Tests +# ============================================================================ + + +@pytest.mark.unit +class TestLogEntry: + """Tests for the LogEntry class.""" + + def test_log_entry_initialization(self): + """Test LogEntry can be initialized with all fields.""" + entry = LogEntry( + timestamp="2024-01-01T10:00:00Z", + level="INFO", + source="app.main", + message="Test message", + raw='{"timestamp":"2024-01-01T10:00:00Z"}', + ) + assert entry.timestamp == "2024-01-01T10:00:00Z" + assert entry.level == "INFO" + assert entry.source == "app.main" + assert entry.message == "Test message" + assert entry.raw == '{"timestamp":"2024-01-01T10:00:00Z"}' + + def test_log_entry_with_none_values(self): + """Test LogEntry can be initialized with None values for optional fields.""" + entry = LogEntry( + timestamp=None, level=None, source=None, message="Message only", raw="raw" + ) + assert entry.timestamp is None + assert entry.level is None + assert entry.source is None + assert entry.message == "Message only" + assert entry.raw == "raw" + + +# ============================================================================ +# parse_log_line JSON Format Tests +# ============================================================================ + + +@pytest.mark.unit +class TestParseLogLineJSON: + """Tests for parse_log_line with JSON format.""" + + def test_parse_valid_json_log(self, sample_json_logs): + """Test parsing a valid JSON log line.""" + entry = parse_log_line(sample_json_logs[0], fmt="json") + assert entry is not None + assert entry.timestamp == "2024-01-01T10:00:00Z" + assert entry.level == "INFO" + assert entry.source == "app.main" + assert entry.message == "Application started" + assert entry.raw == sample_json_logs[0] + + def test_parse_json_with_all_fields(self): + """Test parsing JSON with all expected fields.""" + json_line = ( + '{"timestamp":"2024-02-09","level":"ERROR",' + '"source":"test","message":"Error occurred"}' + ) + entry = parse_log_line(json_line, fmt="json") + assert entry is not None + assert entry.timestamp == "2024-02-09" + assert entry.level == "ERROR" + assert entry.source == "test" + assert entry.message == "Error occurred" + + def test_parse_json_missing_optional_fields(self): + """Test parsing JSON with missing optional fields (timestamp, level, source).""" + json_line = '{"message":"Just a message"}' + entry = parse_log_line(json_line, fmt="json") + assert entry is not None + assert entry.timestamp is None + assert entry.level is None + assert entry.source is None + assert entry.message == "Just a message" + + def test_parse_json_missing_message_field(self): + """Test parsing JSON without a message field uses str(data) as message.""" + json_line = '{"timestamp":"2024-01-01","level":"INFO"}' + entry = parse_log_line(json_line, fmt="json") + assert entry is not None + assert "timestamp" in entry.message + assert "level" in entry.message + + def test_parse_json_with_extra_fields(self): + """Test parsing JSON with extra fields (they are ignored).""" + json_line = ( + '{"timestamp":"2024-01-01","level":"INFO","source":"app",' + '"message":"Test","extra":"field","another":123}' + ) + entry = parse_log_line(json_line, fmt="json") + assert entry is not None + assert entry.message == "Test" + + def test_parse_malformed_json(self, malformed_logs): + """Test parsing malformed JSON returns None.""" + entry = parse_log_line(malformed_logs["malformed_json"], fmt="json") + assert entry is None + + def test_parse_json_array(self, malformed_logs): + """Test parsing JSON array returns None.""" + entry = parse_log_line(malformed_logs["json_array"], fmt="json") + assert entry is None + + def test_parse_empty_json_object(self): + """Test parsing empty JSON object.""" + entry = parse_log_line("{}", fmt="json") + assert entry is not None + assert entry.message == "{}" # Uses str(data) when message is missing + + def test_parse_json_with_unicode(self): + """Test parsing JSON with unicode characters.""" + json_line = '{"message":"Unicode test: δ½ ε₯½δΈ–η•Œ πŸš€"}' + entry = parse_log_line(json_line, fmt="json") + assert entry is not None + assert "δ½ ε₯½δΈ–η•Œ" in entry.message + assert "πŸš€" in entry.message + + def test_parse_json_with_escaped_characters(self): + """Test parsing JSON with escaped characters.""" + json_line = '{"message":"Line 1\\nLine 2\\tTabbed"}' + entry = parse_log_line(json_line, fmt="json") + assert entry is not None + assert "Line 1" in entry.message + + +# ============================================================================ +# parse_log_line Text Format Tests +# ============================================================================ + + +@pytest.mark.unit +class TestParseLogLineText: + """Tests for parse_log_line with text/CSV format.""" + + def test_parse_valid_text_log(self, sample_text_logs): + """Test parsing a valid CSV-style text log.""" + entry = parse_log_line(sample_text_logs[0], fmt="text") + assert entry is not None + assert entry.timestamp is None # Text format doesn't parse fields + assert entry.level is None + assert entry.source is None + assert entry.message == sample_text_logs[0] + assert entry.raw == sample_text_logs[0] + + def test_parse_text_requires_two_commas(self): + """Test text format requires at least 2 commas to be valid.""" + # 0 commas - invalid + entry = parse_log_line("No commas here", fmt="text") + assert entry is None + + # 1 comma - invalid + entry = parse_log_line("One,comma", fmt="text") + assert entry is None + + # 2 commas - valid + entry = parse_log_line("Two,commas,here", fmt="text") + assert entry is not None + + # 3+ commas - valid + entry = parse_log_line("Many,commas,in,this,line", fmt="text") + assert entry is not None + + def test_parse_text_with_many_fields(self): + """Test parsing text log with many comma-separated fields.""" + text_line = "2024-01-01,INFO,app.main,User logged in,user_id=123,session=abc" + entry = parse_log_line(text_line, fmt="text") + assert entry is not None + assert entry.message == text_line + + def test_parse_text_with_unicode(self, malformed_logs): + """Test parsing text with unicode characters.""" + entry = parse_log_line(malformed_logs["unicode"], fmt="text") + assert entry is not None + assert "δ½ ε₯½δΈ–η•Œ" in entry.message + + +# ============================================================================ +# parse_log_line Auto Format Detection Tests +# ============================================================================ + + +@pytest.mark.unit +class TestParseLogLineAuto: + """Tests for parse_log_line with automatic format detection.""" + + def test_auto_detects_json(self, sample_json_logs): + """Test auto format detects and parses JSON.""" + entry = parse_log_line(sample_json_logs[0], fmt="auto") + assert entry is not None + assert entry.timestamp == "2024-01-01T10:00:00Z" + assert entry.level == "INFO" + + def test_auto_detects_text(self, sample_text_logs): + """Test auto format detects and parses text.""" + entry = parse_log_line(sample_text_logs[0], fmt="auto") + assert entry is not None + assert entry.message == sample_text_logs[0] + + def test_auto_with_json_whitespace(self): + """Test auto detection works with leading whitespace before JSON.""" + json_line = " \t " + '{"message":"test"}' + entry = parse_log_line(json_line, fmt="auto") + assert entry is not None + assert entry.message == "test" + + def test_auto_rejects_invalid_formats(self, malformed_logs): + """Test auto format rejects logs that don't match either format.""" + # No commas and not JSON + entry = parse_log_line(malformed_logs["text_no_commas"], fmt="auto") + assert entry is None + + # Only one comma + entry = parse_log_line(malformed_logs["text_one_comma"], fmt="auto") + assert entry is None + + +# ============================================================================ +# parse_log_line Edge Cases +# ============================================================================ + + +@pytest.mark.unit +class TestParseLogLineEdgeCases: + """Tests for edge cases in parse_log_line.""" + + def test_parse_empty_string(self, malformed_logs): + """Test parsing empty string returns None.""" + entry = parse_log_line(malformed_logs["empty"]) + assert entry is None + + def test_parse_whitespace_only(self, malformed_logs): + """Test parsing whitespace-only string returns None.""" + entry = parse_log_line(malformed_logs["whitespace_only"]) + assert entry is None + + def test_parse_very_long_line(self, malformed_logs): + """Test parsing very long line (10K+ chars).""" + entry = parse_log_line(malformed_logs["very_long"], fmt="text") + assert entry is not None + assert len(entry.message) > 10000 + + def test_parse_line_with_newline_characters(self): + """Test parsing line with embedded newline characters.""" + # In JSON, newlines are escaped + json_line = '{"message":"Line 1\\nLine 2"}' + entry = parse_log_line(json_line, fmt="json") + assert entry is not None + + # In text, actual newlines in the string + text_line = "A,B,C,with,embedded,newline" + entry = parse_log_line(text_line, fmt="text") + assert entry is not None + + def test_parse_line_with_special_characters(self): + """Test parsing line with various special characters.""" + special_chars = '{"message":"Special: !@#$%^&*()[]{}|\\\\/\'\\""}' + entry = parse_log_line(special_chars, fmt="json") + assert entry is not None + + def test_parse_with_trailing_whitespace(self): + """Test parsing handles trailing whitespace.""" + json_line = '{"message":"test"} \n\t ' + entry = parse_log_line(json_line, fmt="json") + assert entry is not None + assert entry.message == "test" + + def test_parse_text_exactly_two_commas(self): + """Test edge case: exactly 2 commas (minimum for text format).""" + entry = parse_log_line("A,B,C", fmt="text") + assert entry is not None + assert entry.message == "A,B,C" + + +# ============================================================================ +# parse_logs Iterator Tests +# ============================================================================ + + +@pytest.mark.unit +class TestParseLogs: + """Tests for the parse_logs iterator function.""" + + def test_parse_logs_with_valid_json_lines(self, sample_json_logs): + """Test parsing multiple valid JSON log lines.""" + entries = list(parse_logs(iter(sample_json_logs), fmt="json")) + assert len(entries) == 4 + assert entries[0].level == "INFO" + assert entries[1].level == "WARNING" + assert entries[2].level == "ERROR" + assert entries[3].level == "DEBUG" + + def test_parse_logs_with_valid_text_lines(self, sample_text_logs): + """Test parsing multiple valid text log lines.""" + entries = list(parse_logs(iter(sample_text_logs), fmt="text")) + assert len(entries) == 4 + assert all(entry.message is not None for entry in entries) + + def test_parse_logs_with_mixed_valid_invalid(self): + """Test parsing mixed valid and invalid lines (invalid are skipped).""" + lines = [ + '{"message":"valid json"}', + "invalid line", + '{"message":"another valid"}', + "", + '{"message":"third valid"}', + ] + entries = list(parse_logs(iter(lines), fmt="json")) + assert len(entries) == 3 + assert entries[0].message == "valid json" + assert entries[1].message == "another valid" + assert entries[2].message == "third valid" + + def test_parse_logs_empty_iterator(self): + """Test parsing empty iterator returns no entries.""" + entries = list(parse_logs(iter([]), fmt="json")) + assert len(entries) == 0 + + def test_parse_logs_all_invalid_lines(self): + """Test parsing all invalid lines returns no entries.""" + lines = ["invalid", "no commas", "single,comma"] + entries = list(parse_logs(iter(lines), fmt="text")) + assert len(entries) == 0 + + def test_parse_logs_is_lazy(self): + """Test that parse_logs is a lazy iterator (doesn't consume all at once).""" + lines = iter(['{"message":"test1"}', '{"message":"test2"}']) + entries_iter = parse_logs(lines, fmt="json") + + # Get first entry + first = next(entries_iter) + assert first.message == "test1" + + # Get second entry + second = next(entries_iter) + assert second.message == "test2" + + # No more entries + with pytest.raises(StopIteration): + next(entries_iter) + + def test_parse_logs_auto_format(self, sample_json_logs, sample_text_logs): + """Test parse_logs with auto format detection.""" + # Mix JSON and text logs + mixed_lines = [ + sample_json_logs[0], + sample_text_logs[0], + sample_json_logs[1], + ] + entries = list(parse_logs(iter(mixed_lines), fmt="auto")) + assert len(entries) == 3 + + def test_parse_logs_preserves_order(self): + """Test that parse_logs preserves the order of entries.""" + lines = [ + '{"message":"first"}', + '{"message":"second"}', + '{"message":"third"}', + ] + entries = list(parse_logs(iter(lines), fmt="json")) + assert entries[0].message == "first" + assert entries[1].message == "second" + assert entries[2].message == "third" + + def test_parse_logs_with_whitespace_lines(self): + """Test parsing logs with whitespace-only lines (should be skipped).""" + lines = [ + '{"message":"test1"}', + " ", + "\t\n", + '{"message":"test2"}', + "", + ] + entries = list(parse_logs(iter(lines), fmt="json")) + assert len(entries) == 2 + assert entries[0].message == "test1" + assert entries[1].message == "test2" diff --git a/tests/test_postprocessor.py b/tests/test_postprocessor.py new file mode 100644 index 0000000..6a95681 --- /dev/null +++ b/tests/test_postprocessor.py @@ -0,0 +1,146 @@ +"""Unit tests for postprocessor module.""" + +import pytest + +from logpilot.postprocessor import aggregate_responses + + +@pytest.mark.unit +class TestAggregateResponses: + """Tests for the aggregate_responses function.""" + + def test_aggregate_single_response(self): + """Test aggregating a single response.""" + responses = ["Response 1"] + result = aggregate_responses(responses) + + assert result == "Response 1" + + def test_aggregate_multiple_responses(self): + """Test aggregating multiple responses.""" + responses = ["Response 1", "Response 2", "Response 3"] + result = aggregate_responses(responses) + + expected = "Response 1\n---\n" + "Response 2\n---\n" + "Response 3" + assert result == expected + + def test_aggregate_two_responses(self): + """Test aggregating exactly two responses.""" + responses = ["First", "Second"] + result = aggregate_responses(responses) + + assert result == "First\n---\nSecond" + assert result.count("---") == 1 + + def test_aggregate_empty_list(self): + """Test aggregating an empty list of responses.""" + responses = [] + result = aggregate_responses(responses) + + assert result == "" + + def test_aggregate_responses_with_newlines(self): + """Test aggregating responses that contain newlines.""" + responses = [ + "Response 1\nLine 2", + "Response 2\nLine 2\nLine 3", + "Response 3", + ] + result = aggregate_responses(responses) + + # Verify all responses are included + assert "Response 1\nLine 2" in result + assert "Response 2\nLine 2\nLine 3" in result + assert "Response 3" in result + # Verify separator is present + assert result.count("\n---\n") == 2 + + def test_aggregate_responses_with_empty_strings(self): + """Test aggregating responses with empty strings.""" + responses = ["Response 1", "", "Response 3"] + result = aggregate_responses(responses) + + assert result == "Response 1\n---\n\n---\nResponse 3" + # Empty response should still create a section + + def test_aggregate_responses_all_empty(self): + """Test aggregating all empty responses.""" + responses = ["", "", ""] + result = aggregate_responses(responses) + + assert result == "\n---\n\n---\n" + + def test_aggregate_responses_with_separator_in_content(self): + """Test aggregating when responses already contain '---'.""" + responses = ["Response with --- in it", "Normal response"] + result = aggregate_responses(responses) + + # The function should still add separators + # Count total occurrences of '---' + assert result.count("---") == 2 # One in content, one as separator + + def test_aggregate_responses_preserves_order(self): + """Test that aggregation preserves the order of responses.""" + responses = ["First", "Second", "Third", "Fourth"] + result = aggregate_responses(responses) + + # Find positions to verify order + first_pos = result.find("First") + second_pos = result.find("Second") + third_pos = result.find("Third") + fourth_pos = result.find("Fourth") + + assert first_pos < second_pos < third_pos < fourth_pos + + def test_aggregate_responses_with_unicode(self): + """Test aggregating responses with unicode characters.""" + responses = ["Response with δ½ ε₯½", "Response with πŸš€"] + result = aggregate_responses(responses) + + assert "δ½ ε₯½" in result + assert "πŸš€" in result + assert "\n---\n" in result + + def test_aggregate_responses_with_special_characters(self): + """Test aggregating responses with special characters.""" + responses = [ + "Response with !@#$%", + "Response with ^&*()", + "Response with {}[]|\\", + ] + result = aggregate_responses(responses) + + assert "!@#$%" in result + assert "^&*()" in result + assert "{}[]|\\" in result + + def test_aggregate_large_number_of_responses(self): + """Test aggregating a large number of responses.""" + responses = [f"Response {i}" for i in range(100)] + result = aggregate_responses(responses) + + # Verify all responses are present + for i in range(100): + assert f"Response {i}" in result + + # Verify separator count (should be 99 separators for 100 responses) + assert result.count("\n---\n") == 99 + + def test_aggregate_responses_with_very_long_responses(self): + """Test aggregating very long responses.""" + long_response = "x" * 10000 + responses = [long_response, "Short", long_response] + result = aggregate_responses(responses) + + assert len(result) > 20000 + assert result.count("\n---\n") == 2 + + def test_aggregate_responses_separator_format(self): + """Test that the separator format is exactly '\\n---\\n'.""" + responses = ["A", "B"] + result = aggregate_responses(responses) + + # Verify the exact separator format + assert result == "A\n---\nB" + # Verify no extra spaces or characters + assert "A\n---\nB" == result diff --git a/tests/test_prompt_engine.py b/tests/test_prompt_engine.py new file mode 100644 index 0000000..a9a5281 --- /dev/null +++ b/tests/test_prompt_engine.py @@ -0,0 +1,143 @@ +"""Unit tests for prompt_engine module.""" + +import pytest + +from logpilot.log_parser import LogEntry +from logpilot.prompt_engine import PROMPT_TEMPLATE, format_prompt + + +@pytest.mark.unit +class TestFormatPrompt: + """Tests for the format_prompt function.""" + + def test_format_prompt_with_sample_entries(self, sample_log_entries): + """Test formatting prompt with sample log entries.""" + prompt = format_prompt(sample_log_entries) + + # Verify template structure is present + assert "You are analyzing application logs." in prompt + assert "Logs:" in prompt + assert "Tasks:" in prompt + assert "Identify critical issues" in prompt + + # Verify all log raw data is included + for entry in sample_log_entries: + assert entry.raw in prompt + + def test_format_prompt_single_entry(self): + """Test formatting prompt with a single entry.""" + entry = LogEntry( + timestamp="2024-01-01", + level="ERROR", + source="test", + message="Error occurred", + raw='{"timestamp":"2024-01-01","level":"ERROR","message":"Error occurred"}', + ) + + prompt = format_prompt([entry]) + + assert entry.raw in prompt + assert "You are analyzing application logs." in prompt + + def test_format_prompt_empty_chunk(self): + """Test formatting prompt with empty chunk.""" + prompt = format_prompt([]) + + # Should still have the template structure + assert "You are analyzing application logs." in prompt + assert "Logs:" in prompt + # The chunk section should be empty + assert prompt.count("\n\n") >= 1 + + def test_format_prompt_preserves_order(self): + """Test that format_prompt preserves the order of entries.""" + entries = [ + LogEntry(None, None, None, "First", "First log"), + LogEntry(None, None, None, "Second", "Second log"), + LogEntry(None, None, None, "Third", "Third log"), + ] + + prompt = format_prompt(entries) + + # Find positions of each log in the prompt + first_pos = prompt.find("First log") + second_pos = prompt.find("Second log") + third_pos = prompt.find("Third log") + + assert first_pos < second_pos < third_pos + + def test_format_prompt_with_special_characters(self): + """Test formatting prompt with special characters in logs.""" + entry = LogEntry( + None, + None, + None, + "Special chars", + "Log with special chars: !@#$%^&*(){}[]|\\", + ) + + prompt = format_prompt([entry]) + + assert entry.raw in prompt + assert "!@#$%^&*()" in prompt + + def test_format_prompt_with_unicode(self): + """Test formatting prompt with unicode characters.""" + entry = LogEntry(None, None, None, "Unicode", "Unicode log: δ½ ε₯½δΈ–η•Œ πŸš€") + + prompt = format_prompt([entry]) + + assert "δ½ ε₯½δΈ–η•Œ" in prompt + assert "πŸš€" in prompt + + def test_format_prompt_with_multiline_raw(self): + """Test formatting prompt with entries containing newlines.""" + entry = LogEntry(None, None, None, "Multiline", "Line 1\nLine 2\nLine 3") + + prompt = format_prompt([entry]) + + assert "Line 1" in prompt + assert "Line 2" in prompt + assert "Line 3" in prompt + + def test_format_prompt_uses_template(self): + """Test that format_prompt uses the PROMPT_TEMPLATE constant.""" + entry = LogEntry(None, None, None, "Test", "Test log") + + prompt = format_prompt([entry]) + + # Verify all parts of the template are present + assert "You are analyzing application logs." in prompt + assert "1. Identify critical issues" in prompt + assert "2. Explain likely causes" in prompt + assert "3. Suggest next debugging steps" in prompt + + def test_format_prompt_entries_separated_by_newlines(self): + """Test that entries are separated by newlines in the prompt.""" + entries = [ + LogEntry(None, None, None, "Log 1", "Raw 1"), + LogEntry(None, None, None, "Log 2", "Raw 2"), + LogEntry(None, None, None, "Log 3", "Raw 3"), + ] + + prompt = format_prompt(entries) + + # The chunk should have entries joined by newlines + assert "Raw 1\nRaw 2\nRaw 3" in prompt + + def test_format_prompt_with_very_long_entries(self): + """Test formatting prompt with very long log entries.""" + long_raw = "x" * 10000 + entry = LogEntry(None, None, None, "Long", long_raw) + + prompt = format_prompt([entry]) + + assert long_raw in prompt + assert len(prompt) > 10000 + + def test_prompt_template_constant(self): + """Test that PROMPT_TEMPLATE is properly defined.""" + assert isinstance(PROMPT_TEMPLATE, str) + assert "{chunk}" in PROMPT_TEMPLATE + assert "analyzing" in PROMPT_TEMPLATE.lower() + assert "logs" in PROMPT_TEMPLATE.lower() diff --git a/tests/test_utils.py b/tests/test_utils.py new file mode 100644 index 0000000..6a8ee60 --- /dev/null +++ b/tests/test_utils.py @@ -0,0 +1,166 @@ +"""Unit tests for utils module.""" + +import pytest + +from logpilot.utils import space_indent, tab_indent + + +@pytest.mark.unit +class TestSpaceIndent: + """Tests for the space_indent function.""" + + def test_space_indent_single_line(self): + """Test indenting a single line.""" + text = "Single line" + result = space_indent(text) + + assert result == " Single line" + + def test_space_indent_multiple_lines(self): + """Test indenting multiple lines.""" + text = "Line 1\nLine 2\nLine 3" + result = space_indent(text) + + expected = " Line 1\n Line 2\n Line 3" + assert result == expected + + def test_space_indent_empty_string(self): + """Test indenting an empty string.""" + text = "" + result = space_indent(text) + + assert result == "" + + def test_space_indent_empty_lines(self): + """Test indenting text with empty lines.""" + text = "Line 1\n\nLine 3" + result = space_indent(text) + + # Empty lines should remain empty (no spaces added) + expected = " Line 1\n\n Line 3" + assert result == expected + + def test_space_indent_whitespace_only_lines(self): + """Test indenting lines with only whitespace.""" + text = "Line 1\n \nLine 3" + result = space_indent(text) + + # Lines with only whitespace should remain empty after strip check + lines = result.split("\n") + assert lines[0] == " Line 1" + assert lines[1] == "" # Whitespace-only line becomes empty + assert lines[2] == " Line 3" + + def test_space_indent_already_indented(self): + """Test indenting text that is already indented.""" + text = " Already indented\n Second line" + result = space_indent(text) + + # Should add 4 more spaces to already indented lines + expected = " Already indented\n Second line" + assert result == expected + + def test_space_indent_tabs(self): + """Test indenting text with tabs.""" + text = "\tTabbed line" + result = space_indent(text) + + # Should add 4 spaces before the tab + assert result == " \tTabbed line" + + def test_space_indent_mixed_content(self): + """Test indenting text with mixed content.""" + text = "Normal\n Indented\n\nEmpty above\n \nWhitespace above" + result = space_indent(text) + + lines = result.split("\n") + assert lines[0] == " Normal" + assert lines[1] == " Indented" + assert lines[2] == "" # Empty line + assert lines[3] == " Empty above" + assert lines[4] == "" # Whitespace-only becomes empty + assert lines[5] == " Whitespace above" + + def test_space_indent_with_trailing_newline(self): + """Test indenting text with trailing newline.""" + text = "Line 1\nLine 2\n" + result = space_indent(text) + + # splitlines() doesn't include the trailing newline's empty string + # So the result won't have a trailing newline + assert result == " Line 1\n Line 2" + + def test_space_indent_single_word(self): + """Test indenting a single word.""" + text = "word" + result = space_indent(text) + + assert result == " word" + + def test_space_indent_with_unicode(self): + """Test indenting text with unicode characters.""" + text = "Unicode: δ½ ε₯½\nEmoji: πŸš€" + result = space_indent(text) + + expected = " Unicode: δ½ ε₯½\n Emoji: πŸš€" + assert result == expected + + def test_space_indent_special_characters(self): + """Test indenting text with special characters.""" + text = "Special: !@#$%\nMore: ^&*()" + result = space_indent(text) + + expected = " Special: !@#$%\n More: ^&*()" + assert result == expected + + def test_space_indent_uses_four_spaces(self): + """Test that space_indent uses exactly 4 spaces.""" + text = "test" + result = space_indent(text) + + # Verify it starts with exactly 4 spaces + assert result.startswith(" ") + assert len(result) == len(" test") + # Check the spaces are at the beginning + assert result == " test" + + def test_space_indent_long_lines(self): + """Test indenting very long lines.""" + text = "x" * 1000 + result = space_indent(text) + + assert result == " " + ("x" * 1000) + assert len(result) == 1004 # 4 spaces + 1000 x's + + def test_space_indent_many_lines(self): + """Test indenting many lines.""" + lines = [f"Line {i}" for i in range(100)] + text = "\n".join(lines) + result = space_indent(text) + + result_lines = result.split("\n") + assert len(result_lines) == 100 + for i, line in enumerate(result_lines): + assert line == f" Line {i}" + + +@pytest.mark.unit +class TestTabIndent: + """Tests for the tab_indent function.""" + + def test_tab_indent_not_implemented(self): + """Test that tab_indent is not yet implemented.""" + # The function currently just has 'pass', so it returns None + result = tab_indent("test") + assert result is None + + @pytest.mark.skip(reason="tab_indent is not implemented yet") + def test_tab_indent_future_implementation(self): + """Placeholder test for when tab_indent is implemented.""" + # When implemented, this test should be updated + text = "Line 1\nLine 2" + result = tab_indent(text) + + # Expected behavior: indent with tabs + expected = "\tLine 1\n\tLine 2" + assert result == expected From ece1aa3e2378539b6b3423921041f19d85f73224 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 17 Feb 2026 12:59:21 +0000 Subject: [PATCH 4/4] Merge origin/main to resolve conflicts completely Co-authored-by: colinmakerofthings <66311935+colinmakerofthings@users.noreply.github.com> --- tests/test_cli.py | 60 +++++++++++++++++++++++------------------------ 1 file changed, 30 insertions(+), 30 deletions(-) diff --git a/tests/test_cli.py b/tests/test_cli.py index 6258c94..70366d7 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -843,33 +843,33 @@ def test_analyze_without_mock_llm(self, tmp_path, monkeypatch): # Skip this test - when monkeypatch deletes LOGPILOT_MOCK_LLM, # the conftest has already set it so it won't affect the run_cli_direct behavior pytest.skip("Cannot easily test non-mock path with conftest env var") - - -def test_cli_model_option(tmp_path): - """Test that the --model option is accepted and works""" - log_file = tmp_path / "test.log" - log_file.write_text( - ( - '{"timestamp": "2026-01-28T12:00:00Z", ' - '"level": "ERROR", ' - '"message": "Something failed"}\n' - ) - ) - result = run_cli(["analyze", str(log_file), "--model", "gpt-3.5-turbo"]) - assert result.returncode == 0, f"stdout: {result.stdout}\nstderr: {result.stderr}" - assert "Mocked summary" in result.stdout - - -def test_cli_model_option_default(tmp_path): - """Test that the default model (gpt-4) still works without specifying --model""" - log_file = tmp_path / "test.log" - log_file.write_text( - ( - '{"timestamp": "2026-01-28T12:00:00Z", ' - '"level": "ERROR", ' - '"message": "Something failed"}\n' - ) - ) - result = run_cli(["analyze", str(log_file)]) - assert result.returncode == 0, f"stdout: {result.stdout}\nstderr: {result.stderr}" - assert "Mocked summary" in result.stdout + + +def test_cli_model_option(tmp_path): + """Test that the --model option is accepted and works""" + log_file = tmp_path / "test.log" + log_file.write_text( + ( + '{"timestamp": "2026-01-28T12:00:00Z", ' + '"level": "ERROR", ' + '"message": "Something failed"}\n' + ) + ) + result = run_cli(["analyze", str(log_file), "--model", "gpt-3.5-turbo"]) + assert result.returncode == 0, f"stdout: {result.stdout}\nstderr: {result.stderr}" + assert "Mocked summary" in result.stdout + + +def test_cli_model_option_default(tmp_path): + """Test that the default model (gpt-4) still works without specifying --model""" + log_file = tmp_path / "test.log" + log_file.write_text( + ( + '{"timestamp": "2026-01-28T12:00:00Z", ' + '"level": "ERROR", ' + '"message": "Something failed"}\n' + ) + ) + result = run_cli(["analyze", str(log_file)]) + assert result.returncode == 0, f"stdout: {result.stdout}\nstderr: {result.stderr}" + assert "Mocked summary" in result.stdout