test_wire_pack_post.py
python
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9
docs: expand cache plan with all seven testing tiers and do…
Sonnet 4.6
131 days ago
| 1 | """TDD — CF Worker pack-receiver routing (Phase D). |
| 2 | |
| 3 | Tests that FAIL before implementation (functions don't exist yet): |
| 4 | TestPostObjectPack::* — _post_object_pack not yet in push.py |
| 5 | TestPushStreamWorkerRouting::* — _push_stream doesn't accept pack_origin yet |
| 6 | |
| 7 | Tests that PASS before implementation (already wired on server side): |
| 8 | TestParseRemoteInfoPackOrigin::* — _parse_remote_info already reads pack_origin |
| 9 | |
| 10 | After implementation all tests must pass. |
| 11 | """ |
| 12 | from __future__ import annotations |
| 13 | |
| 14 | import pathlib |
| 15 | |
| 16 | import msgpack |
| 17 | import pytest |
| 18 | from unittest.mock import AsyncMock, MagicMock, patch |
| 19 | |
| 20 | from muse.core.pack import ObjectPayload, RemoteInfo |
| 21 | from muse.core._types import MsgpackDict |
| 22 | from muse.core.transport import _parse_remote_info |
| 23 | |
| 24 | |
| 25 | # ── RemoteInfo / refs parsing ───────────────────────────────────────────────── |
| 26 | |
| 27 | |
| 28 | class TestParseRemoteInfoPackOrigin: |
| 29 | def _raw(self, **extra) -> bytes: |
| 30 | return msgpack.packb( |
| 31 | {"repo_id": "r1", "domain": "code", "default_branch": "main", |
| 32 | "branch_heads": {}, **extra}, |
| 33 | use_bin_type=True, |
| 34 | ) |
| 35 | |
| 36 | def test_pack_origin_parsed(self) -> None: |
| 37 | info = _parse_remote_info(self._raw(pack_origin="https://pack.staging.musehub.ai")) |
| 38 | assert info.get("pack_origin") == "https://pack.staging.musehub.ai" |
| 39 | |
| 40 | def test_pack_origin_absent_returns_none(self) -> None: |
| 41 | info = _parse_remote_info(self._raw()) |
| 42 | assert info.get("pack_origin") is None |
| 43 | |
| 44 | def test_pack_origin_whitespace_only_ignored(self) -> None: |
| 45 | info = _parse_remote_info(self._raw(pack_origin=" ")) |
| 46 | assert info.get("pack_origin") is None |
| 47 | |
| 48 | def test_pack_origin_stored_in_remote_info(self) -> None: |
| 49 | info = _parse_remote_info(self._raw(pack_origin="https://pack.musehub.ai")) |
| 50 | assert "pack_origin" in info |
| 51 | |
| 52 | |
| 53 | # ── _post_object_pack unit tests ────────────────────────────────────────────── |
| 54 | |
| 55 | |
| 56 | class TestPostObjectPack: |
| 57 | """Unit tests for the Worker upload coroutine.""" |
| 58 | |
| 59 | def _make_objects(self, n: int = 2) -> list[ObjectPayload]: |
| 60 | return [ |
| 61 | ObjectPayload( |
| 62 | object_id="sha256:" + hex(i)[2:].zfill(64), |
| 63 | content=f"content{i}".encode(), |
| 64 | path=f"track{i}.mid", |
| 65 | ) |
| 66 | for i in range(n) |
| 67 | ] |
| 68 | |
| 69 | def _mock_client(self, status: int = 200, body: MsgpackDict | None = None) -> AsyncMock: |
| 70 | mock_resp = MagicMock() |
| 71 | mock_resp.status_code = status |
| 72 | mock_resp.headers = {"content-type": "application/json"} |
| 73 | mock_resp.json.return_value = body or {"stored": len(self._make_objects()), "skipped": 0} |
| 74 | mock_resp.text = "error" if status >= 400 else "" |
| 75 | client = AsyncMock() |
| 76 | client.post.return_value = mock_resp |
| 77 | return client |
| 78 | |
| 79 | @pytest.mark.asyncio |
| 80 | async def test_posts_to_provided_url(self) -> None: |
| 81 | from muse.cli.commands.push import _post_object_pack |
| 82 | client = self._mock_client(body={"stored": 1, "skipped": 0}) |
| 83 | await _post_object_pack( |
| 84 | "https://pack.musehub.ai/gabriel/my-repo/push/object-pack", |
| 85 | self._make_objects(1), None, client, |
| 86 | ) |
| 87 | client.post.assert_called_once() |
| 88 | assert client.post.call_args[0][0] == ( |
| 89 | "https://pack.musehub.ai/gabriel/my-repo/push/object-pack" |
| 90 | ) |
| 91 | |
| 92 | @pytest.mark.asyncio |
| 93 | async def test_body_is_msgpack_with_objects_key(self) -> None: |
| 94 | from muse.cli.commands.push import _post_object_pack |
| 95 | objs = self._make_objects(2) |
| 96 | client = self._mock_client(body={"stored": 2, "skipped": 0}) |
| 97 | await _post_object_pack("https://pack.example.com/a/b/push/object-pack", objs, None, client) |
| 98 | |
| 99 | sent_body: bytes = client.post.call_args.kwargs["content"] |
| 100 | decoded = msgpack.unpackb(sent_body, raw=False) |
| 101 | assert "objects" in decoded |
| 102 | assert len(decoded["objects"]) == 2 |
| 103 | assert decoded["objects"][0]["object_id"] == objs[0]["object_id"] |
| 104 | assert decoded["objects"][0]["content"] == b"content0" |
| 105 | assert decoded["objects"][0]["path"] == "track0.mid" |
| 106 | |
| 107 | @pytest.mark.asyncio |
| 108 | async def test_returns_stored_skipped_counts(self) -> None: |
| 109 | from muse.cli.commands.push import _post_object_pack |
| 110 | client = self._mock_client(body={"stored": 3, "skipped": 1}) |
| 111 | result = await _post_object_pack( |
| 112 | "https://pack.example.com/a/b/push/object-pack", [], None, client, |
| 113 | ) |
| 114 | assert result == {"stored": 3, "skipped": 1} |
| 115 | |
| 116 | @pytest.mark.asyncio |
| 117 | async def test_accepts_msgpack_response(self) -> None: |
| 118 | from muse.cli.commands.push import _post_object_pack |
| 119 | client = AsyncMock() |
| 120 | mock_resp = MagicMock() |
| 121 | mock_resp.status_code = 200 |
| 122 | mock_resp.headers = {"content-type": "application/x-msgpack"} |
| 123 | mock_resp.content = msgpack.packb({"stored": 5, "skipped": 2}, use_bin_type=True) |
| 124 | client.post.return_value = mock_resp |
| 125 | result = await _post_object_pack("https://pack.example.com/x/y/push/object-pack", [], None, client) |
| 126 | assert result == {"stored": 5, "skipped": 2} |
| 127 | |
| 128 | @pytest.mark.asyncio |
| 129 | async def test_raises_transport_error_on_4xx(self) -> None: |
| 130 | from muse.cli.commands.push import _post_object_pack |
| 131 | from muse.core.transport import TransportError |
| 132 | client = self._mock_client(status=403) |
| 133 | with pytest.raises(TransportError) as exc_info: |
| 134 | await _post_object_pack("https://pack.example.com/x/y/push/object-pack", [], None, client) |
| 135 | assert exc_info.value.status_code == 403 |
| 136 | |
| 137 | @pytest.mark.asyncio |
| 138 | async def test_raises_transport_error_on_5xx(self) -> None: |
| 139 | from muse.cli.commands.push import _post_object_pack |
| 140 | from muse.core.transport import TransportError |
| 141 | client = self._mock_client(status=502) |
| 142 | with pytest.raises(TransportError) as exc_info: |
| 143 | await _post_object_pack("https://pack.example.com/x/y/push/object-pack", [], None, client) |
| 144 | assert exc_info.value.status_code == 502 |
| 145 | |
| 146 | @pytest.mark.asyncio |
| 147 | async def test_content_type_header_is_msgpack(self) -> None: |
| 148 | from muse.cli.commands.push import _post_object_pack |
| 149 | client = self._mock_client(body={"stored": 0, "skipped": 0}) |
| 150 | await _post_object_pack("https://pack.example.com/a/b/push/object-pack", [], None, client) |
| 151 | sent_headers = client.post.call_args.kwargs.get("headers", {}) |
| 152 | assert sent_headers.get("Content-Type") == "application/x-msgpack" |
| 153 | |
| 154 | |
| 155 | # ── _push_stream Worker routing integration ─────────────────────────────────── |
| 156 | |
| 157 | |
| 158 | def _fresh_repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path: |
| 159 | """Create a repo with one commit and one object.""" |
| 160 | from tests.cli_test_helper import CliRunner |
| 161 | runner = CliRunner() |
| 162 | cli = None |
| 163 | env = {"MUSE_REPO_ROOT": str(tmp_path)} |
| 164 | monkeypatch.chdir(tmp_path) |
| 165 | monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path)) |
| 166 | r = runner.invoke(cli, ["init"], env=env, catch_exceptions=False) |
| 167 | assert r.exit_code == 0, r.output |
| 168 | (tmp_path / "track.mid").write_bytes(b"\x00MIDI" + b"\xff" * 200) |
| 169 | r = runner.invoke(cli, ["code", "add", "track.mid"], env=env, catch_exceptions=False) |
| 170 | assert r.exit_code == 0, r.output |
| 171 | r = runner.invoke(cli, ["commit", "-m", "add track"], env=env, catch_exceptions=False) |
| 172 | assert r.exit_code == 0, r.output |
| 173 | return tmp_path |
| 174 | |
| 175 | |
| 176 | class TestPushStreamWorkerRouting: |
| 177 | """_push_stream routes correctly based on whether pack_origin is set.""" |
| 178 | |
| 179 | def _fake_transport(self) -> MagicMock: |
| 180 | transport = MagicMock() |
| 181 | transport.fetch_remote_info.return_value = RemoteInfo( |
| 182 | repo_id="r1", domain="midi", default_branch="main", branch_heads={}, |
| 183 | ) |
| 184 | from muse.core.pack import PushResult |
| 185 | ok_result = PushResult(ok=True, message="pushed", branch_heads={"main": "sha256:" + "a" * 64}) |
| 186 | coro = AsyncMock(return_value=ok_result) |
| 187 | transport.push_stream_coro = coro |
| 188 | return transport |
| 189 | |
| 190 | def test_push_stream_accepts_pack_origin_param(self, tmp_path: pathlib.Path) -> None: |
| 191 | """_push_stream signature must accept pack_origin keyword argument.""" |
| 192 | import inspect |
| 193 | from muse.cli.commands.push import _push_stream |
| 194 | sig = inspect.signature(_push_stream) |
| 195 | assert "pack_origin" in sig.parameters, ( |
| 196 | "_push_stream must accept pack_origin parameter" |
| 197 | ) |
| 198 | |
| 199 | def test_objects_sent_to_worker_not_stream_when_pack_origin_set( |
| 200 | self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch, |
| 201 | ) -> None: |
| 202 | """When pack_origin is set, objects go to Worker; push_stream_coro gets objects=[].""" |
| 203 | repo = _fresh_repo(tmp_path, monkeypatch) |
| 204 | |
| 205 | from muse.core.store import get_head_commit_id |
| 206 | local_head = get_head_commit_id(repo, "main") |
| 207 | assert local_head |
| 208 | |
| 209 | transport = self._fake_transport() |
| 210 | worker_calls: list[tuple] = [] |
| 211 | |
| 212 | async def fake_post_object_pack(worker_url, objects, signing, client): |
| 213 | worker_calls.append((worker_url, list(objects))) |
| 214 | return {"stored": len(objects), "skipped": 0} |
| 215 | |
| 216 | from muse.cli.commands.push import _push_stream |
| 217 | with patch("muse.cli.commands.push._post_object_pack", fake_post_object_pack): |
| 218 | result, _commits, _objects = _push_stream( |
| 219 | transport, |
| 220 | url="https://staging.musehub.ai/gabriel/test-repo", |
| 221 | signing=None, |
| 222 | root=repo, |
| 223 | local_head=local_head, |
| 224 | have=[], |
| 225 | branch="main", |
| 226 | force=False, |
| 227 | pack_origin="https://pack.staging.musehub.ai", |
| 228 | ) |
| 229 | |
| 230 | # Worker must have been called with objects |
| 231 | assert len(worker_calls) >= 1, "Worker was never called" |
| 232 | all_worker_objects = [o for _, objs in worker_calls for o in objs] |
| 233 | assert len(all_worker_objects) >= 1, "No objects sent to Worker" |
| 234 | |
| 235 | # push_stream_coro must have been called with objects=[] |
| 236 | stream_call = transport.push_stream_coro.call_args |
| 237 | assert stream_call is not None, "push_stream_coro was never called" |
| 238 | assert stream_call.kwargs.get("objects") == [], ( |
| 239 | f"Expected objects=[] in stream call, got {stream_call.kwargs.get('objects')}" |
| 240 | ) |
| 241 | |
| 242 | def test_worker_url_constructed_from_pack_origin_and_repo_path( |
| 243 | self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch, |
| 244 | ) -> None: |
| 245 | """Worker URL = {pack_origin}{/owner/slug}/push/object-pack.""" |
| 246 | repo = _fresh_repo(tmp_path, monkeypatch) |
| 247 | |
| 248 | from muse.core.store import get_head_commit_id |
| 249 | local_head = get_head_commit_id(repo, "main") |
| 250 | |
| 251 | transport = self._fake_transport() |
| 252 | captured_urls: list[str] = [] |
| 253 | |
| 254 | async def fake_post_object_pack(worker_url, objects, signing, client): |
| 255 | captured_urls.append(worker_url) |
| 256 | return {"stored": len(objects), "skipped": 0} |
| 257 | |
| 258 | from muse.cli.commands.push import _push_stream |
| 259 | with patch("muse.cli.commands.push._post_object_pack", fake_post_object_pack): |
| 260 | _push_stream( |
| 261 | transport, |
| 262 | url="https://staging.musehub.ai/gabriel/test-repo", |
| 263 | signing=None, |
| 264 | root=repo, |
| 265 | local_head=local_head, |
| 266 | have=[], |
| 267 | branch="main", |
| 268 | force=False, |
| 269 | pack_origin="https://pack.staging.musehub.ai", |
| 270 | ) |
| 271 | |
| 272 | assert len(captured_urls) >= 1 |
| 273 | assert captured_urls[0] == ( |
| 274 | "https://pack.staging.musehub.ai/gabriel/test-repo/push/object-pack" |
| 275 | ) |
| 276 | |
| 277 | def test_no_worker_call_when_pack_origin_absent( |
| 278 | self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch, |
| 279 | ) -> None: |
| 280 | """Without pack_origin, _post_object_pack is never called.""" |
| 281 | repo = _fresh_repo(tmp_path, monkeypatch) |
| 282 | |
| 283 | from muse.core.store import get_head_commit_id |
| 284 | local_head = get_head_commit_id(repo, "main") |
| 285 | |
| 286 | transport = self._fake_transport() |
| 287 | worker_calls: list[str] = [] |
| 288 | |
| 289 | async def fake_post_object_pack(worker_url, objects, signing, client): |
| 290 | worker_calls.append(worker_url) |
| 291 | return {"stored": 0, "skipped": 0} |
| 292 | |
| 293 | from muse.cli.commands.push import _push_stream |
| 294 | with patch("muse.cli.commands.push._post_object_pack", fake_post_object_pack): |
| 295 | _push_stream( |
| 296 | transport, |
| 297 | url="https://staging.musehub.ai/gabriel/test-repo", |
| 298 | signing=None, |
| 299 | root=repo, |
| 300 | local_head=local_head, |
| 301 | have=[], |
| 302 | branch="main", |
| 303 | force=False, |
| 304 | pack_origin=None, # no Worker |
| 305 | ) |
| 306 | |
| 307 | assert worker_calls == [], "Worker must not be called when pack_origin is absent" |
| 308 | |
| 309 | def test_stream_carries_objects_when_no_pack_origin( |
| 310 | self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch, |
| 311 | ) -> None: |
| 312 | """Without pack_origin, push_stream_coro receives the objects.""" |
| 313 | repo = _fresh_repo(tmp_path, monkeypatch) |
| 314 | |
| 315 | from muse.core.store import get_head_commit_id |
| 316 | local_head = get_head_commit_id(repo, "main") |
| 317 | |
| 318 | transport = self._fake_transport() |
| 319 | |
| 320 | from muse.cli.commands.push import _push_stream |
| 321 | _push_stream( |
| 322 | transport, |
| 323 | url="https://staging.musehub.ai/gabriel/test-repo", |
| 324 | signing=None, |
| 325 | root=repo, |
| 326 | local_head=local_head, |
| 327 | have=[], |
| 328 | branch="main", |
| 329 | force=False, |
| 330 | pack_origin=None, |
| 331 | ) |
| 332 | |
| 333 | stream_call = transport.push_stream_coro.call_args |
| 334 | assert stream_call is not None |
| 335 | stream_objects = stream_call.kwargs.get("objects", []) |
| 336 | assert len(stream_objects) >= 1, ( |
| 337 | "Without pack_origin, objects must travel via push/stream" |
| 338 | ) |
File History
1 commit
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9
docs: expand cache plan with all seven testing tiers and do…
Sonnet 4.6
131 days ago