Viewing files at commit 2735d7f5558eCommit detailsView latest

Browse files
tests/test_fleet_lane.py 189 lines · 6.9 KB · python
1
"""Tests for the agent-fleet (workspace_per_task) lane plumbing."""
2
3
from __future__ import annotations
4
5
import sys
6
import subprocess
7
import tempfile
8
import unittest
9
from pathlib import Path
10
11
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
12
13
import parallel_contention as lane
14
15
16
class VirtualBranchParseTest(unittest.TestCase):
17
    def test_parses_mount_status_line(self) -> None:
18
        text = "Mount: oak/bench-fleet-tmp@cafb68ba0b1e (virtual branch fleet-mnt-0--fe49717c)\n\nWorking tree clean.\n"
19
        match = lane.VIRTUAL_BRANCH_RE.search(text)
20
        self.assertIsNotNone(match)
21
        self.assertEqual(match.group(1), "fleet-mnt-0--fe49717c")
22
23
    def test_no_match_on_plain_status(self) -> None:
24
        self.assertIsNone(lane.VIRTUAL_BRANCH_RE.search("On branch main\nWorking tree clean.\n"))
25
26
27
class WorkerPayloadPrefixTest(unittest.TestCase):
28
    def test_default_prefix_preserves_legacy_path_shape(self) -> None:
29
        # ADR-0005: the git path's payload location must not move.
30
        result = lane.WorkerResult(worker=0)
31
32
        class FailFast(Exception):
33
            pass
34
35
        # Exercise only the path construction: a missing workspace makes the
36
        # first write raise, recorded as a worker error, never an exception.
37
        lane.worker_loop(
38
            _subject_stub(), Path("/nonexistent-oakbench-test"), 7, 1, result
39
        )
40
        self.assertTrue(result.errors)
41
42
    def test_run_scoped_prefix_lands_in_rel_path(self) -> None:
43
        # The oak path namespaces payloads per run so merges into the
44
        # accumulating disposable main stay file-disjoint across runs.
45
        prefix = "workers/20260611T000000Z-r0"
46
        rel = f"{prefix}/w{3:03d}/f{1:03d}.txt"
47
        self.assertEqual(rel, "workers/20260611T000000Z-r0/w003/f001.txt")
48
49
50
class WorkerFailureContextTest(unittest.TestCase):
51
    def test_failed_worker_row_fields_include_command_stderr_and_path(self) -> None:
52
        with tempfile.TemporaryDirectory() as tmp:
53
            root = Path(tmp)
54
            fake_git = root / "fake-git"
55
            fake_git.write_text("#!/bin/sh\necho fake-git-boom >&2\nexit 2\n")
56
            fake_git.chmod(0o755)
57
            workspace = root / "workspace"
58
            workspace.mkdir()
59
            result = lane.WorkerResult(worker=4)
60
61
            lane.worker_loop(_subject_stub(fake_git), workspace, 4, 1, result)
62
            fields = lane.worker_failure_fields(result, workspace)
63
64
        self.assertEqual(result.commits_attempted, 1)
65
        self.assertEqual(result.commits_succeeded, 0)
66
        self.assertEqual(fields["failure_reason"], "snapshot_command_failed")
67
        self.assertEqual(fields["worker_id"], 4)
68
        self.assertIn("workers/w004/f000.txt", fields["affected_paths"])
69
        self.assertIn("fake-git-boom", fields["stderr"])
70
        self.assertTrue(fields["command"])
71
        self.assertTrue(fields["failure_contexts"])
72
73
    def test_terminal_failure_context_marks_worker_failed_without_missing_commits(self) -> None:
74
        result = lane.WorkerResult(worker=2, commits_attempted=1, commits_succeeded=1)
75
        proc = subprocess.CompletedProcess(
76
            ["git", "merge", "--no-edit", "train-002"],
77
            1,
78
            stdout="",
79
            stderr="merge conflict",
80
        )
81
        lane.record_worker_failure(
82
            result,
83
            "train_merge_failed",
84
            command=["git", "merge", "--no-edit", "train-002"],
85
            proc=proc,
86
            workspace=Path("/tmp/workspace"),
87
        )
88
89
        fields = lane.worker_failure_fields(result, Path("/tmp/workspace"))
90
91
        self.assertTrue(lane.worker_result_failed(result))
92
        self.assertEqual(fields["failure_reason"], "train_merge_failed")
93
        self.assertIn("merge conflict", fields["stderr"])
94
        self.assertEqual(fields["command"], ["git", "merge", "--no-edit", "train-002"])
95
96
    def test_worktree_merge_failure_matches_oak_workspace_merge_semantics(self) -> None:
97
        result = lane.WorkerResult(worker=1, commits_attempted=1, commits_succeeded=1)
98
        proc = subprocess.CompletedProcess(
99
            ["git", "merge", "--no-edit", "task-001"],
100
            1,
101
            stdout="",
102
            stderr="worktree merge conflict",
103
        )
104
        lane.record_worker_failure(
105
            result,
106
            "worktree_merge_failed",
107
            command=["git", "merge", "--no-edit", "task-001"],
108
            proc=proc,
109
            workspace=Path("/tmp/workspace"),
110
        )
111
112
        fields = lane.worker_failure_fields(result, Path("/tmp/workspace"))
113
114
        self.assertTrue(lane.worker_result_failed(result))
115
        self.assertEqual(fields["failure_reason"], "worktree_merge_failed")
116
        self.assertIn("worktree merge conflict", fields["stderr"])
117
        self.assertEqual(fields["command"], ["git", "merge", "--no-edit", "task-001"])
118
119
    def test_recovered_snapshot_failure_context_is_not_fatal(self) -> None:
120
        result = lane.WorkerResult(worker=3, commits_attempted=1, commits_succeeded=1)
121
        proc = subprocess.CompletedProcess(
122
            ["git", "commit", "-m", "retry"],
123
            1,
124
            stdout="transient stdout",
125
            stderr="transient stderr",
126
        )
127
        lane.record_worker_failure(
128
            result,
129
            "snapshot_command_failed",
130
            command=["git", "commit", "-m", "retry"],
131
            proc=proc,
132
            workspace=Path("/tmp/workspace"),
133
            affected_path="workers/w003/f000.txt",
134
        )
135
136
        fields = lane.worker_failure_fields(result, Path("/tmp/workspace"))
137
138
        self.assertFalse(lane.worker_result_failed(result))
139
        self.assertNotIn("failure_reason", fields)
140
        self.assertTrue(fields["failure_contexts"])
141
142
143
def _subject_stub(bin_path: Path = Path("/usr/bin/false")) -> object:
144
    class Stub:
145
        name = "stub"
146
        kind = "git"
147
        label = "stub"
148
        bin = bin_path
149
        git_mode = None
150
151
    return Stub()
152
153
154
class SkipReasonTest(unittest.TestCase):
155
    def test_unresolved_fleet_remote_yields_structured_skip(self) -> None:
156
        import os
157
        from unittest import mock
158
159
        with mock.patch.dict(os.environ, {}, clear=False):
160
            os.environ.pop("OAK_BENCH_FLEET_REPO", None)
161
            os.environ.pop("OAK_BENCH_REMOTE", None)
162
            rows = lane.run_workspace_per_task_oak(
163
                _oak_subject_stub(),
164
                Path("/nonexistent"),
165
                2,
166
                1,
167
                Path("/tmp"),
168
                0,
169
                {"bench_id": "t"},
170
            )
171
        self.assertEqual(len(rows), 1)
172
        self.assertEqual(rows[0]["returncode"], lane.SKIP_RETURNCODE)
173
        self.assertTrue(rows[0]["skipped"])
174
        self.assertIn("OAK_BENCH_FLEET_REPO", rows[0]["skip_reason"])
175
176
177
def _oak_subject_stub() -> object:
178
    class Stub:
179
        name = "oak_installed"
180
        kind = "oak"
181
        label = "oak (installed)"
182
        bin = Path("/usr/bin/false")
183
        git_mode = None
184
185
    return Stub()
186
187
188
if __name__ == "__main__":
189
    unittest.main()