test_cmd_coord_sync.py
python
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6
chore(tests): add docstring to tests/__init__.py so rc14 tr…
Human
26 days ago
| 1 | """Comprehensive tests for ``muse coord sync push`` and ``muse coord sync pull``. |
| 2 | |
| 3 | Coverage matrix |
| 4 | --------------- |
| 5 | Unit |
| 6 | ~~~~ |
| 7 | * _gather_local_records — reads reservations from disk |
| 8 | * _gather_local_records — reads heartbeats from disk |
| 9 | * _gather_local_records — kinds filter respected |
| 10 | * _gather_local_records — corrupt file skipped gracefully |
| 11 | * _gather_local_records — claims use claimer_run_id field |
| 12 | * _gather_local_records — all 7 kinds gathered |
| 13 | * _write_remote_records — writes correct paths under remote/ |
| 14 | * _write_remote_records — overwrites existing files |
| 15 | * _write_remote_records — record with no uuid skipped |
| 16 | * _write_remote_records — unknown kind rejected (path traversal prevention) |
| 17 | * _write_remote_records — unsafe record_uuid rejected (path traversal prevention) |
| 18 | * _write_remote_records — compact JSON written (no indent) |
| 19 | |
| 20 | Integration (all network calls mocked) |
| 21 | ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ |
| 22 | * push with no local records — text says "(no local coordination records to push)" |
| 23 | * push with reservations — calls push_to_hub with correct args |
| 24 | * push CoordBusError — exits 1 with error message |
| 25 | * pull empty result — prints "Pulled 0 new record(s)" |
| 26 | * pull with records — writes files to remote dir |
| 27 | * pull CoordBusError — exits 1 with error message |
| 28 | * --format json push — valid JSON with inserted/skipped/total/failed/elapsed/schema_version |
| 29 | * --format json pull — valid JSON with count/cursor/records/elapsed/schema_version |
| 30 | * --json shorthand push — same as --format json |
| 31 | * --json shorthand pull — same as --format json |
| 32 | * --since-id N — passed through to pull_from_hub |
| 33 | * --kinds filter push — restricts gathered kinds |
| 34 | * --limit N — passed through to pull_from_hub |
| 35 | * elapsed_seconds present in both push and pull JSON output |
| 36 | |
| 37 | Input validation |
| 38 | ~~~~~~~~~~~~~~~~ |
| 39 | * --owner too long → exit 1 before any I/O |
| 40 | * --slug too long → exit 1 before any I/O |
| 41 | * --since-id negative → exit 1 before any I/O |
| 42 | * --limit = 0 → exit 1 before any I/O |
| 43 | * --limit > 1000 → exit 1 before any I/O |
| 44 | * --limit at boundary 1 → accepted |
| 45 | * --limit at boundary 1000 → accepted |
| 46 | * push owner/slug validation fires before require_repo |
| 47 | * pull since-id/limit validation fires before require_repo |
| 48 | |
| 49 | Security |
| 50 | ~~~~~~~~ |
| 51 | * owner/slug with path traversal chars are passed as strings to push_to_hub |
| 52 | * token not echoed in output |
| 53 | * _write_remote_records rejects unknown kind (prevents escaping remote/) |
| 54 | * _write_remote_records rejects traversal in record_uuid |
| 55 | |
| 56 | Stress |
| 57 | ~~~~~~ |
| 58 | * push 600 records splits into multiple batches |
| 59 | * pull returns 1000 records, all written to disk |
| 60 | """ |
| 61 | |
| 62 | from __future__ import annotations |
| 63 | |
| 64 | import json |
| 65 | import pathlib |
| 66 | import uuid |
| 67 | from typing import TYPE_CHECKING |
| 68 | |
| 69 | import pytest |
| 70 | from unittest.mock import patch, MagicMock, call |
| 71 | |
| 72 | from tests.cli_test_helper import CliRunner |
| 73 | from muse.core._types import MsgpackDict |
| 74 | from muse.core.coord_bus import CoordBusError, JsonDict |
| 75 | |
| 76 | if TYPE_CHECKING: |
| 77 | from muse.core.transport import SigningIdentity |
| 78 | from muse.cli.commands.coord_sync import ( |
| 79 | _MAX_OWNER_LEN, |
| 80 | _MAX_SLUG_LEN, |
| 81 | _MAX_PULL_LIMIT, |
| 82 | _ALL_KINDS, |
| 83 | ) |
| 84 | |
| 85 | runner = CliRunner() |
| 86 | cli = None |
| 87 | |
| 88 | _PUSH_TARGET = "muse.cli.commands.coord_sync.push_to_hub" |
| 89 | _PULL_TARGET = "muse.cli.commands.coord_sync.pull_from_hub" |
| 90 | |
| 91 | |
| 92 | # ── Fixtures ────────────────────────────────────────────────────────────────── |
| 93 | |
| 94 | |
| 95 | @pytest.fixture() |
| 96 | def repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path: |
| 97 | muse_dir = tmp_path / ".muse" |
| 98 | muse_dir.mkdir() |
| 99 | (muse_dir / "HEAD").write_text("ref: refs/heads/main\n") |
| 100 | monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path)) |
| 101 | return tmp_path |
| 102 | |
| 103 | |
| 104 | # ── Helpers ─────────────────────────────────────────────────────────────────── |
| 105 | |
| 106 | |
| 107 | def _write_local_claim(repo: pathlib.Path, task_id: str | None = None, run_id: str = "worker-1") -> str: |
| 108 | claim_dir = repo / ".muse" / "coordination" / "claims" |
| 109 | claim_dir.mkdir(parents=True, exist_ok=True) |
| 110 | tid = task_id or str(uuid.uuid4()) |
| 111 | data = { |
| 112 | "task_id": tid, |
| 113 | "claimer_run_id": run_id, |
| 114 | "claimed_at": "2026-01-01T00:00:00+00:00", |
| 115 | "expires_at": "2026-12-31T00:00:00+00:00", |
| 116 | } |
| 117 | (claim_dir / f"{tid}.json").write_text(json.dumps(data)) |
| 118 | return tid |
| 119 | |
| 120 | |
| 121 | def _write_local_reservation(repo: pathlib.Path, run_id: str = "agent-1") -> str: |
| 122 | coord_dir = repo / ".muse" / "coordination" / "reservations" |
| 123 | coord_dir.mkdir(parents=True, exist_ok=True) |
| 124 | rid = str(uuid.uuid4()) |
| 125 | data = { |
| 126 | "reservation_id": rid, |
| 127 | "run_id": run_id, |
| 128 | "branch": "main", |
| 129 | "addresses": ["src/x.py::foo"], |
| 130 | "operation": None, |
| 131 | "created_at": "2026-01-01T00:00:00+00:00", |
| 132 | "expires_at": "2026-12-31T00:00:00+00:00", |
| 133 | } |
| 134 | (coord_dir / f"{rid}.json").write_text(json.dumps(data)) |
| 135 | return rid |
| 136 | |
| 137 | |
| 138 | def _write_local_heartbeat(repo: pathlib.Path, run_id: str = "agent-1") -> str: |
| 139 | hb_dir = repo / ".muse" / "coordination" / "heartbeats" |
| 140 | hb_dir.mkdir(parents=True, exist_ok=True) |
| 141 | data = { |
| 142 | "run_id": run_id, |
| 143 | "last_seen": "2026-01-01T00:01:00+00:00", |
| 144 | "expires_at": "2026-12-31T00:00:00+00:00", |
| 145 | } |
| 146 | (hb_dir / f"{run_id}.json").write_text(json.dumps(data)) |
| 147 | return run_id |
| 148 | |
| 149 | |
| 150 | _PUSH_ARGS = [ |
| 151 | "coord", "sync", "push", |
| 152 | "--hub", "http://localhost:10003", |
| 153 | "--owner", "gabriel", |
| 154 | "--slug", "myrepo", |
| 155 | |
| 156 | ] |
| 157 | |
| 158 | _PULL_ARGS = [ |
| 159 | "coord", "sync", "pull", |
| 160 | "--hub", "http://localhost:10003", |
| 161 | "--owner", "gabriel", |
| 162 | "--slug", "myrepo", |
| 163 | |
| 164 | "--since-id", "0", |
| 165 | ] |
| 166 | |
| 167 | |
| 168 | def _push_ok(inserted: int = 1, skipped: int = 0) -> MsgpackDict: |
| 169 | return {"inserted": inserted, "skipped": skipped} |
| 170 | |
| 171 | |
| 172 | def _pull_ok(records: list[MsgpackDict] | None = None, cursor: int = 0) -> MsgpackDict: |
| 173 | return {"records": records or [], "cursor": cursor} |
| 174 | |
| 175 | |
| 176 | # ── Unit: _gather_local_records ─────────────────────────────────────────────── |
| 177 | |
| 178 | |
| 179 | class TestGatherLocalRecords: |
| 180 | def test_empty_coordination_dir_returns_empty(self, repo: pathlib.Path) -> None: |
| 181 | from muse.cli.commands.coord_sync import _gather_local_records |
| 182 | records = _gather_local_records(repo, kinds=["reservation"]) |
| 183 | assert records == [] |
| 184 | |
| 185 | def test_reads_reservation_from_disk(self, repo: pathlib.Path) -> None: |
| 186 | from muse.cli.commands.coord_sync import _gather_local_records |
| 187 | rid = _write_local_reservation(repo) |
| 188 | records = _gather_local_records(repo, kinds=["reservation"]) |
| 189 | assert len(records) == 1 |
| 190 | assert records[0]["kind"] == "reservation" |
| 191 | assert records[0]["record_uuid"] == rid |
| 192 | |
| 193 | def test_reads_heartbeat_from_disk(self, repo: pathlib.Path) -> None: |
| 194 | from muse.cli.commands.coord_sync import _gather_local_records |
| 195 | run_id = _write_local_heartbeat(repo, "hb-agent") |
| 196 | records = _gather_local_records(repo, kinds=["heartbeat"]) |
| 197 | assert len(records) == 1 |
| 198 | assert records[0]["kind"] == "heartbeat" |
| 199 | assert records[0]["run_id"] == run_id |
| 200 | |
| 201 | def test_kinds_filter_excludes_heartbeats(self, repo: pathlib.Path) -> None: |
| 202 | from muse.cli.commands.coord_sync import _gather_local_records |
| 203 | _write_local_reservation(repo) |
| 204 | _write_local_heartbeat(repo) |
| 205 | records = _gather_local_records(repo, kinds=["reservation"]) |
| 206 | assert all(r["kind"] == "reservation" for r in records) |
| 207 | |
| 208 | def test_kinds_filter_excludes_reservations(self, repo: pathlib.Path) -> None: |
| 209 | from muse.cli.commands.coord_sync import _gather_local_records |
| 210 | _write_local_reservation(repo) |
| 211 | _write_local_heartbeat(repo) |
| 212 | records = _gather_local_records(repo, kinds=["heartbeat"]) |
| 213 | assert all(r["kind"] == "heartbeat" for r in records) |
| 214 | |
| 215 | def test_corrupt_file_skipped_gracefully(self, repo: pathlib.Path) -> None: |
| 216 | from muse.cli.commands.coord_sync import _gather_local_records |
| 217 | coord_dir = repo / ".muse" / "coordination" / "reservations" |
| 218 | coord_dir.mkdir(parents=True, exist_ok=True) |
| 219 | (coord_dir / "bad.json").write_text("not-valid-json{{{{") |
| 220 | records = _gather_local_records(repo, kinds=["reservation"]) |
| 221 | assert records == [] |
| 222 | |
| 223 | def test_multiple_records_all_returned(self, repo: pathlib.Path) -> None: |
| 224 | from muse.cli.commands.coord_sync import _gather_local_records |
| 225 | for i in range(5): |
| 226 | _write_local_reservation(repo, run_id=f"agent-{i}") |
| 227 | records = _gather_local_records(repo, kinds=["reservation"]) |
| 228 | assert len(records) == 5 |
| 229 | |
| 230 | def test_payload_field_contains_original_data(self, repo: pathlib.Path) -> None: |
| 231 | from muse.cli.commands.coord_sync import _gather_local_records |
| 232 | rid = _write_local_reservation(repo, run_id="my-agent") |
| 233 | records = _gather_local_records(repo, kinds=["reservation"]) |
| 234 | assert records[0]["payload"]["reservation_id"] == rid |
| 235 | assert records[0]["payload"]["run_id"] == "my-agent" |
| 236 | |
| 237 | |
| 238 | # ── Unit: _write_remote_records ─────────────────────────────────────────────── |
| 239 | |
| 240 | |
| 241 | class TestWriteRemoteRecords: |
| 242 | def test_writes_file_to_correct_path(self, repo: pathlib.Path) -> None: |
| 243 | from muse.cli.commands.coord_sync import _write_remote_records |
| 244 | rec = {"kind": "reservation", "record_uuid": "abc-123", "payload": {"x": 1}} |
| 245 | _write_remote_records(repo, [rec]) |
| 246 | target = repo / ".muse" / "coordination" / "remote" / "reservation" / "abc-123.json" |
| 247 | assert target.exists() |
| 248 | |
| 249 | def test_written_file_contains_correct_data(self, repo: pathlib.Path) -> None: |
| 250 | from muse.cli.commands.coord_sync import _write_remote_records |
| 251 | rec = {"kind": "reservation", "record_uuid": "def-456", "payload": {"y": 2}} |
| 252 | _write_remote_records(repo, [rec]) |
| 253 | target = repo / ".muse" / "coordination" / "remote" / "reservation" / "def-456.json" |
| 254 | data = json.loads(target.read_text()) |
| 255 | assert data["payload"]["y"] == 2 |
| 256 | |
| 257 | def test_overwrites_existing_file(self, repo: pathlib.Path) -> None: |
| 258 | from muse.cli.commands.coord_sync import _write_remote_records |
| 259 | kind_dir = repo / ".muse" / "coordination" / "remote" / "reservation" |
| 260 | kind_dir.mkdir(parents=True, exist_ok=True) |
| 261 | (kind_dir / "ghi-789.json").write_text('{"old": true}') |
| 262 | rec = {"kind": "reservation", "record_uuid": "ghi-789", "payload": {"new": True}} |
| 263 | _write_remote_records(repo, [rec]) |
| 264 | data = json.loads((kind_dir / "ghi-789.json").read_text()) |
| 265 | assert data["payload"]["new"] is True |
| 266 | |
| 267 | def test_record_without_uuid_skipped(self, repo: pathlib.Path) -> None: |
| 268 | from muse.cli.commands.coord_sync import _write_remote_records |
| 269 | rec = {"kind": "reservation", "record_uuid": "", "payload": {}} |
| 270 | _write_remote_records(repo, [rec]) |
| 271 | remote_dir = repo / ".muse" / "coordination" / "remote" |
| 272 | assert not remote_dir.exists() or not any(remote_dir.rglob("*.json")) |
| 273 | |
| 274 | def test_multiple_kinds_written_to_separate_dirs(self, repo: pathlib.Path) -> None: |
| 275 | from muse.cli.commands.coord_sync import _write_remote_records |
| 276 | records = [ |
| 277 | {"kind": "reservation", "record_uuid": "r1", "payload": {}}, |
| 278 | {"kind": "heartbeat", "record_uuid": "h1", "payload": {}}, |
| 279 | ] |
| 280 | _write_remote_records(repo, records) |
| 281 | assert (repo / ".muse" / "coordination" / "remote" / "reservation" / "r1.json").exists() |
| 282 | assert (repo / ".muse" / "coordination" / "remote" / "heartbeat" / "h1.json").exists() |
| 283 | |
| 284 | |
| 285 | # ── Integration: push ───────────────────────────────────────────────────────── |
| 286 | |
| 287 | |
| 288 | class TestCoordSyncPushIntegration: |
| 289 | def test_push_no_local_records_text_output(self, repo: pathlib.Path) -> None: |
| 290 | with patch(_PUSH_TARGET) as mock_push: |
| 291 | result = runner.invoke(cli, _PUSH_ARGS) |
| 292 | assert result.exit_code == 0 |
| 293 | assert "no local coordination records to push" in result.output |
| 294 | mock_push.assert_not_called() |
| 295 | |
| 296 | def test_push_with_reservation_calls_push_to_hub(self, repo: pathlib.Path) -> None: |
| 297 | _write_local_reservation(repo) |
| 298 | with patch(_PUSH_TARGET, return_value=_push_ok(1, 0)) as mock_push: |
| 299 | result = runner.invoke(cli, _PUSH_ARGS) |
| 300 | assert result.exit_code == 0 |
| 301 | mock_push.assert_called_once() |
| 302 | call_args = mock_push.call_args |
| 303 | assert call_args[0][1] == "gabriel" # owner |
| 304 | assert call_args[0][2] == "myrepo" # slug |
| 305 | |
| 306 | def test_push_text_output_contains_inserted_skipped(self, repo: pathlib.Path) -> None: |
| 307 | _write_local_reservation(repo) |
| 308 | with patch(_PUSH_TARGET, return_value=_push_ok(1, 0)): |
| 309 | result = runner.invoke(cli, _PUSH_ARGS) |
| 310 | assert result.exit_code == 0 |
| 311 | assert "inserted" in result.output |
| 312 | assert "skipped" in result.output |
| 313 | |
| 314 | def test_push_coord_bus_error_exits_1(self, repo: pathlib.Path) -> None: |
| 315 | _write_local_reservation(repo) |
| 316 | with patch(_PUSH_TARGET, side_effect=CoordBusError("hub down")): |
| 317 | result = runner.invoke(cli, _PUSH_ARGS) |
| 318 | assert result.exit_code == 1 |
| 319 | assert "error" in result.output.lower() or "hub down" in result.output |
| 320 | |
| 321 | def test_push_format_json_valid_structure(self, repo: pathlib.Path) -> None: |
| 322 | _write_local_reservation(repo) |
| 323 | with patch(_PUSH_TARGET, return_value=_push_ok(1, 0)): |
| 324 | result = runner.invoke(cli, _PUSH_ARGS + ["--format", "json"]) |
| 325 | assert result.exit_code == 0 |
| 326 | data = json.loads(result.output.strip()) |
| 327 | assert "inserted" in data |
| 328 | assert "skipped" in data |
| 329 | assert "total" in data |
| 330 | assert "failed" in data |
| 331 | |
| 332 | def test_push_json_shorthand(self, repo: pathlib.Path) -> None: |
| 333 | _write_local_reservation(repo) |
| 334 | with patch(_PUSH_TARGET, return_value=_push_ok(2, 1)): |
| 335 | r1 = runner.invoke(cli, _PUSH_ARGS + ["--format", "json"]) |
| 336 | r2 = runner.invoke(cli, _PUSH_ARGS + ["--json"]) |
| 337 | assert json.loads(r1.output.strip()) == json.loads(r2.output.strip()) |
| 338 | |
| 339 | def test_push_json_no_records_returns_zeros(self, repo: pathlib.Path) -> None: |
| 340 | with patch(_PUSH_TARGET): |
| 341 | result = runner.invoke(cli, _PUSH_ARGS + ["--json"]) |
| 342 | assert result.exit_code == 0 |
| 343 | data = json.loads(result.output.strip()) |
| 344 | assert data["total"] == 0 |
| 345 | assert data["inserted"] == 0 |
| 346 | |
| 347 | def test_push_kinds_filter_passed_through(self, repo: pathlib.Path) -> None: |
| 348 | _write_local_reservation(repo) |
| 349 | _write_local_heartbeat(repo) |
| 350 | with patch(_PUSH_TARGET, return_value=_push_ok(1, 0)) as mock_push: |
| 351 | result = runner.invoke(cli, _PUSH_ARGS + ["--kinds", "reservation"]) |
| 352 | assert result.exit_code == 0 |
| 353 | # Only 1 kind → batch should contain only reservations |
| 354 | batch_arg = mock_push.call_args[0][3] |
| 355 | assert all(r["kind"] == "reservation" for r in batch_arg) |
| 356 | |
| 357 | def test_push_coord_bus_error_json_failed_true(self, repo: pathlib.Path) -> None: |
| 358 | _write_local_reservation(repo) |
| 359 | with patch(_PUSH_TARGET, side_effect=CoordBusError("oops")): |
| 360 | result = runner.invoke(cli, _PUSH_ARGS + ["--json"]) |
| 361 | # exit_code 1; summary JSON (with failed=True) is the last JSON line |
| 362 | assert result.exit_code == 1 |
| 363 | json_lines = [ln for ln in result.output.splitlines() if ln.startswith("{")] |
| 364 | assert json_lines, "Expected at least one JSON output line" |
| 365 | # The summary JSON is last; the error JSON ({"error": ...}) may appear before it |
| 366 | summary = json.loads(json_lines[-1]) |
| 367 | assert summary["failed"] is True |
| 368 | |
| 369 | |
| 370 | # ── Integration: pull ───────────────────────────────────────────────────────── |
| 371 | |
| 372 | |
| 373 | class TestCoordSyncPullIntegration: |
| 374 | def test_pull_empty_result_exits_0(self, repo: pathlib.Path) -> None: |
| 375 | with patch(_PULL_TARGET, return_value=_pull_ok()): |
| 376 | result = runner.invoke(cli, _PULL_ARGS) |
| 377 | assert result.exit_code == 0 |
| 378 | assert "Pulled 0 new record(s)" in result.output |
| 379 | |
| 380 | def test_pull_with_records_writes_files(self, repo: pathlib.Path) -> None: |
| 381 | records = [{"kind": "reservation", "record_uuid": "r1", "payload": {"x": 1}}] |
| 382 | with patch(_PULL_TARGET, return_value=_pull_ok(records, cursor=1)): |
| 383 | result = runner.invoke(cli, _PULL_ARGS) |
| 384 | assert result.exit_code == 0 |
| 385 | target = repo / ".muse" / "coordination" / "remote" / "reservation" / "r1.json" |
| 386 | assert target.exists() |
| 387 | |
| 388 | def test_pull_text_output_contains_cursor(self, repo: pathlib.Path) -> None: |
| 389 | records = [{"kind": "reservation", "record_uuid": "r42", "payload": {}}] |
| 390 | with patch(_PULL_TARGET, return_value=_pull_ok(records, cursor=42)): |
| 391 | result = runner.invoke(cli, _PULL_ARGS) |
| 392 | assert result.exit_code == 0 |
| 393 | assert "cursor: 42" in result.output |
| 394 | |
| 395 | def test_pull_coord_bus_error_exits_1(self, repo: pathlib.Path) -> None: |
| 396 | with patch(_PULL_TARGET, side_effect=CoordBusError("connection refused")): |
| 397 | result = runner.invoke(cli, _PULL_ARGS) |
| 398 | assert result.exit_code == 1 |
| 399 | assert "connection refused" in result.output |
| 400 | |
| 401 | def test_pull_format_json_valid_structure(self, repo: pathlib.Path) -> None: |
| 402 | with patch(_PULL_TARGET, return_value=_pull_ok(cursor=7)): |
| 403 | result = runner.invoke(cli, _PULL_ARGS + ["--format", "json"]) |
| 404 | assert result.exit_code == 0 |
| 405 | data = json.loads(result.output.strip()) |
| 406 | assert "count" in data |
| 407 | assert "cursor" in data |
| 408 | assert "records" in data |
| 409 | |
| 410 | def test_pull_json_shorthand(self, repo: pathlib.Path) -> None: |
| 411 | with patch(_PULL_TARGET, return_value=_pull_ok(cursor=3)): |
| 412 | r1 = runner.invoke(cli, _PULL_ARGS + ["--format", "json"]) |
| 413 | r2 = runner.invoke(cli, _PULL_ARGS + ["--json"]) |
| 414 | assert json.loads(r1.output.strip()) == json.loads(r2.output.strip()) |
| 415 | |
| 416 | def test_pull_since_id_passed_to_pull_from_hub(self, repo: pathlib.Path) -> None: |
| 417 | with patch(_PULL_TARGET, return_value=_pull_ok()) as mock_pull: |
| 418 | runner.invoke(cli, _PULL_ARGS[:-2] + ["--since-id", "99"]) |
| 419 | assert mock_pull.called |
| 420 | call_args = mock_pull.call_args[0] |
| 421 | assert call_args[3] == 99 # since_id |
| 422 | |
| 423 | def test_pull_limit_passed_to_pull_from_hub(self, repo: pathlib.Path) -> None: |
| 424 | with patch(_PULL_TARGET, return_value=_pull_ok()) as mock_pull: |
| 425 | runner.invoke(cli, _PULL_ARGS + ["--limit", "42"]) |
| 426 | call_args = mock_pull.call_args[0] |
| 427 | assert call_args[5] == 42 # limit |
| 428 | |
| 429 | def test_pull_kinds_filter_passed_to_pull_from_hub(self, repo: pathlib.Path) -> None: |
| 430 | with patch(_PULL_TARGET, return_value=_pull_ok()) as mock_pull: |
| 431 | runner.invoke(cli, _PULL_ARGS + ["--kinds", "reservation", "heartbeat"]) |
| 432 | call_args = mock_pull.call_args[0] |
| 433 | assert "reservation" in call_args[4] |
| 434 | assert "heartbeat" in call_args[4] |
| 435 | |
| 436 | def test_pull_json_count_matches_records_length(self, repo: pathlib.Path) -> None: |
| 437 | records = [ |
| 438 | {"kind": "reservation", "record_uuid": f"r{i}", "payload": {}} |
| 439 | for i in range(5) |
| 440 | ] |
| 441 | with patch(_PULL_TARGET, return_value=_pull_ok(records, cursor=5)): |
| 442 | result = runner.invoke(cli, _PULL_ARGS + ["--json"]) |
| 443 | data = json.loads(result.output.strip()) |
| 444 | assert data["count"] == 5 |
| 445 | assert len(data["records"]) == 5 |
| 446 | |
| 447 | def test_pull_signing_passed_to_pull_from_hub(self, repo: pathlib.Path) -> None: |
| 448 | with patch(_PULL_TARGET, return_value=_pull_ok()) as mock_pull: |
| 449 | runner.invoke(cli, _PULL_ARGS) |
| 450 | call_args = mock_pull.call_args[0] |
| 451 | assert call_args[6] is None # signing (no identity configured in test) |
| 452 | |
| 453 | |
| 454 | # ── Security ────────────────────────────────────────────────────────────────── |
| 455 | |
| 456 | |
| 457 | class TestCoordSyncSecurity: |
| 458 | def test_push_traversal_owner_passed_as_string(self, repo: pathlib.Path) -> None: |
| 459 | """Path-traversal chars in owner are passed as-is to push_to_hub (encoding is push_to_hub's job).""" |
| 460 | _write_local_reservation(repo) |
| 461 | evil_owner = "../evil" |
| 462 | args = [ |
| 463 | "coord", "sync", "push", |
| 464 | "--hub", "http://localhost:10003", |
| 465 | "--owner", evil_owner, |
| 466 | "--slug", "myrepo", |
| 467 | |
| 468 | ] |
| 469 | with patch(_PUSH_TARGET, return_value=_push_ok()) as mock_push: |
| 470 | runner.invoke(cli, args) |
| 471 | if mock_push.called: |
| 472 | call_args = mock_push.call_args[0] |
| 473 | assert call_args[1] == evil_owner # owner string passed verbatim |
| 474 | |
| 475 | def test_token_not_in_output(self, repo: pathlib.Path) -> None: |
| 476 | _write_local_reservation(repo) |
| 477 | with patch(_PUSH_TARGET, return_value=_push_ok(1, 0)): |
| 478 | result = runner.invoke(cli, _PUSH_ARGS) |
| 479 | assert "tok" not in result.output |
| 480 | |
| 481 | def test_pull_token_not_in_output(self, repo: pathlib.Path) -> None: |
| 482 | with patch(_PULL_TARGET, return_value=_pull_ok()): |
| 483 | result = runner.invoke(cli, _PULL_ARGS) |
| 484 | assert "tok" not in result.output |
| 485 | |
| 486 | def test_write_remote_records_rejects_unknown_kind(self, repo: pathlib.Path) -> None: |
| 487 | """Server-supplied kind '../../evil' must not escape remote/ directory.""" |
| 488 | from muse.cli.commands.coord_sync import _write_remote_records |
| 489 | rec = {"kind": "../../evil", "record_uuid": "abc123", "payload": {}} |
| 490 | _write_remote_records(repo, [rec]) |
| 491 | remote_dir = repo / ".muse" / "coordination" / "remote" |
| 492 | assert not remote_dir.exists() or not any(remote_dir.rglob("*.json")) |
| 493 | |
| 494 | def test_write_remote_records_rejects_traversal_uuid(self, repo: pathlib.Path) -> None: |
| 495 | """Server-supplied record_uuid '../../../etc/passwd' must not escape kind dir.""" |
| 496 | from muse.cli.commands.coord_sync import _write_remote_records |
| 497 | rec = {"kind": "reservation", "record_uuid": "../../../etc/passwd", "payload": {}} |
| 498 | _write_remote_records(repo, [rec]) |
| 499 | remote_dir = repo / ".muse" / "coordination" / "remote" |
| 500 | assert not remote_dir.exists() or not any(remote_dir.rglob("*.json")) |
| 501 | |
| 502 | |
| 503 | # ── Unit: _gather_local_records — claim field name ───────────────────────────── |
| 504 | |
| 505 | |
| 506 | class TestGatherLocalRecordsClaims: |
| 507 | def test_claim_uses_claimer_run_id_field(self, repo: pathlib.Path) -> None: |
| 508 | """Bug fix: claims must read 'claimer_run_id', not 'claimed_by'.""" |
| 509 | from muse.cli.commands.coord_sync import _gather_local_records |
| 510 | tid = _write_local_claim(repo, run_id="correct-worker") |
| 511 | records = _gather_local_records(repo, kinds=["claim"]) |
| 512 | assert len(records) == 1 |
| 513 | assert records[0]["run_id"] == "correct-worker" |
| 514 | |
| 515 | def test_claim_record_uuid_is_task_id(self, repo: pathlib.Path) -> None: |
| 516 | from muse.cli.commands.coord_sync import _gather_local_records |
| 517 | tid = _write_local_claim(repo) |
| 518 | records = _gather_local_records(repo, kinds=["claim"]) |
| 519 | assert records[0]["record_uuid"] == tid |
| 520 | |
| 521 | def test_claim_expires_at_included(self, repo: pathlib.Path) -> None: |
| 522 | from muse.cli.commands.coord_sync import _gather_local_records |
| 523 | _write_local_claim(repo) |
| 524 | records = _gather_local_records(repo, kinds=["claim"]) |
| 525 | assert records[0]["expires_at"] is not None |
| 526 | |
| 527 | |
| 528 | # ── Unit: _write_remote_records — compact JSON ──────────────────────────────── |
| 529 | |
| 530 | |
| 531 | class TestWriteRemoteRecordsCompact: |
| 532 | def test_written_json_is_compact(self, repo: pathlib.Path) -> None: |
| 533 | """No indent=2 — remote files must be compact single-line JSON.""" |
| 534 | from muse.cli.commands.coord_sync import _write_remote_records |
| 535 | rec = {"kind": "reservation", "record_uuid": "compact-test", "payload": {"x": 1}} |
| 536 | _write_remote_records(repo, [rec]) |
| 537 | target = repo / ".muse" / "coordination" / "remote" / "reservation" / "compact-test.json" |
| 538 | raw = target.read_text().strip() |
| 539 | # Compact JSON has no interior newlines |
| 540 | assert "\n" not in raw |
| 541 | |
| 542 | def test_valid_kinds_all_accepted(self, repo: pathlib.Path) -> None: |
| 543 | """All 7 kinds are accepted by the allowlist.""" |
| 544 | from muse.cli.commands.coord_sync import _write_remote_records |
| 545 | records = [ |
| 546 | {"kind": k, "record_uuid": f"id-{i}", "payload": {}} |
| 547 | for i, k in enumerate(_ALL_KINDS) |
| 548 | ] |
| 549 | _write_remote_records(repo, records) |
| 550 | remote_dir = repo / ".muse" / "coordination" / "remote" |
| 551 | written = list(remote_dir.rglob("*.json")) |
| 552 | assert len(written) == len(_ALL_KINDS) |
| 553 | |
| 554 | def test_empty_uuid_still_skipped(self, repo: pathlib.Path) -> None: |
| 555 | from muse.cli.commands.coord_sync import _write_remote_records |
| 556 | rec = {"kind": "reservation", "record_uuid": "", "payload": {}} |
| 557 | _write_remote_records(repo, [rec]) |
| 558 | remote_dir = repo / ".muse" / "coordination" / "remote" |
| 559 | assert not remote_dir.exists() or not any(remote_dir.rglob("*.json")) |
| 560 | |
| 561 | def test_uuid_with_dots_rejected(self, repo: pathlib.Path) -> None: |
| 562 | """Dots in record_uuid are not allowed (could be used for traversal).""" |
| 563 | from muse.cli.commands.coord_sync import _write_remote_records |
| 564 | rec = {"kind": "reservation", "record_uuid": "..evil", "payload": {}} |
| 565 | _write_remote_records(repo, [rec]) |
| 566 | remote_dir = repo / ".muse" / "coordination" / "remote" |
| 567 | assert not remote_dir.exists() or not any(remote_dir.rglob("*.json")) |
| 568 | |
| 569 | def test_uuid_with_slash_rejected(self, repo: pathlib.Path) -> None: |
| 570 | from muse.cli.commands.coord_sync import _write_remote_records |
| 571 | rec = {"kind": "reservation", "record_uuid": "a/b", "payload": {}} |
| 572 | _write_remote_records(repo, [rec]) |
| 573 | remote_dir = repo / ".muse" / "coordination" / "remote" |
| 574 | assert not remote_dir.exists() or not any(remote_dir.rglob("*.json")) |
| 575 | |
| 576 | def test_uuid_too_long_rejected(self, repo: pathlib.Path) -> None: |
| 577 | """UUIDs over 128 chars are rejected.""" |
| 578 | from muse.cli.commands.coord_sync import _write_remote_records |
| 579 | rec = {"kind": "reservation", "record_uuid": "a" * 129, "payload": {}} |
| 580 | _write_remote_records(repo, [rec]) |
| 581 | remote_dir = repo / ".muse" / "coordination" / "remote" |
| 582 | assert not remote_dir.exists() or not any(remote_dir.rglob("*.json")) |
| 583 | |
| 584 | |
| 585 | # ── Input validation ────────────────────────────────────────────────────────── |
| 586 | |
| 587 | |
| 588 | class TestSyncInputValidation: |
| 589 | def _push_args(self, owner: str = "gabriel", slug: str = "myrepo", extra: list[str] | None = None) -> list[str]: |
| 590 | args = [ |
| 591 | "coord", "sync", "push", |
| 592 | "--hub", "http://localhost:10003", |
| 593 | "--owner", owner, |
| 594 | "--slug", slug, |
| 595 | |
| 596 | ] |
| 597 | if extra: |
| 598 | args.extend(extra) |
| 599 | return args |
| 600 | |
| 601 | def _pull_args(self, owner: str = "gabriel", slug: str = "myrepo", extra: list[str] | None = None) -> list[str]: |
| 602 | args = [ |
| 603 | "coord", "sync", "pull", |
| 604 | "--hub", "http://localhost:10003", |
| 605 | "--owner", owner, |
| 606 | "--slug", slug, |
| 607 | |
| 608 | ] |
| 609 | if extra: |
| 610 | args.extend(extra) |
| 611 | return args |
| 612 | |
| 613 | def test_push_owner_too_long_exits_1(self, repo: pathlib.Path) -> None: |
| 614 | owner = "x" * (_MAX_OWNER_LEN + 1) |
| 615 | result = runner.invoke(cli, self._push_args(owner=owner)) |
| 616 | assert result.exit_code == 1 |
| 617 | |
| 618 | def test_push_slug_too_long_exits_1(self, repo: pathlib.Path) -> None: |
| 619 | slug = "x" * (_MAX_SLUG_LEN + 1) |
| 620 | result = runner.invoke(cli, self._push_args(slug=slug)) |
| 621 | assert result.exit_code == 1 |
| 622 | |
| 623 | def test_push_owner_at_max_accepted(self, repo: pathlib.Path) -> None: |
| 624 | owner = "x" * _MAX_OWNER_LEN |
| 625 | with patch(_PUSH_TARGET, return_value=_push_ok()): |
| 626 | result = runner.invoke(cli, self._push_args(owner=owner)) |
| 627 | # No validation error — exits 0 (no records to push) |
| 628 | assert result.exit_code == 0 |
| 629 | |
| 630 | def test_push_slug_at_max_accepted(self, repo: pathlib.Path) -> None: |
| 631 | slug = "x" * _MAX_SLUG_LEN |
| 632 | with patch(_PUSH_TARGET, return_value=_push_ok()): |
| 633 | result = runner.invoke(cli, self._push_args(slug=slug)) |
| 634 | assert result.exit_code == 0 |
| 635 | |
| 636 | def test_push_owner_too_long_json_error(self, repo: pathlib.Path) -> None: |
| 637 | owner = "x" * (_MAX_OWNER_LEN + 1) |
| 638 | result = runner.invoke(cli, self._push_args(owner=owner) + ["--json"]) |
| 639 | assert result.exit_code == 1 |
| 640 | data = json.loads(result.output.strip()) |
| 641 | assert data["status"] == "bad_args" |
| 642 | |
| 643 | def test_pull_owner_too_long_exits_1(self, repo: pathlib.Path) -> None: |
| 644 | owner = "x" * (_MAX_OWNER_LEN + 1) |
| 645 | result = runner.invoke(cli, self._pull_args(owner=owner)) |
| 646 | assert result.exit_code == 1 |
| 647 | |
| 648 | def test_pull_slug_too_long_exits_1(self, repo: pathlib.Path) -> None: |
| 649 | slug = "x" * (_MAX_SLUG_LEN + 1) |
| 650 | result = runner.invoke(cli, self._pull_args(slug=slug)) |
| 651 | assert result.exit_code == 1 |
| 652 | |
| 653 | def test_pull_since_id_negative_exits_1(self, repo: pathlib.Path) -> None: |
| 654 | result = runner.invoke(cli, self._pull_args(extra=["--since-id", "-1"])) |
| 655 | assert result.exit_code == 1 |
| 656 | |
| 657 | def test_pull_since_id_negative_json_error(self, repo: pathlib.Path) -> None: |
| 658 | result = runner.invoke(cli, self._pull_args(extra=["--since-id", "-1", "--json"])) |
| 659 | assert result.exit_code == 1 |
| 660 | data = json.loads(result.output.strip()) |
| 661 | assert data["status"] == "bad_args" |
| 662 | |
| 663 | def test_pull_limit_zero_exits_1(self, repo: pathlib.Path) -> None: |
| 664 | result = runner.invoke(cli, self._pull_args(extra=["--limit", "0"])) |
| 665 | assert result.exit_code == 1 |
| 666 | |
| 667 | def test_pull_limit_over_max_exits_1(self, repo: pathlib.Path) -> None: |
| 668 | result = runner.invoke(cli, self._pull_args(extra=["--limit", str(_MAX_PULL_LIMIT + 1)])) |
| 669 | assert result.exit_code == 1 |
| 670 | |
| 671 | def test_pull_limit_at_min_accepted(self, repo: pathlib.Path) -> None: |
| 672 | with patch(_PULL_TARGET, return_value=_pull_ok()): |
| 673 | result = runner.invoke(cli, self._pull_args(extra=["--limit", "1"])) |
| 674 | assert result.exit_code == 0 |
| 675 | |
| 676 | def test_pull_limit_at_max_accepted(self, repo: pathlib.Path) -> None: |
| 677 | with patch(_PULL_TARGET, return_value=_pull_ok()): |
| 678 | result = runner.invoke(cli, self._pull_args(extra=["--limit", str(_MAX_PULL_LIMIT)])) |
| 679 | assert result.exit_code == 0 |
| 680 | |
| 681 | def test_pull_limit_over_max_json_error(self, repo: pathlib.Path) -> None: |
| 682 | result = runner.invoke( |
| 683 | cli, |
| 684 | self._pull_args(extra=["--limit", str(_MAX_PULL_LIMIT + 1), "--json"]), |
| 685 | ) |
| 686 | assert result.exit_code == 1 |
| 687 | data = json.loads(result.output.strip()) |
| 688 | assert data["status"] == "bad_args" |
| 689 | |
| 690 | def test_push_validation_fires_before_repo_lookup(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None: |
| 691 | """Validation must not touch filesystem — no repo needed.""" |
| 692 | monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path)) # no .muse dir |
| 693 | owner = "x" * (_MAX_OWNER_LEN + 1) |
| 694 | result = runner.invoke(cli, [ |
| 695 | "coord", "sync", "push", |
| 696 | "--hub", "http://localhost:10003", |
| 697 | "--owner", owner, |
| 698 | "--slug", "myrepo", |
| 699 | |
| 700 | ]) |
| 701 | assert result.exit_code == 1 |
| 702 | |
| 703 | def test_pull_validation_fires_before_repo_lookup(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None: |
| 704 | monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path)) # no .muse dir |
| 705 | result = runner.invoke(cli, [ |
| 706 | "coord", "sync", "pull", |
| 707 | "--hub", "http://localhost:10003", |
| 708 | "--owner", "gabriel", |
| 709 | "--slug", "myrepo", |
| 710 | |
| 711 | "--since-id", "-5", |
| 712 | ]) |
| 713 | assert result.exit_code == 1 |
| 714 | |
| 715 | |
| 716 | # ── JSON schema: schema_version and elapsed_seconds ────────────────────────── |
| 717 | |
| 718 | |
| 719 | class TestSyncJsonSchema: |
| 720 | def test_push_json_includes_schema_version(self, repo: pathlib.Path) -> None: |
| 721 | _write_local_reservation(repo) |
| 722 | with patch(_PUSH_TARGET, return_value=_push_ok(1, 0)): |
| 723 | result = runner.invoke(cli, _PUSH_ARGS + ["--json"]) |
| 724 | assert result.exit_code == 0 |
| 725 | data = json.loads(result.output.strip()) |
| 726 | assert "schema_version" in data |
| 727 | |
| 728 | def test_push_json_includes_elapsed_seconds(self, repo: pathlib.Path) -> None: |
| 729 | _write_local_reservation(repo) |
| 730 | with patch(_PUSH_TARGET, return_value=_push_ok(1, 0)): |
| 731 | result = runner.invoke(cli, _PUSH_ARGS + ["--json"]) |
| 732 | assert result.exit_code == 0 |
| 733 | data = json.loads(result.output.strip()) |
| 734 | assert "elapsed_seconds" in data |
| 735 | assert isinstance(data["elapsed_seconds"], float) |
| 736 | |
| 737 | def test_push_json_no_records_includes_elapsed(self, repo: pathlib.Path) -> None: |
| 738 | with patch(_PUSH_TARGET): |
| 739 | result = runner.invoke(cli, _PUSH_ARGS + ["--json"]) |
| 740 | assert result.exit_code == 0 |
| 741 | data = json.loads(result.output.strip()) |
| 742 | assert "elapsed_seconds" in data |
| 743 | |
| 744 | def test_pull_json_includes_schema_version(self, repo: pathlib.Path) -> None: |
| 745 | with patch(_PULL_TARGET, return_value=_pull_ok(cursor=3)): |
| 746 | result = runner.invoke(cli, _PULL_ARGS + ["--json"]) |
| 747 | assert result.exit_code == 0 |
| 748 | data = json.loads(result.output.strip()) |
| 749 | assert "schema_version" in data |
| 750 | |
| 751 | def test_pull_json_includes_elapsed_seconds(self, repo: pathlib.Path) -> None: |
| 752 | with patch(_PULL_TARGET, return_value=_pull_ok(cursor=3)): |
| 753 | result = runner.invoke(cli, _PULL_ARGS + ["--json"]) |
| 754 | assert result.exit_code == 0 |
| 755 | data = json.loads(result.output.strip()) |
| 756 | assert "elapsed_seconds" in data |
| 757 | assert isinstance(data["elapsed_seconds"], float) |
| 758 | |
| 759 | def test_push_text_includes_elapsed(self, repo: pathlib.Path) -> None: |
| 760 | _write_local_reservation(repo) |
| 761 | with patch(_PUSH_TARGET, return_value=_push_ok(1, 0)): |
| 762 | result = runner.invoke(cli, _PUSH_ARGS) |
| 763 | assert result.exit_code == 0 |
| 764 | # elapsed is formatted as (Ns) |
| 765 | assert "s)" in result.output |
| 766 | |
| 767 | def test_pull_text_includes_elapsed(self, repo: pathlib.Path) -> None: |
| 768 | with patch(_PULL_TARGET, return_value=_pull_ok(cursor=1)): |
| 769 | result = runner.invoke(cli, _PULL_ARGS) |
| 770 | assert result.exit_code == 0 |
| 771 | assert "s)" in result.output |
| 772 | |
| 773 | |
| 774 | # ── Gather all 7 kinds ──────────────────────────────────────────────────────── |
| 775 | |
| 776 | |
| 777 | class TestGatherAllKinds: |
| 778 | def _write_intent(self, repo: pathlib.Path) -> str: |
| 779 | d = repo / ".muse" / "coordination" / "intents" |
| 780 | d.mkdir(parents=True, exist_ok=True) |
| 781 | iid = str(uuid.uuid4()) |
| 782 | data = {"intent_id": iid, "run_id": "agent", "expires_at": None} |
| 783 | (d / f"{iid}.json").write_text(json.dumps(data)) |
| 784 | return iid |
| 785 | |
| 786 | def _write_release(self, repo: pathlib.Path) -> str: |
| 787 | d = repo / ".muse" / "coordination" / "releases" |
| 788 | d.mkdir(parents=True, exist_ok=True) |
| 789 | rid = str(uuid.uuid4()) |
| 790 | data = {"release_id": rid, "run_id": "agent"} |
| 791 | (d / f"{rid}.json").write_text(json.dumps(data)) |
| 792 | return rid |
| 793 | |
| 794 | def _write_dependency(self, repo: pathlib.Path) -> str: |
| 795 | d = repo / ".muse" / "coordination" / "dependencies" |
| 796 | d.mkdir(parents=True, exist_ok=True) |
| 797 | rid = str(uuid.uuid4()) |
| 798 | data = {"reservation_id": rid} |
| 799 | (d / f"{rid}.json").write_text(json.dumps(data)) |
| 800 | return rid |
| 801 | |
| 802 | def _write_task(self, repo: pathlib.Path) -> str: |
| 803 | d = repo / ".muse" / "coordination" / "tasks" |
| 804 | d.mkdir(parents=True, exist_ok=True) |
| 805 | tid = str(uuid.uuid4()) |
| 806 | data = {"task_id": tid, "run_id": "creator"} |
| 807 | (d / f"{tid}.json").write_text(json.dumps(data)) |
| 808 | return tid |
| 809 | |
| 810 | def test_all_kinds_gathered(self, repo: pathlib.Path) -> None: |
| 811 | from muse.cli.commands.coord_sync import _gather_local_records |
| 812 | _write_local_reservation(repo) |
| 813 | _write_local_heartbeat(repo, "hb-1") |
| 814 | self._write_intent(repo) |
| 815 | self._write_release(repo) |
| 816 | self._write_dependency(repo) |
| 817 | self._write_task(repo) |
| 818 | _write_local_claim(repo) |
| 819 | records = _gather_local_records(repo, kinds=list(_ALL_KINDS)) |
| 820 | kinds_found = {r["kind"] for r in records} |
| 821 | assert kinds_found == set(_ALL_KINDS) |
| 822 | |
| 823 | def test_each_kind_has_correct_record_uuid(self, repo: pathlib.Path) -> None: |
| 824 | from muse.cli.commands.coord_sync import _gather_local_records |
| 825 | rid = _write_local_reservation(repo) |
| 826 | records = _gather_local_records(repo, kinds=["reservation"]) |
| 827 | assert records[0]["record_uuid"] == rid |
| 828 | |
| 829 | def test_release_has_none_expires_at(self, repo: pathlib.Path) -> None: |
| 830 | from muse.cli.commands.coord_sync import _gather_local_records |
| 831 | self._write_release(repo) |
| 832 | records = _gather_local_records(repo, kinds=["release"]) |
| 833 | assert records[0]["expires_at"] is None |
| 834 | |
| 835 | def test_dependency_has_none_expires_at(self, repo: pathlib.Path) -> None: |
| 836 | from muse.cli.commands.coord_sync import _gather_local_records |
| 837 | self._write_dependency(repo) |
| 838 | records = _gather_local_records(repo, kinds=["dependency"]) |
| 839 | assert records[0]["expires_at"] is None |
| 840 | |
| 841 | def test_task_has_none_expires_at(self, repo: pathlib.Path) -> None: |
| 842 | from muse.cli.commands.coord_sync import _gather_local_records |
| 843 | self._write_task(repo) |
| 844 | records = _gather_local_records(repo, kinds=["task"]) |
| 845 | assert records[0]["expires_at"] is None |
| 846 | |
| 847 | |
| 848 | # ── Stress tests ────────────────────────────────────────────────────────────── |
| 849 | |
| 850 | |
| 851 | class TestSyncStress: |
| 852 | def test_push_600_records_batched(self, repo: pathlib.Path) -> None: |
| 853 | """600 records must be split across ≥ 2 batches of MAX_PUSH_BATCH.""" |
| 854 | from muse.core.coord_bus import MAX_PUSH_BATCH |
| 855 | for i in range(600): |
| 856 | _write_local_reservation(repo, run_id=f"agent-{i}") |
| 857 | call_count = 0 |
| 858 | |
| 859 | def fake_push(hub: str, owner: str, slug: str, batch: list[JsonDict], token: SigningIdentity | None) -> MsgpackDict: |
| 860 | nonlocal call_count |
| 861 | call_count += 1 |
| 862 | assert len(batch) <= MAX_PUSH_BATCH |
| 863 | return {"inserted": len(batch), "skipped": 0} |
| 864 | |
| 865 | with patch(_PUSH_TARGET, side_effect=fake_push): |
| 866 | result = runner.invoke(cli, _PUSH_ARGS + ["--kinds", "reservation"]) |
| 867 | assert result.exit_code == 0 |
| 868 | assert call_count >= 2 |
| 869 | |
| 870 | def test_push_600_records_inserted_count_correct(self, repo: pathlib.Path) -> None: |
| 871 | for i in range(600): |
| 872 | _write_local_reservation(repo, run_id=f"agent-{i}") |
| 873 | with patch(_PUSH_TARGET, return_value=_push_ok(inserted=1, skipped=0)) as mock: |
| 874 | result = runner.invoke(cli, _PUSH_ARGS + ["--kinds", "reservation", "--json"]) |
| 875 | assert result.exit_code == 0 |
| 876 | data = json.loads(result.output.strip()) |
| 877 | assert data["total"] == 600 |
| 878 | |
| 879 | def test_pull_1000_records_all_written(self, repo: pathlib.Path) -> None: |
| 880 | records = [ |
| 881 | {"kind": "reservation", "record_uuid": str(uuid.uuid4()), "payload": {}} |
| 882 | for _ in range(1000) |
| 883 | ] |
| 884 | with patch(_PULL_TARGET, return_value=_pull_ok(records, cursor=1000)): |
| 885 | result = runner.invoke(cli, _PULL_ARGS + ["--limit", "1000"]) |
| 886 | assert result.exit_code == 0 |
| 887 | remote_dir = repo / ".muse" / "coordination" / "remote" / "reservation" |
| 888 | written = list(remote_dir.glob("*.json")) |
| 889 | assert len(written) == 1000 |
| 890 | |
| 891 | def test_pull_mixed_invalid_records_skipped(self, repo: pathlib.Path) -> None: |
| 892 | """Records with invalid kind/uuid are skipped; valid ones still written.""" |
| 893 | good = {"kind": "reservation", "record_uuid": "good-uuid-1", "payload": {}} |
| 894 | bad_kind = {"kind": "../evil", "record_uuid": "evil-uuid", "payload": {}} |
| 895 | bad_uuid = {"kind": "reservation", "record_uuid": "../etc/passwd", "payload": {}} |
| 896 | with patch(_PULL_TARGET, return_value=_pull_ok([good, bad_kind, bad_uuid], cursor=3)): |
| 897 | result = runner.invoke(cli, _PULL_ARGS) |
| 898 | assert result.exit_code == 0 |
| 899 | good_file = repo / ".muse" / "coordination" / "remote" / "reservation" / "good-uuid-1.json" |
| 900 | assert good_file.exists() |
| 901 | # Only 1 file written (the good record) |
| 902 | remote_dir = repo / ".muse" / "coordination" / "remote" |
| 903 | all_files = list(remote_dir.rglob("*.json")) |
| 904 | assert len(all_files) == 1 |
| 905 | |
| 906 | def test_push_partial_failure_reports_failed_true(self, repo: pathlib.Path) -> None: |
| 907 | """If one batch fails, failed=True in JSON even if other batches succeed.""" |
| 908 | from muse.core.coord_bus import MAX_PUSH_BATCH |
| 909 | # Write enough for 2 batches |
| 910 | for i in range(MAX_PUSH_BATCH + 1): |
| 911 | _write_local_reservation(repo, run_id=f"agent-{i}") |
| 912 | |
| 913 | call_count = 0 |
| 914 | def sometimes_fail(hub: str, owner: str, slug: str, batch: list[JsonDict], token: SigningIdentity | None) -> MsgpackDict: |
| 915 | nonlocal call_count |
| 916 | call_count += 1 |
| 917 | if call_count == 1: |
| 918 | raise CoordBusError("first batch failed") |
| 919 | return {"inserted": len(batch), "skipped": 0} |
| 920 | |
| 921 | with patch(_PUSH_TARGET, side_effect=sometimes_fail): |
| 922 | result = runner.invoke( |
| 923 | cli, _PUSH_ARGS + ["--kinds", "reservation", "--json"] |
| 924 | ) |
| 925 | assert result.exit_code == 1 |
| 926 | # Find JSON line (error from first batch goes to stdout too in JSON mode) |
| 927 | json_lines = [ln for ln in result.output.splitlines() if ln.startswith("{")] |
| 928 | final = json.loads(json_lines[-1]) |
| 929 | assert final["failed"] is True |
| 930 | |
| 931 | |
| 932 | # --------------------------------------------------------------------------- |
| 933 | # Extended — muse coord sync push |
| 934 | # --------------------------------------------------------------------------- |
| 935 | |
| 936 | |
| 937 | class TestCoordSyncPushExtended: |
| 938 | def test_j_alias_works(self, repo: pathlib.Path) -> None: |
| 939 | """-j is equivalent to --json for push.""" |
| 940 | with patch(_PUSH_TARGET, return_value=_push_ok(0, 0)): |
| 941 | result = runner.invoke(cli, _PUSH_ARGS + ["-j"]) |
| 942 | assert result.exit_code == 0, result.output |
| 943 | data = json.loads(result.output.strip()) |
| 944 | assert "inserted" in data |
| 945 | |
| 946 | def test_help_flag(self, repo: pathlib.Path) -> None: |
| 947 | result = runner.invoke(cli, ["coord", "sync", "push", "--help"]) |
| 948 | assert result.exit_code == 0 |
| 949 | |
| 950 | def test_json_compact_single_line(self, repo: pathlib.Path) -> None: |
| 951 | """JSON output is a single compact line — no indent=2.""" |
| 952 | _write_local_reservation(repo) |
| 953 | with patch(_PUSH_TARGET, return_value=_push_ok(1, 0)): |
| 954 | result = runner.invoke(cli, _PUSH_ARGS + ["-j"]) |
| 955 | assert result.exit_code == 0 |
| 956 | lines = [l for l in result.output.splitlines() if l.strip().startswith("{")] |
| 957 | assert len(lines) == 1, f"Expected compact JSON, got: {result.output!r}" |
| 958 | |
| 959 | def test_json_all_required_fields(self, repo: pathlib.Path) -> None: |
| 960 | """JSON always has schema_version, inserted, skipped, total, failed, elapsed_seconds.""" |
| 961 | _write_local_reservation(repo) |
| 962 | with patch(_PUSH_TARGET, return_value=_push_ok(1, 0)): |
| 963 | result = runner.invoke(cli, _PUSH_ARGS + ["-j"]) |
| 964 | data = json.loads(result.output.strip()) |
| 965 | for field in ("schema_version", "inserted", "skipped", "total", "failed", "elapsed_seconds"): |
| 966 | assert field in data, f"Missing field: {field}" |
| 967 | |
| 968 | def test_json_inserted_is_int(self, repo: pathlib.Path) -> None: |
| 969 | _write_local_reservation(repo) |
| 970 | with patch(_PUSH_TARGET, return_value=_push_ok(1, 0)): |
| 971 | result = runner.invoke(cli, _PUSH_ARGS + ["-j"]) |
| 972 | data = json.loads(result.output.strip()) |
| 973 | assert isinstance(data["inserted"], int) |
| 974 | assert isinstance(data["skipped"], int) |
| 975 | assert isinstance(data["total"], int) |
| 976 | |
| 977 | def test_json_failed_is_bool(self, repo: pathlib.Path) -> None: |
| 978 | with patch(_PUSH_TARGET, return_value=_push_ok(0, 0)): |
| 979 | result = runner.invoke(cli, _PUSH_ARGS + ["-j"]) |
| 980 | data = json.loads(result.output.strip()) |
| 981 | assert isinstance(data["failed"], bool) |
| 982 | assert data["failed"] is False |
| 983 | |
| 984 | def test_json_elapsed_seconds_is_number(self, repo: pathlib.Path) -> None: |
| 985 | with patch(_PUSH_TARGET, return_value=_push_ok(0, 0)): |
| 986 | result = runner.invoke(cli, _PUSH_ARGS + ["-j"]) |
| 987 | data = json.loads(result.output.strip()) |
| 988 | assert isinstance(data["elapsed_seconds"], (int, float)) |
| 989 | assert data["elapsed_seconds"] >= 0 |
| 990 | |
| 991 | def test_json_schema_version_is_string(self, repo: pathlib.Path) -> None: |
| 992 | with patch(_PUSH_TARGET, return_value=_push_ok(0, 0)): |
| 993 | result = runner.invoke(cli, _PUSH_ARGS + ["-j"]) |
| 994 | data = json.loads(result.output.strip()) |
| 995 | assert isinstance(data["schema_version"], str) |
| 996 | assert len(data["schema_version"]) > 0 |
| 997 | |
| 998 | def test_json_total_matches_gathered_records(self, repo: pathlib.Path) -> None: |
| 999 | _write_local_reservation(repo) |
| 1000 | _write_local_reservation(repo) |
| 1001 | with patch(_PUSH_TARGET, return_value=_push_ok(2, 0)): |
| 1002 | result = runner.invoke(cli, _PUSH_ARGS + ["-j"]) |
| 1003 | data = json.loads(result.output.strip()) |
| 1004 | assert data["total"] == 2 |
| 1005 | |
| 1006 | def test_json_no_records_total_is_zero(self, repo: pathlib.Path) -> None: |
| 1007 | result = runner.invoke(cli, _PUSH_ARGS + ["-j"]) |
| 1008 | assert result.exit_code == 0 |
| 1009 | data = json.loads(result.output.strip()) |
| 1010 | assert data["total"] == 0 |
| 1011 | assert data["inserted"] == 0 |
| 1012 | assert data["skipped"] == 0 |
| 1013 | |
| 1014 | def test_idempotent_second_push_all_skipped(self, repo: pathlib.Path) -> None: |
| 1015 | """Second push: all records skipped (hub already has them).""" |
| 1016 | _write_local_reservation(repo) |
| 1017 | with patch(_PUSH_TARGET, return_value=_push_ok(0, 1)): |
| 1018 | result = runner.invoke(cli, _PUSH_ARGS + ["-j"]) |
| 1019 | data = json.loads(result.output.strip()) |
| 1020 | assert data["skipped"] == 1 |
| 1021 | assert data["inserted"] == 0 |
| 1022 | |
| 1023 | def test_all_7_kinds_pushed(self, repo: pathlib.Path) -> None: |
| 1024 | """Push with all 7 kinds gathers records from every kind directory.""" |
| 1025 | coord_dir = repo / ".muse" / "coordination" |
| 1026 | kind_dirs = { |
| 1027 | "reservations": ("reservation_id", "run_id"), |
| 1028 | "heartbeats": ("run_id", "run_id"), |
| 1029 | "intents": ("intent_id", "run_id"), |
| 1030 | "releases": ("release_id", "run_id"), |
| 1031 | "dependencies": ("reservation_id", "reservation_id"), |
| 1032 | "tasks": ("task_id", "run_id"), |
| 1033 | "claims": ("task_id", "claimer_run_id"), |
| 1034 | } |
| 1035 | import uuid as _uuid |
| 1036 | for subdir, (id_field, run_field) in kind_dirs.items(): |
| 1037 | d = coord_dir / subdir |
| 1038 | d.mkdir(parents=True, exist_ok=True) |
| 1039 | rid = str(_uuid.uuid4()) |
| 1040 | (d / f"{rid}.json").write_text(json.dumps({id_field: rid, run_field: "r1"})) |
| 1041 | |
| 1042 | with patch(_PUSH_TARGET, return_value=_push_ok(7, 0)) as mock_push: |
| 1043 | result = runner.invoke(cli, _PUSH_ARGS + ["-j"]) |
| 1044 | assert result.exit_code == 0 |
| 1045 | data = json.loads(result.output.strip()) |
| 1046 | assert data["total"] == 7 |
| 1047 | |
| 1048 | def test_text_output_shows_owner_slug(self, repo: pathlib.Path) -> None: |
| 1049 | _write_local_reservation(repo) |
| 1050 | with patch(_PUSH_TARGET, return_value=_push_ok(1, 0)): |
| 1051 | result = runner.invoke(cli, _PUSH_ARGS) |
| 1052 | assert "gabriel" in result.output |
| 1053 | assert "myrepo" in result.output |
| 1054 | |
| 1055 | def test_text_output_shows_checkmark_on_success(self, repo: pathlib.Path) -> None: |
| 1056 | _write_local_reservation(repo) |
| 1057 | with patch(_PUSH_TARGET, return_value=_push_ok(1, 0)): |
| 1058 | result = runner.invoke(cli, _PUSH_ARGS) |
| 1059 | assert "✅" in result.output |
| 1060 | |
| 1061 | def test_text_output_shows_cross_on_failure(self, repo: pathlib.Path) -> None: |
| 1062 | _write_local_reservation(repo) |
| 1063 | with patch(_PUSH_TARGET, side_effect=CoordBusError("hub down")): |
| 1064 | result = runner.invoke(cli, _PUSH_ARGS) |
| 1065 | assert "❌" in result.output or result.exit_code == 1 |
| 1066 | |
| 1067 | def test_help_shows_agent_quickstart(self, repo: pathlib.Path) -> None: |
| 1068 | result = runner.invoke(cli, ["coord", "sync", "push", "--help"]) |
| 1069 | assert "Agent quickstart" in result.output |
| 1070 | |
| 1071 | def test_help_shows_json_schema(self, repo: pathlib.Path) -> None: |
| 1072 | result = runner.invoke(cli, ["coord", "sync", "push", "--help"]) |
| 1073 | assert "JSON output schema" in result.output |
| 1074 | |
| 1075 | def test_help_shows_exit_codes(self, repo: pathlib.Path) -> None: |
| 1076 | result = runner.invoke(cli, ["coord", "sync", "push", "--help"]) |
| 1077 | assert "Exit codes" in result.output |
| 1078 | |
| 1079 | |
| 1080 | # --------------------------------------------------------------------------- |
| 1081 | # Security — muse coord sync push |
| 1082 | # --------------------------------------------------------------------------- |
| 1083 | |
| 1084 | |
| 1085 | class TestCoordSyncPushSecurity: |
| 1086 | def test_ansi_in_owner_sanitized_in_text_output(self, repo: pathlib.Path) -> None: |
| 1087 | """ANSI codes in --owner must not bleed into text output.""" |
| 1088 | _write_local_reservation(repo) |
| 1089 | ansi_owner = "\x1b[31mevil\x1b[0m" |
| 1090 | push_args = [ |
| 1091 | "coord", "sync", "push", |
| 1092 | "--hub", "http://localhost:10003", |
| 1093 | "--owner", ansi_owner, |
| 1094 | "--slug", "myrepo", |
| 1095 | |
| 1096 | ] |
| 1097 | with patch(_PUSH_TARGET, return_value=_push_ok(1, 0)): |
| 1098 | result = runner.invoke(cli, push_args) |
| 1099 | assert "\x1b" not in result.output |
| 1100 | |
| 1101 | def test_ansi_in_slug_sanitized_in_text_output(self, repo: pathlib.Path) -> None: |
| 1102 | _write_local_reservation(repo) |
| 1103 | ansi_slug = "\x1b[32minjected\x1b[0m" |
| 1104 | push_args = [ |
| 1105 | "coord", "sync", "push", |
| 1106 | "--hub", "http://localhost:10003", |
| 1107 | "--owner", "gabriel", |
| 1108 | "--slug", ansi_slug, |
| 1109 | |
| 1110 | ] |
| 1111 | with patch(_PUSH_TARGET, return_value=_push_ok(1, 0)): |
| 1112 | result = runner.invoke(cli, push_args) |
| 1113 | assert "\x1b" not in result.output |
| 1114 | |
| 1115 | def test_token_not_in_json_output(self, repo: pathlib.Path) -> None: |
| 1116 | """Auth token must never appear in JSON output.""" |
| 1117 | _write_local_reservation(repo) |
| 1118 | secret = "super-secret-token-xyz" |
| 1119 | push_args = [ |
| 1120 | "coord", "sync", "push", |
| 1121 | "--hub", "http://localhost:10003", |
| 1122 | "--owner", "gabriel", |
| 1123 | "--slug", "myrepo", |
| 1124 | |
| 1125 | "--json", |
| 1126 | ] |
| 1127 | with patch(_PUSH_TARGET, return_value=_push_ok(1, 0)): |
| 1128 | result = runner.invoke(cli, push_args) |
| 1129 | assert secret not in result.output |
| 1130 | |
| 1131 | def test_token_not_in_text_output(self, repo: pathlib.Path) -> None: |
| 1132 | _write_local_reservation(repo) |
| 1133 | secret = "super-secret-token-abc" |
| 1134 | push_args = [ |
| 1135 | "coord", "sync", "push", |
| 1136 | "--hub", "http://localhost:10003", |
| 1137 | "--owner", "gabriel", |
| 1138 | "--slug", "myrepo", |
| 1139 | |
| 1140 | ] |
| 1141 | with patch(_PUSH_TARGET, return_value=_push_ok(1, 0)): |
| 1142 | result = runner.invoke(cli, push_args) |
| 1143 | assert secret not in result.output |
| 1144 | |
| 1145 | def test_no_traceback_on_coord_bus_error(self, repo: pathlib.Path) -> None: |
| 1146 | _write_local_reservation(repo) |
| 1147 | with patch(_PUSH_TARGET, side_effect=CoordBusError("network failure")): |
| 1148 | result = runner.invoke(cli, _PUSH_ARGS) |
| 1149 | assert "Traceback" not in result.output |
| 1150 | |
| 1151 | def test_owner_length_cap_before_io(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None: |
| 1152 | """Owner length check fires before any file system access.""" |
| 1153 | monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path)) |
| 1154 | long_owner = "x" * (_MAX_OWNER_LEN + 1) |
| 1155 | push_args = [ |
| 1156 | "coord", "sync", "push", |
| 1157 | "--hub", "http://localhost:10003", |
| 1158 | "--owner", long_owner, |
| 1159 | "--slug", "myrepo", |
| 1160 | |
| 1161 | ] |
| 1162 | result = runner.invoke(cli, push_args) |
| 1163 | assert result.exit_code == 1 |
| 1164 | assert "Traceback" not in result.output |
| 1165 | |
| 1166 | |
| 1167 | # --------------------------------------------------------------------------- |
| 1168 | # Stress — muse coord sync push |
| 1169 | # --------------------------------------------------------------------------- |
| 1170 | |
| 1171 | |
| 1172 | class TestCoordSyncPushStress: |
| 1173 | def test_50_sequential_push_calls_no_records(self, repo: pathlib.Path) -> None: |
| 1174 | """50 sequential pushes with no records all exit 0.""" |
| 1175 | for i in range(50): |
| 1176 | result = runner.invoke(cli, _PUSH_ARGS + ["-j"]) |
| 1177 | assert result.exit_code == 0, f"Call {i}: {result.output}" |
| 1178 | data = json.loads(result.output.strip()) |
| 1179 | assert data["total"] == 0 |
| 1180 | |
| 1181 | def test_push_1200_records_correct_batch_count(self, repo: pathlib.Path) -> None: |
| 1182 | """1200 records → ceil(1200/500) = 3 batches.""" |
| 1183 | from muse.core.coord_bus import MAX_PUSH_BATCH |
| 1184 | for i in range(1200): |
| 1185 | _write_local_reservation(repo, run_id=f"agent-{i}") |
| 1186 | call_count = 0 |
| 1187 | def counting_push(hub: str, owner: str, slug: str, batch: list[JsonDict], token: SigningIdentity | None) -> MsgpackDict: |
| 1188 | nonlocal call_count |
| 1189 | call_count += 1 |
| 1190 | return {"inserted": len(batch), "skipped": 0} |
| 1191 | with patch(_PUSH_TARGET, side_effect=counting_push): |
| 1192 | result = runner.invoke(cli, _PUSH_ARGS + ["--kinds", "reservation", "-j"]) |
| 1193 | assert result.exit_code == 0 |
| 1194 | expected_batches = -(-1200 // MAX_PUSH_BATCH) # ceil division |
| 1195 | assert call_count == expected_batches |
| 1196 | data = json.loads(result.output.strip()) |
| 1197 | assert data["inserted"] == 1200 |
| 1198 | |
| 1199 | def test_concurrent_push_8_threads(self, repo: pathlib.Path) -> None: |
| 1200 | """8 threads each call run_push directly; patches applied at test level. |
| 1201 | |
| 1202 | The goal is to verify that concurrent calls to run_push do not crash |
| 1203 | or corrupt internal state. All threads share the same repo fixture; |
| 1204 | per-thread module mutation is intentionally avoided here because |
| 1205 | unguarded write-then-restore of a module attribute across threads is a |
| 1206 | race condition that can leave the module permanently patched after the |
| 1207 | test completes, polluting later tests. |
| 1208 | """ |
| 1209 | import argparse |
| 1210 | import threading |
| 1211 | |
| 1212 | from muse.cli.commands.coord_sync import run_push |
| 1213 | |
| 1214 | _write_local_reservation(repo, run_id="shared-agent") |
| 1215 | |
| 1216 | errors: list[str] = [] |
| 1217 | |
| 1218 | def worker(idx: int) -> None: |
| 1219 | args = argparse.Namespace( |
| 1220 | hub="http://localhost:10003", |
| 1221 | owner="gabriel", |
| 1222 | slug="myrepo", |
| 1223 | signing=None, |
| 1224 | kinds=list(_ALL_KINDS), |
| 1225 | fmt="json", |
| 1226 | ) |
| 1227 | try: |
| 1228 | run_push(args) |
| 1229 | except SystemExit as exc: |
| 1230 | if exc.code != 0: |
| 1231 | errors.append(f"Thread {idx}: exit {exc.code}") |
| 1232 | except Exception as exc: |
| 1233 | errors.append(f"Thread {idx}: {exc}") |
| 1234 | |
| 1235 | def fake_push(hub: str, owner: str, slug: str, batch: list[JsonDict], token: SigningIdentity | None) -> MsgpackDict: |
| 1236 | return {"inserted": len(batch), "skipped": 0} |
| 1237 | |
| 1238 | push_p = patch(_PUSH_TARGET, side_effect=fake_push) |
| 1239 | repo_p = patch("muse.cli.commands.coord_sync.require_repo", return_value=repo) |
| 1240 | push_p.start() |
| 1241 | repo_p.start() |
| 1242 | try: |
| 1243 | threads = [threading.Thread(target=worker, args=(i,)) for i in range(8)] |
| 1244 | for t in threads: |
| 1245 | t.start() |
| 1246 | for t in threads: |
| 1247 | t.join() |
| 1248 | finally: |
| 1249 | push_p.stop() |
| 1250 | repo_p.stop() |
| 1251 | assert not errors, f"Concurrent failures: {errors}" |
| 1252 | |
| 1253 | |
| 1254 | # --------------------------------------------------------------------------- |
| 1255 | # Extended — muse coord sync pull |
| 1256 | # --------------------------------------------------------------------------- |
| 1257 | |
| 1258 | |
| 1259 | class TestCoordSyncPullExtended: |
| 1260 | def test_j_alias_works(self, repo: pathlib.Path) -> None: |
| 1261 | """-j is equivalent to --json for pull.""" |
| 1262 | with patch(_PULL_TARGET, return_value=_pull_ok()): |
| 1263 | result = runner.invoke(cli, _PULL_ARGS + ["-j"]) |
| 1264 | assert result.exit_code == 0, result.output |
| 1265 | data = json.loads(result.output.strip()) |
| 1266 | assert "count" in data |
| 1267 | |
| 1268 | def test_help_flag(self, repo: pathlib.Path) -> None: |
| 1269 | result = runner.invoke(cli, ["coord", "sync", "pull", "--help"]) |
| 1270 | assert result.exit_code == 0 |
| 1271 | |
| 1272 | def test_json_compact_single_line(self, repo: pathlib.Path) -> None: |
| 1273 | """JSON output is a single compact line — no indent=2.""" |
| 1274 | with patch(_PULL_TARGET, return_value=_pull_ok()): |
| 1275 | result = runner.invoke(cli, _PULL_ARGS + ["-j"]) |
| 1276 | assert result.exit_code == 0 |
| 1277 | lines = [l for l in result.output.splitlines() if l.strip().startswith("{")] |
| 1278 | assert len(lines) == 1, f"Expected compact JSON, got: {result.output!r}" |
| 1279 | |
| 1280 | def test_json_all_required_fields(self, repo: pathlib.Path) -> None: |
| 1281 | """JSON always has schema_version, count, cursor, records, elapsed_seconds.""" |
| 1282 | with patch(_PULL_TARGET, return_value=_pull_ok()): |
| 1283 | result = runner.invoke(cli, _PULL_ARGS + ["-j"]) |
| 1284 | data = json.loads(result.output.strip()) |
| 1285 | for field in ("schema_version", "count", "cursor", "records", "elapsed_seconds"): |
| 1286 | assert field in data, f"Missing field: {field}" |
| 1287 | |
| 1288 | def test_json_count_is_int(self, repo: pathlib.Path) -> None: |
| 1289 | with patch(_PULL_TARGET, return_value=_pull_ok()): |
| 1290 | result = runner.invoke(cli, _PULL_ARGS + ["-j"]) |
| 1291 | data = json.loads(result.output.strip()) |
| 1292 | assert isinstance(data["count"], int) |
| 1293 | assert isinstance(data["cursor"], int) |
| 1294 | |
| 1295 | def test_json_records_is_list(self, repo: pathlib.Path) -> None: |
| 1296 | with patch(_PULL_TARGET, return_value=_pull_ok()): |
| 1297 | result = runner.invoke(cli, _PULL_ARGS + ["-j"]) |
| 1298 | data = json.loads(result.output.strip()) |
| 1299 | assert isinstance(data["records"], list) |
| 1300 | |
| 1301 | def test_json_elapsed_seconds_is_number(self, repo: pathlib.Path) -> None: |
| 1302 | with patch(_PULL_TARGET, return_value=_pull_ok()): |
| 1303 | result = runner.invoke(cli, _PULL_ARGS + ["-j"]) |
| 1304 | data = json.loads(result.output.strip()) |
| 1305 | assert isinstance(data["elapsed_seconds"], (int, float)) |
| 1306 | assert data["elapsed_seconds"] >= 0 |
| 1307 | |
| 1308 | def test_json_schema_version_is_string(self, repo: pathlib.Path) -> None: |
| 1309 | with patch(_PULL_TARGET, return_value=_pull_ok()): |
| 1310 | result = runner.invoke(cli, _PULL_ARGS + ["-j"]) |
| 1311 | data = json.loads(result.output.strip()) |
| 1312 | assert isinstance(data["schema_version"], str) |
| 1313 | assert len(data["schema_version"]) > 0 |
| 1314 | |
| 1315 | def test_json_count_matches_records_length(self, repo: pathlib.Path) -> None: |
| 1316 | fake_records = [ |
| 1317 | {"kind": "reservation", "record_uuid": str(uuid.uuid4()), "run_id": "r1", "payload": {}}, |
| 1318 | {"kind": "reservation", "record_uuid": str(uuid.uuid4()), "run_id": "r2", "payload": {}}, |
| 1319 | ] |
| 1320 | with patch(_PULL_TARGET, return_value=_pull_ok(fake_records, cursor=2)): |
| 1321 | result = runner.invoke(cli, _PULL_ARGS + ["-j"]) |
| 1322 | data = json.loads(result.output.strip()) |
| 1323 | assert data["count"] == 2 |
| 1324 | assert len(data["records"]) == 2 |
| 1325 | |
| 1326 | def test_json_cursor_reflects_hub_cursor(self, repo: pathlib.Path) -> None: |
| 1327 | with patch(_PULL_TARGET, return_value=_pull_ok([], cursor=42)): |
| 1328 | result = runner.invoke(cli, _PULL_ARGS + ["-j"]) |
| 1329 | data = json.loads(result.output.strip()) |
| 1330 | assert data["cursor"] == 42 |
| 1331 | |
| 1332 | def test_zero_records_exits_0(self, repo: pathlib.Path) -> None: |
| 1333 | """0 records returned is a valid success.""" |
| 1334 | with patch(_PULL_TARGET, return_value=_pull_ok()): |
| 1335 | result = runner.invoke(cli, _PULL_ARGS + ["-j"]) |
| 1336 | assert result.exit_code == 0 |
| 1337 | data = json.loads(result.output.strip()) |
| 1338 | assert data["count"] == 0 |
| 1339 | |
| 1340 | def test_incremental_pull_since_id_forwarded(self, repo: pathlib.Path) -> None: |
| 1341 | """--since-id is passed through to pull_from_hub.""" |
| 1342 | with patch(_PULL_TARGET, return_value=_pull_ok()) as mock_pull: |
| 1343 | runner.invoke(cli, _PULL_ARGS + ["--since-id", "99"]) |
| 1344 | _, kwargs = mock_pull.call_args |
| 1345 | assert mock_pull.call_args[0][3] == 99 or kwargs.get("since_id") == 99 or mock_pull.call_args[0][3] == 99 |
| 1346 | |
| 1347 | def test_records_written_to_remote_dir(self, repo: pathlib.Path) -> None: |
| 1348 | """Records returned by hub are written to .muse/coordination/remote/.""" |
| 1349 | rid = str(uuid.uuid4()) |
| 1350 | fake_records = [{"kind": "reservation", "record_uuid": rid, "run_id": "r1", "payload": {}}] |
| 1351 | with patch(_PULL_TARGET, return_value=_pull_ok(fake_records)): |
| 1352 | result = runner.invoke(cli, _PULL_ARGS) |
| 1353 | assert result.exit_code == 0 |
| 1354 | written = repo / ".muse" / "coordination" / "remote" / "reservation" / f"{rid}.json" |
| 1355 | assert written.exists() |
| 1356 | |
| 1357 | def test_text_output_shows_owner_slug(self, repo: pathlib.Path) -> None: |
| 1358 | with patch(_PULL_TARGET, return_value=_pull_ok()): |
| 1359 | result = runner.invoke(cli, _PULL_ARGS) |
| 1360 | assert "gabriel" in result.output |
| 1361 | assert "myrepo" in result.output |
| 1362 | |
| 1363 | def test_text_output_shows_cursor(self, repo: pathlib.Path) -> None: |
| 1364 | with patch(_PULL_TARGET, return_value=_pull_ok([], cursor=7)): |
| 1365 | result = runner.invoke(cli, _PULL_ARGS) |
| 1366 | assert "7" in result.output |
| 1367 | |
| 1368 | def test_text_output_shows_remote_path_when_records(self, repo: pathlib.Path) -> None: |
| 1369 | rid = str(uuid.uuid4()) |
| 1370 | fake_records = [{"kind": "reservation", "record_uuid": rid, "run_id": "r1", "payload": {}}] |
| 1371 | with patch(_PULL_TARGET, return_value=_pull_ok(fake_records)): |
| 1372 | result = runner.invoke(cli, _PULL_ARGS) |
| 1373 | assert "remote" in result.output.lower() |
| 1374 | |
| 1375 | def test_help_shows_agent_quickstart(self, repo: pathlib.Path) -> None: |
| 1376 | result = runner.invoke(cli, ["coord", "sync", "pull", "--help"]) |
| 1377 | assert "Agent quickstart" in result.output |
| 1378 | |
| 1379 | def test_help_shows_json_schema(self, repo: pathlib.Path) -> None: |
| 1380 | result = runner.invoke(cli, ["coord", "sync", "pull", "--help"]) |
| 1381 | assert "JSON output schema" in result.output |
| 1382 | |
| 1383 | def test_help_shows_exit_codes(self, repo: pathlib.Path) -> None: |
| 1384 | result = runner.invoke(cli, ["coord", "sync", "pull", "--help"]) |
| 1385 | assert "Exit codes" in result.output |
| 1386 | |
| 1387 | |
| 1388 | # --------------------------------------------------------------------------- |
| 1389 | # Security — muse coord sync pull |
| 1390 | # --------------------------------------------------------------------------- |
| 1391 | |
| 1392 | |
| 1393 | class TestCoordSyncPullSecurity: |
| 1394 | def test_ansi_in_owner_sanitized_in_text_output(self, repo: pathlib.Path) -> None: |
| 1395 | """ANSI codes in --owner must not bleed into text output.""" |
| 1396 | ansi_owner = "\x1b[31mevil\x1b[0m" |
| 1397 | pull_args = [ |
| 1398 | "coord", "sync", "pull", |
| 1399 | "--hub", "http://localhost:10003", |
| 1400 | "--owner", ansi_owner, |
| 1401 | "--slug", "myrepo", |
| 1402 | |
| 1403 | "--since-id", "0", |
| 1404 | ] |
| 1405 | with patch(_PULL_TARGET, return_value=_pull_ok()): |
| 1406 | result = runner.invoke(cli, pull_args) |
| 1407 | assert "\x1b" not in result.output |
| 1408 | |
| 1409 | def test_ansi_in_slug_sanitized_in_text_output(self, repo: pathlib.Path) -> None: |
| 1410 | ansi_slug = "\x1b[32minjected\x1b[0m" |
| 1411 | pull_args = [ |
| 1412 | "coord", "sync", "pull", |
| 1413 | "--hub", "http://localhost:10003", |
| 1414 | "--owner", "gabriel", |
| 1415 | "--slug", ansi_slug, |
| 1416 | |
| 1417 | "--since-id", "0", |
| 1418 | ] |
| 1419 | with patch(_PULL_TARGET, return_value=_pull_ok()): |
| 1420 | result = runner.invoke(cli, pull_args) |
| 1421 | assert "\x1b" not in result.output |
| 1422 | |
| 1423 | def test_token_not_in_json_output(self, repo: pathlib.Path) -> None: |
| 1424 | """Auth token must never appear in JSON output.""" |
| 1425 | secret = "super-secret-pull-token" |
| 1426 | pull_args = [ |
| 1427 | "coord", "sync", "pull", |
| 1428 | "--hub", "http://localhost:10003", |
| 1429 | "--owner", "gabriel", |
| 1430 | "--slug", "myrepo", |
| 1431 | |
| 1432 | "--since-id", "0", |
| 1433 | "--json", |
| 1434 | ] |
| 1435 | with patch(_PULL_TARGET, return_value=_pull_ok()): |
| 1436 | result = runner.invoke(cli, pull_args) |
| 1437 | assert secret not in result.output |
| 1438 | |
| 1439 | def test_no_traceback_on_coord_bus_error(self, repo: pathlib.Path) -> None: |
| 1440 | with patch(_PULL_TARGET, side_effect=CoordBusError("timeout")): |
| 1441 | result = runner.invoke(cli, _PULL_ARGS) |
| 1442 | assert "Traceback" not in result.output |
| 1443 | |
| 1444 | def test_remote_records_with_traversal_uuid_skipped(self, repo: pathlib.Path) -> None: |
| 1445 | """A record with path-traversal UUID must not escape remote/.""" |
| 1446 | evil_records = [ |
| 1447 | {"kind": "reservation", "record_uuid": "../../evil", "run_id": "r1", "payload": {}}, |
| 1448 | ] |
| 1449 | with patch(_PULL_TARGET, return_value=_pull_ok(evil_records)): |
| 1450 | result = runner.invoke(cli, _PULL_ARGS) |
| 1451 | assert result.exit_code == 0 |
| 1452 | evil_path = repo / ".muse" / "coordination" / "remote" / "reservation" / "../../evil.json" |
| 1453 | assert not evil_path.exists() |
| 1454 | # Confirm nothing was written at all |
| 1455 | remote_dir = repo / ".muse" / "coordination" / "remote" |
| 1456 | if remote_dir.exists(): |
| 1457 | assert list(remote_dir.rglob("*.json")) == [] |
| 1458 | |
| 1459 | def test_remote_records_with_unknown_kind_skipped(self, repo: pathlib.Path) -> None: |
| 1460 | """A record with an unknown kind must not be written anywhere.""" |
| 1461 | evil_records = [ |
| 1462 | {"kind": "../evil_dir", "record_uuid": "safe-uuid", "run_id": "r1", "payload": {}}, |
| 1463 | ] |
| 1464 | with patch(_PULL_TARGET, return_value=_pull_ok(evil_records)): |
| 1465 | result = runner.invoke(cli, _PULL_ARGS) |
| 1466 | assert result.exit_code == 0 |
| 1467 | remote_dir = repo / ".muse" / "coordination" / "remote" |
| 1468 | if remote_dir.exists(): |
| 1469 | assert list(remote_dir.rglob("*.json")) == [] |
| 1470 | |
| 1471 | |
| 1472 | # --------------------------------------------------------------------------- |
| 1473 | # Stress — muse coord sync pull |
| 1474 | # --------------------------------------------------------------------------- |
| 1475 | |
| 1476 | |
| 1477 | class TestCoordSyncPullStress: |
| 1478 | def test_50_sequential_pull_calls_no_records(self, repo: pathlib.Path) -> None: |
| 1479 | """50 sequential pulls with no records all exit 0.""" |
| 1480 | for i in range(50): |
| 1481 | with patch(_PULL_TARGET, return_value=_pull_ok([], cursor=i)): |
| 1482 | result = runner.invoke(cli, _PULL_ARGS + ["-j"]) |
| 1483 | assert result.exit_code == 0, f"Call {i}: {result.output}" |
| 1484 | data = json.loads(result.output.strip()) |
| 1485 | assert data["count"] == 0 |
| 1486 | |
| 1487 | def test_incremental_cursor_chain_100_steps(self, repo: pathlib.Path) -> None: |
| 1488 | """100 incremental pulls each use the cursor from the previous step.""" |
| 1489 | cursor = 0 |
| 1490 | for i in range(100): |
| 1491 | rid = str(uuid.uuid4()) |
| 1492 | fake_records = [{"kind": "reservation", "record_uuid": rid, "run_id": f"r{i}", "payload": {}}] |
| 1493 | with patch(_PULL_TARGET, return_value=_pull_ok(fake_records, cursor=cursor + 1)): |
| 1494 | args = _PULL_ARGS + ["--since-id", str(cursor), "-j"] |
| 1495 | result = runner.invoke(cli, args) |
| 1496 | assert result.exit_code == 0, f"Step {i}: {result.output}" |
| 1497 | data = json.loads(result.output.strip()) |
| 1498 | cursor = data["cursor"] |
| 1499 | assert cursor == 100 |
| 1500 | |
| 1501 | def test_concurrent_pull_8_threads(self, repo: pathlib.Path) -> None: |
| 1502 | """8 threads each call run_pull concurrently; patches applied at test level.""" |
| 1503 | import argparse |
| 1504 | import threading |
| 1505 | |
| 1506 | from muse.cli.commands.coord_sync import run_pull |
| 1507 | |
| 1508 | errors: list[str] = [] |
| 1509 | |
| 1510 | def worker(idx: int) -> None: |
| 1511 | args = argparse.Namespace( |
| 1512 | hub="http://localhost:10003", |
| 1513 | owner="gabriel", |
| 1514 | slug="myrepo", |
| 1515 | signing=None, |
| 1516 | since_id=0, |
| 1517 | kinds=[], |
| 1518 | limit=500, |
| 1519 | fmt="json", |
| 1520 | ) |
| 1521 | try: |
| 1522 | run_pull(args) |
| 1523 | except SystemExit as exc: |
| 1524 | if exc.code != 0: |
| 1525 | errors.append(f"Thread {idx}: exit {exc.code}") |
| 1526 | except Exception as exc: |
| 1527 | errors.append(f"Thread {idx}: {exc}") |
| 1528 | |
| 1529 | pull_p = patch(_PULL_TARGET, return_value=_pull_ok([], cursor=0)) |
| 1530 | repo_p = patch("muse.cli.commands.coord_sync.require_repo", return_value=repo) |
| 1531 | pull_p.start() |
| 1532 | repo_p.start() |
| 1533 | try: |
| 1534 | threads = [threading.Thread(target=worker, args=(i,)) for i in range(8)] |
| 1535 | for t in threads: |
| 1536 | t.start() |
| 1537 | for t in threads: |
| 1538 | t.join() |
| 1539 | finally: |
| 1540 | pull_p.stop() |
| 1541 | repo_p.stop() |
| 1542 | assert not errors, f"Concurrent failures: {errors}" |
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