From f8a9bcc4e3f9304fd72ae6bfdd4f1a60d0711432 Mon Sep 17 00:00:00 2001 From: Minxi Hou Date: Thu, 29 Oct 2026 06:36:25 -0400 Subject: [PATCH 1/8] fixval: count a reverted pytest exit as RED only with JUnit evidence (#28) Exits 1 and 2 on the reverted code used to count as "the test fails without the fix" no matter why pytest stopped. A broken fixture, a syntax error or an unrelated import in the test file therefore passed FIXVAL. The reverted pytest run now writes a JUnit report to a private temp file. The run is RED when the report has a , or a collection or setup that is an import of a name the fix adds: a module-level name bound on a "+" line and not on a "-" line of the same file, or a module the patch creates ("cannot import name", "has no attribute", "No module named", "bad magic number"). Teardown errors never count. Anything else, including a missing or unparseable report, is FIXVAL_INCONCLUSIVE. Non-pytest runners and the require_witness path are unchanged. Refs #28 --- src/code_forge/_fixval_junit.py | 148 ++++++++++++++ src/code_forge/fixval.py | 74 ++++++- tests/test_fixval.py | 9 + tests/test_fixval_exit_classification.py | 20 +- tests/test_fixval_junit_witness.py | 235 +++++++++++++++++++++++ tests/test_fixval_test_support.py | 4 +- 6 files changed, 472 insertions(+), 18 deletions(-) create mode 100644 src/code_forge/_fixval_junit.py create mode 100644 tests/test_fixval_junit_witness.py diff --git a/src/code_forge/_fixval_junit.py b/src/code_forge/_fixval_junit.py new file mode 100644 index 00000000..3ee2e13c --- /dev/null +++ b/src/code_forge/_fixval_junit.py @@ -0,0 +1,148 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright (c) 2026, Minxi Hou +"""RED evidence from a reverted-code pytest run's JUnit XML (#28). + +A reverted run counts as failing only when the report shows a failed test, +or a collection or setup error that is an import of a name the fix adds: +the fix defines `helper`, the test does `from mod import helper`, and the +reverted code no longer has it. Any other error (a broken fixture, a syntax +error, an unrelated import) did not judge the test. +""" + +from __future__ import annotations + +import keyword +import re +import xml.etree.ElementTree as ElementTree +from dataclasses import dataclass + +import unidiff + +from .diff import iter_diff_sections, patched_file_path + +# Top-level definitions on an added or removed line of a production module. +_DEFINITION = re.compile(r"(?:async\s+def|def|class)\s+([A-Za-z_]\w*)") +_ASSIGNMENT = re.compile(r"([A-Za-z_]\w*)\s*(?::[^=]*)?=(?!=)") +_FROM_IMPORT = re.compile(r"from\s+[\w.]+\s+import\s+\(?([^)#]*)") +_IMPORT = re.compile(r"import\s+([^#]*)") +_ALIAS = re.compile(r"[\w.]+(?:\s+as\s+(\w+))?") + +_CANNOT_IMPORT = re.compile(r"cannot import name '(\w+)' from '([\w.]+)'") +_NO_ATTRIBUTE = re.compile(r"module '([\w.]+)' has no attribute '(\w+)'") +_NO_MODULE = re.compile(r"No module named '([\w.]+)'|bad magic number in '([\w.]+)'") + + +@dataclass(frozen=True) +class AddedNames: + """Module-level names and modules that exist only with the fix applied.""" + + names: frozenset[tuple[str, str]] # (module, name) + modules: frozenset[str] + + +def _module_names(path: str) -> set[str]: + """Dotted names `path` can be imported as, with and without `src.`.""" + if not path.endswith(".py"): + return set() + parts = path[: -len(".py")].split("/") + if parts[-1] == "__init__": + parts = parts[:-1] + if not parts or not all(part.isidentifier() for part in parts): + return set() + names = {".".join(parts)} + if parts[0] == "src" and len(parts) > 1: + names.add(".".join(parts[1:])) + return names + + +def _imported_names(line: str) -> set[str]: + match = _FROM_IMPORT.match(line) + if match: + clause = match.group(1) + else: + match = _IMPORT.match(line) + if not match: + return set() + clause = match.group(1) + names = set() + for item in clause.split(","): + alias = _ALIAS.fullmatch(item.strip()) + if alias: + names.add(alias.group(1) or item.strip().split(".")[0]) + return names + + +def _defined(line: str) -> set[str]: + """Module-level names a line binds; indented lines bind none.""" + for pattern in (_DEFINITION, _ASSIGNMENT): + match = pattern.match(line) + if match: + return set() if keyword.iskeyword(match.group(1)) else {match.group(1)} + return _imported_names(line) + + +def added_names(patch: str) -> AddedNames: + """Names the fix in `patch` adds: bound on an added line and not on a + removed line of the same file. A file the patch creates adds its module. + An unreadable patch adds nothing, so no import error counts as RED.""" + names: set[tuple[str, str]] = set() + modules: set[str] = set() + try: + blocks = [block for _path, block in iter_diff_sections(patch)] + entries = [entry for block in blocks for entry in unidiff.PatchSet(block)] + except (unidiff.errors.UnidiffParseError, ValueError): + return AddedNames(frozenset(), frozenset()) + for entry in entries: + path = patched_file_path(entry) + dotted = _module_names(path or "") + if not dotted: + continue + if patched_file_path(entry, source=True) is None: + modules.update(dotted) + added: set[str] = set() + removed: set[str] = set() + for hunk in entry: + for line in hunk: + if line.is_added: + added |= _defined(line.value) + elif line.is_removed: + removed |= _defined(line.value) + names.update((module, name) for module in dotted for name in added - removed) + return AddedNames(frozenset(names), frozenset(modules)) + + +def _is_added_import_failure(text: str, added: AddedNames) -> bool: + for name, module in _CANNOT_IMPORT.findall(text): + if (module, name) in added.names: + return True + for module, name in _NO_ATTRIBUTE.findall(text): + if (module, name) in added.names: + return True + # A module the fix adds is missing, or only a stale foreign .pyc is left. + return any(set(modules) & added.modules for modules in _NO_MODULE.findall(text)) + + +def red_witness(junit_xml: str, added: AddedNames) -> bool: + """Whether the report shows the reverted code failing a test. + + A `` is a failed test call. An `` counts only when it is + a collection or setup error whose text is an import of a name in + `added`; teardown errors never count. A report that does not parse + shows nothing. + """ + try: + # pytest wrote it; ElementTree resolves no external entities and the + # bundled expat caps entity expansion, so a hostile conftest gains nothing. + root = ElementTree.fromstring(junit_xml) # noqa: S314 + except ElementTree.ParseError: + return False + for case in root.iter("testcase"): + if case.find("failure") is not None: + return True + for error in case.findall("error"): + message = error.get("message", "") + if message.startswith("failed on teardown"): + continue + if _is_added_import_failure(message + "\n" + (error.text or ""), added): + return True + return False diff --git a/src/code_forge/fixval.py b/src/code_forge/fixval.py index aadc88ee..5a8ab79c 100644 --- a/src/code_forge/fixval.py +++ b/src/code_forge/fixval.py @@ -34,6 +34,7 @@ from . import progress from .advisory import AdvisoryFinding +from ._fixval_junit import AddedNames, added_names, red_witness from ._fixval_transaction import FixvalTransaction, TransactionError from ._mutation_process import child_sigint_options from .disposition import Disposition @@ -327,9 +328,9 @@ def _transaction_block(message: str, recovery: str) -> FixvalResult: # #252: only a judged test failure counts as "fails on the reverted code". # pytest's exit codes have fixed meanings; other runners' codes don't (make # uses 2 for a failed recipe), so they keep "nonzero = fails" except signals. -# Exit 1 (tests failed) and exit 2 (interrupted or collection error) count as -# failing, exactly as before #252: a test importing a name the fix adds exits -# 2 on the reverted code. Whether collection errors are RED witnesses is #28. +# Exits 1 (tests failed) and 2 (interrupted or collection error) can be +# failing; the JUnit report then decides whether they are (#28, see +# red_witness in _fixval_junit). _PYTEST_JUDGED = frozenset({1, 2}) _PYTEST_EXIT = { 3: "internal error", @@ -502,8 +503,9 @@ def _runs_pytest(cmd: list[str]) -> bool: def _judged_failure(cmd: list[str], result: subprocess.CompletedProcess) -> str | None: """Return None when a nonzero run judged the tests as failing, else why not. - A signal death never judged anything. For pytest, exits 1 and 2 are - judgements (2 pending #28); 3, 4, 5 and unknown codes are not. + A signal death never judged anything. For pytest, exits 1 and 2 may be + judgements (the JUnit report decides, #28); 3, 4, 5 and unknown codes + are not. """ code = result.returncode if code < 0: @@ -586,17 +588,66 @@ def _witnessed_failure(cmd: list[str], result: subprocess.CompletedProcess) -> b return _PYTEST_FAILED_ENTRY.search(output) is not None +def _read_report(path: str) -> str: + try: + with open(path, encoding="utf-8", errors="replace") as fh: + return fh.read() + except OSError: + return "" + + def _test_reverted_candidate( - candidate, scoped_cmd, run_env, repo_root, reverted_files=None, *, require_witness=False, timeout=600 + candidate, + scoped_cmd, + run_env, + repo_root, + reverted_files=None, + *, + require_witness=False, + timeout=600, + added: AddedNames | None = None, ) -> FixvalResult | None: """Run the tests on reverted code. With `require_witness`, a nonzero run - that did not show a failing test returns None (no verdict).""" + that did not show a failing test returns None (no verdict). + + Otherwise a pytest run writes a JUnit report, and exit 1 or 2 is RED + only when that report shows a failed test or an import of a name in + `added` (#28); without it the run is inconclusive.""" # #183: each bounded FIXVAL test run starts with a progress line, so the # MCP job's inactivity watchdog never sees two runs as one silent span. progress.emit("fixval: reverted-fix run") + report = None + if not require_witness and _runs_pytest(scoped_cmd): + fd, report = tempfile.mkstemp(prefix="fixval-junit-", suffix=".xml") + os.close(fd) + os.unlink(report) # pytest writes it; a stale empty file would read as a report try: - result = subprocess.run( + return _judge_reverted_run( + candidate, scoped_cmd, + run_env, + repo_root, + reverted_files, + require_witness, + timeout, + added, + report, + ) + finally: + if report is not None: + try: + os.unlink(report) + except FileNotFoundError: + pass + + +def _judge_reverted_run( + candidate, scoped_cmd, run_env, repo_root, reverted_files, require_witness, timeout, added, report +) -> FixvalResult | None: + cmd = scoped_cmd if report is None else [*scoped_cmd, "--junitxml=%s" % report] + try: + result = subprocess.run( + cmd, env=run_env, capture_output=True, text=True, @@ -619,6 +670,12 @@ def _test_reverted_candidate( return None return FixvalResult(status=FixvalStatus.PASS, findings=[], advisories=[]) reason = _judged_failure(scoped_cmd, result) + if reason is None and report is not None: + if not red_witness(_read_report(report), added or AddedNames(frozenset(), frozenset())): + reason = ( + "pytest exit %d without a failed test or an import of a name the fix adds" + % result.returncode + ) if reason is not None: return _inconclusive_block(reason, result.stderr or result.stdout or "") return FixvalResult(status=FixvalStatus.PASS, findings=[], advisories=[]) @@ -1032,6 +1089,7 @@ def _reverted_run( changed_files_in_order(non_test_patch), require_witness=require_witness, timeout=timeout, + added=added_names(non_test_patch), ) except (OSError, ValueError, TransactionError, subprocess.SubprocessError) as exc: errors.append(str(exc)) diff --git a/tests/test_fixval.py b/tests/test_fixval.py index 2addb300..14861b62 100644 --- a/tests/test_fixval.py +++ b/tests/test_fixval.py @@ -157,6 +157,12 @@ def _make_candidate(): ) +FAILED_TEST_REPORT = ( + '' + '' +) + + def _private_reverse_results(results): """Model Git's private output as well as its exit status in orchestration tests.""" from pathlib import Path @@ -171,6 +177,9 @@ def run(argv, **kwargs): source = image / "src/foo.py" source.parent.mkdir(parents=True, exist_ok=True) source.write_bytes(b"old\n") + report = [arg.split("=", 1)[1] for arg in argv if str(arg).startswith("--junitxml=")] + if report and result.returncode == 1: # pytest's report of a failed test (#28) + Path(report[0]).write_text(FAILED_TEST_REPORT, encoding="utf-8") return result return run diff --git a/tests/test_fixval_exit_classification.py b/tests/test_fixval_exit_classification.py index 13b616f8..9ebb5613 100644 --- a/tests/test_fixval_exit_classification.py +++ b/tests/test_fixval_exit_classification.py @@ -2,9 +2,9 @@ """#252: an unjudged reverted-code run is not FIXVAL PASS. A signal death is inconclusive for any runner. For pytest, exit 1 (tests -failed) and exit 2 (interrupted or collection error) count as failing exactly as -on main, pending #28; 3-5 and other codes are inconclusive. Other runners keep -"nonzero = fails".""" +failed) and exit 2 (interrupted or collection error) count as failing only when +the JUnit report shows it (#28, tests/test_fixval_junit_witness.py); 3-5 and +other codes are inconclusive. Other runners keep "nonzero = fails".""" import os import stat @@ -73,8 +73,12 @@ def test_pytest_non_failure_exit_is_inconclusive(tmp_path, code, label): @pytest.mark.parametrize("code", [1, 2]) -def test_pytest_exit_1_and_2_are_pass_as_on_main(tmp_path, code): - assert _reverted(tmp_path, _fake_pytest(tmp_path, code)).status == FixvalStatus.PASS +def test_pytest_exit_1_and_2_without_a_report_are_inconclusive(tmp_path, code): + # #28: the fake pytest writes no JUnit report, so nothing shows a failed test + _assert_inconclusive( + _reverted(tmp_path, _fake_pytest(tmp_path, code)), + "pytest exit %d without a failed test or an import of a name the fix adds" % code, + ) @pytest.mark.parametrize("code", [1, 2, 3, 5, 7, 127]) @@ -150,10 +154,10 @@ def test_real_pytest_failing_test_is_pass(tmp_path): ) -def test_real_pytest_import_of_name_the_fix_adds_is_pass_pending_28(tmp_path): - # the reverted code lacks `helper`: collection error, exit 2, counted as failing as on main +def test_real_pytest_import_error_without_added_names_is_inconclusive(tmp_path): + # #28: with no diff saying the fix adds `helper`, this collection error judged nothing cmd = _real_pytest(tmp_path, "from added_by_fix import helper\ndef test_it():\n helper()\n") - assert _reverted(tmp_path, cmd).status == FixvalStatus.PASS + _assert_inconclusive(_reverted(tmp_path, cmd), "pytest exit 2 without a failed test") def test_real_pytest_no_tests_is_inconclusive(tmp_path): diff --git a/tests/test_fixval_junit_witness.py b/tests/test_fixval_junit_witness.py new file mode 100644 index 00000000..9be6560a --- /dev/null +++ b/tests/test_fixval_junit_witness.py @@ -0,0 +1,235 @@ +# SPDX-License-Identifier: Apache-2.0 +"""#28: a reverted pytest run is RED only when its JUnit report shows a failed +test, or a collection or setup error importing a name the fix adds.""" + +import os +import subprocess +import sys +import tempfile + +import pytest + +from code_forge._fixval_junit import AddedNames, added_names, red_witness +from code_forge.fixval import ( + FixvalCandidate, + FixvalStatus, + _test_reverted_candidate, + classify_fixval_candidate, + run_fixval, +) + +CANDIDATE = FixvalCandidate(["test_probe.py"], ["src/calc.py"]) +CALC_HELPER = AddedNames(frozenset({("calc", "helper")}), frozenset()) +NEW_MODULE = AddedNames(frozenset(), frozenset({"pkg.newmod"})) +NOTHING = AddedNames(frozenset(), frozenset()) + + +def _patch(path, removed, added, new_file=False): + source = "/dev/null" if new_file else "a/" + path + header = "diff --git a/%s b/%s\n" % (path, path) + if new_file: + header += "new file mode 100644\n" + body = "".join("-%s\n" % line for line in removed) + "".join("+%s\n" % line for line in added) + hunk = "@@ -%s +1,%d @@\n" % ("0,0" if new_file else "1,%d" % len(removed), len(added)) + return header + "--- %s\n+++ b/%s\n" % (source, path) + hunk + body + + +@pytest.mark.parametrize( + ("line", "names"), + [ + ("def helper(x):", {"helper"}), + ("async def fetch():", {"fetch"}), + ("class Store:", {"Store"}), + ("LIMIT = 3", {"LIMIT"}), + ("LIMIT: int = 3", {"LIMIT"}), + ("from .impl import run, stop as halt", {"run", "halt"}), + ("import json", {"json"}), + (" def method(self):", set()), + (" value = 1", set()), + ("value == 1", set()), + ("else: x = 1", set()), + ("# def commented():", set()), + ], +) +def test_added_names_reads_module_level_bindings(line, names): + found = added_names(_patch("src/calc.py", ["pass"], [line])) + assert {name for _module, name in found.names} == names + assert found.modules == frozenset() + + +def test_added_names_maps_paths_to_modules_and_skips_names_that_already_existed(): + patch = _patch("src/calc.py", ["def helper():", "OLD = 1"], ["def helper(x):", "NEW = 2"]) + patch += _patch("lib/pkg/__init__.py", [], ["def entry():"], new_file=True) + patch += _patch("docs/notes.md", [], ["def nothing():"]) + found = added_names(patch) + assert found.names == { + ("src.calc", "NEW"), + ("calc", "NEW"), + ("lib.pkg", "entry"), + } + assert found.modules == {"lib.pkg"} + + +def test_unreadable_patch_adds_nothing(): + assert added_names("@@ not a diff\n+def helper():\n") == NOTHING + + +def _report(*cases): + return "%s" % "".join( + '%s' % item for item in enumerate(cases) + ) + + +def _error(message, text=""): + return '%s' % (message, text) + + +IMPORT_HELPER = "E ImportError: cannot import name 'helper' from 'calc' (/r/src/calc.py)" + + +@pytest.mark.parametrize( + ("report", "added", "red"), + [ + (_report(''), NOTHING, True), + (_report(_error("collection failure", IMPORT_HELPER)), CALC_HELPER, True), + (_report(_error("failed on setup with "%s"" % IMPORT_HELPER)), CALC_HELPER, True), + ( + _report( + _error("collection failure", "E AttributeError: module 'calc' has no attribute 'helper'") + ), + CALC_HELPER, + True, + ), + ( + _report(_error("collection failure", "E ModuleNotFoundError: No module named 'pkg.newmod'")), + NEW_MODULE, + True, + ), + (_report("", _error("collection failure", IMPORT_HELPER)), CALC_HELPER, True), + # counterexamples: errors that did not judge the test + (_report(_error("failed on teardown with "%s"" % IMPORT_HELPER)), CALC_HELPER, False), + ( + _report(_error("collection failure", IMPORT_HELPER.replace("'calc'", "'other'"))), + CALC_HELPER, + False, + ), + ( + _report(_error("collection failure", IMPORT_HELPER.replace("helper", "helpers"))), + CALC_HELPER, + False, + ), + ( + _report( + _error("collection failure", "E AttributeError: module 'calc' has no attribute 'other'") + ), + CALC_HELPER, + False, + ), + (_report(_error("collection failure", "E SyntaxError: invalid syntax")), CALC_HELPER, False), + (_report(_error("failed on setup with "RuntimeError: db down"")), CALC_HELPER, False), + ( + _report(_error("collection failure", "E ModuleNotFoundError: No module named 'pkg.other'")), + NEW_MODULE, + False, + ), + (_report(_error("collection failure", IMPORT_HELPER)), NOTHING, False), + (_report('', ""), CALC_HELPER, False), + (_report(), CALC_HELPER, False), + ("", CALC_HELPER, False), + ("", CALC_HELPER, False), + ], +) +def test_red_witness(report, added, red): + assert red_witness(report, added) is red + + +PYTEST = [sys.executable, "-B", "-m", "pytest", "-q", "-p", "no:cacheprovider", "-p", "no:xdist"] + + +def _reverted(tmp_path, body, added=CALC_HELPER, conftest=None): + (tmp_path / "test_probe.py").write_text(body, encoding="utf-8") + if conftest is not None: + (tmp_path / "conftest.py").write_text(conftest, encoding="utf-8") + (tmp_path / "calc.py").write_text("value = 1\n", encoding="utf-8") + env = dict(os.environ, PYTHONPATH=str(tmp_path)) + return _test_reverted_candidate( + CANDIDATE, [*PYTEST, "test_probe.py"], env, str(tmp_path), added=added + ) + + +def _inconclusive(result, code): + assert result.status is FixvalStatus.BLOCK, result + assert [f.id for f in result.findings] == ["FIXVAL_INCONCLUSIVE"] + assert "pytest exit %d without a failed test" % code in result.block_message + + +def test_real_collection_import_of_an_added_name_is_red(tmp_path): + result = _reverted(tmp_path, "from calc import helper\n\ndef test_it():\n helper()\n") + assert result.status is FixvalStatus.PASS, result.block_message + + +def test_real_collection_import_of_another_name_is_inconclusive(tmp_path): + _inconclusive(_reverted(tmp_path, "from calc import other\n\ndef test_it():\n other()\n"), 2) + + +def test_real_collection_error_that_is_not_an_import_is_inconclusive(tmp_path): + _inconclusive(_reverted(tmp_path, "raise RuntimeError('broken')\n\ndef test_it():\n pass\n"), 2) + + +FIXTURE = "import pytest\n\n@pytest.fixture\ndef dep():\n %s\n" + + +def test_real_setup_import_of_an_added_name_is_red(tmp_path): + conftest = FIXTURE % "from calc import helper\n return helper" + result = _reverted(tmp_path, "def test_it(dep):\n dep()\n", conftest=conftest) + assert result.status is FixvalStatus.PASS, result.block_message + + +def test_real_setup_error_that_is_not_an_import_is_inconclusive(tmp_path): + conftest = FIXTURE % "raise RuntimeError('db down')" + _inconclusive(_reverted(tmp_path, "def test_it(dep):\n pass\n", conftest=conftest), 1) + + +def test_real_teardown_error_beside_a_passing_test_is_inconclusive(tmp_path): + conftest = FIXTURE % "yield\n from calc import helper" + _inconclusive(_reverted(tmp_path, "def test_it(dep):\n pass\n", conftest=conftest), 1) + + +def test_real_failed_test_is_red_and_the_report_is_removed(tmp_path, monkeypatch): + reports = tmp_path / "reports" + reports.mkdir() + monkeypatch.setattr(tempfile, "tempdir", str(reports)) + result = _reverted(tmp_path, "def test_it():\n assert False\n", added=None) + assert result.status is FixvalStatus.PASS, result.block_message + assert list(reports.iterdir()) == [] + + +def _git(root, *arguments): + return subprocess.run( + ["git", "-c", "user.name=t", "-c", "user.email=t@t", "-c", "commit.gpgsign=false", *arguments], + cwd=root, + capture_output=True, + text=True, + check=True, + ).stdout + + +def test_run_fixval_reverted_module_that_breaks_at_import_is_inconclusive(tmp_path): + # main counted this exit 2 as RED; the fix removed a raise, it added no name the test imports + _git(tmp_path, "init", "-q") + (tmp_path / "src").mkdir() + (tmp_path / "tests").mkdir() + (tmp_path / "src/calc.py").write_text("raise RuntimeError('broken')\nvalue = 1\n", encoding="utf-8") + (tmp_path / "tests/test_calc.py").write_text("def test_placeholder():\n pass\n", encoding="utf-8") + _git(tmp_path, "add", "--all") + _git(tmp_path, "commit", "-q", "-m", "base") + (tmp_path / "src/calc.py").write_text("value = 1\n", encoding="utf-8") + (tmp_path / "tests/test_calc.py").write_text( + "from calc import value\n\ndef test_calc():\n assert value == 1\n", encoding="utf-8" + ) + _git(tmp_path, "add", "--all") + diff = _git(tmp_path, "diff", "--cached") + candidate = classify_fixval_candidate(["src/calc.py", "tests/test_calc.py"]) + result = run_fixval(candidate, [*PYTEST, "-q"], tmp_path, "fix calc", diff) + _inconclusive(result, 2) + assert (tmp_path / "src/calc.py").read_text(encoding="utf-8") == "value = 1\n" diff --git a/tests/test_fixval_test_support.py b/tests/test_fixval_test_support.py index 26d44808..01b3e6be 100644 --- a/tests/test_fixval_test_support.py +++ b/tests/test_fixval_test_support.py @@ -492,8 +492,8 @@ def first_run(*args, **kwargs): assert [finding.id for finding in result.findings] == expected -def test_first_run_collection_error_stays_pass_as_on_main(tmp_path): - # the test imports a name the fix adds: exit 2 on reverted code, judged as on main (#28) +def test_first_run_import_of_a_name_the_fix_adds_is_pass(tmp_path): + # the test imports a name the fix adds: exit 2 on reverted code, RED evidence (#28) fixed = { "src/calc.py": "value = 2\n\ndef helper():\n return value\n", "tests/test_calc.py": "from calc import helper\n\ndef test_calc():\n assert helper() == 2\n", From 590054a91db128c92b2f62ff44f19ee6898ed290 Mon Sep 17 00:00:00 2001 From: Minxi Hou Date: Thu, 29 Oct 2026 06:36:40 -0400 Subject: [PATCH 2/8] tests: mark the AST-rename exec in test_fixval.py as intended (S102) Touching tests/test_fixval.py makes lint_changed.py check the whole file, and this pre-existing exec of a fixed literal tripped ruff S102. Refs #28 --- tests/test_fixval.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_fixval.py b/tests/test_fixval.py index 14861b62..926a58b3 100644 --- a/tests/test_fixval.py +++ b/tests/test_fixval.py @@ -684,7 +684,7 @@ def test_renamed_code_executes_without_name_error(self): _ast.fix_missing_locations(transformed) out = _ast.unparse(transformed) ns: dict = {} - exec(compile(out, "", "exec"), ns) + exec(compile(out, "", "exec"), ns) # noqa: S102 - runs the literal above, renamed assert ns["compute"]() == 42 From e127032ab06bc59c27b1a78c272bdeb66a4dd4a3 Mon Sep 17 00:00:00 2001 From: Minxi Hou Date: Thu, 29 Oct 2026 08:05:37 -0400 Subject: [PATCH 3/8] fixval: skip unexpected passes as RED and keep the JUnit report in a private dir (#28) Review follow-up: - red_witness no longer counts a whose message is a strict XPASS ('[XPASS(strict)]') or a unittest unexpected success ('Failed: Unexpected success'); #28 treats both as invalid witnesses. Tested with reports from real pytest runs. - The reverted run writes junit.xml inside a fresh mkdtemp directory that is removed with rmtree, instead of unlinking a predictable mkstemp path in the shared TMPDIR. - New tests pin that a require_witness run passes no --junitxml, that the report directory is private (0700), empty before the run and removed after, and that a path that is not an importable module (pkg/foo-bar.py) adds no names. --- src/code_forge/_fixval_junit.py | 17 ++++-- src/code_forge/fixval.py | 15 ++--- tests/test_fixval_junit_witness.py | 94 ++++++++++++++++++++++++++++++ 3 files changed, 112 insertions(+), 14 deletions(-) diff --git a/src/code_forge/_fixval_junit.py b/src/code_forge/_fixval_junit.py index 3ee2e13c..261d19ce 100644 --- a/src/code_forge/_fixval_junit.py +++ b/src/code_forge/_fixval_junit.py @@ -31,6 +31,11 @@ _NO_ATTRIBUTE = re.compile(r"module '([\w.]+)' has no attribute '(\w+)'") _NO_MODULE = re.compile(r"No module named '([\w.]+)'|bad magic number in '([\w.]+)'") +# A with one of these messages is a test that passed when it was +# expected to fail: a strict xfail that passed, or a unittest expectedFailure +# that succeeded. #28 counts neither as a RED witness. +_UNEXPECTED_PASS = ("[XPASS(strict)]", "Failed: Unexpected success") + @dataclass(frozen=True) class AddedNames: @@ -125,9 +130,10 @@ def _is_added_import_failure(text: str, added: AddedNames) -> bool: def red_witness(junit_xml: str, added: AddedNames) -> bool: """Whether the report shows the reverted code failing a test. - A `` is a failed test call. An `` counts only when it is - a collection or setup error whose text is an import of a name in - `added`; teardown errors never count. A report that does not parse + A `` is a failed test call, unless it reports an unexpected + pass (a strict XPASS or a unittest unexpected success). An `` + counts only when it is a collection or setup error whose text is an + import of a name in `added`; teardown errors never count. A report that does not parse shows nothing. """ try: @@ -137,8 +143,9 @@ def red_witness(junit_xml: str, added: AddedNames) -> bool: except ElementTree.ParseError: return False for case in root.iter("testcase"): - if case.find("failure") is not None: - return True + for failure in case.findall("failure"): + if not failure.get("message", "").startswith(_UNEXPECTED_PASS): + return True for error in case.findall("error"): message = error.get("message", "") if message.startswith("failed on teardown"): diff --git a/src/code_forge/fixval.py b/src/code_forge/fixval.py index 5a8ab79c..f86eeee0 100644 --- a/src/code_forge/fixval.py +++ b/src/code_forge/fixval.py @@ -20,6 +20,7 @@ import logging import os import re +import shutil import signal import subprocess import sys @@ -616,11 +617,10 @@ def _test_reverted_candidate( # #183: each bounded FIXVAL test run starts with a progress line, so the # MCP job's inactivity watchdog never sees two runs as one silent span. progress.emit("fixval: reverted-fix run") - report = None + report_dir = report = None if not require_witness and _runs_pytest(scoped_cmd): - fd, report = tempfile.mkstemp(prefix="fixval-junit-", suffix=".xml") - os.close(fd) - os.unlink(report) # pytest writes it; a stale empty file would read as a report + report_dir = tempfile.mkdtemp(prefix="fixval-junit-") + report = os.path.join(report_dir, "junit.xml") try: return _judge_reverted_run( candidate, @@ -634,11 +634,8 @@ def _test_reverted_candidate( report, ) finally: - if report is not None: - try: - os.unlink(report) - except FileNotFoundError: - pass + if report_dir is not None: + shutil.rmtree(report_dir, ignore_errors=True) def _judge_reverted_run( diff --git a/tests/test_fixval_junit_witness.py b/tests/test_fixval_junit_witness.py index 9be6560a..fb01174f 100644 --- a/tests/test_fixval_junit_witness.py +++ b/tests/test_fixval_junit_witness.py @@ -70,6 +70,12 @@ def test_added_names_maps_paths_to_modules_and_skips_names_that_already_existed( assert found.modules == {"lib.pkg"} +def test_a_path_that_is_not_an_importable_module_adds_nothing(): + # pkg/foo-bar.py cannot be imported, so no import error can name it + assert added_names(_patch("pkg/foo-bar.py", ["pass"], ["def helper():"])) == NOTHING + assert added_names(_patch("pkg/foo-bar.py", [], ["def helper():"], new_file=True)) == NOTHING + + def test_unreadable_patch_adds_nothing(): assert added_names("@@ not a diff\n+def helper():\n") == NOTHING @@ -143,6 +149,53 @@ def test_red_witness(report, added, red): assert red_witness(report, added) is red +UNEXPECTED_PASSES = """\ +import unittest + +import pytest + + +@pytest.mark.xfail(strict=True) +def test_strict_xpass(): + pass + + +@pytest.mark.xfail(strict=True, reason="bug 7") +def test_strict_xpass_with_reason(): + pass + + +class Case(unittest.TestCase): + @unittest.expectedFailure + def test_unexpected_success(self): + pass +""" + + +def _real_report(tmp_path, body): + (tmp_path / "test_probe.py").write_text(body, encoding="utf-8") + report = tmp_path / "junit.xml" + run = subprocess.run( + [*PYTEST, "--junitxml=%s" % report, "test_probe.py"], + cwd=tmp_path, + capture_output=True, + text=True, + ) + assert run.returncode == 1, run.stdout + run.stderr + return report.read_text(encoding="utf-8") + + +def test_real_unexpected_passes_are_not_red(tmp_path): + report = _real_report(tmp_path, UNEXPECTED_PASSES) + assert report.count(" Date: Thu, 29 Oct 2026 08:23:03 -0400 Subject: [PATCH 4/8] Test every is read and a stuck report dir keeps the verdict (#28) Two unit cases put an unexpected pass before a real in one testcase and expect RED, so red_witness must keep reading past the first . A new test makes os.rmdir fail on the private report directory and checks _test_reverted_candidate still returns its verdict, which needs ignore_errors=True on the cleanup rmtree. --- tests/test_fixval_junit_witness.py | 31 ++++++++++++++++++++++++++++++ 1 file changed, 31 insertions(+) diff --git a/tests/test_fixval_junit_witness.py b/tests/test_fixval_junit_witness.py index fb01174f..39f10021 100644 --- a/tests/test_fixval_junit_witness.py +++ b/tests/test_fixval_junit_witness.py @@ -98,6 +98,17 @@ def _error(message, text=""): [ (_report(''), NOTHING, True), (_report(_error("collection failure", IMPORT_HELPER)), CALC_HELPER, True), + # one testcase: an unexpected pass, then a real failure (every is read) + ( + _report(''), + NOTHING, + True, + ), + ( + _report(''), + NOTHING, + True, + ), (_report(_error("failed on setup with "%s"" % IMPORT_HELPER)), CALC_HELPER, True), ( _report( @@ -298,6 +309,26 @@ def test_report_goes_in_a_fresh_private_directory_that_is_removed(tmp_path, monk assert list(reports.iterdir()) == [] +def test_a_report_directory_that_cannot_be_removed_does_not_lose_the_verdict(tmp_path, monkeypatch): + reports = tmp_path / "reports" + reports.mkdir() + monkeypatch.setattr(tempfile, "tempdir", str(reports)) + real_rmdir = os.rmdir + + def rmdir(path, *args, **kwargs): + if "fixval-junit-" in str(path): + raise PermissionError(13, "injected rmdir failure", str(path)) + return real_rmdir(path, *args, **kwargs) + + calls = _spy_run(monkeypatch, 1) + monkeypatch.setattr(os, "rmdir", rmdir) + result = _test_reverted_candidate(CANDIDATE, [*PYTEST, "t.py"], {}, str(tmp_path)) + monkeypatch.setattr(os, "rmdir", real_rmdir) + _inconclusive(result, 1) + [_argv, (_tag, directory, _mode, _entries)] = calls + assert os.listdir(directory) == [] # only the empty directory is left behind + + def _git(root, *arguments): return subprocess.run( ["git", "-c", "user.name=t", "-c", "user.email=t@t", "-c", "commit.gpgsign=false", *arguments], From bb90558fc727a52c25de76d814c8c8a076baa0ba Mon Sep 17 00:00:00 2001 From: Minxi Hou Date: Thu, 29 Oct 2026 09:37:28 -0400 Subject: [PATCH 5/8] fixval: count imports of new submodules and renamed files as RED, keep --junitxml before -- (#28) - `from pkg import newmod` (or `pkg.newmod` attribute access) where the fix creates pkg/newmod.py is now RED instead of inconclusive. - added_names reads multi-line parenthesised imports, keeping separate state for the pre- and post-image of each hunk. - A renamed or copied file adds its module, minus the dotted names the old path already had. - A pytest command containing `--` gets --junitxml before that `--`; appended after it, pytest took the flag as a path and exited 4. - Known limits are now documented in comments: only `src/` is stripped from module paths, a missing namespace-package parent is not RED, a hunk that starts inside a parenthesised import misses its names, and entity-expansion caps need expat 2.4.0 or later. Tests: a real failure before an unexpected pass in one testcase (kills the only-last- mutant), the cases above, and a real pytest run with `--` in the command. --- src/code_forge/_fixval_junit.py | 71 ++++++++++++----- src/code_forge/fixval.py | 19 ++++- tests/test_fixval_junit_witness.py | 119 +++++++++++++++++++++++++++++ 3 files changed, 191 insertions(+), 18 deletions(-) diff --git a/src/code_forge/_fixval_junit.py b/src/code_forge/_fixval_junit.py index 261d19ce..c45f67a3 100644 --- a/src/code_forge/_fixval_junit.py +++ b/src/code_forge/_fixval_junit.py @@ -26,6 +26,8 @@ _FROM_IMPORT = re.compile(r"from\s+[\w.]+\s+import\s+\(?([^)#]*)") _IMPORT = re.compile(r"import\s+([^#]*)") _ALIAS = re.compile(r"[\w.]+(?:\s+as\s+(\w+))?") +# `from x import (` with the closing parenthesis on a later line. +_OPEN_FROM_IMPORT = re.compile(r"from\s+[\w.]+\s+import\s*\([^)#]*(?:#.*)?$") _CANNOT_IMPORT = re.compile(r"cannot import name '(\w+)' from '([\w.]+)'") _NO_ATTRIBUTE = re.compile(r"module '([\w.]+)' has no attribute '(\w+)'") @@ -55,20 +57,16 @@ def _module_names(path: str) -> set[str]: if not parts or not all(part.isidentifier() for part in parts): return set() names = {".".join(parts)} + # Known limit: only `src/` is stripped. A module under another root + # (`lib/pkg/mod.py` imported as `pkg.mod`) is matched by its full dotted + # path alone, so an import error that names the shorter form is not RED + # and the run stays inconclusive. Guessing roots would widen RED. if parts[0] == "src" and len(parts) > 1: names.add(".".join(parts[1:])) return names -def _imported_names(line: str) -> set[str]: - match = _FROM_IMPORT.match(line) - if match: - clause = match.group(1) - else: - match = _IMPORT.match(line) - if not match: - return set() - clause = match.group(1) +def _clause_names(clause: str) -> set[str]: names = set() for item in clause.split(","): alias = _ALIAS.fullmatch(item.strip()) @@ -77,6 +75,31 @@ def _imported_names(line: str) -> set[str]: return names +def _imported_names(line: str) -> set[str]: + match = _FROM_IMPORT.match(line) or _IMPORT.match(line) + return _clause_names(match.group(1)) if match else set() + + +class _Side: + """One side (pre- or post-image) of a hunk, read in order, so the + lines inside a multi-line `from x import (...)` bind their names. + + Known limit: a hunk that starts inside such an import does not see its + opening line, so the names there are not read (no RED from them).""" + + def __init__(self): + self.open = False + + def names(self, line: str) -> set[str]: + if self.open: + inside, closed, _rest = line.split("#", 1)[0].partition(")") + self.open = not closed + return _clause_names(inside) + if _OPEN_FROM_IMPORT.match(line): + self.open = True + return _defined(line) + + def _defined(line: str) -> set[str]: """Module-level names a line binds; indented lines bind none.""" for pattern in (_DEFINITION, _ASSIGNMENT): @@ -102,28 +125,39 @@ def added_names(patch: str) -> AddedNames: dotted = _module_names(path or "") if not dotted: continue - if patched_file_path(entry, source=True) is None: - modules.update(dotted) + source = patched_file_path(entry, source=True) + if source is None or source != path: + # A created, renamed or copied file: its module exists only + # with the fix (a rename's old name is gone, not added). + modules.update(dotted - _module_names(source or "")) added: set[str] = set() removed: set[str] = set() for hunk in entry: + new, old = _Side(), _Side() for line in hunk: if line.is_added: - added |= _defined(line.value) + added |= new.names(line.value) elif line.is_removed: - removed |= _defined(line.value) + removed |= old.names(line.value) + else: + new.names(line.value) + old.names(line.value) names.update((module, name) for module in dotted for name in added - removed) return AddedNames(frozenset(names), frozenset(modules)) def _is_added_import_failure(text: str, added: AddedNames) -> bool: + # `from pkg import name` names a binding or a submodule the fix adds. for name, module in _CANNOT_IMPORT.findall(text): - if (module, name) in added.names: + if (module, name) in added.names or "%s.%s" % (module, name) in added.modules: return True for module, name in _NO_ATTRIBUTE.findall(text): - if (module, name) in added.names: + if (module, name) in added.names or "%s.%s" % (module, name) in added.modules: return True # A module the fix adds is missing, or only a stale foreign .pyc is left. + # Known limit: a missing parent (`No module named 'pkg.new'` when the fix + # creates `pkg/new/mod.py` as a namespace package) is not RED, because + # the patch alone does not show whether `pkg/new/` existed before. return any(set(modules) & added.modules for modules in _NO_MODULE.findall(text)) @@ -137,8 +171,11 @@ def red_witness(junit_xml: str, added: AddedNames) -> bool: shows nothing. """ try: - # pytest wrote it; ElementTree resolves no external entities and the - # bundled expat caps entity expansion, so a hostile conftest gains nothing. + # pytest wrote it into a private directory. ElementTree resolves no + # external entities. Entity expansion (billion laughs) is capped only + # by expat 2.4.0 and later (pyexpat.EXPAT_VERSION); with an older + # expat a crafted report could use much memory. Accepted: writing + # one takes code that already runs in the test process (conftest). root = ElementTree.fromstring(junit_xml) # noqa: S314 except ElementTree.ParseError: return False diff --git a/src/code_forge/fixval.py b/src/code_forge/fixval.py index f86eeee0..a0bdfb65 100644 --- a/src/code_forge/fixval.py +++ b/src/code_forge/fixval.py @@ -638,10 +638,27 @@ def _test_reverted_candidate( shutil.rmtree(report_dir, ignore_errors=True) +def _with_junitxml(cmd: list[str], report: str) -> list[str]: + """`cmd` with `--junitxml=` added as a pytest option. + + pytest reads every argument after a bare `--` as a path, so the flag + goes before the first `--` among pytest's own arguments. A `--` that + belongs to a wrapper (`uv run -- pytest`) comes before the pytest part + and is left alone.""" + flag = "--junitxml=%s" % report + argv = _unwrap_runner([str(arg) for arg in cmd]) + # argv is a suffix of cmd; `coverage run` swaps its prefix for one + # "python", so the offset holds for every argument after the program. + if "--" in argv: + at = len(cmd) - len(argv) + argv.index("--") + return [*cmd[:at], flag, *cmd[at:]] + return [*cmd, flag] + + def _judge_reverted_run( candidate, scoped_cmd, run_env, repo_root, reverted_files, require_witness, timeout, added, report ) -> FixvalResult | None: - cmd = scoped_cmd if report is None else [*scoped_cmd, "--junitxml=%s" % report] + cmd = scoped_cmd if report is None else _with_junitxml(scoped_cmd, report) try: result = subprocess.run( cmd, diff --git a/tests/test_fixval_junit_witness.py b/tests/test_fixval_junit_witness.py index 39f10021..a2395525 100644 --- a/tests/test_fixval_junit_witness.py +++ b/tests/test_fixval_junit_witness.py @@ -76,6 +76,66 @@ def test_a_path_that_is_not_an_importable_module_adds_nothing(): assert added_names(_patch("pkg/foo-bar.py", [], ["def helper():"], new_file=True)) == NOTHING +def _hunk(path, lines): + """A modification hunk; each line starts with "+", "-" or " ".""" + old = sum(not line.startswith("+") for line in lines) + new = sum(not line.startswith("-") for line in lines) + head = "diff --git a/%s b/%s\n--- a/%s\n+++ b/%s\n" % (path, path, path, path) + return head + "@@ -1,%d +1,%d @@\n" % (old, new) + "".join(line + "\n" for line in lines) + + +@pytest.mark.parametrize( + ("lines", "names"), + [ + # a new multi-line import binds every name inside the parentheses + (["+from .impl import (", "+ run, # see (a)", "+ stop as halt,", "+)"], {"run", "halt"}), + # each side keeps its own state: the removed lines leave the old import open + (["-from .impl import (", "- run,", "+from .impl import (stop,", " )"], {"stop"}), + # a name added inside an import whose opening line is context + ([" from .impl import (", " run,", "+ stop,", " )"], {"stop"}), + # a name moved from one line of the import to another is not new + ([" from .impl import (", "- run, stop", "+ run,", "+ stop,", " )"], set()), + (["+from .impl import (run,", "+ stop)", "+def after():"], {"run", "stop", "after"}), + # after the closing line, indented lines bind nothing again + ([" from .impl import (", " run,", " )", "+ value = 1"], set()), + ], +) +def test_added_names_reads_multi_line_imports(lines, names): + found = added_names(_hunk("src/calc.py", lines)) + assert {name for _module, name in found.names} == names + + +RENAME = """\ +diff --git a/src/old.py b/src/pkg/newmod.py +similarity index 90% +rename from src/old.py +rename to src/pkg/newmod.py +--- a/src/old.py ++++ b/src/pkg/newmod.py +@@ -1 +1 @@ +-X = 1 ++X = 2 +""" + + +def test_a_renamed_or_copied_file_adds_its_module(): + assert added_names(RENAME).modules == {"src.pkg.newmod", "pkg.newmod"} + copy = RENAME.replace("rename ", "copy ") + assert added_names(copy).modules == {"src.pkg.newmod", "pkg.newmod"} + + +def test_a_rename_to_a_name_the_module_already_had_adds_no_module(): + # src/calc.py was importable as `calc` before the fix moved it to calc.py + rename = RENAME.replace("src/old.py", "src/calc.py").replace("src/pkg/newmod.py", "calc.py") + assert added_names(rename).modules == frozenset() + + +def test_a_modified_file_adds_no_module(): + assert added_names(_hunk("src/calc.py", ["-X = 1", "+X = 2"])).modules == frozenset() + plain = "--- a/src/calc.py\n+++ b/src/calc.py\n@@ -1 +1 @@\n-X = 1\n+X = 2\n" + assert added_names(plain).modules == frozenset() + + def test_unreadable_patch_adds_nothing(): assert added_names("@@ not a diff\n+def helper():\n") == NOTHING @@ -91,6 +151,7 @@ def _error(message, text=""): IMPORT_HELPER = "E ImportError: cannot import name 'helper' from 'calc' (/r/src/calc.py)" +IMPORT_NEWMOD = "E ImportError: cannot import name 'newmod' from 'pkg' (/r/src/pkg/__init__.py)" @pytest.mark.parametrize( @@ -109,7 +170,27 @@ def _error(message, text=""): NOTHING, True, ), + # a real failure, then an unexpected pass (the last is not the only one read) + ( + _report(''), + NOTHING, + True, + ), (_report(_error("failed on setup with "%s"" % IMPORT_HELPER)), CALC_HELPER, True), + # `from pkg import newmod`, where the fix creates the module pkg/newmod.py + (_report(_error("collection failure", IMPORT_NEWMOD)), NEW_MODULE, True), + ( + _report( + _error("collection failure", "E AttributeError: module 'pkg' has no attribute 'newmod'") + ), + NEW_MODULE, + True, + ), + ( + _report(_error("collection failure", IMPORT_NEWMOD.replace("newmod", "other"))), + NEW_MODULE, + False, + ), ( _report( _error("collection failure", "E AttributeError: module 'calc' has no attribute 'helper'") @@ -268,6 +349,44 @@ def test_real_failed_test_is_red_and_the_report_is_removed(tmp_path, monkeypatch assert list(reports.iterdir()) == [] +def test_real_import_of_a_module_the_fix_creates_is_red(tmp_path): + (tmp_path / "pkg").mkdir() + (tmp_path / "pkg" / "__init__.py").write_text("", encoding="utf-8") + body = "from pkg import newmod\n\ndef test_it():\n newmod.run()\n" + result = _reverted(tmp_path, body, added=NEW_MODULE) + assert result.status is FixvalStatus.PASS, result.block_message + + +def test_real_command_with_a_double_dash_still_gets_its_report(tmp_path): + (tmp_path / "test_probe.py").write_text("def test_it():\n assert False\n", encoding="utf-8") + cmd = [*PYTEST, "--", "test_probe.py"] + result = _test_reverted_candidate(CANDIDATE, cmd, dict(os.environ), str(tmp_path), added=NOTHING) + assert result.status is FixvalStatus.PASS, result.block_message + + +@pytest.mark.parametrize( + ("cmd", "expected"), + [ + (["pytest", "-q", "--", "t.py"], ["pytest", "-q", "J", "--", "t.py"]), + ( + ["python", "-B", "-m", "pytest", "--", "pytest"], + ["python", "-B", "-m", "pytest", "J", "--", "pytest"], + ), + (["uv", "run", "--", "pytest", "t.py"], ["uv", "run", "--", "pytest", "t.py", "J"]), + ( + ["coverage", "run", "-m", "pytest", "--", "t.py"], + ["coverage", "run", "-m", "pytest", "J", "--", "t.py"], + ), + (["pytest", "t.py"], ["pytest", "t.py", "J"]), + ], +) +def test_junitxml_goes_before_a_double_dash_of_pytest(tmp_path, monkeypatch, cmd, expected): + calls = _spy_run(monkeypatch, 1) + _test_reverted_candidate(CANDIDATE, cmd, {}, str(tmp_path)) + argv = [("J" if arg.startswith("--junitxml=") else arg) for arg in calls[0]] + assert argv == expected + + def test_real_unexpected_passes_alone_are_inconclusive(tmp_path): _inconclusive(_reverted(tmp_path, UNEXPECTED_PASSES), 1) From f39df7dd1184b4d108a3211c0759637884fa34c6 Mon Sep 17 00:00:00 2001 From: Minxi Hou Date: Thu, 29 Oct 2026 10:35:03 -0400 Subject: [PATCH 6/8] fixval: read backslash-continued imports, add boundary tests A module-level import line that ends with a backslash now carries on to the next line, so `from pkg.o import a, \` followed by `newname` binds `newname`. Before, that name was not read and an import error for it stayed inconclusive instead of RED. Only an import at column 0 starts a continuation; an assignment or an indented import does not. New tests close gaps that surviving mutants showed: - a comment after `from x import (` still opens the import; - a parenthesised import inside a function binds nothing; - an import left open at the end of one hunk does not carry into the next hunk, with an indented name and with a column-0 binding there; - `cannot import name` and `has no attribute` need the module to be exactly `M.X` (not a longer name, not another parent), and `No module named` needs a module the fix creates, not one it starts with. Refs #28 --- src/code_forge/_fixval_junit.py | 14 ++++- tests/test_fixval_junit_witness.py | 82 ++++++++++++++++++++++++++++++ 2 files changed, 94 insertions(+), 2 deletions(-) diff --git a/src/code_forge/_fixval_junit.py b/src/code_forge/_fixval_junit.py index c45f67a3..d5f3aae6 100644 --- a/src/code_forge/_fixval_junit.py +++ b/src/code_forge/_fixval_junit.py @@ -82,21 +82,31 @@ def _imported_names(line: str) -> set[str]: class _Side: """One side (pre- or post-image) of a hunk, read in order, so the - lines inside a multi-line `from x import (...)` bind their names. + lines of a multi-line import bind their names: inside the parentheses + of `from x import (...)`, or after a module-level import line that ends + with a backslash. Known limit: a hunk that starts inside such an import does not see its opening line, so the names there are not read (no RED from them).""" def __init__(self): - self.open = False + self.open = False # inside `from x import (` + self.continued = False # the previous import line ended with `\` def names(self, line: str) -> set[str]: if self.open: inside, closed, _rest = line.split("#", 1)[0].partition(")") self.open = not closed return _clause_names(inside) + if self.continued: + # A trailing `\` follows a comma, so it is an item of its own + # that matches no name. + self.continued = line.rstrip().endswith("\\") + return _clause_names(line.split("#", 1)[0]) if _OPEN_FROM_IMPORT.match(line): self.open = True + elif (_FROM_IMPORT.match(line) or _IMPORT.match(line)) and line.rstrip().endswith("\\"): + self.continued = True return _defined(line) diff --git a/tests/test_fixval_junit_witness.py b/tests/test_fixval_junit_witness.py index a2395525..274f79e3 100644 --- a/tests/test_fixval_junit_witness.py +++ b/tests/test_fixval_junit_witness.py @@ -105,6 +105,67 @@ def test_added_names_reads_multi_line_imports(lines, names): assert {name for _module, name in found.names} == names +def test_a_comment_on_the_opening_line_still_opens_the_import(): + lines = ["+from .impl import ( # noqa: F401", "+ helper,", "+)"] + found = added_names(_hunk("src/calc.py", lines)) + assert found.names == {("src.calc", "helper"), ("calc", "helper")} + # the reverted run's import error for that name is RED + assert red_witness(_report(_error("collection failure", IMPORT_HELPER)), found) is True + + +def test_a_function_level_parenthesised_import_binds_nothing(): + lines = [ + "+def load():", + "+ from .impl import (", + "+ helper,", + "+ )", + "+ return helper", + ] + found = added_names(_hunk("src/calc.py", lines)) + assert {name for _module, name in found.names} == {"load"} + + +@pytest.mark.parametrize( + ("lines", "names"), + [ + (["+from pkg.o import a, \\", "+ newname"], {"a", "newname"}), + ( + ["+from pkg.o import \\", "+ a, \\", "+ newname", "+VALUE = 1"], + {"a", "newname", "VALUE"}, + ), + (["+import json, \\", "+ csv as table"], {"json", "table"}), + (["+from pkg.o import a,\\", "+ b,\\", "+ newname # noqa"], {"a", "b", "newname"}), + ([" from pkg.o import a, \\", "+ newname"], {"newname"}), + # only an import line continues; an assignment's backslash does not + (["+total = 1 + \\", "+ newname"], {"total"}), + (["+def f():", "+ from pkg.o import a, \\", "+ b"], {"f"}), + ], +) +def test_added_names_reads_backslash_continued_imports(lines, names): + found = added_names(_hunk("src/calc.py", lines)) + assert {name for _module, name in found.names} == names + + +def test_a_backslash_continued_import_of_an_added_name_is_red(): + found = added_names(_hunk("src/calc.py", [" from pkg.o import a, \\", "+ helper"])) + assert red_witness(_report(_error("collection failure", IMPORT_HELPER)), found) is True + + +def test_an_import_left_open_at_the_end_of_a_hunk_does_not_leak_into_the_next(): + # The first hunk stops inside `from .impl import (`; the second hunk is + # elsewhere in the file, inside a call, where `stop,` binds nothing. + patch = ( + "diff --git a/src/calc.py b/src/calc.py\n--- a/src/calc.py\n+++ b/src/calc.py\n" + "@@ -1,1 +1,2 @@\n from .impl import (\n+ run,\n" + "@@ -20,2 +21,3 @@\n call(\n+ stop,\n )\n" + ) + found = added_names(patch) + assert {name for _module, name in found.names} == {"run"} + # the same with a module-level binding at column 0 in the second hunk + found = added_names(patch.replace("+ stop,\n )\n", "+LIMIT = 3\n )\n")) + assert {name for _module, name in found.names} == {"run", "LIMIT"} + + RENAME = """\ diff --git a/src/old.py b/src/pkg/newmod.py similarity index 90% @@ -152,6 +213,8 @@ def _error(message, text=""): IMPORT_HELPER = "E ImportError: cannot import name 'helper' from 'calc' (/r/src/calc.py)" IMPORT_NEWMOD = "E ImportError: cannot import name 'newmod' from 'pkg' (/r/src/pkg/__init__.py)" +NO_ATTRIBUTE_NEWMOD = "E AttributeError: module 'pkg' has no attribute 'newmod'" +NEWMOD2 = AddedNames(frozenset(), frozenset({"pkg.newmod2"})) @pytest.mark.parametrize( @@ -191,6 +254,25 @@ def _error(message, text=""): NEW_MODULE, False, ), + # false-RED boundaries: the module must be exactly `M.X`, and `No module + # named` must name a module the fix creates, not one that starts with it + (_report(_error("collection failure", IMPORT_NEWMOD)), NEWMOD2, False), + ( + _report(_error("collection failure", IMPORT_NEWMOD.replace("'pkg'", "'other'"))), + NEW_MODULE, + False, + ), + (_report(_error("collection failure", NO_ATTRIBUTE_NEWMOD)), NEWMOD2, False), + ( + _report(_error("collection failure", NO_ATTRIBUTE_NEWMOD.replace("'pkg'", "'other'"))), + NEW_MODULE, + False, + ), + ( + _report(_error("collection failure", "E ModuleNotFoundError: No module named 'pkg.newer'")), + AddedNames(frozenset(), frozenset({"pkg.new"})), + False, + ), ( _report( _error("collection failure", "E AttributeError: module 'calc' has no attribute 'helper'") From bdcdba7f888c1b2118400d44ab2f7f7f480e1c1e Mon Sep 17 00:00:00 2001 From: Minxi Hou Date: Thu, 29 Oct 2026 11:08:20 -0400 Subject: [PATCH 7/8] fixval: a backslash at the end of a comment continues no import `import os # see \` was read as a continued import, so the next line (`def helper():`) was parsed as import names and `helper` was lost. An import error for `helper` then stayed inconclusive; at bb90558 it was RED. The backslash check now looks at the line with its comment removed, for both the opening line and continued lines. New tests: - `import os # see \` followed by `def helper():` binds both names, and the import error for `helper` is RED; - a continued line whose comment ends in a backslash ends the import; - a plain import does not continue onto the next line; - a parenthesised import with a `\` inside ends at its `)`; - `No module named 'pkg'` is not RED when the fix adds `pkg.newmod`, and `No module named 'pkg.new'` is not RED when it adds `pkg.new.mod` (the namespace parent limit). Refs #28 --- src/code_forge/_fixval_junit.py | 11 +++++++---- tests/test_fixval_junit_witness.py | 24 ++++++++++++++++++++++++ 2 files changed, 31 insertions(+), 4 deletions(-) diff --git a/src/code_forge/_fixval_junit.py b/src/code_forge/_fixval_junit.py index d5f3aae6..90512825 100644 --- a/src/code_forge/_fixval_junit.py +++ b/src/code_forge/_fixval_junit.py @@ -94,18 +94,21 @@ def __init__(self): self.continued = False # the previous import line ended with `\` def names(self, line: str) -> set[str]: + code = line.split("#", 1)[0] if self.open: - inside, closed, _rest = line.split("#", 1)[0].partition(")") + inside, closed, _rest = code.partition(")") self.open = not closed return _clause_names(inside) + # A `\` at the end of a comment continues nothing. + ends_in_backslash = code.rstrip().endswith("\\") if self.continued: # A trailing `\` follows a comma, so it is an item of its own # that matches no name. - self.continued = line.rstrip().endswith("\\") - return _clause_names(line.split("#", 1)[0]) + self.continued = ends_in_backslash + return _clause_names(code) if _OPEN_FROM_IMPORT.match(line): self.open = True - elif (_FROM_IMPORT.match(line) or _IMPORT.match(line)) and line.rstrip().endswith("\\"): + elif ends_in_backslash and (_FROM_IMPORT.match(line) or _IMPORT.match(line)): self.continued = True return _defined(line) diff --git a/tests/test_fixval_junit_witness.py b/tests/test_fixval_junit_witness.py index 274f79e3..5293de74 100644 --- a/tests/test_fixval_junit_witness.py +++ b/tests/test_fixval_junit_witness.py @@ -139,6 +139,13 @@ def test_a_function_level_parenthesised_import_binds_nothing(): # only an import line continues; an assignment's backslash does not (["+total = 1 + \\", "+ newname"], {"total"}), (["+def f():", "+ from pkg.o import a, \\", "+ b"], {"f"}), + # a plain import does not continue onto the next line + (["+import json", "+VALUE = 1"], {"json", "VALUE"}), + # a parenthesised import is closed by its `)`, even with a `\\` inside + (["+from pkg.o import (a, \\", "+ newname)", "+VALUE = 1"], {"a", "newname", "VALUE"}), + # a backslash at the end of a comment continues nothing + (["+import os # see \\", "+def helper():"], {"os", "helper"}), + (["+from pkg.o import a, \\", "+ b # c \\", "+VALUE = 1"], {"a", "b", "VALUE"}), ], ) def test_added_names_reads_backslash_continued_imports(lines, names): @@ -151,6 +158,12 @@ def test_a_backslash_continued_import_of_an_added_name_is_red(): assert red_witness(_report(_error("collection failure", IMPORT_HELPER)), found) is True +def test_a_definition_after_an_import_whose_comment_ends_in_a_backslash_is_red(): + lines = ["+import os # see \\", "+def helper():", "+ return os.sep"] + found = added_names(_hunk("src/calc.py", lines)) + assert red_witness(_report(_error("collection failure", IMPORT_HELPER)), found) is True + + def test_an_import_left_open_at_the_end_of_a_hunk_does_not_leak_into_the_next(): # The first hunk stops inside `from .impl import (`; the second hunk is # elsewhere in the file, inside a call, where `stop,` binds nothing. @@ -254,6 +267,17 @@ def _error(message, text=""): NEW_MODULE, False, ), + # a parent of a created module is not RED, nor is a namespace parent + ( + _report(_error("collection failure", "E ModuleNotFoundError: No module named 'pkg'")), + NEW_MODULE, + False, + ), + ( + _report(_error("collection failure", "E ModuleNotFoundError: No module named 'pkg.new'")), + AddedNames(frozenset(), frozenset({"pkg.new.mod"})), + False, + ), # false-RED boundaries: the module must be exactly `M.X`, and `No module # named` must name a module the fix creates, not one that starts with it (_report(_error("collection failure", IMPORT_NEWMOD)), NEWMOD2, False), From e4032a90d1e736aa199d7922f79a66f103ccabab Mon Sep 17 00:00:00 2001 From: Codex Date: Sun, 1 Nov 2026 03:30:54 -0500 Subject: [PATCH 8/8] fix(fixval): bind import witnesses to complete source and terminal errors --- src/code_forge/_fixval_junit.py | 242 ++++++---- src/code_forge/_fixval_transaction.py | 43 ++ src/code_forge/fixval.py | 8 +- .../test_fixval_junit_evidence_boundaries.py | 450 ++++++++++++++++++ tests/test_fixval_junit_witness.py | 212 ++++----- 5 files changed, 740 insertions(+), 215 deletions(-) create mode 100644 tests/test_fixval_junit_evidence_boundaries.py diff --git a/src/code_forge/_fixval_junit.py b/src/code_forge/_fixval_junit.py index 90512825..adb83751 100644 --- a/src/code_forge/_fixval_junit.py +++ b/src/code_forge/_fixval_junit.py @@ -11,7 +11,8 @@ from __future__ import annotations -import keyword +import ast +from collections.abc import Mapping import re import xml.etree.ElementTree as ElementTree from dataclasses import dataclass @@ -20,18 +21,12 @@ from .diff import iter_diff_sections, patched_file_path -# Top-level definitions on an added or removed line of a production module. -_DEFINITION = re.compile(r"(?:async\s+def|def|class)\s+([A-Za-z_]\w*)") -_ASSIGNMENT = re.compile(r"([A-Za-z_]\w*)\s*(?::[^=]*)?=(?!=)") -_FROM_IMPORT = re.compile(r"from\s+[\w.]+\s+import\s+\(?([^)#]*)") -_IMPORT = re.compile(r"import\s+([^#]*)") -_ALIAS = re.compile(r"[\w.]+(?:\s+as\s+(\w+))?") -# `from x import (` with the closing parenthesis on a later line. -_OPEN_FROM_IMPORT = re.compile(r"from\s+[\w.]+\s+import\s*\([^)#]*(?:#.*)?$") - -_CANNOT_IMPORT = re.compile(r"cannot import name '(\w+)' from '([\w.]+)'") +_CANNOT_IMPORT = re.compile(r"cannot import name '(\w+)' from '([\w.]+)'(?: \([^\r\n]*\))?") _NO_ATTRIBUTE = re.compile(r"module '([\w.]+)' has no attribute '(\w+)'") -_NO_MODULE = re.compile(r"No module named '([\w.]+)'|bad magic number in '([\w.]+)'") +_NO_MODULE = re.compile( + r"No module named '([\w.]+)'(?:; '[\w.]+' is not a package)?" + r"|bad magic number in '([\w.]+)'(?:: b[^\r\n]+)?" +) # A with one of these messages is a test that passed when it was # expected to fail: a strict xfail that passed, or a unittest expectedFailure @@ -66,112 +61,154 @@ def _module_names(path: str) -> set[str]: return names -def _clause_names(clause: str) -> set[str]: - names = set() - for item in clause.split(","): - alias = _ALIAS.fullmatch(item.strip()) - if alias: - names.add(alias.group(1) or item.strip().split(".")[0]) +def _bindings(node: ast.AST) -> set[str]: + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): + return {node.name} + if isinstance(node, (ast.Import, ast.ImportFrom)): + return {alias.asname or alias.name.split(".")[0] for alias in node.names if alias.name != "*"} + if isinstance(node, (ast.Assign, ast.AnnAssign)): + if isinstance(node, ast.AnnAssign) and node.value is None: + return set() + targets = node.targets if isinstance(node, ast.Assign) else [node.target] + return { + child.id + for target in targets + for child in ast.walk(target) + if isinstance(child, ast.Name) and isinstance(child.ctx, ast.Store) + } + return set() + + +def _possible_bindings(tree: ast.AST) -> set[str] | None: + """Conservatively include every syntactic binding in the old image.""" + if any( + isinstance(node, ast.ImportFrom) and any(alias.name == "*" for alias in node.names) + for node in ast.walk(tree) + ): + return None + names = {name for node in ast.walk(tree) for name in _bindings(node)} + for node in ast.walk(tree): + if isinstance(node, ast.Name) and isinstance(node.ctx, ast.Store): + names.add(node.id) + elif isinstance(node, (ast.ExceptHandler, ast.MatchAs, ast.MatchStar)) and node.name: + names.add(node.name) + elif isinstance(node, ast.MatchMapping) and node.rest: + names.add(node.rest) + elif isinstance(node, ast.arg): + names.add(node.arg) return names -def _imported_names(line: str) -> set[str]: - match = _FROM_IMPORT.match(line) or _IMPORT.match(line) - return _clause_names(match.group(1)) if match else set() - - -class _Side: - """One side (pre- or post-image) of a hunk, read in order, so the - lines of a multi-line import bind their names: inside the parentheses - of `from x import (...)`, or after a module-level import line that ends - with a backslash. - - Known limit: a hunk that starts inside such an import does not see its - opening line, so the names there are not read (no RED from them).""" - - def __init__(self): - self.open = False # inside `from x import (` - self.continued = False # the previous import line ended with `\` - - def names(self, line: str) -> set[str]: - code = line.split("#", 1)[0] - if self.open: - inside, closed, _rest = code.partition(")") - self.open = not closed - return _clause_names(inside) - # A `\` at the end of a comment continues nothing. - ends_in_backslash = code.rstrip().endswith("\\") - if self.continued: - # A trailing `\` follows a comma, so it is an item of its own - # that matches no name. - self.continued = ends_in_backslash - return _clause_names(code) - if _OPEN_FROM_IMPORT.match(line): - self.open = True - elif ends_in_backslash and (_FROM_IMPORT.match(line) or _IMPORT.match(line)): - self.continued = True - return _defined(line) - - -def _defined(line: str) -> set[str]: - """Module-level names a line binds; indented lines bind none.""" - for pattern in (_DEFINITION, _ASSIGNMENT): - match = pattern.match(line) - if match: - return set() if keyword.iskeyword(match.group(1)) else {match.group(1)} - return _imported_names(line) - - -def added_names(patch: str) -> AddedNames: - """Names the fix in `patch` adds: bound on an added line and not on a - removed line of the same file. A file the patch creates adds its module. - An unreadable patch adds nothing, so no import error counts as RED.""" +def added_names( + patch: str, + *, + before: Mapping[str, bytes | None] | None = None, + after: Mapping[str, bytes | None] | None = None, +) -> AddedNames: + """Compare complete transaction-owned Python images, never isolated hunk lines. + + Missing images, invalid Python, and ambiguous bindings earn no import + witness. Static bindings are evidence of a change, not proof of execution. + Only direct module statements introduce candidates; all possible prior + bindings (including nested/conditional ones) conservatively exclude them. + """ names: set[tuple[str, str]] = set() modules: set[str] = set() + if before is None or after is None: + return AddedNames(frozenset(), frozenset()) try: - blocks = [block for _path, block in iter_diff_sections(patch)] - entries = [entry for block in blocks for entry in unidiff.PatchSet(block)] + entries = [ + entry for _path, block in iter_diff_sections(patch) for entry in unidiff.PatchSet(block) + ] except (unidiff.errors.UnidiffParseError, ValueError): return AddedNames(frozenset(), frozenset()) for entry in entries: path = patched_file_path(entry) dotted = _module_names(path or "") - if not dotted: - continue source = patched_file_path(entry, source=True) + if not dotted or path not in before or path not in after: + continue + old, new = before[path], after[path] + if new is None: + continue + try: + new_tree = ast.parse(new) + old_tree = ast.parse(old) if old is not None else ast.Module(body=[], type_ignores=[]) + except (SyntaxError, ValueError, RecursionError): + continue if source is None or source != path: - # A created, renamed or copied file: its module exists only - # with the fix (a rename's old name is gone, not added). - modules.update(dotted - _module_names(source or "")) - added: set[str] = set() - removed: set[str] = set() - for hunk in entry: - new, old = _Side(), _Side() - for line in hunk: - if line.is_added: - added |= new.names(line.value) - elif line.is_removed: - removed |= old.names(line.value) - else: - new.names(line.value) - old.names(line.value) - names.update((module, name) for module in dotted for name in added - removed) + if old is None: + modules.update(dotted - _module_names(source or "")) + previous = _possible_bindings(old_tree) + if previous is None: + continue + candidates = {name for node in new_tree.body for name in _bindings(node)} + deleted = { + node.id + for node in ast.walk(new_tree) + if isinstance(node, ast.Name) and isinstance(node.ctx, ast.Del) + } + for module in dotted: + prior = previous + if source != path and module in _module_names(source or ""): + # Moving src/mod.py to mod.py does not create a new binding + # under the shared import name. Its old image must be known. + if source not in before or before[source] is None: + continue + try: + source_tree = ast.parse(before[source]) + except (SyntaxError, ValueError, RecursionError): + continue + source_bindings = _possible_bindings(source_tree) + if source_bindings is None: + continue + prior = prior | source_bindings + names.update((module, name) for name in candidates - prior - deleted) return AddedNames(frozenset(names), frozenset(modules)) +def _terminal_error(message: str, text: str) -> str | None: + """Read pytest's terminal exception, not source lines or earlier chained errors. + + Multiline exception messages/notes, exception groups and unknown renderings + are deliberately inconclusive. JUnit remains evidence from the test process, + not an authenticated boundary against a plugin that forges the XML itself. + """ + prefix = 'failed on setup with "' + if message.startswith(prefix) and message.endswith('"'): + diagnostic = message[len(prefix) : -1] + return diagnostic if "\n" not in diagnostic and "\r" not in diagnostic else None + if message != "collection failure": + return None + lines = text.rstrip().splitlines() + if not lines: + return None + # The terminal E block must contain one complete exception line. Earlier + # blocks may describe a chained exception; only the final block counts. + last = len(lines) - 1 + if not re.match(r"^E\s+", lines[last]): + return None + if last and re.match(r"^E(?:\s|$)", lines[last - 1]): + return None + return re.sub(r"^E\s+", "", lines[last]) + + def _is_added_import_failure(text: str, added: AddedNames) -> bool: - # `from pkg import name` names a binding or a submodule the fix adds. - for name, module in _CANNOT_IMPORT.findall(text): - if (module, name) in added.names or "%s.%s" % (module, name) in added.modules: - return True - for module, name in _NO_ATTRIBUTE.findall(text): - if (module, name) in added.names or "%s.%s" % (module, name) in added.modules: - return True - # A module the fix adds is missing, or only a stale foreign .pyc is left. - # Known limit: a missing parent (`No module named 'pkg.new'` when the fix - # creates `pkg/new/mod.py` as a namespace package) is not RED, because - # the patch alone does not show whether `pkg/new/` existed before. - return any(set(modules) & added.modules for modules in _NO_MODULE.findall(text)) + kind, separator, detail = text.partition(": ") + if not separator: + return False + # Only one anchored terminal diagnostic supplies the name/module pair. + match = _CANNOT_IMPORT.fullmatch(detail) if kind == "ImportError" else None + if match is not None: + name, module = match.groups() + return (module, name) in added.names or "%s.%s" % (module, name) in added.modules + match = _NO_ATTRIBUTE.fullmatch(detail) if kind == "AttributeError" else None + if match is not None: + module, name = match.groups() + return (module, name) in added.names or "%s.%s" % (module, name) in added.modules + # A missing parent is not evidence that the exact created module is absent. + match = _NO_MODULE.fullmatch(detail) if kind in {"ImportError", "ModuleNotFoundError"} else None + return match is not None and bool(set(match.groups()) & added.modules) def red_witness(junit_xml: str, added: AddedNames) -> bool: @@ -200,6 +237,7 @@ def red_witness(junit_xml: str, added: AddedNames) -> bool: message = error.get("message", "") if message.startswith("failed on teardown"): continue - if _is_added_import_failure(message + "\n" + (error.text or ""), added): + diagnostic = _terminal_error(message, error.text or "") + if diagnostic is not None and _is_added_import_failure(diagnostic, added): return True return False diff --git a/src/code_forge/_fixval_transaction.py b/src/code_forge/_fixval_transaction.py index 4d3476f4..cc0499a2 100644 --- a/src/code_forge/_fixval_transaction.py +++ b/src/code_forge/_fixval_transaction.py @@ -572,6 +572,49 @@ def prepare(self) -> None: self._move(entry.parent, entry.leaf, entry.saved, entry.original) entry.saved_identity = self.saved_entries[entry.saved] + def python_sources(self, *, reverted: bool = False) -> dict[str, bytes | None]: + """Read complete Python images through this transaction's bound identities. + + None proves absence; an omitted key is not usable source evidence. + Oversized or non-regular images cannot establish an import witness. + """ + self._check_directory() + self._check_parents() + if reverted and not self.reverted: + raise TransactionError("reverse image is not available") + sources: dict[str, bytes | None] = {} + for entry in self.entries: + if not entry.path.endswith(".py"): + continue + if reverted: + parent, leaf, expected = entry.parent, entry.leaf, entry.reverted + if self._entry_identity(entry.path, parent) != expected: + raise TransactionError("reverse source changed: " + entry.path) + else: + parent, leaf = self.saved_fd, entry.snapshot + expected = self.saved_entries.get(leaf) if leaf is not None else None + if entry.original is not None and expected is None: + raise TransactionError("source snapshot is missing: " + entry.path) + if expected is None: + sources[entry.path] = None + continue + if not stat.S_ISREG(expected[2]) or expected[4] > 1024 * 1024: + continue + fd = os.open(leaf, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=parent) + with os.fdopen(fd, "rb") as stream: + opened = os.fstat(fd) + if not stat.S_ISREG(opened.st_mode) or (opened.st_dev, opened.st_ino) != expected[:2]: + raise TransactionError("source image changed while opening: " + entry.path) + payload = stream.read(1024 * 1024 + 1) + if len(payload) > 1024 * 1024 or hashlib.sha256(payload).hexdigest() != expected[-1]: + raise TransactionError("source image bytes changed: " + entry.path) + if _identity(parent, leaf) != expected: + raise TransactionError("source image changed while reading: " + entry.path) + sources[entry.path] = payload + self._check_directory() + self._check_parents() + return sources + def git_environment(self, image=None): """Scope local apply to its explicit working tree, without inherited routing.""" environment = {key: value for key, value in os.environ.items() if not key.startswith("GIT_")} diff --git a/src/code_forge/fixval.py b/src/code_forge/fixval.py index a0bdfb65..d7410428 100644 --- a/src/code_forge/fixval.py +++ b/src/code_forge/fixval.py @@ -1088,6 +1088,8 @@ def _reverted_run( preservation = {"recovery_parent": recovery_parent} if recovery_parent is not None else {} transaction = FixvalTransaction(cwd, non_test_patch, **preservation) transaction.prepare() + collect_imports = not require_witness and _runs_pytest(scoped_cmd) + fixed_sources = transaction.python_sources() if collect_imports else {} revert_result = transaction.reverse(patch_path) if revert_result.returncode != 0: errors.append("revert patch failed: %s" % revert_result.stderr[:200]) @@ -1103,7 +1105,11 @@ def _reverted_run( changed_files_in_order(non_test_patch), require_witness=require_witness, timeout=timeout, - added=added_names(non_test_patch), + added=added_names( + non_test_patch, + before=transaction.python_sources(reverted=True) if collect_imports else {}, + after=fixed_sources, + ), ) except (OSError, ValueError, TransactionError, subprocess.SubprocessError) as exc: errors.append(str(exc)) diff --git a/tests/test_fixval_junit_evidence_boundaries.py b/tests/test_fixval_junit_evidence_boundaries.py new file mode 100644 index 00000000..a4161f35 --- /dev/null +++ b/tests/test_fixval_junit_evidence_boundaries.py @@ -0,0 +1,450 @@ +# SPDX-License-Identifier: Apache-2.0 +"""Complete source images and real exception types bound FIXVAL's import evidence.""" + +from contextlib import contextmanager +import difflib +import os + +import pytest + +from code_forge._fixval_junit import AddedNames, added_names +from code_forge._fixval_transaction import FixvalTransaction, TransactionError +from code_forge.fixval import FixvalStatus, _reverted_run, classify_fixval_candidate +from tests.test_fixval_junit_witness import ( + CALC_HELPER, + FIXTURE, + NOTHING, + PYTEST, + _added, + _git, + _inconclusive, + _patch, + _reverted, +) + + +@pytest.mark.parametrize("quote", ['"""', "'''"]) +@pytest.mark.parametrize("inside", ["def helper(): pass", "helper = 1", "from x import helper"]) +def test_string_contents_are_not_added_bindings(quote, inside): + found = _added("value = 1\n", f"value = 1\nDOC = {quote}\n{inside}\n{quote}\n") + assert ("calc", "helper") not in found.names + assert {name for _, name in found.names} == {"DOC"} + + +def test_hunk_starting_inside_a_docstring_uses_complete_images(): + before = '"""\n' + "description\n" * 20 + '"""\n' + after = before.replace("description\n", "def helper(): pass\n", 1) + patch = "diff --git a/src/calc.py b/src/calc.py\n--- a/src/calc.py\n+++ b/src/calc.py\n@@ -2 +2 @@\n-description\n+def helper(): pass\n" + assert ( + added_names( + patch, before={"src/calc.py": before.encode()}, after={"src/calc.py": after.encode()} + ) + == NOTHING + ) + + +@pytest.mark.parametrize( + "before", + [ + "helper = 1\n" + "# outside hunk\n" * 20, + "if flag:\n helper = 1\n", + "def f():\n helper = 1\n", + "class C:\n helper = 1\n", + ], +) +def test_possible_existing_bindings_do_not_become_new(before): + found = _added(before, before + "def helper(): pass\n") + assert ("calc", "helper") not in found.names + + +@pytest.mark.parametrize( + "after", + [ + "helper: int\n", + "if flag:\n helper = 1\n", + "def helper(): pass\ndel helper\n", + "def helper(): pass\nif flag:\n del helper\n", + ], +) +def test_ambiguous_or_removed_bindings_do_not_count(after): + assert ("calc", "helper") not in _added("", after).names + + +@pytest.mark.parametrize("missing", ["both", "before", "after", "key"]) +def test_missing_complete_images_are_inconclusive(missing): + patch = _patch("src/calc.py", "pass\n", "def helper(): pass\n") + kwargs = {"before": {"src/calc.py": b"pass\n"}, "after": {"src/calc.py": b"def helper(): pass\n"}} + if missing == "both": + kwargs = {} + elif missing == "key": + kwargs["before"] = {} + else: + del kwargs[missing] + assert added_names(patch, **kwargs) == NOTHING + + +@pytest.mark.parametrize("side", ["before", "after"]) +@pytest.mark.parametrize("bad", [b"def incomplete(:\n", b"\x00", b"# coding: nonexistent\n"]) +def test_unparseable_complete_image_is_inconclusive(side, bad): + kwargs = {"before": {"src/calc.py": b"pass\n"}, "after": {"src/calc.py": b"def helper(): pass\n"}} + kwargs[side]["src/calc.py"] = bad + assert added_names(_patch("src/calc.py", "pass\n", "def helper(): pass\n"), **kwargs) == NOTHING + + +def test_python_encoding_cookie_in_complete_source_is_supported(): + before = b"# coding: latin-1\n# caf\xe9\n" + after = before + b"def helper(): pass\n" + found = added_names( + _patch("src/calc.py", "", "def helper(): pass\n"), + before={"src/calc.py": before}, + after={"src/calc.py": after}, + ) + assert ("calc", "helper") in found.names + + +_IMPORT_TEXT = "cannot import name 'helper' from 'calc'" + + +@pytest.mark.parametrize( + "message", + [ + _IMPORT_TEXT, + "x\nImportError: " + _IMPORT_TEXT, + "x\nE ImportError: " + _IMPORT_TEXT, + "ImportError: " + _IMPORT_TEXT, + "x\n\nImportError: " + _IMPORT_TEXT, + ], +) +@pytest.mark.parametrize("phase", ["collection", "setup"]) +def test_real_runtime_error_with_import_wording_is_not_red(tmp_path, message, phase): + statement = "raise RuntimeError(%r)" % message + if phase == "collection": + result = _reverted(tmp_path, statement + "\n\ndef test_it():\n pass\n") + else: + result = _reverted(tmp_path, "def test_it(dep):\n pass\n", conftest=FIXTURE % statement) + _inconclusive(result, 2 if phase == "collection" else 1) + + +@pytest.mark.parametrize("phase", ["collection", "setup"]) +def test_real_chained_import_then_runtime_error_is_not_red(tmp_path, phase): + body = "try:\n from calc import helper\nexcept ImportError as exc:\n raise RuntimeError('later error') from exc\n" + if phase == "collection": + result = _reverted(tmp_path, body + "\ndef test_it():\n pass\n") + else: + fixture = "import pytest\n@pytest.fixture\ndef dep():\n" + "".join( + " " + line + "\n" for line in body.splitlines() + ) + result = _reverted(tmp_path, "def test_it(dep):\n pass\n", conftest=fixture) + _inconclusive(result, 2 if phase == "collection" else 1) + + +@pytest.mark.parametrize("phase", ["collection", "setup"]) +@pytest.mark.parametrize( + "operation,added", + [ + ("from calc import helper", CALC_HELPER), + ("import calc; calc.helper", CALC_HELPER), + ("import missing_added_module", AddedNames(frozenset(), frozenset({"missing_added_module"}))), + ], +) +def test_real_supported_exception_types_keep_positive_evidence(tmp_path, phase, operation, added): + if phase == "collection": + result = _reverted(tmp_path, operation + "\ndef test_it():\n pass\n", added=added) + else: + result = _reverted( + tmp_path, "def test_it(dep):\n pass\n", added=added, conftest=FIXTURE % operation + ) + assert result.status is FixvalStatus.PASS, result.block_message + + +@pytest.mark.parametrize("operation", ["from calc import helper", "import calc; calc.helper"]) +def test_real_teardown_import_or_attribute_error_is_not_red(tmp_path, operation): + result = _reverted( + tmp_path, "def test_it(dep):\n pass\n", conftest=FIXTURE % ("yield\n " + operation) + ) + _inconclusive(result, 1) + + +@contextmanager +def _transaction(root, patch): + transaction = FixvalTransaction(root, patch) + transaction.prepare() + try: + yield transaction + finally: + errors = transaction.restore(False) + try: + transaction.close() + finally: + assert not errors, errors + + +def test_transaction_returns_fixed_and_reverted_owned_source_images(tmp_path): + path = tmp_path / "calc.py" + before, after = "value = 1\n", "value = 1\ndef helper(): pass\n" + path.write_text(after) + patch = _patch("calc.py", before, after) + patch_path = tmp_path / "change.patch" + patch_path.write_text(patch) + with _transaction(tmp_path, patch) as transaction: + assert transaction.python_sources() == {"calc.py": after.encode()} + assert transaction.reverse(str(patch_path)).returncode == 0 + transaction.mark_reverted() + assert transaction.python_sources(reverted=True) == {"calc.py": before.encode()} + assert transaction.python_sources() == {"calc.py": after.encode()} + assert path.read_text() == after + + +@pytest.mark.parametrize("kind", ["symlink", "oversized"]) +def test_transaction_does_not_supply_unusable_python_image(tmp_path, kind): + path = tmp_path / "calc.py" + if kind == "symlink": + target = tmp_path / "target" + target.write_text("def helper(): pass\n") + path.symlink_to(target) + patch = "diff --git a/calc.py b/calc.py\nold mode 100644\nnew mode 120000\n--- a/calc.py\n+++ b/calc.py\n@@ -1 +1 @@\n-pass\n+target\n" + else: + path.write_bytes(b"#" * (1024 * 1024 + 1)) + patch = _patch("calc.py", "pass\n", "# large\n") + with _transaction(tmp_path, patch) as transaction: + assert "calc.py" not in transaction.python_sources() + + +def test_reverse_image_cannot_be_requested_before_reverse(tmp_path): + (tmp_path / "calc.py").write_text("value = 2\n") + with _transaction(tmp_path, _patch("calc.py", "value = 1\n", "value = 2\n")) as transaction: + with pytest.raises(TransactionError, match="reverse image is not available"): + transaction.python_sources(reverted=True) + + +@pytest.mark.parametrize( + "addition,red", + [('DOC = """\ndef helper(): pass\n"""\n', False), ("def helper(): pass\n", True)], +) +def test_complete_transaction_distinguishes_string_from_binding(tmp_path, addition, red): + _git(tmp_path, "init", "-q") + (tmp_path / "src").mkdir() + (tmp_path / "tests").mkdir() + before = "value = 1\n" + after = before + addition + (tmp_path / "src/calc.py").write_text(before) + (tmp_path / "tests/test_calc.py").write_text("def test_it(): pass\n") + _git(tmp_path, "add", "--all") + _git(tmp_path, "commit", "-q", "-m", "base") + (tmp_path / "src/calc.py").write_text(after) + # Only the real function addition makes this import possible. + (tmp_path / "tests/test_calc.py").write_text("from calc import helper\ndef test_it(): pass\n") + _git(tmp_path, "add", "--all") + diff = _git(tmp_path, "diff", "--cached") + candidate = classify_fixval_candidate(["src/calc.py", "tests/test_calc.py"]) + # Exercise the revert transaction directly; run_fixval's preliminary green + # check correctly refuses this test before it reaches the RED phase. + + result = _reverted_run( + candidate, + _patch("src/calc.py", before, after), + [*PYTEST, "tests/test_calc.py"], + {**os.environ, "PYTHONPATH": str(tmp_path / "src")}, + tmp_path, + None, + ) + if red: + assert result.status is FixvalStatus.PASS, result.block_message + else: + _inconclusive(result, 2) + assert (tmp_path / "src/calc.py").read_text() == after + assert diff + + +@pytest.mark.parametrize("source", ["obj.helper = 1\n", "obj[helper] = 1\n"]) +def test_assignment_target_loads_do_not_bind_module_names(source): + assert _added("", source) == NOTHING + + +@pytest.mark.parametrize( + "source,names", + [ + ("one, *rest = items\n", {"one", "rest"}), + ("one, (two, three) = items\n", {"one", "two", "three"}), + ("from __future__ import annotations\n", {"annotations"}), + ], +) +def test_actual_assignment_and_future_import_bindings_are_retained(source, names): + assert {name for _, name in _added("", source).names} == names + + +@pytest.mark.parametrize("kind", ["created", "deleted", "renamed"]) +def test_transaction_preserves_absence_and_rename_source_images(tmp_path, kind): + if kind == "created": + patch = "diff --git a/calc.py b/calc.py\nnew file mode 100644\n--- /dev/null\n+++ b/calc.py\n@@ -0,0 +1 @@\n+value = 2\n" + (tmp_path / "calc.py").write_text("value = 2\n") + fixed, reverted = {"calc.py": b"value = 2\n"}, {"calc.py": None} + elif kind == "deleted": + patch = "diff --git a/calc.py b/calc.py\ndeleted file mode 100644\n--- a/calc.py\n+++ /dev/null\n@@ -1 +0,0 @@\n-value = 1\n" + fixed, reverted = {"calc.py": None}, {"calc.py": b"value = 1\n"} + else: + patch = "diff --git a/old.py b/calc.py\nsimilarity index 90%\nrename from old.py\nrename to calc.py\n--- a/old.py\n+++ b/calc.py\n@@ -1 +1 @@\n-value = 1\n+value = 2\n" + (tmp_path / "calc.py").write_text("value = 2\n") + fixed = {"calc.py": b"value = 2\n", "old.py": None} + reverted = {"calc.py": None, "old.py": b"value = 1\n"} + patch_path = tmp_path / "change.patch" + patch_path.write_text(patch) + with _transaction(tmp_path, patch) as transaction: + assert transaction.python_sources() == fixed + assert transaction.reverse(str(patch_path)).returncode == 0 + transaction.mark_reverted() + assert transaction.python_sources(reverted=True) == reverted + assert transaction.python_sources() == fixed + + +@pytest.mark.parametrize("replacement", ["file", "fifo"]) +def test_transaction_rejects_replaced_snapshot_without_blocking(tmp_path, monkeypatch, replacement): + (tmp_path / "calc.py").write_text("value = 2\n") + with _transaction(tmp_path, _patch("calc.py", "value = 1\n", "value = 2\n")) as transaction: + entry = next(entry for entry in transaction.entries if entry.path == "calc.py") + original = entry.snapshot + os.rename( + original, + "saved-test-original", + src_dir_fd=transaction.saved_fd, + dst_dir_fd=transaction.saved_fd, + ) + try: + if replacement == "file": + fd = os.open( + original, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600, dir_fd=transaction.saved_fd + ) + with os.fdopen(fd, "wb") as stream: + stream.write(b"def helper(): pass\n") + else: + os.mkfifo(original, dir_fd=transaction.saved_fd) + # Guard the actual open flags, failing before opening the FIFO + # if the nonblocking requirement regresses. + real_open = os.open + + def checked_open(path, flags, *args, **kwargs): + if path == original and kwargs.get("dir_fd") == transaction.saved_fd: + assert flags & os.O_NONBLOCK + return real_open(path, flags, *args, **kwargs) + + monkeypatch.setattr(os, "open", checked_open) + with pytest.raises(TransactionError): + transaction.python_sources() + finally: + if replacement == "fifo": + monkeypatch.setattr(os, "open", real_open) + os.unlink(original, dir_fd=transaction.saved_fd) + os.rename( + "saved-test-original", + original, + src_dir_fd=transaction.saved_fd, + dst_dir_fd=transaction.saved_fd, + ) + + +@pytest.mark.parametrize( + "before_source,expected_names", + [ + (b"def helper(): pass\n", frozenset()), + (b"value = 1\n", frozenset({("calc", "helper")})), + ], +) +def test_rename_with_shared_module_alias_compares_original_bindings(before_source, expected_names): + patch = _rename_patch("src/calc.py", "calc.py", before_source, b"def helper(): pass\n") + found = added_names( + patch, + before={"src/calc.py": before_source, "calc.py": None}, + after={"src/calc.py": None, "calc.py": b"def helper(): pass\n"}, + ) + assert found == AddedNames(expected_names, frozenset()) + + +@pytest.mark.parametrize("old_image", ["missing", None, b"def broken("]) +def test_rename_with_shared_module_alias_requires_usable_original_image(old_image): + before = {"calc.py": None} + if old_image != "missing": + before["src/calc.py"] = old_image + found = added_names( + _rename_patch("src/calc.py", "calc.py", b"value = 1\n", b"def helper(): pass\n"), + before=before, + after={"calc.py": b"def helper(): pass\n"}, + ) + assert found == NOTHING + + +def test_rename_to_distinct_module_alias_still_introduces_module_and_binding(): + source = b"def helper(): pass\n" + found = added_names( + _rename_patch("src/calc.py", "newcalc.py", source, source), + before={"src/calc.py": source, "newcalc.py": None}, + after={"src/calc.py": None, "newcalc.py": source}, + ) + assert found == AddedNames(frozenset({("newcalc", "helper")}), frozenset({"newcalc"})) + + +def _rename_patch(source, target, before, after): + return f"diff --git a/{source} b/{target}\nrename from {source}\nrename to {target}\n" + "".join( + difflib.unified_diff( + before.decode().splitlines(keepends=True), + after.decode().splitlines(keepends=True), + fromfile=f"a/{source}", + tofile=f"b/{target}", + ) + ) + + +@pytest.mark.parametrize( + "prior", + [ + "for helper in items:\n pass\n", + "with manager() as helper:\n pass\n", + "if (helper := value):\n pass\n", + "match value:\n case helper:\n pass\n", + "match value:\n case [*helper]:\n pass\n", + "match value:\n case {**helper}:\n pass\n", + "try:\n action()\nexcept Exception as helper:\n pass\n", + "def existing(helper):\n pass\n", + ], + ids=["for", "with", "walrus", "match-as", "match-star", "match-rest", "except", "argument"], +) +@pytest.mark.parametrize("renamed", [False, True], ids=["same-path", "shared-rename-alias"]) +def test_all_possible_prior_bindings_exclude_new_binding_claim(prior, renamed): + # This is the conservative AddedNames API contract, not a claim that all + # possible prior bindings execute or are module-visible at runtime. + after = prior + "def helper(): pass\n" + if renamed: + found = added_names( + _rename_patch("src/calc.py", "calc.py", prior.encode(), after.encode()), + before={"src/calc.py": prior.encode(), "calc.py": None}, + after={"src/calc.py": None, "calc.py": after.encode()}, + ) + else: + found = _added(prior, after) + assert not any(name == "helper" for _, name in found.names) + assert found.modules == frozenset() + + +@pytest.mark.parametrize("renamed", [False, True], ids=["same-path", "shared-rename-alias"]) +def test_prior_star_import_makes_added_binding_claim_uncertain(renamed): + prior = b"from provider import *\n" + after = prior + b"def helper(): pass\n" + if renamed: + found = added_names( + _rename_patch("src/calc.py", "calc.py", prior, after), + before={"src/calc.py": prior, "calc.py": None}, + after={"src/calc.py": None, "calc.py": after}, + ) + else: + found = _added(prior.decode(), after.decode()) + assert found == NOTHING + + +def test_prior_star_import_does_not_erase_distinct_new_module_evidence(): + prior = b"from provider import *\n" + after = prior + b"def helper(): pass\n" + found = added_names( + _rename_patch("src/calc.py", "newcalc.py", prior, after), + before={"src/calc.py": prior, "newcalc.py": None}, + after={"src/calc.py": None, "newcalc.py": after}, + ) + assert found.modules == frozenset({"newcalc"}) diff --git a/tests/test_fixval_junit_witness.py b/tests/test_fixval_junit_witness.py index 5293de74..2ddcaf49 100644 --- a/tests/test_fixval_junit_witness.py +++ b/tests/test_fixval_junit_witness.py @@ -2,6 +2,8 @@ """#28: a reverted pytest run is RED only when its JUnit report shows a failed test, or a collection or setup error importing a name the fix adds.""" +import ast +import difflib import os import subprocess import sys @@ -24,158 +26,137 @@ NOTHING = AddedNames(frozenset(), frozenset()) -def _patch(path, removed, added, new_file=False): - source = "/dev/null" if new_file else "a/" + path - header = "diff --git a/%s b/%s\n" % (path, path) - if new_file: - header += "new file mode 100644\n" - body = "".join("-%s\n" % line for line in removed) + "".join("+%s\n" % line for line in added) - hunk = "@@ -%s +1,%d @@\n" % ("0,0" if new_file else "1,%d" % len(removed), len(added)) - return header + "--- %s\n+++ b/%s\n" % (source, path) + hunk + body +def _patch(path, before, after): + return "diff --git a/%s b/%s\n" % (path, path) + "".join( + difflib.unified_diff( + before.splitlines(keepends=True), + after.splitlines(keepends=True), + fromfile="a/" + path, + tofile="b/" + path, + ) + ) + + +def _added(before, after, path="src/calc.py"): + # Fixtures are complete source images, not snippets inferred from hunks. + ast.parse(before) + ast.parse(after) + return added_names( + _patch(path, before, after), + before={path: before.encode()}, + after={path: after.encode()}, + ) @pytest.mark.parametrize( - ("line", "names"), + ("source", "names"), [ - ("def helper(x):", {"helper"}), - ("async def fetch():", {"fetch"}), - ("class Store:", {"Store"}), - ("LIMIT = 3", {"LIMIT"}), - ("LIMIT: int = 3", {"LIMIT"}), - ("from .impl import run, stop as halt", {"run", "halt"}), - ("import json", {"json"}), - (" def method(self):", set()), - (" value = 1", set()), - ("value == 1", set()), - ("else: x = 1", set()), - ("# def commented():", set()), + ("def helper(x): pass\n", {"helper"}), + ("async def fetch(): pass\n", {"fetch"}), + ("class Store: pass\n", {"Store"}), + ("LIMIT = 3\n", {"LIMIT"}), + ("LIMIT: int = 3\n", {"LIMIT"}), + ("from .impl import run, stop as halt\n", {"run", "halt"}), + ("import json\n", {"json"}), + ("def outer():\n def method(self): pass\n", {"outer"}), + ("def outer():\n value = 1\n", {"outer"}), + ("value == 1\n", set()), + ("if False:\n x = 1\n", set()), + ("# def commented():\n", set()), ], ) -def test_added_names_reads_module_level_bindings(line, names): - found = added_names(_patch("src/calc.py", ["pass"], [line])) +def test_added_names_reads_module_level_bindings(source, names): + found = _added("pass\n", source) assert {name for _module, name in found.names} == names assert found.modules == frozenset() def test_added_names_maps_paths_to_modules_and_skips_names_that_already_existed(): - patch = _patch("src/calc.py", ["def helper():", "OLD = 1"], ["def helper(x):", "NEW = 2"]) - patch += _patch("lib/pkg/__init__.py", [], ["def entry():"], new_file=True) - patch += _patch("docs/notes.md", [], ["def nothing():"]) - found = added_names(patch) - assert found.names == { - ("src.calc", "NEW"), - ("calc", "NEW"), - ("lib.pkg", "entry"), - } + before = "def helper(): pass\nOLD = 1\n" + after = "def helper(x): pass\nNEW = 2\n" + found = _added(before, after) + assert found.names == {("src.calc", "NEW"), ("calc", "NEW")} + created = "diff --git a/lib/pkg/__init__.py b/lib/pkg/__init__.py\nnew file mode 100644\n--- /dev/null\n+++ b/lib/pkg/__init__.py\n@@ -0,0 +1 @@\n+def entry(): pass\n" + found = added_names( + created, + before={"lib/pkg/__init__.py": None}, + after={"lib/pkg/__init__.py": b"def entry(): pass\n"}, + ) + assert found.names == {("lib.pkg", "entry")} assert found.modules == {"lib.pkg"} + assert _added("pass\n", "def nothing(): pass\n", "docs/notes.md") == NOTHING def test_a_path_that_is_not_an_importable_module_adds_nothing(): - # pkg/foo-bar.py cannot be imported, so no import error can name it - assert added_names(_patch("pkg/foo-bar.py", ["pass"], ["def helper():"])) == NOTHING - assert added_names(_patch("pkg/foo-bar.py", [], ["def helper():"], new_file=True)) == NOTHING - - -def _hunk(path, lines): - """A modification hunk; each line starts with "+", "-" or " ".""" - old = sum(not line.startswith("+") for line in lines) - new = sum(not line.startswith("-") for line in lines) - head = "diff --git a/%s b/%s\n--- a/%s\n+++ b/%s\n" % (path, path, path, path) - return head + "@@ -1,%d +1,%d @@\n" % (old, new) + "".join(line + "\n" for line in lines) + assert _added("pass\n", "def helper(): pass\n", "pkg/foo-bar.py") == NOTHING @pytest.mark.parametrize( - ("lines", "names"), + ("before", "after", "names"), [ - # a new multi-line import binds every name inside the parentheses - (["+from .impl import (", "+ run, # see (a)", "+ stop as halt,", "+)"], {"run", "halt"}), - # each side keeps its own state: the removed lines leave the old import open - (["-from .impl import (", "- run,", "+from .impl import (stop,", " )"], {"stop"}), - # a name added inside an import whose opening line is context - ([" from .impl import (", " run,", "+ stop,", " )"], {"stop"}), - # a name moved from one line of the import to another is not new - ([" from .impl import (", "- run, stop", "+ run,", "+ stop,", " )"], set()), - (["+from .impl import (run,", "+ stop)", "+def after():"], {"run", "stop", "after"}), - # after the closing line, indented lines bind nothing again - ([" from .impl import (", " run,", " )", "+ value = 1"], set()), + ("", "from .impl import (\n run, # see (a)\n stop as halt,\n)\n", {"run", "halt"}), + ("from .impl import (\n run,\n)\n", "from .impl import (stop,\n)\n", {"stop"}), + ("from .impl import (\n run,\n)\n", "from .impl import (\n run,\n stop,\n)\n", {"stop"}), + ("from .impl import (\n run, stop\n)\n", "from .impl import (\n run,\n stop,\n)\n", set()), + ("", "from .impl import (run,\n stop)\ndef after(): pass\n", {"run", "stop", "after"}), + ("from .impl import (run,)\n", "from .impl import (run,)\nif False:\n value = 1\n", set()), ], ) -def test_added_names_reads_multi_line_imports(lines, names): - found = added_names(_hunk("src/calc.py", lines)) - assert {name for _module, name in found.names} == names +def test_added_names_reads_multi_line_imports(before, after, names): + assert {name for _module, name in _added(before, after).names} == names def test_a_comment_on_the_opening_line_still_opens_the_import(): - lines = ["+from .impl import ( # noqa: F401", "+ helper,", "+)"] - found = added_names(_hunk("src/calc.py", lines)) + found = _added("", "from .impl import ( # noqa: F401\n helper,\n)\n") assert found.names == {("src.calc", "helper"), ("calc", "helper")} - # the reverted run's import error for that name is RED assert red_witness(_report(_error("collection failure", IMPORT_HELPER)), found) is True def test_a_function_level_parenthesised_import_binds_nothing(): - lines = [ - "+def load():", - "+ from .impl import (", - "+ helper,", - "+ )", - "+ return helper", - ] - found = added_names(_hunk("src/calc.py", lines)) + found = _added( + "", "def load():\n from .impl import (\n helper,\n )\n return helper\n" + ) assert {name for _module, name in found.names} == {"load"} @pytest.mark.parametrize( - ("lines", "names"), + ("before", "after", "names"), [ - (["+from pkg.o import a, \\", "+ newname"], {"a", "newname"}), - ( - ["+from pkg.o import \\", "+ a, \\", "+ newname", "+VALUE = 1"], - {"a", "newname", "VALUE"}, - ), - (["+import json, \\", "+ csv as table"], {"json", "table"}), - (["+from pkg.o import a,\\", "+ b,\\", "+ newname # noqa"], {"a", "b", "newname"}), - ([" from pkg.o import a, \\", "+ newname"], {"newname"}), - # only an import line continues; an assignment's backslash does not - (["+total = 1 + \\", "+ newname"], {"total"}), - (["+def f():", "+ from pkg.o import a, \\", "+ b"], {"f"}), - # a plain import does not continue onto the next line - (["+import json", "+VALUE = 1"], {"json", "VALUE"}), - # a parenthesised import is closed by its `)`, even with a `\\` inside - (["+from pkg.o import (a, \\", "+ newname)", "+VALUE = 1"], {"a", "newname", "VALUE"}), - # a backslash at the end of a comment continues nothing - (["+import os # see \\", "+def helper():"], {"os", "helper"}), - (["+from pkg.o import a, \\", "+ b # c \\", "+VALUE = 1"], {"a", "b", "VALUE"}), + ("", "from pkg.o import a, \\\n newname\n", {"a", "newname"}), + ("", "from pkg.o import \\\n a, \\\n newname\nVALUE = 1\n", {"a", "newname", "VALUE"}), + ("", "import json, \\\n csv as table\n", {"json", "table"}), + ("", "from pkg.o import a,\\\n b,\\\n newname # noqa\n", {"a", "b", "newname"}), + ("from pkg.o import a\n", "from pkg.o import a, \\\n newname\n", {"newname"}), + ("", "total = 1 + \\\n newname\n", {"total"}), + ("", "def f():\n from pkg.o import a, \\\n b\n", {"f"}), + ("", "import json\nVALUE = 1\n", {"json", "VALUE"}), + ("", "from pkg.o import (a, \\\n newname)\nVALUE = 1\n", {"a", "newname", "VALUE"}), + ("", "import os # see \\\ndef helper(): pass\n", {"os", "helper"}), + ("", "from pkg.o import a, \\\n b # c \\\nVALUE = 1\n", {"a", "b", "VALUE"}), ], ) -def test_added_names_reads_backslash_continued_imports(lines, names): - found = added_names(_hunk("src/calc.py", lines)) - assert {name for _module, name in found.names} == names +def test_added_names_reads_backslash_continued_imports(before, after, names): + assert {name for _module, name in _added(before, after).names} == names def test_a_backslash_continued_import_of_an_added_name_is_red(): - found = added_names(_hunk("src/calc.py", [" from pkg.o import a, \\", "+ helper"])) + found = _added("from pkg.o import a\n", "from pkg.o import a, \\\n helper\n") assert red_witness(_report(_error("collection failure", IMPORT_HELPER)), found) is True def test_a_definition_after_an_import_whose_comment_ends_in_a_backslash_is_red(): - lines = ["+import os # see \\", "+def helper():", "+ return os.sep"] - found = added_names(_hunk("src/calc.py", lines)) + found = _added("", "import os # see \\\ndef helper():\n return os.sep\n") assert red_witness(_report(_error("collection failure", IMPORT_HELPER)), found) is True def test_an_import_left_open_at_the_end_of_a_hunk_does_not_leak_into_the_next(): - # The first hunk stops inside `from .impl import (`; the second hunk is - # elsewhere in the file, inside a call, where `stop,` binds nothing. - patch = ( - "diff --git a/src/calc.py b/src/calc.py\n--- a/src/calc.py\n+++ b/src/calc.py\n" - "@@ -1,1 +1,2 @@\n from .impl import (\n+ run,\n" - "@@ -20,2 +21,3 @@\n call(\n+ stop,\n )\n" + before = "from .impl import (\n old,\n)\n" + "# padding\n" * 16 + "call(\n old,\n)\n" + after = before.replace("from .impl import (\n", "from .impl import (\n run,\n").replace( + "call(\n", "call(\n stop,\n" ) - found = added_names(patch) + found = _added(before, after) assert {name for _module, name in found.names} == {"run"} - # the same with a module-level binding at column 0 in the second hunk - found = added_names(patch.replace("+ stop,\n )\n", "+LIMIT = 3\n )\n")) + found = _added(before, after + "LIMIT = 3\n") assert {name for _module, name in found.names} == {"run", "LIMIT"} @@ -193,25 +174,28 @@ def test_an_import_left_open_at_the_end_of_a_hunk_does_not_leak_into_the_next(): def test_a_renamed_or_copied_file_adds_its_module(): - assert added_names(RENAME).modules == {"src.pkg.newmod", "pkg.newmod"} - copy = RENAME.replace("rename ", "copy ") - assert added_names(copy).modules == {"src.pkg.newmod", "pkg.newmod"} + for patch in (RENAME, RENAME.replace("rename ", "copy ")): + found = added_names( + patch, before={"src/pkg/newmod.py": None}, after={"src/pkg/newmod.py": b"X = 2\n"} + ) + assert found.modules == {"src.pkg.newmod", "pkg.newmod"} def test_a_rename_to_a_name_the_module_already_had_adds_no_module(): - # src/calc.py was importable as `calc` before the fix moved it to calc.py rename = RENAME.replace("src/old.py", "src/calc.py").replace("src/pkg/newmod.py", "calc.py") - assert added_names(rename).modules == frozenset() + found = added_names(rename, before={"calc.py": None}, after={"calc.py": b"X = 2\n"}) + assert found.modules == frozenset() def test_a_modified_file_adds_no_module(): - assert added_names(_hunk("src/calc.py", ["-X = 1", "+X = 2"])).modules == frozenset() + assert _added("X = 1\n", "X = 2\n").modules == frozenset() plain = "--- a/src/calc.py\n+++ b/src/calc.py\n@@ -1 +1 @@\n-X = 1\n+X = 2\n" - assert added_names(plain).modules == frozenset() + found = added_names(plain, before={"src/calc.py": b"X = 1\n"}, after={"src/calc.py": b"X = 2\n"}) + assert found.modules == frozenset() def test_unreadable_patch_adds_nothing(): - assert added_names("@@ not a diff\n+def helper():\n") == NOTHING + assert added_names("@@ not a diff\n+def helper():\n", before={}, after={}) == NOTHING def _report(*cases): @@ -252,7 +236,11 @@ def _error(message, text=""): NOTHING, True, ), - (_report(_error("failed on setup with "%s"" % IMPORT_HELPER)), CALC_HELPER, True), + ( + _report(_error("failed on setup with "%s"" % IMPORT_HELPER.removeprefix("E "))), + CALC_HELPER, + True, + ), # `from pkg import newmod`, where the fix creates the module pkg/newmod.py (_report(_error("collection failure", IMPORT_NEWMOD)), NEW_MODULE, True), (