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