gabriel / musehub public
test_wire_push_stream.py python
1,136 lines 41.5 KB
Raw
sha256:9590cee1e0ccd6c76528f005b95d634d80f5019f0dcb7c371e149adc31d1fb65 refactor: enforce gRPC framing on all MWP wire traffic Sonnet 4.6 minor ⚠ breaking 156 days ago
1 """TDD — MWP v2 streaming push: eight-tier test coverage.
2
3 Tier map
4 --------
5 T1 Unit — frame codec helpers (_sp, _prog, _err, _result)
6 T2 Unit — server protocol state machine (pure frame dispatch, no DB/R2)
7 T3 Component — object validation (hash check, size limits, enc modes)
8 T4 Component — commit-pack validation (schema, limits, signature gate)
9 T5 Service — wire_push_stream() async generator against in-memory stubs
10 T6 Integration — service layer against real test DB, stub R2 backend
11 T7 Route — POST /push/stream via ASGI test client (StreamingResponse)
12 T8 E2E — complete round-trip: push objects + commits → GET /refs confirms head
13
14 All frame construction uses the canonical SFRAME_* constants from
15 ``musehub.models.wire`` so the tests act as a contract for the wire format
16 itself — any change to the frame shape will break these tests first.
17 """
18 from __future__ import annotations
19
20 import asyncio
21 import hashlib
22 import zlib
23 from collections.abc import AsyncGenerator, AsyncIterator
24 from datetime import datetime, timezone
25 from typing import Any
26 from unittest.mock import AsyncMock, MagicMock, patch
27
28 import msgpack
29 import pytest
30 from httpx import AsyncClient
31 from sqlalchemy.ext.asyncio import AsyncSession
32
33 from musehub.models.wire import (
34 SFRAME_COMMIT_PACK,
35 SFRAME_END,
36 SFRAME_ERROR,
37 SFRAME_HEADER,
38 SFRAME_OBJECT,
39 SFRAME_PROGRESS,
40 SFRAME_RESULT,
41 STREAM_MAX_COMMITS,
42 STREAM_MAX_OBJECT_WIRE_BYTES,
43 STREAM_MAX_OBJECTS,
44 )
45 from muse.core.mpack import GRPC_CONTENT_TYPE, MuseWireFrameWriter, grpc_frame
46 from tests.factories import create_repo
47
48 _fw = MuseWireFrameWriter()
49
50
51 # ---------------------------------------------------------------------------
52 # Shared codec helpers
53 # ---------------------------------------------------------------------------
54
55 def _pack(data: object) -> bytes:
56 """Encode one msgpack frame payload (without transport envelope)."""
57 return msgpack.packb(data, use_bin_type=True)
58
59
60 def _wrap(ft: str, data: object) -> bytes:
61 """Encode, wrap in MWP envelope, and add gRPC length-prefix.
62
63 wire_push_stream calls iter_wire_frames_grpc, which expects each MWP
64 frame prefixed with a 5-byte gRPC header (compress_flag + uint32 length).
65 """
66 return grpc_frame(_fw.wrap(frame_type=ft, payload=_pack(data)))
67
68
69 def _unpack_all(raw: bytes) -> list[dict]:
70 """Decode all concatenated gRPC-framed bytes into a list of dicts.
71
72 Handles two formats:
73 - gRPC-wrapped MWP: 0x00 + length(4) + b"muse" envelope → parse inner MWP
74 - gRPC-wrapped msgpack: 0x00 + length(4) + msgpack payload → parse as dict
75 """
76 import struct
77 results = []
78 offset = 0
79 while offset < len(raw):
80 if offset + 5 > len(raw):
81 break
82 if raw[offset] != 0x00:
83 break
84 grpc_length = struct.unpack(">I", raw[offset + 1:offset + 5])[0]
85 if offset + 5 + grpc_length > len(raw):
86 break
87 inner = raw[offset + 5: offset + 5 + grpc_length]
88 offset += 5 + grpc_length
89 if inner[:4] == b"muse":
90 # gRPC-wrapped MWP envelope — parse inner MWP frame
91 header_len = struct.unpack(">I", inner[5:9])[0]
92 payload_start = 9 + header_len + 8
93 payload_len = struct.unpack(">Q", inner[9 + header_len:payload_start])[0]
94 payload = inner[payload_start:payload_start + payload_len]
95 results.append(msgpack.unpackb(payload, raw=False))
96 else:
97 # gRPC-wrapped raw msgpack — server response frames (P/X/R)
98 decoded = msgpack.unpackb(inner, raw=False)
99 if isinstance(decoded, dict):
100 results.append(decoded)
101 return results
102
103
104 def _sha256_oid(raw: bytes) -> str:
105 return "sha256:" + hashlib.sha256(raw).hexdigest()
106
107
108 def _utc() -> str:
109 return datetime.now(tz=timezone.utc).isoformat()
110
111
112 def _make_obj_bytes(content: bytes = b"hello world") -> tuple[str, bytes]:
113 """Return (sha256_oid, raw_content) for a test object."""
114 oid = _sha256_oid(content)
115 return oid, content
116
117
118 def _header_frame(
119 branch: str = "main",
120 force: bool = False,
121 have: list[str] | None = None,
122 head: str = "sha256:abc",
123 n_objects: int = 0,
124 n_commits: int = 1,
125 ) -> bytes:
126 return _wrap(SFRAME_HEADER, {
127 "t": SFRAME_HEADER,
128 "branch": branch,
129 "force": force,
130 "have": have or [],
131 "head": head,
132 "n_objects": n_objects,
133 "n_commits": n_commits,
134 })
135
136
137 def _object_frame(oid: str, content: bytes, path: str = "track.wav", enc: str = "raw") -> bytes:
138 return _wrap(SFRAME_OBJECT, {
139 "t": SFRAME_OBJECT,
140 "id": oid,
141 "content": content,
142 "path": path,
143 "enc": enc,
144 })
145
146
147 def _commit_pack_frame(commits: list[dict], snapshots: list[dict] | None = None) -> bytes:
148 return _wrap(SFRAME_COMMIT_PACK, {
149 "t": SFRAME_COMMIT_PACK,
150 "commits": commits,
151 "snapshots": snapshots or [],
152 })
153
154
155 def _end_frame() -> bytes:
156 return _wrap(SFRAME_END, {"t": SFRAME_END})
157
158
159 def _make_commit(
160 commit_id: str | None = None,
161 parent_ids: list[str] | None = None,
162 snapshot_id: str | None = None,
163 branch: str = "main",
164 author: str = "gabriel",
165 ) -> dict:
166 cid = commit_id or _sha256_oid(f"commit-{_utc()}".encode())
167 return {
168 "commit_id": cid,
169 "parent_ids": parent_ids or [],
170 "snapshot_id": snapshot_id or _sha256_oid(b"default-snap"),
171 "branch": branch,
172 "message": "test commit",
173 "author": author,
174 "committed_at": _utc(),
175 "signature": "",
176 "signer_key_id": "",
177 "agent_id": "",
178 "model_id": "",
179 "metadata": {},
180 }
181
182
183 def _make_snapshot(snapshot_id: str, manifest: dict | None = None) -> dict:
184 return {
185 "snapshot_id": snapshot_id,
186 "manifest": manifest or {},
187 "committed_at": _utc(),
188 }
189
190
191 async def _collect_frames(gen: AsyncGenerator[bytes, None]) -> list[dict]:
192 """Drain an async generator of frame bytes into a list of decoded dicts."""
193 raw = b""
194 async for chunk in gen:
195 raw += chunk
196 return _unpack_all(raw)
197
198
199 async def _body_iter(*frames: bytes) -> AsyncIterator[bytes]:
200 """Wrap pre-built frames as an async iterator for wire_push_stream()."""
201 for f in frames:
202 yield f
203
204
205 # ---------------------------------------------------------------------------
206 # T1 — Unit: frame codec helpers
207 # ---------------------------------------------------------------------------
208
209 class TestT1FrameCodec:
210 """Tier 1: verify frame construction and MIME constant."""
211
212 def test_stream_mime_type(self) -> None:
213 assert GRPC_CONTENT_TYPE == "application/grpc+muse"
214
215 def test_sframe_constants_are_single_chars(self) -> None:
216 for const in (
217 SFRAME_HEADER, SFRAME_OBJECT, SFRAME_COMMIT_PACK, SFRAME_END,
218 SFRAME_PROGRESS, SFRAME_ERROR, SFRAME_RESULT,
219 ):
220 assert len(const) == 1, f"{const!r} should be a single character"
221
222 def test_client_server_frame_tags_are_disjoint(self) -> None:
223 client_tags = {SFRAME_HEADER, SFRAME_OBJECT, SFRAME_COMMIT_PACK, SFRAME_END}
224 server_tags = {SFRAME_PROGRESS, SFRAME_ERROR, SFRAME_RESULT}
225 assert client_tags.isdisjoint(server_tags), (
226 "client and server frame tags must not overlap"
227 )
228
229 def test_header_frame_round_trips(self) -> None:
230 raw = _header_frame(branch="dev", n_objects=3, n_commits=2)
231 frames = _unpack_all(raw)
232 assert len(frames) == 1
233 f = frames[0]
234 assert f["t"] == SFRAME_HEADER
235 assert f["branch"] == "dev"
236 assert f["n_objects"] == 3
237 assert f["n_commits"] == 2
238
239 def test_object_frame_round_trips(self) -> None:
240 oid, content = _make_obj_bytes(b"muse track data")
241 raw = _object_frame(oid, content, path="beat.mid")
242 frames = _unpack_all(raw)
243 f = frames[0]
244 assert f["t"] == SFRAME_OBJECT
245 assert f["id"] == oid
246 assert bytes(f["content"]) == content
247
248 def test_commit_pack_frame_round_trips(self) -> None:
249 commit = _make_commit()
250 raw = _commit_pack_frame([commit])
251 frames = _unpack_all(raw)
252 f = frames[0]
253 assert f["t"] == SFRAME_COMMIT_PACK
254 assert len(f["commits"]) == 1
255
256 def test_end_frame_round_trips(self) -> None:
257 frames = _unpack_all(_end_frame())
258 assert frames[0]["t"] == SFRAME_END
259
260 def test_multiple_frames_concatenated_parse_correctly(self) -> None:
261 body = (
262 _header_frame()
263 + _object_frame(*_make_obj_bytes())
264 + _end_frame()
265 )
266 frames = _unpack_all(body)
267 assert [f["t"] for f in frames] == [SFRAME_HEADER, SFRAME_OBJECT, SFRAME_END]
268
269 def test_stream_limits_are_positive(self) -> None:
270 assert STREAM_MAX_OBJECTS > 0
271 assert STREAM_MAX_COMMITS > 0
272 assert STREAM_MAX_OBJECT_WIRE_BYTES > 0
273
274
275 # ---------------------------------------------------------------------------
276 # T2 — Unit: server helper functions
277 # ---------------------------------------------------------------------------
278
279 class TestT2ServerHelpers:
280 """Tier 2: test _sp, _prog, _err, _result frame builders."""
281
282 def _import_helpers(self):
283 from musehub.services.musehub_wire import _sp, _prog, _err, _result
284 return _sp, _prog, _err, _result
285
286 def test_prog_encodes_progress_frame(self) -> None:
287 _, _prog, _, _ = self._import_helpers()
288 frames = _unpack_all(_prog("uploading objects"))
289 assert frames[0]["t"] == SFRAME_PROGRESS
290 assert frames[0]["msg"] == "uploading objects"
291
292 def test_err_encodes_error_frame_with_code(self) -> None:
293 _, _, _err, _ = self._import_helpers()
294 frames = _unpack_all(_err("repo not found", 404))
295 f = frames[0]
296 assert f["t"] == SFRAME_ERROR
297 assert f["code"] == 404
298 assert "repo not found" in f["msg"]
299
300 def test_err_default_code_is_400(self) -> None:
301 _, _, _err, _ = self._import_helpers()
302 frames = _unpack_all(_err("bad request"))
303 assert frames[0]["code"] == 400
304
305 def test_result_ok_encodes_correctly(self) -> None:
306 _, _, _, _result = self._import_helpers()
307 heads = {"main": "sha256:abc"}
308 frames = _unpack_all(_result(True, "pushed", heads, "sha256:abc"))
309 f = frames[0]
310 assert f["t"] == SFRAME_RESULT
311 assert f["ok"] is True
312 assert f["heads"] == heads
313
314 def test_result_failure_encodes_correctly(self) -> None:
315 _, _, _, _result = self._import_helpers()
316 frames = _unpack_all(_result(False, "rejected", {}, ""))
317 assert frames[0]["ok"] is False
318
319
320 # ---------------------------------------------------------------------------
321 # T3 — Component: object validation edge cases
322 # ---------------------------------------------------------------------------
323
324 class TestT3ObjectValidation:
325 """Tier 3: hash mismatch, size limits, compression encoding."""
326
327 def _oid_for(self, raw: bytes) -> str:
328 return _sha256_oid(raw)
329
330 def test_sha256_oid_format(self) -> None:
331 raw = b"guitar riff"
332 oid = self._oid_for(raw)
333 assert oid.startswith("sha256:")
334 assert len(oid) == len("sha256:") + 64
335
336 def test_zlib_object_frame_decompresses_correctly(self) -> None:
337 raw = b"MIDI note data " * 50
338 compressed = zlib.compress(raw)
339 oid, _ = _make_obj_bytes(raw) # sha256 of raw (before compression)
340 frame_bytes = _object_frame(oid, compressed, enc="zlib")
341 frames = _unpack_all(frame_bytes)
342 f = frames[0]
343 assert f["enc"] == "zlib"
344 assert zlib.decompress(bytes(f["content"])) == raw
345
346 def test_wire_size_limit_constant_is_reasonable(self) -> None:
347 # Must be at least 1 MB and at most 512 MB per object wire payload.
348 assert 1 * 1024 * 1024 <= STREAM_MAX_OBJECT_WIRE_BYTES <= 512 * 1024 * 1024
349
350 def test_object_frame_with_empty_content_is_encodable(self) -> None:
351 raw = b""
352 oid = _sha256_oid(raw)
353 frame_bytes = _object_frame(oid, raw)
354 frames = _unpack_all(frame_bytes)
355 assert bytes(frames[0]["content"]) == b""
356
357 def test_large_object_frame_exceeds_limit_is_detectable(self) -> None:
358 """Frame content larger than STREAM_MAX_OBJECT_WIRE_BYTES should be flagged."""
359 oversized = b"x" * (STREAM_MAX_OBJECT_WIRE_BYTES + 1)
360 oid = _sha256_oid(oversized)
361 frame = _unpack_all(_object_frame(oid, oversized))[0]
362 assert len(bytes(frame["content"])) > STREAM_MAX_OBJECT_WIRE_BYTES
363
364 def test_raw_encoding_preserves_exact_bytes(self) -> None:
365 raw = b"\x00\x01\x02\x03" * 100
366 oid = _sha256_oid(raw)
367 frame = _unpack_all(_object_frame(oid, raw, enc="raw"))[0]
368 assert bytes(frame["content"]) == raw
369
370
371 # ---------------------------------------------------------------------------
372 # T4 — Component: commit-pack schema validation
373 # ---------------------------------------------------------------------------
374
375 class TestT4CommitPackValidation:
376 """Tier 4: WireCommit/WireSnapshot schema, commit-count limit."""
377
378 def test_minimal_commit_passes_model_validate(self) -> None:
379 from musehub.models.wire import WireCommit
380 commit = _make_commit()
381 obj = WireCommit.model_validate(commit)
382 assert obj.commit_id == commit["commit_id"]
383
384 def test_commit_without_required_fields_raises(self) -> None:
385 from musehub.models.wire import WireCommit
386 import pydantic
387 with pytest.raises((pydantic.ValidationError, Exception)):
388 WireCommit.model_validate({"message": "incomplete"})
389
390 def test_snapshot_passes_model_validate(self) -> None:
391 from musehub.models.wire import WireSnapshot
392 sid = _sha256_oid(b"snap")
393 snap = _make_snapshot(sid, {"file.wav": sid})
394 obj = WireSnapshot.model_validate(snap)
395 assert obj.snapshot_id == sid
396
397 def test_commit_pack_limit_constant(self) -> None:
398 assert STREAM_MAX_COMMITS >= 1_000
399
400 def test_commit_pack_frame_encodes_many_commits(self) -> None:
401 commits = [_make_commit() for _ in range(10)]
402 frame = _unpack_all(_commit_pack_frame(commits))[0]
403 assert len(frame["commits"]) == 10
404
405 def test_signed_commit_has_signature_fields(self) -> None:
406 from musehub.models.wire import WireCommit
407 commit = _make_commit()
408 commit["signature"] = "sig_base64"
409 commit["signer_key_id"] = "key123"
410 commit["agent_id"] = "agent-1"
411 commit["model_id"] = "claude-opus-4-6"
412 obj = WireCommit.model_validate(commit)
413 assert obj.signature == "sig_base64"
414
415
416 # ---------------------------------------------------------------------------
417 # T5 — Service: wire_push_stream() with stubbed DB and R2 backend
418 # ---------------------------------------------------------------------------
419
420 class TestT5ServiceStream:
421 """Tier 5: wire_push_stream() async generator against minimal stubs.
422
423 We patch the storage backend and DB session so tests run without
424 infrastructure — only the frame-parsing and protocol state machine
425 are exercised here.
426 """
427
428 @pytest.fixture()
429 def stub_backend(self, monkeypatch: pytest.MonkeyPatch) -> MagicMock:
430 backend = AsyncMock()
431 backend.exists = AsyncMock(return_value=False)
432 backend.put = AsyncMock(return_value="https://r2.example.com/obj")
433 backend.get = AsyncMock(return_value=b"raw bytes")
434 monkeypatch.setattr(
435 "musehub.services.musehub_wire.get_backend",
436 lambda: backend,
437 )
438 return backend
439
440 @pytest.fixture()
441 def stub_session(self) -> AsyncMock:
442 session = AsyncMock(spec=AsyncSession)
443 session.execute = AsyncMock(return_value=MagicMock(scalar=lambda: None, fetchall=lambda: []))
444 session.commit = AsyncMock()
445 session.add = MagicMock()
446 return session
447
448 @pytest.mark.asyncio
449 async def test_missing_header_yields_error(
450 self, stub_backend: MagicMock, stub_session: AsyncMock
451 ) -> None:
452 from musehub.services.musehub_wire import wire_push_stream
453
454 async def body():
455 yield _object_frame(*_make_obj_bytes()) + _end_frame()
456
457 frames = await _collect_frames(
458 wire_push_stream(stub_session, "repo-id", body(), "gabriel")
459 )
460 error_frames = [f for f in frames if f.get("t") == SFRAME_ERROR]
461 assert error_frames, "expected an ERROR frame when OBJECT sent before HEADER"
462 assert "OBJECT frame before HEADER" in error_frames[0]["msg"]
463
464 @pytest.mark.asyncio
465 async def test_missing_end_frame_yields_error(
466 self, stub_backend: MagicMock, stub_session: AsyncMock
467 ) -> None:
468 from musehub.services.musehub_wire import wire_push_stream
469
470 commit = _make_commit()
471 snap_id = _sha256_oid(b"snap")
472 snap = _make_snapshot(snap_id)
473 commit["snapshot_id"] = snap_id
474
475 async def body():
476 yield _header_frame() + _commit_pack_frame([commit], [snap])
477 # no END frame
478
479 frames = await _collect_frames(
480 wire_push_stream(stub_session, "repo-id", body(), "gabriel")
481 )
482 error_frames = [f for f in frames if f.get("t") == SFRAME_ERROR]
483 assert error_frames
484 assert "without END frame" in error_frames[0]["msg"]
485
486 @pytest.mark.asyncio
487 async def test_missing_commit_pack_yields_error(
488 self, stub_backend: MagicMock, stub_session: AsyncMock
489 ) -> None:
490 from musehub.services.musehub_wire import wire_push_stream
491
492 async def body():
493 yield _header_frame() + _end_frame()
494
495 frames = await _collect_frames(
496 wire_push_stream(stub_session, "repo-id", body(), "gabriel")
497 )
498 error_frames = [f for f in frames if f.get("t") == SFRAME_ERROR]
499 assert error_frames
500 assert "COMMIT_PACK" in error_frames[0]["msg"]
501
502 @pytest.mark.asyncio
503 async def test_object_hash_mismatch_yields_error(
504 self, stub_backend: MagicMock, stub_session: AsyncMock
505 ) -> None:
506 from musehub.services.musehub_wire import wire_push_stream
507
508 oid = "sha256:" + "a" * 64 # wrong hash
509 content = b"this content does not match the oid"
510
511 async def body():
512 yield _header_frame(n_objects=1) + _object_frame(oid, content)
513
514 frames = await _collect_frames(
515 wire_push_stream(stub_session, "repo-id", body(), "gabriel")
516 )
517 error_frames = [f for f in frames if f.get("t") == SFRAME_ERROR]
518 assert error_frames
519
520 @pytest.mark.asyncio
521 async def test_progress_frames_emitted_on_header(
522 self, stub_backend: MagicMock, stub_session: AsyncMock
523 ) -> None:
524 from musehub.services.musehub_wire import wire_push_stream
525
526 commit = _make_commit()
527 snap_id = _sha256_oid(b"snap-data")
528 snap = _make_snapshot(snap_id)
529 commit["snapshot_id"] = snap_id
530
531 async def body():
532 yield _header_frame() + _commit_pack_frame([commit], [snap]) + _end_frame()
533
534 frames = await _collect_frames(
535 wire_push_stream(stub_session, "repo-id", body(), "gabriel")
536 )
537 progress_frames = [f for f in frames if f.get("t") == SFRAME_PROGRESS]
538 assert progress_frames, "expected at least one PROGRESS frame"
539
540
541 # ---------------------------------------------------------------------------
542 # T6 — Integration: service against real DB, stub R2
543 # ---------------------------------------------------------------------------
544
545 def _stub_r2_backend(monkeypatch: pytest.MonkeyPatch) -> None:
546 """Patch the R2 backend with an in-memory dict store."""
547 _store: dict[str, bytes] = {}
548
549 async def _exists(oid: str) -> bool:
550 return oid in _store
551
552 async def _put(oid: str, data: bytes) -> str:
553 _store[oid] = data
554 return f"https://r2.fake/{oid}"
555
556 async def _get(oid: str) -> bytes | None:
557 return _store.get(oid)
558
559 backend = AsyncMock()
560 backend.exists = _exists
561 backend.put = _put
562 backend.get = _get
563 monkeypatch.setattr(
564 "musehub.services.musehub_wire.get_backend",
565 lambda: backend,
566 )
567
568
569 async def _make_repo(db_session: AsyncSession, name: str, owner: str = "gabriel") -> Any:
570 """Create a repo row + main branch, committed and visible to any session."""
571 import uuid as _uuid_mod
572 from datetime import datetime, timezone
573 from musehub.db.musehub_models import MusehubRepo, MusehubBranch
574 from musehub.core.genesis import compute_repo_id, compute_branch_id
575 owner_user_id = str(_uuid_mod.uuid4())
576 slug = name.lower().replace(" ", "-")
577 created_at = datetime.now(tz=timezone.utc)
578 repo_id = compute_repo_id(owner_user_id, slug, "", created_at.isoformat())
579 repo = MusehubRepo(
580 repo_id=repo_id,
581 name=name,
582 owner=owner,
583 slug=slug,
584 visibility="public",
585 owner_user_id=owner_user_id,
586 description="",
587 tags=[],
588 created_at=created_at,
589 )
590 db_session.add(repo)
591 await db_session.commit()
592 branch = MusehubBranch(
593 branch_id=compute_branch_id(repo_id, "main"),
594 repo_id=repo_id,
595 name="main",
596 )
597 db_session.add(branch)
598 await db_session.commit()
599 await db_session.refresh(repo)
600 return repo
601
602
603 @pytest.mark.asyncio
604 async def test_t6_push_single_commit_no_objects(
605 db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
606 ) -> None:
607 """T6: push one commit with no objects against the real test DB."""
608 from musehub.services.musehub_wire import wire_push_stream
609
610 _stub_r2_backend(monkeypatch)
611 repo = await _make_repo(db_session, "T6 Single Commit")
612
613 snap_id = _sha256_oid(b"t6-snap-1")
614 commit = _make_commit(snapshot_id=snap_id)
615 snap = _make_snapshot(snap_id)
616
617 async def body():
618 yield _header_frame(n_commits=1) + _commit_pack_frame([commit], [snap]) + _end_frame()
619
620 frames = await _collect_frames(
621 wire_push_stream(db_session, str(repo.repo_id), body(), "gabriel")
622 )
623 result_frames = [f for f in frames if f.get("t") == SFRAME_RESULT]
624 assert result_frames, f"no RESULT frame; frames: {[f.get('t') for f in frames]}"
625 assert result_frames[0]["ok"] is True
626
627
628 @pytest.mark.asyncio
629 async def test_t6_push_with_object(
630 db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
631 ) -> None:
632 """T6: push one object + commit; confirm RESULT.ok."""
633 from musehub.services.musehub_wire import wire_push_stream
634
635 _stub_r2_backend(monkeypatch)
636 repo = await _make_repo(db_session, "T6 With Object")
637
638 raw = b"audio data for test track"
639 oid = _sha256_oid(raw)
640 snap_id = _sha256_oid(b"t6-snap-obj")
641 commit = _make_commit(snapshot_id=snap_id)
642 snap = _make_snapshot(snap_id, {"track.wav": oid})
643
644 async def body():
645 yield (
646 _header_frame(n_objects=1, n_commits=1)
647 + _object_frame(oid, raw)
648 + _commit_pack_frame([commit], [snap])
649 + _end_frame()
650 )
651
652 frames = await _collect_frames(
653 wire_push_stream(db_session, str(repo.repo_id), body(), "gabriel")
654 )
655 result_frames = [f for f in frames if f.get("t") == SFRAME_RESULT]
656 assert result_frames and result_frames[0]["ok"] is True
657
658
659 @pytest.mark.asyncio
660 async def test_t6_push_zlib_compressed_object(
661 db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
662 ) -> None:
663 """T6: push a zlib-compressed object; server decompresses and confirms."""
664 from musehub.services.musehub_wire import wire_push_stream
665
666 _stub_r2_backend(monkeypatch)
667 repo = await _make_repo(db_session, "T6 Zlib")
668
669 raw = b"raw MIDI data " * 100
670 compressed = zlib.compress(raw)
671 oid = _sha256_oid(raw)
672 snap_id = _sha256_oid(b"t6-snap-zlib")
673 commit = _make_commit(snapshot_id=snap_id)
674 snap = _make_snapshot(snap_id, {"beat.mid": oid})
675
676 async def body():
677 yield (
678 _header_frame(n_objects=1, n_commits=1)
679 + _object_frame(oid, compressed, enc="zlib")
680 + _commit_pack_frame([commit], [snap])
681 + _end_frame()
682 )
683
684 frames = await _collect_frames(
685 wire_push_stream(db_session, str(repo.repo_id), body(), "gabriel")
686 )
687 result = [f for f in frames if f.get("t") == SFRAME_RESULT]
688 assert result and result[0]["ok"] is True
689
690
691 @pytest.mark.asyncio
692 async def test_t6_force_push_advances_branch(
693 db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
694 ) -> None:
695 """T6: two sequential pushes — second uses force=True to advance divergent branch."""
696 from musehub.services.musehub_wire import wire_push_stream
697
698 _stub_r2_backend(monkeypatch)
699 repo = await _make_repo(db_session, "T6 Force Push")
700
701 snap_id = _sha256_oid(b"t6-snap-force-1")
702 commit1 = _make_commit(snapshot_id=snap_id)
703 snap1 = _make_snapshot(snap_id)
704
705 async def body1():
706 yield _header_frame() + _commit_pack_frame([commit1], [snap1]) + _end_frame()
707
708 frames1 = await _collect_frames(
709 wire_push_stream(db_session, str(repo.repo_id), body1(), "gabriel")
710 )
711 assert any(f.get("t") == SFRAME_RESULT and f["ok"] for f in frames1)
712
713 snap_id2 = _sha256_oid(b"t6-snap-force-2")
714 commit2 = _make_commit(snapshot_id=snap_id2)
715 snap2 = _make_snapshot(snap_id2)
716
717 async def body2():
718 yield _header_frame(force=True) + _commit_pack_frame([commit2], [snap2]) + _end_frame()
719
720 frames2 = await _collect_frames(
721 wire_push_stream(db_session, str(repo.repo_id), body2(), "gabriel")
722 )
723 result2 = [f for f in frames2 if f.get("t") == SFRAME_RESULT]
724 assert result2 and result2[0]["ok"] is True
725
726
727 # ---------------------------------------------------------------------------
728 # T7 — Route: ASGI test client hitting POST /{owner}/{slug}/push/stream
729 # ---------------------------------------------------------------------------
730
731 @pytest.mark.asyncio
732 async def test_t7_push_stream_returns_200_with_packstream_content_type(
733 client: AsyncClient, db_session: AsyncSession, auth_headers: dict,
734 monkeypatch: pytest.MonkeyPatch,
735 ) -> None:
736 """T7: route returns 200 with application/x-muse-packstream content-type."""
737 _stub_r2_backend(monkeypatch)
738 repo = await _make_repo(db_session, "T7 Route Test 1", owner="testuser")
739
740 snap_id = _sha256_oid(b"t7-snap-ct")
741 commit = _make_commit(snapshot_id=snap_id)
742 snap = _make_snapshot(snap_id)
743 body = _header_frame() + _commit_pack_frame([commit], [snap]) + _end_frame()
744
745 resp = await client.post(
746 f"/{repo.owner}/{repo.slug}/push/stream",
747 content=body,
748 headers={**auth_headers, "Content-Type": GRPC_CONTENT_TYPE},
749 )
750 assert resp.status_code == 200
751 assert GRPC_CONTENT_TYPE in resp.headers.get("content-type", "")
752
753
754 @pytest.mark.asyncio
755 async def test_t7_push_stream_response_contains_result_frame(
756 client: AsyncClient, db_session: AsyncSession, auth_headers: dict,
757 monkeypatch: pytest.MonkeyPatch,
758 ) -> None:
759 """T7: response body is a frame stream that ends with a RESULT frame."""
760 _stub_r2_backend(monkeypatch)
761 repo = await _make_repo(db_session, "T7 Route Test 2", owner="testuser")
762
763 snap_id = _sha256_oid(b"t7-snap-result")
764 commit = _make_commit(snapshot_id=snap_id)
765 snap = _make_snapshot(snap_id)
766 body = _header_frame() + _commit_pack_frame([commit], [snap]) + _end_frame()
767
768 resp = await client.post(
769 f"/{repo.owner}/{repo.slug}/push/stream",
770 content=body,
771 headers={**auth_headers, "Content-Type": GRPC_CONTENT_TYPE},
772 )
773 frames = _unpack_all(resp.content)
774 result_frames = [f for f in frames if f.get("t") == SFRAME_RESULT]
775 assert result_frames, f"no RESULT frame; frames: {[f.get('t') for f in frames]}"
776
777
778 @pytest.mark.asyncio
779 async def test_t7_push_stream_requires_auth(
780 client: AsyncClient, db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch,
781 ) -> None:
782 """T7: unauthenticated push yields error."""
783 _stub_r2_backend(monkeypatch)
784 repo = await _make_repo(db_session, "T7 Route Auth", owner="testuser")
785
786 body = _header_frame() + _end_frame()
787 resp = await client.post(
788 f"/{repo.owner}/{repo.slug}/push/stream",
789 content=body,
790 headers={"Content-Type": GRPC_CONTENT_TYPE},
791 )
792 assert resp.status_code in (200, 401, 403)
793 if resp.status_code == 200:
794 frames = _unpack_all(resp.content)
795 error_frames = [f for f in frames if f.get("t") == SFRAME_ERROR]
796 assert error_frames, "unauthenticated push should yield error frame"
797
798
799 @pytest.mark.asyncio
800 async def test_t7_push_stream_404_for_missing_repo(
801 client: AsyncClient, auth_headers: dict,
802 ) -> None:
803 """T7: push to a repo that doesn't exist yields 404 or error frame."""
804 body = _header_frame() + _end_frame()
805 resp = await client.post(
806 "/gabriel/nonexistent-repo-xyz-t7/push/stream",
807 content=body,
808 headers={**auth_headers, "Content-Type": GRPC_CONTENT_TYPE},
809 )
810 if resp.status_code == 200:
811 frames = _unpack_all(resp.content)
812 err = [f for f in frames if f.get("t") == SFRAME_ERROR]
813 assert err and err[0].get("code") == 404
814 else:
815 assert resp.status_code == 404
816
817
818 @pytest.mark.asyncio
819 async def test_t7_old_push_endpoints_deleted(
820 client: AsyncClient, db_session: AsyncSession, auth_headers: dict,
821 ) -> None:
822 """T7: all MWP v1 push endpoints return 404 — they were deleted in MWP v2."""
823 repo = await _make_repo(db_session, "T7 Route v1 Deleted", owner="testuser")
824
825 deleted_paths = [
826 f"/{repo.owner}/{repo.slug}/filter-objects",
827 f"/{repo.owner}/{repo.slug}/presign-objects",
828 f"/{repo.owner}/{repo.slug}/push/objects",
829 f"/{repo.owner}/{repo.slug}/push/object-pack",
830 f"/{repo.owner}/{repo.slug}/push/objects/confirm",
831 f"/{repo.owner}/{repo.slug}/push",
832 ]
833 for path in deleted_paths:
834 resp = await client.post(path, headers=auth_headers, content=b"{}")
835 assert resp.status_code == 404, (
836 f"Expected 404 for deleted endpoint {path}, got {resp.status_code}"
837 )
838
839
840 # ---------------------------------------------------------------------------
841 # T8 — E2E: full push → GET /refs confirms branch head updated
842 # ---------------------------------------------------------------------------
843
844 @pytest.mark.asyncio
845 async def test_t8_push_then_refs_show_commit(
846 client: AsyncClient, db_session: AsyncSession, auth_headers: dict,
847 ) -> None:
848 """T8: push one commit; GET /refs confirms branch head updated."""
849 repo = await _make_repo(db_session, "T8 E2E Refs", owner="testuser")
850
851 snap_id = _sha256_oid(b"t8-e2e-snap-1")
852 commit = _make_commit(snapshot_id=snap_id)
853 snap = _make_snapshot(snap_id)
854 body = (
855 _header_frame(head=commit["commit_id"])
856 + _commit_pack_frame([commit], [snap])
857 + _end_frame()
858 )
859
860 push_resp = await client.post(
861 f"/{repo.owner}/{repo.slug}/push/stream",
862 content=body,
863 headers={**auth_headers, "Content-Type": GRPC_CONTENT_TYPE},
864 )
865 assert push_resp.status_code == 200
866 frames = _unpack_all(push_resp.content)
867 result = next((f for f in frames if f.get("t") == SFRAME_RESULT), None)
868 assert result is not None, f"no RESULT frame; got: {[f.get('t') for f in frames]}"
869 assert result["ok"] is True
870
871 refs_resp = await client.get(
872 f"/{repo.owner}/{repo.slug}/refs",
873 headers=auth_headers,
874 )
875 assert refs_resp.status_code == 200
876 branch_heads = refs_resp.json().get("branch_heads", {})
877 assert branch_heads.get("main") == commit["commit_id"], (
878 f"Expected main head={commit['commit_id']!r}, got {branch_heads}"
879 )
880
881
882 @pytest.mark.asyncio
883 async def test_t8_push_with_objects_then_fetch_objects(
884 client: AsyncClient, db_session: AsyncSession, auth_headers: dict,
885 ) -> None:
886 """T8: push an object; fetch it back and verify content integrity."""
887 repo = await _make_repo(db_session, "T8 E2E Fetch", owner="testuser")
888
889 raw = b"audio track bytes for e2e test"
890 oid = _sha256_oid(raw)
891 snap_id = _sha256_oid(b"t8-e2e-obj-snap")
892 commit = _make_commit(snapshot_id=snap_id)
893 snap = _make_snapshot(snap_id, {"track.wav": oid})
894 body = (
895 _header_frame(n_objects=1)
896 + _object_frame(oid, raw)
897 + _commit_pack_frame([commit], [snap])
898 + _end_frame()
899 )
900
901 push_resp = await client.post(
902 f"/{repo.owner}/{repo.slug}/push/stream",
903 content=body,
904 headers={**auth_headers, "Content-Type": GRPC_CONTENT_TYPE},
905 )
906 assert push_resp.status_code == 200
907 result = next(
908 (f for f in _unpack_all(push_resp.content) if f.get("t") == SFRAME_RESULT), None
909 )
910 assert result and result["ok"] is True
911
912 fetch_resp = await client.get(
913 f"/o/{oid}",
914 headers=auth_headers,
915 )
916 assert fetch_resp.status_code == 200
917 assert fetch_resp.content == raw
918
919
920 @pytest.mark.asyncio
921 async def test_t8_push_chain_of_commits_parent_linking(
922 client: AsyncClient, db_session: AsyncSession, auth_headers: dict,
923 ) -> None:
924 """T8: push two chained commits; branch head advances to the child."""
925 repo = await _make_repo(db_session, "T8 E2E Chain", owner="testuser")
926
927 snap1_id = _sha256_oid(b"t8-e2e-snap-chain-1")
928 commit1 = _make_commit(snapshot_id=snap1_id)
929 snap1 = _make_snapshot(snap1_id)
930
931 snap2_id = _sha256_oid(b"t8-e2e-snap-chain-2")
932 commit2 = _make_commit(snapshot_id=snap2_id, parent_ids=[commit1["commit_id"]])
933 snap2 = _make_snapshot(snap2_id)
934
935 body = (
936 _header_frame(n_commits=2, head=commit2["commit_id"])
937 + _commit_pack_frame([commit1, commit2], [snap1, snap2])
938 + _end_frame()
939 )
940
941 push_resp = await client.post(
942 f"/{repo.owner}/{repo.slug}/push/stream",
943 content=body,
944 headers={**auth_headers, "Content-Type": GRPC_CONTENT_TYPE},
945 )
946 assert push_resp.status_code == 200
947 result = next(
948 (f for f in _unpack_all(push_resp.content) if f.get("t") == SFRAME_RESULT), None
949 )
950 assert result and result["ok"] is True, f"push failed: {result}"
951
952 refs_resp = await client.get(
953 f"/{repo.owner}/{repo.slug}/refs",
954 headers=auth_headers,
955 )
956 refs = refs_resp.json().get("branch_heads", {})
957 assert refs.get("main") == commit2["commit_id"]
958
959
960 # ---------------------------------------------------------------------------
961 # T9 — Regression: MPackStreamWriter frames with compressed binary content
962 # must not raise UnicodeDecodeError on the server.
963 #
964 # Root cause of staging bug:
965 # 'utf-8' codec can't decode byte 0xad in position 2: invalid start byte
966 #
967 # The client sends O frames where the "content" field is zlib-compressed
968 # binary bytes packed with use_bin_type=True (msgpack bin type 0xc4/c5/c6).
969 # If the server Unpacker is misconfigured (raw=True, or content field encoded
970 # as str/fixstr), raw=False raises UnicodeDecodeError on non-UTF-8 bytes.
971 #
972 # These tests use MPackStreamWriter (the actual client encoder) to build the
973 # exact bytes the client sends, then POST them through the ASGI app.
974 # ---------------------------------------------------------------------------
975
976 @pytest.mark.asyncio
977 async def test_t9_mpackstreamwriter_compressed_frames_decode_clean(
978 client: AsyncClient, db_session: AsyncSession, auth_headers: dict,
979 ) -> None:
980 """T9: server must decode MPackStreamWriter O frames with zlib-compressed binary content.
981
982 Regression for: 'utf-8' codec can't decode byte 0xad in position 2.
983 Client uses MPackStreamWriter (use_bin_type=True); server Unpacker uses raw=False.
984 Binary content including byte 0xad must arrive as bytes, not trigger UnicodeDecodeError.
985 """
986 from muse.core.mpack import MPackStreamWriter
987 from muse.core._types import blob_id
988
989 repo = await _make_repo(db_session, "T9 Compressed Binary Frames", owner="testuser")
990 w = MPackStreamWriter()
991
992 # Content that includes 0xad and other non-UTF-8 bytes — the exact failing case
993 content = bytes(range(256)) * 10
994 oid = blob_id(content)
995
996 snap_id = _sha256_oid(b"t9-snap-compressed")
997 commit = _make_commit(snapshot_id=snap_id)
998 snap = _make_snapshot(snap_id, {"file.bin": oid})
999
1000 body = (
1001 grpc_frame(_fw.wrap(frame_type="H", payload=w.write_header(op="push", branch="main", n_objects=1, n_commits=1)))
1002 + grpc_frame(_fw.wrap(frame_type="O", payload=w.write_object_raw(object_id=oid, raw_bytes=content, compress="zlib")))
1003 + grpc_frame(_fw.wrap(frame_type="C", payload=w.write_commit_pack(commits=[commit], snapshots=[snap])))
1004 + grpc_frame(_fw.wrap(frame_type="E", payload=w.write_end(n_objects=1, n_commits=1)))
1005 )
1006
1007 resp = await client.post(
1008 f"/{repo.owner}/{repo.slug}/push/stream",
1009 content=body,
1010 headers={**auth_headers, "Content-Type": GRPC_CONTENT_TYPE},
1011 )
1012 assert resp.status_code == 200
1013 frames = _unpack_all(resp.content)
1014
1015 error_frames = [f for f in frames if f.get("t") == SFRAME_ERROR]
1016 assert not error_frames, (
1017 f"Server returned error frame(s): {[f.get('msg') for f in error_frames]}"
1018 )
1019 result = next((f for f in frames if f.get("t") == SFRAME_RESULT), None)
1020 assert result is not None, f"No RESULT frame; got: {[f.get('t') for f in frames]}"
1021 assert result["ok"] is True
1022
1023
1024 @pytest.mark.asyncio
1025 async def test_t9_920_objects_with_binary_content_no_unicode_error(
1026 client: AsyncClient, db_session: AsyncSession, auth_headers: dict,
1027 ) -> None:
1028 """T9: 920 objects with full byte range (0x00-0xff) — none may produce UnicodeDecodeError.
1029
1030 Reproduces the staging scenario: ~900 small objects each containing binary
1031 data. The server must process all O frames without any 'utf-8 codec' error.
1032 """
1033 from muse.core.mpack import MPackStreamWriter
1034 from muse.core._types import blob_id
1035
1036 n = 920
1037 repo = await _make_repo(db_session, "T9 920 Binary Objects", owner="testuser")
1038 w = MPackStreamWriter()
1039
1040 snap_id = _sha256_oid(b"t9-snap-920")
1041 commit = _make_commit(snapshot_id=snap_id)
1042 manifest = {}
1043 parts = [grpc_frame(_fw.wrap(frame_type="H", payload=w.write_header(op="push", branch="main", n_objects=n, n_commits=1)))]
1044
1045 for i in range(n):
1046 raw_content = (bytes(range(256)) * 2)[i % 256: i % 256 + 256] + i.to_bytes(4, "big")
1047 oid = blob_id(raw_content)
1048 manifest[f"file_{i}.bin"] = oid
1049 parts.append(grpc_frame(_fw.wrap(frame_type="O", payload=w.write_object_raw(object_id=oid, raw_bytes=raw_content, compress="zlib"))))
1050
1051 snap = _make_snapshot(snap_id, manifest)
1052 parts.append(grpc_frame(_fw.wrap(frame_type="C", payload=w.write_commit_pack(commits=[commit], snapshots=[snap]))))
1053 parts.append(grpc_frame(_fw.wrap(frame_type="E", payload=w.write_end(n_objects=n, n_commits=1))))
1054
1055 body = b"".join(parts)
1056 resp = await client.post(
1057 f"/{repo.owner}/{repo.slug}/push/stream",
1058 content=body,
1059 headers={**auth_headers, "Content-Type": GRPC_CONTENT_TYPE},
1060 )
1061 assert resp.status_code == 200
1062 frames = _unpack_all(resp.content)
1063
1064 error_frames = [f for f in frames if f.get("t") == SFRAME_ERROR]
1065 assert not error_frames, (
1066 f"Server returned error frame(s): {[f.get('msg') for f in error_frames]}"
1067 )
1068 result = next((f for f in frames if f.get("t") == SFRAME_RESULT), None)
1069 assert result is not None, f"No RESULT frame; got: {[f.get('t') for f in frames]}"
1070 assert result["ok"] is True
1071
1072
1073 # ---------------------------------------------------------------------------
1074 # T10 — Regression: server response frames must use only string map keys.
1075 #
1076 # Root cause of staging bug:
1077 # stream read error: int is not allowed for map key when strict_map_key=True
1078 #
1079 # MPackStreamReader (muse/core/mpack.py) uses msgpack.Unpacker with the
1080 # default strict_map_key=True. If the server sends any frame where a map
1081 # key is an integer, the client raises and the push fails.
1082 #
1083 # This test decodes the server response with strict_map_key=True — the exact
1084 # setting the client uses — and asserts every key in every frame is a str.
1085 # ---------------------------------------------------------------------------
1086
1087 def _unpack_all_strict(raw: bytes) -> list:
1088 """Decode all msgpack frames with strict_map_key=True (same as MPackStreamReader)."""
1089 unpacker = msgpack.Unpacker(raw=False, strict_map_key=True)
1090 unpacker.feed(raw)
1091 return list(unpacker)
1092
1093
1094 @pytest.mark.asyncio
1095 async def test_t10_response_frames_use_only_string_map_keys(
1096 client: AsyncClient, db_session: AsyncSession, auth_headers: dict,
1097 ) -> None:
1098 """T10: all server response frame map keys must be strings.
1099
1100 MPackStreamReader uses strict_map_key=True (msgpack default). Any integer
1101 key in a server frame raises 'int is not allowed for map key' on the client
1102 and aborts the push. This test uses the same strict decoder to catch the
1103 mismatch at the server level before it reaches production.
1104 """
1105 repo = await _make_repo(db_session, "T10 String Map Keys", owner="testuser")
1106
1107 snap_id = _sha256_oid(b"t10-snap-1")
1108 commit = _make_commit(snapshot_id=snap_id)
1109 snap = _make_snapshot(snap_id)
1110 body = _header_frame() + _commit_pack_frame([commit], [snap]) + _end_frame()
1111
1112 resp = await client.post(
1113 f"/{repo.owner}/{repo.slug}/push/stream",
1114 content=body,
1115 headers={**auth_headers, "Content-Type": GRPC_CONTENT_TYPE},
1116 )
1117 assert resp.status_code == 200
1118
1119 # Use the same strict decoder the client uses — must not raise.
1120 try:
1121 frames = _unpack_all_strict(resp.content)
1122 except Exception as exc:
1123 raise AssertionError(
1124 f"Server response failed strict msgpack decode (same as MPackStreamReader): {exc}\n"
1125 f"Raw response (first 512 bytes): {resp.content[:512]!r}"
1126 ) from exc
1127
1128 # Belt-and-suspenders: assert every key in every frame dict is a str.
1129 for i, frame in enumerate(frames):
1130 if not isinstance(frame, dict):
1131 continue
1132 int_keys = [k for k in frame if not isinstance(k, str)]
1133 assert not int_keys, (
1134 f"Frame {i} (t={frame.get('t')!r}) has integer map keys: {int_keys!r}\n"
1135 f"Full frame: {frame!r}"
1136 )
File History 1 commit
sha256:9590cee1e0ccd6c76528f005b95d634d80f5019f0dcb7c371e149adc31d1fb65 refactor: enforce gRPC framing on all MWP wire traffic Sonnet 4.6 minor ⚠ 156 days ago