gabriel / musehub public
test_wire_push_stream.py python
1,516 lines 56.4 KB
Raw
sha256:bb2baaabdd19320bde50cb69d447fd1c1e571467df729be1a23b5e93064b046a feat(intel): standardize headers, gauge icon, velocity card… Sonnet 4.6 minor ⚠ breaking 143 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 zlib
22 from collections.abc import AsyncGenerator, AsyncIterator
23 from datetime import datetime, timezone
24 from unittest.mock import AsyncMock, MagicMock, patch
25
26 import msgpack
27 import pytest
28 from httpx import AsyncClient
29 from sqlalchemy.ext.asyncio import AsyncSession
30
31 from muse.core.types import blob_id, fake_id
32 from musehub.db.musehub_models import MusehubRepo
33 from musehub.types.json_types import JSONObject, JSONValue, StrDict
34 from musehub.models.wire import (
35 SFRAME_COMMIT_PACK,
36 SFRAME_END,
37 SFRAME_ERROR,
38 SFRAME_HEADER,
39 SFRAME_OBJECT,
40 SFRAME_PROGRESS,
41 SFRAME_RESULT,
42 STREAM_MAX_COMMITS,
43 STREAM_MAX_OBJECT_WIRE_BYTES,
44 STREAM_MAX_OBJECTS,
45 )
46 from muse.core.mpack import WIRE_CONTENT_TYPE, MuseWireFrameWriter
47 from tests.factories import create_repo
48
49 _fw = MuseWireFrameWriter()
50
51
52 # ---------------------------------------------------------------------------
53 # Shared codec helpers
54 # ---------------------------------------------------------------------------
55
56 def _pack(data: JSONValue) -> bytes:
57 """Encode one msgpack frame payload (without transport envelope)."""
58 return msgpack.packb(data, use_bin_type=True)
59
60
61 def _wrap(ft: str, data: JSONValue) -> bytes:
62 """Encode and wrap in a raw MWP envelope — no gRPC prefix."""
63 return _fw.wrap(frame_type=ft, payload=_pack(data))
64
65
66 def _unpack_all(raw: bytes) -> list[dict]:
67 """Decode concatenated raw MWP frames into a list of payload dicts.
68
69 Each MWP frame: magic(4=b"muse") | version(1) | header_len(4) |
70 header(N) | payload_len(8) | payload(M)
71 """
72 import struct
73 results = []
74 offset = 0
75 while offset + 17 <= len(raw):
76 if raw[offset:offset + 4] != b"muse":
77 break
78 header_len = struct.unpack(">I", raw[offset + 5:offset + 9])[0]
79 pl_start = offset + 9 + header_len
80 if pl_start + 8 > len(raw):
81 break
82 payload_len = struct.unpack(">Q", raw[pl_start:pl_start + 8])[0]
83 payload = raw[pl_start + 8:pl_start + 8 + payload_len]
84 results.append(msgpack.unpackb(payload, raw=False))
85 offset = pl_start + 8 + payload_len
86 return results
87
88
89 def _last_frame(raw: bytes) -> JSONObject:
90 """Return the last msgpack object from a push-stream response body."""
91 unpacker = msgpack.Unpacker(raw=False)
92 unpacker.feed(raw)
93 last: JSONObject = {}
94 for frame in unpacker:
95 last = frame
96 return last
97
98
99 def _sha256_oid(raw: bytes) -> str:
100 return blob_id(raw)
101
102
103 def _utc() -> str:
104 return datetime.now(tz=timezone.utc).isoformat()
105
106
107 def _make_obj_bytes(content: bytes = b"hello world") -> tuple[str, bytes]:
108 """Return (sha256_oid, raw_content) for a test object."""
109 oid = _sha256_oid(content)
110 return oid, content
111
112
113 def _header_frame(
114 branch: str = "main",
115 force: bool = False,
116 have: list[str] | None = None,
117 head: str = "sha256:abc",
118 n_objects: int = 0,
119 n_commits: int = 1,
120 ) -> bytes:
121 return _wrap(SFRAME_HEADER, {
122 "t": SFRAME_HEADER,
123 "branch": branch,
124 "force": force,
125 "have": have or [],
126 "head": head,
127 "n_objects": n_objects,
128 "n_commits": n_commits,
129 })
130
131
132 def _object_frame(oid: str, content: bytes, path: str = "track.wav", enc: str = "raw") -> bytes:
133 return _wrap(SFRAME_OBJECT, {
134 "t": SFRAME_OBJECT,
135 "id": oid,
136 "content": content,
137 "path": path,
138 "enc": enc,
139 })
140
141
142 def _commit_pack_frame(commits: list[dict], snapshots: list[dict] | None = None) -> bytes:
143 return _wrap(SFRAME_COMMIT_PACK, {
144 "t": SFRAME_COMMIT_PACK,
145 "commits": commits,
146 "snapshots": snapshots or [],
147 })
148
149
150 def _end_frame(n_objects: int = 0, n_commits: int = 0) -> bytes:
151 return _wrap(SFRAME_END, {"t": SFRAME_END, "n_objects": n_objects, "n_commits": n_commits})
152
153
154 def _make_commit(
155 commit_id: str | None = None,
156 parent_ids: list[str] | None = None,
157 snapshot_id: str | None = None,
158 branch: str = "main",
159 author: str = "gabriel",
160 ) -> JSONObject:
161 cid = commit_id or _sha256_oid(f"commit-{_utc()}".encode())
162 pids = parent_ids or []
163 return {
164 "commit_id": cid,
165 "parent_ids": pids,
166 # WireCommit uses parent_commit_id / parent2_commit_id, not parent_ids
167 "parent_commit_id": pids[0] if len(pids) > 0 else None,
168 "parent2_commit_id": pids[1] if len(pids) > 1 else None,
169 "snapshot_id": snapshot_id or _sha256_oid(b"default-snap"),
170 "branch": branch,
171 "message": "test commit",
172 "author": author,
173 "committed_at": _utc(),
174 "signature": "",
175 "signer_key_id": "",
176 "agent_id": "",
177 "model_id": "",
178 "metadata": {},
179 }
180
181
182 def _make_snapshot(snapshot_id: str, manifest: JSONObject | None = None) -> JSONObject:
183 return {
184 "snapshot_id": snapshot_id,
185 "manifest": manifest or {},
186 "committed_at": _utc(),
187 }
188
189
190 async def _collect_frames(gen: AsyncGenerator[bytes, None]) -> list[dict]:
191 """Drain an async generator of plain msgpack frame bytes into a list of dicts."""
192 unpacker = msgpack.Unpacker(raw=False)
193 async for chunk in gen:
194 unpacker.feed(chunk)
195 return list(unpacker)
196
197
198 async def _body_iter(*frames: bytes) -> AsyncIterator[bytes]:
199 """Wrap pre-built frames as an async iterator for wire_push_stream()."""
200 for f in frames:
201 yield f
202
203
204 # ---------------------------------------------------------------------------
205 # T1 — Unit: frame codec helpers
206 # ---------------------------------------------------------------------------
207
208 class TestT1FrameCodec:
209 """Tier 1: verify frame construction and MIME constant."""
210
211 def test_stream_mime_type(self) -> None:
212 assert WIRE_CONTENT_TYPE == "application/x-muse-wire"
213
214 def test_sframe_constants_are_single_chars(self) -> None:
215 for const in (
216 SFRAME_HEADER, SFRAME_OBJECT, SFRAME_COMMIT_PACK, SFRAME_END,
217 SFRAME_PROGRESS, SFRAME_ERROR, SFRAME_RESULT,
218 ):
219 assert len(const) == 1, f"{const!r} should be a single character"
220
221 def test_client_server_frame_tags_are_disjoint(self) -> None:
222 client_tags = {SFRAME_HEADER, SFRAME_OBJECT, SFRAME_COMMIT_PACK, SFRAME_END}
223 server_tags = {SFRAME_PROGRESS, SFRAME_ERROR, SFRAME_RESULT}
224 assert client_tags.isdisjoint(server_tags), (
225 "client and server frame tags must not overlap"
226 )
227
228 def test_header_frame_round_trips(self) -> None:
229 raw = _header_frame(branch="dev", n_objects=3, n_commits=2)
230 frames = _unpack_all(raw)
231 assert len(frames) == 1
232 f = frames[0]
233 assert f["t"] == SFRAME_HEADER
234 assert f["branch"] == "dev"
235 assert f["n_objects"] == 3
236 assert f["n_commits"] == 2
237
238 def test_object_frame_round_trips(self) -> None:
239 oid, content = _make_obj_bytes(b"muse track data")
240 raw = _object_frame(oid, content, path="beat.mid")
241 frames = _unpack_all(raw)
242 f = frames[0]
243 assert f["t"] == SFRAME_OBJECT
244 assert f["id"] == oid
245 assert bytes(f["content"]) == content
246
247 def test_commit_pack_frame_round_trips(self) -> None:
248 commit = _make_commit()
249 raw = _commit_pack_frame([commit])
250 frames = _unpack_all(raw)
251 f = frames[0]
252 assert f["t"] == SFRAME_COMMIT_PACK
253 assert len(f["commits"]) == 1
254
255 def test_end_frame_round_trips(self) -> None:
256 frames = _unpack_all(_end_frame())
257 assert frames[0]["t"] == SFRAME_END
258
259 def test_multiple_frames_concatenated_parse_correctly(self) -> None:
260 body = (
261 _header_frame()
262 + _object_frame(*_make_obj_bytes())
263 + _end_frame()
264 )
265 frames = _unpack_all(body)
266 assert [f["t"] for f in frames] == [SFRAME_HEADER, SFRAME_OBJECT, SFRAME_END]
267
268 def test_stream_limits_are_positive(self) -> None:
269 assert STREAM_MAX_OBJECTS > 0
270 assert STREAM_MAX_COMMITS > 0
271 assert STREAM_MAX_OBJECT_WIRE_BYTES > 0
272
273
274 # ---------------------------------------------------------------------------
275 # T2 — Unit: server helper functions
276 # ---------------------------------------------------------------------------
277
278 class TestT2ServerHelpers:
279 """Tier 2: test _sp, _prog, _err, _result frame builders."""
280
281 def _import_helpers(self) -> None:
282 from musehub.services.musehub_wire import _sp, _prog, _err, _result
283 return _sp, _prog, _err, _result
284
285 def test_prog_encodes_progress_frame(self) -> None:
286 _, _prog, _, _ = self._import_helpers()
287 f = msgpack.unpackb(_prog("uploading objects"), raw=False)
288 assert f["t"] == SFRAME_PROGRESS
289 assert f["msg"] == "uploading objects"
290
291 def test_err_encodes_error_frame_with_code(self) -> None:
292 _, _, _err, _ = self._import_helpers()
293 f = msgpack.unpackb(_err("repo not found", 404), raw=False)
294 assert f["t"] == SFRAME_ERROR
295 assert f["code"] == 404
296 assert "repo not found" in f["msg"]
297
298 def test_err_default_code_is_400(self) -> None:
299 _, _, _err, _ = self._import_helpers()
300 f = msgpack.unpackb(_err("bad request"), raw=False)
301 assert f["code"] == 400
302
303 def test_result_ok_encodes_correctly(self) -> None:
304 _, _, _, _result = self._import_helpers()
305 heads = {"main": "sha256:abc"}
306 f = msgpack.unpackb(_result(True, "pushed", heads, "sha256:abc"), raw=False)
307 assert f["t"] == SFRAME_RESULT
308 assert f["ok"] is True
309 assert f["heads"] == heads
310
311 def test_result_failure_encodes_correctly(self) -> None:
312 _, _, _, _result = self._import_helpers()
313 f = msgpack.unpackb(_result(False, "rejected", {}, ""), raw=False)
314 assert f["ok"] is False
315
316
317 # ---------------------------------------------------------------------------
318 # T3 — Component: object validation edge cases
319 # ---------------------------------------------------------------------------
320
321 class TestT3ObjectValidation:
322 """Tier 3: hash mismatch, size limits, compression encoding."""
323
324 def _oid_for(self, raw: bytes) -> str:
325 return _sha256_oid(raw)
326
327 def test_sha256_oid_format(self) -> None:
328 raw = b"guitar riff"
329 oid = self._oid_for(raw)
330 assert oid.startswith("sha256:")
331 assert len(oid) == len("sha256:") + 64
332
333 def test_zlib_object_frame_decompresses_correctly(self) -> None:
334 raw = b"MIDI note data " * 50
335 compressed = zlib.compress(raw)
336 oid, _ = _make_obj_bytes(raw) # sha256 of raw (before compression)
337 frame_bytes = _object_frame(oid, compressed, enc="zlib")
338 frames = _unpack_all(frame_bytes)
339 f = frames[0]
340 assert f["enc"] == "zlib"
341 assert zlib.decompress(bytes(f["content"])) == raw
342
343 def test_wire_size_limit_constant_is_reasonable(self) -> None:
344 # Must be at least 1 MB and at most 512 MB per object wire payload.
345 assert 1 * 1024 * 1024 <= STREAM_MAX_OBJECT_WIRE_BYTES <= 512 * 1024 * 1024
346
347 def test_object_frame_with_empty_content_is_encodable(self) -> None:
348 raw = b""
349 oid = _sha256_oid(raw)
350 frame_bytes = _object_frame(oid, raw)
351 frames = _unpack_all(frame_bytes)
352 assert bytes(frames[0]["content"]) == b""
353
354 def test_large_object_frame_exceeds_limit_is_detectable(self) -> None:
355 """Frame content larger than STREAM_MAX_OBJECT_WIRE_BYTES should be flagged."""
356 oversized = b"x" * (STREAM_MAX_OBJECT_WIRE_BYTES + 1)
357 oid = _sha256_oid(oversized)
358 frame = _unpack_all(_object_frame(oid, oversized))[0]
359 assert len(bytes(frame["content"])) > STREAM_MAX_OBJECT_WIRE_BYTES
360
361 def test_raw_encoding_preserves_exact_bytes(self) -> None:
362 raw = b"\x00\x01\x02\x03" * 100
363 oid = _sha256_oid(raw)
364 frame = _unpack_all(_object_frame(oid, raw, enc="raw"))[0]
365 assert bytes(frame["content"]) == raw
366
367
368 # ---------------------------------------------------------------------------
369 # T4 — Component: commit-pack schema validation
370 # ---------------------------------------------------------------------------
371
372 class TestT4CommitPackValidation:
373 """Tier 4: WireCommit/WireSnapshot schema, commit-count limit."""
374
375 def test_minimal_commit_passes_model_validate(self) -> None:
376 from musehub.models.wire import WireCommit
377 commit = _make_commit()
378 obj = WireCommit.model_validate(commit)
379 assert obj.commit_id == commit["commit_id"]
380
381 def test_commit_without_required_fields_raises(self) -> None:
382 from musehub.models.wire import WireCommit
383 import pydantic
384 with pytest.raises((pydantic.ValidationError, Exception)):
385 WireCommit.model_validate({"message": "incomplete"})
386
387 def test_snapshot_passes_model_validate(self) -> None:
388 from musehub.models.wire import WireSnapshot
389 sid = _sha256_oid(b"snap")
390 snap = _make_snapshot(sid, {"file.wav": sid})
391 obj = WireSnapshot.model_validate(snap)
392 assert obj.snapshot_id == sid
393
394 def test_commit_pack_limit_constant(self) -> None:
395 assert STREAM_MAX_COMMITS >= 1_000
396
397 def test_commit_pack_frame_encodes_many_commits(self) -> None:
398 commits = [_make_commit() for _ in range(10)]
399 frame = _unpack_all(_commit_pack_frame(commits))[0]
400 assert len(frame["commits"]) == 10
401
402 def test_signed_commit_has_signature_fields(self) -> None:
403 from musehub.models.wire import WireCommit
404 commit = _make_commit()
405 commit["signature"] = "sig_base64"
406 commit["signer_key_id"] = "key123"
407 commit["agent_id"] = "agent-1"
408 commit["model_id"] = "claude-opus-4-6"
409 obj = WireCommit.model_validate(commit)
410 assert obj.signature == "sig_base64"
411
412
413 # ---------------------------------------------------------------------------
414 # T5 — Service: wire_push_stream() with stubbed DB and R2 backend
415 # ---------------------------------------------------------------------------
416
417 class TestT5ServiceStream:
418 """Tier 5: wire_push_stream() async generator against minimal stubs.
419
420 We patch the storage backend and DB session so tests run without
421 infrastructure — only the frame-parsing and protocol state machine
422 are exercised here.
423 """
424
425 @pytest.fixture()
426 def stub_backend(self, monkeypatch: pytest.MonkeyPatch) -> MagicMock:
427 backend = AsyncMock()
428 backend.exists = AsyncMock(return_value=False)
429 backend.put = AsyncMock(return_value="https://r2.example.com/obj")
430 backend.get = AsyncMock(return_value=b"raw bytes")
431 monkeypatch.setattr(
432 "musehub.services.musehub_wire.get_backend",
433 lambda: backend,
434 )
435 return backend
436
437 @pytest.fixture()
438 def stub_session(self) -> AsyncMock:
439 session = AsyncMock(spec=AsyncSession)
440 session.execute = AsyncMock(return_value=MagicMock(scalar=lambda: None, fetchall=lambda: []))
441 session.commit = AsyncMock()
442 session.add = MagicMock()
443 return session
444
445 @pytest.mark.asyncio
446 async def test_missing_header_yields_error(
447 self, stub_backend: MagicMock, stub_session: AsyncMock
448 ) -> None:
449 from musehub.services.musehub_wire import wire_push_stream
450
451 async def body() -> None:
452 yield _object_frame(*_make_obj_bytes()) + _end_frame()
453
454 frames = await _collect_frames(
455 wire_push_stream(stub_session, "repo-id", body(), "gabriel")
456 )
457 error_frames = [f for f in frames if f.get("t") == SFRAME_ERROR]
458 assert error_frames, "expected an ERROR frame when OBJECT sent before HEADER"
459 assert "OBJECT frame before HEADER" in error_frames[0]["msg"]
460
461 @pytest.mark.asyncio
462 async def test_missing_end_frame_yields_error(
463 self, stub_backend: MagicMock, stub_session: AsyncMock
464 ) -> None:
465 from musehub.services.musehub_wire import wire_push_stream
466
467 commit = _make_commit()
468 snap_id = _sha256_oid(b"snap")
469 snap = _make_snapshot(snap_id)
470 commit["snapshot_id"] = snap_id
471
472 async def body() -> None:
473 yield _header_frame() + _commit_pack_frame([commit], [snap])
474 # no END frame
475
476 frames = await _collect_frames(
477 wire_push_stream(stub_session, "repo-id", body(), "gabriel")
478 )
479 error_frames = [f for f in frames if f.get("t") == SFRAME_ERROR]
480 assert error_frames
481 assert "without END frame" in error_frames[0]["msg"]
482
483 @pytest.mark.asyncio
484 async def test_missing_commit_pack_yields_error(
485 self, stub_backend: MagicMock, stub_session: AsyncMock
486 ) -> None:
487 from musehub.services.musehub_wire import wire_push_stream
488
489 async def body() -> None:
490 yield _header_frame() + _end_frame()
491
492 frames = await _collect_frames(
493 wire_push_stream(stub_session, "repo-id", body(), "gabriel")
494 )
495 error_frames = [f for f in frames if f.get("t") == SFRAME_ERROR]
496 assert error_frames
497 assert "COMMIT_PACK" in error_frames[0]["msg"]
498
499 @pytest.mark.asyncio
500 async def test_object_hash_mismatch_yields_error(
501 self, stub_backend: MagicMock, stub_session: AsyncMock
502 ) -> None:
503 from musehub.services.musehub_wire import wire_push_stream
504
505 oid = fake_id("wrong-hash-object") # wrong hash
506 content = b"this content does not match the oid"
507
508 async def body() -> None:
509 yield _header_frame(n_objects=1) + _object_frame(oid, content)
510
511 frames = await _collect_frames(
512 wire_push_stream(stub_session, "repo-id", body(), "gabriel")
513 )
514 error_frames = [f for f in frames if f.get("t") == SFRAME_ERROR]
515 assert error_frames
516
517 @pytest.mark.asyncio
518 async def test_progress_frames_emitted_on_header(
519 self, stub_backend: MagicMock, stub_session: AsyncMock
520 ) -> None:
521 from musehub.services.musehub_wire import wire_push_stream
522
523 commit = _make_commit()
524 snap_id = _sha256_oid(b"snap-data")
525 snap = _make_snapshot(snap_id)
526 commit["snapshot_id"] = snap_id
527
528 async def body() -> None:
529 yield _header_frame() + _commit_pack_frame([commit], [snap]) + _end_frame()
530
531 frames = await _collect_frames(
532 wire_push_stream(stub_session, "repo-id", body(), "gabriel")
533 )
534 progress_frames = [f for f in frames if f.get("t") == SFRAME_PROGRESS]
535 assert progress_frames, "expected at least one PROGRESS frame"
536
537
538 # ---------------------------------------------------------------------------
539 # T6 — Integration: service against real DB, stub R2
540 # ---------------------------------------------------------------------------
541
542 def _stub_r2_backend(monkeypatch: pytest.MonkeyPatch) -> None:
543 """Patch the R2 backend with an in-memory dict store."""
544 _store: dict[str, bytes] = {}
545
546 async def _exists(oid: str) -> bool:
547 return oid in _store
548
549 async def _put(oid: str, data: bytes) -> str:
550 _store[oid] = data
551 return f"https://r2.fake/{oid}"
552
553 async def _get(oid: str) -> bytes | None:
554 return _store.get(oid)
555
556 backend = AsyncMock()
557 backend.exists = _exists
558 backend.put = _put
559 backend.get = _get
560 monkeypatch.setattr(
561 "musehub.services.musehub_wire.get_backend",
562 lambda: backend,
563 )
564
565
566 async def _make_repo(db_session: AsyncSession, name: str, owner: str = "gabriel") -> MusehubRepo:
567 """Create a repo row + main branch, committed and visible to any session."""
568 from datetime import datetime, timezone
569 from musehub.db.musehub_models import MusehubRepo, MusehubBranch
570 from musehub.core.genesis import compute_identity_id, compute_repo_id, compute_branch_id
571 owner_user_id = compute_identity_id(owner.encode())
572 slug = name.lower().replace(" ", "-")
573 created_at = datetime.now(tz=timezone.utc)
574 repo_id = compute_repo_id(owner_user_id, slug, "code", created_at.isoformat())
575 repo = MusehubRepo(
576 repo_id=repo_id,
577 name=name,
578 owner=owner,
579 slug=slug,
580 visibility="public",
581 owner_user_id=owner_user_id,
582 description="",
583 tags=[],
584 created_at=created_at,
585 )
586 db_session.add(repo)
587 await db_session.commit()
588 branch = MusehubBranch(
589 branch_id=compute_branch_id(repo_id, "main"),
590 repo_id=repo_id,
591 name="main",
592 )
593 db_session.add(branch)
594 await db_session.commit()
595 await db_session.refresh(repo)
596 return repo
597
598
599 @pytest.mark.asyncio
600 async def test_t6_push_single_commit_no_objects(
601 db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
602 ) -> None:
603 """T6: push one commit with no objects against the real test DB."""
604 from musehub.services.musehub_wire import wire_push_stream
605
606 _stub_r2_backend(monkeypatch)
607 repo = await _make_repo(db_session, "T6 Single Commit")
608
609 snap_id = _sha256_oid(b"t6-snap-1")
610 commit = _make_commit(snapshot_id=snap_id)
611 snap = _make_snapshot(snap_id)
612
613 async def body() -> None:
614 yield _header_frame(n_commits=1) + _commit_pack_frame([commit], [snap]) + _end_frame(n_commits=1)
615
616 frames = await _collect_frames(
617 wire_push_stream(db_session, str(repo.repo_id), body(), "gabriel")
618 )
619 result_frames = [f for f in frames if f.get("t") == SFRAME_RESULT]
620 assert result_frames, f"no RESULT frame; frames: {[f.get('t') for f in frames]}"
621 assert result_frames[0]["ok"] is True
622
623
624 @pytest.mark.asyncio
625 async def test_t6_push_with_object(
626 db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
627 ) -> None:
628 """T6: push one object + commit; confirm RESULT.ok."""
629 from musehub.services.musehub_wire import wire_push_stream
630
631 _stub_r2_backend(monkeypatch)
632 repo = await _make_repo(db_session, "T6 With Object")
633
634 raw = b"audio data for test track"
635 oid = _sha256_oid(raw)
636 snap_id = _sha256_oid(b"t6-snap-obj")
637 commit = _make_commit(snapshot_id=snap_id)
638 snap = _make_snapshot(snap_id, {"track.wav": oid})
639
640 async def body() -> None:
641 yield (
642 _header_frame(n_objects=1, n_commits=1)
643 + _object_frame(oid, raw)
644 + _commit_pack_frame([commit], [snap])
645 + _end_frame(n_objects=1, n_commits=1)
646 )
647
648 frames = await _collect_frames(
649 wire_push_stream(db_session, str(repo.repo_id), body(), "gabriel")
650 )
651 result_frames = [f for f in frames if f.get("t") == SFRAME_RESULT]
652 assert result_frames and result_frames[0]["ok"] is True
653
654
655 @pytest.mark.asyncio
656 async def test_t6_push_zlib_compressed_object(
657 db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
658 ) -> None:
659 """T6: push a zlib-compressed object; server decompresses and confirms."""
660 from musehub.services.musehub_wire import wire_push_stream
661
662 _stub_r2_backend(monkeypatch)
663 repo = await _make_repo(db_session, "T6 Zlib")
664
665 raw = b"raw MIDI data " * 100
666 compressed = zlib.compress(raw)
667 oid = _sha256_oid(raw)
668 snap_id = _sha256_oid(b"t6-snap-zlib")
669 commit = _make_commit(snapshot_id=snap_id)
670 snap = _make_snapshot(snap_id, {"beat.mid": oid})
671
672 async def body() -> None:
673 yield (
674 _header_frame(n_objects=1, n_commits=1)
675 + _object_frame(oid, compressed, enc="zlib")
676 + _commit_pack_frame([commit], [snap])
677 + _end_frame(n_objects=1, n_commits=1)
678 )
679
680 frames = await _collect_frames(
681 wire_push_stream(db_session, str(repo.repo_id), body(), "gabriel")
682 )
683 result = [f for f in frames if f.get("t") == SFRAME_RESULT]
684 assert result and result[0]["ok"] is True
685
686
687 @pytest.mark.asyncio
688 async def test_t6_force_push_advances_branch(
689 db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
690 ) -> None:
691 """T6: two sequential pushes — second uses force=True to advance divergent branch."""
692 from musehub.services.musehub_wire import wire_push_stream
693
694 _stub_r2_backend(monkeypatch)
695 repo = await _make_repo(db_session, "T6 Force Push")
696
697 snap_id = _sha256_oid(b"t6-snap-force-1")
698 commit1 = _make_commit(snapshot_id=snap_id)
699 snap1 = _make_snapshot(snap_id)
700
701 async def body1() -> None:
702 yield _header_frame() + _commit_pack_frame([commit1], [snap1]) + _end_frame(n_commits=1)
703
704 frames1 = await _collect_frames(
705 wire_push_stream(db_session, str(repo.repo_id), body1(), "gabriel")
706 )
707 assert any(f.get("t") == SFRAME_RESULT and f["ok"] for f in frames1)
708
709 snap_id2 = _sha256_oid(b"t6-snap-force-2")
710 commit2 = _make_commit(snapshot_id=snap_id2)
711 snap2 = _make_snapshot(snap_id2)
712
713 async def body2() -> None:
714 yield _header_frame(force=True) + _commit_pack_frame([commit2], [snap2]) + _end_frame(n_commits=1)
715
716 frames2 = await _collect_frames(
717 wire_push_stream(db_session, str(repo.repo_id), body2(), "gabriel")
718 )
719 result2 = [f for f in frames2 if f.get("t") == SFRAME_RESULT]
720 assert result2 and result2[0]["ok"] is True
721
722
723 # ---------------------------------------------------------------------------
724 # T7 — Route: ASGI test client hitting POST /{owner}/{slug}/push/stream
725 # ---------------------------------------------------------------------------
726
727 @pytest.mark.asyncio
728 async def test_t7_push_stream_returns_200_with_packstream_content_type(
729 client: AsyncClient, db_session: AsyncSession, auth_headers: StrDict,
730 monkeypatch: pytest.MonkeyPatch,
731 ) -> None:
732 """T7: route returns 200 with application/x-muse-packstream content-type."""
733 _stub_r2_backend(monkeypatch)
734 repo = await _make_repo(db_session, "T7 Route Test 1", owner="testuser")
735
736 snap_id = _sha256_oid(b"t7-snap-ct")
737 commit = _make_commit(snapshot_id=snap_id)
738 snap = _make_snapshot(snap_id)
739 body = _header_frame() + _commit_pack_frame([commit], [snap]) + _end_frame(n_commits=1)
740
741 resp = await client.post(
742 f"/{repo.owner}/{repo.slug}/push/stream",
743 content=body,
744 headers={**auth_headers, "Content-Type": WIRE_CONTENT_TYPE},
745 )
746 assert resp.status_code == 200
747 assert resp.headers.get("content-type", "").startswith(WIRE_CONTENT_TYPE)
748
749
750 @pytest.mark.asyncio
751 async def test_t7_push_stream_response_contains_result_frame(
752 client: AsyncClient, db_session: AsyncSession, auth_headers: StrDict,
753 monkeypatch: pytest.MonkeyPatch,
754 ) -> None:
755 """T7: response body is a plain msgpack dict with t=RESULT."""
756 _stub_r2_backend(monkeypatch)
757 repo = await _make_repo(db_session, "T7 Route Test 2", owner="testuser")
758
759 snap_id = _sha256_oid(b"t7-snap-result")
760 commit = _make_commit(snapshot_id=snap_id)
761 snap = _make_snapshot(snap_id)
762 body = _header_frame() + _commit_pack_frame([commit], [snap]) + _end_frame(n_commits=1)
763
764 resp = await client.post(
765 f"/{repo.owner}/{repo.slug}/push/stream",
766 content=body,
767 headers={**auth_headers, "Content-Type": WIRE_CONTENT_TYPE},
768 )
769 result = _last_frame(resp.content)
770 assert result.get("ok") is True, f"expected ok=True result, got: {result}"
771
772
773 @pytest.mark.asyncio
774 async def test_t7_push_stream_requires_auth(
775 client: AsyncClient, db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch,
776 ) -> None:
777 """T7: unauthenticated push yields error."""
778 _stub_r2_backend(monkeypatch)
779 repo = await _make_repo(db_session, "T7 Route Auth", owner="testuser")
780
781 body = _header_frame() + _end_frame()
782 resp = await client.post(
783 f"/{repo.owner}/{repo.slug}/push/stream",
784 content=body,
785 headers={"Content-Type": WIRE_CONTENT_TYPE},
786 )
787 assert resp.status_code in (200, 401, 403)
788 if resp.status_code == 200:
789 result = _last_frame(resp.content)
790 assert result.get("t") == SFRAME_ERROR, "unauthenticated push should yield error frame"
791
792
793 @pytest.mark.asyncio
794 async def test_t7_push_stream_404_for_missing_repo(
795 client: AsyncClient, auth_headers: StrDict,
796 ) -> None:
797 """T7: push to a repo that doesn't exist yields 404 or error frame."""
798 body = _header_frame() + _end_frame()
799 resp = await client.post(
800 "/gabriel/nonexistent-repo-xyz-t7/push/stream",
801 content=body,
802 headers={**auth_headers, "Content-Type": WIRE_CONTENT_TYPE},
803 )
804 if resp.status_code == 200:
805 result = _last_frame(resp.content)
806 assert result.get("t") == SFRAME_ERROR and result.get("code") == 404
807 else:
808 assert resp.status_code == 404
809
810
811 @pytest.mark.asyncio
812 async def test_t7_old_push_endpoints_deleted(
813 client: AsyncClient, db_session: AsyncSession, auth_headers: StrDict,
814 ) -> None:
815 """T7: all MWP v1 push endpoints return 404 — they were deleted in MWP v2."""
816 repo = await _make_repo(db_session, "T7 Route v1 Deleted", owner="testuser")
817
818 deleted_paths = [
819 f"/{repo.owner}/{repo.slug}/filter-objects",
820 f"/{repo.owner}/{repo.slug}/presign-objects",
821 f"/{repo.owner}/{repo.slug}/presign",
822 f"/{repo.owner}/{repo.slug}/push/objects",
823 f"/{repo.owner}/{repo.slug}/push/objects/confirm",
824 f"/{repo.owner}/{repo.slug}/push",
825 ]
826 for path in deleted_paths:
827 resp = await client.post(path, headers=auth_headers, content=b"{}")
828 assert resp.status_code == 404, (
829 f"Expected 404 for deleted endpoint {path}, got {resp.status_code}"
830 )
831
832
833 # ---------------------------------------------------------------------------
834 # T8 — E2E: full push → GET /refs confirms branch head updated
835 # ---------------------------------------------------------------------------
836
837 @pytest.mark.asyncio
838 async def test_t8_push_then_refs_show_commit(
839 client: AsyncClient, db_session: AsyncSession, auth_headers: StrDict,
840 ) -> None:
841 """T8: push one commit; GET /refs confirms branch head updated."""
842 repo = await _make_repo(db_session, "T8 E2E Refs", owner="testuser")
843
844 snap_id = _sha256_oid(b"t8-e2e-snap-1")
845 commit = _make_commit(snapshot_id=snap_id)
846 snap = _make_snapshot(snap_id)
847 body = (
848 _header_frame(head=commit["commit_id"])
849 + _commit_pack_frame([commit], [snap])
850 + _end_frame(n_commits=1)
851 )
852
853 push_resp = await client.post(
854 f"/{repo.owner}/{repo.slug}/push/stream",
855 content=body,
856 headers={**auth_headers, "Content-Type": WIRE_CONTENT_TYPE},
857 )
858 assert push_resp.status_code == 200
859 result = _last_frame(push_resp.content)
860 assert result.get("ok") is True, f"push failed: {result}"
861
862 refs_resp = await client.get(
863 f"/{repo.owner}/{repo.slug}/refs",
864 headers=auth_headers,
865 )
866 assert refs_resp.status_code == 200
867 branch_heads = refs_resp.json().get("branch_heads", {})
868 assert branch_heads.get("main") == commit["commit_id"], (
869 f"Expected main head={commit['commit_id']!r}, got {branch_heads}"
870 )
871
872
873 @pytest.mark.asyncio
874 async def test_t8_push_with_objects_then_fetch_objects(
875 client: AsyncClient, db_session: AsyncSession, auth_headers: StrDict,
876 ) -> None:
877 """T8: push an object; fetch it back and verify content integrity."""
878 repo = await _make_repo(db_session, "T8 E2E Fetch", owner="testuser")
879
880 raw = b"audio track bytes for e2e test"
881 oid = _sha256_oid(raw)
882 snap_id = _sha256_oid(b"t8-e2e-obj-snap")
883 commit = _make_commit(snapshot_id=snap_id)
884 snap = _make_snapshot(snap_id, {"track.wav": oid})
885 body = (
886 _header_frame(n_objects=1)
887 + _object_frame(oid, raw)
888 + _commit_pack_frame([commit], [snap])
889 + _end_frame(n_objects=1, n_commits=1)
890 )
891
892 push_resp = await client.post(
893 f"/{repo.owner}/{repo.slug}/push/stream",
894 content=body,
895 headers={**auth_headers, "Content-Type": WIRE_CONTENT_TYPE},
896 )
897 assert push_resp.status_code == 200
898 result = _last_frame(push_resp.content)
899 assert result.get("ok") is True, f"push failed: {result}"
900
901 fetch_resp = await client.get(
902 f"/o/{oid}",
903 headers=auth_headers,
904 )
905 assert fetch_resp.status_code == 200
906 assert fetch_resp.content == raw
907
908
909 @pytest.mark.asyncio
910 async def test_t8_push_chain_of_commits_parent_linking(
911 client: AsyncClient, db_session: AsyncSession, auth_headers: StrDict,
912 ) -> None:
913 """T8: push two chained commits; branch head advances to the child."""
914 repo = await _make_repo(db_session, "T8 E2E Chain", owner="testuser")
915
916 snap1_id = _sha256_oid(b"t8-e2e-snap-chain-1")
917 commit1 = _make_commit(snapshot_id=snap1_id)
918 snap1 = _make_snapshot(snap1_id)
919
920 snap2_id = _sha256_oid(b"t8-e2e-snap-chain-2")
921 commit2 = _make_commit(snapshot_id=snap2_id, parent_ids=[commit1["commit_id"]])
922 snap2 = _make_snapshot(snap2_id)
923
924 body = (
925 _header_frame(n_commits=2, head=commit2["commit_id"])
926 + _commit_pack_frame([commit1, commit2], [snap1, snap2])
927 + _end_frame(n_commits=2)
928 )
929
930 push_resp = await client.post(
931 f"/{repo.owner}/{repo.slug}/push/stream",
932 content=body,
933 headers={**auth_headers, "Content-Type": WIRE_CONTENT_TYPE},
934 )
935 assert push_resp.status_code == 200
936 result = _last_frame(push_resp.content)
937 assert result.get("ok") is True, f"push failed: {result}"
938
939 refs_resp = await client.get(
940 f"/{repo.owner}/{repo.slug}/refs",
941 headers=auth_headers,
942 )
943 refs = refs_resp.json().get("branch_heads", {})
944 assert refs.get("main") == commit2["commit_id"]
945
946
947 # ---------------------------------------------------------------------------
948 # T9 — Regression: MPackStreamWriter frames with compressed binary content
949 # must not raise UnicodeDecodeError on the server.
950 #
951 # Root cause of staging bug:
952 # 'utf-8' codec can't decode byte 0xad in position 2: invalid start byte
953 #
954 # The client sends O frames where the "content" field is zlib-compressed
955 # binary bytes packed with use_bin_type=True (msgpack bin type 0xc4/c5/c6).
956 # If the server Unpacker is misconfigured (raw=True, or content field encoded
957 # as str/fixstr), raw=False raises UnicodeDecodeError on non-UTF-8 bytes.
958 #
959 # These tests use MPackStreamWriter (the actual client encoder) to build the
960 # exact bytes the client sends, then POST them through the ASGI app.
961 # ---------------------------------------------------------------------------
962
963 @pytest.mark.asyncio
964 async def test_t9_mpackstreamwriter_compressed_frames_decode_clean(
965 client: AsyncClient, db_session: AsyncSession, auth_headers: StrDict,
966 ) -> None:
967 """T9: server must decode MPackStreamWriter O frames with zlib-compressed binary content.
968
969 Regression for: 'utf-8' codec can't decode byte 0xad in position 2.
970 Client uses MPackStreamWriter (use_bin_type=True); server Unpacker uses raw=False.
971 Binary content including byte 0xad must arrive as bytes, not trigger UnicodeDecodeError.
972 """
973 from muse.core.mpack import MPackStreamWriter
974 from muse.core.types import blob_id
975
976 repo = await _make_repo(db_session, "T9 Compressed Binary Frames", owner="testuser")
977 w = MPackStreamWriter()
978
979 # Content that includes 0xad and other non-UTF-8 bytes — the exact failing case
980 content = bytes(range(256)) * 10
981 oid = blob_id(content)
982
983 snap_id = _sha256_oid(b"t9-snap-compressed")
984 commit = _make_commit(snapshot_id=snap_id)
985 snap = _make_snapshot(snap_id, {"file.bin": oid})
986
987 body = (
988 _fw.wrap(frame_type="H", payload=w.write_header(op="push", branch="main", n_objects=1, n_commits=1))
989 + _fw.wrap(frame_type="O", payload=w.write_object_raw(object_id=oid, raw_bytes=content, compress="zlib"))
990 + _fw.wrap(frame_type="C", payload=w.write_commit_pack(commits=[commit], snapshots=[snap]))
991 + _fw.wrap(frame_type="E", payload=w.write_end(n_objects=1, n_commits=1))
992 )
993
994 resp = await client.post(
995 f"/{repo.owner}/{repo.slug}/push/stream",
996 content=body,
997 headers={**auth_headers, "Content-Type": WIRE_CONTENT_TYPE},
998 )
999 assert resp.status_code == 200
1000 result = _last_frame(resp.content)
1001 assert result.get("t") != SFRAME_ERROR, f"Server returned error: {result.get('msg')}"
1002 assert result.get("ok") is True, f"No ok=True in result: {result}"
1003
1004
1005 @pytest.mark.asyncio
1006 async def test_t9_920_objects_with_binary_content_no_unicode_error(
1007 client: AsyncClient, db_session: AsyncSession, auth_headers: StrDict,
1008 ) -> None:
1009 """T9: 920 objects with full byte range (0x00-0xff) — none may produce UnicodeDecodeError.
1010
1011 Reproduces the staging scenario: ~900 small objects each containing binary
1012 data. The server must process all O frames without any 'utf-8 codec' error.
1013 """
1014 from muse.core.mpack import MPackStreamWriter
1015 from muse.core.types import blob_id
1016
1017 n = 920
1018 repo = await _make_repo(db_session, "T9 920 Binary Objects", owner="testuser")
1019 w = MPackStreamWriter()
1020
1021 snap_id = _sha256_oid(b"t9-snap-920")
1022 commit = _make_commit(snapshot_id=snap_id)
1023 manifest = {}
1024 parts = [_fw.wrap(frame_type="H", payload=w.write_header(op="push", branch="main", n_objects=n, n_commits=1))]
1025
1026 for i in range(n):
1027 raw_content = (bytes(range(256)) * 2)[i % 256: i % 256 + 256] + i.to_bytes(4, "big")
1028 oid = blob_id(raw_content)
1029 manifest[f"file_{i}.bin"] = oid
1030 parts.append(_fw.wrap(frame_type="O", payload=w.write_object_raw(object_id=oid, raw_bytes=raw_content, compress="zlib")))
1031
1032 snap = _make_snapshot(snap_id, manifest)
1033 parts.append(_fw.wrap(frame_type="C", payload=w.write_commit_pack(commits=[commit], snapshots=[snap])))
1034 parts.append(_fw.wrap(frame_type="E", payload=w.write_end(n_objects=n, n_commits=1)))
1035
1036 body = b"".join(parts)
1037 resp = await client.post(
1038 f"/{repo.owner}/{repo.slug}/push/stream",
1039 content=body,
1040 headers={**auth_headers, "Content-Type": WIRE_CONTENT_TYPE},
1041 )
1042 assert resp.status_code == 200
1043 result = _last_frame(resp.content)
1044 assert result.get("t") != SFRAME_ERROR, f"Server returned error: {result.get('msg')}"
1045 assert result.get("ok") is True, f"No ok=True in result: {result}"
1046
1047
1048 # ---------------------------------------------------------------------------
1049 # T10 — Regression: server response frames must use only string map keys.
1050 #
1051 # Root cause of staging bug:
1052 # stream read error: int is not allowed for map key when strict_map_key=True
1053 #
1054 # MPackStreamReader (muse/core/mpack.py) uses msgpack.Unpacker with the
1055 # default strict_map_key=True. If the server sends any frame where a map
1056 # key is an integer, the client raises and the push fails.
1057 #
1058 # This test decodes the server response with strict_map_key=True — the exact
1059 # setting the client uses — and asserts every key in every frame is a str.
1060 # ---------------------------------------------------------------------------
1061
1062 def _unpack_all_strict(raw: bytes) -> list[JSONValue]:
1063 """Decode all msgpack frames with strict_map_key=True (same as MPackStreamReader)."""
1064 unpacker = msgpack.Unpacker(raw=False, strict_map_key=True)
1065 unpacker.feed(raw)
1066 return list(unpacker)
1067
1068
1069 @pytest.mark.asyncio
1070 async def test_t10_response_frames_use_only_string_map_keys(
1071 client: AsyncClient, db_session: AsyncSession, auth_headers: StrDict,
1072 ) -> None:
1073 """T10: all server response frame map keys must be strings.
1074
1075 MPackStreamReader uses strict_map_key=True (msgpack default). Any integer
1076 key in a server frame raises 'int is not allowed for map key' on the client
1077 and aborts the push. This test uses the same strict decoder to catch the
1078 mismatch at the server level before it reaches production.
1079 """
1080 repo = await _make_repo(db_session, "T10 String Map Keys", owner="testuser")
1081
1082 snap_id = _sha256_oid(b"t10-snap-1")
1083 commit = _make_commit(snapshot_id=snap_id)
1084 snap = _make_snapshot(snap_id)
1085 body = _header_frame() + _commit_pack_frame([commit], [snap]) + _end_frame(n_commits=1)
1086
1087 resp = await client.post(
1088 f"/{repo.owner}/{repo.slug}/push/stream",
1089 content=body,
1090 headers={**auth_headers, "Content-Type": WIRE_CONTENT_TYPE},
1091 )
1092 assert resp.status_code == 200
1093
1094 # Use the same strict decoder the client uses — must not raise on any frame.
1095 try:
1096 frames = _unpack_all_strict(resp.content)
1097 except Exception as exc:
1098 raise AssertionError(
1099 f"Server response failed strict msgpack decode: {exc}\n"
1100 f"Raw response (first 512 bytes): {resp.content[:512]!r}"
1101 ) from exc
1102
1103 assert frames, "Expected at least one frame in response"
1104 for result in frames:
1105 assert isinstance(result, dict), f"Expected dict frame, got {type(result)}"
1106 int_keys = [k for k in result if not isinstance(k, str)]
1107 assert not int_keys, (
1108 f"Response frame has integer map keys: {int_keys!r}\nFull frame: {result!r}"
1109 )
1110
1111
1112 # ---------------------------------------------------------------------------
1113 # T11 — Phase 5: E frame count verification
1114 #
1115 # Server must reject a push where the E frame's n_objects or n_commits
1116 # field does not match the actual number of frames received.
1117 # ---------------------------------------------------------------------------
1118
1119 @pytest.mark.asyncio
1120 async def test_t11_e_frame_wrong_n_objects_rejected(
1121 client: AsyncClient, db_session: AsyncSession, auth_headers: StrDict,
1122 monkeypatch: pytest.MonkeyPatch,
1123 ) -> None:
1124 """T11: E frame claiming wrong n_objects is rejected with an error."""
1125 _stub_r2_backend(monkeypatch)
1126 repo = await _make_repo(db_session, "T11 E Objects Mismatch", owner="testuser")
1127
1128 raw = b"audio bytes t11"
1129 oid = _sha256_oid(raw)
1130 snap_id = _sha256_oid(b"t11-snap-objs")
1131 commit = _make_commit(snapshot_id=snap_id)
1132 snap = _make_snapshot(snap_id, {"a.wav": oid})
1133
1134 # Send 1 object frame but E frame claims 0
1135 body = (
1136 _header_frame(n_objects=1, n_commits=1)
1137 + _object_frame(oid, raw)
1138 + _commit_pack_frame([commit], [snap])
1139 + _end_frame(n_objects=0, n_commits=1)
1140 )
1141 resp = await client.post(
1142 f"/{repo.owner}/{repo.slug}/push/stream",
1143 content=body,
1144 headers={**auth_headers, "Content-Type": WIRE_CONTENT_TYPE},
1145 )
1146 result = _last_frame(resp.content)
1147 assert resp.status_code == 400 or result.get("t") == SFRAME_ERROR, (
1148 f"Expected 400 or error frame, got status={resp.status_code} result={result}"
1149 )
1150 assert "count mismatch" in result.get("msg", "").lower() or result.get("code") == 400
1151
1152
1153 @pytest.mark.asyncio
1154 async def test_t11_e_frame_wrong_n_commits_rejected(
1155 client: AsyncClient, db_session: AsyncSession, auth_headers: StrDict,
1156 monkeypatch: pytest.MonkeyPatch,
1157 ) -> None:
1158 """T11: E frame claiming wrong n_commits is rejected with an error."""
1159 _stub_r2_backend(monkeypatch)
1160 repo = await _make_repo(db_session, "T11 E Commits Mismatch", owner="testuser")
1161
1162 snap_id = _sha256_oid(b"t11-snap-commits")
1163 commit = _make_commit(snapshot_id=snap_id)
1164 snap = _make_snapshot(snap_id)
1165
1166 # Send 1 commit but E frame claims 2
1167 body = (
1168 _header_frame(n_commits=1)
1169 + _commit_pack_frame([commit], [snap])
1170 + _end_frame(n_objects=0, n_commits=2)
1171 )
1172 resp = await client.post(
1173 f"/{repo.owner}/{repo.slug}/push/stream",
1174 content=body,
1175 headers={**auth_headers, "Content-Type": WIRE_CONTENT_TYPE},
1176 )
1177 result = _last_frame(resp.content)
1178 assert resp.status_code == 400 or result.get("t") == SFRAME_ERROR, (
1179 f"Expected 400 or error frame, got status={resp.status_code} result={result}"
1180 )
1181 assert "count mismatch" in result.get("msg", "").lower() or result.get("code") == 400
1182
1183
1184 @pytest.mark.asyncio
1185 async def test_t11_e_frame_overstated_objects_rejected(
1186 client: AsyncClient, db_session: AsyncSession, auth_headers: StrDict,
1187 monkeypatch: pytest.MonkeyPatch,
1188 ) -> None:
1189 """T11: E frame claiming more objects than received is rejected."""
1190 _stub_r2_backend(monkeypatch)
1191 repo = await _make_repo(db_session, "T11 E Overstated Objects", owner="testuser")
1192
1193 snap_id = _sha256_oid(b"t11-snap-over")
1194 commit = _make_commit(snapshot_id=snap_id)
1195 snap = _make_snapshot(snap_id)
1196
1197 # Send 0 objects but E frame claims 5
1198 body = (
1199 _header_frame(n_commits=1)
1200 + _commit_pack_frame([commit], [snap])
1201 + _end_frame(n_objects=5, n_commits=1)
1202 )
1203 resp = await client.post(
1204 f"/{repo.owner}/{repo.slug}/push/stream",
1205 content=body,
1206 headers={**auth_headers, "Content-Type": WIRE_CONTENT_TYPE},
1207 )
1208 result = _last_frame(resp.content)
1209 assert resp.status_code == 400 or result.get("t") == SFRAME_ERROR, (
1210 f"Expected 400 or error frame, got status={resp.status_code} result={result}"
1211 )
1212 assert "count mismatch" in result.get("msg", "").lower() or result.get("code") == 400
1213
1214
1215 # ---------------------------------------------------------------------------
1216 # T12 — Phase 6: Ingest transaction model
1217 #
1218 # P6A: snapshot referential integrity — reject if snapshot references an
1219 # object not in the push bundle and not in storage.
1220 # P6B: snapshot references an object from a PRIOR push — accepted (already
1221 # in storage path).
1222 # P6C: atomicity — branch ref remains unchanged after a rejected push.
1223 # ---------------------------------------------------------------------------
1224
1225 @pytest.mark.asyncio
1226 async def test_t12_p6a_snapshot_missing_object_rejected(
1227 client: AsyncClient, db_session: AsyncSession, auth_headers: StrDict,
1228 monkeypatch: pytest.MonkeyPatch,
1229 ) -> None:
1230 """T12/P6A: snapshot manifest references an object not in bundle or storage → 422."""
1231 _stub_r2_backend(monkeypatch)
1232 repo = await _make_repo(db_session, "T12 P6A Missing Object", owner="testuser")
1233
1234 ghost_oid = _sha256_oid(b"ghost-object-never-pushed")
1235 snap_id = _sha256_oid(b"t12-p6a-snap")
1236 commit = _make_commit(snapshot_id=snap_id)
1237 snap = _make_snapshot(snap_id, {"missing.wav": ghost_oid})
1238
1239 body = (
1240 _header_frame(n_objects=0, n_commits=1)
1241 + _commit_pack_frame([commit], [snap])
1242 + _end_frame(n_objects=0, n_commits=1)
1243 )
1244 resp = await client.post(
1245 f"/{repo.owner}/{repo.slug}/push/stream",
1246 content=body,
1247 headers={**auth_headers, "Content-Type": WIRE_CONTENT_TYPE},
1248 )
1249 result = _last_frame(resp.content)
1250 assert resp.status_code in (200, 422), f"Unexpected status: {resp.status_code}"
1251 assert resp.status_code == 422 or result.get("t") == SFRAME_ERROR, (
1252 f"Expected 422 or error frame for missing snapshot object, got: {result}"
1253 )
1254 msg = result.get("msg", "")
1255 assert "missing" in msg.lower() or result.get("code") in (422, 400), (
1256 f"Error message should mention missing object: {msg!r}"
1257 )
1258
1259
1260 @pytest.mark.asyncio
1261 async def test_t12_p6b_snapshot_object_from_prior_push_accepted(
1262 client: AsyncClient, db_session: AsyncSession, auth_headers: StrDict,
1263 monkeypatch: pytest.MonkeyPatch,
1264 ) -> None:
1265 """T12/P6B: snapshot references an object stored in a previous push → accepted."""
1266 _stub_r2_backend(monkeypatch)
1267 repo = await _make_repo(db_session, "T12 P6B Prior Object", owner="testuser")
1268
1269 raw = b"audio content from push 1"
1270 oid = _sha256_oid(raw)
1271 snap1_id = _sha256_oid(b"t12-p6b-snap1")
1272 commit1 = _make_commit(snapshot_id=snap1_id)
1273 snap1 = _make_snapshot(snap1_id, {"track.wav": oid})
1274
1275 body1 = (
1276 _header_frame(n_objects=1, n_commits=1)
1277 + _object_frame(oid, raw)
1278 + _commit_pack_frame([commit1], [snap1])
1279 + _end_frame(n_objects=1, n_commits=1)
1280 )
1281 resp1 = await client.post(
1282 f"/{repo.owner}/{repo.slug}/push/stream",
1283 content=body1,
1284 headers={**auth_headers, "Content-Type": WIRE_CONTENT_TYPE},
1285 )
1286 assert resp1.status_code == 200
1287 result1 = _last_frame(resp1.content)
1288 assert result1.get("ok") is True, f"Push 1 failed: {result1}"
1289
1290 # Push 2: new commit referencing the SAME object — not re-sent in bundle
1291 snap2_id = _sha256_oid(b"t12-p6b-snap2")
1292 commit2 = _make_commit(snapshot_id=snap2_id, parent_ids=[commit1["commit_id"]])
1293 snap2 = _make_snapshot(snap2_id, {"track.wav": oid})
1294
1295 body2 = (
1296 _header_frame(n_objects=0, n_commits=1)
1297 + _commit_pack_frame([commit2], [snap2])
1298 + _end_frame(n_objects=0, n_commits=1)
1299 )
1300 resp2 = await client.post(
1301 f"/{repo.owner}/{repo.slug}/push/stream",
1302 content=body2,
1303 headers={**auth_headers, "Content-Type": WIRE_CONTENT_TYPE},
1304 )
1305 assert resp2.status_code == 200
1306 result2 = _last_frame(resp2.content)
1307 assert result2.get("ok") is True, (
1308 f"Push 2 should succeed — object already in storage. Got: {result2}"
1309 )
1310
1311
1312 @pytest.mark.asyncio
1313 async def test_t12_p6c_failed_push_leaves_branch_unchanged(
1314 client: AsyncClient, db_session: AsyncSession, auth_headers: StrDict,
1315 monkeypatch: pytest.MonkeyPatch,
1316 ) -> None:
1317 """T12/P6C: a rejected push must not advance the branch head (atomicity)."""
1318 _stub_r2_backend(monkeypatch)
1319 repo = await _make_repo(db_session, "T12 P6C Atomicity", owner="testuser")
1320
1321 snap1_id = _sha256_oid(b"t12-p6c-snap1")
1322 commit1 = _make_commit(snapshot_id=snap1_id)
1323 snap1 = _make_snapshot(snap1_id)
1324
1325 body1 = (
1326 _header_frame(n_commits=1, head=commit1["commit_id"])
1327 + _commit_pack_frame([commit1], [snap1])
1328 + _end_frame(n_commits=1)
1329 )
1330 resp1 = await client.post(
1331 f"/{repo.owner}/{repo.slug}/push/stream",
1332 content=body1,
1333 headers={**auth_headers, "Content-Type": WIRE_CONTENT_TYPE},
1334 )
1335 assert resp1.status_code == 200
1336 result1 = _last_frame(resp1.content)
1337 assert result1.get("ok") is True, f"Push 1 failed: {result1}"
1338
1339 refs1 = (await client.get(f"/{repo.owner}/{repo.slug}/refs", headers=auth_headers)).json()
1340 head_after_push1 = refs1.get("branch_heads", {}).get("main")
1341 assert head_after_push1 == commit1["commit_id"]
1342
1343 # Push 2: snapshot references a ghost object — must be rejected
1344 ghost_oid = _sha256_oid(b"t12-p6c-ghost-never-exists")
1345 snap2_id = _sha256_oid(b"t12-p6c-snap2")
1346 commit2 = _make_commit(snapshot_id=snap2_id, parent_ids=[commit1["commit_id"]])
1347 snap2 = _make_snapshot(snap2_id, {"ghost.wav": ghost_oid})
1348
1349 body2 = (
1350 _header_frame(n_objects=0, n_commits=1)
1351 + _commit_pack_frame([commit2], [snap2])
1352 + _end_frame(n_objects=0, n_commits=1)
1353 )
1354 resp2 = await client.post(
1355 f"/{repo.owner}/{repo.slug}/push/stream",
1356 content=body2,
1357 headers={**auth_headers, "Content-Type": WIRE_CONTENT_TYPE},
1358 )
1359 result2 = _last_frame(resp2.content)
1360 assert resp2.status_code in (200, 422) and (
1361 resp2.status_code == 422 or result2.get("t") == SFRAME_ERROR
1362 ), f"Push 2 should be rejected, got: status={resp2.status_code} result={result2}"
1363
1364 # Branch head must still be commit1
1365 refs2 = (await client.get(f"/{repo.owner}/{repo.slug}/refs", headers=auth_headers)).json()
1366 head_after_push2 = refs2.get("branch_heads", {}).get("main")
1367 assert head_after_push2 == commit1["commit_id"], (
1368 f"Branch head must remain commit1 after rejected push. "
1369 f"Got: {head_after_push2!r}"
1370 )
1371
1372
1373 # ---------------------------------------------------------------------------
1374 # Phase 4 — Server-side branch ref CAS (SELECT FOR UPDATE)
1375 # ---------------------------------------------------------------------------
1376
1377 @pytest.mark.asyncio
1378 async def test_phase4_sequential_pushes_both_advance_branch(
1379 db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
1380 ) -> None:
1381 """Phase 4 / Data: two sequential pushes must each advance the branch.
1382
1383 The SELECT FOR UPDATE on the branch row ensures that concurrent pushes
1384 are serialized at the DB level. This test uses sequential pushes with
1385 separate sessions to verify the core invariant: the second push sees the
1386 branch head left by the first push and advances it correctly.
1387 """
1388 from musehub.services.musehub_wire import wire_push_stream
1389
1390 _stub_r2_backend(monkeypatch)
1391 repo = await _make_repo(db_session, "P4 Sequential CAS")
1392
1393 snap1_id = _sha256_oid(b"p4-snap-1")
1394 commit1 = _make_commit(snapshot_id=snap1_id)
1395 snap1 = _make_snapshot(snap1_id)
1396
1397 # Push 1
1398 async def body1() -> None:
1399 yield (
1400 _header_frame(n_commits=1, head=commit1["commit_id"])
1401 + _commit_pack_frame([commit1], [snap1])
1402 + _end_frame(n_commits=1)
1403 )
1404
1405 frames1 = await _collect_frames(
1406 wire_push_stream(db_session, str(repo.repo_id), body1(), "gabriel")
1407 )
1408 result1 = [f for f in frames1 if f.get("t") == SFRAME_RESULT]
1409 assert result1 and result1[0]["ok"] is True, f"Push 1 failed: {frames1}"
1410
1411 # Push 2 — parent is commit1
1412 snap2_id = _sha256_oid(b"p4-snap-2")
1413 commit2 = _make_commit(
1414 snapshot_id=snap2_id,
1415 parent_ids=[commit1["commit_id"]],
1416 )
1417 snap2 = _make_snapshot(snap2_id)
1418
1419 async def body2() -> None:
1420 yield (
1421 _header_frame(n_commits=1, head=commit2["commit_id"])
1422 + _commit_pack_frame([commit2], [snap2])
1423 + _end_frame(n_commits=1)
1424 )
1425
1426 frames2 = await _collect_frames(
1427 wire_push_stream(db_session, str(repo.repo_id), body2(), "gabriel")
1428 )
1429 result2 = [f for f in frames2 if f.get("t") == SFRAME_RESULT]
1430 assert result2 and result2[0]["ok"] is True, f"Push 2 failed: {frames2}"
1431
1432 # Branch must point to commit2 (the latest)
1433 from musehub.db.musehub_models import MusehubBranch
1434 from sqlalchemy import select as _select
1435 branch = (await db_session.execute(
1436 _select(MusehubBranch).where(
1437 MusehubBranch.repo_id == repo.repo_id,
1438 MusehubBranch.name == "main",
1439 )
1440 )).scalar_one()
1441 assert branch.head_commit_id == commit2["commit_id"], (
1442 f"Branch should point to commit2 after second push. "
1443 f"Got: {branch.head_commit_id!r}"
1444 )
1445
1446
1447 @pytest.mark.asyncio
1448 async def test_phase4_non_ff_push_rejected_after_concurrent_advance(
1449 db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
1450 ) -> None:
1451 """Phase 4 / Unit: a push that tries to set a non-FF head is rejected.
1452
1453 This is the key invariant the SELECT FOR UPDATE protects: if another push
1454 advances the branch between our read and our write, our push fails the
1455 fast-forward check on the now-current head rather than silently overwriting.
1456 """
1457 from musehub.services.musehub_wire import wire_push_stream
1458 from musehub.db.musehub_models import MusehubBranch
1459 from sqlalchemy import select as _select
1460
1461 _stub_r2_backend(monkeypatch)
1462 repo = await _make_repo(db_session, "P4 Non-FF After Advance")
1463
1464 # Establish an initial commit on the branch
1465 snap0_id = _sha256_oid(b"p4-nff-snap-0")
1466 commit0 = _make_commit(snapshot_id=snap0_id)
1467 snap0 = _make_snapshot(snap0_id)
1468
1469 async def body0() -> None:
1470 yield (
1471 _header_frame(n_commits=1)
1472 + _commit_pack_frame([commit0], [snap0])
1473 + _end_frame(n_commits=1)
1474 )
1475
1476 frames0 = await _collect_frames(
1477 wire_push_stream(db_session, str(repo.repo_id), body0(), "gabriel")
1478 )
1479 assert [f for f in frames0 if f.get("t") == SFRAME_RESULT and f["ok"]], (
1480 f"Initial push failed: {frames0}"
1481 )
1482
1483 # Manually advance the branch to simulate a concurrent push winning
1484 branch_row = (await db_session.execute(
1485 _select(MusehubBranch).where(
1486 MusehubBranch.repo_id == repo.repo_id,
1487 MusehubBranch.name == "main",
1488 )
1489 )).scalar_one()
1490 concurrent_commit_id = _sha256_oid(b"p4-concurrent-winner")
1491 branch_row.head_commit_id = concurrent_commit_id
1492 await db_session.commit()
1493
1494 # Now push a commit that is a child of commit0 — diverges from the branch
1495 snap1_id = _sha256_oid(b"p4-nff-snap-1")
1496 commit1 = _make_commit(snapshot_id=snap1_id, parent_ids=[commit0["commit_id"]])
1497 snap1 = _make_snapshot(snap1_id)
1498
1499 async def body1() -> None:
1500 yield (
1501 _header_frame(n_commits=1, force=False)
1502 + _commit_pack_frame([commit1], [snap1])
1503 + _end_frame(n_commits=1)
1504 )
1505
1506 frames1 = await _collect_frames(
1507 wire_push_stream(db_session, str(repo.repo_id), body1(), "gabriel")
1508 )
1509 error_frames = [f for f in frames1 if f.get("t") == SFRAME_ERROR]
1510 assert error_frames, (
1511 "Expected non-FF push to be rejected after branch was concurrently advanced. "
1512 f"Got frames: {[f.get('t') for f in frames1]}"
1513 )
1514 assert "non-fast-forward" in error_frames[0]["msg"].lower(), (
1515 f"Expected non-fast-forward error, got: {error_frames[0]['msg']!r}"
1516 )
File History 1 commit
sha256:bb2baaabdd19320bde50cb69d447fd1c1e571467df729be1a23b5e93064b046a feat(intel): standardize headers, gauge icon, velocity card… Sonnet 4.6 minor ⚠ 143 days ago