Log in
benchmarksPublic

Viewing files at commit 5717cc9fc7edCommit detailsView latest

Browse files
tests/test_sync_recovery.py 153 lines · 5.5 KB · python
1
"""Tests for sync/divergence recovery workflows and the recovery-cost row."""
2
3
from __future__ import annotations
4
5
import sys
6
import unittest
7
from pathlib import Path
8
9
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
10
11
import workflow_ab as lane
12
13
14
class StubSubject:
15
    name = "git"
16
    kind = "git"
17
    label = "git"
18
    bin = Path("/usr/bin/git")
19
    git_mode = None
20
21
22
def _row(operation: str, **extra) -> dict:
23
    return {"operation": operation, **extra}
24
25
26
class RecoveryMetricsRowTest(unittest.TestCase):
27
    def test_counts_errors_and_recovery_steps(self) -> None:
28
        rows = [
29
            _row("edit.local_change", estimated_tokens_output=5),
30
            _row(
31
                "error.push_diverged",
32
                estimated_tokens_output=120,
33
                error_mentions_recovery_command=True,
34
            ),
35
            _row(
36
                "error.pull_unconfigured",
37
                estimated_tokens_output=80,
38
                error_mentions_recovery_command=False,
39
            ),
40
            _row(
41
                "recovery.pull_rebase",
42
                elapsed_ms=10.0,
43
                estimated_cost_weighted_tokens_with_envelope=200.0,
44
            ),
45
            _row(
46
                "recovery.push_after",
47
                elapsed_ms=5.0,
48
                estimated_cost_weighted_tokens_with_envelope=100.0,
49
            ),
50
            _row("verify.payload"),
51
        ]
52
        row = lane.recovery_metrics_row(
53
            rows, {"bench_id": "t"}, StubSubject(), lane.WORKFLOWS["sync_push_divergence"], 0
54
        )
55
        self.assertIsNotNone(row)
56
        assert row is not None
57
        self.assertEqual(row["error_steps_count"], 2)
58
        self.assertEqual(row["error_output_tokens"], 200)
59
        self.assertEqual(row["recovery_steps_count"], 2)
60
        self.assertEqual(row["recovery_tokens_total"], 300.0)
61
        self.assertEqual(row["error_steps_with_hint"], 2)
62
        self.assertEqual(row["error_steps_hint_actionable"], 1)
63
        self.assertFalse(row["error_actionability"])
64
        self.assertEqual(row["tool_call_count"], 0)
65
66
    def test_actionability_null_when_no_hints(self) -> None:
67
        rows = [_row("error.push_diverged", estimated_tokens_output=10)]
68
        row = lane.recovery_metrics_row(
69
            rows, {}, StubSubject(), lane.WORKFLOWS["sync_push_divergence"], 0
70
        )
71
        self.assertIsNotNone(row)
72
        assert row is not None
73
        self.assertIsNone(row["error_actionability"])
74
75
    def test_none_when_workflow_has_no_failure_rows(self) -> None:
76
        rows = [_row("vcs.status"), _row("sync.pull_upstream"), _row("verify.payload")]
77
        self.assertIsNone(
78
            lane.recovery_metrics_row(
79
                rows, {}, StubSubject(), lane.WORKFLOWS["sync_pull_upstream"], 0
80
            )
81
        )
82
83
    def test_redo_steps_count_as_recovery(self) -> None:
84
        rows = [
85
            _row("error.push_diverged", estimated_tokens_output=10),
86
            _row("recovery.pull_force", elapsed_ms=1.0),
87
            _row("edit.redo_change", elapsed_ms=1.0),
88
            _row("vcs.snapshot.redo", elapsed_ms=1.0),
89
        ]
90
        row = lane.recovery_metrics_row(
91
            rows, {}, StubSubject(), lane.WORKFLOWS["sync_push_divergence"], 0
92
        )
93
        self.assertIsNotNone(row)
94
        assert row is not None
95
        self.assertEqual(row["recovery_steps_count"], 3)
96
97
    def test_recovery_free_steps_emit_zero_cost_summary(self) -> None:
98
        rows = [
99
            _row("edit.fix_last_commit"),
100
            _row("vcs.snapshot"),
101
            _row("recovery_free.desc_update", elapsed_ms=1.0),
102
            _row("recovery_free.push_after", elapsed_ms=1.0),
103
            _row("verify.payload"),
104
        ]
105
        row = lane.recovery_metrics_row(
106
            rows, {}, StubSubject(), lane.WORKFLOWS["sync_non_ff_after_amend"], 0
107
        )
108
        self.assertIsNotNone(row)
109
        assert row is not None
110
        self.assertEqual(row["error_steps_count"], 0)
111
        self.assertEqual(row["recovery_steps_count"], 0)
112
        self.assertEqual(
113
            row["summarized_operations"],
114
            ["recovery_free.desc_update", "recovery_free.push_after"],
115
        )
116
117
118
class SyncWorkflowRegistryTest(unittest.TestCase):
119
    def test_all_sync_workflows_registered_with_remote_purpose(self) -> None:
120
        for name in (
121
            "sync_push_divergence",
122
            "sync_pull_upstream",
123
            "sync_pull_dirty",
124
            "sync_non_ff_after_amend",
125
        ):
126
            self.assertIn(name, lane.WORKFLOWS)
127
            self.assertEqual(lane.WORKFLOWS[name].remote_purpose, "sync")
128
129
    def test_legacy_error_recovery_untouched(self) -> None:
130
        wf = lane.WORKFLOWS["vcs_error_recovery"]
131
        steps = wf.steps(StubSubject(), "agent-default")
132
        ops = [step.operation for step in steps]
133
        self.assertIn("error.commit_nothing", ops)
134
        self.assertIn("error.unknown_subcommand", ops)
135
        self.assertNotIn("error.push_diverged", ops)
136
137
    def test_divergence_error_steps_carry_hints(self) -> None:
138
        ctx = {
139
            "branch": "b",
140
            "remote_dir": "../r.git",
141
            "clone_b": "../b",
142
            "oak_repo": "oak/bench-sync-tmp",
143
            "clone_a": "../a",
144
            "repo_dir": "run-0",
145
            "measure_dir": "../a",
146
        }
147
        steps = lane.sync_push_divergence_steps(StubSubject(), "agent-default", ctx)
148
        hints = {s.operation: s.recovery_hint for s in steps if s.operation.startswith("error.")}
149
        self.assertTrue(all(hint for hint in hints.values()))
150
151
152
if __name__ == "__main__":
153
    unittest.main()