Log in
benchmarksPublic

Viewing files at commit 63ed254f5df1Commit detailsView latest

Browse files
tests/test_task_loop.py 111 lines · 4.0 KB · python
1
"""Tests for the task-loop lane's derived fields and lane registration."""
2
3
from __future__ import annotations
4
5
import sys
6
import tempfile
7
import unittest
8
from pathlib import Path
9
from unittest.mock import patch
10
11
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
12
13
import task_loop as lane
14
from oakbench.execution import TimedCapture
15
from oakbench.rows import LANE_REQUIRED_KEYS
16
from oakbench.subjects import Subject
17
18
19
class LoopSummaryFieldsTest(unittest.TestCase):
20
    def test_warm_delta_from_acquire_pair(self) -> None:
21
        rows = [
22
            {"operation": "task.acquire", "elapsed_ms": 1000.0},
23
            {"operation": "task.edit", "elapsed_ms": 5.0},
24
            {"operation": "followup.acquire", "elapsed_ms": 600.0},
25
            {"operation": "followup.edit", "elapsed_ms": 5.0},
26
        ]
27
        fields = lane.loop_summary_fields(rows)
28
        self.assertEqual(fields["followup_warm_delta_ms"], -400.0)
29
        self.assertEqual(fields["task_phase_ms"], 1005.0)
30
        self.assertEqual(fields["followup_phase_ms"], 605.0)
31
32
    def test_missing_acquire_yields_null_not_zero(self) -> None:
33
        fields = lane.loop_summary_fields([{"operation": "task.edit", "elapsed_ms": 5.0}])
34
        self.assertIsNone(fields["followup_warm_delta_ms"])
35
36
37
class LaneRegistrationTest(unittest.TestCase):
38
    def test_task_loop_lane_registered(self) -> None:
39
        self.assertIn("task-loop", LANE_REQUIRED_KEYS)
40
        self.assertIn("command", LANE_REQUIRED_KEYS["task-loop"])
41
42
43
class VirtualBranchReTest(unittest.TestCase):
44
    def test_parses_mount_status(self) -> None:
45
        text = "Mount: oak/bench-sync-tmp@abc (virtual branch tl-x-r0-task--9f00aa11)"
46
        match = lane.VIRTUAL_BRANCH_RE.search(text)
47
        self.assertEqual(match.group(1), "tl-x-r0-task--9f00aa11")
48
49
50
class FailurePolicyTest(unittest.TestCase):
51
    def test_step_row_marks_unexpected_zero_exit_as_failure(self) -> None:
52
        subject = Subject("git", "git", "Git", Path("/usr/bin/git"))
53
        capture = TimedCapture(
54
            elapsed_ms=1.0,
55
            returncode=0,
56
            stdout_text="",
57
            stdout_bytes=0,
58
            stdout_truncated=False,
59
            stderr_text="",
60
            stderr_bytes=0,
61
            stderr_truncated=False,
62
        )
63
        with tempfile.TemporaryDirectory() as tmp, patch.object(lane, "run_timed", return_value=capture):
64
            row = lane.step_row(
65
                subject,
66
                "task_loop",
67
                "expect.fail",
68
                ["sh", "-c", "true"],
69
                Path(tmp),
70
                0,
71
                {
72
                    "bench_id": "test",
73
                    "profile": "task-loop",
74
                    "timestamp_utc": "test",
75
                    "benchmark_track": "agent-default",
76
                },
77
                100,
78
                expected_returncodes=(1,),
79
            )
80
81
        self.assertEqual(row["process_returncode"], 0)
82
        self.assertEqual(row["returncode"], 1)
83
84
    def test_oracle_false_marks_total_failed(self) -> None:
85
        total = {
86
            **lane.phase_total_row(
87
                [{"operation": "task.acquire", "elapsed_ms": 1.0, "returncode": 0, "command": []}],
88
                Subject("oak", "oak", "Oak", Path("/usr/bin/oak")),
89
                "task_loop",
90
                0,
91
                {
92
                    "bench_id": "test",
93
                    "profile": "task-loop",
94
                    "timestamp_utc": "test",
95
                    "benchmark_track": "agent-default",
96
                },
97
                "task_loop.total",
98
            ),
99
            "oracle_payload_on_remote_branch": False,
100
        }
101
        lane.enforce_oracle_result(total, False)
102
        self.assertEqual(total["returncode"], 1)
103
        self.assertTrue(lane.has_lane_failures([total]))
104
105
    def test_lane_failures_ignore_skips_but_fail_on_real_errors(self) -> None:
106
        self.assertFalse(lane.has_lane_failures([{"returncode": lane.SKIP_RETURNCODE, "skipped": True}]))
107
        self.assertTrue(lane.has_lane_failures([{"returncode": 1}]))
108
109
110
if __name__ == "__main__":
111
    unittest.main()