diff --git a/README.md b/README.md index 488bfd2..f9770d6 100644 --- a/README.md +++ b/README.md @@ -384,13 +384,14 @@ Keep `.env` local. It is already ignored by the repository template and should n - Applies the gold patch. - Runs the configured test command in Docker after the patch. - Runs the post-patch command a second time as a deterministic rerun check. +- For pytest commands, collects a per-test JUnit report from the before/after runs and derives the targeted test sets `fail_to_pass` (failed before, pass after) and `pass_to_pass` (pass both) — the SWE-bench-style scoping spec. - Writes `.forge/tasks//verification.json`. ### `forge package ` Creates `taskpacks//` with: -- `task.json` +- `task.json` (task metadata plus an `eval` block carrying the `fail_to_pass`/`pass_to_pass` test sets) - `prompt.md` - `gold.patch` - `reward.py` @@ -412,10 +413,10 @@ Runs the taskpack's standalone reward script and returns JSON: } ``` -For v1, the reward is binary: +The reward is binary: -- `1.0` if the configured test command exits with code `0` -- `0.0` otherwise +- If the task has a scoping spec (`eval.fail_to_pass` from a pytest task), the score is `1.0` only when every targeted test passes — all `fail_to_pass` tests pass **and** no `pass_to_pass` test regresses — so unrelated failing or flaky tests in the repo don't mask the signal. The reward re-collects a per-test JUnit report and scores against exactly those node ids. +- Otherwise (non-pytest tasks, or no spec) it falls back to whole-suite scoring: `1.0` if the configured test command exits with code `0`, else `0.0`. `error` is `null` on success. On a plain test failure it carries `"Test command exited with code "`; on an infrastructure failure (missing artifacts, Docker build failure, or timeout) it carries the corresponding message. @@ -493,6 +494,7 @@ The configured test command runs inside the container with a timeout, with no ne ## Current Limitations - Python-only Dockerfile generation. +- Test-scoped reward (`fail_to_pass`/`pass_to_pass`) is derived for pytest commands only; other tasks fall back to whole-suite exit-code scoring. The scoping is computed from the combined gold patch (test files are not yet separated from the fix). - No hidden tests or leakage detection. - No agent rollout orchestration. - No model training, GRPO, LoRA, or fine-tuning pipeline. diff --git a/forge/docker_runner.py b/forge/docker_runner.py index 94f2737..2c948a5 100644 --- a/forge/docker_runner.py +++ b/forge/docker_runner.py @@ -97,8 +97,14 @@ def run_tests_in_docker( timeout_seconds: int, image_tag_prefix: str, dockerfile_path: Path | None = None, + exec_command: str | None = None, ) -> DockerRunResult: - """Build a Docker image from repo_path and run the configured test command.""" + """Build a Docker image from repo_path and run the configured test command. + + ``exec_command`` overrides the shell command actually run in the container + (e.g. a pytest command wrapped to emit a JUnit report); the recorded + ``command`` stays ``test_command`` for display. + """ started_at = time.monotonic() if shutil.which("docker") is None: @@ -171,7 +177,7 @@ def run_tests_in_docker( duration_seconds=time.monotonic() - started_at, ) - run_command = ["docker", "run", "--rm", "--name", container_name, *CONTAINER_RUN_FLAGS, image_tag, "sh", "-lc", test_command] + run_command = ["docker", "run", "--rm", "--name", container_name, *CONTAINER_RUN_FLAGS, image_tag, "sh", "-lc", exec_command or test_command] try: run = subprocess.run( run_command, diff --git a/forge/task_builder.py b/forge/task_builder.py index b95fcb1..a5707f4 100644 --- a/forge/task_builder.py +++ b/forge/task_builder.py @@ -1,5 +1,6 @@ from __future__ import annotations +import json import shutil from pathlib import Path from typing import Callable @@ -10,6 +11,7 @@ from forge.github_pr import fetch_pr_diff, fetch_pull_request from forge.patch_utils import apply_patch_file, check_patch from forge.task_schema import CommandRun, TaskMetadata, VerificationResult, detect_test_infrastructure_failure +from forge.test_report import derive_test_deltas, extract_report, parse_junit_xml, wrap_pytest_command FORGE_DIR = Path(".forge") @@ -145,6 +147,12 @@ def verify_task(task_id: str, *, root: Path | None = None, log: LogFn = _noop_lo before_patch: CommandRun | None = None after_patch: CommandRun | None = None deterministic_rerun: CommandRun | None = None + before_results: dict[str, str] = {} + after_results: dict[str, str] = {} + + # Wrap pytest commands to emit a per-test JUnit report; None for non-pytest + # commands, which fall back to whole-suite exit-code scoring. + exec_command = wrap_pytest_command(metadata.test_command) try: ensure_commit_available(repo_dir, metadata.base_commit) @@ -174,8 +182,10 @@ def verify_task(task_id: str, *, root: Path | None = None, log: LogFn = _noop_lo metadata.test_command, timeout_seconds=metadata.timeout_seconds, image_tag_prefix=f"forge-{task_id}-before", + exec_command=exec_command, ) before_patch = _command_run("before_patch", before) + before_results = parse_junit_xml(extract_report(before.stdout)) if base_commit_found and patch_applies: log("Applying the gold patch") @@ -189,8 +199,10 @@ def verify_task(task_id: str, *, root: Path | None = None, log: LogFn = _noop_lo metadata.test_command, timeout_seconds=metadata.timeout_seconds, image_tag_prefix=f"forge-{task_id}-after", + exec_command=exec_command, ) after_patch = _command_run("after_patch", after) + after_results = parse_junit_xml(extract_report(after.stdout)) # Attests post-patch idempotence (same patched tree, freshly rebuilt # image, run twice). It does NOT re-check the pre-patch baseline, so a @@ -201,6 +213,7 @@ def verify_task(task_id: str, *, root: Path | None = None, log: LogFn = _noop_lo metadata.test_command, timeout_seconds=metadata.timeout_seconds, image_tag_prefix=f"forge-{task_id}-rerun", + exec_command=exec_command, ) deterministic_rerun = _command_run("deterministic_rerun", rerun) finally: @@ -229,6 +242,13 @@ def verify_task(task_id: str, *, root: Path | None = None, log: LogFn = _noop_lo tests_pass_after_patch = after_patch is not None and test_environment_success and after_patch.docker_build_success and after_patch.exit_code == 0 deterministic_rerun_success = deterministic_rerun is not None and test_environment_success and deterministic_rerun.docker_build_success and deterministic_rerun.exit_code == 0 + # Derive the targeted test sets only when both runs produced a parseable report; + # otherwise leave them empty so reward falls back to whole-suite scoring. + if before_results and after_results and test_environment_success: + fail_to_pass, pass_to_pass = derive_test_deltas(before_results, after_results) + else: + fail_to_pass, pass_to_pass = [], [] + verification = VerificationResult( task_id=task_id, base_commit_found=base_commit_found, @@ -238,6 +258,8 @@ def verify_task(task_id: str, *, root: Path | None = None, log: LogFn = _noop_lo docker_build_success=docker_build_success, test_environment_success=test_environment_success, deterministic_rerun_success=deterministic_rerun_success, + fail_to_pass=fail_to_pass, + pass_to_pass=pass_to_pass, before_patch=before_patch, after_patch=after_patch, deterministic_rerun=deterministic_rerun, @@ -292,8 +314,17 @@ def _reward_script() -> str: import tempfile import time import uuid +import xml.etree.ElementTree as ET from pathlib import Path +# Sentinels for embedding a JUnit report in container stdout (kept in sync with +# forge.test_report). Targeted scoring uses the report; otherwise we fall back to +# the test command's exit code. +REPORT_PATH = "/tmp/forge-report.xml" +REPORT_START = "<<>>" +REPORT_END = "<<>>" +_PYTEST_RE = re.compile(r"(^|\s)(pytest|py\.test)(\s|$)|python[0-9.]*\s+-m\s+pytest") + def emit(score: float, tests_passed: bool, error: str | None) -> int: print(json.dumps({"score": score, "tests_passed": tests_passed, "error": error})) @@ -305,6 +336,55 @@ def slug(value: str) -> str: return normalized or "task" +def is_pytest_command(test_command: str) -> bool: + return bool(_PYTEST_RE.search(test_command or "")) + + +def wrap_pytest_command(test_command: str) -> str: + inner = f"{test_command} --junitxml={REPORT_PATH}" + return ( + f"{inner}; __forge_code=$?; " + f"echo '{REPORT_START}'; cat {REPORT_PATH} 2>/dev/null; echo '{REPORT_END}'; " + f"exit $__forge_code" + ) + + +def extract_report(stdout: str): + if not stdout or REPORT_START not in stdout or REPORT_END not in stdout: + return None + start = stdout.index(REPORT_START) + len(REPORT_START) + end = stdout.index(REPORT_END, start) + xml = stdout[start:end].strip() + return xml or None + + +def parse_junit_xml(xml_text): + results = {} + if not xml_text: + return results + try: + root = ET.fromstring(xml_text) + except ET.ParseError: + return results + for case in root.iter("testcase"): + classname = case.get("classname") or "" + name = case.get("name") or "" + nodeid = f"{classname}::{name}" if classname else name + status = "passed" + for child in case: + tag = child.tag.lower() + if tag in ("failure", "error", "skipped"): + status = "failed" if tag == "failure" else tag + break + results[nodeid] = status + return results + + +def score_targeted(results, fail_to_pass, pass_to_pass): + ok = {nodeid for nodeid, status in results.items() if status == "passed"} + return all(t in ok for t in fail_to_pass) and all(t in ok for t in pass_to_pass) + + def stage_repo_with_patch(repo_dir: Path, patch_file: Path) -> tuple[Path | None, str | None]: """Copy repo_dir into a temp dir and apply the candidate patch. Returns (work_dir, error).""" @@ -367,6 +447,12 @@ def main() -> int: image_tag = f"forge-reward-{slug(task['id'])}-{uuid.uuid4().hex[:12]}" container_name = f"{image_tag}-run" + eval_spec = task.get("eval") or {} + fail_to_pass = list(eval_spec.get("fail_to_pass") or []) + pass_to_pass = list(eval_spec.get("pass_to_pass") or []) + scoped = bool(fail_to_pass) and is_pytest_command(test_command) + exec_command = wrap_pytest_command(test_command) if scoped else test_command + started_at = time.monotonic() try: build = subprocess.run( @@ -390,7 +476,7 @@ def main() -> int: docker_bin, "run", "--rm", "--name", container_name, "--network", "none", "--memory", "4g", "--cpus", "2", "--pids-limit", "512", "--security-opt", "no-new-privileges", - image_tag, "sh", "-lc", test_command, + image_tag, "sh", "-lc", exec_command, ], text=True, encoding="utf-8", @@ -398,6 +484,15 @@ def main() -> int: capture_output=True, timeout=timeout_seconds, ) + if scoped: + results = parse_junit_xml(extract_report(run.stdout)) + if results: + if score_targeted(results, fail_to_pass, pass_to_pass): + return emit(1.0, True, None) + missing = [t for t in fail_to_pass if results.get(t) != "passed"] + regressed = [t for t in pass_to_pass if results.get(t) != "passed"] + return emit(0.0, False, f"Targeted tests not satisfied (fail_to_pass missing: {len(missing)}, pass_to_pass regressed: {len(regressed)})") + # No parseable report (e.g. collection error): fall back to exit code. if run.returncode == 0: return emit(1.0, True, None) return emit(0.0, False, f"Test command exited with code {run.returncode}") @@ -417,6 +512,19 @@ def main() -> int: ''' +def _eval_spec(verification_file: Path) -> dict[str, list[str]]: + """Read the targeted test sets from verification.json, robust to partial files.""" + + try: + data = json.loads(verification_file.read_text(encoding="utf-8")) + except (OSError, ValueError): + data = {} + return { + "fail_to_pass": list(data.get("fail_to_pass") or []), + "pass_to_pass": list(data.get("pass_to_pass") or []), + } + + def package_task(task_id: str, *, root: Path | None = None, log: LogFn = _noop_log) -> Path: root = root or Path.cwd() metadata = load_metadata(root, task_id) @@ -437,7 +545,11 @@ def package_task(task_id: str, *, root: Path | None = None, log: LogFn = _noop_l log(f"Checking out base commit {metadata.base_commit} for package snapshot") checkout_clean(repo_dir, metadata.base_commit) - (package_dir / "task.json").write_text(metadata.model_dump_json(indent=2), encoding="utf-8") + # The packed task.json is a superset of the metadata: it also carries the eval + # spec (targeted test sets) so the standalone reward.py can score without forge. + task_payload = json.loads(metadata.model_dump_json()) + task_payload["eval"] = _eval_spec(verification_file) + (package_dir / "task.json").write_text(json.dumps(task_payload, indent=2), encoding="utf-8") (package_dir / "prompt.md").write_text(_prompt(metadata), encoding="utf-8") (package_dir / "Dockerfile").write_text(generate_python_dockerfile(), encoding="utf-8") (package_dir / "reward.py").write_text(_reward_script(), encoding="utf-8") diff --git a/forge/task_schema.py b/forge/task_schema.py index f7fe717..a753953 100644 --- a/forge/task_schema.py +++ b/forge/task_schema.py @@ -93,6 +93,8 @@ class VerificationResult(BaseModel): docker_build_success: bool = False test_environment_success: bool = True deterministic_rerun_success: bool = False + fail_to_pass: list[str] = Field(default_factory=list) + pass_to_pass: list[str] = Field(default_factory=list) before_patch: CommandRun | None = None after_patch: CommandRun | None = None deterministic_rerun: CommandRun | None = None diff --git a/forge/test_report.py b/forge/test_report.py new file mode 100644 index 0000000..90e4203 --- /dev/null +++ b/forge/test_report.py @@ -0,0 +1,111 @@ +"""Per-test result capture and scoping for pytest-based tasks. + +The verifier runs the configured test command before and after the gold patch. +With these helpers it can additionally collect a per-test JUnit report from each +run and derive the SWE-bench-style sets: + +- ``fail_to_pass``: tests that did not pass before the patch but pass after it — + the tests the fix is responsible for. +- ``pass_to_pass``: tests that pass both before and after — regression guards. + +A candidate fix is then scored against exactly those tests instead of the whole +suite's exit code, so unrelated failing/flaky tests in the repo don't mask the +signal. Everything here is pure and stdlib-only so the same logic can be inlined +into the standalone ``reward.py`` template. +""" + +from __future__ import annotations + +import re +import xml.etree.ElementTree as ET + +# A pytest run is wrapped so it writes a JUnit report and echoes it to stdout +# between these sentinels, preserving the original exit code. stdout is already +# captured by the Docker runner, so no volume mount or `docker cp` is needed. +REPORT_PATH = "/tmp/forge-report.xml" +REPORT_START = "<<>>" +REPORT_END = "<<>>" + +_PYTEST_RE = re.compile(r"(^|\s)(pytest|py\.test)(\s|$)|python[0-9.]*\s+-m\s+pytest") + + +def is_pytest_command(test_command: str) -> bool: + return bool(_PYTEST_RE.search(test_command or "")) + + +def wrap_pytest_command(test_command: str) -> str | None: + """Wrap a pytest command to emit a JUnit report between sentinels on stdout. + + Returns None for non-pytest commands (the caller should run them unwrapped and + fall back to whole-suite scoring). The original exit code is preserved so the + existing fail-before/pass-after signals keep working. + """ + + if not is_pytest_command(test_command): + return None + inner = f"{test_command} --junitxml={REPORT_PATH}" + return ( + f"{inner}; __forge_code=$?; " + f"echo '{REPORT_START}'; cat {REPORT_PATH} 2>/dev/null; echo '{REPORT_END}'; " + f"exit $__forge_code" + ) + + +def extract_report(stdout: str) -> str | None: + """Pull the JUnit XML back out from between the sentinels in captured stdout.""" + + if not stdout or REPORT_START not in stdout or REPORT_END not in stdout: + return None + start = stdout.index(REPORT_START) + len(REPORT_START) + end = stdout.index(REPORT_END, start) + xml = stdout[start:end].strip() + return xml or None + + +def parse_junit_xml(xml_text: str | None) -> dict[str, str]: + """Map ``classname::name`` -> status in {passed, failed, error, skipped}. + + The key is a stable identifier consistent across runs (we control the report + format), not necessarily a runnable pytest node id. + """ + + results: dict[str, str] = {} + if not xml_text: + return results + try: + root = ET.fromstring(xml_text) + except ET.ParseError: + return results + for case in root.iter("testcase"): + classname = case.get("classname") or "" + name = case.get("name") or "" + nodeid = f"{classname}::{name}" if classname else name + status = "passed" + for child in case: + tag = child.tag.lower() + if tag in ("failure", "error", "skipped"): + status = "failed" if tag == "failure" else tag + break + results[nodeid] = status + return results + + +def passed_tests(results: dict[str, str]) -> set[str]: + return {nodeid for nodeid, status in results.items() if status == "passed"} + + +def derive_test_deltas(before: dict[str, str], after: dict[str, str]) -> tuple[list[str], list[str]]: + """Return (fail_to_pass, pass_to_pass) from before/after per-test result maps.""" + + before_pass = passed_tests(before) + after_pass = passed_tests(after) + fail_to_pass = sorted(after_pass - before_pass) + pass_to_pass = sorted(after_pass & before_pass) + return fail_to_pass, pass_to_pass + + +def score_results(results: dict[str, str], fail_to_pass: list[str], pass_to_pass: list[str]) -> bool: + """True iff every targeted test (fail_to_pass + pass_to_pass) passed.""" + + ok = passed_tests(results) + return all(t in ok for t in fail_to_pass) and all(t in ok for t in pass_to_pass) diff --git a/tests/test_reward_rollout.py b/tests/test_reward_rollout.py index ac4db38..1365f73 100644 --- a/tests/test_reward_rollout.py +++ b/tests/test_reward_rollout.py @@ -91,6 +91,35 @@ def test_generated_reward_script_is_valid_and_supports_patch() -> None: assert "stage_repo_with_patch" in src +def test_reward_template_declares_scoped_scoring() -> None: + src = _reward_script() + assert "score_targeted" in src + assert "FORGE_JUNIT_START" in src + assert 'task.get("eval")' in src + + +def test_reward_template_scoping_helpers_match_forge() -> None: + from forge import test_report as tr + + ns: dict = {} + exec(_reward_script(), ns) # defines helpers; main() only runs under __main__ + + xml = '' \ + '' \ + '' \ + '' + + assert ns["parse_junit_xml"](xml) == tr.parse_junit_xml(xml) + assert ns["wrap_pytest_command"]("pytest -q") == tr.wrap_pytest_command("pytest -q") + + stdout = f'{ns["REPORT_START"]}{xml}{ns["REPORT_END"]}' + assert ns["extract_report"](stdout) == tr.extract_report(stdout) + + # targeted scoring: pass only when every fail_to_pass + pass_to_pass passed + assert ns["score_targeted"]({"pkg::a": "passed", "pkg::b": "passed"}, ["pkg::a"], ["pkg::b"]) is True + assert ns["score_targeted"]({"pkg::a": "passed", "pkg::b": "failed"}, ["pkg::a"], ["pkg::b"]) is False + + @requires_git def test_good_candidate_patch_applies_then_reaches_docker(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: package_dir = _build_taskpack(tmp_path, monkeypatch) diff --git a/tests/test_task_builder.py b/tests/test_task_builder.py index d419012..c7c758d 100644 --- a/tests/test_task_builder.py +++ b/tests/test_task_builder.py @@ -1,5 +1,6 @@ from __future__ import annotations +import json from pathlib import Path import pytest @@ -169,6 +170,28 @@ def test_package_task_strips_upstream_dockerignore(tmp_path: Path, monkeypatch: assert not (package_dir / "repo" / ".dockerignore").exists() +def test_package_task_writes_eval_spec_into_task_json(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + metadata = _metadata() + metadata.write_json(metadata_path(tmp_path, metadata.id)) + gold_patch_path(tmp_path, metadata.id).write_bytes(b"diff --git a b\n") + verification_path(tmp_path, metadata.id).write_text( + json.dumps({"task_id": metadata.id, "fail_to_pass": ["pkg::test_fix"], "pass_to_pass": ["pkg::test_keep"]}), + encoding="utf-8", + ) + repo_dir = tmp_path / ".forge" / "repos" / metadata.id + repo_dir.mkdir(parents=True) + (repo_dir / "m.py").write_text("value = 1\n", encoding="utf-8") + monkeypatch.setattr(task_builder, "checkout_clean", lambda *args, **kwargs: None) + + package_dir = package_task(metadata.id, root=tmp_path) + + task_json = json.loads((package_dir / "task.json").read_text(encoding="utf-8")) + assert task_json["eval"] == {"fail_to_pass": ["pkg::test_fix"], "pass_to_pass": ["pkg::test_keep"]} + # still a superset of the task metadata + assert task_json["test_command"] == metadata.test_command + assert task_json["id"] == metadata.id + + def test_package_task_overwrites_existing_taskpack(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: metadata = _metadata() metadata.write_json(metadata_path(tmp_path, metadata.id)) diff --git a/tests/test_test_report.py b/tests/test_test_report.py new file mode 100644 index 0000000..e6e7a0a --- /dev/null +++ b/tests/test_test_report.py @@ -0,0 +1,94 @@ +from __future__ import annotations + +from forge.test_report import ( + REPORT_END, + REPORT_PATH, + REPORT_START, + derive_test_deltas, + extract_report, + is_pytest_command, + parse_junit_xml, + score_results, + wrap_pytest_command, +) + +SAMPLE_XML = f""" + + + trace + trace + +""" + + +def test_is_pytest_command() -> None: + assert is_pytest_command("pytest") + assert is_pytest_command("pytest -q tests/") + assert is_pytest_command("python -m pytest") + assert is_pytest_command("python3 -m pytest -k foo") + assert not is_pytest_command("tox") + assert not is_pytest_command("make test") + assert not is_pytest_command("") + + +def test_wrap_pytest_command_roundtrips_through_extract() -> None: + wrapped = wrap_pytest_command("pytest -q") + assert wrapped is not None + assert REPORT_PATH in wrapped + assert "--junitxml" in wrapped + assert "exit $__forge_code" in wrapped + + # Simulate the container stdout: test output, then the sentinel-wrapped report. + stdout = f"collected 4 items\n.F\n{REPORT_START}\n{SAMPLE_XML}\n{REPORT_END}\n" + assert extract_report(stdout).strip().startswith(" None: + assert wrap_pytest_command("make test") is None + + +def test_extract_report_handles_missing_sentinels() -> None: + assert extract_report("no markers here") is None + assert extract_report("") is None + + +def test_parse_junit_xml_classifies_each_status() -> None: + results = parse_junit_xml(SAMPLE_XML) + + assert results == { + "tests.test_m::test_pass": "passed", + "tests.test_m::test_fail": "failed", + "tests.test_m::test_err": "error", + "tests.test_m::test_skip": "skipped", + } + + +def test_parse_junit_xml_is_lenient_on_garbage() -> None: + assert parse_junit_xml("not xml") == {} + assert parse_junit_xml(None) == {} + + +def test_derive_test_deltas() -> None: + before = {"a": "failed", "b": "passed", "c": "failed"} + after = {"a": "passed", "b": "passed", "c": "failed", "d": "passed"} + + fail_to_pass, pass_to_pass = derive_test_deltas(before, after) + + assert fail_to_pass == ["a", "d"] # a: fail->pass, d: new and passing + assert pass_to_pass == ["b"] # passed both runs + assert "c" not in fail_to_pass and "c" not in pass_to_pass # still failing + + +def test_score_results() -> None: + fail_to_pass = ["a", "d"] + pass_to_pass = ["b"] + + assert score_results({"a": "passed", "b": "passed", "d": "passed"}, fail_to_pass, pass_to_pass) + # one fail_to_pass still failing -> not scored + assert not score_results({"a": "passed", "b": "passed", "d": "failed"}, fail_to_pass, pass_to_pass) + # a regression in pass_to_pass -> not scored + assert not score_results({"a": "passed", "b": "failed", "d": "passed"}, fail_to_pass, pass_to_pass) + + +def test_sentinels_are_distinct() -> None: + assert REPORT_START != REPORT_END diff --git a/tests/test_verify_task.py b/tests/test_verify_task.py index 39c9308..af1db80 100644 --- a/tests/test_verify_task.py +++ b/tests/test_verify_task.py @@ -19,6 +19,24 @@ from forge.quality_report import recommend_status from forge.task_builder import gold_patch_path, metadata_path, verification_path, verify_task from forge.task_schema import TaskMetadata +from forge.test_report import REPORT_END, REPORT_START + + +def _junit(cases: dict[str, str]) -> str: + parts = [] + for key, status in cases.items(): + classname, name = key.split("::") + body = {"passed": "", "failed": "", "skipped": "", "error": ""}[status] + parts.append(f'{body}') + return "" + "".join(parts) + "" + + +def _docker_report(exit_code: int, cases: dict[str, str], *, build: bool = True) -> DockerRunResult: + stdout = f"collected\n{REPORT_START}\n{_junit(cases)}\n{REPORT_END}\n" + return DockerRunResult( + command="pytest", exit_code=exit_code, timed_out=False, stdout=stdout, stderr="", + error=None, docker_build_success=build, image_tag="img", duration_seconds=1.0, + ) def _metadata(task_id: str = "ver-001") -> TaskMetadata: @@ -84,7 +102,7 @@ def ensure(*_args, **_kwargs): def _phase_runner(results: dict[str, DockerRunResult]): """Dispatch run_tests_in_docker results by the phase encoded in image_tag_prefix.""" - def runner(repo_path, test_command, *, timeout_seconds, image_tag_prefix, dockerfile_path=None): + def runner(repo_path, test_command, *, timeout_seconds, image_tag_prefix, dockerfile_path=None, exec_command=None): phase = image_tag_prefix.rsplit("-", 1)[-1] return results[phase] @@ -187,6 +205,33 @@ def test_verify_nondeterministic_rerun_is_needs_review(tmp_path: Path, monkeypat assert recommend_status(result) == "needs_review" +def test_verify_derives_fail_to_pass_and_pass_to_pass(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + _setup_task(tmp_path) # default test_command is "pytest" -> reports are collected + before = _docker_report(1, {"pkg::test_keep": "passed", "pkg::test_fix": "failed"}) + after = _docker_report(0, {"pkg::test_keep": "passed", "pkg::test_fix": "passed"}) + rerun = _docker_report(0, {"pkg::test_keep": "passed", "pkg::test_fix": "passed"}) + _patch_collaborators(monkeypatch, runner=_phase_runner({"before": before, "after": after, "rerun": rerun})) + + result = verify_task("ver-001", root=tmp_path) + + assert result.fail_to_pass == ["pkg::test_fix"] + assert result.pass_to_pass == ["pkg::test_keep"] + assert recommend_status(result) == "usable" + + +def test_verify_leaves_targeted_sets_empty_without_reports(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + _setup_task(tmp_path) + _patch_collaborators( + monkeypatch, + runner=_phase_runner({"before": _docker(1), "after": _docker(0), "rerun": _docker(0)}), + ) + + result = verify_task("ver-001", root=tmp_path) + + assert result.fail_to_pass == [] + assert result.pass_to_pass == [] + + def test_verify_docker_build_failure_is_invalid(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: _setup_task(tmp_path) failed = _docker(None, build=False, error="Docker build failed with exit code 1")