gabriel / muse public
test_mpack_transport.py python
859 lines 33.0 KB
Raw
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3 docs: docstring sprint contract→find-symbol — idiomatic run… Sonnet 4.6 patch 138 days ago
1 """TDD tests for MPack integration in transport.py.
2
3 These tests verify that HttpTransport.push_stream and fetch_stream use the
4 MPackStreamWriter/Reader API (muse.core.mpack) rather than the old inline
5 frame construction, and that LocalFileTransport has a push_stream method.
6
7 Red phase: all tests below are written against the *desired* behaviour.
8 Tests marked with their expected failure reason:
9 - content-type tests: current code uses "application/x-muse-packstream"
10 - H-frame "v" tests: current H frame omits the version field
11 - O-frame "sz" tests: current O frame omits the uncompressed size field
12 - compression tests: current code hardcodes zlib, not choose_compression()
13 - zstd decompression: fetch_stream only handles zlib today
14 - LocalFileTransport.push_stream: method does not exist yet
15 """
16
17 from __future__ import annotations
18
19 import hashlib
20 import json
21 import pathlib
22 import unittest.mock
23 from io import BytesIO
24
25 import msgpack
26 import pytest
27
28 from muse.core._types import long_id
29 from muse.core.mpack import (
30 MPACK_CONTENT_TYPE,
31 MPACK_VERSION,
32 MPackStreamReader,
33 MPackStreamWriter,
34 )
35 from muse.core.transport import (
36 HttpTransport,
37 LocalFileTransport,
38 PushResult,
39 SigningIdentity,
40 TransportError,
41 )
42
43
44 # ---------------------------------------------------------------------------
45 # Shared helpers
46 # ---------------------------------------------------------------------------
47
48
49 def _make_signing() -> SigningIdentity:
50 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
51 return SigningIdentity(handle="testuser", private_key=Ed25519PrivateKey.generate())
52
53
54 def _mock_stream_response(
55 frames_bytes: bytes,
56 status: int = 200,
57 content_type: str = MPACK_CONTENT_TYPE,
58 ) -> unittest.mock.MagicMock:
59 """Streaming response that returns frames_bytes then b""."""
60 resp = unittest.mock.MagicMock()
61 resp.read.side_effect = [frames_bytes, b""]
62 resp.headers = {"Content-Type": content_type}
63 resp.__enter__ = lambda s: s
64 resp.__exit__ = unittest.mock.MagicMock(return_value=False)
65 return resp
66
67
68 class _MockPushConn:
69 """Minimal http.client.HTTPConnection stand-in for push_stream tests."""
70
71 def __init__(self, response_body: bytes) -> None:
72 self.headers: dict[str, str] = {}
73 self.sends: list[bytes] = []
74 self._response_body = response_body
75 self.closed = False
76
77 def putrequest(self, method: str, path: str, **kw: object) -> None:
78 pass
79
80 def putheader(self, key: str, value: str) -> None:
81 self.headers[key.lower()] = value
82
83 def endheaders(self, body: bytes | None = None) -> None:
84 pass
85
86 def send(self, data: bytes) -> None:
87 self.sends.append(bytes(data))
88
89 def getresponse(self) -> object:
90 class _Resp:
91 def __init__(self, body: bytes) -> None:
92 self.status = 200
93 self._buf = body
94 self._pos = 0
95
96 def read(self, n: int = -1) -> bytes:
97 if n == -1:
98 out = self._buf[self._pos:]
99 self._pos = len(self._buf)
100 return out
101 out = self._buf[self._pos : self._pos + n]
102 self._pos += n
103 return out
104
105 def getheader(self, name: str, default: str | None = None) -> str | None:
106 return default
107 return _Resp(self._response_body)
108
109 def close(self) -> None:
110 self.closed = True
111
112
113 def _decode_push_frames(mock_conn: _MockPushConn) -> list[dict]:
114 """Reassemble frame bytes from chunked sends and decode MPack frames."""
115 import msgpack as _msgpack
116 frames = []
117 for raw in mock_conn.sends:
118 if raw == b"0\r\n\r\n":
119 continue
120 try:
121 # HTTP chunk: {hex_size}\r\n{data}\r\n
122 nl = raw.index(b"\r\n")
123 payload = raw[nl + 2 : -2]
124 obj = _msgpack.unpackb(payload, raw=False)
125 if isinstance(obj, dict):
126 frames.append(obj)
127 except Exception: # noqa: BLE001
128 pass
129 return frames
130
131
132 def _capture_push_request(
133 transport: HttpTransport,
134 *,
135 objects: list | None = None,
136 commits: list | None = None,
137 snapshots: list | None = None,
138 branch: str = "main",
139 force: bool = False,
140 have: list | None = None,
141 local_head: str | None = None,
142 server_result: bytes | None = None,
143 ) -> dict:
144 """Call push_stream via chunked _open_chunked_connection seam and capture state.
145
146 Returns a dict with keys:
147 content_type — Content-Type header sent on connection
148 accept — Accept header sent on connection
149 frames — list of decoded MPack frame dicts (from chunked sends)
150 """
151 writer = MPackStreamWriter()
152 result_frame = writer.write_result(ok=True, msg="ok", head=long_id("a" * 64))
153 response_body = server_result or result_frame
154
155 mock_conn = _MockPushConn(response_body)
156
157 def factory(host, port, *, use_ssl, timeout, context=None):
158 return mock_conn
159
160 with unittest.mock.patch(
161 "muse.core.transport._open_chunked_connection", side_effect=factory
162 ):
163 try:
164 transport.push_stream(
165 url="http://localhost:10003",
166 signing=None,
167 objects=objects or [],
168 commits=commits or [],
169 snapshots=snapshots or [],
170 branch=branch,
171 force=force,
172 have=have or [],
173 local_head=local_head,
174 )
175 except TransportError:
176 pass
177
178 captured: dict = {
179 "content_type": mock_conn.headers.get("content-type", ""),
180 "accept": mock_conn.headers.get("accept", ""),
181 "frames": _decode_push_frames(mock_conn),
182 }
183 return captured
184
185
186 def _run_push_via_chunked(
187 transport: HttpTransport,
188 *,
189 objects: list | None = None,
190 commits: list | None = None,
191 snapshots: list | None = None,
192 branch: str = "main",
193 signing: SigningIdentity | None = None,
194 response_body: bytes | None = None,
195 expect_error: bool = False,
196 ) -> tuple[_MockPushConn, dict]:
197 """Invoke push_stream via the chunked connection seam. Returns (conn, result)."""
198 writer = MPackStreamWriter()
199 default_resp = writer.write_result(ok=True, msg="ok", head=long_id("a" * 64))
200 body = response_body or default_resp
201 conn = _MockPushConn(body)
202
203 def factory(host, port, *, use_ssl, timeout, context=None):
204 return conn
205
206 result: dict = {}
207 with unittest.mock.patch(
208 "muse.core.transport._open_chunked_connection", side_effect=factory
209 ):
210 try:
211 result = transport.push_stream(
212 url="http://localhost:10003",
213 signing=signing,
214 objects=objects or [],
215 commits=commits or [],
216 snapshots=snapshots or [],
217 branch=branch,
218 force=False,
219 have=[],
220 )
221 except TransportError:
222 if not expect_error:
223 raise
224 return conn, result
225
226
227 def _capture_fetch_request(
228 transport: HttpTransport,
229 *,
230 want: list | None = None,
231 have: list | None = None,
232 server_frames: bytes | None = None,
233 ) -> dict:
234 """Call fetch_stream and capture outgoing request headers."""
235 writer = MPackStreamWriter()
236 end_frame = writer.write_end(n_objects=0, n_commits=0)
237 commit_frame = writer.write_commit_pack(commits=[], snapshots=[])
238 h_frame = writer.write_header(
239 op="fetch", branch="main", n_objects=0, n_commits=0,
240 branch_heads={}, repo_id="r", domain="code", default_branch="main",
241 )
242 response_body = server_frames or (h_frame + commit_frame + end_frame)
243
244 captured: dict = {}
245
246 def fake_open(req, timeout):
247 captured["content_type"] = req.get_header("Content-type")
248 captured["accept"] = req.get_header("Accept")
249 captured["body"] = req.data
250 return _mock_stream_response(response_body)
251
252 with unittest.mock.patch(
253 "muse.core.transport._open_url", side_effect=fake_open
254 ):
255 try:
256 transport.fetch_stream(
257 url="http://localhost:10003",
258 signing=None,
259 want=want or [],
260 have=have or [],
261 )
262 except (TransportError, Exception):
263 pass
264
265 return captured
266
267
268 def _make_repo(tmp_path: pathlib.Path) -> pathlib.Path:
269 repo = tmp_path / "repo"
270 muse = repo / ".muse"
271 for sub in ("objects", "commits", "snapshots", "refs/heads"):
272 (muse / sub).mkdir(parents=True)
273 (muse / "HEAD").write_text("ref: refs/heads/main")
274 (muse / "repo.json").write_text(
275 json.dumps({"repo_id": "test-repo", "domain": "code", "default_branch": "main"})
276 )
277 return repo
278
279
280 # ---------------------------------------------------------------------------
281 # HttpTransport.push_stream — content-type
282 # ---------------------------------------------------------------------------
283
284
285 class TestPushStreamContentType:
286 def test_push_stream_uses_mpack_content_type(self) -> None:
287 """push_stream must set Content-Type: application/x-muse-mpack."""
288 t = HttpTransport()
289 cap = _capture_push_request(t)
290 assert cap.get("content_type", "").lower() == MPACK_CONTENT_TYPE.lower(), (
291 f"Expected Content-Type {MPACK_CONTENT_TYPE!r}, got {cap.get('content_type')!r}"
292 )
293
294 def test_push_stream_accept_uses_mpack_content_type(self) -> None:
295 """push_stream must set Accept: application/x-muse-mpack."""
296 t = HttpTransport()
297 cap = _capture_push_request(t)
298 assert cap.get("accept", "").lower() == MPACK_CONTENT_TYPE.lower(), (
299 f"Expected Accept {MPACK_CONTENT_TYPE!r}, got {cap.get('accept')!r}"
300 )
301
302
303 # ---------------------------------------------------------------------------
304 # HttpTransport.push_stream — H frame
305 # ---------------------------------------------------------------------------
306
307
308 class TestPushStreamHFrame:
309 def test_h_frame_is_first(self) -> None:
310 """First frame must be H."""
311 t = HttpTransport()
312 cap = _capture_push_request(t)
313 frames = cap["frames"]
314 assert frames, "No frames decoded from push body"
315 assert frames[0].get("t") == "H", f"Expected H frame first, got {frames[0].get('t')!r}"
316
317 def test_h_frame_has_protocol_version(self) -> None:
318 """H frame must carry v=MPACK_VERSION."""
319 t = HttpTransport()
320 cap = _capture_push_request(t)
321 h = next((f for f in cap["frames"] if f.get("t") == "H"), None)
322 assert h is not None, "No H frame in push body"
323 assert h.get("v") == MPACK_VERSION, (
324 f"H frame missing v={MPACK_VERSION!r}, got v={h.get('v')!r}"
325 )
326
327 def test_h_frame_has_op_push(self) -> None:
328 """H frame op must be 'push'."""
329 t = HttpTransport()
330 cap = _capture_push_request(t)
331 h = next((f for f in cap["frames"] if f.get("t") == "H"), None)
332 assert h is not None
333 assert h.get("op") == "push", f"H frame op={h.get('op')!r}, expected 'push'"
334
335 def test_h_frame_branch_matches(self) -> None:
336 """H frame branch must match the requested branch."""
337 t = HttpTransport()
338 cap = _capture_push_request(t, branch="dev")
339 h = next((f for f in cap["frames"] if f.get("t") == "H"), None)
340 assert h is not None
341 assert h.get("branch") == "dev"
342
343 def test_h_frame_n_objects_advisory_count(self) -> None:
344 """H frame n_objects must equal the number of objects passed."""
345 t = HttpTransport()
346 content = b"hello mpack"
347 oid = long_id(hashlib.sha256(content).hexdigest())
348 objects = [{"object_id": oid, "content": content, "path": "a.txt", "encoding": "raw"}]
349 cap = _capture_push_request(t, objects=objects)
350 h = next((f for f in cap["frames"] if f.get("t") == "H"), None)
351 assert h is not None
352 assert h.get("n_objects") == 1
353
354
355 # ---------------------------------------------------------------------------
356 # HttpTransport.push_stream — O frames
357 # ---------------------------------------------------------------------------
358
359
360 class TestPushStreamOFrames:
361 def test_o_frame_has_sz_field(self) -> None:
362 """O frame must carry sz = uncompressed byte length."""
363 t = HttpTransport()
364 content = b"object content for sz test"
365 oid = long_id(hashlib.sha256(content).hexdigest())
366 objects = [{"object_id": oid, "content": content, "path": "f.txt", "encoding": "raw"}]
367 cap = _capture_push_request(t, objects=objects)
368 o_frames = [f for f in cap["frames"] if f.get("t") == "O"]
369 assert o_frames, "No O frames in push body"
370 for o in o_frames:
371 assert "sz" in o, f"O frame missing 'sz' field: {list(o.keys())}"
372 assert o["sz"] == len(content), (
373 f"O frame sz={o['sz']}, expected {len(content)}"
374 )
375
376 def test_o_frame_id_matches_object_id(self) -> None:
377 """O frame 'id' must match the object_id."""
378 t = HttpTransport()
379 content = b"id match test"
380 oid = long_id(hashlib.sha256(content).hexdigest())
381 objects = [{"object_id": oid, "content": content, "path": "g.txt", "encoding": "raw"}]
382 cap = _capture_push_request(t, objects=objects)
383 o_frames = [f for f in cap["frames"] if f.get("t") == "O"]
384 assert o_frames
385 assert o_frames[0].get("id") == oid
386
387 def test_o_frame_content_decompresses_to_original(self) -> None:
388 """O frame content must decompress back to the original bytes."""
389 t = HttpTransport()
390 content = b"round-trip compression test " * 20
391 oid = long_id(hashlib.sha256(content).hexdigest())
392 objects = [{"object_id": oid, "content": content, "path": "h.txt", "encoding": "raw"}]
393 cap = _capture_push_request(t, objects=objects)
394 o_frames = [f for f in cap["frames"] if f.get("t") == "O"]
395 assert o_frames
396 reader = MPackStreamReader()
397 recovered = reader.decompress_object(o_frames[0])
398 assert recovered == content, "Decompressed O frame content does not match original"
399
400 def test_o_frame_uses_best_compression(self) -> None:
401 """O frame enc must be whatever choose_compression() returns (zstd or zlib)."""
402 from muse.core.compression import choose_compression
403 t = HttpTransport()
404 content = b"compression selection test " * 10
405 oid = long_id(hashlib.sha256(content).hexdigest())
406 objects = [{"object_id": oid, "content": content, "path": "i.txt", "encoding": "raw"}]
407 cap = _capture_push_request(t, objects=objects)
408 o_frames = [f for f in cap["frames"] if f.get("t") == "O"]
409 assert o_frames
410 expected_enc = choose_compression()
411 assert o_frames[0].get("enc") == expected_enc, (
412 f"O frame enc={o_frames[0].get('enc')!r}, expected {expected_enc!r} "
413 f"from choose_compression()"
414 )
415
416 def test_multiple_objects_all_have_sz(self) -> None:
417 """All O frames in a multi-object push must have sz."""
418 t = HttpTransport()
419 objects = []
420 for i in range(5):
421 content = f"object-{i}".encode() * 10
422 oid = long_id(hashlib.sha256(content).hexdigest())
423 objects.append({"object_id": oid, "content": content, "path": f"{i}.txt", "encoding": "raw"})
424 cap = _capture_push_request(t, objects=objects)
425 o_frames = [f for f in cap["frames"] if f.get("t") == "O"]
426 assert len(o_frames) == 5
427 for o in o_frames:
428 assert "sz" in o, f"O frame missing sz: {list(o.keys())}"
429
430
431 # ---------------------------------------------------------------------------
432 # HttpTransport.push_stream — frame sequence
433 # ---------------------------------------------------------------------------
434
435
436 class TestPushStreamFrameSequence:
437 def test_frame_sequence_h_c_e(self) -> None:
438 """Frame sequence for empty push must be H → C → E."""
439 t = HttpTransport()
440 cap = _capture_push_request(t)
441 tags = [f.get("t") for f in cap["frames"]]
442 assert "H" in tags, f"Missing H frame. Tags: {tags}"
443 assert "C" in tags, f"Missing C frame. Tags: {tags}"
444 assert "E" in tags, f"Missing E frame. Tags: {tags}"
445 assert tags.index("H") < tags.index("C") < tags.index("E"), (
446 f"Frame order wrong: {tags}"
447 )
448
449 def test_frame_sequence_h_o_c_e_with_objects(self) -> None:
450 """Frame sequence with objects must be H → O... → C → E."""
451 t = HttpTransport()
452 content = b"seq test"
453 oid = long_id(hashlib.sha256(content).hexdigest())
454 objects = [{"object_id": oid, "content": content, "path": "s.txt", "encoding": "raw"}]
455 cap = _capture_push_request(t, objects=objects)
456 tags = [f.get("t") for f in cap["frames"]]
457 assert tags[0] == "H", f"First frame must be H, got {tags[0]!r}"
458 assert tags[-1] == "E", f"Last frame must be E, got {tags[-1]!r}"
459 assert "O" in tags
460 assert "C" in tags
461 h_idx = tags.index("H")
462 o_idx = tags.index("O")
463 c_idx = tags.index("C")
464 e_idx = tags.index("E")
465 assert h_idx < o_idx < c_idx < e_idx, f"Frame order wrong: {tags}"
466
467 def test_e_frame_n_objects_matches_actual(self) -> None:
468 """E frame n_objects must equal the actual number of O frames sent."""
469 t = HttpTransport()
470 objects = []
471 for i in range(3):
472 content = f"obj-{i}".encode()
473 oid = long_id(hashlib.sha256(content).hexdigest())
474 objects.append({"object_id": oid, "content": content, "path": f"{i}.txt", "encoding": "raw"})
475 cap = _capture_push_request(t, objects=objects)
476 e = next((f for f in cap["frames"] if f.get("t") == "E"), None)
477 assert e is not None, "No E frame found"
478 assert e.get("n_objects") == 3, f"E frame n_objects={e.get('n_objects')}, expected 3"
479
480
481 # ---------------------------------------------------------------------------
482 # HttpTransport.push_stream — response parsing
483 # ---------------------------------------------------------------------------
484
485
486 class TestPushStreamResponseParsing:
487 def test_progress_frame_does_not_raise(self) -> None:
488 """P frames in server response must not raise."""
489 t = HttpTransport()
490 writer = MPackStreamWriter()
491 response = (
492 writer.write_progress(msg="resolving objects", pct=50.0)
493 + writer.write_result(ok=True, msg="ok", head=long_id("a" * 64))
494 )
495 # Should not raise
496 _capture_push_request(t, server_result=response)
497
498 def test_error_frame_raises_transport_error(self) -> None:
499 """X frame in server response must raise TransportError."""
500 t = HttpTransport()
501 writer = MPackStreamWriter()
502 response = writer.write_error(msg="conflict", code=409)
503 _, result_dict = _run_push_via_chunked(t, response_body=response, expect_error=True)
504 # error path — TransportError is caught; result_dict is empty
505 assert result_dict == {}
506
507 def test_error_frame_status_code(self) -> None:
508 """X frame code=409 must become TransportError with status_code=409."""
509 t = HttpTransport()
510 writer = MPackStreamWriter()
511 response = writer.write_error(msg="conflict", code=409)
512
513 def factory(host, port, *, use_ssl, timeout, context=None):
514 return _MockPushConn(response)
515
516 with unittest.mock.patch(
517 "muse.core.transport._open_chunked_connection", side_effect=factory
518 ):
519 with pytest.raises(TransportError) as exc_info:
520 t.push_stream(
521 url="http://localhost:10003",
522 signing=None,
523 objects=[],
524 commits=[],
525 snapshots=[],
526 branch="main",
527 force=False,
528 have=[],
529 )
530 assert exc_info.value.status_code == 409
531
532 def test_result_frame_ok_returns_push_result(self) -> None:
533 """R frame with ok=True returns PushResult(ok=True)."""
534 t = HttpTransport()
535 writer = MPackStreamWriter()
536 tip = long_id("b" * 64)
537 response = writer.write_result(ok=True, msg="pushed", heads={"main": tip}, head=tip)
538 _, result = _run_push_via_chunked(t, response_body=response)
539 assert isinstance(result, dict)
540 assert result["ok"] is True
541
542
543 # ---------------------------------------------------------------------------
544 # HttpTransport.fetch_stream — content-type
545 # ---------------------------------------------------------------------------
546
547
548 class TestFetchStreamContentType:
549 def test_fetch_stream_uses_mpack_content_type(self) -> None:
550 """fetch_stream must set Content-Type: application/x-muse-mpack."""
551 t = HttpTransport()
552 cap = _capture_fetch_request(t)
553 assert cap.get("content_type", "").lower() == MPACK_CONTENT_TYPE.lower(), (
554 f"Expected {MPACK_CONTENT_TYPE!r}, got {cap.get('content_type')!r}"
555 )
556
557 def test_fetch_stream_accept_uses_mpack_content_type(self) -> None:
558 """fetch_stream must set Accept: application/x-muse-mpack."""
559 t = HttpTransport()
560 cap = _capture_fetch_request(t)
561 assert cap.get("accept", "").lower() == MPACK_CONTENT_TYPE.lower(), (
562 f"Expected {MPACK_CONTENT_TYPE!r}, got {cap.get('accept')!r}"
563 )
564
565
566 # ---------------------------------------------------------------------------
567 # HttpTransport.fetch_stream — O frame decompression
568 # ---------------------------------------------------------------------------
569
570
571 class TestFetchStreamDecompression:
572 def _make_fetch_response_with_object(self, enc: str, content: bytes, oid: str) -> bytes:
573 """Build a fetch response stream containing one O frame with given enc."""
574 from muse.core.compression import compress_zlib, compress_zstd
575 writer = MPackStreamWriter()
576 if enc == "zstd":
577 wire = compress_zstd(content)
578 elif enc == "zlib":
579 wire = compress_zlib(content)
580 else:
581 wire = content
582
583 h = writer.write_header(
584 op="fetch", branch="main", n_objects=1, n_commits=0,
585 repo_id="r", domain="code", default_branch="main", branch_heads={},
586 )
587 o = writer.write_object(
588 object_id=oid, content=wire, enc=enc, sz=len(content), path="a.txt",
589 )
590 c = writer.write_commit_pack(commits=[], snapshots=[])
591 e = writer.write_end(n_objects=1, n_commits=0)
592 return h + o + c + e
593
594 def test_fetch_stream_decompresses_zlib_objects(self) -> None:
595 """fetch_stream must decompress zlib-encoded O frames correctly."""
596 content = b"zlib content for fetch " * 5
597 oid = long_id(hashlib.sha256(content).hexdigest())
598 server_frames = self._make_fetch_response_with_object("zlib", content, oid)
599
600 received: list = []
601 t = HttpTransport()
602 with unittest.mock.patch(
603 "muse.core.transport._open_url",
604 return_value=_mock_stream_response(server_frames),
605 ):
606 t.fetch_stream(
607 url="http://localhost:10003",
608 signing=None,
609 want=[oid],
610 have=[],
611 on_object=received.append,
612 )
613 assert len(received) == 1
614 assert received[0]["content"] == content
615
616 def test_fetch_stream_decompresses_zstd_objects(self) -> None:
617 """fetch_stream must decompress zstd-encoded O frames (via MPackStreamReader)."""
618 from muse.core.compression import ZSTD_AVAILABLE
619 if not ZSTD_AVAILABLE:
620 pytest.skip("zstd not available on this machine")
621
622 content = b"zstd content for fetch " * 5
623 oid = long_id(hashlib.sha256(content).hexdigest())
624 server_frames = self._make_fetch_response_with_object("zstd", content, oid)
625
626 received: list = []
627 t = HttpTransport()
628 with unittest.mock.patch(
629 "muse.core.transport._open_url",
630 return_value=_mock_stream_response(server_frames),
631 ):
632 t.fetch_stream(
633 url="http://localhost:10003",
634 signing=None,
635 want=[oid],
636 have=[],
637 on_object=received.append,
638 )
639 assert len(received) == 1, f"Expected 1 object, got {len(received)}"
640 assert received[0]["content"] == content, "Decompressed content mismatch"
641
642 def test_fetch_stream_objects_received_count(self) -> None:
643 """fetch_stream result.objects_received must equal actual O frames."""
644 content = b"count test"
645 oid = long_id(hashlib.sha256(content).hexdigest())
646 server_frames = self._make_fetch_response_with_object("zlib", content, oid)
647
648 t = HttpTransport()
649 with unittest.mock.patch(
650 "muse.core.transport._open_url",
651 return_value=_mock_stream_response(server_frames),
652 ):
653 result = t.fetch_stream(
654 url="http://localhost:10003",
655 signing=None,
656 want=[oid],
657 have=[],
658 )
659 assert result["objects_received"] == 1
660
661
662 # ---------------------------------------------------------------------------
663 # LocalFileTransport.push_stream
664 # ---------------------------------------------------------------------------
665
666
667 class TestLocalFileTransportPushStream:
668 def test_push_stream_method_exists(self) -> None:
669 """LocalFileTransport must have a push_stream method."""
670 t = LocalFileTransport()
671 assert hasattr(t, "push_stream"), (
672 "LocalFileTransport is missing push_stream — method not implemented"
673 )
674 assert callable(t.push_stream)
675
676 def test_push_stream_writes_commit_to_remote(self, tmp_path: pathlib.Path) -> None:
677 """push_stream must write commits to the remote store."""
678 import datetime
679 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
680 from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot
681
682 src = _make_repo(tmp_path / "src")
683 dst = _make_repo(tmp_path / "dst")
684
685 snap_id = compute_snapshot_id({})
686 write_snapshot(src, SnapshotRecord(
687 snapshot_id=snap_id,
688 manifest={},
689 created_at=datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc),
690 ))
691 committed_at = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
692 commit_id = compute_commit_id([], snap_id, "test push_stream", committed_at.isoformat())
693 write_commit(src, CommitRecord(
694 commit_id=commit_id,
695 repo_id="test-repo",
696 branch="main",
697 snapshot_id=snap_id,
698 message="test push_stream",
699 committed_at=committed_at,
700 parent_commit_id=None,
701 ))
702
703 # Build commits/snapshots as dicts to pass to push_stream
704 commits = [{"commit_id": commit_id, "repo_id": "test-repo", "branch": "main",
705 "snapshot_id": snap_id, "message": "test push_stream",
706 "committed_at": "2026-01-01T00:00:00+00:00", "parent_commit_id": None,
707 "parent2_commit_id": None, "agent_id": "", "model_id": "", "tags": []}]
708 snapshots = [{"snapshot_id": snap_id, "manifest": {},
709 "created_at": "2026-01-01T00:00:00+00:00"}]
710
711 t = LocalFileTransport()
712 result = t.push_stream(
713 url=dst.as_uri(),
714 signing=None,
715 objects=[],
716 commits=commits,
717 snapshots=snapshots,
718 branch="main",
719 force=False,
720 have=[],
721 local_head=commit_id,
722 )
723 assert isinstance(result, dict)
724 assert result["ok"], f"push_stream returned ok=False: {result.get('message')}"
725
726 # Commit must exist in dst store
727 from muse.core.store import _commit_path
728 dst_commit_path = _commit_path(dst, commit_id)
729 assert dst_commit_path.exists(), (
730 f"Commit not found in dst store at {dst_commit_path}"
731 )
732
733 def test_push_stream_writes_objects_to_remote(self, tmp_path: pathlib.Path) -> None:
734 """push_stream must write objects to the remote object store."""
735 import datetime
736 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
737 from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot
738
739 src = _make_repo(tmp_path / "src")
740 dst = _make_repo(tmp_path / "dst")
741
742 content = b"binary object for push_stream"
743 oid = long_id(hashlib.sha256(content).hexdigest())
744
745 snap_id = compute_snapshot_id({"file.bin": oid})
746 write_snapshot(src, SnapshotRecord(
747 snapshot_id=snap_id,
748 manifest={"file.bin": oid},
749 created_at=datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc),
750 ))
751 committed_at = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
752 commit_id = compute_commit_id([], snap_id, "obj test", committed_at.isoformat())
753 write_commit(src, CommitRecord(
754 commit_id=commit_id,
755 repo_id="test-repo",
756 branch="main",
757 snapshot_id=snap_id,
758 message="obj test",
759 committed_at=committed_at,
760 parent_commit_id=None,
761 ))
762
763 commits = [{"commit_id": commit_id, "repo_id": "test-repo", "branch": "main",
764 "snapshot_id": snap_id, "message": "obj test",
765 "committed_at": "2026-01-01T00:00:00+00:00", "parent_commit_id": None,
766 "parent2_commit_id": None, "agent_id": "", "model_id": "", "tags": []}]
767 snapshots = [{"snapshot_id": snap_id, "manifest": {"file.bin": oid},
768 "created_at": "2026-01-01T00:00:00+00:00"}]
769 objects = [{"object_id": oid, "content": content, "path": "file.bin", "encoding": "raw"}]
770
771 t = LocalFileTransport()
772 result = t.push_stream(
773 url=dst.as_uri(),
774 signing=None,
775 objects=objects,
776 commits=commits,
777 snapshots=snapshots,
778 branch="main",
779 force=False,
780 have=[],
781 local_head=commit_id,
782 )
783 assert result["ok"], f"push_stream returned ok=False: {result.message}"
784
785 # Object must exist in dst object store
786 from muse.core.object_store import object_path
787 dst_obj = object_path(dst, oid)
788 assert dst_obj.exists(), f"Object not found in dst at {dst_obj}"
789
790 def test_push_stream_updates_branch_ref(self, tmp_path: pathlib.Path) -> None:
791 """push_stream must update the remote branch ref to the new tip."""
792 import datetime
793 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
794 from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot
795
796 src = _make_repo(tmp_path / "src")
797 dst = _make_repo(tmp_path / "dst")
798
799 snap_id = compute_snapshot_id({})
800 write_snapshot(src, SnapshotRecord(
801 snapshot_id=snap_id,
802 manifest={},
803 created_at=datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc),
804 ))
805 committed_at = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
806 commit_id = compute_commit_id([], snap_id, "branch ref test", committed_at.isoformat())
807 write_commit(src, CommitRecord(
808 commit_id=commit_id,
809 repo_id="test-repo",
810 branch="main",
811 snapshot_id=snap_id,
812 message="branch ref test",
813 committed_at=committed_at,
814 parent_commit_id=None,
815 ))
816
817 commits = [{"commit_id": commit_id, "repo_id": "test-repo", "branch": "main",
818 "snapshot_id": snap_id, "message": "branch ref test",
819 "committed_at": "2026-01-01T00:00:00+00:00", "parent_commit_id": None,
820 "parent2_commit_id": None, "agent_id": "", "model_id": "", "tags": []}]
821 snapshots = [{"snapshot_id": snap_id, "manifest": {},
822 "created_at": "2026-01-01T00:00:00+00:00"}]
823
824 t = LocalFileTransport()
825 result = t.push_stream(
826 url=dst.as_uri(),
827 signing=None,
828 objects=[],
829 commits=commits,
830 snapshots=snapshots,
831 branch="main",
832 force=False,
833 have=[],
834 local_head=commit_id,
835 )
836 assert result["ok"]
837
838 # Branch ref must point to the commit
839 ref_path = dst / ".muse" / "refs" / "heads" / "main"
840 assert ref_path.exists(), "Branch ref not created"
841 assert ref_path.read_text().strip() == commit_id
842
843 def test_push_stream_returns_push_result(self, tmp_path: pathlib.Path) -> None:
844 """push_stream must return a PushResult (dict with ok, message, branch_heads)."""
845 dst = _make_repo(tmp_path / "dst")
846 t = LocalFileTransport()
847 result = t.push_stream(
848 url=dst.as_uri(),
849 signing=None,
850 objects=[],
851 commits=[],
852 snapshots=[],
853 branch="main",
854 force=False,
855 have=[],
856 )
857 assert isinstance(result, dict)
858 assert "ok" in result
859 assert "message" in result or "branch_heads" in result
File History 1 commit
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3 docs: docstring sprint contract→find-symbol — idiomatic run… Sonnet 4.6 patch 138 days ago