gabriel / muse public
test_cli_stream_push.py python
758 lines 30.3 KB
Raw
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3 docs: docstring sprint contract→find-symbol — idiomatic run… Sonnet 4.6 patch 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 io
19 import json
20 import pathlib
21 import zlib
22 from typing import Any
23 from unittest.mock import MagicMock, patch, call
24
25 import msgpack
26 import pytest
27
28 from muse._version import __version__
29 from muse.core.pack import ObjectPayload, PushResult
30 from muse.core.transport import TransportError
31 from tests.cli_test_helper import CliRunner
32 from muse.core._types import blob_id
33
34 runner = CliRunner()
35
36
37 # ---------------------------------------------------------------------------
38 # Shared helpers
39 # ---------------------------------------------------------------------------
40
41 def _sha256_oid(raw: bytes) -> str:
42 return blob_id(raw)
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_passed_as_raw_to_transport(self, repo: pathlib.Path) -> None:
353 """Objects are passed raw to push_stream; the transport applies compression.
354
355 The CLI delegates compression to the transport layer so it can choose
356 the best algorithm (zstd when available, zlib otherwise) at send time,
357 rather than always pre-compressing with zlib here.
358 """
359 from muse.cli.commands.push import _push_stream
360
361 local_head = (repo / ".muse" / "refs" / "heads" / "main").read_text().strip()
362 transport = _mock_transport()
363 _push_stream(
364 transport=transport,
365 url="http://localhost:10003/gabriel/test",
366 signing=None,
367 root=repo,
368 local_head=local_head,
369 have=[],
370 branch="main",
371 force=False,
372 )
373 call_kwargs = transport.push_stream.call_args.kwargs
374 objects: list[ObjectPayload] = call_kwargs.get("objects", [])
375 for obj in objects:
376 assert obj.get("encoding") == "raw", (
377 f"object {obj['object_id'][:16]} must be raw — transport chooses compression"
378 )
379 # Verify the raw bytes content-address matches the OID.
380 raw = obj["content"]
381 assert isinstance(raw, (bytes, bytearray)), "content must be bytes"
382 assert _sha256_oid(raw) == obj["object_id"]
383
384 def test_commits_sent_oldest_first(self, repo: pathlib.Path) -> None:
385 """Commits must be topologically sorted oldest-first (parent before child)."""
386 import datetime
387 from muse.cli.commands.push import _push_stream
388 from muse.core.object_store import write_object
389 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
390 from muse.core.store import write_commit, write_snapshot, CommitRecord, SnapshotRecord
391
392 root = repo
393 head = (root / ".muse" / "refs" / "heads" / "main").read_text().strip()
394
395 # Add a second commit on top
396 raw2 = b"second commit content"
397 oid2 = _sha256_oid(raw2)
398 write_object(root, oid2, raw2)
399 manifest2 = {oid2: oid2}
400 snap2 = compute_snapshot_id(manifest2)
401 write_snapshot(root, SnapshotRecord(snapshot_id=snap2, manifest=manifest2))
402 now2 = datetime.datetime.now(tz=datetime.timezone.utc)
403 commit2 = compute_commit_id(
404 parent_ids=[head], snapshot_id=snap2,
405 message="second", committed_at_iso=now2.isoformat(),
406 )
407 record2 = CommitRecord(
408 commit_id=commit2,
409 repo_id="test-repo-stream",
410 branch="main",
411 snapshot_id=snap2,
412 message="second",
413 committed_at=now2,
414 parent_commit_id=head,
415 author="gabriel",
416 )
417 write_commit(root, record2)
418 (root / ".muse" / "refs" / "heads" / "main").write_text(commit2)
419
420 transport = _mock_transport()
421 _push_stream(
422 transport=transport,
423 url="http://localhost:10003/gabriel/test",
424 signing=None,
425 root=root,
426 local_head=commit2,
427 have=[],
428 branch="main",
429 force=False,
430 )
431 kwargs = transport.push_stream.call_args.kwargs
432 raw_commits = kwargs.get("commits", [])
433 if len(raw_commits) >= 2:
434 def _cid(c):
435 # CommitRecord dataclass or dict — handle both
436 if hasattr(c, "commit_id"):
437 return c.commit_id
438 return c.get("commit_id") or c.get("id")
439 ids = [_cid(c) for c in raw_commits]
440 # parent (head) must come before child (commit2)
441 assert ids.index(head) < ids.index(commit2), (
442 "commits must be oldest-first (parent before child)"
443 )
444
445
446 # ---------------------------------------------------------------------------
447 # T6 — Integration: HttpTransport.push_stream() frame encoding
448 # ---------------------------------------------------------------------------
449
450 _SFRAME_HEADER = "H"
451 _SFRAME_OBJECT = "O"
452 _SFRAME_COMMIT_PACK = "C"
453 _SFRAME_END = "E"
454 _SFRAME_RESULT = "R"
455
456
457 class TestT6MPackFrameEncoding:
458 """Tier 6: MPackStreamWriter frame encoding — the contract for the push wire format.
459
460 Tests the frame builder (``MPackStreamWriter``) directly. This decouples
461 wire-format correctness from HTTP transport mechanics: the same writer is
462 used by ``HttpTransport.push_stream()`` for the chunked network path and
463 verified here without any network stack involvement.
464
465 Frame sequence for a push:
466 H (HEADER) → O (OBJECT) × N → C (COMMIT_PACK) → E (END)
467
468 Each frame is a msgpack map. Key field contracts:
469 H: {"t": "H", "branch": str, "force": bool, "have": [...]}
470 O: {"t": "O", "id": str, "b": bytes, "enc": str, "sz": int}
471 C: {"t": "C", "commits": [...], "snapshots": [...]}
472 E: {"t": "E", "n_objects": int, "n_commits": int}
473 """
474
475 def _build_frames(
476 self,
477 objects: list[ObjectPayload] | None = None,
478 commits: list[dict] | None = None,
479 snapshots: list[dict] | None = None,
480 branch: str = "main",
481 force: bool = False,
482 have: list[str] | None = None,
483 ) -> list[dict]:
484 """Build and unpack a complete push frame sequence using MPackStreamWriter.
485
486 MPackStreamWriter is stateless — each write_* call returns raw bytes.
487 We concatenate them and feed into an Unpacker to recover the frame dicts.
488 """
489 from muse.core.mpack import MPackStreamWriter
490
491 writer = MPackStreamWriter()
492 raw_objs = objects or []
493 raw_commits = commits or []
494
495 parts: list[bytes] = []
496 parts.append(writer.write_header(
497 op="push",
498 branch=branch,
499 force=force,
500 have=have or [],
501 n_objects=len(raw_objs),
502 n_commits=len(raw_commits),
503 ))
504 for obj in raw_objs:
505 enc = obj.get("encoding", "raw")
506 content: bytes = obj.get("content") or b""
507 if enc == "raw":
508 parts.append(writer.write_object_raw(
509 object_id=obj["object_id"],
510 raw_bytes=content,
511 path=obj.get("path", ""),
512 ))
513 else:
514 parts.append(writer.write_object(
515 object_id=obj["object_id"],
516 content=content,
517 enc=enc,
518 sz=obj.get("sz", len(content)),
519 path=obj.get("path", ""),
520 ))
521 parts.append(writer.write_commit_pack(
522 commits=list(raw_commits),
523 snapshots=list(snapshots or []),
524 ))
525 parts.append(writer.write_end(
526 n_objects=len(raw_objs),
527 n_commits=len(raw_commits),
528 ))
529
530 unpacker = msgpack.Unpacker(raw=False)
531 for chunk in parts:
532 unpacker.feed(chunk)
533 return list(unpacker)
534
535 def test_first_frame_is_header(self) -> None:
536 frames = self._build_frames()
537 assert frames, "expected at least one frame"
538 assert frames[0].get("t") == _SFRAME_HEADER
539
540 def test_header_contains_branch(self) -> None:
541 frames = self._build_frames(branch="dev")
542 header = next(f for f in frames if f.get("t") == _SFRAME_HEADER)
543 assert header["branch"] == "dev"
544
545 def test_header_force_field(self) -> None:
546 frames = self._build_frames(force=True)
547 header = next(f for f in frames if f.get("t") == _SFRAME_HEADER)
548 assert header.get("force") is True
549
550 def test_header_have_list(self) -> None:
551 have = [_sha256_oid(b"ancestor")]
552 frames = self._build_frames(have=have)
553 header = next(f for f in frames if f.get("t") == _SFRAME_HEADER)
554 assert have[0] in (header.get("have") or [])
555
556 def test_raw_object_frame_present(self) -> None:
557 raw = b"guitar sample"
558 oid = _sha256_oid(raw)
559 obj = ObjectPayload(object_id=oid, content=raw, encoding="raw")
560 frames = self._build_frames(objects=[obj])
561 obj_frames = [f for f in frames if f.get("t") == _SFRAME_OBJECT]
562 assert len(obj_frames) == 1
563 assert obj_frames[0]["id"] == oid
564
565 def test_raw_object_content_compressable(self) -> None:
566 """write_object_raw applies compression; the enc field reflects the algorithm."""
567 raw = b"a" * 1024 # highly compressible
568 oid = _sha256_oid(raw)
569 obj = ObjectPayload(object_id=oid, content=raw, encoding="raw")
570 frames = self._build_frames(objects=[obj])
571 obj_frame = next(f for f in frames if f.get("t") == _SFRAME_OBJECT)
572 # enc must be a compression algorithm (zlib or zstd), not "raw"
573 assert obj_frame.get("enc") in ("zlib", "zstd"), (
574 "raw objects should be compressed before wire transmission"
575 )
576
577 def test_zlib_object_enc_field_preserved(self) -> None:
578 """Pre-compressed zlib objects retain their enc field on the wire."""
579 raw = b"bass line"
580 compressed = zlib.compress(raw)
581 oid = _sha256_oid(raw)
582 obj = ObjectPayload(object_id=oid, content=compressed, encoding="zlib")
583 frames = self._build_frames(objects=[obj])
584 obj_frames = [f for f in frames if f.get("t") == _SFRAME_OBJECT]
585 assert obj_frames[0].get("enc") == "zlib"
586
587 def test_commit_pack_frame_present(self) -> None:
588 commit_dict: dict = {
589 "commit_id": _sha256_oid(b"commit"),
590 "branch": "main",
591 "message": "test",
592 }
593 frames = self._build_frames(commits=[commit_dict])
594 cp_frames = [f for f in frames if f.get("t") == _SFRAME_COMMIT_PACK]
595 assert len(cp_frames) == 1
596
597 def test_commit_pack_contains_commit(self) -> None:
598 cid = _sha256_oid(b"commit-x")
599 commit_dict: dict = {"commit_id": cid, "branch": "main", "message": "x"}
600 frames = self._build_frames(commits=[commit_dict])
601 cp = next(f for f in frames if f.get("t") == _SFRAME_COMMIT_PACK)
602 commit_ids = [c.get("commit_id") for c in (cp.get("commits") or [])]
603 assert cid in commit_ids
604
605 def test_last_frame_is_end(self) -> None:
606 frames = self._build_frames()
607 assert frames[-1].get("t") == _SFRAME_END, (
608 f"last frame should be END, got {frames[-1].get('t')!r}"
609 )
610
611 def test_end_frame_object_count(self) -> None:
612 raw = b"piano roll"
613 oid = _sha256_oid(raw)
614 obj = ObjectPayload(object_id=oid, content=raw, encoding="raw")
615 frames = self._build_frames(objects=[obj])
616 end = frames[-1]
617 assert end.get("n_objects") == 1
618
619 def test_frame_order_h_o_c_e(self) -> None:
620 """Frame sequence must be H → O → C → E (server relies on this order)."""
621 raw = b"melody"
622 oid = _sha256_oid(raw)
623 obj = ObjectPayload(object_id=oid, content=raw, encoding="raw")
624 commit_dict: dict = {"commit_id": _sha256_oid(b"c"), "branch": "main", "message": "m"}
625 frames = self._build_frames(objects=[obj], commits=[commit_dict])
626 types = [f.get("t") for f in frames]
627 h = types.index(_SFRAME_HEADER)
628 o = types.index(_SFRAME_OBJECT)
629 c = types.index(_SFRAME_COMMIT_PACK)
630 e = types.index(_SFRAME_END)
631 assert h < o < c < e, f"expected H<O<C<E, got positions {h},{o},{c},{e}"
632
633
634 # ---------------------------------------------------------------------------
635 # T7 — CLI: ``muse push`` full flow through run()
636 # ---------------------------------------------------------------------------
637
638 class TestT7CliPushCommand:
639 """Tier 7: CLI run() → _push_stream() with mocked HttpTransport."""
640
641 @pytest.fixture(autouse=True)
642 def _patch_transport(self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
643 monkeypatch.chdir(repo)
644 self._transport = _mock_transport()
645 monkeypatch.setattr(
646 "muse.cli.commands.push.make_transport",
647 lambda url: self._transport,
648 )
649
650 def test_push_exits_zero_on_success(self, repo: pathlib.Path) -> None:
651 self._transport.get_refs.return_value = {"branch_heads": {}, "pack_origin": None}
652 self._transport.push_stream.return_value = _make_push_result()
653 result = runner.invoke(None, ["push", "local", "main"])
654 assert result.exit_code == 0, f"unexpected exit: {result.output}"
655
656 def test_push_calls_push_stream_not_push_pack(self, repo: pathlib.Path) -> None:
657 self._transport.push_stream.return_value = _make_push_result()
658 runner.invoke(None, ["push", "local", "main"])
659 self._transport.push_stream.assert_called_once()
660 assert not getattr(self._transport, "push_pack", MagicMock()).called
661
662 def test_push_non_fast_forward_exits_nonzero(self, repo: pathlib.Path) -> None:
663 self._transport.push_stream.side_effect = TransportError(
664 "non-fast-forward update rejected", 409
665 )
666 result = runner.invoke(None, ["push", "local", "main"])
667 assert result.exit_code != 0
668
669 def test_push_401_exits_nonzero_with_message(
670 self, repo: pathlib.Path, capsys: pytest.CaptureFixture
671 ) -> None:
672 self._transport.push_stream.side_effect = TransportError("Unauthorized", 401)
673 result = runner.invoke(None, ["push", "local", "main"])
674 assert result.exit_code != 0
675
676 def test_push_404_prints_helpful_message(
677 self, repo: pathlib.Path, capsys: pytest.CaptureFixture
678 ) -> None:
679 self._transport.push_stream.side_effect = TransportError("Not Found", 404)
680 result = runner.invoke(None, ["push", "local", "main"])
681 assert result.exit_code != 0
682
683 def test_push_force_flag_passed_to_transport(self, repo: pathlib.Path) -> None:
684 self._transport.push_stream.return_value = _make_push_result()
685 runner.invoke(None, ["push", "--force", "local", "main"])
686 kwargs = self._transport.push_stream.call_args.kwargs
687 assert kwargs.get("force") is True
688
689 def test_push_dry_run_does_not_call_transport(self, repo: pathlib.Path) -> None:
690 runner.invoke(None, ["push", "--dry-run", "local", "main"])
691 self._transport.push_stream.assert_not_called()
692
693
694 # ---------------------------------------------------------------------------
695 # T8 — CLI: error-path coverage
696 # ---------------------------------------------------------------------------
697
698 class TestT8ErrorPaths:
699 """Tier 8: error conditions in run() — remote not found, auth, conflicts."""
700
701 @pytest.fixture(autouse=True)
702 def _patch_transport(self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
703 monkeypatch.chdir(repo)
704 self._transport = _mock_transport()
705 monkeypatch.setattr(
706 "muse.cli.commands.push.make_transport",
707 lambda url: self._transport,
708 )
709
710 def test_push_to_unconfigured_remote_exits_nonzero(self, repo: pathlib.Path) -> None:
711 result = runner.invoke(None, ["push", "nonexistent-remote-xyz", "main"])
712 assert result.exit_code != 0
713
714 def test_push_stream_transport_error_generic(self, repo: pathlib.Path) -> None:
715 self._transport.push_stream.side_effect = TransportError("server error", 500)
716 result = runner.invoke(None, ["push", "local", "main"])
717 assert result.exit_code != 0
718
719 def test_push_up_to_date_is_reported(self, repo: pathlib.Path) -> None:
720 """When remote already has local head, push should report 'up to date'."""
721 local_head = (repo / ".muse" / "refs" / "heads" / "main").read_text().strip()
722 self._transport.get_refs.return_value = {
723 "branch_heads": {"main": local_head},
724 "pack_origin": None,
725 }
726 self._transport.push_stream.return_value = _make_push_result()
727 result = runner.invoke(None, ["push", "local", "main"])
728 # Should exit 0 (already up to date is not an error)
729 assert result.exit_code == 0
730
731 def test_push_with_json_format_returns_valid_json(self, repo: pathlib.Path) -> None:
732 self._transport.push_stream.return_value = _make_push_result()
733 result = runner.invoke(None, ["push", "--format", "json", "local", "main"])
734 if result.exit_code == 0:
735 # Find JSON in output — may have stderr mixed in
736 for line in (result.output or "").splitlines():
737 try:
738 data = json.loads(line)
739 assert isinstance(data, dict)
740 break
741 except (json.JSONDecodeError, ValueError):
742 pass
743
744 def test_push_invalid_format_exits_nonzero(self, repo: pathlib.Path) -> None:
745 result = runner.invoke(None, ["push", "--format", "xml", "local", "main"])
746 assert result.exit_code != 0
747
748 def test_push_delete_branch_does_not_call_push_stream(
749 self, repo: pathlib.Path
750 ) -> None:
751 """Branch deletion is a separate code path — push_stream must not be called."""
752 # Create a branch to delete (so we don't try to delete main)
753 (repo / ".muse" / "refs" / "heads" / "feat-x").write_text(
754 (repo / ".muse" / "refs" / "heads" / "main").read_text()
755 )
756 self._transport.delete_branch = MagicMock(return_value={"deleted": True})
757 runner.invoke(None, ["push", "--delete", "local", "feat-x"])
758 self._transport.push_stream.assert_not_called()
File History 2 commits
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3 docs: docstring sprint contract→find-symbol — idiomatic run… Sonnet 4.6 patch 146 days ago
sha256:a09b1b4f6838754495547f200aa0ce88e2f56ffc5b20b900f6f0cff2c3cdede9 fix(cursorignore): remove git-ism (.git/worktrees) Human minor ⚠ 149 days ago