diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..4a88c99 --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,99 @@ +# CLAUDE.md + +This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository. + +## Project Overview + +Refactron is a Python library and CLI tool for code analysis, refactoring, and technical debt elimination. It uses AST-based analysis (libcst, astroid), AI-powered suggestions (Groq/Llama 3), and a RAG system (ChromaDB) for context-aware code intelligence. Version 1.0.15 → 1.1.0 (MVP in progress), Python 3.8+, MIT license. + +**Active development branch:** `main` (MVP v1.1.0 — adding Verification Engine, `--dry-run`, `--verify`). See `dev-notes/Refactron_Comprehensive_MVP.md` for the full roadmap. + +## Build & Development Commands + +```bash +# Install in development mode +pip install -e ".[dev]" + +# Run all tests (with coverage by default via pyproject.toml addopts) +pytest + +# Run a single test file +pytest tests/test_analyzers.py + +# Run a single test +pytest tests/test_analyzers.py::TestClassName::test_method_name + +# Run tests without coverage +pytest --no-cov + +# Formatting +black refactron tests --line-length=100 +isort refactron tests --profile=black + +# Linting +flake8 refactron --max-line-length=100 + +# Type checking (strict for src, relaxed for tests/examples/benchmarks) +mypy refactron + +# Run all pre-commit hooks +pre-commit run --all-files + +# CLI entry point +refactron +``` + +## Code Style + +- **Line length**: 100 (enforced by black, isort, flake8) +- **Formatter**: black with `target-version = ["py38", "py39", "py310", "py311"]` +- **Import sorting**: isort with `profile = "black"` +- **Type annotations**: Required in `refactron/` (mypy `disallow_untyped_defs = true`), not required in tests/examples/benchmarks + +## Architecture + +### Entry Points +- **CLI**: `refactron/cli/` package → `refactron/cli/main.py:main` (Click group). Entry point: `refactron.cli:main`. Subcommands split across `analysis.py`, `refactor.py`, `patterns.py`, `rag.py`, `cicd.py`, `repo.py`, `auth.py` +- **Python API**: `Refactron` class from `refactron/core/refactron.py`, exported via `__init__.py` using lazy imports (heavy modules only load when accessed) + +### Core Modules +- **`core/`** — Central orchestration: `Refactron` main class, `RefactronConfig` (YAML-backed with versioning), models, backup/rollback, AST cache, incremental analysis, parallel processing, false positive tracking, workspace management, credentials/device auth, logging config, Prometheus metrics, telemetry, OpenTelemetry +- **`analyzers/`** — Plugin-based code analyzers (7: complexity, code smell, security, performance, dead code, dependency, type hint). All extend `BaseAnalyzer` +- **`refactorers/`** — Plugin-based refactorers (5: extract method, add docstring, magic number, reduce parameters, simplify conditionals). All extend `BaseRefactorer` +- **`analysis/`** — Semantic analysis layer: CFG builder (`cfg/`), data flow analysis, taint analysis (source-to-sink tracking), symbol table (maps classes/functions/variables across codebase), type inference engine (`core/inference.py`) +- **`autofix/`** — Automated fix engine with fixers, risk scoring, and safety previews +- **`patterns/`** — Pattern learning system: fingerprinting, learning, matching, ranking, storage, tuning +- **`llm/`** — LLM integration: orchestrator, Groq client, backend client, prompts, safety checks +- **`rag/`** — RAG system: code indexing, chunking, parsing, retrieval (ChromaDB + sentence-transformers) +- **`cicd/`** — CI/CD integrations: GitHub Actions, GitLab CI, pre-commit hooks, PR integration, quality gates + +### Key Design Patterns +- **Plugin architecture**: Analyzers and refactorers register via base classes and are discovered by the core `Refactron` class +- **Safety-first refactoring**: All refactoring goes through preview → backup → apply → optional rollback +- **AST-based**: Uses libcst for concrete syntax tree manipulation (preserves formatting) and astroid for advanced analysis +- **Auth-gated CLI**: All CLI commands except `login`/`logout`/`auth` check for a valid access token (stored via `core/credentials.py`) before proceeding + +### MVP v1.1.0 Work in Progress + +| Day | Feature | Status | +|-----|---------|--------| +| 1 | Exception isolation for TaintAnalyzer/DataFlowAnalyzer (`AnalysisSkipWarning`, skip-rate tracking) | ✅ Done | +| 2 | SHA-256 hardening for `IncrementalAnalysisTracker`; backup integrity validation in `BackupManager` | ✅ Done | +| 3 | `--dry-run` flag for `refactron autofix`; `generate_diff()` in `autofix/file_ops.py`; `AutoFixEngine.fix_file()` | ✅ Done | +| 4 | Test fixture files in `tests/fixtures/` (6 files, 8 fixture validation tests) | ✅ Done | +| 5 | Phase 1 gate: 758 tests green, self-analysis 96 files/0 crashes, added `--no-cache` flag to `analyze` | ✅ Done | +| 6–15 | Verification Engine (`refactron/verification/`) | Pending | + +**Key new APIs (Day 1–3):** +- `AnalysisSkipWarning` dataclass in `core/models.py` — surfaced on `AnalysisResult.semantic_skip_warnings` +- `BackupManager.validate_backup_integrity(session_id)` → `(valid_paths, corrupt_paths)` +- `generate_diff(original, modified, filename)` in `autofix/file_ops.py` — returns unified diff string +- `AutoFixEngine.fix_file(file_path, issues, dry_run=True)` → `(fixed_code, diff_or_None)` + +**CLI output overhaul:** +- `refactron analyze` now shows an **interactive issue viewer** (TTY) with severity-grouped navigation (`[1-4]` to drill in, `[n/p/b/q]` to navigate) +- Non-interactive fallback for CI/CD (piped output or `--no-interactive` flag) +- Issues grouped by severity (CRITICAL → ERROR → WARNING → INFO) +- Relative file paths instead of absolute +- Code context (3 lines) for critical/warning/error issues +- Logger noise suppressed (use `--log-level INFO` to re-enable) diff --git a/dev-notes/Refactron_Comprehensive_MVP.docx b/dev-notes/Refactron_Comprehensive_MVP.docx new file mode 100644 index 0000000..e2c32c0 Binary files /dev/null and b/dev-notes/Refactron_Comprehensive_MVP.docx differ diff --git a/dev-notes/Refactron_Comprehensive_MVP.md b/dev-notes/Refactron_Comprehensive_MVP.md new file mode 100644 index 0000000..fca1775 --- /dev/null +++ b/dev-notes/Refactron_Comprehensive_MVP.md @@ -0,0 +1,1067 @@ ++-----------------------------------------------------------------------+ +| **REFACTRON** | +| | +| Comprehensive MVP Plan | +| | +| *Safety-First Refactoring for Production Codebases* | +| | +| | +| ---------------------- ---------------------- ---------------------- | +| **v1.1.0** **5 Weeks** **March 2026** | +| | +| | +| ---------------------- ---------------------- ---------------------- | +| | +| Repo Connect: ✅ COMPLETE \| Current: v1.0.15 on PyPI | ++-----------------------------------------------------------------------+ + ++-----------------------------------------------------------------------+ +| **🎯 North Star --- The One Sentence That Governs Every Decision** | +| | +| \"I ran Refactron on my production codebase. It found real issues, | +| fixed the safe ones, | +| | +| and proved --- with evidence --- that nothing would break.\" | +| | +| Every feature in this plan exists to make a developer able to say | +| that sentence. | +| | +| If a feature does not serve that sentence, it is not in this MVP. | ++-----------------------------------------------------------------------+ + +**Contents** + +**Part 1 --- What We Are Starting From** + +**1.1 What Is Confirmed Built and Stable** + +The following modules are live at v1.0.15 on PyPI. They are tested, +functional, and ship as-is in the MVP. The mandate is to stabilise them +--- not rewrite them. + + -------------------------- ------------------------ -------------------------- + **Module** **Location** **Status** + + **Core Orchestrator** *core/refactron.py* **✅ Stable --- ship + as-is** + + **8 Rule Analyzers** *analyzers/* **✅ Stable --- 96.8% test + coverage** + + **5 Complex Refactorers** *refactorers/* **✅ Stable** + + **14 Auto-Fixers** *autofix/* **✅ Stable --- needs + \--dry-run flag added** + + **BackupRollbackSystem** *core/backup.py* **✅ Stable --- reactive + only, pre-MVP** + + **AST Cache + *core/cache.py* **✅ Stable --- needs + Incremental** SHA-256 hash fix** + + **CI/CD Gateway** *cicd/* **✅ Stable** + + **CLI (10 commands)** *cli.py* **✅ Stable** + + **Config System (YAML)** *core/config\*.py* **✅ Stable** + + **Repo Connect** *cli/repo.py* **✅ COMPLETE --- excluded + from MVP scope** + -------------------------- ------------------------ -------------------------- + +**1.2 What Is Partial and Needs Gating** + +These modules exist but are not reliable enough to ship as default +behavior. They will be wrapped in exception isolation and kept in +dispatch with visible warnings (Option B decision). They are NOT +disabled --- they fail transparently. + + ---------------------- ------------------------- -------------------------- + **Module** **Location** **Required Action** + + **LLM Orchestrator** *llm/* **🟡 No reliability layer + --- exclude from MVP + entirely** + + **RAG System** *rag/* **🟡 No quality validation + --- exclude from MVP + entirely** + + **TaintAnalyzer** *analysis/taint.py* **🟡 Wrap in try/except + --- show visible skip + warning** + + **DataFlowAnalyzer** *analysis/data_flow.py* **🟡 Wrap in try/except + --- show visible skip + warning** + + **SafetyGate** *llm/safety.py* **🟡 Guards LLM only --- + not relevant without LLM + in MVP** + ---------------------- ------------------------- -------------------------- + ++-----------------------------------------------------------------------+ +| **⚠️ Option B Decision --- Transparent Degradation, Not Silent | +| Failure** | +| | +| TaintAnalyzer and DataFlowAnalyzer stay in the dispatch pipeline, | +| exception-isolated. | +| | +| When they hit an unusual AST node, output must show: | +| | +| ⚠ TaintAnalyzer skipped src/api/views.py --- Unsupported AST node | +| (walrus operator, line 47) | +| | +| ⚠ This file was NOT checked for taint vulnerabilities. | +| | +| A file that was skipped must NEVER appear the same as a file that was | +| checked and found clean. | +| | +| Track and display the skip rate. If it exceeds 10% of files, escalate | +| to hardening immediately. | ++-----------------------------------------------------------------------+ + +**1.3 The Critical Gap --- What Is Completely Missing** + ++-----------------------------------------------------------------------+ +| **🔴 The Gap Between the Motto and the Reality** | +| | +| Motto: \"Safety-First Refactoring for Production Codebases\" | +| | +| Current reality: BackupRollback (recovers after failure). | +| FixRiskLevel (a label, not a proof). | +| | +| Neither of these PROVES a transform is safe before the filesystem is | +| modified. | +| | +| The Verification Engine --- the system that actually proves safety | +| --- does not exist yet. | +| | +| Building it is the entire MVP. | ++-----------------------------------------------------------------------+ + + ---------------------- ------------------------------------------------ + **Missing System** **Why This Is Critical** + + **Verification Nothing currently proves a transform is safe + Engine** before the real file is touched + + **\--dry-run on Developers cannot preview changes without + autofix** writing to disk --- table-stakes for any autofix + tool + + **Test Suite Gate** Existing tests are never run after a transform + --- regression breakage goes undetected + + **refactron verify The standalone CLI command that proves safety + command** does not exist + + **Import Integrity No detection of broken imports or new circular + Check** dependencies after a transform + + **SHA-256 cache mtime-only cache will serve stale ASTs after + invalidation** file edits --- wrong analysis results + ---------------------- ------------------------------------------------ + +**Part 2 --- Definitive MVP Scope** + ++-----------------------------------------------------------------------+ +| **📌 The Single Decision Filter** | +| | +| \"Does this feature directly prove that a refactoring is safe?\" | +| | +| If yes → it is in the MVP. | +| | +| If no → write it in BACKLOG.md and do not touch it until after the | +| MVP ships. | +| | +| This filter has already been applied. Everything below passed it. | ++-----------------------------------------------------------------------+ + +**2.1 The Five Hero Commands --- Nothing More** + +These five commands are the entire MVP. They must work flawlessly on any +real Python codebase before v1.1.0 ships. Not four of five. All five. + + ---------------------------- ------------------------------------------ + **Command** **What It Proves** + + **refactron analyze .** Deep static analysis --- 8 analyzers, rich + output grouped by severity, actionable + messages + + **refactron autofix . Preview every proposed fix as a unified + \--dry-run** diff --- nothing written to disk + + **refactron autofix . Apply only fixes that pass all 3 + \--verify** verification checks --- block everything + else + + **refactron verify \ Standalone verification --- runs all 3 + \--against \** checks, usable in CI without autofix + + **refactron rollback** Undo any applied fix --- restores from + backup, zero developer action required + ---------------------------- ------------------------------------------ + +**2.2 What Is Explicitly NOT In This MVP** + +These belong in BACKLOG.md. Writing them here so they are acknowledged +and then closed. + + ------------------------ ---------------------------------------------- + **Feature** **Why Post-MVP** + + LLM / RAG Integration Verification must work perfectly without AI. + Intelligence is layered on top of proven + safety. + + Pattern Learning Engine Requires real user feedback data to produce + anything meaningful. Useless before users + exist. + + Semantic Equivalence Post-MVP verification layer. Added after test + Check gate is proven solid. + + Type Consistency Check Optional, configurable. Not needed to prove + (mypy) basic safety. + + SARIF Output Format Enterprise CI unlock. Prioritise after SMB + validation. + + Prometheus / Telemetry No users means nothing meaningful to observe. + + refactron init command Quality of life. Does not prove safety. + + VSCode Extension CLI must be perfect first. Extension is CLI + under the hood. + + Multi-Language Support Python-only until the model is validated. + ------------------------ ---------------------------------------------- + +**2.3 Version: v1.1.0, Not v2.0.0** + +The Verification Engine is additive functionality. It is a new +subcommand and a new flag on autofix. No existing command behavior +changes. No existing config format changes. No breaking API changes. +**Per SemVer, this is v1.1.0.** + +A v2.0.0 bump would scare existing users and break anyone with +refactron\>=1.0,\<2 in their requirements. Reserve v2.0.0 for when the +core CLI interface needs fundamental restructuring. + +**Part 3 --- Verification Engine Architecture** + +**This is the most important engineering work in the entire MVP. +Everything else is polish. The Verification Engine is the reason v1.1.0 +is a different product from v1.0.15.** + +**3.1 The Inviolable Safety Rule** + ++-----------------------------------------------------------------------+ +| **The Original File Is Never Modified During Verification** | +| | +| All verification checks run against a temp file in an isolated | +| subprocess. | +| | +| The real file is written to disk ONLY after safe_to_apply is True. | +| | +| *This single rule --- not a sandbox, not a container --- is the | +| safety guarantee.* | ++-----------------------------------------------------------------------+ + +**3.2 Module Structure** + ++-----------------------------------------------------------------------+ +| **refactron/verification/** | +| | +| ├── \_\_init\_\_.py | +| | +| ├── engine.py ← VerificationEngine (main orchestrator --- called by | +| AutoFixEngine) | +| | +| ├── result.py ← VerificationResult (locked data contract --- define | +| before any code) | +| | +| ├── report.py ← VerificationReport (human-readable CLI output) | +| | +| └── checks/ | +| | +| ├── syntax.py ← SyntaxVerifier (Check 1 --- always runs, \< 50ms) | +| | +| ├── imports.py ← ImportIntegrityVerifier (Check 2 --- always runs, \< | +| 100ms) | +| | +| └── test_gate.py ← TestSuiteGate (Check 3 --- runs when tests exist, | +| 2--30s) | ++-----------------------------------------------------------------------+ + +**3.3 The VerificationResult Contract** + +Lock this dataclass before writing a single line of verification logic. +**Every downstream module --- AutoFixEngine, CLI output, tests --- +depends on this schema. Do not change it once locked.** + + ---------------------- ------------------- ------------------------------------ + **Field** **Type** **Purpose** + + **safe_to_apply** *bool* THE only field AutoFixEngine reads. + False = nothing is written. No + exceptions in code. + + **passed** *bool* All checks completed without a + blocking failure + + **checks_run** *List\[str\]* Names of every check that executed + in this run + + **checks_passed** *List\[str\]* Names of checks that returned a + clean result + + **checks_failed** *List\[str\]* Names of checks that blocked the + transform + + **blocking_reason** *Optional\[str\]* Shown to developer when blocked --- + must be actionable, not a stack + trace + + **confidence_score** *float 0.0--1.0* Composite confidence across all + checks that passed + + **verification_ms** *int* Total wall-clock time for the full + verification run in milliseconds + + **skipped_checks** *List\[str\]* Checks that were skipped (e.g. no + tests found) --- distinct from + passed or failed + ---------------------- ------------------- ------------------------------------ + +**3.4 The Three MVP Verification Checks** + + ------- ----------------- --------------- ---------------------------------- + **Check Name** **Speed** **What It Catches** + + **1** **Syntax \< 50ms Always Parse errors, CST corruption from + Validation** runs transform bugs, newly introduced + eval()/exec() calls, import count + drop + + **2** **Import \< 100ms Always Removed imports still used in + Integrity** runs code, new imports that cannot + resolve, circular dependencies + introduced + + **3** **Test Suite 2--30s Runs if Any test breakage in files that + Gate** tests exist import the changed module. + Import-graph mapped --- NOT full + suite. Hard kill at 45s. + ------- ----------------- --------------- ---------------------------------- + +**3.5 How Check 3 Maps Tests to Changed Files** + +Running the full test suite after every autofix is not practical. The +Test Suite Gate uses an import-graph mapper to find only the tests that +actually import the changed module. + +- Build a reverse import graph at analysis time: for every .py file, + record which modules it imports. + +- When a file is transformed, look up that file in the reverse graph + to find all files that import it. + +- Filter those files to only test files (names starting with test\_ or + in a tests/ directory). + +- Run only those test files via subprocess: pytest \--timeout=30 -x -q + \ + +- If no test files import the changed module --- pass with note \"No + tests cover this module.\" + +- If pytest exits non-zero for any reason --- block. Show the first + failing test output. + +- Hard kill at 45 seconds regardless of outcome --- never block a + developer workflow. + +**3.6 The Atomic Write Protocol** + +This is the technical implementation of the safety rule from Section +3.1. The sequence is non-negotiable. + + ------- ------------------------------------------------------------------- + **1** Generate the transformed code in memory. The real file has not been + touched. + + **2** Write transformed code to a temp file in the SAME directory as the + target (same filesystem required for atomic move). + + **3** Run all 3 verification checks against the temp file path, not the + real path. + + **4** If ANY check sets safe_to_apply = False: delete the temp file, show + blocking_reason, restore from backup if needed. Done. + + **5** If safe_to_apply = True: call os.replace(tmp_path, real_path). This + is atomic on POSIX --- no partial write is possible. + + **6** Preserve original file permissions on the replaced file. + + **7** Delete the temp file in a finally block --- it must never be left + on disk regardless of outcome. + ------- ------------------------------------------------------------------- + +**Part 4 --- The 5-Week Execution Plan** + ++-----------------------------------------------------------------------+ +| **⚠️ Why 5 Weeks Is The Right Number** | +| | +| The previous 16-week plan was a v2.0 product roadmap. An MVP ships in | +| 5 weeks. | +| | +| Repo connect is already done --- this removes \~4 days from the | +| original critical path. | +| | +| 5 weeks gives buffer for the hardest part (Test Suite Gate) without | +| over-engineering everything else. | +| | +| Week 5 is validation, not building. You ship what Week 4 produces, | +| not what Week 5 imagines. | ++-----------------------------------------------------------------------+ + ++-----------------------------------------------------------------------+ +| **PHASE 1** Stabilize What Exists Week 1 · Days 1--5 | ++-----------------------------------------------------------------------+ +| **Goal: Zero crashes running refactron analyze refactron/ on the | +| library itself** | +| | +| This is the stability baseline. No new features. No Verification | +| Engine code. Just making what exists rock-solid. | +| | +| **Day 1 --- Exception Isolation for All Analyzers** | +| | +| - Wrap every analyzer in try/except in the dispatcher --- one bad | +| AST node must NEVER crash the full run | +| | +| - TaintAnalyzer and DataFlowAnalyzer: on exception, emit a visible | +| ⚠ warning showing the file and reason | +| | +| - Warning format: *⚠ TaintAnalyzer skipped src/api.py --- | +| Unsupported node (line 47). This file was NOT checked for taint.* | +| | +| - Add skip_rate counter --- track files analyzed vs skipped per | +| analyzer run | +| | +| - If skip_rate \> 10% at run end, add a summary warning in the CLI | +| output | +| | +| **Day 2 --- Cache Hardening + Backup Validation** | +| | +| - Replace mtime-only AST cache invalidation with SHA-256 file hash | +| --- mtime alone is unreliable | +| | +| - Add \--no-cache flag for debugging and reproducibility in CI | +| environments | +| | +| - Test BackupRollbackSystem when git is NOT initialized --- verify | +| plain file backup path works correctly | +| | +| - Add file integrity hash check before and after every transform | +| --- detect silent corruption | +| | +| **Day 3 --- Dry-Run Mode (Table-Stakes for Any Autofix Tool)** | +| | +| - Add \--dry-run flag to **refactron autofix** --- shows unified | +| diff, writes nothing to disk | +| | +| - Dry-run output format: coloured unified diff per file, summary of | +| what would change | +| | +| - Add \--diff flag as alias for \--dry-run for discoverability | +| | +| - This is the single most important UX addition --- no developer | +| trusts autofix without preview | +| | +| **Day 4 --- Test Harness Construction** | +| | +| - Create tests/fixtures/ directory with 5 Python files containing | +| known issues: | +| | +| - fixture_sql_injection.py --- raw SQL with f-string user input | +| | +| - fixture_unused_imports.py --- 6 unused imports across 3 files | +| | +| - fixture_complexity.py --- function with cyclomatic complexity | +| 18 | +| | +| - fixture_bad_extract.py --- function extraction that changes | +| return type (known-bad transform) | +| | +| - fixture_safe_extract.py --- function extraction that is | +| provably safe (known-good transform) | +| | +| - Write integration test: full analyze → autofix → rollback cycle | +| on fixture project passes | +| | +| **Day 5 --- Gate Check** | +| | +| - Run: **refactron analyze refactron/** --- the library analyzing | +| itself | +| | +| - Fix every crash, every confusing error message, every missing | +| exception handler found | +| | +| - All 135+ existing tests must still pass --- no regressions from | +| Days 1--4 | +| | +| - Record baseline: how many files analyzed, how many skipped, what | +| the output looks like | +| | +| +------------------------------------------------------------------+ | +| | **✅ Week 1 Gate --- Must Pass Before Week 2 Starts** | | +| | | | +| | refactron analyze refactron/ completes with zero crashes. | | +| | | | +| | All 135+ existing tests pass. | | +| | | | +| | TaintAnalyzer skip warnings are visible and correctly formatted. | | +| | | | +| | \--dry-run shows a diff without writing any files. | | +| +------------------------------------------------------------------+ | ++-----------------------------------------------------------------------+ + ++-----------------------------------------------------------------------+ +| **PHASE 2** Build the Verification Engine Weeks 2--3 · Days 6--15 | ++-----------------------------------------------------------------------+ +| **Goal: Verification Engine blocks 100% of known-bad transforms in | +| the fixture test suite** | +| | +| This is the hardest engineering work in the MVP. Ten days. Three | +| checks. One data contract. Nothing else. | +| | +| **Days 6--7 --- Lock the Contract, Write the Failing Test** | +| | +| - Create refactron/verification/ directory structure as specified | +| in Section 3.2 | +| | +| - Define **VerificationResult** in result.py --- lock every field | +| as specified in Section 3.3 | +| | +| - Write tests/verification/test_blocks_bad_transform.py --- this | +| test MUST fail on Day 6 | +| | +| - The fixture: fixture_bad_extract.py contains a function | +| extraction that changes the return type | +| | +| - The test asserts that VerificationEngine.verify() returns | +| safe_to_apply=False on this fixture | +| | +| - **The entire Days 8--15 effort is making this single test pass.** | +| | +| **Days 8--9 --- Check 1: Syntax Validation** | +| | +| - SyntaxVerifier.verify(original_code, transformed_code) → | +| VerificationResult | +| | +| - Step 1: Attempt libcst.parse_module(transformed_code) --- | +| ParserSyntaxError = block immediately | +| | +| - Step 2: CST round-trip --- parse → unparse → re-parse. If second | +| parse fails, CST was corrupted by transform | +| | +| - Step 3: AST walk on transformed code --- block if any new | +| ast.Call to eval/exec/os.system found | +| | +| - Step 4: Compare import counts --- if transformed has fewer import | +| statements, check if removed imports are still referenced | +| | +| - Target: \< 50ms on a 500-line file. If slower, profile and | +| optimise before moving to Check 2. | +| | +| **Days 10--11 --- Check 2: Import Integrity** | +| | +| - ImportIntegrityVerifier.verify(original_path, transformed_code) → | +| VerificationResult | +| | +| - Step 1: Extract all import statements from original and | +| transformed using ast.walk() | +| | +| - Step 2: For every import that existed in original but is absent | +| in transformed --- check if that name is still referenced in the | +| code. If yes, block. | +| | +| - Step 3: For every new import in transformed that was not in | +| original --- attempt importlib.util.find_spec(module_name). If | +| returns None, block. | +| | +| - Step 4: Build directed import graph, run DFS cycle detection. If | +| new cycle introduced, block. | +| | +| - Edge cases to handle: TYPE_CHECKING blocks, relative imports, | +| conditional imports inside if statements --- warn on these, do | +| not block | +| | +| **Days 12--14 --- Check 3: Test Suite Gate** | +| | +| - TestSuiteGate.verify(changed_file, transformed_code) → | +| VerificationResult | +| | +| - Step 1: Write transformed_code to a NamedTemporaryFile in the | +| same directory as changed_file (delete=False) | +| | +| - Step 2: Build reverse import graph from project root --- find all | +| test files that transitively import changed_file | +| | +| - Step 3: If no test files found → return VerificationResult with | +| checks_passed=\[\"test_gate\"\], note=\"No tests cover this | +| module\" | +| | +| - Step 4: Run subprocess.run(\[\"pytest\", \"\--timeout=30\", | +| \"-x\", \"-q\", \*relevant_test_files\], timeout=45, | +| capture_output=True) | +| | +| - Step 5: returncode == 0 → pass. Any other returncode → block with | +| first 500 chars of stdout as blocking_reason | +| | +| - Step 6: Always unlink temp file in finally block --- no orphaned | +| files on disk ever | +| | +| - Step 7: Hard kill: if subprocess times out at 45s → block with | +| \"Test suite gate timed out (45s limit)\" | +| | +| **Day 15 --- Wire into AutoFixEngine + Integration Tests** | +| | +| - AutoFixEngine calls VerificationEngine.verify(original_code, | +| transformed_code, file_path) before every os.write call | +| | +| - If result.safe_to_apply is False: log result.blocking_reason, | +| restore from backup, show in CLI output | +| | +| - If result.safe_to_apply is True: execute atomic write protocol | +| from Section 3.6 | +| | +| - Run full fixture test suite: fixture_bad_extract must be blocked, | +| fixture_safe_extract must be allowed | +| | +| - Verify that the inviolable safety rule holds: original file is | +| never modified when any check fails | +| | +| +------------------------------------------------------------------+ | +| | **✅ Week 3 Gate --- Must Pass Before Week 4 Starts** | | +| | | | +| | VerificationEngine correctly blocks fixture_bad_extract.py | | +| | (type-changing extraction). | | +| | | | +| | VerificationEngine correctly allows fixture_safe_extract.py | | +| | (unused import removal). | | +| | | | +| | 100% accuracy on the complete fixture test suite --- no | | +| | exceptions. | | +| | | | +| | Original file integrity verified: never modified when | | +| | safe_to_apply is False. | | +| | | | +| | test_blocks_bad_transform.py is now GREEN. | | +| +------------------------------------------------------------------+ | ++-----------------------------------------------------------------------+ + ++-----------------------------------------------------------------------+ +| **PHASE 3** Hero Commands + Output Quality + Real-World Testing Week | +| 4 · Days 16--20 | ++-----------------------------------------------------------------------+ +| **Goal: pip install refactron && refactron analyze . produces useful, | +| verified output in under 2 minutes** | +| | +| Engineering is done. This week is about whether a developer who has | +| never seen Refactron can use it without help. | +| | +| **Day 16 --- Output Quality Overhaul** | +| | +| - Group issues by file, then by severity: CRITICAL → HIGH → MEDIUM | +| → LOW | +| | +| - Show the problematic line of code inline beneath each issue --- | +| not just a line number | +| | +| - Add summary line at the bottom: \"3 critical · 7 high · 14 medium | +| (run \--autofix \--verify to fix 16)\" | +| | +| - Add \--format json for CI parsing and script consumption | +| | +| - Add \--fail-on CRITICAL flag for CI quality gate exit codes | +| (exits 1 if any CRITICAL found) | +| | +| - Error messages: replace every stack trace with a human-readable | +| message + suggested action | +| | +| **Day 17 --- refactron verify as Standalone Command** | +| | +| - **refactron verify \ \--against \** --- | +| standalone command that runs all 3 checks independently | +| | +| - Output format: | +| | +| - ✅ Syntax check (12ms) | +| | +| - ✅ Import integrity (8ms) | +| | +| - ✅ Tests passed (ran 4 test files in 6.2s) | +| | +| - Safe to apply. Confidence: 94.1% \| Total: 6.3s | +| | +| - This command works in CI on any PR that touches Python files --- | +| independent of autofix | +| | +| - This is the CI/CD integration story for the MVP --- document it | +| prominently | +| | +| **Days 18--19 --- Real-World Repo Testing** | +| | +| **Run Refactron on 5 real open-source Python repos. Do not ship until | +| all 5 pass without crashes.** | +| | +| | +| ---------------- ----------------------- --------------------------- | +| **Repository** **What to Test For** **Success Criteria** | +| | +| **Flask (80k Analyzer coverage, *Zero crashes. Skip | +| lines)** TaintAnalyzer skip warnings displayed | +| rate, import graph correctly.* | +| accuracy | +| | +| **Requests (15k Output quality, issue *Clean output in under 30 | +| lines)** grouping, summary seconds.* | +| accuracy | +| | +| **FastAPI (40k Autofix dry-run, diff *Diff is readable and | +| lines)** quality, no false correct.* | +| positives on modern | +| syntax | +| | +| **Httpx (25k Verification Engine on *Blocks 0 good transforms. | +| lines)** real autofix candidates Allows safe ones.* | +| | +| **Black (30k Self-hosting edge case *Zero crashes. Ironic but | +| lines)** --- a formatter critical.* | +| formatting itself | +| | +| ---------------- ----------------------- --------------------------- | +| | +| **Day 20 --- CLAUDE.md Update + v1.1.0 Pre-Release** | +| | +| - Update CLAUDE.md to include refactron/verification/ directory, | +| VerificationResult contract, and the inviolable safety rule | +| | +| - Add verification module architecture to CLAUDE.md architecture | +| section --- Claude Code needs this context | +| | +| - Create CHANGELOG.md entry for v1.1.0 listing all new capabilities | +| | +| - Publish v1.1.0-beta.1 to TestPyPI --- verify install and basic | +| commands work on a clean machine | +| | +| - Do NOT publish to PyPI yet --- Week 5 validation gates the real | +| release | +| | +| +------------------------------------------------------------------+ | +| | **✅ Week 4 Gate --- Must Pass Before Week 5 Starts** | | +| | | | +| | Fresh pip install on a real Python project → useful output in | | +| | under 2 minutes. | | +| | | | +| | refactron verify works as a standalone CI command. | | +| | | | +| | Zero crashes on all 5 real-world repos. | | +| | | | +| | All error messages are human-readable --- no stack traces | | +| | visible to end users. | | +| | | | +| | v1.1.0-beta.1 installs cleanly from TestPyPI. | | +| +------------------------------------------------------------------+ | ++-----------------------------------------------------------------------+ + ++-----------------------------------------------------------------------+ +| **PHASE 4** Real User Validation → Ship v1.1.0 Week 5 · Days 21--25 | ++-----------------------------------------------------------------------+ +| **Goal: 3 of 5 external developers say \"I would use this on my | +| production codebase\"** | +| | +| **You are not building this week.** You are watching real developers | +| use what you built and deciding whether it is ready to ship. | +| | +| **Days 21--22 --- Find and Brief 5 Target Developers** | +| | +| - Profile: senior engineers at scale-ups, 50k--500k line Python | +| codebases, real technical debt pain | +| | +| - Not colleagues who will be polite. Developers who will tell you | +| when something is broken. | +| | +| - Brief: \"Install it, run it on your codebase, tell me what | +| happens. I will not explain anything.\" | +| | +| - Give zero guidance --- the points of confusion they hit are your | +| next release priority list | +| | +| - Record sessions (with permission) or pair silently --- do not | +| explain, do not interrupt | +| | +| **Days 23--24 --- Observe, Document, Decide** | +| | +| - Watch where they get confused --- that is a UX bug, not user | +| error | +| | +| - Watch what they try first --- that tells you what your mental | +| model of the product is wrong about | +| | +| - Watch what makes them say something positive --- that is your | +| actual value proposition | +| | +| - Document every piece of feedback verbatim --- this becomes the | +| v1.2.0 backlog | +| | +| - Track: crashes found, confusing outputs, missing docs, unexpected | +| behaviors | +| | +| **Day 25 --- The Go/No-Go Decision** | +| | +| --------------------------------- --------------------------------- | +| **SHIP --- Conditions Met** **DO NOT SHIP --- Fix First** | +| | +| 3+ of 5 say \"I would use this on Fewer than 3 of 5 validate --- | +| my real codebase\" find the blocking problem | +| | +| Zero P0 crashes during any user Any crash that a user hits during | +| session testing | +| | +| Verification Engine blocks at Verification engine is bypassed | +| least 1 real bad transform or throws | +| | +| At least 1 developer asks \"When Users finish sessions confused | +| can my team get this?\" about what the tool does | +| --------------------------------- --------------------------------- | +| | +| - **If SHIP:** Publish v1.1.0 to PyPI. Post to Hacker News Show HN. | +| Write the launch blog post on refactron.dev/blog. | +| | +| - **If NO GO:** Do not publish. Fix the single most blocking issue | +| found. Retest with 2 of the same developers. Re-evaluate. | +| | +| +------------------------------------------------------------------+ | +| | **✅ Week 5 Gate --- The MVP Is Shipped** | | +| | | | +| | 3 of 5 external developers validated the core value proposition. | | +| | | | +| | Zero P0 crashes during all user sessions. | | +| | | | +| | v1.1.0 published to PyPI. | | +| | | | +| | Launch blog post live on refactron.dev/blog. | | +| | | | +| | Show HN post submitted. | | +| +------------------------------------------------------------------+ | ++-----------------------------------------------------------------------+ + +**Part 5 --- How Refactron Is Different** + +**5.1 The Only Column That Matters** + +Every competitor has at least one of these columns. No competitor has +all five. The Verifies column is empty across the entire market. That is +the product. + + ---------------- ----------- ----------- -------------- ------------ ----------- + **Tool** **Finds** **Fixes** **Verifies** **Learns** **Offline + CI** + + Claude Code \~ ✅ ❌ ❌ ❌ + + Cursor \~ ✅ ❌ \~ ❌ + + GitHub Copilot \~ \~ ❌ \~ ❌ + + Amazon Q ✅ \~ ❌ ❌ \~ + + SonarQube ✅ ❌ ❌ ❌ ✅ + + Semgrep ✅ ❌ ❌ ❌ ✅ + + Bandit \~ ❌ ❌ ❌ ✅ + + **Refactron ✅ ✅ **✅** ❌ ✅ + v1.1.0** + ---------------- ----------- ----------- -------------- ------------ ----------- + ++-----------------------------------------------------------------------+ +| **💡 The Research-Backed Positioning Numbers** | +| | +| AI tools break code in 63% of refactoring attempts --- CodeScene, | +| 100,000+ samples | +| | +| No single SAST tool detects more than 40% of real-world Python | +| vulnerabilities --- ICSE 2026 | +| | +| Developers with AI tools were 19% slower, but believed they were 20% | +| faster --- METR 2025 RCT | +| | +| Semgrep\'s experimental autofix produces incorrect code in 3.6% of | +| cases --- no verification step | +| | +| These numbers are your entire marketing message. Lead with them. | ++-----------------------------------------------------------------------+ + +**5.2 Claude Code Is Not a Competitor** + +This distinction matters for positioning and should be part of every +conversation about Refactron. + + ----------------------------------- ----------------------------------- + **Claude Code** **Refactron** + + You prompt it conversationally Runs automatically --- no prompting + + You review every diff manually Verification Engine proves it is + safe + + Non-deterministic --- different Deterministic --- identical result + result each run every time + + Requires Anthropic API to function Fully offline --- air-gapped CI + ready + + No static analysis engine Deep taint, data flow, CFG analysis + + No CI/CD quality gates Native PR blocking on CRITICAL + issues + + A smart colleague you pair with A safety inspector that runs + autonomously + + ***These are not competing. Claude ***A developer can use both in the + Code = active pairing. Refactron = same workflow without conflict.*** + autonomous safety enforcement.*** + ----------------------------------- ----------------------------------- + +**Part 6 --- Post-MVP Backlog (Priority Order)** + +Do not start any of these until the Week 5 gate is cleared and real user +feedback is in hand. The order below is based on what the research says +has the most impact, not on what is most interesting to build. + + -------- -------------------- ------------------------- ----------------------- ---------- + **\#** **Feature** **Why This Order** **Success Signal** **v** + + **1** **Harden Skip rate data from real *Skip rate drops below **v1.2** + TaintAnalyzer + users will show which AST 5% on real repos* + DataFlowAnalyzer** patterns to fix. Build + this based on evidence, + not guesses. + + **2** **Semantic Deepest verification *Blocks 1 transform **v1.2** + Equivalence Check layer. Only viable after that test gate missed* + (Check 4)** Test Suite Gate is proven + solid across real + codebases. + + **3** **LLM Suggestions Intelligence on top of *80%+ suggestion **v1.3** + (Groq, hardened)** proven safety. Never the approval rate in user + other way. LLM = advisor, testing* + never operator. + + **4** **RAG Context Reduces LLM *Measurable suggestion **v1.3** + (ChromaDB, hallucinations 60--80%. quality improvement* + project-aware)** Only valuable once the + LLM layer is stable. + + **5** **SARIF Output** Single biggest enterprise *Passes GitHub Advanced **v1.4** + CI/CD unlock. GitHub Security validation* + Advanced Security, Azure + DevOps, GitLab all + consume it. + + **6** **Pattern Learning Needs real user *Measurable noise **v1.5** + Engine** accept/reject data to reduction after 30 + produce anything. Useless days* + before you have users. + + **7** **refactron init Quality of life --- *Time-to-first-output **v1.5** + Command** auto-detect framework, under 90 seconds* + build RAG index, write + config. Not needed to + prove core value. + + **8** **Type Consistency Optional verification *Zero new false **v1.5** + (mypy, Check 5)** layer. Configurable, positives on real + default off. Catches type codebases* + regressions from + refactors. + + **9** **VSCode Extension** Only after CLI is *500+ installs in first **v2.0** + perfect. Extension is CLI month* + under the hood. Builds on + all the above. + -------- -------------------- ------------------------- ----------------------- ---------- + +**Part 7 --- Non-Negotiables and Today\'s Actions** + +**7.1 Five Rules That Cannot Be Broken** + +These are architectural invariants and process rules. If any of them +slip, the MVP is not a safety-first product. + + ------- --------------------------- ---------------------------------------- + **1** **Original file never All checks run against a temp file. The + touched during real file is written ONLY after + verification** safe_to_apply is True. This must be + enforced in code --- no way to bypass + it. + + **2** **Verification blocks 100% One bad transform slipping through + of known-bad transforms** destroys trust permanently. 99% accuracy + is not acceptable for a safety-first + tool. Validate against fixtures every CI + run. + + **3** **Zero crashes on all 5 Run on Flask, Requests, FastAPI, Httpx, + real-world repos before and Black before v1.1.0 goes to PyPI. + shipping** Fix every crash found. No exceptions. + + **4** **\--dry-run shows diff Developers must see what will change + before any write** before anything changes. No way to + bypass this. Autofix without dry-run is + not an MVP feature. + + **5** **External user validation 3 of 5 external developers must + before PyPI publish** validate. Internal testing is not + validation. The people who built it + cannot be the only people who test it. + ------- --------------------------- ---------------------------------------- + +**7.2 The Three Actions for Today --- Right Now** + + ------- ------------------------------ ------------------------------------------------- + **1** **Run: refactron analyze Point the library at itself. Every crash is Day 1 + refactron/** work. This is the fastest way to find what needs + fixing before building anything new. + + **2** **Create result.py and lock Define every field in the dataclass. Lock it. + VerificationResult** Every module downstream depends on this contract. + Build nothing else until it is locked. + + **3** **Write tests/verification/test_blocks_bad_transform.py + test_blocks_bad_transform.py must fail today. That failing test is the target. + --- make it fail** Weeks 2--3 are about making it pass. + ------- ------------------------------ ------------------------------------------------- + ++-----------------------------------------------------------------------+ +| The entire MVP is one sentence: | +| | +| **\"I ran Refactron on my production codebase.** | +| | +| ***It found real issues, fixed the safe ones, and proved --- with | +| evidence --- that nothing would break.\"*** | +| | +| Build only what is required for a developer to say that sentence. | +| | +| Repo connect is done. The plan is final. Open the terminal. | ++-----------------------------------------------------------------------+ diff --git a/refactron/analysis/__init__.py b/refactron/analysis/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/refactron/analysis/cfg/__init__.py b/refactron/analysis/cfg/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/refactron/analysis/cfg/builder.py b/refactron/analysis/cfg/builder.py new file mode 100644 index 0000000..8e1b2be --- /dev/null +++ b/refactron/analysis/cfg/builder.py @@ -0,0 +1,231 @@ +""" +Control Flow Graph Builder. +Converts Python AST into a Control Flow Graph. +""" + +import ast +from typing import List, Optional, Tuple + +from .node import CFGNode, EdgeType + + +class CFGBuilder: + def __init__(self) -> None: + self.nodes: List[CFGNode] = [] + self.current_id = 0 + self.current_block: Optional[CFGNode] = None + + # Stack for managing control flow targets + # loop_stack stores (break_target, continue_target) + self.loop_stack: List[Tuple[CFGNode, CFGNode]] = [] + + def _new_block(self) -> CFGNode: + """Create a new basic block.""" + node = CFGNode(id=self.current_id) + self.current_id += 1 + self.nodes.append(node) + return node + + def build_from_source(self, source_code: str) -> CFGNode: + """Build CFG from source code string.""" + tree = ast.parse(source_code) + return self.build_from_ast(tree) + + def build_from_ast(self, tree: ast.AST) -> CFGNode: + """Build CFG from AST.""" + entry_block = self._new_block() + self.current_block = entry_block + + # We handle function definitions specially if we want interprocedural analysis later + # For now, we process top-level code or body of functions + if isinstance(tree, ast.Module): + self._process_statements(tree.body) + elif isinstance(tree, (ast.FunctionDef, ast.AsyncFunctionDef)): + self._process_statements(tree.body) + else: + # Fallback for snippets + self._visit(tree) + + return entry_block + + def _process_statements(self, statements: List[ast.stmt]) -> None: + """Process a list of statements sequentially.""" + for stmt in statements: + self._visit(stmt) + + def _visit(self, node: ast.AST) -> None: + """Dispatch visitor method.""" + method_name = f"_visit_{node.__class__.__name__}" + visitor = getattr(self, method_name, self._visit_generic) + visitor(node) + + def _visit_generic(self, node: ast.AST) -> None: + """Default visitor for simple statements.""" + if self.current_block is None: + # Unreachable code or detached block + self.current_block = self._new_block() + + self.current_block.statements.append(node) + + def _visit_If(self, node: ast.If) -> None: + """Handle if statements.""" + if self.current_block is None: + return + + # Condition evaluation is in the current block + self.current_block.statements.append(node.test) + condition_block = self.current_block + + # Prepare blocks + then_block = self._new_block() + else_block = self._new_block() if node.orelse else None + join_block = self._new_block() + + # Connect condition to branches + condition_block.add_successor(then_block, EdgeType.TRUE) + if else_block: + condition_block.add_successor(else_block, EdgeType.FALSE) + else: + condition_block.add_successor(join_block, EdgeType.FALSE) + + # Process THEN branch + self.current_block = then_block + self._process_statements(node.body) + if self.current_block: # If didn't return/break/raise + self.current_block.add_successor(join_block, EdgeType.NORMAL) + + # Process ELSE branch + if else_block: + self.current_block = else_block + self._process_statements(node.orelse) + if self.current_block: + self.current_block.add_successor(join_block, EdgeType.NORMAL) + + self.current_block = join_block + + def _visit_For(self, node: ast.For) -> None: + """Handle for loops.""" + if self.current_block is None: + return + + # Initialization logic (iterating) happens in loop_head + loop_head = self._new_block() + self.current_block.add_successor(loop_head, EdgeType.NORMAL) + + loop_body = self._new_block() + loop_exit = self._new_block() + + # Head connects to Body (True) and Exit (False/Done) + loop_head.statements.append(node.iter) # Approximation + loop_head.add_successor(loop_body, EdgeType.TRUE) + loop_head.add_successor(loop_exit, EdgeType.FALSE) + + # Push loop context for break/continue + self.loop_stack.append((loop_exit, loop_head)) + + # Process Body + self.current_block = loop_body + # Assignment of target happens at start of body + self.current_block.statements.append(node.target) + self._process_statements(node.body) + + # Loop back + if self.current_block: + self.current_block.add_successor(loop_head, EdgeType.NORMAL) + + # Handle orelse (executed if loop finishes normally, not via break) + if node.orelse: + orelse_block = self._new_block() + # The "False" edge from head actually goes to orelse if present + # Fix previous connection + loop_head.successors = [s for s in loop_head.successors if s[1] != EdgeType.FALSE] + loop_head.add_successor(orelse_block, EdgeType.FALSE) + + self.current_block = orelse_block + self._process_statements(node.orelse) + if self.current_block: + self.current_block.add_successor(loop_exit, EdgeType.NORMAL) + + self.loop_stack.pop() + self.current_block = loop_exit + + def _visit_While(self, node: ast.While) -> None: + """Handle while loops.""" + if self.current_block is None: + return + + loop_head = self._new_block() + self.current_block.add_successor(loop_head, EdgeType.NORMAL) + + loop_body = self._new_block() + loop_exit = self._new_block() + + # Head evaluates condition + loop_head.statements.append(node.test) + loop_head.add_successor(loop_body, EdgeType.TRUE) + loop_head.add_successor(loop_exit, EdgeType.FALSE) + + self.loop_stack.append((loop_exit, loop_head)) + + # Body + self.current_block = loop_body + self._process_statements(node.body) + if self.current_block: + self.current_block.add_successor(loop_head, EdgeType.NORMAL) + + # Orelse + if node.orelse: + orelse_block = self._new_block() + loop_head.successors = [s for s in loop_head.successors if s[1] != EdgeType.FALSE] + loop_head.add_successor(orelse_block, EdgeType.FALSE) + + self.current_block = orelse_block + self._process_statements(node.orelse) + if self.current_block: + self.current_block.add_successor(loop_exit, EdgeType.NORMAL) + + self.loop_stack.pop() + self.current_block = loop_exit + + def _visit_Break(self, node: ast.Break) -> None: + """Handle break statement.""" + if self.current_block is None: + return + + self.current_block.statements.append(node) + if self.loop_stack: + break_target, _ = self.loop_stack[-1] + self.current_block.add_successor(break_target, EdgeType.NORMAL) + + # Code after break is unreachable in this block + self.current_block = None + + def _visit_Continue(self, node: ast.Continue) -> None: + """Handle continue statement.""" + if self.current_block is None: + return + + self.current_block.statements.append(node) + if self.loop_stack: + _, continue_target = self.loop_stack[-1] + self.current_block.add_successor(continue_target, EdgeType.NORMAL) + + self.current_block = None + + def _visit_Return(self, node: ast.Return) -> None: + """Handle return statement.""" + if self.current_block is None: + return + + self.current_block.statements.append(node) + # In a full implementation, this connects to the Exit block of the function + self.current_block = None + + def _visit_Raise(self, node: ast.Raise) -> None: + """Handle raise statement.""" + if self.current_block is None: + return + + self.current_block.statements.append(node) + # Connects to exception handler or Exit + self.current_block = None diff --git a/refactron/analysis/cfg/node.py b/refactron/analysis/cfg/node.py new file mode 100644 index 0000000..72e7b16 --- /dev/null +++ b/refactron/analysis/cfg/node.py @@ -0,0 +1,33 @@ +""" +Control Flow Graph Node definition. +Represents a basic block in the control flow graph. +""" + +from dataclasses import dataclass, field +from enum import Enum +from typing import Any, List, Tuple + + +class EdgeType(Enum): + NORMAL = "normal" + TRUE = "true" # Branch taken + FALSE = "false" # Branch not taken + EXCEPTION = "exception" + + +@dataclass +class CFGNode: + id: int + statements: List[Any] = field(default_factory=list) # AST nodes in this block + predecessors: List["CFGNode"] = field(default_factory=list) + successors: List[Tuple["CFGNode", EdgeType]] = field(default_factory=list) + + def add_successor(self, node: "CFGNode", edge_type: EdgeType = EdgeType.NORMAL) -> None: + self.successors.append((node, edge_type)) + node.predecessors.append(self) + + def __hash__(self) -> int: + return self.id + + def __repr__(self) -> str: + return f"CFGNode(id={self.id}, stmts={len(self.statements)})" diff --git a/refactron/analysis/data_flow.py b/refactron/analysis/data_flow.py new file mode 100644 index 0000000..d591896 --- /dev/null +++ b/refactron/analysis/data_flow.py @@ -0,0 +1,117 @@ +""" +Data Flow Analysis Engine. +Implements standard data flow analyses like Reaching Definitions. +""" + +import ast +from collections import defaultdict +from typing import Dict, List, Set, Tuple + +from .cfg.node import CFGNode + + +class DataFlowAnalyzer: + def __init__(self, cfg_entry: CFGNode): + self.entry_node = cfg_entry + self.nodes = self._collect_nodes(cfg_entry) + + def _collect_nodes(self, entry: CFGNode) -> List[CFGNode]: + """BFS to collect all reachable nodes.""" + nodes = [] + visited = set() + queue = [entry] + visited.add(entry.id) + + while queue: + node = queue.pop(0) + nodes.append(node) + for succ, _ in node.successors: + if succ.id not in visited: + visited.add(succ.id) + queue.append(succ) + return sorted(nodes, key=lambda n: n.id) + + def compute_reaching_definitions(self) -> Dict[int, Set[Tuple[str, int]]]: + """ + Compute Reaching Definitions for each block. + Returns a map: node_id -> set of (variable_name, definition_node_id) + definition_node_id can be the CFG node ID where it was defined. + """ + # specialized sets for gen/kill + # gen[n]: definitions generated in block n + # kill[n]: definitions killed in block n + gen: Dict[int, Set[Tuple[str, int]]] = defaultdict(set) + kill: Dict[int, Set[str]] = defaultdict(set) + + # 1. Initialize Gen/Kill sets for each block + for node in self.nodes: + node_gen: Set[Tuple[str, int]] = set() + node_kill = set() + + # Iterate statements to find assignments + for stmt in node.statements: + if isinstance(stmt, (ast.Assign, ast.AnnAssign)): + targets = [] + if isinstance(stmt, ast.Assign): + targets = stmt.targets + else: + targets = [stmt.target] + + for target in targets: + if isinstance(target, ast.Name): + var_name = target.id + # This definition kills previous definitions of 'var_name' + node_kill.add(var_name) + # And generates a new definition at this node + # Filter out any previous gen for the same var in this block + node_gen = {(v, d) for v, d in node_gen if v != var_name} + node_gen.add((var_name, node.id)) + + gen[node.id] = node_gen + kill[node.id] = node_kill + + # 2. Iterative Worklist Algorithm + # in_set[n] = union(out_set[p] for p in predecessors) + # out_set[n] = gen[n] union (in_set[n] - kill[n]) + + in_sets: Dict[int, Set[Tuple[str, int]]] = defaultdict(set) + out_sets: Dict[int, Set[Tuple[str, int]]] = defaultdict(set) + + changed = True + while changed: + changed = False + for node in self.nodes: + # Compute IN set + new_in = set() + for pred in node.predecessors: + new_in.update(out_sets[pred.id]) + + if new_in != in_sets[node.id]: + in_sets[node.id] = new_in + # recompute OUT causes change? + # actually OUT depends on IN, so we just check if OUT changes + + # Compute OUT set + # kill[node.id] is set of var names. + # We remove any definition (v, d) where v is in kill set. + preserved = {(v, d) for v, d in new_in if v not in kill[node.id]} + new_out = gen[node.id].union(preserved) + + if new_out != out_sets[node.id]: + out_sets[node.id] = new_out + changed = True + + return in_sets + + def find_variable_usages(self) -> Dict[str, List[int]]: + """ + Find where variables are used. + Returns: var_name -> list of node_ids where used + """ + usages = defaultdict(list) + for node in self.nodes: + for stmt in node.statements: + for child in ast.walk(stmt): + if isinstance(child, ast.Name) and isinstance(child.ctx, ast.Load): + usages[child.id].append(node.id) + return usages diff --git a/refactron/analysis/symbol_table.py b/refactron/analysis/symbol_table.py new file mode 100644 index 0000000..de9bbf1 --- /dev/null +++ b/refactron/analysis/symbol_table.py @@ -0,0 +1,255 @@ +""" +Symbol Table implementation for semantic analysis. +Maps classes, functions, variables, and their relationships across the codebase. +""" + +import json +import logging +from dataclasses import dataclass, field +from enum import Enum +from pathlib import Path +from typing import Any, Dict, List, Optional + +from refactron.core.inference import InferenceEngine + +logger = logging.getLogger(__name__) + + +class SymbolType(str, Enum): + CLASS = "class" + FUNCTION = "function" + VARIABLE = "variable" + MODULE = "module" + IMPORT = "import" + + +@dataclass +class Symbol: + name: str + type: SymbolType + file_path: str + line_number: int + scope: str # "global", "class:ClassName", "function:func_name" + definition_node: Any = field(default=None, repr=False) # AST/Astroid node + references: List[tuple] = field(default_factory=list) # List of (file_path, line_number) + + # Type inference data + inferred_type: Optional[str] = None + + def to_dict(self) -> Dict[str, Any]: + """Serialize for caching (excluding the definition node).""" + return { + "name": self.name, + "type": self.type.value, + "file_path": self.file_path, + "line_number": self.line_number, + "scope": self.scope, + "references": self.references, + "inferred_type": self.inferred_type, + } + + @classmethod + def from_dict(cls, data: Dict[str, Any]) -> "Symbol": + """Deserialize from dictionary.""" + data["type"] = SymbolType(data["type"]) + return cls(**data) + + +@dataclass +class SymbolTable: + # Map: file_path -> { scope -> { name -> Symbol } } + symbols: Dict[str, Dict[str, Dict[str, Symbol]]] = field(default_factory=dict) + # Map: global_name -> Symbol (for easy cross-file lookup of exports) + exports: Dict[str, Symbol] = field(default_factory=dict) + + def add_symbol(self, symbol: Symbol) -> None: + """Add a symbol to the table.""" + if symbol.file_path not in self.symbols: + self.symbols[symbol.file_path] = {} + + if symbol.scope not in self.symbols[symbol.file_path]: + self.symbols[symbol.file_path][symbol.scope] = {} + + self.symbols[symbol.file_path][symbol.scope][symbol.name] = symbol + + # Track global exports (top-level functions and classes) + if symbol.scope == "global" and symbol.type in ( + SymbolType.CLASS, + SymbolType.FUNCTION, + SymbolType.VARIABLE, + ): + # Key by module path + name? Or just name for now? + # Using simple name collision strategy for MVP + self.exports[symbol.name] = symbol + + def get_symbol(self, file_path: str, name: str, scope: str = "global") -> Optional[Symbol]: + """Retrieve a symbol.""" + return self.symbols.get(file_path, {}).get(scope, {}).get(name) + + def resolve_reference( + self, name: str, current_file: str, current_scope: str + ) -> Optional[Symbol]: + """ + Attempt to resolve a name to a definition. + 1. Check local scope + 2. Check global scope of current file + 3. Check exports (cross-file) + """ + # 1. Local scope + local = self.get_symbol(current_file, name, current_scope) + if local: + return local + + # 2. File Global scope + if current_scope != "global": + file_global = self.get_symbol(current_file, name, "global") + if file_global: + return file_global + + # 3. Cross-file exports (Naive implementation) + # TODO: Enhance this with proper import resolution + return self.exports.get(name) + + +class SymbolTableBuilder: + """Builds and manages the project-wide symbol table.""" + + def __init__(self, cache_dir: Optional[Path] = None): + self.symbol_table = SymbolTable() + self.cache_dir = cache_dir + self.inference_engine = InferenceEngine() + + def build_for_project(self, project_root: Path) -> SymbolTable: + """Scan project and build symbol table.""" + if self.cache_dir: + cached = self._load_cache() + if cached: + # TODO: Implement incremental update logic here + return cached + + python_files = list(project_root.rglob("*.py")) + for file_path in python_files: + self._analyze_file(file_path) + + if self.cache_dir: + self._save_cache() + + return self.symbol_table + + def _analyze_file(self, file_path: Path) -> None: + """Analyze a single file and populate symbols.""" + try: + # We use astroid for better inference capabilities later + tree = self.inference_engine.parse_file(str(file_path)) + + # Walk the tree + self._visit_node(tree, str(file_path), "global") + + except Exception as e: + logger.warning(f"Failed to build symbol table for {file_path}: {e}") + + def _visit_node(self, node: Any, file_path: str, scope: str) -> None: + """Recursive node visitor.""" + import astroid.nodes as nodes + + new_scope = scope + + if isinstance(node, (nodes.ClassDef, nodes.FunctionDef)): + # Register the definition itself in the CURRENT scope + symbol_type = ( + SymbolType.CLASS if isinstance(node, nodes.ClassDef) else SymbolType.FUNCTION + ) + symbol = Symbol( + name=node.name, + type=symbol_type, + file_path=file_path, + line_number=node.lineno, + scope=scope, + definition_node=node, + ) + self.symbol_table.add_symbol(symbol) + + # Enter new scope + prefix = "class" if isinstance(node, nodes.ClassDef) else "function" + new_scope = f"{prefix}:{node.name}" + + elif isinstance(node, nodes.AssignName): + # Register variable assignment + symbol = Symbol( + name=node.name, + type=SymbolType.VARIABLE, + file_path=file_path, + line_number=node.lineno, + scope=scope, + definition_node=node, + ) + # Try to infer type + try: + symbol.inferred_type = self.inference_engine.get_node_type_name(node) + except Exception: + pass + + self.symbol_table.add_symbol(symbol) + + # Recurse children + if hasattr(node, "get_children"): + for child in node.get_children(): + self._visit_node(child, file_path, new_scope) + + def _save_cache(self) -> None: + """Save symbol table to cache.""" + if not self.cache_dir: + return + + try: + self.cache_dir.mkdir(parents=True, exist_ok=True) + cache_file = self.cache_dir / "symbols.json" + + data = { + "symbols": { + f: { + s: {n: sym.to_dict() for n, sym in names.items()} + for s, names in scopes.items() + } + for f, scopes in self.symbol_table.symbols.items() + }, + "exports": {n: sym.to_dict() for n, sym in self.symbol_table.exports.items()}, + } + + with open(cache_file, "w") as f: + json.dump(data, f) + except Exception as e: + logger.warning(f"Failed to save symbol table cache: {e}") + + def _load_cache(self) -> Optional[SymbolTable]: + """Load symbol table from cache.""" + if not self.cache_dir: + return None + + cache_file = self.cache_dir / "symbols.json" + if not cache_file.exists(): + return None + + try: + with open(cache_file, "r") as f: + data = json.load(f) + + table = SymbolTable() + + # Reconstruct symbols + for f_path, scopes in data.get("symbols", {}).items(): + table.symbols[f_path] = {} + for scope_name, names in scopes.items(): + table.symbols[f_path][scope_name] = {} + for name, sym_data in names.items(): + table.symbols[f_path][scope_name][name] = Symbol.from_dict(sym_data) + + # Reconstruct exports + for name, sym_data in data.get("exports", {}).items(): + table.exports[name] = Symbol.from_dict(sym_data) + + return table + + except Exception as e: + logger.warning(f"Failed to load symbol table cache: {e}") + return None diff --git a/refactron/analysis/taint.py b/refactron/analysis/taint.py new file mode 100644 index 0000000..92da4f1 --- /dev/null +++ b/refactron/analysis/taint.py @@ -0,0 +1,273 @@ +""" +Taint Analysis Engine. +Tracks data flow from untrusted sources to sensitive sinks. +""" + +import ast +from collections import defaultdict +from dataclasses import dataclass, field +from typing import Dict, List, NamedTuple, Set, Tuple + +from .cfg.node import CFGNode +from .data_flow import DataFlowAnalyzer + + +class TaintSource(NamedTuple): + name: str # Function/Attribute name or variable name + type: str # 'function', 'variable', 'attribute' + + +class TaintSink(NamedTuple): + name: str + type: str + arg_index: int = 0 # Which argument is sensitive? + + +@dataclass +class TaintConfig: + sources: List[TaintSource] = field(default_factory=list) + sinks: List[TaintSink] = field(default_factory=list) + sanitizers: List[str] = field(default_factory=list) # Functions that clean taint + + +# Default configuration for Python web/security context +DEFAULT_TAINT_CONFIG = TaintConfig( + sources=[ + TaintSource("request", "variable"), # Flask/Django request + TaintSource("input", "function"), # Standard input + TaintSource("os.environ", "variable"), + TaintSource("sys.argv", "variable"), + TaintSource("argparse.ArgumentParser.parse_args", "function"), + TaintSource("parse_args", "function"), # Common alias for parser.parse_args() + ], + sinks=[ + TaintSink("eval", "function", 0), + TaintSink("exec", "function", 0), + TaintSink("os.system", "function", 0), + TaintSink("subprocess.call", "function", 0), + TaintSink("subprocess.run", "function", 0), + TaintSink("sqlite3.execute", "function", 0), # SQL Injection + ], + sanitizers=[ + "escape", + "int", # Casting to int usually sanitizes injection + "float", + ], +) + + +@dataclass +class TaintVulnerability: + source: str + sink: str + node_id: int + line_number: int + variable: str + message: str + + +class TaintAnalyzer: + def __init__(self, cfg_entry: CFGNode, config: TaintConfig = DEFAULT_TAINT_CONFIG): + self.cfg_entry = cfg_entry + self.config = config + self.data_flow = DataFlowAnalyzer(cfg_entry) + + # Helper maps for fast lookup + self._sources = {s.name for s in config.sources} + self._sinks = {s.name: s for s in config.sinks} + self._sanitizers = set(config.sanitizers) + + def analyze(self) -> List[TaintVulnerability]: + """ + Perform taint analysis. + Returns a list of detected vulnerabilities. + """ + # 1. Compute reaching definitions to help propagation + # reaching_defs = self.data_flow.compute_reaching_definitions() + + # 2. Iterative Taint Propagation + # taint_set[n] = set of tainted variable names at start of block n + tainted_vars: Dict[int, Set[str]] = defaultdict(set) + + vulnerabilities = [] + + # Simple iterative fixed-point might not be enough if we want path sensitivity + # For MVP, we use a block-level propagation + + # Initialize worklist with all nodes (or just entry) + # Using a topological sort would be better, but standard BFS/worklist works + + # We'll re-use the nodes list from data flow for iteration order + nodes = self.data_flow.nodes + + changed = True + while changed: + changed = False + for node in nodes: + # Compute IN taint from predecessors + incoming_taint = set() + for pred in node.predecessors: + incoming_taint.update(tainted_vars[pred.id]) + + # Process block statements + current_taint = incoming_taint.copy() + + for stmt in node.statements: + # Check for Sink usage + vuls = self._check_sink(stmt, current_taint, node.id) + # We accumulate vulnerabilities but continue analysis + # De-duplicate vuls? + for v in vuls: + if v not in vulnerabilities: + vulnerabilities.append(v) + + # Update Taint (Sources & Propagation) + new_taints, cleansed = self._propagate_taint(stmt, current_taint) + current_taint.update(new_taints) + current_taint.difference_update(cleansed) + + # If OUT taint changed, we might need to re-process successors? + # For this simple implementation, we assume intra-block propagation is enough + # IF variables persist across blocks. + + # Optimization: check if OUT set changed + if current_taint != tainted_vars[node.id]: + tainted_vars[node.id] = current_taint + changed = True + + return vulnerabilities + + def _propagate_taint(self, stmt: ast.AST, current_taint: Set[str]) -> Tuple[Set[str], Set[str]]: + """ + Analyze a statement and return (newly_tainted_vars, cleansed_vars). + """ + generated = set() + killed = set() + + if isinstance(stmt, (ast.Assign, ast.AnnAssign)): + targets = [] + if isinstance(stmt, ast.Assign): + targets = stmt.targets + else: + targets = [stmt.target] + + value = stmt.value # type: ignore[union-attr, attr-defined] + is_tainted = self._is_expression_tainted(value, current_taint) # type: ignore[arg-type] + + for target in targets: + if isinstance(target, ast.Name): + if is_tainted: + generated.add(target.id) + else: + # If assigning a clean value to a variable, it cleanses it + killed.add(target.id) + + return generated, killed + + def _is_expression_tainted(self, expr: ast.AST, current_taint: Set[str]) -> bool: + """Check if an expression evaluates to a tainted value.""" + if isinstance(expr, ast.Name): + # Check if variable is already tainted + if expr.id in current_taint: + return True + # Check if it's a direct source (e.g. 'request') + if expr.id in self._sources: + return True + + elif isinstance(expr, ast.Call): + # Check if function call returns taint (Source) + func_name = self._get_call_name(expr) + if func_name in self._sources: + return True + + # Check if function call propagates taint (Sanitizer check) + if func_name in self._sanitizers: + return False + + # Default: propagate if any arg is tainted + for arg in expr.args: + if self._is_expression_tainted(arg, current_taint): + return True + + elif isinstance(expr, ast.BinOp): + # Binary op is tainted if either side is tainted + return self._is_expression_tainted( + expr.left, current_taint + ) or self._is_expression_tainted(expr.right, current_taint) + + elif isinstance(expr, ast.JoinedStr): + # f-string: tainted if any value in it is tainted + for value in expr.values: + if isinstance(value, ast.FormattedValue): + if self._is_expression_tainted(value.value, current_taint): + return True + elif self._is_expression_tainted(value, current_taint): + return True + + elif isinstance(expr, ast.Subscript): + # Propagate from value (e.g. args.input) + if self._is_expression_tainted(expr.value, current_taint): + return True + + elif isinstance(expr, ast.Attribute): + # Check specific attributes (e.g. os.environ) + full_name = self._get_attribute_name(expr) + if full_name in self._sources: + return True + # Propagate from object (simple object taint) + if self._is_expression_tainted(expr.value, current_taint): + return True + + return False + + def _check_sink( + self, stmt: ast.AST, current_taint: Set[str], node_id: int + ) -> List[TaintVulnerability]: + """Check if a statement uses a tainted variable in a sink.""" + vuls = [] + + # We need to traverse the statement to find Call nodes + for node in ast.walk(stmt): + if isinstance(node, ast.Call): + func_name = self._get_call_name(node) + if func_name in self._sinks: + sink_def = self._sinks[func_name] + # Check the sensitive argument + if len(node.args) > sink_def.arg_index: + arg = node.args[sink_def.arg_index] + if self._is_expression_tainted(arg, current_taint): + # Identify which variable caused it for reporting + var_name = "expression" + if isinstance(arg, ast.Name): + var_name = arg.id + + vuls.append( + TaintVulnerability( + source="untrusted input", # Simplification + sink=func_name, + node_id=node_id, + line_number=getattr(node, "lineno", 0), + variable=var_name, + message=( + f"Tainted data from untrusted source reaches" + f" sensitive sink '{func_name}' via '{var_name}'" + ), + ) + ) + return vuls + + def _get_call_name(self, node: ast.Call) -> str: + """Extract function name from Call node.""" + if isinstance(node.func, ast.Name): + return node.func.id + elif isinstance(node.func, ast.Attribute): + return self._get_attribute_name(node.func) + return "" + + def _get_attribute_name(self, node: ast.Attribute) -> str: + """Recursive attribute name extraction (e.g. os.path.join).""" + if isinstance(node.value, ast.Name): + return f"{node.value.id}.{node.attr}" + elif isinstance(node.value, ast.Attribute): + return f"{self._get_attribute_name(node.value)}.{node.attr}" + return node.attr diff --git a/refactron/autofix/engine.py b/refactron/autofix/engine.py index 95a43a2..b16bea9 100644 --- a/refactron/autofix/engine.py +++ b/refactron/autofix/engine.py @@ -5,7 +5,9 @@ automatic fixes without requiring expensive AI APIs. """ -from typing import Dict +import tempfile +from pathlib import Path +from typing import Dict, List, Optional, Tuple from refactron.autofix.models import FixResult, FixRiskLevel from refactron.core.models import CodeIssue @@ -138,6 +140,61 @@ def fix_all(self, issues: list, code: str, preview: bool = True) -> Dict[int, Fi return results + def fix_file( + self, + file_path: Path, + issues: List[CodeIssue], + dry_run: bool = True, + ) -> Tuple[str, Optional[str]]: + """ + Apply all fixable issues to a file. + + In dry_run=True mode the file is never touched; the fixed code and a + unified diff are returned for display only. In dry_run=False mode the + fixed content is written atomically (temp-file → os.replace). + + Args: + file_path: Path to the Python file to fix. + issues: List of CodeIssue objects to attempt to fix. + dry_run: When True, no bytes are written to disk. + + Returns: + Tuple of (fixed_code, diff). diff is None/empty when no changes + were made; otherwise it is a unified-diff string. + """ + from refactron.autofix.file_ops import generate_diff + + code = file_path.read_text(encoding="utf-8") + current_code = code + + for issue in issues: + if not self.can_fix(issue): + continue + result = self.fix(issue, current_code, preview=False) + if result.success and result.fixed is not None: + current_code = result.fixed + + diff = generate_diff(code, current_code, file_path.name) + + if not dry_run and current_code != code: + # Atomic write: temp file in same directory → os.replace + tmp_fd, tmp_path = tempfile.mkstemp( + dir=file_path.parent, + prefix=f".{file_path.name}.", + suffix=".tmp", + ) + try: + with open(tmp_fd, "w", encoding="utf-8") as fh: + fh.write(current_code) + Path(tmp_path).replace(file_path) + finally: + try: + Path(tmp_path).unlink(missing_ok=True) + except Exception: + pass + + return current_code, diff if diff else None + class BaseFixer: """Base class for all automatic fixers.""" diff --git a/refactron/autofix/file_ops.py b/refactron/autofix/file_ops.py index 06b9b2d..6ab59e4 100644 --- a/refactron/autofix/file_ops.py +++ b/refactron/autofix/file_ops.py @@ -2,6 +2,7 @@ File operations for auto-fix system with backup and rollback support. """ +import difflib import json import shutil import tempfile @@ -10,6 +11,35 @@ from typing import Any, Dict, List, Optional +def generate_diff(original: str, modified: str, filename: str = "") -> str: + """ + Generate a unified diff between two code strings. + + Args: + original: The original file content. + modified: The modified file content. + filename: Filename shown in the diff header. + + Returns: + Unified diff string, or empty string when there are no differences. + """ + if original == modified: + return "" + + original_lines = original.splitlines(keepends=True) + modified_lines = modified.splitlines(keepends=True) + + diff_lines = list( + difflib.unified_diff( + original_lines, + modified_lines, + fromfile=f"a/{filename}", + tofile=f"b/{filename}", + ) + ) + return "".join(diff_lines) + + class FileOperations: """Handle file operations with safety guarantees.""" diff --git a/refactron/cli/analysis.py b/refactron/cli/analysis.py index dd29556..de25ac5 100644 --- a/refactron/cli/analysis.py +++ b/refactron/cli/analysis.py @@ -3,6 +3,8 @@ Commands for analyzing code, generating reports, and metrics. """ +import logging +import sys from pathlib import Path from typing import Optional @@ -14,6 +16,7 @@ _auth_banner, _create_summary_table, _interactive_file_selector, + _interactive_issue_viewer, _print_detailed_issues, _print_file_count, _print_helpful_tips, @@ -82,6 +85,18 @@ "overrides the selected profile." ), ) +@click.option( + "--no-cache", + is_flag=True, + default=False, + help="Disable incremental analysis cache — re-analyze all files from scratch", +) +@click.option( + "--no-interactive", + is_flag=True, + default=False, + help="Disable interactive mode — dump all issues (for CI/CD or piped output)", +) def analyze( target: Optional[str], config: Optional[str], @@ -92,6 +107,8 @@ def analyze( show_metrics: bool, profile: Optional[str], environment: Optional[str], + no_cache: bool, + no_interactive: bool, ) -> None: """ Analyze code for issues and technical debt. @@ -141,10 +158,13 @@ def analyze( # Override config with CLI options if log_level: cfg.log_level = log_level + logging.getLogger("refactron").setLevel(getattr(logging, log_level.upper())) if log_format: cfg.log_format = log_format if metrics is not None: cfg.enable_metrics = metrics + if no_cache: + cfg.enable_incremental_analysis = False _print_file_count(target_path) @@ -160,15 +180,21 @@ def analyze( # Display results summary = result.summary() - console.print(_create_summary_table(summary)) - console.print() + use_interactive = sys.stdout.isatty() and not no_interactive + + if use_interactive: + _interactive_issue_viewer(result, target_path) + else: + # Non-interactive: grouped dump for CI/CD or piped output + console.print(_create_summary_table(summary)) + console.print() - _print_status_messages(summary) + _print_status_messages(summary) - if detailed and result.all_issues: - _print_detailed_issues(result) + if detailed and result.all_issues: + _print_detailed_issues(result) - _print_helpful_tips(summary, detailed) + _print_helpful_tips(summary, detailed) # Show metrics if requested if show_metrics and cfg.enable_metrics: diff --git a/refactron/cli/refactor.py b/refactron/cli/refactor.py index fda3213..48315e6 100644 --- a/refactron/cli/refactor.py +++ b/refactron/cli/refactor.py @@ -242,6 +242,12 @@ def refactor( default=True, help="Preview fixes or apply them", ) +@click.option( + "--dry-run", + is_flag=True, + default=False, + help="Show a unified diff of what would change — writes nothing to disk", +) @click.option( "--safety-level", "-s", @@ -255,6 +261,7 @@ def autofix( profile: Optional[str], environment: Optional[str], preview: bool, + dry_run: bool, safety_level: str, ) -> None: """ @@ -287,8 +294,9 @@ def autofix( # Initialize auto-fix engine engine = AutoFixEngine(safety_level=safety) - if preview: - console.print("[warning]Preview mode: No changes will be applied[/warning]\n") + # --dry-run implies preview (no writes) + if dry_run or preview: + console.print("[warning]Dry-run mode: No changes will be written to disk[/warning]\n") else: console.print("[success]Apply mode: Changes will be written to files[/success]\n") diff --git a/refactron/cli/ui.py b/refactron/cli/ui.py index a27f902..e121c40 100644 --- a/refactron/cli/ui.py +++ b/refactron/cli/ui.py @@ -7,7 +7,8 @@ import random import sys import time -from typing import TYPE_CHECKING, Any, Optional +from dataclasses import dataclass, field +from typing import TYPE_CHECKING, Any, List, Optional, Set, Tuple import click from rich import box @@ -16,12 +17,14 @@ from rich.live import Live from rich.panel import Panel from rich.prompt import IntPrompt, Prompt +from rich.rule import Rule from rich.table import Table from rich.text import Text from rich.theme import Theme from refactron import __version__ from refactron.core.analysis_result import AnalysisResult +from refactron.core.models import IssueLevel from refactron.core.refactor_result import RefactorResult if TYPE_CHECKING: @@ -160,22 +163,372 @@ def _print_status_messages(summary: dict) -> None: ) -def _print_detailed_issues(result: AnalysisResult) -> None: - """Print detailed issues list.""" - console.print("[primary bold]Detailed Issues:[/primary bold]\n") - - for issue in result.all_issues: - style = ( - "error" - if issue.level.value in ("critical", "error") - else "warning" if issue.level.value == "warning" else "info" +def _relative_path(file_path: Any) -> str: + """Convert an absolute path to a relative one (from cwd), or just the filename.""" + from pathlib import Path as _Path + + try: + return str(_Path(file_path).relative_to(_Path.cwd())) + except ValueError: + return _Path(file_path).name + + +def _severity_style(level_name: str) -> str: + """Map severity name to a Rich style string.""" + return { + "critical": "bold red", + "error": "red", + "warning": "bold yellow", + "info": "cyan", + }.get(level_name, "dim") + + +_SEVERITY_ORDER = [ + ("critical", IssueLevel.CRITICAL), + ("error", IssueLevel.ERROR), + ("warning", IssueLevel.WARNING), + ("info", IssueLevel.INFO), +] + +# ─── Key constants ─────────────────────────────────────────────────── + +KEY_UP = "\x1b[A" +KEY_DOWN = "\x1b[B" +KEY_ENTER = "\r" + +# Type alias: list of (level_name, issues_list) tuples +TuiGroups = List[Tuple[str, list]] + + +@dataclass +class TuiState: + """Immutable-ish state for the TUI issue viewer.""" + + groups: TuiGroups + screen: str = "summary" # "summary" or "group" + cursor: int = 0 + current_group: int = 0 + expanded: Set[Tuple[int, int]] = field(default_factory=set) + quit: bool = False + + +def _build_tui_groups(result: AnalysisResult) -> TuiGroups: + """Build ordered list of (level_name, issues) for non-empty severity groups.""" + groups: TuiGroups = [] + for name, level in _SEVERITY_ORDER: + issues = result.issues_by_level(level) + if issues: + groups.append((name, issues)) + return groups + + +def _handle_key(state: TuiState, key: str) -> TuiState: + """Pure state transition: given current state + key, return new state.""" + if key == "q": + return TuiState( + groups=state.groups, + screen=state.screen, + cursor=state.cursor, + current_group=state.current_group, + expanded=state.expanded, + quit=True, + ) + + if state.screen == "summary": + return _handle_summary_key(state, key) + elif state.screen == "group": + return _handle_group_key(state, key) + + return state + + +def _handle_summary_key(state: TuiState, key: str) -> TuiState: + """Handle key press on the summary screen.""" + max_idx = len(state.groups) - 1 + + if key == KEY_DOWN: + new_cursor = min(state.cursor + 1, max_idx) + return TuiState( + groups=state.groups, + screen="summary", + cursor=new_cursor, + current_group=state.current_group, + expanded=state.expanded, + ) + elif key == KEY_UP: + new_cursor = max(state.cursor - 1, 0) + return TuiState( + groups=state.groups, + screen="summary", + cursor=new_cursor, + current_group=state.current_group, + expanded=state.expanded, + ) + elif key == KEY_ENTER: + return TuiState( + groups=state.groups, + screen="group", + cursor=0, + current_group=state.cursor, + expanded=state.expanded, + ) + + return state + + +def _handle_group_key(state: TuiState, key: str) -> TuiState: + """Handle key press on a group detail screen.""" + _, issues = state.groups[state.current_group] + max_idx = len(issues) - 1 + + if key == KEY_DOWN: + new_cursor = min(state.cursor + 1, max_idx) + return TuiState( + groups=state.groups, + screen="group", + cursor=new_cursor, + current_group=state.current_group, + expanded=state.expanded, ) - level_label = f"[{issue.level.value.upper()}]" + elif key == KEY_UP: + new_cursor = max(state.cursor - 1, 0) + return TuiState( + groups=state.groups, + screen="group", + cursor=new_cursor, + current_group=state.current_group, + expanded=state.expanded, + ) + elif key == KEY_ENTER: + toggle_key = (state.current_group, state.cursor) + new_expanded = set(state.expanded) + if toggle_key in new_expanded: + new_expanded.discard(toggle_key) + else: + new_expanded.add(toggle_key) + return TuiState( + groups=state.groups, + screen="group", + cursor=state.cursor, + current_group=state.current_group, + expanded=new_expanded, + ) + elif key == "b": + return TuiState( + groups=state.groups, + screen="summary", + cursor=state.current_group, + current_group=state.current_group, + expanded=state.expanded, + ) + elif key == "n": + if state.current_group < len(state.groups) - 1: + return TuiState( + groups=state.groups, + screen="group", + cursor=0, + current_group=state.current_group + 1, + expanded=state.expanded, + ) + return state + elif key == "p": + if state.current_group > 0: + return TuiState( + groups=state.groups, + screen="group", + cursor=0, + current_group=state.current_group - 1, + expanded=state.expanded, + ) + return state + + return state + + +def _read_key() -> str: + """Read a single keypress from stdin, handling escape sequences for arrow keys.""" + import termios + import tty + + fd = sys.stdin.fileno() + old_settings = termios.tcgetattr(fd) + try: + tty.setraw(fd) + ch = sys.stdin.read(1) + if ch == "\x1b": + ch2 = sys.stdin.read(1) + if ch2 == "[": + ch3 = sys.stdin.read(1) + return "\x1b[" + ch3 + return ch + ch2 + if ch == "\n": + return KEY_ENTER + return ch + finally: + termios.tcsetattr(fd, termios.TCSADRAIN, old_settings) + + +def _render_tui_summary(state: TuiState, target_path: Any) -> Text: + """Render the summary screen as a Rich Text object.""" + from pathlib import Path as _Path + + target = _Path(target_path) + label = target.name if target.is_file() else str(_relative_path(target)) + + total = sum(len(issues) for _, issues in state.groups) + + output = Text() + output.append(f"\n {total} issues", style="bold") + output.append(" found in ", style="") + output.append(f"{label}\n\n", style="bold") + + for idx, (level_name, issues) in enumerate(state.groups): + style = _severity_style(level_name) + is_selected = idx == state.cursor + prefix = " > " if is_selected else " " + row_style = "bold " + style if is_selected else style + + count = len(issues) + suffix = "" + if level_name == "critical" and count > 0: + suffix = " <-- needs attention" + + line = f"{prefix}{level_name.upper():<12} {count:>4}{suffix}\n" + output.append(line, style=row_style) + + output.append("\n ") + output.append("↑↓", style="bold") + output.append(" Navigate ", style="dim") + output.append("Enter", style="bold") + output.append(" Select ", style="dim") + output.append("q", style="bold") + output.append(" Quit\n", style="dim") + return output + + +def _render_tui_group(state: TuiState) -> Text: + """Render a severity group detail screen as a Rich Text object.""" + level_name, issues = state.groups[state.current_group] + style = _severity_style(level_name) + + output = Text() + output.append(f"\n ── {level_name.upper()} ({len(issues)}) ", style=style) + output.append("─" * 40 + "\n\n", style="dim") + + for idx, issue in enumerate(issues): + is_selected = idx == state.cursor + is_expanded = (state.current_group, idx) in state.expanded + prefix = " > " if is_selected else " " + rel = _relative_path(issue.file_path) + + # Issue line + issue_style = "bold " + style if is_selected else style + output.append(f"{prefix}{rel}:{issue.line_number}", style=issue_style) + output.append(f" {issue.message}\n", style="") + + if is_expanded: + snippet = _read_code_context(issue.file_path, issue.line_number, context=1) + if snippet: + for line_no, line_text in snippet: + marker = ">" if line_no == issue.line_number else " " + output.append(f" {marker} {line_no:>4}| {line_text}\n", style="dim") + if issue.suggestion: + output.append(f" Tip: {issue.suggestion}\n", style="dim") + output.append("\n") + + # Navigation bar + output.append("\n ") + output.append("↑↓", style="bold") + output.append(" Navigate ", style="dim") + output.append("Enter", style="bold") + output.append(" Expand/Collapse ", style="dim") + output.append("b", style="bold") + output.append(" Back ", style="dim") + if state.current_group < len(state.groups) - 1: + next_name = state.groups[state.current_group + 1][0].upper() + output.append("n", style="bold") + output.append(f" Next ({next_name}) ", style="dim") + if state.current_group > 0: + prev_name = state.groups[state.current_group - 1][0].upper() + output.append("p", style="bold") + output.append(f" Prev ({prev_name}) ", style="dim") + output.append("q", style="bold") + output.append(" Quit\n", style="dim") + + return output + + +def _read_code_context(file_path: Any, line_number: int, context: int = 1) -> Optional[list]: + """Read a few lines around *line_number* from the source file. + + Returns a list of ``(lineno, text)`` tuples, or ``None`` on failure. + """ + from pathlib import Path as _Path + + try: + lines = _Path(file_path).read_text(encoding="utf-8").splitlines() + start = max(0, line_number - 1 - context) + end = min(len(lines), line_number + context) + return [(i + 1, lines[i]) for i in range(start, end)] + except Exception: + return None + + +def _print_single_issue(issue: Any, show_code: bool = False) -> None: + """Print one issue with optional code context.""" + rel = _relative_path(issue.file_path) + style = _severity_style(issue.level.value) - console.print(f"[{style}]{level_label} {issue}[/{style}]") - if issue.suggestion: - console.print(f" [secondary]Tip: {issue.suggestion}[/secondary]") - console.print() + console.print(f" [{style}]{rel}:{issue.line_number}[/{style}] {issue.message}") + + if show_code: + snippet = _read_code_context(issue.file_path, issue.line_number, context=1) + if snippet: + for line_no, line_text in snippet: + marker = ">" if line_no == issue.line_number else " " + console.print(f" [dim] {marker} {line_no:>4}[/dim]| {line_text}") + + if issue.suggestion: + console.print(f" [dim] Tip: {issue.suggestion}[/dim]") + + console.print() + + +def _print_severity_group(level_name: str, issues: list) -> None: + """Print all issues for one severity group.""" + style = _severity_style(level_name) + show_code = level_name in ("critical", "error", "warning") + + console.print() + console.print(Rule(f" {level_name.upper()} ({len(issues)}) ", style=style)) + console.print() + + for issue in issues: + _print_single_issue(issue, show_code=show_code) + + +def _group_issues(result: AnalysisResult) -> dict: + """Group issues by severity level. Returns {name: [issues]} for non-empty groups.""" + groups: dict = {} + for name, level in _SEVERITY_ORDER: + issues = result.issues_by_level(level) + if issues: + groups[name] = issues + return groups + + +# ─── Non-interactive output (CI/CD, piped) ──────────────────────────────── + + +def _print_detailed_issues(result: AnalysisResult) -> None: + """Print issues grouped by severity (non-interactive mode).""" + groups = _group_issues(result) + if not groups: + return + + for level_name in ["critical", "error", "warning", "info"]: + if level_name in groups: + _print_severity_group(level_name, groups[level_name]) def _print_helpful_tips(summary: dict, detailed: bool) -> None: @@ -185,10 +538,74 @@ def _print_helpful_tips(summary: dict, detailed: bool) -> None: if summary["total_issues"] > 5: console.print( - "[secondary]Tip: Run 'refactron refactor --preview' to see suggested fixes[/secondary]" + "[secondary]Tip: Run 'refactron autofix --dry-run'" + " to preview fixes[/secondary]" ) +# ─── Interactive viewer ─────────────────────────────────────────────────── + + +def _erase_lines(n: int) -> None: + """Move cursor up *n* lines and clear everything below.""" + if n > 0: + sys.stdout.write(f"\x1b[{n}A\x1b[J") + sys.stdout.flush() + + +def _interactive_issue_viewer(result: AnalysisResult, target_path: Any) -> None: + """Interactive TUI issue browser with arrow key navigation (TTY only). + + Uses termios raw mode for key capture and manual cursor erasure for + flicker-free re-rendering. Arrow keys navigate, Enter expands/collapses, + q quits. + """ + groups = _build_tui_groups(result) + + if not groups: + console.print( + Panel( + "[success]Excellent! No issues found.[/success]", + box=box.ROUNDED, + border_style="success", + ) + ) + return + + state = TuiState(groups=groups) + last_height = 0 + + try: + while not state.quit: + # Erase previous frame + _erase_lines(last_height) + + # Build renderable + if state.screen == "summary": + renderable = _render_tui_summary(state, target_path) + else: + renderable = _render_tui_group(state) + + # Capture to count lines, then write to terminal + with console.capture() as cap: + console.print(renderable, highlight=False) + output = cap.get() + last_height = output.count("\n") + sys.stdout.write(output) + sys.stdout.flush() + + # Wait for keypress + key = _read_key() + state = _handle_key(state, key) + + # Erase the TUI on quit, leave a clean summary + _erase_lines(last_height) + total = sum(len(iss) for _, iss in groups) + console.print(f" [dim]{total} issues found. Done.[/dim]") + except (KeyboardInterrupt, EOFError): + _erase_lines(last_height) + + def _print_refactor_filters(types: tuple) -> None: """Print operation type filters if specified.""" if types: diff --git a/refactron/cli/utils.py b/refactron/cli/utils.py index d034390..025ff32 100644 --- a/refactron/cli/utils.py +++ b/refactron/cli/utils.py @@ -78,12 +78,24 @@ def _validate_api_key( def _setup_logging(verbose: bool = False) -> None: """Setup logging configuration.""" level = logging.DEBUG if verbose else logging.INFO + + # Remove duplicate handlers that accumulate when basicConfig is called + # multiple times (e.g. from tests or repeated CLI invocations). + root = logging.getLogger() + root.handlers.clear() + logging.basicConfig( level=level, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", datefmt="%Y-%m-%d %H:%M:%S", + force=True, ) + # Suppress refactron internal logs during CLI (INFO/DEBUG are diagnostics, not user output). + # Users can override with --log-level DEBUG/INFO. + if not verbose: + logging.getLogger("refactron").setLevel(logging.WARNING) + # Suppress noisy third-party libraries if not verbose: # Standard logging suppression diff --git a/refactron/core/analysis_result.py b/refactron/core/analysis_result.py index 9893708..9b51576 100644 --- a/refactron/core/analysis_result.py +++ b/refactron/core/analysis_result.py @@ -4,7 +4,7 @@ from pathlib import Path from typing import Dict, List, Optional -from refactron.core.models import CodeIssue, FileMetrics, IssueLevel +from refactron.core.models import AnalysisSkipWarning, CodeIssue, FileMetrics, IssueLevel @dataclass @@ -25,6 +25,8 @@ class AnalysisResult: total_files: int = 0 total_issues: int = 0 failed_files: List[FileAnalysisError] = field(default_factory=list) + semantic_skip_warnings: List[AnalysisSkipWarning] = field(default_factory=list) + semantic_skip_summary: Optional[str] = None @property def files_analyzed_successfully(self) -> int: diff --git a/refactron/core/backup.py b/refactron/core/backup.py index 40fc216..f4e20e7 100644 --- a/refactron/core/backup.py +++ b/refactron/core/backup.py @@ -7,6 +7,7 @@ - Rollback capability to restore original files """ +import hashlib import json import logging import re @@ -116,6 +117,8 @@ def backup_file(self, file_path: Path, session_id: str) -> Path: backup_path.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(file_path, backup_path) + file_sha256 = hashlib.sha256(backup_path.read_bytes()).hexdigest() + for session in self._index["sessions"]: if session["id"] == session_id: session["files"].append( @@ -124,6 +127,7 @@ def backup_file(self, file_path: Path, session_id: str) -> Path: "backup": str(backup_path), "relative_path": str(relative_path), "size": file_path.stat().st_size, + "sha256": file_sha256, } ) break @@ -180,12 +184,23 @@ def rollback_session(self, session_id: Optional[str] = None) -> Tuple[int, List[ if session is None: return 0, [] + # Validate backup integrity before restoring anything + _, corrupt_paths = self.validate_backup_integrity(session["id"]) + corrupt_set = set(corrupt_paths) + restored_count = 0 failed_files = [] for file_info in session["files"]: backup_path = Path(file_info["backup"]) original_path = Path(file_info["original"]) + if str(original_path) in corrupt_set: + logger.warning( + f"Skipping restore of {original_path}: backup failed integrity check" + ) + failed_files.append(str(original_path)) + continue + if backup_path.exists(): original_path.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(backup_path, original_path) @@ -220,6 +235,54 @@ def get_session(self, session_id: str) -> Optional[Dict[str, Any]]: return session # type: ignore[no-any-return] return None + def validate_backup_integrity(self, session_id: str) -> Tuple[List[str], List[str]]: + """Verify that every backup file in a session matches its recorded SHA-256. + + Args: + session_id: Session ID to validate. + + Returns: + Tuple of (valid_original_paths, corrupt_original_paths). + A path is "corrupt" if its backup file is missing or its hash doesn't match. + """ + session = self.get_session(session_id) + if session is None: + return [], [] + + valid: List[str] = [] + corrupt: List[str] = [] + + for file_info in session["files"]: + original_path = file_info["original"] + backup_path = Path(file_info["backup"]) + stored_hash = file_info.get("sha256") + + if not backup_path.exists(): + logger.warning(f"Backup file missing for integrity check: {backup_path}") + corrupt.append(original_path) + continue + + if stored_hash is None: + # Legacy backup without hash — treat as valid (best-effort) + valid.append(original_path) + continue + + try: + current_hash = hashlib.sha256(backup_path.read_bytes()).hexdigest() + if current_hash == stored_hash: + valid.append(original_path) + else: + logger.warning( + f"Backup integrity failure for {original_path}: " + f"expected {stored_hash[:8]}…, got {current_hash[:8]}…" + ) + corrupt.append(original_path) + except Exception as e: + logger.warning(f"Failed to verify backup for {original_path}: {e}") + corrupt.append(original_path) + + return valid, corrupt + def clear_session(self, session_id: str) -> bool: """ Clear a specific backup session. diff --git a/refactron/core/incremental.py b/refactron/core/incremental.py index 7978368..afa6a4b 100644 --- a/refactron/core/incremental.py +++ b/refactron/core/incremental.py @@ -1,5 +1,6 @@ """Incremental analysis tracking for performance optimization.""" +import hashlib import json import logging import threading @@ -38,8 +39,9 @@ def __init__( else: self.state_file = Path(state_file) - # State tracking: file_path -> (mtime, size, hash) - self._state: Dict[str, Dict[str, float]] = {} + # State tracking: file_path -> {mtime, size, sha256} + # sha256 is the authoritative change signal; mtime/size are fast pre-checks only. + self._state: Dict[str, Dict[str, object]] = {} # Thread lock for safe concurrent access self._lock = threading.Lock() @@ -100,18 +102,36 @@ def has_file_changed(self, file_path: Path) -> bool: logger.warning(f"Failed to get stats for {file_path}: {e}") return True # Assume changed if we can't get stats - # Check if file is new or changed + # Check if file is new if file_path_str not in self._state: logger.debug(f"New file detected: {file_path}") return True previous = self._state[file_path_str] - previous_mtime = previous.get("mtime", 0) previous_size = previous.get("size", 0) - # File changed if mtime or size is different - if current_mtime != previous_mtime or current_size != previous_size: - logger.debug(f"Changed file detected: {file_path}") + # Fast pre-check: size difference means content definitely changed + if current_size != previous_size: + logger.debug(f"Changed file detected (size mismatch): {file_path}") + return True + + # Authoritative check: compare SHA-256 hashes when available + stored_hash = previous.get("sha256") + if stored_hash: + try: + current_hash = hashlib.sha256(file_path.read_bytes()).hexdigest() + if current_hash != stored_hash: + logger.debug(f"Changed file detected (hash mismatch): {file_path}") + return True + return False # Same size AND same hash → unchanged + except Exception as e: + logger.warning(f"Failed to hash {file_path} for change detection: {e}") + # Fall through to mtime comparison on hash failure + + # Legacy fallback: no hash stored — use mtime (pre-SHA-256 state entries) + previous_mtime = previous.get("mtime", 0) + if current_mtime != previous_mtime: + logger.debug(f"Changed file detected (mtime mismatch, no hash): {file_path}") return True return False @@ -157,10 +177,12 @@ def update_file_state(self, file_path: Path) -> None: try: stat = file_path.stat() + content_hash = hashlib.sha256(file_path.read_bytes()).hexdigest() with self._lock: self._state[file_path_str] = { "mtime": stat.st_mtime, "size": stat.st_size, + "sha256": content_hash, } except Exception as e: logger.warning(f"Failed to update state for {file_path}: {e}") diff --git a/refactron/core/inference.py b/refactron/core/inference.py new file mode 100644 index 0000000..99c42f9 --- /dev/null +++ b/refactron/core/inference.py @@ -0,0 +1,90 @@ +""" +Inference engine wrapping astroid for semantic analysis. +Provides capabilities to infer types, values, and resolve symbols. +""" + +from typing import Any, List, Optional + +import astroid +from astroid import nodes +from astroid.context import InferenceContext +from astroid.exceptions import InferenceError + + +class InferenceEngine: + """ + Wrapper around astroid to provide high-level semantic analysis capabilities. + """ + + @staticmethod + def parse_string(code: str, module_name: str = "") -> nodes.Module: + """Parse source code string into an astroid node tree.""" + try: + return astroid.parse(code, module_name=module_name) + except Exception as e: + # Fallback or re-raise with better context if needed + raise ValueError(f"Failed to parse code with astroid: {e}") + + @staticmethod + def parse_file(file_path: str) -> nodes.Module: + """Parse a file into an astroid node tree.""" + builder = astroid.builder.AstroidBuilder(astroid.MANAGER) + with open(file_path, "r", encoding="utf-8") as f: + code = f.read() + return builder.string_build(code, modname=file_path) + + @staticmethod + def infer_node(node: nodes.NodeNG, context: Optional[InferenceContext] = None) -> List[Any]: + """ + Attempt to infer the value/type of a given node. + Returns a list of potential values (astroid nodes). + """ + try: + return list(node.infer(context=context)) + except InferenceError: + return [] + + @staticmethod + def get_node_type_name(node: nodes.NodeNG) -> str: + """Get the string representation of the inferred type.""" + inferred = InferenceEngine.infer_node(node) + if not inferred: + return "Uninferred" + + # Simple heuristic: take the first inference result + obj = inferred[0] + if isinstance(obj, nodes.Const): + return type(obj.value).__name__ + if isinstance(obj, nodes.ClassDef): + return str(obj.name) + if isinstance(obj, nodes.FunctionDef): + return "function" + if isinstance(obj, nodes.Module): + return "module" + if obj is astroid.Uninferable: + return "Uninferred" + + return str(getattr(obj, "name", str(type(obj)))) + + @staticmethod + def is_subtype_of(node: nodes.NodeNG, type_name: str) -> bool: + """Check if node infers to a specific type name (e.g. 'str', 'int', 'MyClass').""" + inferred_list = InferenceEngine.infer_node(node) + for obj in inferred_list: + if obj is astroid.Uninferable: + continue + + # Check direct type name + if getattr(obj, "name", "") == type_name: + return True + + # Check instance class + if isinstance(obj, nodes.Instance): + if obj.name == type_name: + return True + # Check ancestry + for ancestor in obj.ancestors(): + if ancestor.name == type_name: + return True + + return False diff --git a/refactron/core/models.py b/refactron/core/models.py index 78d0273..e76a984 100644 --- a/refactron/core/models.py +++ b/refactron/core/models.py @@ -85,6 +85,23 @@ def error_issues(self) -> List[CodeIssue]: return [i for i in self.issues if i.level == IssueLevel.ERROR] +@dataclass +class AnalysisSkipWarning: + """Emitted when a semantic analyzer is skipped for a file due to an unexpected error. + + This is NOT a hard failure — regular analyzers still run. + The warning is surfaced to the user so they know the file was not + fully checked by the semantic analysis layer. + """ + + file_path: Path + analyzer_name: str # e.g. "taint", "data_flow" + reason: str # Human-readable exception summary (never a full stack trace) + + def __str__(self) -> str: + return f"⚠ {self.analyzer_name} skipped {self.file_path} — {self.reason}" + + @dataclass class RefactoringOperation: """Represents a refactoring operation to be applied.""" diff --git a/refactron/core/refactron.py b/refactron/core/refactron.py index 48580b2..573004b 100644 --- a/refactron/core/refactron.py +++ b/refactron/core/refactron.py @@ -6,6 +6,8 @@ from pathlib import Path from typing import List, Optional, Tuple, Union +from refactron.analysis.cfg.builder import CFGBuilder +from refactron.analysis.taint import TaintAnalyzer from refactron.analyzers.base_analyzer import BaseAnalyzer from refactron.analyzers.code_smell_analyzer import CodeSmellAnalyzer from refactron.analyzers.complexity_analyzer import ComplexityAnalyzer @@ -22,7 +24,7 @@ from refactron.core.logging_config import setup_logging from refactron.core.memory_profiler import MemoryProfiler from refactron.core.metrics import get_metrics_collector -from refactron.core.models import FileMetrics, RefactoringOperation +from refactron.core.models import AnalysisSkipWarning, FileMetrics, RefactoringOperation from refactron.core.parallel import ParallelProcessor from refactron.core.prometheus_metrics import start_metrics_server from refactron.core.refactor_result import RefactorResult @@ -267,7 +269,9 @@ def process_file_wrapper( file_path: Path, ) -> Tuple[Optional[FileMetrics], Optional[FileAnalysisError]]: try: - file_metrics = self._analyze_file(file_path) + file_metrics, skip_warn = self._analyze_file(file_path) + if skip_warn is not None: + result.semantic_skip_warnings.append(skip_warn) # Update incremental tracker if self.incremental_tracker.enabled: @@ -306,9 +310,11 @@ def process_file_wrapper( # Sequential processing for file_path in files: try: - file_metrics = self._analyze_file(file_path) + file_metrics, skip_warn = self._analyze_file(file_path) result.file_metrics.append(file_metrics) result.total_issues += file_metrics.issue_count + if skip_warn is not None: + result.semantic_skip_warnings.append(skip_warn) # Update incremental tracker if self.incremental_tracker.enabled: @@ -335,6 +341,17 @@ def process_file_wrapper( ) ) + # Compute semantic skip summary if skip_rate > 10% + total_analyzed = len(result.file_metrics) + skipped_count = len(result.semantic_skip_warnings) + if total_analyzed > 0 and skipped_count / total_analyzed > 0.10: + result.semantic_skip_summary = ( + f"⚠ Semantic analysis (taint) was skipped for {skipped_count} of " + f"{total_analyzed} files ({skipped_count / total_analyzed * 100:.0f}%). " + "Check logs for details. Common causes: unsupported syntax or very large files." + ) + logger.warning(result.semantic_skip_summary) + # Save incremental state if self.incremental_tracker.enabled: self.incremental_tracker.save() @@ -363,7 +380,32 @@ def process_file_wrapper( return result - def _analyze_file(self, file_path: Path) -> FileMetrics: + def _run_semantic_analysis( + self, file_path: Path, source_code: str + ) -> Tuple[list, Optional["AnalysisSkipWarning"]]: + """Run TaintAnalyzer on *source_code* with full exception isolation. + + Returns: + (vulnerabilities, None) on success, or ([], AnalysisSkipWarning) on any failure. + The caller is guaranteed this method never raises. + """ + try: + cfg = CFGBuilder().build_from_source(source_code) + ta = TaintAnalyzer(cfg) + vulnerabilities = ta.analyze() + return vulnerabilities, None + except Exception as e: + short_reason = f"{type(e).__name__}: {e}" + logger.debug("Semantic analysis (taint) skipped for %s: %s", file_path, short_reason) + return [], AnalysisSkipWarning( + file_path=file_path, + analyzer_name="taint", + reason=short_reason, + ) + + def _analyze_file( + self, file_path: Path + ) -> Tuple["FileMetrics", Optional["AnalysisSkipWarning"]]: """Analyze a single file. Args: @@ -469,6 +511,9 @@ def _analyze_file(self, file_path: Path) -> FileMetrics: ) # Don't raise - allow other analyzers to run + # Run semantic analysis (TaintAnalyzer) with full exception isolation + _, skip_warning = self._run_semantic_analysis(file_path, source_code) + # Record file analysis metrics if self.metrics_collector and self.config.metrics_detailed: analysis_time_ms = (time.time() - start_time) * 1000 @@ -481,7 +526,7 @@ def _analyze_file(self, file_path: Path) -> FileMetrics: success=True, ) - return metrics + return metrics, skip_warning def refactor( self, diff --git a/tests/fixtures/fixture_bad_extract.py b/tests/fixtures/fixture_bad_extract.py new file mode 100644 index 0000000..2c2fc6d --- /dev/null +++ b/tests/fixtures/fixture_bad_extract.py @@ -0,0 +1,21 @@ +"""Module where a naive autofix would produce broken code. + +The ``eval()`` call is flagged by the security analyzer (SEC001), but +removing it naively would eliminate the only code path that computes +``result``, breaking the return value. The Verification Engine's +SyntaxVerifier must block any transform that introduces syntax errors +or new ``eval``/``exec`` calls. +""" + + +def dynamic_dispatch(expression, fallback=0): + result = eval(expression) # SEC001 — dangerous function + if result is None: + result = fallback + return result + + +def build_query(table, columns): + cols = ", ".join(columns) + query = f"SELECT {cols} FROM {table}" # noqa: S608 — intentional SQL pattern + return query diff --git a/tests/fixtures/fixture_clean.py b/tests/fixtures/fixture_clean.py new file mode 100644 index 0000000..b3b9944 --- /dev/null +++ b/tests/fixtures/fixture_clean.py @@ -0,0 +1,32 @@ +"""A clean module with zero code issues. + +This fixture exists as a negative control: analyzers and autofix +should find nothing to report or change. +""" + +from typing import List + +__all__ = ["fibonacci"] + + +def fibonacci(n: int) -> List[int]: + """Return the first *n* Fibonacci numbers. + + Args: + n: How many numbers to generate. + + Returns: + A list of Fibonacci numbers. + """ + if n <= 0: + return [] + + sequence: List[int] = [0] + if n == 1: + return sequence + + sequence.append(1) + for _ in range(2, n): + sequence.append(sequence[-1] + sequence[-2]) + + return sequence diff --git a/tests/fixtures/fixture_import_break.py b/tests/fixtures/fixture_import_break.py new file mode 100644 index 0000000..b3df7d0 --- /dev/null +++ b/tests/fixtures/fixture_import_break.py @@ -0,0 +1,22 @@ +"""Module where removing an import would break runtime behaviour. + +``collections`` is used three lines below its import via +``collections.OrderedDict``, but a naive unused-import fixer that only +checks for bare ``Name`` nodes might miss the dotted attribute access +and try to remove it. + +``sys`` is genuinely unused and safe to remove (DEP001 trigger). +""" + +import collections +import sys # intentionally unused — safe to remove (DEP001) # noqa: F401 + + +def ordered_merge(mapping_a, mapping_b): + merged = collections.OrderedDict() + for key, value in mapping_a.items(): + merged[key] = value + for key, value in mapping_b.items(): + if key not in merged: + merged[key] = value + return merged diff --git a/tests/fixtures/fixture_safe_extract.py b/tests/fixtures/fixture_safe_extract.py new file mode 100644 index 0000000..232acb4 --- /dev/null +++ b/tests/fixtures/fixture_safe_extract.py @@ -0,0 +1,80 @@ +"""Module with three safe-to-fix issues. + +Issues present: + - S004: magic number (42, 3.14, 86400) + - DEP001: unused import (``os``) + - C002: function too long (>20 lines) +""" + +import os # noqa: F401 — intentionally unused (DEP001 trigger) +from typing import List + + +def long_data_pipeline(records: List[dict]) -> dict: + """Process *records* and return a summary. + + This function is intentionally long to trigger C002. + """ + valid = [] + invalid_count = 0 + + for rec in records: + if "id" not in rec: + invalid_count += 1 + continue + if "value" not in rec: + invalid_count += 1 + continue + if rec["value"] < 0: + invalid_count += 1 + continue + valid.append(rec) + + total = 0.0 + for rec in valid: + total += rec["value"] + + average = total / len(valid) if valid else 0.0 + threshold = 42 # magic number → S004 + pi_approx = 3.14 # magic number → S004 + + above = [r for r in valid if r["value"] > threshold] + below = [r for r in valid if r["value"] <= threshold] + + seconds_per_day = 86400 # magic number → S004 + + # Compute percentile breakdown + sorted_values = sorted(r["value"] for r in valid) + n = len(sorted_values) + if n > 0: + p25_idx = max(0, n // 4 - 1) + p50_idx = max(0, n // 2 - 1) + p75_idx = max(0, 3 * n // 4 - 1) + p25 = sorted_values[p25_idx] + p50 = sorted_values[p50_idx] + p75 = sorted_values[p75_idx] + else: + p25 = 0 + p50 = 0 + p75 = 0 + + # Compute variance + variance = 0.0 + for rec in valid: + diff = rec["value"] - average + variance += diff * diff + variance = variance / n if n > 0 else 0.0 + + return { + "total": total, + "average": average, + "variance": variance, + "p25": p25, + "p50": p50, + "p75": p75, + "above_threshold": len(above), + "below_threshold": len(below), + "invalid": invalid_count, + "pi": pi_approx, + "seconds_per_day": seconds_per_day, + } diff --git a/tests/fixtures/fixture_test_break.py b/tests/fixtures/fixture_test_break.py new file mode 100644 index 0000000..2c19fbe --- /dev/null +++ b/tests/fixtures/fixture_test_break.py @@ -0,0 +1,21 @@ +"""Module whose public function is tested by fixture_test_break_test.py. + +If autofix changes the function signature (e.g. extracts ``tax_rate`` +into a module-level constant and removes the parameter), the +companion test file will break. The Verification Engine's +TestSuiteGate must block such transforms. + +Issues present: + - S004: magic number ``100`` inside the function body + - DEP001: unused import ``math`` +""" + +import math # intentionally unused — DEP001 trigger # noqa: F401 + + +def calculate_total(items, tax_rate=0.1): + subtotal = sum(items) + if subtotal > 100: # S004 — magic number + discount = subtotal * 0.05 + subtotal -= discount + return round(subtotal * (1 + tax_rate), 2) diff --git a/tests/fixtures/fixture_test_break_test.py b/tests/fixtures/fixture_test_break_test.py new file mode 100644 index 0000000..a314040 --- /dev/null +++ b/tests/fixtures/fixture_test_break_test.py @@ -0,0 +1,40 @@ +"""Tests for fixture_test_break.calculate_total(). + +These tests exercise the exact public signature of ``calculate_total`` +including the ``tax_rate`` keyword argument. If autofix changes the +function signature, these tests must fail — and the TestSuiteGate +should block the transform. +""" + +import sys +from pathlib import Path + +# Ensure the fixtures directory is importable +sys.path.insert(0, str(Path(__file__).parent)) + +from fixture_test_break import calculate_total # noqa: E402 + + +def test_basic_total(): + assert calculate_total([10, 20, 30]) == 66.0 + + +def test_custom_tax_rate(): + result = calculate_total([10, 20, 30], tax_rate=0.2) + assert result == 72.0 + + +def test_discount_applied_above_threshold(): + # subtotal = 200, discount = 10, taxed = 190 * 1.1 = 209.0 + result = calculate_total([100, 100]) + assert result == 209.0 + + +def test_no_discount_below_threshold(): + # subtotal = 50, no discount, taxed = 50 * 1.1 = 55.0 + result = calculate_total([20, 30]) + assert result == 55.0 + + +def test_empty_items(): + assert calculate_total([]) == 0.0 diff --git a/tests/test_cache_hardening.py b/tests/test_cache_hardening.py new file mode 100644 index 0000000..0d01a90 --- /dev/null +++ b/tests/test_cache_hardening.py @@ -0,0 +1,193 @@ +""" +Day 2 — SHA-256 hardening for IncrementalAnalysisTracker. + +Tests verify that: +- A file whose content changes is detected as changed even if mtime is rolled back +- A file whose mtime changes but content is identical is NOT re-analyzed +- Backup integrity validation catches a corrupted backup file +- Rollback skips corrupt backup files and reports them as failures +""" + +import hashlib +import os +from pathlib import Path + +from refactron.core.backup import BackupManager +from refactron.core.incremental import IncrementalAnalysisTracker + +# ─── helpers ──────────────────────────────────────────────────────────────── + + +def _sha256(path: Path) -> str: + return hashlib.sha256(path.read_bytes()).hexdigest() + + +def _write(path: Path, content: str) -> None: + path.write_text(content, encoding="utf-8") + + +# ─── IncrementalAnalysisTracker (SHA-256 hardening) ───────────────────────── + + +def test_cache_invalidates_on_content_change_not_mtime(tmp_path): + """Content change must trigger re-analysis even when mtime is rolled back to original.""" + state_file = tmp_path / "state.json" + tracker = IncrementalAnalysisTracker(state_file=state_file, enabled=True) + + py_file = tmp_path / "module.py" + _write(py_file, "x = 1\n") + + # Record initial state + tracker.update_file_state(py_file) + original_mtime = py_file.stat().st_mtime + + # Change content + _write(py_file, "x = 999 # changed\n") + + # Roll mtime back to the original value — simulates a git checkout or Docker volume + os.utime(py_file, (original_mtime, original_mtime)) + + # mtime is identical to recorded state, but content changed — must detect it + assert tracker.has_file_changed(py_file), ( + "File with changed content must be detected as changed, " + "even when mtime is rolled back to the original value" + ) + + +def test_cache_stable_when_content_unchanged_but_mtime_changes(tmp_path): + """A file whose mtime is bumped but content is identical must NOT trigger re-analysis.""" + state_file = tmp_path / "state.json" + tracker = IncrementalAnalysisTracker(state_file=state_file, enabled=True) + + py_file = tmp_path / "module.py" + content = "x = 1\n" + _write(py_file, content) + + # Record initial state + tracker.update_file_state(py_file) + + # Advance mtime by 1 second without changing content — simulates a `touch` + new_mtime = py_file.stat().st_mtime + 1.0 + os.utime(py_file, (new_mtime, new_mtime)) + + # Content is identical → should NOT be considered changed + assert not tracker.has_file_changed(py_file), ( + "File with unchanged content must not trigger re-analysis, " + "even when mtime advances (e.g. after git checkout or touch)" + ) + + +def test_state_persists_sha256_across_reload(tmp_path): + """SHA-256 stored by update_file_state must survive save/reload.""" + state_file = tmp_path / "state.json" + py_file = tmp_path / "module.py" + _write(py_file, "y = 42\n") + + tracker1 = IncrementalAnalysisTracker(state_file=state_file, enabled=True) + tracker1.update_file_state(py_file) + tracker1.save() + + # Advance mtime, same content + new_mtime = py_file.stat().st_mtime + 5.0 + os.utime(py_file, (new_mtime, new_mtime)) + + tracker2 = IncrementalAnalysisTracker(state_file=state_file, enabled=True) + assert not tracker2.has_file_changed( + py_file + ), "After reload, unchanged content must still be recognised as unchanged" + + +# ─── BackupManager integrity validation ───────────────────────────────────── + + +def test_backup_stores_sha256_in_index(tmp_path): + """backup_file() must record a 'sha256' field in the backup index.""" + mgr = BackupManager(root_dir=tmp_path) + session_id = mgr.create_backup_session("test") + + source = tmp_path / "app.py" + _write(source, "print('hello')\n") + + mgr.backup_file(source, session_id) + + session = mgr.get_session(session_id) + assert session is not None + file_record = session["files"][0] + assert "sha256" in file_record, "Backup index must include sha256 of backed-up file" + assert file_record["sha256"] == _sha256(source) + + +def test_backup_validation_passes_for_intact_backup(tmp_path): + """validate_backup_integrity() must return True for an unmodified backup.""" + mgr = BackupManager(root_dir=tmp_path) + session_id = mgr.create_backup_session("test") + + source = tmp_path / "app.py" + _write(source, "x = 1\n") + mgr.backup_file(source, session_id) + + valid, corrupt = mgr.validate_backup_integrity(session_id) + assert valid == [str(source)] + assert corrupt == [] + + +def test_backup_validation_catches_corrupted_file(tmp_path): + """validate_backup_integrity() must flag a backup file whose content was altered.""" + mgr = BackupManager(root_dir=tmp_path) + session_id = mgr.create_backup_session("test") + + source = tmp_path / "app.py" + _write(source, "x = 1\n") + backup_path = mgr.backup_file(source, session_id) + + # Corrupt the backup by overwriting it with different content + backup_path.write_text("CORRUPTED\n", encoding="utf-8") + + valid, corrupt = mgr.validate_backup_integrity(session_id) + assert str(source) in corrupt + assert str(source) not in valid + + +def test_rollback_refuses_invalid_backup(tmp_path): + """rollback_session() must skip corrupt backup files and include them in failed_files.""" + mgr = BackupManager(root_dir=tmp_path) + session_id = mgr.create_backup_session("test") + + source = tmp_path / "app.py" + original_content = "x = 1\n" + _write(source, original_content) + backup_path = mgr.backup_file(source, session_id) + + # Modify the original (simulate a refactoring was applied) + _write(source, "x = 999 # refactored\n") + + # Corrupt the backup + backup_path.write_text("CORRUPTED\n", encoding="utf-8") + + restored_count, failed_files = mgr.rollback_session(session_id) + + # The file should NOT be restored from a corrupt backup + assert restored_count == 0 + assert str(source) in failed_files + # Original file should not have been overwritten with corrupted content + assert source.read_text(encoding="utf-8") == "x = 999 # refactored\n" + + +def test_rollback_succeeds_for_intact_backup(tmp_path): + """rollback_session() must restore files correctly when backup is intact.""" + mgr = BackupManager(root_dir=tmp_path) + session_id = mgr.create_backup_session("test") + + source = tmp_path / "app.py" + original_content = "x = 1\n" + _write(source, original_content) + mgr.backup_file(source, session_id) + + # Simulate refactoring changing the file + _write(source, "x = 999 # refactored\n") + + restored_count, failed_files = mgr.rollback_session(session_id) + + assert restored_count == 1 + assert failed_files == [] + assert source.read_text(encoding="utf-8") == original_content diff --git a/tests/test_dry_run.py b/tests/test_dry_run.py new file mode 100644 index 0000000..af5e44a --- /dev/null +++ b/tests/test_dry_run.py @@ -0,0 +1,141 @@ +""" +Day 3 — --dry-run flag for refactron autofix. + +Tests verify that: +- generate_diff() produces a proper unified diff (or empty string when unchanged) +- fix_file() in dry_run=True mode never writes bytes to disk +- fix_file() in dry_run=True returns a diff showing proposed changes +- The fixed code returned in dry_run matches what apply would actually write +- fix_file() with dry_run=False does write the fixed content to disk +""" + +from pathlib import Path + +from refactron.autofix.engine import AutoFixEngine +from refactron.autofix.file_ops import generate_diff +from refactron.autofix.models import FixRiskLevel +from refactron.core.models import CodeIssue, IssueCategory, IssueLevel + +# ─── helpers ──────────────────────────────────────────────────────────────── + + +def _trailing_ws_issue(file_path: Path) -> CodeIssue: + return CodeIssue( + rule_id="remove_trailing_whitespace", + message="Trailing whitespace detected", + file_path=file_path, + line_number=1, + category=IssueCategory.STYLE, + level=IssueLevel.WARNING, + ) + + +# ─── generate_diff ──────────────────────────────────────────────────────────── + + +def test_generate_diff_produces_unified_diff(): + """generate_diff() must return a unified diff string when content changes.""" + original = "x = 1\n" + modified = "x = 2\n" + diff = generate_diff(original, modified, "test.py") + assert diff, "Diff must be non-empty when content changed" + assert "@@" in diff, "Diff must contain unified diff chunk markers" + + +def test_generate_diff_empty_when_no_changes(): + """generate_diff() must return empty string when content is identical.""" + content = "x = 1\n" + diff = generate_diff(content, content, "test.py") + assert diff == "", "Diff must be empty string when content is unchanged" + + +def test_generate_diff_includes_filename(): + """generate_diff() must include the filename in the diff header.""" + diff = generate_diff("a = 1\n", "a = 2\n", "mymodule.py") + assert "mymodule.py" in diff, "Diff header must include the filename" + + +# ─── fix_file (dry_run=True) ───────────────────────────────────────────────── + + +def test_dry_run_writes_no_files(tmp_path): + """fix_file() with dry_run=True must not modify any file on disk.""" + py_file = tmp_path / "module.py" + original_content = "x = 1 \ny = 2 \n" + py_file.write_text(original_content, encoding="utf-8") + + engine = AutoFixEngine(safety_level=FixRiskLevel.SAFE) + engine.fix_file(py_file, [_trailing_ws_issue(py_file)], dry_run=True) + + assert ( + py_file.read_text(encoding="utf-8") == original_content + ), "dry_run=True must not write any bytes to disk" + + +def test_dry_run_returns_unified_diff(tmp_path): + """fix_file() with dry_run=True must return a non-empty diff when changes exist.""" + py_file = tmp_path / "module.py" + py_file.write_text("x = 1 \ny = 2 \n", encoding="utf-8") + + engine = AutoFixEngine(safety_level=FixRiskLevel.SAFE) + _fixed_code, diff = engine.fix_file(py_file, [_trailing_ws_issue(py_file)], dry_run=True) + + assert ( + diff is not None and diff != "" + ), "fix_file must return a non-empty diff in dry_run mode when changes exist" + + +def test_dry_run_returns_none_diff_when_no_changes(tmp_path): + """fix_file() with dry_run=True must return None diff when nothing changes.""" + py_file = tmp_path / "module.py" + py_file.write_text("x = 1\ny = 2\n", encoding="utf-8") # No trailing whitespace + + engine = AutoFixEngine(safety_level=FixRiskLevel.SAFE) + _fixed_code, diff = engine.fix_file(py_file, [_trailing_ws_issue(py_file)], dry_run=True) + + assert ( + diff is None or diff == "" + ), "fix_file must return None/empty diff when no changes are made" + + +# ─── fix_file (apply vs dry-run consistency) ───────────────────────────────── + + +def test_dry_run_diff_matches_what_apply_would_write(tmp_path): + """The fixed code returned in dry_run must match what apply actually writes.""" + content = "x = 1 \ny = 2 \n" + + # Dry-run path + dry_file = tmp_path / "dry.py" + dry_file.write_text(content, encoding="utf-8") + engine = AutoFixEngine(safety_level=FixRiskLevel.SAFE) + fixed_from_dry, _diff = engine.fix_file(dry_file, [_trailing_ws_issue(dry_file)], dry_run=True) + + # Apply path (separate file, same content) + apply_file = tmp_path / "apply.py" + apply_file.write_text(content, encoding="utf-8") + engine.fix_file(apply_file, [_trailing_ws_issue(apply_file)], dry_run=False) + applied_content = apply_file.read_text(encoding="utf-8") + + assert ( + fixed_from_dry == applied_content + ), "dry_run fixed code must exactly match what apply writes to disk" + + +# ─── fix_file (dry_run=False) ──────────────────────────────────────────────── + + +def test_apply_without_dry_run_does_write_files(tmp_path): + """fix_file() with dry_run=False must write the fixed content to disk.""" + py_file = tmp_path / "module.py" + original_content = "x = 1 \ny = 2 \n" + py_file.write_text(original_content, encoding="utf-8") + + engine = AutoFixEngine(safety_level=FixRiskLevel.SAFE) + engine.fix_file(py_file, [_trailing_ws_issue(py_file)], dry_run=False) + + written_content = py_file.read_text(encoding="utf-8") + assert written_content != original_content, "dry_run=False must write the fixed content to disk" + # Confirm trailing whitespace was removed + for line in written_content.splitlines(): + assert line == line.rstrip(), f"Line still has trailing whitespace: {line!r}" diff --git a/tests/test_exception_isolation.py b/tests/test_exception_isolation.py new file mode 100644 index 0000000..6385f2b --- /dev/null +++ b/tests/test_exception_isolation.py @@ -0,0 +1,124 @@ +""" +Day 1 — Exception isolation for TaintAnalyzer / CFGBuilder. + +Tests verify that: +- A TaintAnalyzer crash on a specific file produces an AnalysisSkipWarning, not an exception +- A CFGBuilder crash (upstream of TaintAnalyzer) is also isolated +- When >10% of files skip semantic analysis, a summary warning is added to the result +- Regular analyzers still run and find issues even when semantic analysis is skipped +""" + +from pathlib import Path +from unittest.mock import patch + +from refactron import Refactron +from refactron.core.models import AnalysisSkipWarning # noqa: F401 (import drives the test fail) + +# ─── helpers ──────────────────────────────────────────────────────────────── + + +def _write_py(tmp_path: Path, name: str, content: str) -> Path: + f = tmp_path / name + f.write_text(content) + return f + + +# ─── tests ────────────────────────────────────────────────────────────────── + + +def test_taint_skip_produces_warning_not_crash(tmp_path): + """TaintAnalyzer.analyze() crashing on a file must produce a SkipWarning, not raise.""" + py_file = _write_py(tmp_path, "sample.py", "x = 1\n") + + with patch( + "refactron.core.refactron.TaintAnalyzer.analyze", + side_effect=RuntimeError("Unsupported AST node at line 47"), + ): + result = Refactron().analyze(py_file) + + assert not isinstance(result, Exception) + assert len(result.semantic_skip_warnings) == 1 + + warn = result.semantic_skip_warnings[0] + assert warn.analyzer_name == "taint" + assert "Unsupported AST node" in warn.reason + assert warn.file_path == py_file + + +def test_cfg_build_failure_skip_produces_warning_not_crash(tmp_path): + """CFGBuilder.build_from_source() crashing is isolated to a SkipWarning.""" + py_file = _write_py(tmp_path, "sample.py", "x = 1\n") + + with patch( + "refactron.core.refactron.CFGBuilder.build_from_source", + side_effect=RuntimeError("Unsupported node: MatchAs"), + ): + result = Refactron().analyze(py_file) + + assert not isinstance(result, Exception) + assert len(result.semantic_skip_warnings) == 1 + + warn = result.semantic_skip_warnings[0] + assert warn.analyzer_name == "taint" + assert "Unsupported node" in warn.reason + + +def test_skip_rate_over_10pct_shows_summary(tmp_path): + """When >10% of files have semantic analysis skipped, result.semantic_skip_summary is set.""" + for i in range(10): + _write_py(tmp_path, f"file_{i}.py", f"x = {i}\n") + + with patch( + "refactron.core.refactron.TaintAnalyzer.analyze", + side_effect=RuntimeError("crash"), + ): + result = Refactron().analyze(tmp_path) + + # 10/10 = 100% skip rate → summary must be present + assert result.semantic_skip_summary is not None + summary = result.semantic_skip_summary + assert "10" in summary or "100%" in summary + + +def test_skip_rate_under_threshold_has_no_summary(tmp_path): + """When only 1 of 20 files skips semantic analysis, no summary is shown (skip_rate <= 10%).""" + for i in range(20): + _write_py(tmp_path, f"file_{i}.py", f"x = {i}\n") + + call_count = {"n": 0} + + def maybe_crash(*args, **kwargs): + call_count["n"] += 1 + if call_count["n"] == 1: + raise RuntimeError("one file fails") + return [] # all others succeed + + with patch("refactron.core.refactron.TaintAnalyzer.analyze", side_effect=maybe_crash): + result = Refactron().analyze(tmp_path) + + # 1/20 = 5% → below 10% threshold → no summary + assert result.semantic_skip_summary is None + # But we should still have the single warning + assert len(result.semantic_skip_warnings) == 1 + + +def test_healthy_analyzers_unaffected_by_isolation(tmp_path): + """Regular code-smell / complexity analyzers still run even if semantic analysis fails.""" + # 42 is a magic number → CodeSmellAnalyzer should flag it + py_file = _write_py( + tmp_path, + "sample.py", + "def compute(x):\n return x * 42\n", + ) + + with patch( + "refactron.core.refactron.TaintAnalyzer.analyze", + side_effect=RuntimeError("crash"), + ): + result = Refactron().analyze(py_file) + + # Regular analyzers must still produce issues + assert result.total_issues > 0 + + # Semantic analysis produced a skip warning + assert len(result.semantic_skip_warnings) == 1 diff --git a/tests/test_fixtures_behave_as_expected.py b/tests/test_fixtures_behave_as_expected.py new file mode 100644 index 0000000..3d4c5dc --- /dev/null +++ b/tests/test_fixtures_behave_as_expected.py @@ -0,0 +1,136 @@ +""" +Day 4 — Verify that the test fixtures contain the expected issues. + +These fixtures are the ground truth for the Verification Engine (Days 6-15). +Each fixture is designed to trigger specific analyzer rule_ids so the +engine can prove it blocks bad transforms and allows safe ones. +""" + +from pathlib import Path + +import pytest + +from refactron.analyzers.code_smell_analyzer import CodeSmellAnalyzer +from refactron.analyzers.complexity_analyzer import ComplexityAnalyzer +from refactron.analyzers.dead_code_analyzer import DeadCodeAnalyzer +from refactron.analyzers.dependency_analyzer import DependencyAnalyzer +from refactron.core.config import RefactronConfig + +FIXTURES_DIR = Path(__file__).parent / "fixtures" + + +@pytest.fixture() +def config() -> RefactronConfig: + return RefactronConfig() + + +# ─── fixture_clean.py ──────────────────────────────────────────────────────── + + +def test_clean_fixture_has_zero_issues(config): + """fixture_clean.py must produce zero issues from code-smell & dead-code analyzers.""" + source = (FIXTURES_DIR / "fixture_clean.py").read_text(encoding="utf-8") + path = FIXTURES_DIR / "fixture_clean.py" + + smell_issues = CodeSmellAnalyzer(config).analyze(path, source) + dead_issues = DeadCodeAnalyzer(config).analyze(path, source) + + all_issues = smell_issues + dead_issues + assert ( + len(all_issues) == 0 + ), f"fixture_clean.py must have zero issues, got {len(all_issues)}: " + ", ".join( + str(i) for i in all_issues + ) + + +# ─── fixture_safe_extract.py ───────────────────────────────────────────────── + + +def test_safe_fixture_has_magic_number_issue(config): + """fixture_safe_extract.py must trigger at least one magic-number issue (S004).""" + source = (FIXTURES_DIR / "fixture_safe_extract.py").read_text(encoding="utf-8") + path = FIXTURES_DIR / "fixture_safe_extract.py" + + issues = CodeSmellAnalyzer(config).analyze(path, source) + magic_issues = [i for i in issues if i.rule_id == "S004"] + assert len(magic_issues) >= 1, "Expected at least one magic number issue (S004)" + + +def test_safe_fixture_has_unused_import(config): + """fixture_safe_extract.py must trigger at least one unused-import issue (DEP001).""" + source = (FIXTURES_DIR / "fixture_safe_extract.py").read_text(encoding="utf-8") + path = FIXTURES_DIR / "fixture_safe_extract.py" + + issues = DependencyAnalyzer(config).analyze(path, source) + unused = [i for i in issues if i.rule_id == "DEP001"] + assert len(unused) >= 1, "Expected at least one unused import issue (DEP001)" + + +def test_safe_fixture_has_long_function(config): + """fixture_safe_extract.py must trigger a function-too-long issue (C002).""" + source = (FIXTURES_DIR / "fixture_safe_extract.py").read_text(encoding="utf-8") + path = FIXTURES_DIR / "fixture_safe_extract.py" + + issues = ComplexityAnalyzer(config).analyze(path, source) + long_func = [i for i in issues if i.rule_id == "C002"] + assert len(long_func) >= 1, "Expected at least one long function issue (C002)" + + +# ─── fixture_bad_extract.py ────────────────────────────────────────────────── + + +def test_bad_extract_fixture_contains_known_issues(config): + """fixture_bad_extract.py must trigger at least one issue.""" + source = (FIXTURES_DIR / "fixture_bad_extract.py").read_text(encoding="utf-8") + path = FIXTURES_DIR / "fixture_bad_extract.py" + + smell_issues = CodeSmellAnalyzer(config).analyze(path, source) + dead_issues = DeadCodeAnalyzer(config).analyze(path, source) + + all_issues = smell_issues + dead_issues + assert len(all_issues) >= 1, "fixture_bad_extract.py must have at least one issue" + + +# ─── fixture_import_break.py ───────────────────────────────────────────────── + + +def test_import_break_fixture_has_unused_import(config): + """fixture_import_break.py must have an import the autofix would try to remove.""" + source = (FIXTURES_DIR / "fixture_import_break.py").read_text(encoding="utf-8") + path = FIXTURES_DIR / "fixture_import_break.py" + + issues = DependencyAnalyzer(config).analyze(path, source) + unused = [i for i in issues if i.rule_id == "DEP001"] + assert len(unused) >= 1, "Expected at least one unused-import to trigger autofix" + + +# ─── fixture_test_break.py ─────────────────────────────────────────────────── + + +def test_test_break_fixture_has_issues(config): + """fixture_test_break.py must have at least one issue.""" + source = (FIXTURES_DIR / "fixture_test_break.py").read_text(encoding="utf-8") + path = FIXTURES_DIR / "fixture_test_break.py" + + smell_issues = CodeSmellAnalyzer(config).analyze(path, source) + dep_issues = DependencyAnalyzer(config).analyze(path, source) + + all_issues = smell_issues + dep_issues + assert len(all_issues) >= 1, "fixture_test_break.py must have at least one issue" + + +def test_test_break_test_actually_passes(): + """fixture_test_break_test.py must pass when run against the unmodified fixture.""" + import subprocess + + test_file = FIXTURES_DIR / "fixture_test_break_test.py" + result = subprocess.run( + ["python3", "-m", "pytest", str(test_file), "-x", "--no-header", "-q"], + capture_output=True, + text=True, + timeout=30, + ) + assert result.returncode == 0, ( + f"fixture_test_break_test.py must pass against unmodified fixture.\n" + f"stdout: {result.stdout}\nstderr: {result.stderr}" + ) diff --git a/tests/test_semantic_analysis.py b/tests/test_semantic_analysis.py new file mode 100644 index 0000000..199c102 --- /dev/null +++ b/tests/test_semantic_analysis.py @@ -0,0 +1,137 @@ +""" +Integration tests for the new Semantic Analysis Engine. +Verifies CFG construction, Data Flow Analysis, and Taint Tracking. +""" + +import ast + +from refactron.analysis.cfg.builder import CFGBuilder, EdgeType +from refactron.analysis.data_flow import DataFlowAnalyzer +from refactron.analysis.taint import TaintAnalyzer, TaintConfig, TaintSink, TaintSource + + +def test_cfg_construction_simple_if(): + code = """ +if x: + y = 1 +else: + y = 2 +z = y +""" + builder = CFGBuilder() + entry = builder.build_from_source(code) + + # Entry block contains 'if x' test + assert len(entry.statements) == 1 + assert ( + isinstance(entry.statements[0], ast.If) is False + ) # The If node visitor handles this, stmt list gets 'x' (test) + + # Check successors + assert len(entry.successors) == 2 + true_branch = [n for n, t in entry.successors if t == EdgeType.TRUE][0] + false_branch = [n for n, t in entry.successors if t == EdgeType.FALSE][0] + + assert len(true_branch.statements) == 1 # y = 1 + assert len(false_branch.statements) == 1 # y = 2 + + # Both should converge + assert len(true_branch.successors) == 1 + assert len(false_branch.successors) == 1 + join_block = true_branch.successors[0][0] + assert join_block == false_branch.successors[0][0] + + assert len(join_block.statements) == 1 # z = y + + +def test_reaching_definitions(): + code = """ +x = 1 +y = 2 +if z: + x = 3 +w = x +""" + builder = CFGBuilder() + entry = builder.build_from_source(code) + + dfa = DataFlowAnalyzer(entry) + reaching_defs = dfa.compute_reaching_definitions() + + # Find the block with 'w = x' (should be the last join block) + # entry -> if -> (then) -> join + # entry -> if -> (else) -> join + + # Identify the join block by content 'w = x' + join_block = None + for node in dfa.nodes: + if any(isinstance(s, ast.Assign) and s.targets[0].id == "w" for s in node.statements): + join_block = node + break + + assert join_block is not None + + # In join block, 'x' definition should reach from both entry (x=1) and then_block (x=3) + defs_at_join = reaching_defs[join_block.id] + + # Filter for 'x' + x_defs = {d for v, d in defs_at_join if v == "x"} + + # We expect 2 definitions of x to reach here + # 1. from initial assignment (in entry block) + # 2. from re-assignment (in true branch) + # Wait, standard Reaching Defs is about what reaches the ENTRY of the block. + # Entry block defines x=1. + # True branch defines x=3. + # False branch (implicit) preserves x=1. + # So at Join, we should have {x from entry, x from true_branch} + + assert len(x_defs) >= 2 + + +def test_taint_analysis_sql_injection(): + code = """ +user_input = request.args.get('id') +query = "SELECT * FROM users WHERE id = " + user_input +cursor.execute(query) +""" + builder = CFGBuilder() + entry = builder.build_from_source(code) + + config = TaintConfig( + sources=[TaintSource("request.args.get", "function")], + sinks=[TaintSink("cursor.execute", "function", 0)], + sanitizers=[], + ) + + analyzer = TaintAnalyzer(entry, config) + vulnerabilities = analyzer.analyze() + + assert len(vulnerabilities) == 1 + vuln = vulnerabilities[0] + assert vuln.variable == "query" + assert "cursor.execute" in vuln.sink + assert "request.args.get" in vuln.message or "source" in vuln.message + + +def test_taint_analysis_safe_path(): + code = """ +user_input = request.args.get('id') +clean_input = int(user_input) +query = "SELECT * FROM users WHERE id = " + str(clean_input) +cursor.execute(query) +""" + builder = CFGBuilder() + entry = builder.build_from_source(code) + + config = TaintConfig( + sources=[TaintSource("request.args.get", "function")], + sinks=[TaintSink("cursor.execute", "function", 0)], + sanitizers=["int"], + ) + + analyzer = TaintAnalyzer(entry, config) + vulnerabilities = analyzer.analyze() + + # Should find NO vulnerabilities because 'int()' sanitizes + assert len(vulnerabilities) == 0 diff --git a/tests/test_tui_viewer.py b/tests/test_tui_viewer.py new file mode 100644 index 0000000..822eda0 --- /dev/null +++ b/tests/test_tui_viewer.py @@ -0,0 +1,225 @@ +"""Tests for the TUI issue viewer state machine. + +Tests the pure-logic components: TuiState, _handle_key(), and render functions. +No TTY/termios required — all state transitions are pure functions. +""" + +from pathlib import Path + +from refactron.cli.ui import KEY_DOWN, KEY_ENTER, KEY_UP, TuiState, _build_tui_groups, _handle_key +from refactron.core.analysis_result import AnalysisResult +from refactron.core.models import CodeIssue, FileMetrics, IssueCategory, IssueLevel + +# ── Helpers ────────────────────────────────────────────────────────── + + +def _make_issue( + level: IssueLevel = IssueLevel.WARNING, + message: str = "test issue", + line: int = 10, + suggestion: str = None, +) -> CodeIssue: + return CodeIssue( + category=IssueCategory.STYLE, + level=level, + message=message, + file_path=Path("/tmp/test.py"), + line_number=line, + suggestion=suggestion, + ) + + +def _make_result(*issues: CodeIssue) -> AnalysisResult: + fm = FileMetrics( + file_path=Path("/tmp/test.py"), + lines_of_code=100, + comment_lines=5, + blank_lines=10, + complexity=1.0, + maintainability_index=80.0, + functions=3, + classes=0, + issues=list(issues), + ) + return AnalysisResult( + file_metrics=[fm], + total_files=1, + total_issues=len(issues), + ) + + +def _sample_result() -> AnalysisResult: + """Result with issues in 3 severity groups: critical(1), warning(2), info(1).""" + return _make_result( + _make_issue(IssueLevel.CRITICAL, "critical bug", 5, "fix it now"), + _make_issue(IssueLevel.WARNING, "unused var", 10), + _make_issue(IssueLevel.WARNING, "magic number", 20, "use a constant"), + _make_issue(IssueLevel.INFO, "consider docstring", 30), + ) + + +# ── TuiState creation ─────────────────────────────────────────────── + + +class TestTuiState: + def test_initial_state_is_summary_screen(self): + groups = _build_tui_groups(_sample_result()) + state = TuiState(groups=groups) + assert state.screen == "summary" + assert state.cursor == 0 + + def test_groups_only_contain_nonempty_severities(self): + result = _make_result( + _make_issue(IssueLevel.CRITICAL, "one"), + _make_issue(IssueLevel.INFO, "two"), + ) + groups = _build_tui_groups(result) + # Only critical and info should appear (no warning, no error) + level_names = [g[0] for g in groups] + assert "critical" in level_names + assert "info" in level_names + assert "warning" not in level_names + assert "error" not in level_names + + def test_groups_are_ordered_by_severity(self): + groups = _build_tui_groups(_sample_result()) + level_names = [g[0] for g in groups] + assert level_names == ["critical", "warning", "info"] + + def test_expanded_set_starts_empty(self): + groups = _build_tui_groups(_sample_result()) + state = TuiState(groups=groups) + assert state.expanded == set() + + +# ── Summary screen navigation ─────────────────────────────────────── + + +class TestSummaryNavigation: + def test_arrow_down_moves_cursor(self): + groups = _build_tui_groups(_sample_result()) + state = TuiState(groups=groups) + new = _handle_key(state, KEY_DOWN) + assert new.cursor == 1 + + def test_arrow_up_moves_cursor(self): + groups = _build_tui_groups(_sample_result()) + state = TuiState(groups=groups, cursor=2) + new = _handle_key(state, KEY_UP) + assert new.cursor == 1 + + def test_arrow_up_at_top_stays_at_zero(self): + groups = _build_tui_groups(_sample_result()) + state = TuiState(groups=groups, cursor=0) + new = _handle_key(state, KEY_UP) + assert new.cursor == 0 + + def test_arrow_down_at_bottom_stays(self): + groups = _build_tui_groups(_sample_result()) + state = TuiState(groups=groups, cursor=len(groups) - 1) + new = _handle_key(state, KEY_DOWN) + assert new.cursor == len(groups) - 1 + + def test_enter_drills_into_group(self): + groups = _build_tui_groups(_sample_result()) + state = TuiState(groups=groups, cursor=0) # cursor on "critical" + new = _handle_key(state, KEY_ENTER) + assert new.screen == "group" + assert new.current_group == 0 + assert new.cursor == 0 # reset to first issue in group + + def test_q_signals_quit(self): + groups = _build_tui_groups(_sample_result()) + state = TuiState(groups=groups) + new = _handle_key(state, "q") + assert new.quit is True + + +# ── Group screen navigation ───────────────────────────────────────── + + +class TestGroupNavigation: + def _group_state(self) -> TuiState: + groups = _build_tui_groups(_sample_result()) + return TuiState(groups=groups, screen="group", current_group=1, cursor=0) + # current_group=1 is "warning" which has 2 issues + + def test_arrow_down_moves_between_issues(self): + state = self._group_state() + new = _handle_key(state, KEY_DOWN) + assert new.cursor == 1 + + def test_arrow_down_at_last_issue_stays(self): + state = self._group_state() + state = TuiState(**{**state.__dict__, "cursor": 1}) + new = _handle_key(state, KEY_DOWN) + assert new.cursor == 1 + + def test_enter_toggles_expand(self): + state = self._group_state() + # First Enter expands issue 0 + new = _handle_key(state, KEY_ENTER) + assert (1, 0) in new.expanded # (group_idx, issue_idx) + + # Second Enter collapses it + new2 = _handle_key(new, KEY_ENTER) + assert (1, 0) not in new2.expanded + + def test_b_goes_back_to_summary(self): + state = self._group_state() + new = _handle_key(state, "b") + assert new.screen == "summary" + # Cursor should return to the group we were viewing + assert new.cursor == 1 + + def test_n_goes_to_next_group(self): + state = self._group_state() # group 1 (warning) + new = _handle_key(state, "n") + assert new.screen == "group" + assert new.current_group == 2 # info + assert new.cursor == 0 + + def test_n_at_last_group_stays(self): + groups = _build_tui_groups(_sample_result()) + state = TuiState( + groups=groups, + screen="group", + current_group=len(groups) - 1, + cursor=0, + ) + new = _handle_key(state, "n") + assert new.current_group == len(groups) - 1 + + def test_p_goes_to_prev_group(self): + state = self._group_state() # group 1 (warning) + new = _handle_key(state, "p") + assert new.screen == "group" + assert new.current_group == 0 # critical + assert new.cursor == 0 + + def test_p_at_first_group_stays(self): + groups = _build_tui_groups(_sample_result()) + state = TuiState(groups=groups, screen="group", current_group=0, cursor=0) + new = _handle_key(state, "p") + assert new.current_group == 0 + + def test_q_signals_quit_from_group(self): + state = self._group_state() + new = _handle_key(state, "q") + assert new.quit is True + + +# ── Edge cases ─────────────────────────────────────────────────────── + + +class TestEdgeCases: + def test_unknown_key_is_ignored(self): + groups = _build_tui_groups(_sample_result()) + state = TuiState(groups=groups, cursor=1) + new = _handle_key(state, "x") + assert new == state # unchanged + + def test_empty_result_produces_no_groups(self): + result = _make_result() # no issues + groups = _build_tui_groups(result) + assert groups == []