gabriel / muse public
test_wire_push_plain.py python
72 lines 2.7 KB
Raw
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9 docs: expand cache plan with all seven testing tiers and do… Sonnet 4.6 131 days ago
1 """TDD — push uses plain MWP frames over HTTPS POST.
2
3 Rules:
4
5 P1 _frame_generator yields raw MWP frames — each starts with b"muse" magic.
6
7 P2 push_stream_coro sends Content-Type: application/x-muse-wire.
8
9 P3 push_stream_coro uses client.post(), not client.stream() — upload-first
10 means we POST the full body and read the response after, not concurrently.
11 """
12 from __future__ import annotations
13
14 import inspect
15
16
17 # ---------------------------------------------------------------------------
18 # P1 — _frame_generator yields raw MWP frames
19 # ---------------------------------------------------------------------------
20
21 def test_p1_frame_generator_mwp_magic() -> None:
22 from muse.core.transport import _frame_generator
23
24 frames = list(_frame_generator([], [], [], local_head=None))
25 assert frames, "_frame_generator must yield at least H and E frames"
26 for i, frame in enumerate(frames):
27 assert len(frame) >= 4, f"Frame {i} too short: {len(frame)} bytes"
28 assert frame[:4] == b"muse", (
29 f"Frame {i} starts with {frame[:4]!r} — expected MWP magic b'muse'."
30 )
31
32
33 def test_p1_frame_generator_with_objects() -> None:
34 from muse.core._types import blob_id
35 from muse.core.transport import _frame_generator
36
37 raw = b"x" * 1024
38 oid = blob_id(raw)
39 obj = {"object_id": oid, "content": raw, "path": "a.txt", "encoding": "raw"}
40
41 frames = list(_frame_generator([obj], [], [], local_head=None))
42 assert len(frames) >= 3, "Expected H + O + E frames"
43 for i, frame in enumerate(frames):
44 assert frame[:4] == b"muse", (
45 f"Frame {i} starts with {frame[:4]!r}, not MWP magic."
46 )
47
48
49 # ---------------------------------------------------------------------------
50 # P2 — push_stream_coro uses application/x-muse-wire
51 # ---------------------------------------------------------------------------
52
53 def test_p2_push_uses_wire_content_type() -> None:
54 from muse.core import transport as _t
55
56 source = inspect.getsource(_t.HttpTransport.push_stream_coro)
57 assert "WIRE_CONTENT_TYPE" in source or "application/x-muse-wire" in source, (
58 "push_stream_coro must use WIRE_CONTENT_TYPE (application/x-muse-wire)."
59 )
60
61
62 # ---------------------------------------------------------------------------
63 # P3 — push_stream_coro uses client.post(), not client.stream()
64 # ---------------------------------------------------------------------------
65
66 def test_p3_push_upload_first() -> None:
67 from muse.core import transport as _t
68
69 source = inspect.getsource(_t.HttpTransport.push_stream_coro)
70 assert "client.stream(" not in source, (
71 "push_stream_coro uses client.stream() — replace with client.post()."
72 )
File History 1 commit
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9 docs: expand cache plan with all seven testing tiers and do… Sonnet 4.6 131 days ago