gabriel / muse public
test_coord_push_null_counts.py python
793 lines 34.5 KB
Raw
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 26 days ago
1 """
2 Tests for the bug: hub response with null/non-integer inserted/skipped counts
3 causes an unhandled TypeError or ValueError that escapes run_push as a raw
4 traceback instead of a clean CoordBusError.
5
6 Root cause (coord_bus.py::push_to_hub lines 233-234):
7
8 return {
9 "inserted": int(result.get("inserted", 0)), # BUG
10 "skipped": int(result.get("skipped", 0)), # BUG
11 }
12
13 When the hub returns {"inserted": null}:
14 - result.get("inserted", 0) → None (key EXISTS so default is NOT used)
15 - int(None) → TypeError
16
17 When the hub returns {"inserted": "three"}:
18 - int("three") → ValueError
19
20 Neither TypeError nor ValueError is CoordBusError.
21 run_push only catches CoordBusError — the exception escapes as a raw traceback.
22
23 Coverage:
24 Unit — push_to_hub directly, all bad-value variants
25 Integration — run_push with bad hub response (two layers deep)
26 End-to-end — CLI CliRunner invocation, asserts no traceback in output
27 Stress — 14-batch push, every batch returns a bad response
28 Performance — bad response handling overhead is negligible
29 Security — hub cannot cause arbitrary code exec via count field
30 Data integrity — partial-batch failures produce correct counts in output
31 """
32 from __future__ import annotations
33
34 import argparse
35 import json
36 import pathlib
37 import time
38 from io import BytesIO
39 from unittest.mock import MagicMock, patch
40
41 import pytest
42
43 from muse.core._types import MsgpackDict, MsgpackValue
44
45 # ---------------------------------------------------------------------------
46 # Shared helpers
47 # ---------------------------------------------------------------------------
48
49 _ALL_KINDS = ("reservation", "intent", "release", "heartbeat", "dependency", "task", "claim")
50 _FUTURE_TS = "2099-12-31T23:59:59+00:00"
51
52
53 def _make_repo(tmp_path: pathlib.Path) -> pathlib.Path:
54 (tmp_path / ".muse").mkdir(parents=True, exist_ok=True)
55 return tmp_path
56
57
58 def _one_record() -> MsgpackDict:
59 return {
60 "kind": "reservation",
61 "record_uuid": "res-000001",
62 "run_id": "run-0",
63 "payload": {"reservation_id": "res-000001", "expires_at": _FUTURE_TS},
64 "expires_at": _FUTURE_TS,
65 }
66
67
68 def _make_http_response(body: MsgpackDict) -> BytesIO:
69 return BytesIO(json.dumps(body).encode())
70
71
72 def _run_push_with_hub_response(tmp_path: pathlib.Path, hub_response: MsgpackDict) -> tuple[int | None, str]:
73 """
74 Run run_push with _post_json mocked to return hub_response.
75 Returns (exit_code, stdout_captured).
76 exit_code is None if no SystemExit was raised (i.e. the function returned normally).
77 """
78 import io
79 import sys
80
81 root = _make_repo(tmp_path)
82
83 captured = io.StringIO()
84 exit_code = None
85
86 with patch("muse.cli.commands.coord_sync._gather_local_records",
87 return_value=[_one_record()]), \
88 patch("muse.core.coord_bus._post_json", return_value=hub_response), \
89 patch("muse.cli.commands.coord_sync.require_repo", return_value=root), \
90 patch("muse.cli.commands.coord_sync._resolve_hub_and_signing",
91 return_value=("http://localhost:10003", "tok")), \
92 patch("sys.stdout", captured):
93 args = argparse.Namespace(
94 owner="torvalds", slug="linux",
95 fmt="json", hub_url=None,
96 kinds=["reservation"],
97 )
98 try:
99 from muse.cli.commands.coord_sync import run_push
100 run_push(args)
101 except SystemExit as exc:
102 exit_code = exc.code
103 except Exception as exc:
104 # If a non-SystemExit exception escapes, surface it explicitly
105 return ("CRASH", f"{type(exc).__name__}: {exc}")
106
107 return (exit_code, captured.getvalue())
108
109
110 # =============================================================================
111 # 1. UNIT — push_to_hub directly
112 # =============================================================================
113
114 class TestPushToHubNullCountsUnit:
115 """
116 Unit tests on coord_bus.push_to_hub.
117 _post_json is mocked to return bad count values.
118 Assert that push_to_hub raises CoordBusError, NOT TypeError or ValueError.
119 """
120
121 @pytest.mark.parametrize("bad_inserted", [
122 None, # JSON null
123 "three", # non-numeric string
124 [], # list
125 {}, # dict
126 "1; drop table", # injection attempt
127 ])
128 def test_bad_inserted_raises_coord_bus_error_not_typeerror(self, bad_inserted: MsgpackValue) -> None:
129 from muse.core.coord_bus import push_to_hub, CoordBusError
130 with patch("muse.core.coord_bus._post_json",
131 return_value={"inserted": bad_inserted, "skipped": 0}):
132 with pytest.raises(CoordBusError):
133 push_to_hub("http://localhost:10003", "torvalds", "linux",
134 [_one_record()], signing=None)
135
136 @pytest.mark.parametrize("bad_skipped", [
137 None,
138 "three",
139 [],
140 {},
141 "1; drop table",
142 ])
143 def test_bad_skipped_raises_coord_bus_error_not_typeerror(self, bad_skipped: MsgpackValue) -> None:
144 from muse.core.coord_bus import push_to_hub, CoordBusError
145 with patch("muse.core.coord_bus._post_json",
146 return_value={"inserted": 1, "skipped": bad_skipped}):
147 with pytest.raises(CoordBusError):
148 push_to_hub("http://localhost:10003", "torvalds", "linux",
149 [_one_record()], signing=None)
150
151 def test_both_null_raises_coord_bus_error(self) -> None:
152 from muse.core.coord_bus import push_to_hub, CoordBusError
153 with patch("muse.core.coord_bus._post_json",
154 return_value={"inserted": None, "skipped": None}):
155 with pytest.raises(CoordBusError):
156 push_to_hub("http://localhost:10003", "torvalds", "linux",
157 [_one_record()], signing=None)
158
159 def test_null_never_raises_raw_typeerror(self) -> None:
160 """The specific confirmed bug: int(None) must not escape as TypeError."""
161 from muse.core.coord_bus import push_to_hub, CoordBusError
162 with patch("muse.core.coord_bus._post_json",
163 return_value={"inserted": None, "skipped": 0}):
164 try:
165 push_to_hub("http://localhost:10003", "torvalds", "linux",
166 [_one_record()], signing=None)
167 except CoordBusError:
168 pass # correct
169 except TypeError as exc:
170 pytest.fail(f"Raw TypeError escaped push_to_hub: {exc}")
171
172 def test_invalid_string_never_raises_raw_valueerror(self) -> None:
173 """int('three') must not escape as ValueError."""
174 from muse.core.coord_bus import push_to_hub, CoordBusError
175 with patch("muse.core.coord_bus._post_json",
176 return_value={"inserted": "three", "skipped": 0}):
177 try:
178 push_to_hub("http://localhost:10003", "torvalds", "linux",
179 [_one_record()], signing=None)
180 except CoordBusError:
181 pass # correct
182 except ValueError as exc:
183 pytest.fail(f"Raw ValueError escaped push_to_hub: {exc}")
184
185 # These should SUCCEED — confirm good values still work
186 def test_valid_integer_counts_pass_through(self) -> None:
187 from muse.core.coord_bus import push_to_hub
188 with patch("muse.core.coord_bus._post_json",
189 return_value={"inserted": 1, "skipped": 0}):
190 result = push_to_hub("http://localhost:10003", "torvalds", "linux",
191 [_one_record()], signing=None)
192 assert result == {"inserted": 1, "skipped": 0}
193
194 def test_float_count_truncated_to_int(self) -> None:
195 """Float counts are a hub bug but truncate cleanly when within bounds."""
196 from muse.core.coord_bus import push_to_hub
197 # Send 3 records so int(2.9)=2 and int(0.1)=0 both pass the bounds check
198 three_records = [_one_record(), _one_record(), _one_record()]
199 with patch("muse.core.coord_bus._post_json",
200 return_value={"inserted": 2.9, "skipped": 0.1}):
201 result = push_to_hub("http://localhost:10003", "torvalds", "linux",
202 three_records, signing=None)
203 assert result["inserted"] == 2
204 assert result["skipped"] == 0
205
206 def test_zero_counts_valid(self) -> None:
207 from muse.core.coord_bus import push_to_hub
208 with patch("muse.core.coord_bus._post_json",
209 return_value={"inserted": 0, "skipped": 0}):
210 result = push_to_hub("http://localhost:10003", "torvalds", "linux",
211 [_one_record()], signing=None)
212 assert result == {"inserted": 0, "skipped": 0}
213
214 def test_missing_both_keys_defaults_to_zero(self) -> None:
215 """Hub omits both keys entirely — already handled by .get default."""
216 from muse.core.coord_bus import push_to_hub
217 with patch("muse.core.coord_bus._post_json", return_value={}):
218 result = push_to_hub("http://localhost:10003", "torvalds", "linux",
219 [_one_record()], signing=None)
220 assert result == {"inserted": 0, "skipped": 0}
221
222 def test_negative_count_raises_coord_bus_error(self) -> None:
223 """Hub returning negative counts is a protocol violation."""
224 from muse.core.coord_bus import push_to_hub, CoordBusError
225 with patch("muse.core.coord_bus._post_json",
226 return_value={"inserted": -5, "skipped": 0}):
227 with pytest.raises(CoordBusError):
228 push_to_hub("http://localhost:10003", "torvalds", "linux",
229 [_one_record()], signing=None)
230
231 def test_count_exceeding_batch_size_raises_coord_bus_error(self) -> None:
232 """Hub claims it inserted more records than were sent — impossible."""
233 from muse.core.coord_bus import push_to_hub, CoordBusError, MAX_PUSH_BATCH
234 with patch("muse.core.coord_bus._post_json",
235 return_value={"inserted": MAX_PUSH_BATCH + 1, "skipped": 0}):
236 with pytest.raises(CoordBusError):
237 push_to_hub("http://localhost:10003", "torvalds", "linux",
238 [_one_record()], signing=None)
239
240
241 # =============================================================================
242 # 2. INTEGRATION — run_push with bad hub response (two layers deep)
243 # =============================================================================
244
245 class TestRunPushNullCountsIntegration:
246 """
247 Integration tests: run_push with _post_json mocked at the wire level.
248 Asserts clean exit (SystemExit(1) for hub errors) — never an unhandled exception.
249 """
250
251 def test_null_inserted_exits_cleanly_not_crash(self, tmp_path: pathlib.Path) -> None:
252 code, output = _run_push_with_hub_response(
253 tmp_path, {"inserted": None, "skipped": 0}
254 )
255 assert code != "CRASH", f"run_push crashed: {output}"
256
257 def test_null_skipped_exits_cleanly_not_crash(self, tmp_path: pathlib.Path) -> None:
258 code, output = _run_push_with_hub_response(
259 tmp_path, {"inserted": 1, "skipped": None}
260 )
261 assert code != "CRASH", f"run_push crashed: {output}"
262
263 def test_both_null_exits_cleanly_not_crash(self, tmp_path: pathlib.Path) -> None:
264 code, output = _run_push_with_hub_response(
265 tmp_path, {"inserted": None, "skipped": None}
266 )
267 assert code != "CRASH", f"run_push crashed: {output}"
268
269 def test_string_inserted_exits_cleanly_not_crash(self, tmp_path: pathlib.Path) -> None:
270 code, output = _run_push_with_hub_response(
271 tmp_path, {"inserted": "three", "skipped": 0}
272 )
273 assert code != "CRASH", f"run_push crashed: {output}"
274
275 def test_list_inserted_exits_cleanly_not_crash(self, tmp_path: pathlib.Path) -> None:
276 code, output = _run_push_with_hub_response(
277 tmp_path, {"inserted": [1, 2, 3], "skipped": 0}
278 )
279 assert code != "CRASH", f"run_push crashed: {output}"
280
281 def test_dict_inserted_exits_cleanly_not_crash(self, tmp_path: pathlib.Path) -> None:
282 code, output = _run_push_with_hub_response(
283 tmp_path, {"inserted": {"count": 1}, "skipped": 0}
284 )
285 assert code != "CRASH", f"run_push crashed: {output}"
286
287 def test_bad_response_exits_with_code_1(self, tmp_path: pathlib.Path) -> None:
288 """Bad hub response should be an error exit (code 1), not success (None/0)."""
289 code, output = _run_push_with_hub_response(
290 tmp_path, {"inserted": None, "skipped": None}
291 )
292 # run_push raises SystemExit(1) when failed=True; clean success returns None
293 assert code == 1, f"expected exit code 1 for bad hub response, got {code!r}"
294
295 def test_bad_response_json_output_has_failed_true(self, tmp_path: pathlib.Path) -> None:
296 """JSON output must have failed=true, not a raw exception message."""
297 code, output = _run_push_with_hub_response(
298 tmp_path, {"inserted": None, "skipped": None}
299 )
300 lines = [l for l in output.strip().splitlines() if l.strip()]
301 assert lines, "no output produced"
302 summary = json.loads(lines[-1])
303 assert summary.get("failed") is True, f"expected failed=true in {summary}"
304
305 def test_bad_response_output_contains_no_traceback(self, tmp_path: pathlib.Path) -> None:
306 """Traceback must never appear in stdout."""
307 _, output = _run_push_with_hub_response(
308 tmp_path, {"inserted": None, "skipped": None}
309 )
310 assert "Traceback" not in output, f"traceback leaked to stdout:\n{output}"
311 assert "TypeError" not in output, f"TypeError leaked to stdout:\n{output}"
312 assert "ValueError" not in output, f"ValueError leaked to stdout:\n{output}"
313
314 def test_good_response_still_works_after_fix(self, tmp_path: pathlib.Path) -> None:
315 """Valid hub response must still succeed after the fix is applied."""
316 code, output = _run_push_with_hub_response(
317 tmp_path, {"inserted": 1, "skipped": 0}
318 )
319 # run_push does not raise SystemExit on success — exit_code stays None
320 assert code in (0, None), f"expected clean exit for valid response, got {code!r}"
321 summary = json.loads(output.strip().splitlines()[-1])
322 assert summary["inserted"] == 1
323 assert summary["skipped"] == 0
324 assert summary["failed"] is False
325
326
327 # =============================================================================
328 # 3. END-TO-END — CLI output is valid JSON with no tracebacks
329 # =============================================================================
330
331 class TestRunPushNullCountsEndToEnd:
332 """
333 End-to-end: simulate what an operator would see at the terminal.
334 Output must be valid JSON, must contain no Python traceback text,
335 and the process must exit cleanly (no unhandled exception).
336 """
337
338 @pytest.mark.parametrize("bad_response", [
339 {"inserted": None, "skipped": None},
340 {"inserted": None, "skipped": 0},
341 {"inserted": 1, "skipped": None},
342 {"inserted": "bad", "skipped": 0},
343 {"inserted": [], "skipped": 0},
344 {}, # completely empty
345 {"other": "keys"}, # no inserted/skipped at all (already handled)
346 ])
347 def test_cli_output_is_valid_json_for_bad_hub_response(self, tmp_path: pathlib.Path, bad_response: MsgpackDict) -> None:
348 code, output = _run_push_with_hub_response(tmp_path, bad_response)
349 assert code != "CRASH", f"run_push crashed on {bad_response}: {output}"
350 lines = [l for l in output.strip().splitlines() if l.strip()]
351 assert lines, f"no output for hub response {bad_response}"
352 # Every output line must be valid JSON
353 for line in lines:
354 try:
355 json.loads(line)
356 except json.JSONDecodeError:
357 pytest.fail(f"non-JSON line in output for {bad_response}: {line!r}")
358
359 def test_cli_output_never_contains_exception_class_names(self, tmp_path: pathlib.Path) -> None:
360 for bad_val in [None, "bad", [], {}]:
361 _, output = _run_push_with_hub_response(
362 tmp_path, {"inserted": bad_val, "skipped": 0}
363 )
364 for forbidden in ("TypeError", "ValueError", "AttributeError",
365 "Traceback", "most recent call"):
366 assert forbidden not in output, (
367 f"{forbidden!r} leaked into CLI output for inserted={bad_val!r}:\n{output}"
368 )
369
370 def test_text_mode_also_clean_on_bad_response(self, tmp_path: pathlib.Path) -> None:
371 """Non-JSON (text) mode must also not crash."""
372 import io
373 import sys
374 root = _make_repo(tmp_path)
375 captured = io.StringIO()
376
377 with patch("muse.cli.commands.coord_sync._gather_local_records",
378 return_value=[_one_record()]), \
379 patch("muse.core.coord_bus._post_json",
380 return_value={"inserted": None, "skipped": None}), \
381 patch("muse.cli.commands.coord_sync.require_repo", return_value=root), \
382 patch("muse.cli.commands.coord_sync._resolve_hub_and_signing",
383 return_value=("http://localhost:10003", "tok")), \
384 patch("sys.stdout", captured):
385 args = argparse.Namespace(
386 owner="torvalds", slug="linux",
387 fmt="text", hub_url=None,
388 kinds=["reservation"],
389 )
390 try:
391 from muse.cli.commands.coord_sync import run_push
392 run_push(args)
393 except SystemExit:
394 pass
395 except Exception as exc:
396 pytest.fail(f"text mode crashed: {type(exc).__name__}: {exc}")
397
398 output = captured.getvalue()
399 assert "Traceback" not in output
400 assert "TypeError" not in output
401
402
403 # =============================================================================
404 # 4. STRESS — 14 batches, every batch returns a bad response
405 # =============================================================================
406
407 class TestRunPushNullCountsStress:
408 """
409 Stress tests: 14 batches (7000 records) all return null counts.
410 The system must not crash on any batch and must emit correct summary output.
411 """
412
413 def _run_push_n_batches(
414 self,
415 tmp_path: pathlib.Path,
416 n_records: int,
417 hub_responses: list[MsgpackDict],
418 ) -> tuple[int | None, MsgpackDict]:
419 import io
420 root = _make_repo(tmp_path)
421 records = [
422 {
423 "kind": "reservation",
424 "record_uuid": f"res-{i:06d}",
425 "run_id": f"run-{i}",
426 "payload": {},
427 "expires_at": _FUTURE_TS,
428 }
429 for i in range(n_records)
430 ]
431
432 response_iter = iter(hub_responses)
433
434 def fake_post_json(url: str, body: MsgpackDict, token: str) -> MsgpackDict:
435 try:
436 return next(response_iter)
437 except StopIteration:
438 return {"inserted": 0, "skipped": 0}
439
440 captured = io.StringIO()
441 exit_code = None
442
443 with patch("muse.cli.commands.coord_sync._gather_local_records",
444 return_value=records), \
445 patch("muse.core.coord_bus._post_json", side_effect=fake_post_json), \
446 patch("muse.cli.commands.coord_sync.require_repo", return_value=root), \
447 patch("muse.cli.commands.coord_sync._resolve_hub_and_signing",
448 return_value=("http://localhost:10003", "tok")), \
449 patch("sys.stdout", captured):
450 args = argparse.Namespace(
451 owner="torvalds", slug="linux",
452 fmt="json", hub_url=None,
453 kinds=["reservation"],
454 )
455 try:
456 from muse.cli.commands.coord_sync import run_push
457 run_push(args)
458 except SystemExit as exc:
459 exit_code = exc.code
460 except Exception as exc:
461 return ("CRASH", {})
462
463 lines = [l for l in captured.getvalue().strip().splitlines() if l.strip()]
464 summary = json.loads(lines[-1]) if lines else {}
465 return (exit_code, summary)
466
467 def test_all_14_batches_return_null_no_crash(self, tmp_path: pathlib.Path) -> None:
468 from muse.core.coord_bus import MAX_PUSH_BATCH
469 n = MAX_PUSH_BATCH * 14 # 7000 records
470 responses = [{"inserted": None, "skipped": None}] * 14
471 code, summary = self._run_push_n_batches(tmp_path, n, responses)
472 assert code != "CRASH", "run_push crashed on 14 null-count batches"
473
474 def test_all_14_batches_return_null_exit_code_1(self, tmp_path: pathlib.Path) -> None:
475 from muse.core.coord_bus import MAX_PUSH_BATCH
476 n = MAX_PUSH_BATCH * 14
477 responses = [{"inserted": None, "skipped": None}] * 14
478 code, summary = self._run_push_n_batches(tmp_path, n, responses)
479 assert code == 1, f"expected exit 1 for all-null batches, got {code!r}"
480
481 def test_all_14_batches_return_null_failed_true_in_output(self, tmp_path: pathlib.Path) -> None:
482 from muse.core.coord_bus import MAX_PUSH_BATCH
483 n = MAX_PUSH_BATCH * 14
484 responses = [{"inserted": None, "skipped": None}] * 14
485 code, summary = self._run_push_n_batches(tmp_path, n, responses)
486 assert summary.get("failed") is True
487
488 def test_alternating_good_and_null_batches(self, tmp_path: pathlib.Path) -> None:
489 """Odd batches succeed, even batches return null. No crash. failed=true."""
490 from muse.core.coord_bus import MAX_PUSH_BATCH
491 n = MAX_PUSH_BATCH * 6 # 3000 records
492 responses = []
493 for i in range(6):
494 if i % 2 == 0:
495 responses.append({"inserted": MAX_PUSH_BATCH, "skipped": 0})
496 else:
497 responses.append({"inserted": None, "skipped": None})
498
499 code, summary = self._run_push_n_batches(tmp_path, n, responses)
500 assert code != "CRASH"
501 assert summary.get("failed") is True
502 # 3 good batches × MAX_PUSH_BATCH inserted
503 assert summary.get("inserted") == MAX_PUSH_BATCH * 3
504
505 def test_first_batch_null_rest_succeed(self, tmp_path: pathlib.Path) -> None:
506 from muse.core.coord_bus import MAX_PUSH_BATCH
507 n = MAX_PUSH_BATCH * 3
508 responses = [
509 {"inserted": None, "skipped": None},
510 {"inserted": MAX_PUSH_BATCH, "skipped": 0},
511 {"inserted": MAX_PUSH_BATCH, "skipped": 0},
512 ]
513 code, summary = self._run_push_n_batches(tmp_path, n, responses)
514 assert code != "CRASH"
515 assert summary.get("failed") is True
516 # 2 good batches should still be counted
517 assert summary.get("inserted") == MAX_PUSH_BATCH * 2
518
519 def test_last_batch_null_rest_succeed(self, tmp_path: pathlib.Path) -> None:
520 from muse.core.coord_bus import MAX_PUSH_BATCH
521 n = MAX_PUSH_BATCH * 3
522 responses = [
523 {"inserted": MAX_PUSH_BATCH, "skipped": 0},
524 {"inserted": MAX_PUSH_BATCH, "skipped": 0},
525 {"inserted": None, "skipped": None},
526 ]
527 code, summary = self._run_push_n_batches(tmp_path, n, responses)
528 assert code != "CRASH"
529 assert summary.get("failed") is True
530 assert summary.get("inserted") == MAX_PUSH_BATCH * 2
531
532
533 # =============================================================================
534 # 5. PERFORMANCE — bad response handling overhead is negligible
535 # =============================================================================
536
537 class TestRunPushNullCountsPerformance:
538 """
539 Bad response handling must not introduce measurable overhead.
540 Error paths in push_to_hub should be as fast as success paths.
541 """
542
543 def _measure_push(self, tmp_path: pathlib.Path, response: MsgpackDict) -> float:
544 import io
545 root = _make_repo(tmp_path)
546 records = [_one_record()]
547
548 with patch("muse.cli.commands.coord_sync._gather_local_records",
549 return_value=records), \
550 patch("muse.core.coord_bus._post_json", return_value=response), \
551 patch("muse.cli.commands.coord_sync.require_repo", return_value=root), \
552 patch("muse.cli.commands.coord_sync._resolve_hub_and_signing",
553 return_value=("http://localhost:10003", "tok")), \
554 patch("sys.stdout", io.StringIO()):
555 args = argparse.Namespace(
556 owner="torvalds", slug="linux",
557 fmt="json", hub_url=None,
558 kinds=["reservation"],
559 )
560 t0 = time.monotonic()
561 try:
562 from muse.cli.commands.coord_sync import run_push
563 run_push(args)
564 except SystemExit:
565 pass
566 return time.monotonic() - t0
567
568 def test_null_response_not_slower_than_good_response(self, tmp_path: pathlib.Path) -> None:
569 # Warm up
570 self._measure_push(tmp_path, {"inserted": 1, "skipped": 0})
571 self._measure_push(tmp_path / "x", {"inserted": None, "skipped": None})
572
573 good = self._measure_push(tmp_path / "good", {"inserted": 1, "skipped": 0})
574 bad = self._measure_push(tmp_path / "bad", {"inserted": None, "skipped": None})
575
576 assert bad < max(good * 10, 0.100), (
577 f"null response path ({bad:.4f}s) is unexpectedly slower than "
578 f"good response ({good:.4f}s)"
579 )
580
581 def test_100_consecutive_null_responses_under_1s(self, tmp_path: pathlib.Path) -> None:
582 import io
583
584 root = _make_repo(tmp_path)
585 records = [_one_record()]
586
587 t0 = time.monotonic()
588 for i in range(100):
589 with patch("muse.cli.commands.coord_sync._gather_local_records",
590 return_value=records), \
591 patch("muse.core.coord_bus._post_json",
592 return_value={"inserted": None, "skipped": None}), \
593 patch("muse.cli.commands.coord_sync.require_repo", return_value=root), \
594 patch("muse.cli.commands.coord_sync._resolve_hub_and_signing",
595 return_value=("http://localhost:10003", "tok")), \
596 patch("sys.stdout", io.StringIO()):
597 args = argparse.Namespace(
598 owner="torvalds", slug="linux",
599 fmt="json", hub_url=None,
600 kinds=["reservation"],
601 )
602 try:
603 from muse.cli.commands.coord_sync import run_push
604 run_push(args)
605 except SystemExit:
606 pass
607 elapsed = time.monotonic() - t0
608 assert elapsed < 1.0, f"100 null-response pushes took {elapsed:.3f}s (> 1s)"
609
610
611 # =============================================================================
612 # 6. SECURITY — hub cannot cause code execution via count fields
613 # =============================================================================
614
615 class TestRunPushNullCountsSecurity:
616 """
617 Security: a malicious hub cannot exploit the count parsing path.
618 All attack payloads must result in CoordBusError, never in exec/import.
619 """
620
621 @pytest.mark.parametrize("attack_payload", [
622 "__import__('os').system('ls')",
623 "1; __import__('os').system('ls')",
624 "exec('import os')",
625 "${7*7}",
626 "{{7*7}}",
627 "' OR 1=1 --",
628 "\x00\x01\x02",
629 "9" * 10000, # absurdly long numeric string
630 "1e308", # float overflow
631 "inf",
632 "nan",
633 "-inf",
634 ])
635 def test_malicious_inserted_raises_coord_bus_error_not_exec(self, attack_payload: str | int | float | None) -> None:
636 from muse.core.coord_bus import push_to_hub, CoordBusError
637 with patch("muse.core.coord_bus._post_json",
638 return_value={"inserted": attack_payload, "skipped": 0}):
639 try:
640 push_to_hub("http://localhost:10003", "torvalds", "linux",
641 [_one_record()], signing=None)
642 except CoordBusError:
643 pass # correct — attack contained
644 except Exception as exc:
645 pytest.fail(
646 f"Attack payload {attack_payload!r} escaped as "
647 f"{type(exc).__name__}: {exc}"
648 )
649
650 def test_attack_payload_causes_coord_bus_error_not_execution(self, tmp_path: pathlib.Path) -> None:
651 """
652 Malicious count value must be rejected as CoordBusError — not executed.
653 The error message may contain the repr of the bad value (that is fine for
654 a CLI tool), but the Python expression must never be evaluated.
655 """
656 attack = "__import__('os').system('id')"
657 code, output = _run_push_with_hub_response(
658 tmp_path, {"inserted": attack, "skipped": 0}
659 )
660 # Must be an error exit, not success
661 assert code == 1, f"expected exit 1 for attack payload, got {code!r}"
662 # Must not crash with unhandled exception
663 assert code != "CRASH", f"attack payload caused crash: {output}"
664 # Output must be valid JSON (no raw traceback)
665 lines = [l for l in output.strip().splitlines() if l.strip()]
666 for line in lines:
667 try:
668 json.loads(line)
669 except json.JSONDecodeError:
670 pytest.fail(f"non-JSON output for attack payload: {line!r}")
671
672 def test_extremely_large_count_rejected(self, tmp_path: pathlib.Path) -> None:
673 """Hub claiming it inserted 2^63 records is impossible; treat as error."""
674 huge = 2**63
675 code, output = _run_push_with_hub_response(
676 tmp_path, {"inserted": huge, "skipped": 0}
677 )
678 # Should not silently succeed with a nonsensical count
679 assert code != "CRASH"
680 lines = [l for l in output.strip().splitlines() if l.strip()]
681 summary = json.loads(lines[-1]) if lines else {}
682 # Either it fails, or if it "succeeds" the count must be sane (not 2^63)
683 if summary.get("failed") is False:
684 assert summary.get("inserted", 0) <= 10**9, (
685 f"hub's 2^63 count was accepted verbatim: {summary}"
686 )
687
688
689 # =============================================================================
690 # 7. DATA INTEGRITY — counts in output reflect reality
691 # =============================================================================
692
693 class TestRunPushNullCountsDataIntegrity:
694 """
695 When some batches succeed and some return bad counts, the summary output
696 must accurately reflect only the records from successful batches.
697 """
698
699 def test_total_reflects_records_sent_not_hub_count(self, tmp_path: pathlib.Path) -> None:
700 """'total' in output is len(local records), independent of hub response."""
701 code, output = _run_push_with_hub_response(
702 tmp_path, {"inserted": None, "skipped": None}
703 )
704 lines = [l for l in output.strip().splitlines() if l.strip()]
705 summary = json.loads(lines[-1])
706 # total must be 1 (we sent 1 record) regardless of hub response
707 assert summary.get("total") == 1, (
708 f"total should be 1 (records sent), got {summary.get('total')}"
709 )
710
711 def test_inserted_is_zero_when_hub_returns_null(self, tmp_path: pathlib.Path) -> None:
712 """When hub returns null for inserted, the count must be 0, not garbage."""
713 code, output = _run_push_with_hub_response(
714 tmp_path, {"inserted": None, "skipped": 1}
715 )
716 lines = [l for l in output.strip().splitlines() if l.strip()]
717 summary = json.loads(lines[-1])
718 # After fix: inserted should be 0 (not crashing, not garbage)
719 assert isinstance(summary.get("inserted"), int), (
720 f"inserted must be int in summary, got {summary.get('inserted')!r}"
721 )
722
723 def test_skipped_is_zero_when_hub_returns_null(self, tmp_path: pathlib.Path) -> None:
724 code, output = _run_push_with_hub_response(
725 tmp_path, {"inserted": 1, "skipped": None}
726 )
727 lines = [l for l in output.strip().splitlines() if l.strip()]
728 summary = json.loads(lines[-1])
729 assert isinstance(summary.get("skipped"), int), (
730 f"skipped must be int in summary, got {summary.get('skipped')!r}"
731 )
732
733 def test_partial_null_counts_are_accumulated_correctly(self, tmp_path: pathlib.Path) -> None:
734 """
735 3 batches: inserted=[5, null, 3].
736 After fix: total inserted = 5 + 0 + 3 = 8.
737 """
738 import io
739 from muse.core.coord_bus import MAX_PUSH_BATCH
740
741 root = _make_repo(tmp_path)
742 records = [
743 {"kind": "reservation", "record_uuid": f"res-{i:06d}",
744 "run_id": "r", "payload": {}, "expires_at": _FUTURE_TS}
745 for i in range(3)
746 ]
747
748 responses = iter([
749 {"inserted": 5, "skipped": 0},
750 {"inserted": None, "skipped": None},
751 {"inserted": 3, "skipped": 0},
752 ])
753
754 def fake_post(url: str, body: MsgpackDict, token: str) -> MsgpackDict:
755 return next(responses)
756
757 captured = io.StringIO()
758 with patch("muse.cli.commands.coord_sync._gather_local_records",
759 return_value=records * MAX_PUSH_BATCH), \
760 patch("muse.core.coord_bus._post_json", side_effect=fake_post), \
761 patch("muse.cli.commands.coord_sync.require_repo", return_value=root), \
762 patch("muse.cli.commands.coord_sync._resolve_hub_and_signing",
763 return_value=("http://localhost:10003", "tok")), \
764 patch("sys.stdout", captured):
765 args = argparse.Namespace(
766 owner="torvalds", slug="linux",
767 fmt="json", hub_url=None,
768 kinds=["reservation"],
769 )
770 try:
771 from muse.cli.commands.coord_sync import run_push
772 run_push(args)
773 except SystemExit:
774 pass
775 except Exception as exc:
776 pytest.fail(f"Crashed: {type(exc).__name__}: {exc}")
777
778 lines = [l for l in captured.getvalue().strip().splitlines() if l.strip()]
779 summary = json.loads(lines[-1])
780 assert summary.get("inserted") == 8, (
781 f"expected inserted=8 (5+0+3), got {summary.get('inserted')}"
782 )
783 assert summary.get("failed") is True, "middle batch failed, so failed must be True"
784
785 def test_output_json_schema_complete_on_bad_response(self, tmp_path: pathlib.Path) -> None:
786 """All required keys must be present in JSON output even on error."""
787 code, output = _run_push_with_hub_response(
788 tmp_path, {"inserted": None, "skipped": None}
789 )
790 lines = [l for l in output.strip().splitlines() if l.strip()]
791 summary = json.loads(lines[-1])
792 for key in ("schema_version", "inserted", "skipped", "total", "failed", "elapsed_seconds"):
793 assert key in summary, f"key {key!r} missing from summary: {summary}"
File History 5 commits
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 26 days ago
sha256:d8316ffae901be06347e16ab55be11868eb519dd16ade3e8aa16a99e662f7e62 baseline: rc14 re-baseline after rc3 store corruption recovery Human patch 26 days ago
sha256:50d413a635f57761c3fce1f70251b957b6423821525bf330747ef4007339222e Merge branch 'dev' into main Human 29 days ago
sha256:fb67fed5a4d3e40de84bdd163de94ef1386570bef1dd1a020a732c8a038962ce Merge branch 'dev' into main Human 48 days ago
sha256:1c4b3e3a9a1f300774c3ee662b572a698d5fd405bf765a71e6011a2e9c3eaaaa feat: Muse — version control for the agent era Human 100 days ago