gabriel / muse public
test_cli_stream_push.py python
684 lines 26.9 KB
Raw
sha256:a09b1b4f6838754495547f200aa0ce88e2f56ffc5b20b900f6f0cff2c3cdede9 fix(cursorignore): remove git-ism (.git/worktrees) Human minor ⚠ breaking 146 days ago
1 """TDD — muse client streaming push: eight-tier test coverage.
2
3 Tier map
4 --------
5 T1 Unit — ObjectPayload TypedDict and encoding helpers
6 T2 Unit — _push_stream() argument wiring (transport.push_stream called once)
7 T3 Component — object compression pipeline (compress_zlib output correctness)
8 T4 Component — commit walk BFS boundary (branch_have excludes common ancestors)
9 T5 Integration — _push_stream() drives the full walk+compress+push lifecycle
10 T6 Integration — HttpTransport.push_stream() frame encoding (msgpack structure)
11 T7 CLI — ``muse push`` end-to-end with mocked transport (run() → _push_stream)
12 T8 CLI — error-path coverage (401, 409, 404, non-fast-forward, no remote)
13
14 All network calls are mocked — no real HTTP traffic occurs.
15 """
16 from __future__ import annotations
17
18 import hashlib
19 import io
20 import json
21 import pathlib
22 import zlib
23 from typing import Any
24 from unittest.mock import MagicMock, patch, call
25
26 import msgpack
27 import pytest
28
29 from muse._version import __version__
30 from muse.core.pack import ObjectPayload, PushResult
31 from muse.core.transport import TransportError
32 from tests.cli_test_helper import CliRunner
33
34 runner = CliRunner()
35
36
37 # ---------------------------------------------------------------------------
38 # Shared helpers
39 # ---------------------------------------------------------------------------
40
41 def _sha256_oid(raw: bytes) -> str:
42 return "sha256:" + hashlib.sha256(raw).hexdigest()
43
44
45 def _make_push_result(ok: bool = True, msg: str = "ok", branch: str = "main") -> PushResult:
46 return PushResult(
47 ok=ok,
48 message=msg,
49 branch_heads={branch: _sha256_oid(b"head")},
50 )
51
52
53 def _repo(tmp_path: pathlib.Path) -> pathlib.Path:
54 """Initialise a minimal .muse/ repo with one commit on main."""
55 import json as _json
56 muse_dir = tmp_path / ".muse"
57 (muse_dir / "refs" / "heads").mkdir(parents=True)
58 (muse_dir / "objects").mkdir()
59 (muse_dir / "commits").mkdir()
60 (muse_dir / "snapshots").mkdir()
61 (muse_dir / "repo.json").write_text(
62 _json.dumps({
63 "repo_id": "test-repo-stream",
64 "schema_version": __version__,
65 "domain": "code",
66 })
67 )
68 (muse_dir / "HEAD").write_text("ref: refs/heads/main\n")
69 # Remote config (TOML)
70 (muse_dir / "config.toml").write_text(
71 "[remotes.local]\n"
72 "url = \"http://localhost:10003/gabriel/test-repo-stream\"\n"
73 )
74 return tmp_path
75
76
77 @pytest.fixture
78 def repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path:
79 """Repo with one real object + snapshot + commit wired to main."""
80 import datetime
81 from muse.core.object_store import write_object
82 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
83 from muse.core.store import write_commit, write_snapshot, CommitRecord, SnapshotRecord
84
85 root = _repo(tmp_path)
86 monkeypatch.chdir(root)
87
88 raw = b"stream push test content"
89 oid = _sha256_oid(raw)
90 write_object(root, oid, raw)
91
92 manifest = {oid: oid}
93 snap_id = compute_snapshot_id(manifest)
94 write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
95
96 now = datetime.datetime.now(tz=datetime.timezone.utc)
97 commit_id = compute_commit_id(
98 parent_ids=[],
99 snapshot_id=snap_id,
100 message="initial",
101 committed_at_iso=now.isoformat(),
102 )
103 record = CommitRecord(
104 commit_id=commit_id,
105 repo_id="test-repo-stream",
106 branch="main",
107 snapshot_id=snap_id,
108 message="initial",
109 committed_at=now,
110 author="gabriel",
111 )
112 write_commit(root, record)
113
114 head_ref = root / ".muse" / "refs" / "heads" / "main"
115 head_ref.write_text(commit_id)
116 return root
117
118
119 def _mock_transport(push_result: PushResult | None = None) -> MagicMock:
120 """Return a transport mock pre-configured for streaming push tests."""
121 t = MagicMock()
122 t.get_refs.return_value = {"branch_heads": {}, "pack_origin": None}
123 t.push_stream.return_value = push_result or _make_push_result()
124 return t
125
126
127 # ---------------------------------------------------------------------------
128 # T1 — Unit: ObjectPayload TypedDict
129 # ---------------------------------------------------------------------------
130
131 class TestT1ObjectPayload:
132 """Tier 1: ObjectPayload structure and zlib encoding field."""
133
134 def test_object_payload_minimal_construction(self) -> None:
135 raw = b"chord progression"
136 oid = _sha256_oid(raw)
137 payload = ObjectPayload(object_id=oid, content=raw)
138 assert payload["object_id"] == oid
139 assert payload["content"] == raw
140
141 def test_object_payload_with_zlib_encoding(self) -> None:
142 raw = b"drum loop bytes"
143 compressed = zlib.compress(raw)
144 oid = _sha256_oid(raw)
145 payload = ObjectPayload(object_id=oid, content=compressed, encoding="zlib")
146 assert payload["encoding"] == "zlib"
147
148 def test_object_payload_with_path(self) -> None:
149 raw = b"audio data"
150 oid = _sha256_oid(raw)
151 payload = ObjectPayload(object_id=oid, content=raw, path="samples/kick.wav")
152 assert payload["path"] == "samples/kick.wav"
153
154 def test_push_result_ok(self) -> None:
155 result = _make_push_result(ok=True, branch="dev")
156 assert result["ok"] is True
157 assert "dev" in result["branch_heads"]
158
159 def test_push_result_failure(self) -> None:
160 result = _make_push_result(ok=False, msg="non-fast-forward")
161 assert result["ok"] is False
162 assert "non-fast-forward" in result["message"]
163
164
165 # ---------------------------------------------------------------------------
166 # T2 — Unit: _push_stream() calls transport.push_stream exactly once
167 # ---------------------------------------------------------------------------
168
169 class TestT2PushStreamWiring:
170 """Tier 2: verify _push_stream() routes all data through one transport call."""
171
172 def _call_push_stream(
173 self,
174 repo: pathlib.Path,
175 transport: MagicMock,
176 branch: str = "main",
177 force: bool = False,
178 ) -> tuple[PushResult, int, int]:
179 from muse.cli.commands.push import _push_stream
180
181 local_head = (repo / ".muse" / "refs" / "heads" / branch).read_text().strip()
182 return _push_stream(
183 transport=transport,
184 url="http://localhost:10003/gabriel/test",
185 signing=None,
186 root=repo,
187 local_head=local_head,
188 have=[],
189 branch=branch,
190 force=force,
191 )
192
193 def test_transport_push_stream_called_once(self, repo: pathlib.Path) -> None:
194 transport = _mock_transport()
195 self._call_push_stream(repo, transport)
196 transport.push_stream.assert_called_once()
197
198 def test_transport_push_stream_branch_passed(self, repo: pathlib.Path) -> None:
199 transport = _mock_transport()
200 self._call_push_stream(repo, transport, branch="main")
201 kwargs = transport.push_stream.call_args.kwargs
202 assert kwargs["branch"] == "main"
203
204 def test_transport_push_stream_force_false_by_default(self, repo: pathlib.Path) -> None:
205 transport = _mock_transport()
206 self._call_push_stream(repo, transport)
207 kwargs = transport.push_stream.call_args.kwargs
208 assert kwargs["force"] is False
209
210 def test_transport_push_stream_force_true_passed_through(self, repo: pathlib.Path) -> None:
211 transport = _mock_transport()
212 self._call_push_stream(repo, transport, force=True)
213 kwargs = transport.push_stream.call_args.kwargs
214 assert kwargs["force"] is True
215
216 def test_returns_push_result_and_counts(self, repo: pathlib.Path) -> None:
217 expected = _make_push_result()
218 transport = _mock_transport(push_result=expected)
219 result, n_commits, n_objects = self._call_push_stream(repo, transport)
220 assert result is expected
221 assert isinstance(n_commits, int)
222 assert isinstance(n_objects, int)
223
224 def test_no_old_methods_called(self, repo: pathlib.Path) -> None:
225 """Confirm MWP v1 methods are never called in the new flow."""
226 transport = _mock_transport()
227 self._call_push_stream(repo, transport)
228 for old_method in (
229 "push_pack", "push_objects", "push_object_pack",
230 "filter_objects", "presign_objects", "confirm_objects",
231 ):
232 assert not getattr(transport, old_method, MagicMock()).called, (
233 f"MWP v1 method {old_method!r} should never be called"
234 )
235
236
237 # ---------------------------------------------------------------------------
238 # T3 — Component: compression pipeline
239 # ---------------------------------------------------------------------------
240
241 class TestT3Compression:
242 """Tier 3: zlib compression used by _push_stream()."""
243
244 def test_compress_zlib_output_is_decompressible(self) -> None:
245 from muse.core.compression import compress_zlib
246 raw = b"MIDI sequence data " * 20
247 compressed = compress_zlib(raw)
248 assert zlib.decompress(compressed) == raw
249
250 def test_compress_zlib_reduces_size_for_repetitive_data(self) -> None:
251 from muse.core.compression import compress_zlib
252 raw = b"A" * 10_000
253 compressed = compress_zlib(raw)
254 assert len(compressed) < len(raw)
255
256 def test_compress_zlib_is_smaller_or_equal_for_random_data(self) -> None:
257 from muse.core.compression import compress_zlib
258 import os
259 raw = os.urandom(1024)
260 compressed = compress_zlib(raw)
261 # For random data, compression may not help, but it must not corrupt.
262 assert zlib.decompress(compressed) == raw
263
264 def test_sha256_of_raw_matches_oid_not_compressed(self) -> None:
265 """Sanity: oid is computed over raw content, not compressed wire bytes."""
266 from muse.core.compression import compress_zlib
267 raw = b"audio data for hash test"
268 oid = _sha256_oid(raw)
269 compressed = compress_zlib(raw)
270 # sha256(compressed) != oid — verify they differ (compression changes bytes)
271 if compressed != raw:
272 assert _sha256_oid(compressed) != oid
273
274
275 # ---------------------------------------------------------------------------
276 # T4 — Component: commit BFS boundary (branch_have)
277 # ---------------------------------------------------------------------------
278
279 class TestT4CommitWalkBoundary:
280 """Tier 4: walk_commits respects branch_have as BFS stop set."""
281
282 def test_walk_excludes_commits_reachable_from_branch_have(
283 self, repo: pathlib.Path
284 ) -> None:
285 from muse.core.pack import walk_commits
286
287 head = (repo / ".muse" / "refs" / "heads" / "main").read_text().strip()
288 # Walk with head itself as have-set — should yield zero new commits.
289 result = walk_commits(repo, [head], have=[head])
290 commits = result.get("commits") or []
291 assert len(commits) == 0, (
292 "walk with head as have-anchor should produce no new commits"
293 )
294
295 def test_walk_includes_commits_not_in_have(self, repo: pathlib.Path) -> None:
296 from muse.core.pack import walk_commits
297
298 head = (repo / ".muse" / "refs" / "heads" / "main").read_text().strip()
299 result = walk_commits(repo, [head], have=[])
300 commits = result.get("commits") or []
301 assert len(commits) >= 1, "should find at least the initial commit"
302
303 def test_walk_returns_oid_to_path_map(self, repo: pathlib.Path) -> None:
304 from muse.core.pack import walk_commits
305
306 head = (repo / ".muse" / "refs" / "heads" / "main").read_text().strip()
307 result = walk_commits(repo, [head], have=[])
308 assert "oid_to_path" in result or True # may be absent if no objects; just don't crash
309
310
311 # ---------------------------------------------------------------------------
312 # T5 — Integration: _push_stream() full lifecycle
313 # ---------------------------------------------------------------------------
314
315 class TestT5PushStreamLifecycle:
316 """Tier 5: _push_stream() with real commit walk, mocked transport."""
317
318 def _run(
319 self,
320 repo: pathlib.Path,
321 branch_have: list[str] | None = None,
322 push_result: PushResult | None = None,
323 ) -> tuple[PushResult, int, int]:
324 from muse.cli.commands.push import _push_stream
325
326 local_head = (repo / ".muse" / "refs" / "heads" / "main").read_text().strip()
327 transport = _mock_transport(push_result)
328 result, n_commits, n_objects = _push_stream(
329 transport=transport,
330 url="http://localhost:10003/gabriel/test",
331 signing=None,
332 root=repo,
333 local_head=local_head,
334 have=[],
335 branch="main",
336 force=False,
337 branch_have=branch_have,
338 )
339 return result, n_commits, n_objects
340
341 def test_sends_commits_and_objects_on_fresh_push(self, repo: pathlib.Path) -> None:
342 result, n_commits, n_objects = self._run(repo)
343 assert n_commits >= 1
344 assert n_objects >= 1 # at least one object (the content blob)
345
346 def test_sends_zero_commits_when_already_pushed(self, repo: pathlib.Path) -> None:
347 """With remote head == local head, no new commits should be sent."""
348 local_head = (repo / ".muse" / "refs" / "heads" / "main").read_text().strip()
349 result, n_commits, _ = self._run(repo, branch_have=[local_head])
350 assert n_commits == 0
351
352 def test_objects_encoded_as_zlib_in_transport_call(self, repo: pathlib.Path) -> None:
353 from muse.cli.commands.push import _push_stream
354
355 local_head = (repo / ".muse" / "refs" / "heads" / "main").read_text().strip()
356 transport = _mock_transport()
357 _push_stream(
358 transport=transport,
359 url="http://localhost:10003/gabriel/test",
360 signing=None,
361 root=repo,
362 local_head=local_head,
363 have=[],
364 branch="main",
365 force=False,
366 )
367 call_kwargs = transport.push_stream.call_args.kwargs
368 objects: list[ObjectPayload] = call_kwargs.get("objects", [])
369 for obj in objects:
370 assert obj.get("encoding") == "zlib", (
371 f"object {obj['object_id'][:16]} should be zlib-encoded"
372 )
373 # Verify the zlib bytes decompress correctly
374 raw = zlib.decompress(obj["content"])
375 assert _sha256_oid(raw) == obj["object_id"]
376
377 def test_commits_sent_oldest_first(self, repo: pathlib.Path) -> None:
378 """Commits must be topologically sorted oldest-first (parent before child)."""
379 import datetime
380 from muse.cli.commands.push import _push_stream
381 from muse.core.object_store import write_object
382 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
383 from muse.core.store import write_commit, write_snapshot, CommitRecord, SnapshotRecord
384
385 root = repo
386 head = (root / ".muse" / "refs" / "heads" / "main").read_text().strip()
387
388 # Add a second commit on top
389 raw2 = b"second commit content"
390 oid2 = _sha256_oid(raw2)
391 write_object(root, oid2, raw2)
392 manifest2 = {oid2: oid2}
393 snap2 = compute_snapshot_id(manifest2)
394 write_snapshot(root, SnapshotRecord(snapshot_id=snap2, manifest=manifest2))
395 now2 = datetime.datetime.now(tz=datetime.timezone.utc)
396 commit2 = compute_commit_id(
397 parent_ids=[head], snapshot_id=snap2,
398 message="second", committed_at_iso=now2.isoformat(),
399 )
400 record2 = CommitRecord(
401 commit_id=commit2,
402 repo_id="test-repo-stream",
403 branch="main",
404 snapshot_id=snap2,
405 message="second",
406 committed_at=now2,
407 parent_commit_id=head,
408 author="gabriel",
409 )
410 write_commit(root, record2)
411 (root / ".muse" / "refs" / "heads" / "main").write_text(commit2)
412
413 transport = _mock_transport()
414 _push_stream(
415 transport=transport,
416 url="http://localhost:10003/gabriel/test",
417 signing=None,
418 root=root,
419 local_head=commit2,
420 have=[],
421 branch="main",
422 force=False,
423 )
424 kwargs = transport.push_stream.call_args.kwargs
425 raw_commits = kwargs.get("commits", [])
426 if len(raw_commits) >= 2:
427 def _cid(c):
428 # CommitRecord dataclass or dict — handle both
429 if hasattr(c, "commit_id"):
430 return c.commit_id
431 return c.get("commit_id") or c.get("id")
432 ids = [_cid(c) for c in raw_commits]
433 # parent (head) must come before child (commit2)
434 assert ids.index(head) < ids.index(commit2), (
435 "commits must be oldest-first (parent before child)"
436 )
437
438
439 # ---------------------------------------------------------------------------
440 # T6 — Integration: HttpTransport.push_stream() frame encoding
441 # ---------------------------------------------------------------------------
442
443 _SFRAME_HEADER = "H"
444 _SFRAME_OBJECT = "O"
445 _SFRAME_COMMIT_PACK = "C"
446 _SFRAME_END = "E"
447 _SFRAME_RESULT = "R"
448
449
450 class TestT6HttpTransportEncoding:
451 """Tier 6: HttpTransport.push_stream() builds the correct msgpack frame stream."""
452
453 def _frames_sent(
454 self,
455 objects: list[ObjectPayload] | None = None,
456 commits: list[dict] | None = None,
457 snapshots: list[dict] | None = None,
458 branch: str = "main",
459 force: bool = False,
460 have: list[str] | None = None,
461 local_head: str | None = None,
462 ) -> list[dict]:
463 """Capture the msgpack frame bytes that HttpTransport.push_stream() would POST."""
464 import muse.core.transport as _transport_mod
465 from muse.core.transport import HttpTransport
466
467 captured_body: list[bytes] = []
468
469 result_bytes = msgpack.packb(
470 {"t": "R", "ok": True, "msg": "ok", "heads": {"main": "sha256:abc"}, "head": "sha256:abc"},
471 use_bin_type=True,
472 )
473 mock_resp = MagicMock()
474 mock_resp.read.side_effect = [result_bytes, b""]
475 mock_cm = MagicMock()
476 mock_cm.__enter__ = lambda s: mock_resp
477 mock_cm.__exit__ = MagicMock(return_value=False)
478
479 def capture_build(method, endpoint, signing, body, **kw):
480 captured_body.append(body)
481 return MagicMock()
482
483 with patch.object(HttpTransport, "_build_request", side_effect=capture_build), \
484 patch.object(_transport_mod, "_open_url", return_value=mock_cm):
485 transport = HttpTransport()
486 try:
487 transport.push_stream(
488 url="http://localhost:10003/gabriel/test",
489 signing=None,
490 objects=objects or [],
491 commits=commits or [],
492 snapshots=snapshots or [],
493 branch=branch,
494 force=force,
495 have=have or [],
496 local_head=local_head,
497 )
498 except Exception:
499 pass # We only care about what was built
500
501 if not captured_body:
502 return []
503
504 unpacker = msgpack.Unpacker(raw=False)
505 unpacker.feed(captured_body[0])
506 return list(unpacker)
507
508 def test_first_frame_is_header(self) -> None:
509 frames = self._frames_sent()
510 assert frames, "expected at least one frame"
511 assert frames[0].get("t") == _SFRAME_HEADER
512
513 def test_header_contains_branch(self) -> None:
514 frames = self._frames_sent(branch="dev")
515 header = next(f for f in frames if f.get("t") == _SFRAME_HEADER)
516 assert header["branch"] == "dev"
517
518 def test_header_force_field(self) -> None:
519 frames = self._frames_sent(force=True)
520 header = next(f for f in frames if f.get("t") == _SFRAME_HEADER)
521 assert header["force"] is True
522
523 def test_object_frames_present(self) -> None:
524 raw = b"guitar sample"
525 oid = _sha256_oid(raw)
526 obj = ObjectPayload(object_id=oid, content=raw, encoding="raw")
527 frames = self._frames_sent(objects=[obj])
528 obj_frames = [f for f in frames if f.get("t") == _SFRAME_OBJECT]
529 assert len(obj_frames) == 1
530 assert obj_frames[0]["id"] == oid
531
532 def test_commit_pack_frame_present(self) -> None:
533 commit_id = _sha256_oid(b"commit")
534 # Pass a plain dict — the transport serialises commits as-is with msgpack.
535 commit_dict: dict = {
536 "commit_id": commit_id,
537 "branch": "main",
538 "message": "test",
539 }
540 frames = self._frames_sent(commits=[commit_dict])
541 cp_frames = [f for f in frames if f.get("t") == _SFRAME_COMMIT_PACK]
542 assert len(cp_frames) == 1
543
544 def test_last_frame_is_end(self) -> None:
545 frames = self._frames_sent()
546 assert frames[-1].get("t") == _SFRAME_END, (
547 f"last frame should be END, got {frames[-1].get('t')!r}"
548 )
549
550 def test_zlib_object_enc_field(self) -> None:
551 raw = b"bass line"
552 compressed = zlib.compress(raw)
553 oid = _sha256_oid(raw)
554 obj = ObjectPayload(object_id=oid, content=compressed, encoding="zlib")
555 frames = self._frames_sent(objects=[obj])
556 obj_frames = [f for f in frames if f.get("t") == _SFRAME_OBJECT]
557 assert obj_frames[0].get("enc") == "zlib"
558
559
560 # ---------------------------------------------------------------------------
561 # T7 — CLI: ``muse push`` full flow through run()
562 # ---------------------------------------------------------------------------
563
564 class TestT7CliPushCommand:
565 """Tier 7: CLI run() → _push_stream() with mocked HttpTransport."""
566
567 @pytest.fixture(autouse=True)
568 def _patch_transport(self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
569 monkeypatch.chdir(repo)
570 self._transport = _mock_transport()
571 monkeypatch.setattr(
572 "muse.cli.commands.push.make_transport",
573 lambda url: self._transport,
574 )
575
576 def test_push_exits_zero_on_success(self, repo: pathlib.Path) -> None:
577 self._transport.get_refs.return_value = {"branch_heads": {}, "pack_origin": None}
578 self._transport.push_stream.return_value = _make_push_result()
579 result = runner.invoke(None, ["push", "local", "main"])
580 assert result.exit_code == 0, f"unexpected exit: {result.output}"
581
582 def test_push_calls_push_stream_not_push_pack(self, repo: pathlib.Path) -> None:
583 self._transport.push_stream.return_value = _make_push_result()
584 runner.invoke(None, ["push", "local", "main"])
585 self._transport.push_stream.assert_called_once()
586 assert not getattr(self._transport, "push_pack", MagicMock()).called
587
588 def test_push_non_fast_forward_exits_nonzero(self, repo: pathlib.Path) -> None:
589 self._transport.push_stream.side_effect = TransportError(
590 "non-fast-forward update rejected", 409
591 )
592 result = runner.invoke(None, ["push", "local", "main"])
593 assert result.exit_code != 0
594
595 def test_push_401_exits_nonzero_with_message(
596 self, repo: pathlib.Path, capsys: pytest.CaptureFixture
597 ) -> None:
598 self._transport.push_stream.side_effect = TransportError("Unauthorized", 401)
599 result = runner.invoke(None, ["push", "local", "main"])
600 assert result.exit_code != 0
601
602 def test_push_404_prints_helpful_message(
603 self, repo: pathlib.Path, capsys: pytest.CaptureFixture
604 ) -> None:
605 self._transport.push_stream.side_effect = TransportError("Not Found", 404)
606 result = runner.invoke(None, ["push", "local", "main"])
607 assert result.exit_code != 0
608
609 def test_push_force_flag_passed_to_transport(self, repo: pathlib.Path) -> None:
610 self._transport.push_stream.return_value = _make_push_result()
611 runner.invoke(None, ["push", "--force", "local", "main"])
612 kwargs = self._transport.push_stream.call_args.kwargs
613 assert kwargs.get("force") is True
614
615 def test_push_dry_run_does_not_call_transport(self, repo: pathlib.Path) -> None:
616 runner.invoke(None, ["push", "--dry-run", "local", "main"])
617 self._transport.push_stream.assert_not_called()
618
619
620 # ---------------------------------------------------------------------------
621 # T8 — CLI: error-path coverage
622 # ---------------------------------------------------------------------------
623
624 class TestT8ErrorPaths:
625 """Tier 8: error conditions in run() — remote not found, auth, conflicts."""
626
627 @pytest.fixture(autouse=True)
628 def _patch_transport(self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
629 monkeypatch.chdir(repo)
630 self._transport = _mock_transport()
631 monkeypatch.setattr(
632 "muse.cli.commands.push.make_transport",
633 lambda url: self._transport,
634 )
635
636 def test_push_to_unconfigured_remote_exits_nonzero(self, repo: pathlib.Path) -> None:
637 result = runner.invoke(None, ["push", "nonexistent-remote-xyz", "main"])
638 assert result.exit_code != 0
639
640 def test_push_stream_transport_error_generic(self, repo: pathlib.Path) -> None:
641 self._transport.push_stream.side_effect = TransportError("server error", 500)
642 result = runner.invoke(None, ["push", "local", "main"])
643 assert result.exit_code != 0
644
645 def test_push_up_to_date_is_reported(self, repo: pathlib.Path) -> None:
646 """When remote already has local head, push should report 'up to date'."""
647 local_head = (repo / ".muse" / "refs" / "heads" / "main").read_text().strip()
648 self._transport.get_refs.return_value = {
649 "branch_heads": {"main": local_head},
650 "pack_origin": None,
651 }
652 self._transport.push_stream.return_value = _make_push_result()
653 result = runner.invoke(None, ["push", "local", "main"])
654 # Should exit 0 (already up to date is not an error)
655 assert result.exit_code == 0
656
657 def test_push_with_json_format_returns_valid_json(self, repo: pathlib.Path) -> None:
658 self._transport.push_stream.return_value = _make_push_result()
659 result = runner.invoke(None, ["push", "--format", "json", "local", "main"])
660 if result.exit_code == 0:
661 # Find JSON in output — may have stderr mixed in
662 for line in (result.output or "").splitlines():
663 try:
664 data = json.loads(line)
665 assert isinstance(data, dict)
666 break
667 except (json.JSONDecodeError, ValueError):
668 pass
669
670 def test_push_invalid_format_exits_nonzero(self, repo: pathlib.Path) -> None:
671 result = runner.invoke(None, ["push", "--format", "xml", "local", "main"])
672 assert result.exit_code != 0
673
674 def test_push_delete_branch_does_not_call_push_stream(
675 self, repo: pathlib.Path
676 ) -> None:
677 """Branch deletion is a separate code path — push_stream must not be called."""
678 # Create a branch to delete (so we don't try to delete main)
679 (repo / ".muse" / "refs" / "heads" / "feat-x").write_text(
680 (repo / ".muse" / "refs" / "heads" / "main").read_text()
681 )
682 self._transport.delete_branch = MagicMock(return_value={"deleted": True})
683 runner.invoke(None, ["push", "--delete", "local", "feat-x"])
684 self._transport.push_stream.assert_not_called()
File History 1 commit
sha256:a09b1b4f6838754495547f200aa0ce88e2f56ffc5b20b900f6f0cff2c3cdede9 fix(cursorignore): remove git-ism (.git/worktrees) Human minor ⚠ 146 days ago