Log in
benchmarksPublic

Viewing files at commit 1a441edafc71Commit detailsView latest

Browse files
tests/test_agent_native_json_probe.py 247 lines · 10.2 KB · python
1
"""Oak agent-native JSON capability probe rows."""
2
3
from __future__ import annotations
4
5
import sys
6
import unittest
7
from pathlib import Path
8
from unittest import mock
9
10
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
11
12
import bench  # noqa: E402
13
from oakbench.execution import TimedCapture  # noqa: E402
14
from oakbench.rows import SKIP_RETURNCODE, validate_rows  # noqa: E402
15
from oakbench.subjects import Subject  # noqa: E402
16
17
18
SUBJECT = Subject(name="oak_installed", kind="oak", label="Oak", bin=Path("oak"))
19
SCENARIO = bench.Scenario("tiny_text", 1, 1, 1, False, 1)
20
METADATA = {"bench_id": "test", "profile": "smoke"}
21
22
23
class JsonFieldOracleTest(unittest.TestCase):
24
    def test_parse_success_and_required_fields(self) -> None:
25
        fields = bench.json_field_oracle_fields(
26
            '{"schema_version":1,"branch":"agent-task","changes":[]}',
27
            ["schema_version", "branch", "changes"],
28
        )
29
30
        self.assertTrue(fields["json_parse_success"])
31
        self.assertTrue(fields["json_oracle_passed"])
32
        self.assertEqual(fields["json_field_completeness"], 1.0)
33
        self.assertEqual(fields["json_missing_required_fields"], [])
34
35
    def test_malformed_json_records_failed_oracle(self) -> None:
36
        fields = bench.json_field_oracle_fields("{not json", ["schema_version"])
37
38
        self.assertFalse(fields["json_parse_success"])
39
        self.assertFalse(fields["json_oracle_passed"])
40
        self.assertEqual(fields["json_missing_required_fields"], ["schema_version"])
41
        self.assertIn("json_error", fields)
42
43
    def test_missing_fields_records_completeness(self) -> None:
44
        fields = bench.json_field_oracle_fields('{"schema_version":1}', ["schema_version", "branch"])
45
46
        self.assertTrue(fields["json_parse_success"])
47
        self.assertFalse(fields["json_oracle_passed"])
48
        self.assertEqual(fields["json_present_required_field_count"], 1)
49
        self.assertEqual(fields["json_missing_required_fields"], ["branch"])
50
        self.assertEqual(fields["json_field_completeness"], 0.5)
51
52
53
class AgentNativeJsonProbeRowTest(unittest.TestCase):
54
    def test_capability_probe_uses_isolated_benchmark_environment(self) -> None:
55
        completed = mock.Mock(returncode=0, stdout="Usage: oak status --json\n", stderr="")
56
        with mock.patch.object(bench.subprocess, "run", return_value=completed) as run:
57
            reason = bench.oak_json_probe_skip_reason("oak", ["status", "--help"], ["status", "--json"])
58
59
        self.assertIsNone(reason)
60
        self.assertEqual(run.call_args.kwargs["env"]["GIT_CONFIG_GLOBAL"], bench.os.devnull)
61
        self.assertEqual(run.call_args.kwargs["env"]["LC_ALL"], "C")
62
63
    def test_missing_command_emits_capability_skip_row(self) -> None:
64
        spec = {
65
            "args": ["agent", "state", "--json"],
66
            "help_args": ["agent", "state", "--help"],
67
            "required_fields": ["schema_version"],
68
        }
69
70
        with mock.patch.object(
71
            bench,
72
            "oak_json_probe_skip_reason",
73
            return_value="capability_gap:oak_json_probe_help_failed:agent state --help",
74
        ):
75
            row = bench.run_agent_native_json_probe(
76
                SUBJECT, SCENARIO, "oak.agent.state.json", Path.cwd(), spec, 0, METADATA, 20000
77
            )
78
79
        self.assertEqual(row["returncode"], SKIP_RETURNCODE)
80
        self.assertTrue(row["skipped"])
81
        self.assertEqual(row["command"], [])
82
        self.assertEqual(row["intended_command"], ["oak", "agent", "state", "--json"])
83
        self.assertEqual(validate_rows([row], "core"), [])
84
85
    def test_minimum_oak_version_emits_capability_skip_row(self) -> None:
86
        spec = {
87
            "args": ["agent", "state", "--json"],
88
            "help_args": ["agent", "state", "--help"],
89
            "required_fields": ["schema_version", "recommended_action"],
90
            "minimum_oak_version": "0.98.0",
91
        }
92
93
        with mock.patch.object(bench, "oak_cli_semver", return_value=(0, 97, 0)):
94
            row = bench.run_agent_native_json_probe(
95
                SUBJECT, SCENARIO, "oak.agent.state.json", Path.cwd(), spec, 0, METADATA, 20000
96
            )
97
98
        self.assertEqual(row["returncode"], SKIP_RETURNCODE)
99
        self.assertTrue(row["skipped"])
100
        self.assertEqual(row["command"], [])
101
        self.assertEqual(row["intended_command"], ["oak", "agent", "state", "--json"])
102
        self.assertEqual(row["minimum_oak_version"], "0.98.0")
103
        self.assertIn("requires_oak>=0.98.0", row["skip_reason"])
104
        self.assertEqual(validate_rows([row], "core"), [])
105
106
    def test_present_command_records_tokens_output_and_json_oracle(self) -> None:
107
        spec = {
108
            "args": ["status", "--json"],
109
            "help_args": ["status", "--help"],
110
            "required_fields": ["schema_version", "branch", "changes"],
111
        }
112
        stdout = '{"schema_version":1,"branch":"agent-task","changes":[]}'
113
        capture = TimedCapture(
114
            elapsed_ms=12.34,
115
            returncode=0,
116
            stdout_text=stdout,
117
            stdout_bytes=len(stdout),
118
            stdout_truncated=False,
119
            stderr_text="",
120
            stderr_bytes=0,
121
            stderr_truncated=False,
122
            peak_rss_bytes=1234,
123
        )
124
125
        with mock.patch.object(bench, "oak_json_probe_skip_reason", return_value=None), mock.patch.object(
126
            bench, "run_timed", return_value=capture
127
        ):
128
            row = bench.run_agent_native_json_probe(
129
                SUBJECT, SCENARIO, "oak.status.json", Path.cwd(), spec, 0, METADATA, 20000
130
            )
131
132
        self.assertEqual(row["returncode"], 0)
133
        self.assertTrue(row["json_parse_success"])
134
        self.assertTrue(row["json_oracle_passed"])
135
        self.assertEqual(row["probe_output_bytes"], len(stdout))
136
        self.assertGreater(row["estimated_tokens_total"], 0)
137
        self.assertEqual(row["tool_call_count"], 1)
138
        self.assertEqual(validate_rows([row], "core"), [])
139
140
    def test_json_oracle_parses_full_stdout_when_admitted_output_is_truncated(self) -> None:
141
        spec = {
142
            "args": ["branch", "diff", "agent-task", "--json"],
143
            "help_args": ["branch", "diff", "--help"],
144
            "required_fields": ["schema_version", "branch", "changes"],
145
        }
146
        stdout = '{"schema_version":1,"branch":"agent-task","changes":["' + ("x" * 100) + '"]}'
147
        admitted_stdout = stdout[:32]
148
        capture = TimedCapture(
149
            elapsed_ms=12.34,
150
            returncode=0,
151
            stdout_text=admitted_stdout,
152
            stdout_bytes=len(stdout),
153
            stdout_truncated=True,
154
            stderr_text="",
155
            stderr_bytes=0,
156
            stderr_truncated=False,
157
            peak_rss_bytes=1234,
158
            stdout_full_text=stdout,
159
            stdout_full_text_truncated=False,
160
        )
161
162
        with mock.patch.object(bench, "oak_json_probe_skip_reason", return_value=None), mock.patch.object(
163
            bench, "run_timed", return_value=capture
164
        ) as run_timed:
165
            row = bench.run_agent_native_json_probe(
166
                SUBJECT, SCENARIO, "oak.branch.diff.json", Path.cwd(), spec, 0, METADATA, 32
167
            )
168
169
        self.assertEqual(run_timed.call_args.args[3], 32)
170
        self.assertEqual(run_timed.call_args.kwargs["full_output_bytes"], bench.JSON_PROBE_MAX_BYTES)
171
        self.assertEqual(row["returncode"], 0)
172
        self.assertTrue(row["stdout_truncated"])
173
        self.assertEqual(row["token_estimate"]["stdout_admitted_chars"], len(admitted_stdout))
174
        self.assertTrue(row["json_parse_success"])
175
        self.assertTrue(row["json_oracle_passed"])
176
        self.assertEqual(row["json_validation_source"], "full_stdout_capture")
177
        self.assertEqual(validate_rows([row], "core"), [])
178
179
    def test_json_oracle_is_unmeasured_when_full_stdout_capture_is_truncated(self) -> None:
180
        spec = {
181
            "args": ["branch", "review", "agent-task", "--json"],
182
            "help_args": ["branch", "review", "--help"],
183
            "required_fields": ["schema_version", "branch", "findings"],
184
        }
185
        capture = TimedCapture(
186
            elapsed_ms=12.34,
187
            returncode=0,
188
            stdout_text='{"schema_version":1,"branch":"agent',
189
            stdout_bytes=200_000_000,
190
            stdout_truncated=True,
191
            stderr_text="",
192
            stderr_bytes=0,
193
            stderr_truncated=False,
194
            peak_rss_bytes=1234,
195
            stdout_full_text=None,
196
            stdout_full_text_truncated=True,
197
        )
198
199
        with mock.patch.object(bench, "oak_json_probe_skip_reason", return_value=None), mock.patch.object(
200
            bench, "run_timed", return_value=capture
201
        ):
202
            row = bench.run_agent_native_json_probe(
203
                SUBJECT, SCENARIO, "oak.branch.review.json", Path.cwd(), spec, 0, METADATA, 32
204
            )
205
206
        self.assertEqual(row["returncode"], 0)
207
        self.assertIsNone(row["json_validation_source"])
208
        self.assertIsNone(row["json_parse_success"])
209
        self.assertIsNone(row["json_oracle_passed"])
210
        self.assertEqual(
211
            row["json_validation_unmeasured_reason"],
212
            "stdout_exceeded_json_validation_capture_limit",
213
        )
214
        self.assertEqual(validate_rows([row], "core"), [])
215
216
    def test_present_command_with_incomplete_json_fails_row(self) -> None:
217
        spec = {
218
            "args": ["status", "--json"],
219
            "help_args": ["status", "--help"],
220
            "required_fields": ["schema_version", "branch"],
221
        }
222
        capture = TimedCapture(
223
            elapsed_ms=1.0,
224
            returncode=0,
225
            stdout_text='{"schema_version":1}',
226
            stdout_bytes=len('{"schema_version":1}'),
227
            stdout_truncated=False,
228
            stderr_text="",
229
            stderr_bytes=0,
230
            stderr_truncated=False,
231
        )
232
233
        with mock.patch.object(bench, "oak_json_probe_skip_reason", return_value=None), mock.patch.object(
234
            bench, "run_timed", return_value=capture
235
        ):
236
            row = bench.run_agent_native_json_probe(
237
                SUBJECT, SCENARIO, "oak.status.json", Path.cwd(), spec, 0, METADATA, 20000
238
            )
239
240
        self.assertEqual(row["returncode"], 1)
241
        self.assertFalse(row["json_oracle_passed"])
242
        self.assertEqual(row["json_missing_required_fields"], ["branch"])
243
        self.assertEqual(validate_rows([row], "core"), [])
244
245
246
if __name__ == "__main__":
247
    unittest.main()