Log in
benchmarksPublic

Viewing files at commit dc75e12cb299Commit detailsView latest

Browse files
tests/test_conflict_resolution_lane.py 170 lines · 6.9 KB · python
1
from __future__ import annotations
2
3
import shutil
4
import subprocess
5
import sys
6
import tempfile
7
import unittest
8
import json
9
from pathlib import Path
10
11
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
12
13
import conflict_resolution_lane
14
from oakbench.rows import SKIP_RETURNCODE, validate_rows
15
16
ROOT = Path(__file__).resolve().parents[1]
17
GENERATOR = ROOT / "scripts" / "make_conflict_corpus.py"
18
GIT = shutil.which("git")
19
20
21
@unittest.skipIf(GIT is None, "git not installed")
22
class ConflictResolutionLaneTest(unittest.TestCase):
23
    def setUp(self) -> None:
24
        self.tmp = Path(tempfile.mkdtemp(prefix="oakbench-conflict-lane-test-"))
25
        self.corpus = self.tmp / "bench-conflict-corpus-v1"
26
        proc = subprocess.run(
27
            [
28
                sys.executable,
29
                str(GENERATOR),
30
                "--out",
31
                str(self.corpus),
32
                "--seed",
33
                "test",
34
                "--quick",
35
                "--kinds",
36
                "text-overlap-small,adjacent-lines",
37
            ],
38
            capture_output=True,
39
            text=True,
40
        )
41
        if proc.returncode != 0:
42
            raise AssertionError(f"corpus generation failed:\n{proc.stdout}\n{proc.stderr}")
43
        self.subject = conflict_resolution_lane.Subject("git", "git", "Git", Path(GIT))
44
        self.metadata = {
45
            "bench_id": "test",
46
            "profile": conflict_resolution_lane.PROFILE,
47
            "benchmark_track": "agent-default",
48
            "timestamp_utc": "test",
49
        }
50
51
    def tearDown(self) -> None:
52
        shutil.rmtree(self.tmp, ignore_errors=True)
53
54
    def run_kind(self, kind: str, *, trust_oracle_commands: bool = True) -> list[dict]:
55
        return conflict_resolution_lane.run_subject_kind(
56
            subject=self.subject,
57
            corpus=self.corpus,
58
            kind=kind,
59
            run_index=0,
60
            run_root=self.tmp / "runs",
61
            metadata=self.metadata,
62
            admitted_output_chars=20_000,
63
            trust_oracle_commands=trust_oracle_commands,
64
        )
65
66
    def test_text_conflict_applies_and_verifies_oracle(self) -> None:
67
        rows = self.run_kind("text-overlap-small")
68
69
        self.assertEqual(validate_rows(rows, "workflow"), [])
70
        total = next(row for row in rows if row["operation"] == "workflow.total")
71
        self.assertEqual(total["returncode"], 0)
72
        self.assertEqual(total["conflict_kind"], "text-overlap-small")
73
        self.assertEqual(total["conflict_files_observed"], ["src/module.py"])
74
        self.assertTrue(total["resolution_oracle_hash_verified"])
75
76
        merge = next(row for row in rows if row["operation"] == "conflict.merge")
77
        self.assertEqual(merge["returncode"], 0)
78
        self.assertNotEqual(merge["process_returncode"], 0)
79
        self.assertEqual(merge["conflict_file_count_observed"], 1)
80
81
    def test_lane_failure_predicate_ignores_skips_but_flags_setup_failures(self) -> None:
82
        self.assertFalse(
83
            conflict_resolution_lane.has_lane_failures(
84
                [{"returncode": SKIP_RETURNCODE, "skipped": True}]
85
            )
86
        )
87
        self.assertTrue(conflict_resolution_lane.has_lane_failures([{"operation": "setup.total", "returncode": 1}]))
88
89
    def test_adjacent_lines_records_clean_merge_broken_check_then_fixes(self) -> None:
90
        rows = self.run_kind("adjacent-lines")
91
92
        self.assertEqual(validate_rows(rows, "workflow"), [])
93
        merge = next(row for row in rows if row["operation"] == "conflict.merge")
94
        self.assertEqual(merge["process_returncode"], 0)
95
        self.assertEqual(merge["conflict_files_observed"], [])
96
97
        clean_check = next(row for row in rows if row["operation"] == "conflict.clean_merge_check")
98
        self.assertEqual(clean_check["returncode"], 0)
99
        self.assertEqual(clean_check["process_returncode"], 1)
100
        self.assertEqual(clean_check["post_merge_check_observed_returncode_clean_merge"], 1)
101
102
        total = next(row for row in rows if row["operation"] == "workflow.total")
103
        self.assertEqual(total["returncode"], 0)
104
        self.assertTrue(total["resolution_oracle_hash_verified"])
105
        self.assertEqual(total["post_merge_check_observed_returncode_clean_merge"], 1)
106
107
    def test_unsupported_subject_is_an_honest_skip(self) -> None:
108
        subject = conflict_resolution_lane.Subject("oak", "oak", "Oak", shutil.which("oak") and Path(shutil.which("oak")))
109
        rows = conflict_resolution_lane.run_subject_kind(
110
            subject=subject,
111
            corpus=self.corpus,
112
            kind="text-overlap-small",
113
            run_index=0,
114
            run_root=self.tmp / "runs",
115
            metadata=self.metadata,
116
            admitted_output_chars=20_000,
117
        )
118
119
        self.assertEqual(validate_rows(rows, "workflow"), [])
120
        self.assertEqual(len(rows), 1)
121
        self.assertEqual(rows[0]["returncode"], SKIP_RETURNCODE)
122
        self.assertTrue(rows[0]["skipped"])
123
        self.assertIn("supports_git_local_fixtures_only", rows[0]["skip_reason"])
124
125
    def test_malformed_oracle_is_an_honest_skip(self) -> None:
126
        oracle_path = self.corpus / "text-overlap-small.RESOLUTION.json"
127
        oracle = json.loads(oracle_path.read_text())
128
        oracle.pop("expected_conflict")
129
        oracle_path.write_text(json.dumps(oracle))
130
131
        rows = self.run_kind("text-overlap-small")
132
133
        self.assertEqual(validate_rows(rows, "workflow"), [])
134
        self.assertEqual(len(rows), 1)
135
        self.assertEqual(rows[0]["returncode"], SKIP_RETURNCODE)
136
        self.assertTrue(rows[0]["skipped"])
137
        self.assertIn("conflict_oracle_malformed:missing_or_invalid_expected_conflict", rows[0]["skip_reason"])
138
139
    def test_untrusted_oracle_commands_are_not_executed(self) -> None:
140
        sentinel = self.tmp / "oracle-command-ran"
141
        oracle_path = self.corpus / "text-overlap-small.RESOLUTION.json"
142
        oracle = json.loads(oracle_path.read_text())
143
        oracle["post_merge_check"] = {
144
            "command": [
145
                sys.executable,
146
                "-c",
147
                f"from pathlib import Path; Path({str(sentinel)!r}).write_text('ran')",
148
            ],
149
            "expected_returncode_clean_merge": 0,
150
        }
151
        oracle_path.write_text(json.dumps(oracle))
152
153
        rows = self.run_kind("text-overlap-small", trust_oracle_commands=False)
154
155
        self.assertEqual(validate_rows(rows, "workflow"), [])
156
        self.assertFalse(sentinel.exists())
157
        self.assertNotIn("conflict.clean_merge_check", {row["operation"] for row in rows})
158
        total = next(row for row in rows if row["operation"] == "workflow.total")
159
        self.assertEqual(total["returncode"], 0)
160
161
    def test_resolution_text_content_uses_exact_utf8_bytes_without_newline_normalization(self) -> None:
162
        resolution = {"content": "line one\r\nline two\n"}
163
        self.assertEqual(
164
            conflict_resolution_lane.resolution_bytes(resolution),
165
            b"line one\r\nline two\n",
166
        )
167
168
169
if __name__ == "__main__":
170
    unittest.main()