gabriel / musehub public
test_wire_push_plain.py python
211 lines 7.3 KB
Raw
sha256:a34090cc4a394a78bd72cbbe34b08cc59525141e19135b6c0ab154f10611b9ef debug(push/stream): instrument O-frame decode path with INF… Sonnet 4.6 patch 121 days ago
1 """TDD — server reads raw MWP frames and returns a plain final response.
2
3 Rules:
4
5 S7 push/stream returns Content-Type: application/x-muse-wire.
6
7 S8 A valid push with raw MWP frames yields ok=True.
8 Full round-trip: client sends raw MWP, server parses, returns result.
9
10 S9 Auth is checked before any frames are consumed.
11 """
12 from __future__ import annotations
13
14 import inspect
15 from collections.abc import AsyncIterator
16 from datetime import datetime, timezone
17 from unittest.mock import AsyncMock, MagicMock
18
19 import msgpack
20 import pytest
21 from httpx import AsyncClient
22 from sqlalchemy.ext.asyncio import AsyncSession
23
24 from muse.core.types import blob_id, now_utc_iso
25 from musehub.types.json_types import JSONObject, JSONValue, StrDict
26 from musehub.models.wire import (
27 SFRAME_COMMIT_PACK,
28 SFRAME_END,
29 SFRAME_HEADER,
30 SFRAME_RESULT,
31 )
32 from muse.core.mpack import WIRE_CONTENT_TYPE, MuseWireFrameWriter
33 from tests.factories import create_repo
34
35 _fw = MuseWireFrameWriter()
36
37
38 def _last_frame(raw: bytes) -> JSONObject:
39 unpacker = msgpack.Unpacker(raw=False)
40 unpacker.feed(raw)
41 last: JSONObject = {}
42 for frame in unpacker:
43 last = frame
44 return last
45
46
47 # ---------------------------------------------------------------------------
48 # Helpers — raw MWP (no gRPC prefix)
49 # ---------------------------------------------------------------------------
50
51 def _pack(obj: JSONValue) -> bytes:
52 return msgpack.packb(obj, use_bin_type=True)
53
54
55 def _wrap_raw(ft: str, data: JSONValue) -> bytes:
56 """Wrap a logical frame in a raw MWP envelope — NO gRPC prefix."""
57 return _fw.wrap(frame_type=ft, payload=_pack(data))
58
59
60 def _header_frame(n_objects: int = 0, n_commits: int = 1) -> bytes:
61 return _wrap_raw(SFRAME_HEADER, {
62 "t": SFRAME_HEADER,
63 "branch": "main",
64 "force": False,
65 "have": [],
66 "head": blob_id(b"head"),
67 "n_objects": n_objects,
68 "n_commits": n_commits,
69 })
70
71
72 def _commit_pack_frame(commits: list[dict], snapshots: list[dict]) -> bytes:
73 return _wrap_raw(SFRAME_COMMIT_PACK, {
74 "t": SFRAME_COMMIT_PACK,
75 "commits": commits,
76 "snapshots": snapshots,
77 })
78
79
80 def _end_frame(n_objects: int = 0, n_commits: int = 0) -> bytes:
81 return _wrap_raw(SFRAME_END, {"t": SFRAME_END, "n_objects": n_objects, "n_commits": n_commits})
82
83
84 def _make_commit(snapshot_id: str | None = None) -> JSONObject:
85 snap = snapshot_id or blob_id(b"default-snap")
86 return {
87 "commit_id": blob_id(f"commit-{now_utc_iso()}".encode()),
88 "parent_ids": [],
89 "snapshot_id": snap,
90 "branch": "main",
91 "message": "test commit",
92 "author": "gabriel",
93 "committed_at": now_utc_iso(),
94 "signature": "",
95 "signer_key_id": "",
96 "agent_id": "",
97 "model_id": "",
98 "metadata": {},
99 }
100
101
102 def _make_snapshot(snap_id: str) -> JSONObject:
103 return {"snapshot_id": snap_id, "manifest": {}}
104
105
106 def _stub_r2_backend(monkeypatch: pytest.MonkeyPatch) -> None:
107 backend = MagicMock()
108 backend.store_object = AsyncMock(return_value=None)
109 backend.object_exists = AsyncMock(return_value=False)
110 backend.get_object = AsyncMock(return_value=None)
111 monkeypatch.setattr("musehub.services.musehub_wire.get_backend", lambda: backend)
112
113
114 # ---------------------------------------------------------------------------
115 # S7 — push/stream response Content-Type is application/x-muse-wire
116 # ---------------------------------------------------------------------------
117
118 def test_s7_push_route_response_content_type() -> None:
119 """push/stream response Content-Type must be application/x-muse-wire."""
120 from musehub.api.routes import wire
121
122 source = inspect.getsource(wire.push_stream)
123 assert "WIRE_CONTENT_TYPE" in source or "application/x-muse-wire" in source, (
124 "push/stream route must set Content-Type: application/x-muse-wire. "
125 "Replace GRPC_CONTENT_TYPE with WIRE_CONTENT_TYPE."
126 )
127 assert "GRPC_CONTENT_TYPE" not in source, (
128 "push/stream route still uses GRPC_CONTENT_TYPE. "
129 "Replace with WIRE_CONTENT_TYPE."
130 )
131
132
133 # ---------------------------------------------------------------------------
134 # S8 — valid push with raw MWP frames yields ok=True
135 # ---------------------------------------------------------------------------
136
137 @pytest.mark.asyncio
138 async def test_s8_raw_mwp_push_succeeds(
139 client: AsyncClient,
140 db_session: AsyncSession,
141 auth_headers: StrDict,
142 monkeypatch: pytest.MonkeyPatch,
143 ) -> None:
144 """A push body with raw MWP frames (no gRPC prefix) must yield ok=True.
145
146 This is the core end-to-end test: client builds raw MWP frames,
147 sends them as a plain HTTPS POST, server reads and stores them,
148 returns a plain msgpack result body.
149 """
150 _stub_r2_backend(monkeypatch)
151 repo = await create_repo(db_session, owner="testuser", name="s8-plain-push")
152
153 snap_id = blob_id(b"s8-snap")
154 commit = _make_commit(snapshot_id=snap_id)
155 snap = _make_snapshot(snap_id)
156 body = _header_frame() + _commit_pack_frame([commit], [snap]) + _end_frame(n_commits=1)
157
158 # Confirm body does NOT start with gRPC prefix
159 assert body[:1] != b"\x00", "Test body should be raw MWP, not gRPC-prefixed"
160 assert body[:4] == b"muse", "Test body should start with MWP magic"
161
162 resp = await client.post(
163 f"/{repo.owner}/{repo.slug}/push/stream",
164 content=body,
165 headers={**auth_headers, "Content-Type": WIRE_CONTENT_TYPE},
166 )
167
168 assert resp.status_code == 200, f"Expected 200, got {resp.status_code}: {resp.text[:200]}"
169 assert resp.headers.get("content-type", "").startswith(WIRE_CONTENT_TYPE), (
170 f"Expected Content-Type {WIRE_CONTENT_TYPE}, got {resp.headers.get('content-type')}"
171 )
172
173 # Response is a plain msgpack dict, not a stream of gRPC frames
174 result = _last_frame(resp.content)
175 assert isinstance(result, dict), f"Expected dict response, got {type(result)}"
176 assert result.get("ok") is True, f"Expected ok=True, got: {result}"
177 assert "stored_commits" in result, f"Missing stored_commits in result: {result}"
178
179
180 # ---------------------------------------------------------------------------
181 # S9 — auth is still checked before stream consumption (unchanged)
182 # ---------------------------------------------------------------------------
183
184 @pytest.mark.asyncio
185 async def test_s9_auth_still_checked_before_stream(
186 client: AsyncClient,
187 db_session: AsyncSession,
188 monkeypatch: pytest.MonkeyPatch,
189 ) -> None:
190 """Unauthenticated push must be rejected without consuming the request body."""
191 _stub_r2_backend(monkeypatch)
192 repo = await create_repo(db_session, owner="testuser", name="s9-auth-plain")
193
194 wire_was_called = []
195
196 async def _spy(session: AsyncSession, repo_id: str, body_iter: AsyncIterator[bytes], pusher_id: str | None) -> None:
197 wire_was_called.append(True)
198
199 monkeypatch.setattr("musehub.api.routes.wire.wire_push_stream", _spy)
200
201 body = _header_frame() + _end_frame()
202 resp = await client.post(
203 f"/{repo.owner}/{repo.slug}/push/stream",
204 content=body,
205 headers={"Content-Type": WIRE_CONTENT_TYPE},
206 )
207
208 assert resp.status_code in (401, 403), (
209 f"Unauthenticated push should return 401/403, got {resp.status_code}"
210 )
211 assert not wire_was_called, "wire_push_stream called despite missing auth"
File History 1 commit
sha256:a34090cc4a394a78bd72cbbe34b08cc59525141e19135b6c0ab154f10611b9ef debug(push/stream): instrument O-frame decode path with INF… Sonnet 4.6 patch 121 days ago