From 35a9aa4b244f9c3d9e5d8e3331b952a833b70889 Mon Sep 17 00:00:00 2001 From: Caleb Evans Date: Fri, 22 May 2026 16:25:48 -0600 Subject: [PATCH] feat: harden CLI with error handling, overwrite protection, and tests - Wrap analysis calls in try/except to continue on errors - Add output file write error handling - Add --force flag for output file overwrite protection - Add KeyboardInterrupt handler for clean Ctrl+C exit - Refactor analyze_file into smaller focused functions - Add comprehensive CLI test suite Co-authored-by: Cursor --- space/cordon | 1 + src/cordon/cli.py | 104 ++++++++++++++++---- tests/test_cli.py | 246 ++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 330 insertions(+), 21 deletions(-) create mode 160000 space/cordon create mode 100644 tests/test_cli.py diff --git a/space/cordon b/space/cordon new file mode 160000 index 0000000..3f11ee6 --- /dev/null +++ b/space/cordon @@ -0,0 +1 @@ +Subproject commit 3f11ee6ce01cdb0a4a2af33a526ae556269a7740 diff --git a/src/cordon/cli.py b/src/cordon/cli.py index 0e7de0a..5f61461 100644 --- a/src/cordon/cli.py +++ b/src/cordon/cli.py @@ -137,38 +137,90 @@ def parse_args() -> argparse.Namespace: default=None, help="Save anomalous blocks to file (default: print to stdout)", ) + output_group.add_argument( + "--force", + action="store_true", + help="Overwrite output file if it exists", + ) return parser.parse_args() -def analyze_file( - log_path: Path, analyzer: SemanticLogAnalyzer, detailed: bool, output_path: Path | None = None -) -> None: - """Analyze a single log file and print results. +def _validate_file(log_path: Path) -> bool: + """Check that a log file exists and is a regular file. Args: - log_path: Path to the log file - analyzer: Configured SemanticLogAnalyzer instance - detailed: Whether to show detailed statistics - output_path: Optional path to save anomalous blocks (None = stdout) + log_path: Path to the log file to validate. + + Returns: + True if the file is valid, False otherwise. """ - # verify file exists and is readable if not log_path.exists(): print(f"Error: File not found: {log_path}", file=sys.stderr) - return + return False if not log_path.is_file(): print(f"Error: Not a file: {log_path}", file=sys.stderr) + return False + return True + + +def _write_output(content: str, output_path: Path, force: bool) -> None: + """Write analysis output to a file with overwrite protection. + + Args: + content: The content to write. + output_path: Destination file path. + force: If True, overwrite an existing file. + """ + if output_path.exists() and not force: + print( + f"Error: Output file already exists: {output_path}", + file=sys.stderr, + ) + print("Use --force to overwrite.", file=sys.stderr) + return + + try: + output_path.write_text(content) + print(f"Anomalous blocks written to: {output_path}") + except OSError as error: + print(f"Error writing output file: {error}", file=sys.stderr) + + +def analyze_file( + log_path: Path, + analyzer: SemanticLogAnalyzer, + detailed: bool, + output_path: Path | None = None, + force: bool = False, +) -> None: + """Analyze a single log file and print results. + + Args: + log_path: Path to the log file. + analyzer: Configured SemanticLogAnalyzer instance. + detailed: Whether to show detailed statistics. + output_path: Optional path to save anomalous blocks (None prints to stdout). + force: If True, overwrite an existing output file. + """ + if not _validate_file(log_path): return print("=" * 80) print(f"Analyzing: {log_path}") print("=" * 80) - result = analyzer.analyze_file_detailed(log_path) - - print(f"Total lines: {result.total_lines:,}") + try: + if detailed: + result = analyzer.analyze_file_detailed(log_path) + else: + output = analyzer.analyze_file(log_path) + except Exception as error: + print(f"Error analyzing {log_path}: {error}", file=sys.stderr) + return if detailed: + print(f"Total lines: {result.total_lines:,}") print("\nAnalysis Statistics:") print(f" Total windows created: {result.total_windows:,}") print(f" Significant windows: {result.significant_windows:,}") @@ -184,11 +236,14 @@ def analyze_file( print(f"\n{'Significant Blocks':^80}") print("=" * 80) + content = result.output + else: + content = output + if output_path: - output_path.write_text(result.output) - print(f"Anomalous blocks written to: {output_path}") + _write_output(content, output_path, force) else: - print(result.output) + print(content) print() @@ -223,8 +278,8 @@ def _print_filtering_mode(config: AnalysisConfig) -> None: print(f"Filtering mode: Percentile (top {config.anomaly_percentile*100:.1f}%)") -def main() -> None: - """Main entry point for the CLI.""" +def _main_impl() -> None: + """Implementation of the main CLI entry point.""" args = parse_args() # handle anomaly range vs percentile mutual exclusivity @@ -233,10 +288,8 @@ def main() -> None: anomaly_percentile = args.anomaly_percentile if args.anomaly_range is not None: - # Using range mode anomaly_range_min = args.anomaly_range[0] anomaly_range_max = args.anomaly_range[1] - # Keep default percentile value (not used in range mode) if not isclose(args.anomaly_percentile, 0.1): print( "Warning: --anomaly-percentile is ignored when using --anomaly-range", @@ -288,7 +341,16 @@ def main() -> None: # analyze each log file for log_path in args.logfiles: - analyze_file(log_path, analyzer, args.detailed, args.output) + analyze_file(log_path, analyzer, args.detailed, args.output, args.force) + + +def main() -> None: + """Main entry point for the CLI.""" + try: + _main_impl() + except KeyboardInterrupt: + print("\nInterrupted.", file=sys.stderr) + sys.exit(130) if __name__ == "__main__": diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..a0db220 --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,246 @@ +"""Tests for the CLI module.""" + +import sys +from pathlib import Path +from unittest.mock import MagicMock + +import pytest + +from cordon.cli import analyze_file, parse_args + + +class TestParseArgs: + """Tests for argument parsing.""" + + def test_basic_args(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Test parsing basic positional arguments.""" + monkeypatch.setattr(sys, "argv", ["cordon", "test.log"]) + args = parse_args() + assert args.logfiles == [Path("test.log")] + assert args.backend == "sentence-transformers" + assert args.window_size == 4 + + def test_multiple_files(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Test parsing multiple log files.""" + monkeypatch.setattr(sys, "argv", ["cordon", "a.log", "b.log"]) + args = parse_args() + assert len(args.logfiles) == 2 + + def test_backend_choice(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Test backend selection.""" + monkeypatch.setattr(sys, "argv", ["cordon", "--backend", "remote", "test.log"]) + args = parse_args() + assert args.backend == "remote" + + def test_device_choice(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Test device selection.""" + monkeypatch.setattr(sys, "argv", ["cordon", "--device", "cpu", "test.log"]) + args = parse_args() + assert args.device == "cpu" + + def test_output_flag(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Test output file flag.""" + monkeypatch.setattr(sys, "argv", ["cordon", "-o", "out.xml", "test.log"]) + args = parse_args() + assert args.output == Path("out.xml") + + def test_force_flag(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Test --force flag.""" + monkeypatch.setattr(sys, "argv", ["cordon", "--force", "-o", "out.xml", "test.log"]) + args = parse_args() + assert args.force is True + + def test_force_flag_default(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Test --force defaults to False.""" + monkeypatch.setattr(sys, "argv", ["cordon", "test.log"]) + args = parse_args() + assert args.force is False + + def test_anomaly_range(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Test anomaly range parsing.""" + monkeypatch.setattr( + sys, + "argv", + ["cordon", "--anomaly-range", "0.05", "0.15", "test.log"], + ) + args = parse_args() + assert args.anomaly_range == [0.05, 0.15] + + def test_detailed_flag(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Test --detailed flag.""" + monkeypatch.setattr(sys, "argv", ["cordon", "--detailed", "test.log"]) + args = parse_args() + assert args.detailed is True + + +class TestAnalyzeFile: + """Tests for analyze_file function.""" + + def test_nonexistent_file(self, capsys: pytest.CaptureFixture[str]) -> None: + """Test handling of nonexistent file.""" + analyzer = MagicMock() + analyze_file(Path("/nonexistent/file.log"), analyzer, detailed=False) + captured = capsys.readouterr() + assert "Error: File not found" in captured.err + + def test_not_a_file(self, capsys: pytest.CaptureFixture[str], tmp_path: Path) -> None: + """Test handling of a path that exists but is not a regular file.""" + analyzer = MagicMock() + analyze_file(tmp_path, analyzer, detailed=False) + captured = capsys.readouterr() + assert "Error: Not a file" in captured.err + + def test_analysis_error_continues( + self, capsys: pytest.CaptureFixture[str], tmp_path: Path + ) -> None: + """Test that analysis errors are caught and reported.""" + log_file = tmp_path / "test.log" + log_file.write_text("line 1\nline 2\n") + + analyzer = MagicMock() + analyzer.analyze_file_detailed.side_effect = RuntimeError("Model failed") + + analyze_file(log_file, analyzer, detailed=True) + captured = capsys.readouterr() + assert "Error analyzing" in captured.err + assert "Model failed" in captured.err + + def test_analysis_error_non_detailed( + self, capsys: pytest.CaptureFixture[str], tmp_path: Path + ) -> None: + """Test that analysis errors in non-detailed mode are caught.""" + log_file = tmp_path / "test.log" + log_file.write_text("line 1\nline 2\n") + + analyzer = MagicMock() + analyzer.analyze_file.side_effect = ValueError("Bad input") + + analyze_file(log_file, analyzer, detailed=False) + captured = capsys.readouterr() + assert "Error analyzing" in captured.err + assert "Bad input" in captured.err + + def test_write_failure(self, capsys: pytest.CaptureFixture[str], tmp_path: Path) -> None: + """Test handling of output write failure.""" + log_file = tmp_path / "test.log" + log_file.write_text("line 1\n") + + analyzer = MagicMock() + mock_result = MagicMock() + mock_result.output = "" + mock_result.total_lines = 1 + mock_result.total_windows = 0 + mock_result.significant_windows = 0 + mock_result.merged_blocks = 0 + mock_result.processing_time = 0.1 + mock_result.score_distribution = { + "min": 0, + "max": 0, + "mean": 0, + "median": 0, + "p90": 0, + } + analyzer.analyze_file_detailed.return_value = mock_result + + bad_output = Path("/nonexistent/dir/output.xml") + analyze_file(log_file, analyzer, detailed=True, output_path=bad_output, force=True) + captured = capsys.readouterr() + assert "Error writing output file" in captured.err + + def test_overwrite_protection(self, capsys: pytest.CaptureFixture[str], tmp_path: Path) -> None: + """Test that existing output files are not overwritten without --force.""" + log_file = tmp_path / "test.log" + log_file.write_text("line 1\n") + output_file = tmp_path / "output.xml" + output_file.write_text("existing content") + + analyzer = MagicMock() + analyze_file( + log_file, + analyzer, + detailed=False, + output_path=output_file, + force=False, + ) + + assert output_file.read_text() == "existing content" + captured = capsys.readouterr() + assert "already exists" in captured.err + assert "--force" in captured.err + + def test_force_overwrite(self, tmp_path: Path) -> None: + """Test that --force allows overwriting.""" + log_file = tmp_path / "test.log" + log_file.write_text("line 1\n") + output_file = tmp_path / "output.xml" + output_file.write_text("existing content") + + analyzer = MagicMock() + analyzer.analyze_file.return_value = "new" + + analyze_file( + log_file, + analyzer, + detailed=False, + output_path=output_file, + force=True, + ) + assert output_file.read_text() == "new" + + def test_stdout_output(self, capsys: pytest.CaptureFixture[str], tmp_path: Path) -> None: + """Test that output is printed to stdout when no output path is given.""" + log_file = tmp_path / "test.log" + log_file.write_text("line 1\n") + + analyzer = MagicMock() + analyzer.analyze_file.return_value = "results" + + analyze_file(log_file, analyzer, detailed=False) + captured = capsys.readouterr() + assert "results" in captured.out + + def test_detailed_output(self, capsys: pytest.CaptureFixture[str], tmp_path: Path) -> None: + """Test that detailed mode prints analysis statistics.""" + log_file = tmp_path / "test.log" + log_file.write_text("line 1\n") + + analyzer = MagicMock() + mock_result = MagicMock() + mock_result.output = "" + mock_result.total_lines = 100 + mock_result.total_windows = 25 + mock_result.significant_windows = 3 + mock_result.merged_blocks = 2 + mock_result.processing_time = 1.23 + mock_result.score_distribution = { + "min": 0.1, + "max": 0.9, + "mean": 0.5, + "median": 0.45, + "p90": 0.8, + } + analyzer.analyze_file_detailed.return_value = mock_result + + analyze_file(log_file, analyzer, detailed=True) + captured = capsys.readouterr() + assert "Total lines: 100" in captured.out + assert "Total windows created: 25" in captured.out + assert "Significant windows: 3" in captured.out + assert "Processing time: 1.23s" in captured.out + assert "Score Distribution:" in captured.out + + +class TestMainEntryPoint: + """Tests for the main() entry point.""" + + def test_keyboard_interrupt(self) -> None: + """Test that KeyboardInterrupt results in exit code 130.""" + from cordon.cli import main + + with pytest.raises(SystemExit) as exc_info, pytest.MonkeyPatch.context() as mp: + mp.setattr( + "cordon.cli._main_impl", + MagicMock(side_effect=KeyboardInterrupt), + ) + main() + assert exc_info.value.code == 130