diff --git a/src/code_forge/_fixval_junit.py b/src/code_forge/_fixval_junit.py new file mode 100644 index 00000000..adb83751 --- /dev/null +++ b/src/code_forge/_fixval_junit.py @@ -0,0 +1,243 @@ +# 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 ast +from collections.abc import Mapping +import re +import xml.etree.ElementTree as ElementTree +from dataclasses import dataclass + +import unidiff + +from .diff import iter_diff_sections, patched_file_path + +_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.]+)'(?:; '[\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 +# that succeeded. #28 counts neither as a RED witness. +_UNEXPECTED_PASS = ("[XPASS(strict)]", "Failed: Unexpected success") + + +@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)} + # 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 _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 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: + 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 "") + 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: + 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: + 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: + """Whether the report shows the reverted code failing a test. + + 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: + # 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 + for case in root.iter("testcase"): + 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"): + continue + 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 aadc88ee..d7410428 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 @@ -34,6 +35,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 +329,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 +504,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 +589,79 @@ 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_dir = report = None + if not require_witness and _runs_pytest(scoped_cmd): + report_dir = tempfile.mkdtemp(prefix="fixval-junit-") + report = os.path.join(report_dir, "junit.xml") 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_dir is not None: + 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 _with_junitxml(scoped_cmd, report) + try: + result = subprocess.run( + cmd, env=run_env, capture_output=True, text=True, @@ -619,6 +684,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=[]) @@ -1017,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]) @@ -1032,6 +1105,11 @@ def _reverted_run( changed_files_in_order(non_test_patch), require_witness=require_witness, timeout=timeout, + 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.py b/tests/test_fixval.py index 2addb300..926a58b3 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 @@ -675,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 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_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 new file mode 100644 index 00000000..2ddcaf49 --- /dev/null +++ b/tests/test_fixval_junit_witness.py @@ -0,0 +1,573 @@ +# 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 ast +import difflib +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, 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( + ("source", "names"), + [ + ("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(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(): + 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(): + assert _added("pass\n", "def helper(): pass\n", "pkg/foo-bar.py") == NOTHING + + +@pytest.mark.parametrize( + ("before", "after", "names"), + [ + ("", "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(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(): + found = _added("", "from .impl import ( # noqa: F401\n helper,\n)\n") + assert found.names == {("src.calc", "helper"), ("calc", "helper")} + assert red_witness(_report(_error("collection failure", IMPORT_HELPER)), found) is True + + +def test_a_function_level_parenthesised_import_binds_nothing(): + 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( + ("before", "after", "names"), + [ + ("", "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(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("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(): + 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(): + 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(before, after) + assert {name for _module, name in found.names} == {"run"} + found = _added(before, after + "LIMIT = 3\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% +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(): + 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(): + rename = RENAME.replace("src/old.py", "src/calc.py").replace("src/pkg/newmod.py", "calc.py") + 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("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" + 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", before={}, after={}) == 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)" +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( + ("report", "added", "red"), + [ + (_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, + ), + # 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.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), + ( + _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, + ), + # 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), + ( + _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'") + ), + 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 + + +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("