diff --git a/plugins/codex-security/scripts/workbench_db.py b/plugins/codex-security/scripts/workbench_db.py index 0fcacf4fc..5358c25f1 100644 --- a/plugins/codex-security/scripts/workbench_db.py +++ b/plugins/codex-security/scripts/workbench_db.py @@ -3198,7 +3198,8 @@ def patch_artifact_preview( hunk_lines = 0 elif match := re.match(rb"@@ -\d+(?:,(\d+))? \+\d+(?:,(\d+))? @@", chunk): hunk_lines = sum( - int(count) if count is not None else 1 for count in match.groups() + int(count.lstrip(b"0") or b"0") if count is not None else 1 + for count in match.groups() ) elif hunk_lines == 0 and chunk.startswith(b"+++ "): new_headers += 1 diff --git a/plugins/codex-security/tests/test_workbench_db.py b/plugins/codex-security/tests/test_workbench_db.py index 974dac16f..d151e8811 100644 --- a/plugins/codex-security/tests/test_workbench_db.py +++ b/plugins/codex-security/tests/test_workbench_db.py @@ -3356,7 +3356,7 @@ def test_patch_statistics_count_blank_context_lines(tmp_path: Path, blank_contex assert stats == {"additions": 2, "deletions": 2, "fileCount": 2, "previewTruncated": False} -def test_patch_preview_tolerates_integer_conversion_failure(tmp_path: Path) -> None: +def test_patch_preview_accepts_padded_hunk_counts(tmp_path: Path) -> None: namespace = runpy.run_path(str(SCRIPT)) count = b"0" * 5000 + b"1" patch = ( @@ -3368,7 +3368,68 @@ def test_patch_preview_tolerates_integer_conversion_failure(tmp_path: Path) -> N subprocess.run(["git", "apply", "--check", "patch.diff"], cwd=tmp_path, check=True) assert namespace["patch_artifact_preview"]( tmp_path, "patch.diff", f"sha256:{hashlib.sha256(patch).hexdigest()}" - ) == (None, None) + ) == ( + patch.decode(), + {"additions": 1, "deletions": 1, "fileCount": 1, "previewTruncated": False}, + ) + + +@pytest.mark.parametrize("padded", [False, True]) +def test_saved_scan_remains_readable_with_large_hunk_counts(tmp_path: Path, padded: bool) -> None: + state_dir = tmp_path / "state" + target = tmp_path / "target" + target.mkdir() + (target / "source.txt").write_text("original\n") + scan_id, scan_dir = start_saved_scan(state_dir, target, tmp_path / "scans") + write_completed_contract(scan_dir, scan_id, target, relative_path="source.txt") + completed = run_workbench(state_dir, "complete-scan", "--scan-id", scan_id)["scan"] + occurrence_id = completed["findings"][0]["occurrenceId"] + count = "0" * 4300 + "1" if padded else "9" * 4301 + patch = f"--- a/source.txt\n+++ b/source.txt\n@@ -1,{count} +1,{count} @@\n-original\n+fixed\n" + patch_path = scan_dir / "remediation.patch" + patch_path.write_text(patch) + if padded: + subprocess.run(["git", "apply", "--check", str(patch_path)], cwd=target, check=True) + request_id = str(uuid.uuid4()) + token = str(uuid.uuid4()) + run_workbench( + state_dir, + "request-finding-remediation", + "--occurrence-id", + occurrence_id, + "--request-id", + request_id, + "--action-token", + token, + ) + run_workbench( + state_dir, + "set-finding-remediation", + "--occurrence-id", + occurrence_id, + "--request-id", + request_id, + "--action-token", + token, + "--expected-version", + "1", + "--state", + "generated", + "--patch-path", + patch_path.name, + "--patch-digest", + f"sha256:{hashlib.sha256(patch_path.read_bytes()).hexdigest()}", + ) + saved = run_workbench(state_dir, "get-scan", "--scan-id", scan_id)["scan"] + remediation = saved["findings"][0]["remediationState"] + assert remediation["state"] == "generated" + if padded: + assert remediation["patch"] == patch_path.read_bytes().decode() + assert remediation["patchStats"]["additions"] == 1 + assert remediation["patchStats"]["deletions"] == 1 + else: + assert remediation["patch"] is None + assert remediation["patchStats"] is None @pytest.mark.cross_platform