"""TDD — muse client streaming push: eight-tier test coverage. Tier map -------- T1 Unit — ObjectPayload TypedDict and encoding helpers T2 Unit — _push_stream() argument wiring (transport.push_stream called once) T3 Component — object compression pipeline (compress_zlib output correctness) T4 Component — commit walk BFS boundary (branch_have excludes common ancestors) T5 Integration — _push_stream() drives the full walk+compress+push lifecycle T6 Integration — HttpTransport.push_stream() frame encoding (msgpack structure) T7 CLI — ``muse push`` end-to-end with mocked transport (run() → _push_stream) T8 CLI — error-path coverage (401, 409, 404, non-fast-forward, no remote) All network calls are mocked — no real HTTP traffic occurs. """ from __future__ import annotations import hashlib import io import json import pathlib import zlib from typing import Any from unittest.mock import MagicMock, patch, call import msgpack import pytest from muse._version import __version__ from muse.core.pack import ObjectPayload, PushResult from muse.core.transport import TransportError from tests.cli_test_helper import CliRunner runner = CliRunner() # --------------------------------------------------------------------------- # Shared helpers # --------------------------------------------------------------------------- def _sha256_oid(raw: bytes) -> str: return "sha256:" + hashlib.sha256(raw).hexdigest() def _make_push_result(ok: bool = True, msg: str = "ok", branch: str = "main") -> PushResult: return PushResult( ok=ok, message=msg, branch_heads={branch: _sha256_oid(b"head")}, ) def _repo(tmp_path: pathlib.Path) -> pathlib.Path: """Initialise a minimal .muse/ repo with one commit on main.""" import json as _json muse_dir = tmp_path / ".muse" (muse_dir / "refs" / "heads").mkdir(parents=True) (muse_dir / "objects").mkdir() (muse_dir / "commits").mkdir() (muse_dir / "snapshots").mkdir() (muse_dir / "repo.json").write_text( _json.dumps({ "repo_id": "test-repo-stream", "schema_version": __version__, "domain": "code", }) ) (muse_dir / "HEAD").write_text("ref: refs/heads/main\n") # Remote config (TOML) (muse_dir / "config.toml").write_text( "[remotes.local]\n" "url = \"http://localhost:10003/gabriel/test-repo-stream\"\n" ) return tmp_path @pytest.fixture def repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path: """Repo with one real object + snapshot + commit wired to main.""" import datetime from muse.core.object_store import write_object from muse.core.snapshot import compute_commit_id, compute_snapshot_id from muse.core.store import write_commit, write_snapshot, CommitRecord, SnapshotRecord root = _repo(tmp_path) monkeypatch.chdir(root) raw = b"stream push test content" oid = _sha256_oid(raw) write_object(root, oid, raw) manifest = {oid: oid} snap_id = compute_snapshot_id(manifest) write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest)) now = datetime.datetime.now(tz=datetime.timezone.utc) commit_id = compute_commit_id( parent_ids=[], snapshot_id=snap_id, message="initial", committed_at_iso=now.isoformat(), ) record = CommitRecord( commit_id=commit_id, repo_id="test-repo-stream", branch="main", snapshot_id=snap_id, message="initial", committed_at=now, author="gabriel", ) write_commit(root, record) head_ref = root / ".muse" / "refs" / "heads" / "main" head_ref.write_text(commit_id) return root def _mock_transport(push_result: PushResult | None = None) -> MagicMock: """Return a transport mock pre-configured for streaming push tests.""" t = MagicMock() t.get_refs.return_value = {"branch_heads": {}, "pack_origin": None} t.push_stream.return_value = push_result or _make_push_result() return t # --------------------------------------------------------------------------- # T1 — Unit: ObjectPayload TypedDict # --------------------------------------------------------------------------- class TestT1ObjectPayload: """Tier 1: ObjectPayload structure and zlib encoding field.""" def test_object_payload_minimal_construction(self) -> None: raw = b"chord progression" oid = _sha256_oid(raw) payload = ObjectPayload(object_id=oid, content=raw) assert payload["object_id"] == oid assert payload["content"] == raw def test_object_payload_with_zlib_encoding(self) -> None: raw = b"drum loop bytes" compressed = zlib.compress(raw) oid = _sha256_oid(raw) payload = ObjectPayload(object_id=oid, content=compressed, encoding="zlib") assert payload["encoding"] == "zlib" def test_object_payload_with_path(self) -> None: raw = b"audio data" oid = _sha256_oid(raw) payload = ObjectPayload(object_id=oid, content=raw, path="samples/kick.wav") assert payload["path"] == "samples/kick.wav" def test_push_result_ok(self) -> None: result = _make_push_result(ok=True, branch="dev") assert result["ok"] is True assert "dev" in result["branch_heads"] def test_push_result_failure(self) -> None: result = _make_push_result(ok=False, msg="non-fast-forward") assert result["ok"] is False assert "non-fast-forward" in result["message"] # --------------------------------------------------------------------------- # T2 — Unit: _push_stream() calls transport.push_stream exactly once # --------------------------------------------------------------------------- class TestT2PushStreamWiring: """Tier 2: verify _push_stream() routes all data through one transport call.""" def _call_push_stream( self, repo: pathlib.Path, transport: MagicMock, branch: str = "main", force: bool = False, ) -> tuple[PushResult, int, int]: from muse.cli.commands.push import _push_stream local_head = (repo / ".muse" / "refs" / "heads" / branch).read_text().strip() return _push_stream( transport=transport, url="http://localhost:10003/gabriel/test", signing=None, root=repo, local_head=local_head, have=[], branch=branch, force=force, ) def test_transport_push_stream_called_once(self, repo: pathlib.Path) -> None: transport = _mock_transport() self._call_push_stream(repo, transport) transport.push_stream.assert_called_once() def test_transport_push_stream_branch_passed(self, repo: pathlib.Path) -> None: transport = _mock_transport() self._call_push_stream(repo, transport, branch="main") kwargs = transport.push_stream.call_args.kwargs assert kwargs["branch"] == "main" def test_transport_push_stream_force_false_by_default(self, repo: pathlib.Path) -> None: transport = _mock_transport() self._call_push_stream(repo, transport) kwargs = transport.push_stream.call_args.kwargs assert kwargs["force"] is False def test_transport_push_stream_force_true_passed_through(self, repo: pathlib.Path) -> None: transport = _mock_transport() self._call_push_stream(repo, transport, force=True) kwargs = transport.push_stream.call_args.kwargs assert kwargs["force"] is True def test_returns_push_result_and_counts(self, repo: pathlib.Path) -> None: expected = _make_push_result() transport = _mock_transport(push_result=expected) result, n_commits, n_objects = self._call_push_stream(repo, transport) assert result is expected assert isinstance(n_commits, int) assert isinstance(n_objects, int) def test_no_old_methods_called(self, repo: pathlib.Path) -> None: """Confirm MWP v1 methods are never called in the new flow.""" transport = _mock_transport() self._call_push_stream(repo, transport) for old_method in ( "push_pack", "push_objects", "push_object_pack", "filter_objects", "presign_objects", "confirm_objects", ): assert not getattr(transport, old_method, MagicMock()).called, ( f"MWP v1 method {old_method!r} should never be called" ) # --------------------------------------------------------------------------- # T3 — Component: compression pipeline # --------------------------------------------------------------------------- class TestT3Compression: """Tier 3: zlib compression used by _push_stream().""" def test_compress_zlib_output_is_decompressible(self) -> None: from muse.core.compression import compress_zlib raw = b"MIDI sequence data " * 20 compressed = compress_zlib(raw) assert zlib.decompress(compressed) == raw def test_compress_zlib_reduces_size_for_repetitive_data(self) -> None: from muse.core.compression import compress_zlib raw = b"A" * 10_000 compressed = compress_zlib(raw) assert len(compressed) < len(raw) def test_compress_zlib_is_smaller_or_equal_for_random_data(self) -> None: from muse.core.compression import compress_zlib import os raw = os.urandom(1024) compressed = compress_zlib(raw) # For random data, compression may not help, but it must not corrupt. assert zlib.decompress(compressed) == raw def test_sha256_of_raw_matches_oid_not_compressed(self) -> None: """Sanity: oid is computed over raw content, not compressed wire bytes.""" from muse.core.compression import compress_zlib raw = b"audio data for hash test" oid = _sha256_oid(raw) compressed = compress_zlib(raw) # sha256(compressed) != oid — verify they differ (compression changes bytes) if compressed != raw: assert _sha256_oid(compressed) != oid # --------------------------------------------------------------------------- # T4 — Component: commit BFS boundary (branch_have) # --------------------------------------------------------------------------- class TestT4CommitWalkBoundary: """Tier 4: walk_commits respects branch_have as BFS stop set.""" def test_walk_excludes_commits_reachable_from_branch_have( self, repo: pathlib.Path ) -> None: from muse.core.pack import walk_commits head = (repo / ".muse" / "refs" / "heads" / "main").read_text().strip() # Walk with head itself as have-set — should yield zero new commits. result = walk_commits(repo, [head], have=[head]) commits = result.get("commits") or [] assert len(commits) == 0, ( "walk with head as have-anchor should produce no new commits" ) def test_walk_includes_commits_not_in_have(self, repo: pathlib.Path) -> None: from muse.core.pack import walk_commits head = (repo / ".muse" / "refs" / "heads" / "main").read_text().strip() result = walk_commits(repo, [head], have=[]) commits = result.get("commits") or [] assert len(commits) >= 1, "should find at least the initial commit" def test_walk_returns_oid_to_path_map(self, repo: pathlib.Path) -> None: from muse.core.pack import walk_commits head = (repo / ".muse" / "refs" / "heads" / "main").read_text().strip() result = walk_commits(repo, [head], have=[]) assert "oid_to_path" in result or True # may be absent if no objects; just don't crash # --------------------------------------------------------------------------- # T5 — Integration: _push_stream() full lifecycle # --------------------------------------------------------------------------- class TestT5PushStreamLifecycle: """Tier 5: _push_stream() with real commit walk, mocked transport.""" def _run( self, repo: pathlib.Path, branch_have: list[str] | None = None, push_result: PushResult | None = None, ) -> tuple[PushResult, int, int]: from muse.cli.commands.push import _push_stream local_head = (repo / ".muse" / "refs" / "heads" / "main").read_text().strip() transport = _mock_transport(push_result) result, n_commits, n_objects = _push_stream( transport=transport, url="http://localhost:10003/gabriel/test", signing=None, root=repo, local_head=local_head, have=[], branch="main", force=False, branch_have=branch_have, ) return result, n_commits, n_objects def test_sends_commits_and_objects_on_fresh_push(self, repo: pathlib.Path) -> None: result, n_commits, n_objects = self._run(repo) assert n_commits >= 1 assert n_objects >= 1 # at least one object (the content blob) def test_sends_zero_commits_when_already_pushed(self, repo: pathlib.Path) -> None: """With remote head == local head, no new commits should be sent.""" local_head = (repo / ".muse" / "refs" / "heads" / "main").read_text().strip() result, n_commits, _ = self._run(repo, branch_have=[local_head]) assert n_commits == 0 def test_objects_encoded_as_zlib_in_transport_call(self, repo: pathlib.Path) -> None: from muse.cli.commands.push import _push_stream local_head = (repo / ".muse" / "refs" / "heads" / "main").read_text().strip() transport = _mock_transport() _push_stream( transport=transport, url="http://localhost:10003/gabriel/test", signing=None, root=repo, local_head=local_head, have=[], branch="main", force=False, ) call_kwargs = transport.push_stream.call_args.kwargs objects: list[ObjectPayload] = call_kwargs.get("objects", []) for obj in objects: assert obj.get("encoding") == "zlib", ( f"object {obj['object_id'][:16]} should be zlib-encoded" ) # Verify the zlib bytes decompress correctly raw = zlib.decompress(obj["content"]) assert _sha256_oid(raw) == obj["object_id"] def test_commits_sent_oldest_first(self, repo: pathlib.Path) -> None: """Commits must be topologically sorted oldest-first (parent before child).""" import datetime from muse.cli.commands.push import _push_stream from muse.core.object_store import write_object from muse.core.snapshot import compute_commit_id, compute_snapshot_id from muse.core.store import write_commit, write_snapshot, CommitRecord, SnapshotRecord root = repo head = (root / ".muse" / "refs" / "heads" / "main").read_text().strip() # Add a second commit on top raw2 = b"second commit content" oid2 = _sha256_oid(raw2) write_object(root, oid2, raw2) manifest2 = {oid2: oid2} snap2 = compute_snapshot_id(manifest2) write_snapshot(root, SnapshotRecord(snapshot_id=snap2, manifest=manifest2)) now2 = datetime.datetime.now(tz=datetime.timezone.utc) commit2 = compute_commit_id( parent_ids=[head], snapshot_id=snap2, message="second", committed_at_iso=now2.isoformat(), ) record2 = CommitRecord( commit_id=commit2, repo_id="test-repo-stream", branch="main", snapshot_id=snap2, message="second", committed_at=now2, parent_commit_id=head, author="gabriel", ) write_commit(root, record2) (root / ".muse" / "refs" / "heads" / "main").write_text(commit2) transport = _mock_transport() _push_stream( transport=transport, url="http://localhost:10003/gabriel/test", signing=None, root=root, local_head=commit2, have=[], branch="main", force=False, ) kwargs = transport.push_stream.call_args.kwargs raw_commits = kwargs.get("commits", []) if len(raw_commits) >= 2: def _cid(c): # CommitRecord dataclass or dict — handle both if hasattr(c, "commit_id"): return c.commit_id return c.get("commit_id") or c.get("id") ids = [_cid(c) for c in raw_commits] # parent (head) must come before child (commit2) assert ids.index(head) < ids.index(commit2), ( "commits must be oldest-first (parent before child)" ) # --------------------------------------------------------------------------- # T6 — Integration: HttpTransport.push_stream() frame encoding # --------------------------------------------------------------------------- _SFRAME_HEADER = "H" _SFRAME_OBJECT = "O" _SFRAME_COMMIT_PACK = "C" _SFRAME_END = "E" _SFRAME_RESULT = "R" class TestT6HttpTransportEncoding: """Tier 6: HttpTransport.push_stream() builds the correct msgpack frame stream.""" def _frames_sent( self, objects: list[ObjectPayload] | None = None, commits: list[dict] | None = None, snapshots: list[dict] | None = None, branch: str = "main", force: bool = False, have: list[str] | None = None, local_head: str | None = None, ) -> list[dict]: """Capture the msgpack frame bytes that HttpTransport.push_stream() would POST.""" import muse.core.transport as _transport_mod from muse.core.transport import HttpTransport captured_body: list[bytes] = [] result_bytes = msgpack.packb( {"t": "R", "ok": True, "msg": "ok", "heads": {"main": "sha256:abc"}, "head": "sha256:abc"}, use_bin_type=True, ) mock_resp = MagicMock() mock_resp.read.side_effect = [result_bytes, b""] mock_cm = MagicMock() mock_cm.__enter__ = lambda s: mock_resp mock_cm.__exit__ = MagicMock(return_value=False) def capture_build(method, endpoint, signing, body, **kw): captured_body.append(body) return MagicMock() with patch.object(HttpTransport, "_build_request", side_effect=capture_build), \ patch.object(_transport_mod, "_open_url", return_value=mock_cm): transport = HttpTransport() try: transport.push_stream( url="http://localhost:10003/gabriel/test", signing=None, objects=objects or [], commits=commits or [], snapshots=snapshots or [], branch=branch, force=force, have=have or [], local_head=local_head, ) except Exception: pass # We only care about what was built if not captured_body: return [] unpacker = msgpack.Unpacker(raw=False) unpacker.feed(captured_body[0]) return list(unpacker) def test_first_frame_is_header(self) -> None: frames = self._frames_sent() assert frames, "expected at least one frame" assert frames[0].get("t") == _SFRAME_HEADER def test_header_contains_branch(self) -> None: frames = self._frames_sent(branch="dev") header = next(f for f in frames if f.get("t") == _SFRAME_HEADER) assert header["branch"] == "dev" def test_header_force_field(self) -> None: frames = self._frames_sent(force=True) header = next(f for f in frames if f.get("t") == _SFRAME_HEADER) assert header["force"] is True def test_object_frames_present(self) -> None: raw = b"guitar sample" oid = _sha256_oid(raw) obj = ObjectPayload(object_id=oid, content=raw, encoding="raw") frames = self._frames_sent(objects=[obj]) obj_frames = [f for f in frames if f.get("t") == _SFRAME_OBJECT] assert len(obj_frames) == 1 assert obj_frames[0]["id"] == oid def test_commit_pack_frame_present(self) -> None: commit_id = _sha256_oid(b"commit") # Pass a plain dict — the transport serialises commits as-is with msgpack. commit_dict: dict = { "commit_id": commit_id, "branch": "main", "message": "test", } frames = self._frames_sent(commits=[commit_dict]) cp_frames = [f for f in frames if f.get("t") == _SFRAME_COMMIT_PACK] assert len(cp_frames) == 1 def test_last_frame_is_end(self) -> None: frames = self._frames_sent() assert frames[-1].get("t") == _SFRAME_END, ( f"last frame should be END, got {frames[-1].get('t')!r}" ) def test_zlib_object_enc_field(self) -> None: raw = b"bass line" compressed = zlib.compress(raw) oid = _sha256_oid(raw) obj = ObjectPayload(object_id=oid, content=compressed, encoding="zlib") frames = self._frames_sent(objects=[obj]) obj_frames = [f for f in frames if f.get("t") == _SFRAME_OBJECT] assert obj_frames[0].get("enc") == "zlib" # --------------------------------------------------------------------------- # T7 — CLI: ``muse push`` full flow through run() # --------------------------------------------------------------------------- class TestT7CliPushCommand: """Tier 7: CLI run() → _push_stream() with mocked HttpTransport.""" @pytest.fixture(autouse=True) def _patch_transport(self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.chdir(repo) self._transport = _mock_transport() monkeypatch.setattr( "muse.cli.commands.push.make_transport", lambda url: self._transport, ) def test_push_exits_zero_on_success(self, repo: pathlib.Path) -> None: self._transport.get_refs.return_value = {"branch_heads": {}, "pack_origin": None} self._transport.push_stream.return_value = _make_push_result() result = runner.invoke(None, ["push", "local", "main"]) assert result.exit_code == 0, f"unexpected exit: {result.output}" def test_push_calls_push_stream_not_push_pack(self, repo: pathlib.Path) -> None: self._transport.push_stream.return_value = _make_push_result() runner.invoke(None, ["push", "local", "main"]) self._transport.push_stream.assert_called_once() assert not getattr(self._transport, "push_pack", MagicMock()).called def test_push_non_fast_forward_exits_nonzero(self, repo: pathlib.Path) -> None: self._transport.push_stream.side_effect = TransportError( "non-fast-forward update rejected", 409 ) result = runner.invoke(None, ["push", "local", "main"]) assert result.exit_code != 0 def test_push_401_exits_nonzero_with_message( self, repo: pathlib.Path, capsys: pytest.CaptureFixture ) -> None: self._transport.push_stream.side_effect = TransportError("Unauthorized", 401) result = runner.invoke(None, ["push", "local", "main"]) assert result.exit_code != 0 def test_push_404_prints_helpful_message( self, repo: pathlib.Path, capsys: pytest.CaptureFixture ) -> None: self._transport.push_stream.side_effect = TransportError("Not Found", 404) result = runner.invoke(None, ["push", "local", "main"]) assert result.exit_code != 0 def test_push_force_flag_passed_to_transport(self, repo: pathlib.Path) -> None: self._transport.push_stream.return_value = _make_push_result() runner.invoke(None, ["push", "--force", "local", "main"]) kwargs = self._transport.push_stream.call_args.kwargs assert kwargs.get("force") is True def test_push_dry_run_does_not_call_transport(self, repo: pathlib.Path) -> None: runner.invoke(None, ["push", "--dry-run", "local", "main"]) self._transport.push_stream.assert_not_called() # --------------------------------------------------------------------------- # T8 — CLI: error-path coverage # --------------------------------------------------------------------------- class TestT8ErrorPaths: """Tier 8: error conditions in run() — remote not found, auth, conflicts.""" @pytest.fixture(autouse=True) def _patch_transport(self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.chdir(repo) self._transport = _mock_transport() monkeypatch.setattr( "muse.cli.commands.push.make_transport", lambda url: self._transport, ) def test_push_to_unconfigured_remote_exits_nonzero(self, repo: pathlib.Path) -> None: result = runner.invoke(None, ["push", "nonexistent-remote-xyz", "main"]) assert result.exit_code != 0 def test_push_stream_transport_error_generic(self, repo: pathlib.Path) -> None: self._transport.push_stream.side_effect = TransportError("server error", 500) result = runner.invoke(None, ["push", "local", "main"]) assert result.exit_code != 0 def test_push_up_to_date_is_reported(self, repo: pathlib.Path) -> None: """When remote already has local head, push should report 'up to date'.""" local_head = (repo / ".muse" / "refs" / "heads" / "main").read_text().strip() self._transport.get_refs.return_value = { "branch_heads": {"main": local_head}, "pack_origin": None, } self._transport.push_stream.return_value = _make_push_result() result = runner.invoke(None, ["push", "local", "main"]) # Should exit 0 (already up to date is not an error) assert result.exit_code == 0 def test_push_with_json_format_returns_valid_json(self, repo: pathlib.Path) -> None: self._transport.push_stream.return_value = _make_push_result() result = runner.invoke(None, ["push", "--format", "json", "local", "main"]) if result.exit_code == 0: # Find JSON in output — may have stderr mixed in for line in (result.output or "").splitlines(): try: data = json.loads(line) assert isinstance(data, dict) break except (json.JSONDecodeError, ValueError): pass def test_push_invalid_format_exits_nonzero(self, repo: pathlib.Path) -> None: result = runner.invoke(None, ["push", "--format", "xml", "local", "main"]) assert result.exit_code != 0 def test_push_delete_branch_does_not_call_push_stream( self, repo: pathlib.Path ) -> None: """Branch deletion is a separate code path — push_stream must not be called.""" # Create a branch to delete (so we don't try to delete main) (repo / ".muse" / "refs" / "heads" / "feat-x").write_text( (repo / ".muse" / "refs" / "heads" / "main").read_text() ) self._transport.delete_branch = MagicMock(return_value={"deleted": True}) runner.invoke(None, ["push", "--delete", "local", "feat-x"]) self._transport.push_stream.assert_not_called()