Perf/replace fixer queue preview

refactron/cli/main.py

@@ -143,6 +143,13 @@ def main(ctx: click.Context) -> None:
 except ImportError:
     pass
 
+try:
+    from refactron.cli.run import run
+
+    main.add_command(run)
+except ImportError:
+    pass
+
 try:
     from refactron.cli.cicd import feedback, generate_cicd, init
 

refactron/cli/run.py

@@ -0,0 +1,41 @@
+"""
+Refactron CLI - Run Module.
+Provides the 'run' command to execute Refactron as a connected pipeline.
+"""
+from typing import Optional
+from pathlib import Path
+import click
+
+from refactron.cli.ui import console, _auth_banner
+from refactron.cli.utils import _validate_path
+from refactron.core.pipeline import RefactronPipeline
+
+@click.command()
+@click.argument("target", type=click.Path(exists=True), required=False)
+@click.option(
+    "--incremental/--no-incremental",
+    default=False,
+    help="Enable or disable incremental analysis for the pipeline run (off by default)",
+)
+def run(target: Optional[str], incremental: bool) -> None:
+    """
+    Run Refactron as a connected pipeline session.
+    """
+    console.print()
+    _auth_banner("Pipeline Run")
+    console.print()
+
+    target_path = _validate_path(target) if target else Path.cwd()
+
+    with console.status("[primary]Executing pipeline run...[/primary]"):
+        try:
+            pipeline = RefactronPipeline()
+            result = pipeline.analyze(target_path, use_incremental=incremental)
+
+            console.print(f"[green]Pipeline analysis completed successfully.[/green]")
+            console.print(f"Total files analyzed: {result.total_files}")
+            console.print(f"Total issues found: {result.total_issues}")
+
+        except Exception as e:
+            console.print(f"[red]Pipeline run failed: {e}[/red]")
+            raise SystemExit(1)

refactron/core/pipeline.py

@@ -0,0 +1,97 @@
+"""Pipeline orchestration for automated and session-based flows."""
+
+from pathlib import Path
+from typing import Optional, Union, List
+
+from refactron.core.config import RefactronConfig
+from refactron.core.refactron import Refactron
+from refactron.core.analysis_result import AnalysisResult
+from refactron.core.models import CodeIssue
+from refactron.autofix.engine import AutoFixEngine
+
+class RefactronPipeline:
+    """Automated pipeline execution for session-based and CI/CD flows."""
+
+    def __init__(self, project_root: Optional[Union[str, Path]] = None, config_path: Optional[Union[str, Path]] = None):
+        """
+        Initialize the pipeline.
+
+        Args:
+            project_root: Optional root directory of the project.
+            config_path: Optional explicit configuration path.
+        """
+        self.project_root = Path(project_root) if project_root else None
+        self.config_path = Path(config_path) if config_path else None
+
+        self._config = self._load_config()
+        # Enforce pipeline default behavior (full scan) unless overridden later
+        self._config.enable_incremental_analysis = False
+        self.refactron = Refactron(self._config)
+        self.autofix_engine = AutoFixEngine()
+        self._fixer_cache = {}
+
+    def _load_config(self) -> RefactronConfig:
+        if self.config_path and self.config_path.exists():
+            return RefactronConfig.from_file(self.config_path)
+
+        root = self.project_root or Path.cwd()
+        yaml_path = root / ".refactron.yaml"
+        if yaml_path.exists():
+            return RefactronConfig.from_file(yaml_path)
+        return RefactronConfig.default()
+
+    def analyze(self, target: Union[str, Path], use_incremental: Optional[bool] = None) -> AnalysisResult:
+        """
+        Run analysis as part of a pipeline flow.
+
+        Pipeline-only overrides:
+        - `enable_incremental_analysis` is forced to False by default to guarantee
+          a fully fresh, reproducible analysis in CI/CD or pipeline environments.
+
+        Args:
+            target: Path to file or directory to analyze.
+            use_incremental: Optional override to enable incremental analysis.
+
+        Returns:
+            AnalysisResult containing issues and metrics.
+        """
+        if use_incremental is not None:
+            self.refactron.config.enable_incremental_analysis = use_incremental
+
+        return self.refactron.analyze(target)
+
+    def queue_issues(self, issues: List[CodeIssue]) -> list:
+        """Queue issues mapped to their responsible fixers."""
+        queued = []
+        for issue in issues:
+            fixer_name = self._find_fixer_name(issue)
+            if fixer_name:
+                queued.append({"issue": issue, "fixer_name": fixer_name})
+        return queued
+
+    def _find_fixer_name(self, issue: CodeIssue) -> Optional[str]:
+        """Find the appropriate fixer name for a given issue."""
+        # Use rule_id or category as cache key
+        issue_type = issue.rule_id or str(getattr(issue, 'category', type(issue).__name__))
+        if issue_type in self._fixer_cache:
+            return self._fixer_cache[issue_type]
+
+        candidate_name = None
+
+        # 1. Ask AutoFixEngine directly without preview (O(1) dictionary lookup)
+        if hasattr(self, 'autofix_engine') and self.autofix_engine.can_fix(issue):
+            candidate_name = issue.rule_id
+        else:
+            # 2. Fallback to preview-based resolution for ambiguous cases
+            if hasattr(self, 'autofix_engine'):
+                for name, fixer in self.autofix_engine.fixers.items():
+                    try:
+                        preview_result = fixer.preview(issue, "x = 1\n")
+                        if preview_result and getattr(preview_result, "success", False):
+                            candidate_name = name
+                            break
+                    except Exception:
+                        continue
+
+        self._fixer_cache[issue_type] = candidate_name
+        return candidate_name

tests/test_pipeline.py

@@ -0,0 +1,107 @@
+"""Tests for the Refactron pipeline module."""
+
+import tempfile
+from pathlib import Path
+from unittest.mock import patch, MagicMock
+
+from refactron.core.pipeline import RefactronPipeline
+from refactron.core.config import RefactronConfig
+
+def test_pipeline_loads_project_config():
+    """Test that RefactronPipeline loads configuration from the project root."""
+    with tempfile.TemporaryDirectory() as temp_dir:
+        root_path = Path(temp_dir)
+
+        config_path = root_path / ".refactron.yaml"
+        config_path.write_text("max_function_complexity: 99\nenable_metrics: false\n")
+
+        target_file = root_path / "dummy.py"
+        target_file.write_text("def foo():\n    pass\n")
+
+        with patch("refactron.core.pipeline.Refactron.analyze") as mock_analyze:
+            mock_analyze.return_value = "mock_result"
+            pipeline = RefactronPipeline(project_root=root_path)
+
+            result = pipeline.analyze(target_file)
+
+            assert result == "mock_result"
+            config_passed = pipeline.refactron.config
+
+            assert isinstance(config_passed, RefactronConfig)
+            assert config_passed.max_function_complexity == 99
+            assert config_passed.enable_metrics is False
+            assert config_passed.enable_incremental_analysis is False
+
+def test_pipeline_incremental_override():
+    """Test that RefactronPipeline can override the incremental setting."""
+    with tempfile.TemporaryDirectory() as temp_dir:
+        root_path = Path(temp_dir)
+
+        target_file = root_path / "dummy.py"
+        target_file.write_text("def foo():\n    pass\n")
+
+        with patch("refactron.core.pipeline.Refactron.analyze"):
+            pipeline = RefactronPipeline(project_root=root_path)
+            pipeline.analyze(target_file, use_incremental=True)
+
+            config_passed = pipeline.refactron.config
+            assert config_passed.enable_incremental_analysis is True
+
+def test_pipeline_queue_issues_caching():
+    """Test that queue_issues leverages caching and direct mapping without multiple previews."""
+    from refactron.core.models import CodeIssue, IssueCategory, IssueLevel
+    from refactron.autofix.models import FixResult
+    from pathlib import Path
+
+    with tempfile.TemporaryDirectory() as temp_dir:
+        pipeline = RefactronPipeline(project_root=Path(temp_dir))
+
+        # Mock AutoFixEngine on the pipeline directly
+        pipeline.autofix_engine = MagicMock()
+
+        calls = {"preview": 0}
+
+        # Create a mock fixer that counts preview calls
+        mock_fixer = MagicMock()
+        mock_fixer.name = "mock_rule"
+        def mock_preview(issue, code):
+            calls["preview"] += 1
+            return FixResult(success=True, reason="")
+        mock_fixer.preview.side_effect = mock_preview
+
+        pipeline.autofix_engine.fixers = {"mock_rule": mock_fixer}
+
+        # Mock can_fix to return False so it falls back to preview ONCE per type
+        pipeline.autofix_engine.can_fix.return_value = False
+
+        issue1 = CodeIssue(
+            category=IssueCategory.STYLE,
+            level=IssueLevel.WARNING,
+            message="Test issue",
+            file_path=Path("dummy.py"),
+            line_number=1,
+            rule_id="unknown_rule_1"
+        )
+        issue2 = CodeIssue(
+            category=IssueCategory.STYLE,
+            level=IssueLevel.WARNING,
+            message="Test issue 2",
+            file_path=Path("dummy.py"),
+            line_number=2,
+            rule_id="unknown_rule_1"
+        )
+        issue3 = CodeIssue(
+            category=IssueCategory.STYLE,
+            level=IssueLevel.WARNING,
+            message="Test issue 3",
+            file_path=Path("dummy.py"),
+            line_number=3,
+            rule_id="unknown_rule_1"
+        )
+
+        queued = pipeline.queue_issues([issue1, issue2, issue3])
+
+        assert len(queued) == 3
+        # Should only have run preview 1 time, because the subsequent issues have the same rule_id and hit the cache.
+        assert calls["preview"] == 1
+        assert pipeline._fixer_cache["unknown_rule_1"] == "mock_rule"